如何从业务调用边界建立 VMP 核心函数清单

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

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

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

def digest(path):
    return hashlib.sha256(path.read_bytes()).hexdigest()

def require_string(record, key):
    value = record.get(key)
    if not isinstance(value, str) or not value.strip():
        fail("missing string field: " + key, 5)
    return value.strip()

def string_set(value, field):
    if not isinstance(value, list) or not all(isinstance(x, str) and x.strip() for x in value):
        fail("invalid string list: " + field, 6)
    return {x.strip() for x in value}

def index_mapping(document):
    methods = document.get("methods")
    if not isinstance(methods, list) or not methods:
        fail("mapping methods are missing", 7)
    index = {}
    for item in methods:
        if not isinstance(item, dict):
            fail("mapping entry must be an object", 8)
        original = require_string(item, "original")
        if original in index:
            fail("duplicate mapping method: " + original, 9)
        callers = string_set(item.get("callers", []), "callers")
        side_effects = string_set(item.get("sideEffects", []), "sideEffects")
        index[original] = {
            "obfuscated": require_string(item, "obfuscated"),
            "callers": sorted(callers),
            "sideEffects": sorted(side_effects),
            "frequency": require_string(item, "frequency"),
        }
    return index

def build_inventory(mapping, assets):
    capabilities = assets.get("capabilities")
    if not isinstance(capabilities, list) or not capabilities:
        fail("capabilities are missing", 10)
    records = []
    for capability in capabilities:
        if not isinstance(capability, dict):
            fail("capability must be an object", 11)
        name = require_string(capability, "name")
        entries = string_set(capability.get("entrypoints"), "entrypoints")
        decisions = string_set(capability.get("decisions"), "decisions")
        operations = string_set(capability.get("sensitiveOperations"), "sensitiveOperations")
        exclusions = string_set(capability.get("exclusions", []), "exclusions")
        server_check = capability.get("serverCheck")
        if server_check not in {"required", "offline-limited", "not-applicable"}:
            fail("invalid serverCheck for " + name, 12)
        for method in sorted(decisions | operations | exclusions):
            mapped = mapping.get(method)
            reasons = []
            excluded = method in exclusions
            if mapped is None:
                reasons.append("missing-build-mapping")
            else:
                if not mapped["callers"]:
                    reasons.append("no-recorded-caller")
                if mapped["sideEffects"]:
                    reasons.append("side-effects:" + ",".join(mapped["sideEffects"]))
            if excluded:
                reasons.append("manually-excluded")
            role = "decision" if method in decisions else "sensitive-operation"
            candidate = mapped is not None and not excluded and not reasons
            records.append({
                "capability": name,
                "method": method,
                "compiledMethod": mapped["obfuscated"] if mapped else None,
                "role": role,
                "entrypoints": sorted(entries),
                "callers": mapped["callers"] if mapped else [],
                "serverCheck": server_check,
                "candidate": candidate,
                "reasons": reasons or ["boundary-complete-review-required"],
            })
    return records

def main():
    if len(sys.argv) != 3:
        print("Usage: inventory.py MAPPING_JSON ASSET_JSON", file=sys.stderr)
        raise SystemExit(2)
    mapping_path = Path(sys.argv[1])
    asset_path = Path(sys.argv[2])
    if not mapping_path.is_file() or not asset_path.is_file():
        print("required input file missing", file=sys.stderr)
        raise SystemExit(2)
    mapping_doc = load_object(mapping_path)
    asset_doc = load_object(asset_path)
    inventory = build_inventory(index_mapping(mapping_doc), asset_doc)
    result = {
        "status": "manual-review-required",
        "mappingSha256": digest(mapping_path),
        "assetSha256": digest(asset_path),
        "records": inventory,
    }
    print(json.dumps(result, ensure_ascii=False, indent=2, sort_keys=True))

if __name__ == "__main__":
    main()