如何用 ApplicationExitInfo 复核加固版本异常退出

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

SHA256 = re.compile(r"^[a-f0-9]{64}$")
ALLOWED_REASONS = {
    "crash", "native-signal", "anr", "low-memory",
    "resource-pressure", "user-requested", "system-action", "other"
}
ALLOWED_TRACES = {"none", "java-stack", "anr-trace", "native-tombstone"}

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

def load_json(path):
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        fail("cannot read plan: " + str(exc), 3)
    if not isinstance(value, dict):
        fail("plan root must be an object", 4)
    return value

def require_sha(value, field):
    if not isinstance(value, str) or not SHA256.fullmatch(value):
        fail("invalid SHA-256 field: " + field, 5)
    return value

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

def require_int(record, field):
    value = record.get(field)
    if not isinstance(value, int) or isinstance(value, bool):
        fail("invalid integer field: " + field, 7)
    return value

def load_records(path):
    records = []
    try:
        lines = path.read_text(encoding="utf-8").splitlines()
    except OSError as exc:
        fail("cannot read exit records: " + str(exc), 8)
    for line_number, line in enumerate(lines, 1):
        if not line.strip():
            continue
        try:
            record = json.loads(line)
        except json.JSONDecodeError as exc:
            fail("invalid JSON line " + str(line_number) + ": " + str(exc), 9)
        if not isinstance(record, dict):
            fail("record must be an object", 10)
        reason = require_text(record, "reason")
        trace_kind = require_text(record, "traceKind")
        if reason not in ALLOWED_REASONS:
            fail("unsupported reason: " + reason, 11)
        if trace_kind not in ALLOWED_TRACES:
            fail("unsupported traceKind: " + trace_kind, 12)
        mapping = record.get("mappingSha256")
        if mapping is not None:
            require_sha(mapping, "mappingSha256")
        records.append({
            "artifactSha256": require_sha(record.get("artifactSha256"), "artifactSha256"),
            "versionCode": require_int(record, "versionCode"),
            "processName": require_text(record, "processName"),
            "reason": reason,
            "timestampMs": require_int(record, "timestampMs"),
            "testRunId": require_text(record, "testRunId"),
            "pathName": require_text(record, "pathName"),
            "traceKind": trace_kind,
            "mappingSha256": mapping,
        })
    return records

def validate_plan(plan):
    artifact = require_sha(plan.get("artifactSha256"), "artifactSha256")
    version = require_int(plan, "versionCode")
    start = require_int(plan, "windowStartMs")
    end = require_int(plan, "windowEndMs")
    if start >= end:
        fail("invalid observation window", 13)
    processes = plan.get("processNames")
    runs = plan.get("requiredRunIds")
    if not isinstance(processes, list) or not processes:
        fail("processNames are missing", 14)
    if not isinstance(runs, list) or not runs:
        fail("requiredRunIds are missing", 15)
    if not all(isinstance(x, str) and x.strip() for x in processes + runs):
        fail("plan contains an invalid identifier", 16)
    return artifact, version, start, end, set(processes), set(runs)

def build_report(records, plan):
    artifact, version, start, end, processes, required_runs = validate_plan(plan)
    grouped = defaultdict(lambda: {"count": 0, "traceKinds": set(), "runIds": set()})
    observed_runs = set()
    excluded = 0
    for record in records:
        matches = (
            record["artifactSha256"] == artifact
            and record["versionCode"] == version
            and record["processName"] in processes
            and start <= record["timestampMs"] <= end
        )
        if not matches:
            excluded += 1
            continue
        observed_runs.add(record["testRunId"])
        key = (record["processName"], record["pathName"], record["reason"])
        grouped[key]["count"] += 1
        grouped[key]["traceKinds"].add(record["traceKind"])
        grouped[key]["runIds"].add(record["testRunId"])
    rows = []
    for key in sorted(grouped):
        process, path_name, reason = key
        item = grouped[key]
        rows.append({
            "processName": process,
            "pathName": path_name,
            "reason": reason,
            "count": item["count"],
            "traceKinds": sorted(item["traceKinds"]),
            "runIds": sorted(item["runIds"]),
            "gate": "investigate" if reason not in {"user-requested", "system-action"} else "review-expected-exit",
        })
    missing_runs = sorted(required_runs - observed_runs)
    status = "blocked" if rows or missing_runs else "no-exit-observed"
    return {
        "status": status,
        "artifactSha256": artifact,
        "versionCode": version,
        "rows": rows,
        "missingRunIds": missing_runs,
        "excludedRecordCount": excluded,
    }

def main():
    if len(sys.argv) != 3:
        print("Usage: exit_gate.py EXIT_RECORDS_JSONL TEST_PLAN_JSON", file=sys.stderr)
        raise SystemExit(2)
    records_path = Path(sys.argv[1])
    plan_path = Path(sys.argv[2])
    if not records_path.is_file() or not plan_path.is_file():
        print("required input file missing", file=sys.stderr)
        raise SystemExit(2)
    report = build_report(load_records(records_path), load_json(plan_path))
    print(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True))
    if report["status"] == "blocked":
        raise SystemExit(20)

if __name__ == "__main__":
    main()