SO 加固前怎样检查 JNI 局部与全局引用生命周期

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

ALLOWED_ABIS = {"armeabi-v7a", "arm64-v8a", "x86", "x86_64"}
ALLOWED_OPS = {
    "frame-push", "frame-pop", "local-create", "local-delete",
    "local-retain", "global-create", "global-delete",
    "weak-create", "weak-delete", "weak-promote",
    "weak-use", "temporary-release", "exception-pending",
    "exception-cleared", "owner-end", "scope-end",
}

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 load_document(path):
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, UnicodeError, json.JSONDecodeError) as exc:
        fail("cannot read event document: " + str(exc))
    if not isinstance(value, dict):
        fail("event document must be an object")
    return value

def require_text(event, field):
    value = event.get(field)
    if not isinstance(value, str) or not value.strip():
        fail("event field is missing: " + field)
    return value.strip()

def add_finding(findings, event, kind, detail):
    findings.append({
        "seq": event.get("seq"),
        "kind": kind,
        "detail": detail,
    })

def audit(document):
    artifact = document.get("artifactSha256")
    abi = document.get("abi")
    events = document.get("events")
    if not isinstance(artifact, str) or len(artifact) != 64:
        fail("artifactSha256 must contain 64 hexadecimal characters")
    try:
        int(artifact, 16)
    except ValueError:
        fail("artifactSha256 is not hexadecimal")
    if abi not in ALLOWED_ABIS:
        fail("unsupported or missing ABI")
    if not isinstance(events, list) or not events:
        fail("events must be a non-empty array")

    findings = []
    last_seq = 0
    frames = defaultdict(int)
    locals_by_scope = defaultdict(set)
    retained_locals = set()
    globals_live = {}
    weak_live = {}
    promoted = defaultdict(set)
    pending_exception = set()
    ended_owners = set()

    for event in events:
        if not isinstance(event, dict):
            fail("every event must be an object")
        seq = event.get("seq")
        if not isinstance(seq, int) or seq <= last_seq:
            fail("event seq must be a strictly increasing integer")
        last_seq = seq
        op = require_text(event, "op")
        if op not in ALLOWED_OPS:
            fail("unsupported event operation: " + op)
        thread = require_text(event, "thread")
        scope = require_text(event, "scope")
        scope_key = (thread, scope)
        ref_id = str(event.get("ref", "")).strip()
        owner = str(event.get("owner", "")).strip()

        if op == "frame-push":
            frames[scope_key] += 1
        elif op == "frame-pop":
            if frames[scope_key] == 0:
                add_finding(findings, event, "frame-underflow", str(scope_key))
            else:
                frames[scope_key] -= 1
                locals_by_scope[scope_key].clear()
        elif op == "local-create":
            if not ref_id:
                fail("local-create requires ref")
            if ref_id in locals_by_scope[scope_key]:
                add_finding(findings, event, "duplicate-local", ref_id)
            locals_by_scope[scope_key].add(ref_id)
        elif op == "local-delete":
            if ref_id not in locals_by_scope[scope_key]:
                add_finding(findings, event, "unknown-local-delete", ref_id)
            else:
                locals_by_scope[scope_key].remove(ref_id)
                retained_locals.discard((scope_key, ref_id))
        elif op == "local-retain":
            if ref_id not in locals_by_scope[scope_key]:
                add_finding(findings, event, "retain-unknown-local", ref_id)
            else:
                retained_locals.add((scope_key, ref_id))
        elif op == "global-create":
            if not ref_id or not owner:
                fail("global-create requires ref and owner")
            if ref_id in globals_live:
                add_finding(findings, event, "duplicate-global", ref_id)
            if owner in ended_owners:
                add_finding(findings, event, "create-after-owner-end", owner)
            globals_live[ref_id] = {"owner": owner, "seq": seq}
        elif op == "global-delete":
            if ref_id not in globals_live:
                add_finding(findings, event, "unknown-global-delete", ref_id)
            else:
                del globals_live[ref_id]
        elif op == "weak-create":
            if not ref_id or not owner:
                fail("weak-create requires ref and owner")
            if ref_id in weak_live:
                add_finding(findings, event, "duplicate-weak", ref_id)
            weak_live[ref_id] = {"owner": owner, "seq": seq}
        elif op == "weak-promote":
            if ref_id not in weak_live:
                add_finding(findings, event, "promote-unknown-weak", ref_id)
            else:
                promoted[scope_key].add(ref_id)
        elif op == "weak-use":
            if ref_id not in promoted[scope_key]:
                add_finding(findings, event, "weak-use-without-promotion", ref_id)
        elif op == "temporary-release":
            if ref_id not in promoted[scope_key]:
                add_finding(findings, event, "unknown-temporary-release", ref_id)
            else:
                promoted[scope_key].remove(ref_id)
        elif op == "weak-delete":
            if ref_id not in weak_live:
                add_finding(findings, event, "unknown-weak-delete", ref_id)
            else:
                del weak_live[ref_id]
        elif op == "exception-pending":
            pending_exception.add(scope_key)
        elif op == "exception-cleared":
            if scope_key not in pending_exception:
                add_finding(findings, event, "clear-without-pending", str(scope_key))
            pending_exception.discard(scope_key)
        elif op == "owner-end":
            if not owner:
                fail("owner-end requires owner")
            ended_owners.add(owner)
            for live_ref, record in sorted(globals_live.items()):
                if record["owner"] == owner:
                    add_finding(findings, event, "global-after-owner-end", live_ref)
            for live_ref, record in sorted(weak_live.items()):
                if record["owner"] == owner:
                    add_finding(findings, event, "weak-after-owner-end", live_ref)
        elif op == "scope-end":
            for live_ref in sorted(locals_by_scope[scope_key]):
                add_finding(findings, event, "local-at-scope-end", live_ref)
            if frames[scope_key] != 0:
                add_finding(findings, event, "unbalanced-local-frame", str(frames[scope_key]))
            if promoted[scope_key]:
                add_finding(findings, event, "promoted-weak-at-scope-end", ",".join(sorted(promoted[scope_key])))
            if scope_key in pending_exception:
                add_finding(findings, event, "pending-exception-at-scope-end", str(scope_key))
            for retained_scope, live_ref in sorted(retained_locals):
                if retained_scope == scope_key:
                    add_finding(findings, event, "retained-local-crosses-scope", live_ref)

    for live_ref, record in sorted(globals_live.items()):
        findings.append({"seq": None, "kind": "global-not-released", "detail": live_ref + " owner=" + record["owner"]})
    for live_ref, record in sorted(weak_live.items()):
        findings.append({"seq": None, "kind": "weak-not-released", "detail": live_ref + " owner=" + record["owner"]})

    return {
        "status": "blocked" if findings else "runtime-scope-passed",
        "artifactSha256": artifact.lower(),
        "abi": abi,
        "eventCount": len(events),
        "findings": findings,
        "boundary": "This audit covers only executed and recorded paths.",
    }

def main():
    if len(sys.argv) != 2:
        print("Usage: audit_jni_reference_events.py EVENTS_JSON", file=sys.stderr)
        raise SystemExit(2)
    input_path = Path(sys.argv[1])
    if not input_path.is_file():
        print("event document does not exist", file=sys.stderr)
        raise SystemExit(2)
    document = load_document(input_path)
    result = audit(document)
    result["inputSha256"] = sha256_file(input_path)
    print(json.dumps(result, ensure_ascii=False, indent=2, sort_keys=True))
    if result["status"] == "blocked":
        raise SystemExit(20)

if __name__ == "__main__":
    main()