App 加固后进程重建与状态恢复怎样回归

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

SHA256 = re.compile(r"^[a-f0-9]{64}$")
REQUIRED_EVENTS = ["snapshot-before", "background", "process-death", "relaunch", "snapshot-after"]

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

def load_record(path):
    try:
        value = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        fail("cannot read recovery record: " + str(exc), 3)
    if not isinstance(value, dict):
        fail("recovery record root must be an object", 4)
    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, 5)
    return value.strip()

def require_list(record, field, allow_empty=False):
    value = record.get(field)
    if not isinstance(value, list):
        fail("invalid list field: " + field, 6)
    items = []
    for item in value:
        if not isinstance(item, str) or not item.strip():
            fail("invalid item in: " + field, 7)
        items.append(item.strip())
    if not allow_empty and not items:
        fail("empty required list: " + field, 8)
    if len(items) != len(set(items)):
        fail("duplicate item in: " + field, 9)
    return items

def require_object(record, field):
    value = record.get(field)
    if not isinstance(value, dict):
        fail("invalid object field: " + field, 10)
    return value

def validate_identity(document):
    candidate = require_object(document, "candidate")
    artifact = require_text(candidate, "artifactSha256")
    if not SHA256.fullmatch(artifact):
        fail("invalid artifactSha256", 11)
    version = candidate.get("versionCode")
    if not isinstance(version, int) or isinstance(version, bool):
        fail("invalid versionCode", 12)
    run_id = require_text(document, "testRunId")
    return artifact, version, run_id

def validate_event_order(document):
    scenario = require_object(document, "scenario")
    events = scenario.get("events")
    if not isinstance(events, list):
        fail("scenario events must be a list", 13)
    names = []
    timestamps = []
    for event in events:
        if not isinstance(event, dict):
            fail("scenario event must be an object", 14)
        names.append(require_text(event, "name"))
        timestamp = event.get("timestampMs")
        if not isinstance(timestamp, int) or isinstance(timestamp, bool):
            fail("event timestampMs is invalid", 15)
        timestamps.append(timestamp)
    positions = []
    for required in REQUIRED_EVENTS:
        if required not in names:
            fail("missing scenario event: " + required, 16)
        positions.append(names.index(required))
    if positions != sorted(positions) or timestamps != sorted(timestamps):
        fail("scenario events are out of order", 17)
    return names

def compare_contract(document):
    before = require_object(document, "before")
    after = require_object(document, "after")
    contract = require_object(document, "contract")
    before_state = require_object(before, "state")
    after_state = require_object(after, "state")
    durable = require_list(contract, "durableEqual")
    transient = require_object(contract, "transientReset")
    sensitive = require_list(contract, "sensitiveAbsent", True)
    allowed_navigation = set(require_list(contract, "allowedNavigation"))
    operation_field = require_text(contract, "operationIdField")
    findings = []
    for field in durable:
        if field not in before_state or field not in after_state:
            findings.append({"field": field, "problem": "missing-durable-field"})
        elif before_state[field] != after_state[field]:
            findings.append({"field": field, "problem": "durable-value-changed"})
    for field, expected in transient.items():
        if after_state.get(field) != expected:
            findings.append({"field": field, "problem": "transient-not-reset"})
    for field in sensitive:
        if field in after_state:
            findings.append({"field": field, "problem": "sensitive-state-restored"})
    before_operation = before_state.get(operation_field)
    after_operation = after_state.get(operation_field)
    if not isinstance(before_operation, str) or before_operation != after_operation:
        findings.append({"field": operation_field, "problem": "operation-identity-changed"})
    before_process = require_text(before, "processIdentity")
    after_process = require_text(after, "processIdentity")
    if before_process == after_process:
        findings.append({"field": "processIdentity", "problem": "process-death-not-demonstrated"})
    navigation = require_text(after, "navigationEntry")
    if navigation not in allowed_navigation:
        findings.append({"field": "navigationEntry", "problem": "unexpected-destination"})
    return findings

def main():
    if len(sys.argv) != 2:
        print("Usage: recovery_gate.py RECOVERY_RECORD_JSON", file=sys.stderr)
        raise SystemExit(2)
    record_path = Path(sys.argv[1])
    if not record_path.is_file():
        print("recovery record is missing", file=sys.stderr)
        raise SystemExit(2)
    document = load_record(record_path)
    artifact, version, run_id = validate_identity(document)
    events = validate_event_order(document)
    findings = compare_contract(document)
    result = {
        "status": "blocked" if findings else "contract-passed",
        "artifactSha256": artifact,
        "versionCode": version,
        "testRunId": run_id,
        "events": events,
        "findings": findings,
    }
    print(json.dumps(result, ensure_ascii=False, indent=2, sort_keys=True))
    if findings:
        raise SystemExit(20)

if __name__ == "__main__":
    main()