AAR 内置 SO 冲突怎样在加固前发现

#!/usr/bin/env python3
import hashlib
import json
import re
import subprocess
import sys
import tempfile
import zipfile
from collections import defaultdict
from pathlib import Path

SO_PATH = re.compile(r"^jni/([^/]+)/([^/]+\.so)$")
SAFE_ABI = re.compile(r"^[A-Za-z0-9_-]+$")
MAX_MEMBERS = 20000
MAX_SO_BYTES = 256 * 1024 * 1024

def fail(message, code):
    print(message, file=sys.stderr)
    raise SystemExit(code)

def sha256_file(path):
    digest = hashlib.sha256()
    with path.open("rb") as stream:
        for block in iter(lambda: stream.read(1024 * 1024), b""):
            digest.update(block)
    return digest.hexdigest()

def sha256_bytes(value):
    return hashlib.sha256(value).hexdigest()

def load_required_abis(path):
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        fail("cannot read required ABI file: " + str(exc), 3)
    if not isinstance(value, dict) or not isinstance(value.get("requiredAbis"), list):
        fail("requiredAbis must be a list", 4)
    abis = value["requiredAbis"]
    if not abis or not all(isinstance(x, str) and SAFE_ABI.fullmatch(x) for x in abis):
        fail("requiredAbis contains invalid values", 5)
    if len(abis) != len(set(abis)):
        fail("requiredAbis contains duplicates", 6)
    return sorted(abis)

def parse_readelf(path):
    try:
        header = subprocess.run(
            ["readelf", "--file-header", str(path)],
            check=True, capture_output=True, text=True, timeout=20
        ).stdout
        dynamic = subprocess.run(
            ["readelf", "--dynamic", str(path)],
            check=True, capture_output=True, text=True, timeout=20
        ).stdout
    except (OSError, subprocess.SubprocessError) as exc:
        fail("readelf failed: " + str(exc), 7)
    elf_class = None
    machine = None
    for line in header.splitlines():
        stripped = line.strip()
        if stripped.startswith("Class:"):
            elf_class = stripped.split(":", 1)[1].strip()
        if stripped.startswith("Machine:"):
            machine = stripped.split(":", 1)[1].strip()
    soname = None
    needed = []
    for line in dynamic.splitlines():
        if "(SONAME)" in line and "[" in line and "]" in line:
            soname = line.rsplit("[", 1)[1].split("]", 1)[0]
        if "(NEEDED)" in line and "[" in line and "]" in line:
            needed.append(line.rsplit("[", 1)[1].split("]", 1)[0])
    if not elf_class or not machine:
        fail("ELF header fields are missing", 8)
    return {
        "elfClass": elf_class,
        "machine": machine,
        "soname": soname,
        "needed": sorted(set(needed)),
    }

def scan_aar(aar_path, temporary_root):
    records = []
    try:
        archive = zipfile.ZipFile(aar_path)
    except (OSError, zipfile.BadZipFile) as exc:
        fail("invalid AAR: " + str(exc), 9)
    with archive:
        members = archive.infolist()
        if len(members) > MAX_MEMBERS:
            fail("AAR has too many members: " + aar_path.name, 10)
        for member in members:
            match = SO_PATH.fullmatch(member.filename)
            if not match:
                continue
            abi, name = match.groups()
            if not SAFE_ABI.fullmatch(abi) or Path(name).name != name:
                fail("unsafe SO path: " + member.filename, 11)
            if member.file_size <= 0 or member.file_size > MAX_SO_BYTES:
                fail("invalid SO size: " + member.filename, 12)
            try:
                data = archive.read(member)
            except (OSError, RuntimeError, zipfile.BadZipFile) as exc:
                fail("cannot read SO: " + str(exc), 13)
            target = temporary_root / (sha256_bytes((aar_path.name + member.filename).encode()) + ".so")
            target.write_bytes(data)
            elf = parse_readelf(target)
            records.append({
                "aar": aar_path.name,
                "aarSha256": sha256_file(aar_path),
                "path": member.filename,
                "abi": abi,
                "name": name,
                "soSha256": sha256_bytes(data),
                **elf,
            })
    return records

def analyze(records, required_abis):
    by_path = defaultdict(list)
    by_soname = defaultdict(list)
    by_name_abis = defaultdict(set)
    for record in records:
        by_path[(record["abi"], record["name"])].append(record)
        if record["soname"]:
            by_soname[(record["abi"], record["soname"])].append(record)
        by_name_abis[record["name"]].add(record["abi"])
    duplicates = []
    for key, providers in sorted(by_path.items()):
        if len(providers) < 2:
            continue
        hashes = {item["soSha256"] for item in providers}
        duplicates.append({
            "abi": key[0],
            "name": key[1],
            "kind": "identical-duplicate" if len(hashes) == 1 else "conflicting-content",
            "providers": providers,
        })
    soname_conflicts = []
    for key, providers in sorted(by_soname.items()):
        hashes = {item["soSha256"] for item in providers}
        names = {item["name"] for item in providers}
        if len(hashes) > 1 or len(names) > 1:
            soname_conflicts.append({
                "abi": key[0],
                "soname": key[1],
                "providers": providers,
            })
    gaps = []
    required = set(required_abis)
    for name, present in sorted(by_name_abis.items()):
        missing = sorted(required - present)
        if missing:
            gaps.append({"name": name, "present": sorted(present), "missing": missing})
    return duplicates, soname_conflicts, gaps

def main():
    if len(sys.argv) != 3:
        print("Usage: aar_so_inventory.py AAR_DIRECTORY REQUIRED_ABIS_JSON", file=sys.stderr)
        raise SystemExit(2)
    aar_directory = Path(sys.argv[1])
    abi_path = Path(sys.argv[2])
    if not aar_directory.is_dir() or not abi_path.is_file():
        print("required input is missing", file=sys.stderr)
        raise SystemExit(2)
    aar_files = sorted(path for path in aar_directory.iterdir() if path.is_file() and path.suffix == ".aar")
    if not aar_files:
        fail("AAR directory is empty", 14)
    required_abis = load_required_abis(abi_path)
    with tempfile.TemporaryDirectory(prefix="aar-so-inventory-") as temporary:
        temporary_root = Path(temporary)
        records = []
        for aar_path in aar_files:
            records.extend(scan_aar(aar_path, temporary_root))
    if not records:
        fail("no embedded SO files found", 15)
    duplicates, soname_conflicts, gaps = analyze(records, required_abis)
    blocked = any(item["kind"] == "conflicting-content" for item in duplicates) or bool(soname_conflicts) or bool(gaps)
    result = {
        "status": "blocked" if blocked else "runtime-validation-required",
        "requiredAbis": required_abis,
        "records": records,
        "duplicates": duplicates,
        "sonameConflicts": soname_conflicts,
        "abiGaps": gaps,
    }
    print(json.dumps(result, ensure_ascii=False, indent=2, sort_keys=True))
    if blocked:
        raise SystemExit(20)

if __name__ == "__main__":
    main()