#!/usr/bin/env python3 """只读检查 Mapping 的逐 Style Layer 证据；绝不产生 EXECUTABLE 授权。""" from __future__ import annotations import argparse import hashlib from collections import Counter, defaultdict import json from pathlib import Path from typing import Any RESOURCE_PROPERTIES = { "icon": (("layout", "icon-image"),), "fill": (("paint", "fill-pattern"),), "line": (("paint", "line-pattern"),), } KNOWN_STATUSES = {"VISUAL_OBJECT_MATCH", "LAYER_ONLY"} def _direct_predicate(layer: dict[str, Any], component: dict[str, Any]) -> list[Any] | None: key = component.get("resourceKey") if not isinstance(key, str) or not key: return None for section, property_name in RESOURCE_PROPERTIES.get(component.get("role"), ()): expr = layer.get(section, {}).get(property_name) if isinstance(expr, list) and len(expr) == 2 and expr[0] == "get" and isinstance(expr[1], str) and expr[1]: return ["==", ["get", expr[1]], key] return None def audit_mapping(style: dict[str, Any], mapping: dict[str, Any]) -> dict[str, Any]: """保守静态诊断：DIRECT 仅是构造候选，不是执行批准或对象身份确认。""" layers = style.get("layers") assignments = mapping.get("assignments") if not isinstance(layers, list) or not isinstance(assignments, list): raise ValueError("Style/Mapping 结构无效") records: list[dict[str, Any]] = [] counts: Counter[str] = Counter() layers_to_classes: dict[str, set[str]] = defaultdict(set) seen: set[str] = set() for item in sorted(assignments, key=lambda row: str(row.get("visualObjectId", ""))): object_id = item.get("visualObjectId") if not isinstance(object_id, str) or not object_id or object_id in seen: raise ValueError("Visual Object ID 为空或重复") seen.add(object_id) status = item.get("status") standard = item.get("standardLayer") evidence = item.get("renderEvidence") reasons: set[str] = set() components_out: list[dict[str, Any]] = [] direct_fields: dict[str, set[str]] = defaultdict(set) sources: set[str] = set() has_unkeyed = False if status not in KNOWN_STATUSES or not isinstance(standard, str) or not standard: reasons.add("UNRESOLVED_CLASSIFICATION") if not isinstance(evidence, dict) or not isinstance(evidence.get("components"), list) or not evidence["components"]: reasons.add("MISSING_RENDER_EVIDENCE") else: for component in evidence["components"]: index = component.get("styleIndex") if type(index) is not int or index < 0 or index >= len(layers): raise ValueError(f"{object_id}: Style 索引无效") layer = layers[index] layer_id = component.get("styleLayerId") if not isinstance(layer, dict) or layer.get("id") != layer_id: raise ValueError(f"{object_id}: Style Layer ID 不一致") source = layer.get("source") source_layer = layer.get("source-layer") if not isinstance(source, str) or not source or source_layer != evidence.get("sourceLayer"): reasons.add("SOURCE_OR_SOURCE_LAYER_MISMATCH") else: sources.add(source) if isinstance(layer_id, str): layers_to_classes[layer_id].add(standard if status in KNOWN_STATUSES and standard else "RESIDUAL") pred = _direct_predicate(layer, component) if pred is None: has_unkeyed = True reasons.add("COMPONENT_SELECTOR_NOT_PROVED") else: direct_fields[pred[1][1]].add(pred[2]) components_out.append({ "styleLayerId": layer_id, "styleIndex": index, "drawRole": component.get("role"), "selectorEvidence": "DIRECT_GET_RESOURCE_KEY_CANDIDATE" if pred is not None else "UNPROVED", "candidateSelector": pred, }) if len(sources) > 1: reasons.add("CROSS_SOURCE_COMPONENTS") if any(len(keys) > 1 for keys in direct_fields.values()): reasons.add("CONTRADICTORY_RESOURCE_FIELDS") if len(components_out) > 1: reasons.add("MULTI_COMPONENT_IDENTITY_UNVERIFIED") elif components_out: reasons.add("PRIMARY_IDENTITY_UNDECLARED") if status in KNOWN_STATUSES and standard: # Mapping v2 has draw roles, not reviewed PRIMARY/COMPONENT roles. reasons.add("STANDARD_TO_DISPLAY_GROUP_CONTRACT_UNBOUND") if has_unkeyed: reasons.add("INHERITANCE_REQUIRES_SAME_FEATURE_PROOF") if components_out: reasons.add("ORIGINAL_FILTER_REACHABILITY_UNVERIFIED") if not reasons: reasons.add("EXECUTION_AUTHORITY_UNDEFINED") for reason in reasons: counts[reason] += 1 records.append({ "visualObjectId": object_id, "assignmentStatus": status, "standardLayer": standard, "renderComponents": components_out, "blockingReasons": sorted(reasons), "executionAuthorized": False, }) mixed_layers = sorted(layer_id for layer_id, categories in layers_to_classes.items() if len(categories) > 1) return { "schemaVersion": "navsea-mapping-execution-evidence-audit/v1", "scope": "STATIC_DIAGNOSTIC_ONLY", "mappingObjectCount": len(records), "executionAuthorizedCount": 0, "blockingReasonCounts": dict(sorted(counts.items())), "mixedStyleLayerCount": len(mixed_layers), "mixedStyleLayerIds": mixed_layers, "records": records, } def verify_mapping_style_fingerprint(mapping: dict[str, Any], style_bytes: bytes) -> None: """Fail closed if the declared original Style digest differs from this input.""" source = mapping.get("renderEvidenceSource") if source is None: return # Old mapping versions have no evidence source, and get no authority. expected = source.get("styleSha256") if isinstance(source, dict) else None if not isinstance(expected, str) or len(expected) != 64: raise ValueError("声明的 Style SHA-256 缺失或格式无效") actual = hashlib.sha256(style_bytes).hexdigest() if actual.lower() != expected.lower(): raise ValueError("Mapping 声明的 Style SHA-256 与实际快照不匹配") def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--style", type=Path, required=True) parser.add_argument("--mapping", type=Path, required=True) parser.add_argument("--output", type=Path, help="可选：写入审计 JSON，禁止用于运行时授权") args = parser.parse_args() style_bytes = args.style.read_bytes() mapping = json.loads(args.mapping.read_text(encoding="utf-8")) verify_mapping_style_fingerprint(mapping, style_bytes) result = audit_mapping(json.loads(style_bytes.decode("utf-8")), mapping) data = json.dumps(result, ensure_ascii=False, indent=2) + "\n" if args.output is None: print(data) else: args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(data, encoding="utf-8") print(json.dumps({ "scope": result["scope"], "mappingObjectCount": result["mappingObjectCount"], "executionAuthorizedCount": result["executionAuthorizedCount"], "mixedStyleLayerCount": result["mixedStyleLayerCount"], "output": str(args.output), }, ensure_ascii=False)) return 0 if __name__ == "__main__": raise SystemExit(main())