SO 符号版本与 ABI 兼容契约怎样验收

#!/usr/bin/env python3
import hashlib
import json
import re
import shutil
import subprocess
import sys
from pathlib import Path

SYMBOL_RE = re.compile(
    r"^\s*\d+:\s+([0-9a-fA-F]+)\s+(\d+)\s+(\S+)\s+(\S+)\s+(\S+)\s+(\S+)\s+(.+?)\s*$"
)
SONAME_RE = re.compile(r"\(SONAME\).*?\[(.+?)\]")
NEEDED_RE = re.compile(r"\(NEEDED\).*?\[(.+?)\]")
CLASS_RE = re.compile(r"^\s*Class:\s*(.+)$", re.MULTILINE)
MACHINE_RE = re.compile(r"^\s*Machine:\s*(.+)$", re.MULTILINE)

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

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

def run_readelf(readelf, args, path):
    command = [readelf, *args, str(path)]
    completed = subprocess.run(
        command,
        check=False,
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
        timeout=30,
    )
    if completed.returncode != 0:
        fail("readelf failed for " + path.name + ": " + completed.stderr.strip(), 10)
    return completed.stdout

def load_contract(path):
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, UnicodeError, json.JSONDecodeError) as exc:
        fail("cannot read contract: " + str(exc))
    if not isinstance(value, dict):
        fail("contract must be a JSON object")
    allowed = value.get("allowedRemovedInternalSymbols", [])
    if not isinstance(allowed, list) or any(not isinstance(item, str) or not item for item in allowed):
        fail("allowedRemovedInternalSymbols must be an array of non-empty strings")
    return set(allowed)

def split_name_and_version(raw_name):
    clean = raw_name.split()[0]
    if "@@" in clean:
        name, version = clean.split("@@", 1)
        return name, version, True
    if "@" in clean:
        name, version = clean.split("@", 1)
        return name, version, False
    return clean, "", False

def parse_symbols(text):
    symbols = {}
    for line in text.splitlines():
        match = SYMBOL_RE.match(line)
        if not match:
            continue
        value, size, kind, bind, visibility, ndx, raw_name = match.groups()
        name, version, is_default = split_name_and_version(raw_name)
        if not name or name == "0":
            continue
        key = name + "@" + version
        record = {
            "name": name,
            "version": version,
            "defaultVersion": is_default,
            "value": value,
            "size": int(size),
            "type": kind,
            "bind": bind,
            "visibility": visibility,
            "ndx": ndx,
        }
        if key in symbols and symbols[key] != record:
            fail("duplicate dynamic symbol key: " + key, 11)
        symbols[key] = record
    if not symbols:
        fail("no dynamic symbols were parsed", 12)
    return symbols

def parse_identity(header_and_dynamic):
    class_match = CLASS_RE.search(header_and_dynamic)
    machine_match = MACHINE_RE.search(header_and_dynamic)
    if not class_match or not machine_match:
        fail("ELF Class or Machine is missing", 13)
    soname_match = SONAME_RE.search(header_and_dynamic)
    needed = sorted(set(NEEDED_RE.findall(header_and_dynamic)))
    return {
        "class": class_match.group(1).strip(),
        "machine": machine_match.group(1).strip(),
        "soname": soname_match.group(1).strip() if soname_match else "",
        "needed": needed,
    }

def inspect(readelf, path):
    header_dynamic = run_readelf(readelf, ["--file-header", "--dynamic", "--wide"], path)
    dynsym = run_readelf(readelf, ["--dyn-syms", "--wide"], path)
    versions = run_readelf(readelf, ["--version-info", "--wide"], path)
    return {
        "path": path.name,
        "sha256": sha256_file(path),
        "identity": parse_identity(header_dynamic),
        "symbols": parse_symbols(dynsym),
        "versionInfoSha256": hashlib.sha256(versions.encode("utf-8")).hexdigest(),
    }

def public_symbols(records):
    return {
        key: value
        for key, value in records.items()
        if value["ndx"] != "UND" and value["bind"] in {"GLOBAL", "WEAK"}
    }

def compare(old, candidate, allowed_removed):
    findings = []
    for field in ("class", "machine", "soname"):
        if old["identity"][field] != candidate["identity"][field]:
            findings.append({"kind": "identity-changed", "field": field, "old": old["identity"][field], "new": candidate["identity"][field]})
    old_needed = set(old["identity"]["needed"])
    new_needed = set(candidate["identity"]["needed"])
    for name in sorted(new_needed - old_needed):
        findings.append({"kind": "needed-added", "name": name})
    for name in sorted(old_needed - new_needed):
        findings.append({"kind": "needed-removed", "name": name})

    old_public = public_symbols(old["symbols"])
    new_public = public_symbols(candidate["symbols"])
    for key in sorted(old_public.keys() - new_public.keys()):
        if key not in allowed_removed:
            findings.append({"kind": "public-symbol-removed", "symbol": key})
    for key in sorted(new_public.keys() - old_public.keys()):
        findings.append({"kind": "public-symbol-added-review", "symbol": key})
    compared_fields = ("type", "bind", "visibility", "ndx", "defaultVersion", "size")
    for key in sorted(old_public.keys() & new_public.keys()):
        changes = {
            field: {"old": old_public[key][field], "new": new_public[key][field]}
            for field in compared_fields
            if old_public[key][field] != new_public[key][field]
        }
        if changes:
            findings.append({"kind": "public-symbol-attributes-changed", "symbol": key, "changes": changes})
    if old["versionInfoSha256"] != candidate["versionInfoSha256"]:
        findings.append({"kind": "version-info-changed-review"})
    return findings

def main():
    if len(sys.argv) != 4:
        print("Usage: compare_so_abi.py OLD_SO CANDIDATE_SO CONTRACT_JSON", file=sys.stderr)
        raise SystemExit(2)
    old_path = Path(sys.argv[1])
    candidate_path = Path(sys.argv[2])
    contract_path = Path(sys.argv[3])
    if not old_path.is_file() or not candidate_path.is_file() or not contract_path.is_file():
        print("all three input files must exist", file=sys.stderr)
        raise SystemExit(2)
    readelf = shutil.which("readelf")
    if not readelf:
        fail("readelf is not available", 3)
    allowed_removed = load_contract(contract_path)
    old = inspect(readelf, old_path)
    candidate = inspect(readelf, candidate_path)
    findings = compare(old, candidate, allowed_removed)
    blocking = [item for item in findings if not item["kind"].endswith("-review")]
    result = {
        "status": "static-contract-blocked" if blocking else "static-contract-passed",
        "old": {"sha256": old["sha256"], "identity": old["identity"]},
        "candidate": {"sha256": candidate["sha256"], "identity": candidate["identity"]},
        "findings": findings,
        "boundary": "Static ELF comparison does not prove runtime ABI compatibility.",
    }
    print(json.dumps(result, ensure_ascii=False, indent=2, sort_keys=True))
    if blocking:
        raise SystemExit(20)

if __name__ == "__main__":
    main()