VMP 为什么不能替代服务端授权：客户端保护与最终裁决边界

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

REQUIRED_GRANT_EVIDENCE = {"authenticated-user", "server-entitlement", "freshness"}
ALLOWED_OWNERS = {"client", "server", "joint"}
ALLOWED_SOURCES = {"client", "server", "trusted-service"}

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

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

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

def index_states(document):
    states = document.get("states")
    if not isinstance(states, list) or not states:
        fail("states are missing", 9)
    index = {}
    for state in states:
        if not isinstance(state, dict):
            fail("state must be an object", 10)
        state_id = state.get("id")
        owner = state.get("decisionOwner")
        if not isinstance(state_id, str) or not state_id.strip():
            fail("state id is missing", 11)
        if owner not in ALLOWED_OWNERS:
            fail("invalid decision owner: " + str(state_id), 12)
        if state_id in index:
            fail("duplicate state: " + state_id, 13)
        index[state_id] = {
            "decisionOwner": owner,
            "grants": strings(state, "grants", True),
            "requires": strings(state, "requires", True),
            "terminal": bool(state.get("terminal", False)),
        }
    return index

def index_transitions(document, states):
    transitions = document.get("transitions")
    if not isinstance(transitions, list) or not transitions:
        fail("transitions are missing", 14)
    graph = defaultdict(list)
    incoming_sources = defaultdict(set)
    for transition in transitions:
        if not isinstance(transition, dict):
            fail("transition must be an object", 15)
        source = transition.get("from")
        target = transition.get("to")
        event = transition.get("event")
        evidence_source = transition.get("evidenceSource")
        if source not in states or target not in states:
            fail("transition references unknown state", 16)
        if not isinstance(event, str) or not event.strip():
            fail("transition event is missing", 17)
        if evidence_source not in ALLOWED_SOURCES:
            fail("invalid evidenceSource", 18)
        graph[source].append(target)
        incoming_sources[target].add(evidence_source)
    return graph, incoming_sources

def reachable(initial, graph):
    seen = {initial}
    queue = deque([initial])
    while queue:
        current = queue.popleft()
        for target in graph.get(current, []):
            if target not in seen:
                seen.add(target)
                queue.append(target)
    return seen

def audit(states, incoming_sources, reachable_states):
    findings = []
    for state_id in sorted(reachable_states):
        state = states[state_id]
        if not state["grants"]:
            continue
        missing = sorted(REQUIRED_GRANT_EVIDENCE - set(state["requires"]))
        reasons = []
        if state["decisionOwner"] == "client":
            reasons.append("client-only-decision")
        if missing:
            reasons.append("missing-evidence:" + ",".join(missing))
        sources = incoming_sources.get(state_id, set())
        if "server" not in sources and "trusted-service" not in sources:
            reasons.append("no-server-derived-transition")
        if reasons:
            findings.append({
                "state": state_id,
                "grants": state["grants"],
                "reasons": reasons,
            })
    return findings

def main():
    if len(sys.argv) != 2:
        print("Usage: auth_boundary_audit.py STATE_MACHINE_JSON", file=sys.stderr)
        raise SystemExit(2)
    source_path = Path(sys.argv[1])
    if not source_path.is_file():
        print("state machine file is missing", file=sys.stderr)
        raise SystemExit(2)
    document = load_state_machine(source_path)
    states = index_states(document)
    initial = document.get("initialState")
    if initial not in states:
        fail("initialState is invalid", 19)
    graph, incoming_sources = index_transitions(document, states)
    reachable_states = reachable(initial, graph)
    unreachable = sorted(set(states) - reachable_states)
    findings = audit(states, incoming_sources, reachable_states)
    result = {
        "status": "blocked" if findings else "manual-review-required",
        "clientAuthorizationFindings": findings,
        "unreachableStates": unreachable,
        "checkedStates": len(reachable_states),
    }
    print(json.dumps(result, ensure_ascii=False, indent=2, sort_keys=True))
    if findings:
        raise SystemExit(20)

if __name__ == "__main__":
    main()