diff --git a/tests/test_alpha4_network_release_architecture.py b/tests/test_alpha4_network_release_architecture.py index 5740ffd..62721c4 100644 --- a/tests/test_alpha4_network_release_architecture.py +++ b/tests/test_alpha4_network_release_architecture.py @@ -202,3 +202,51 @@ def test_post_build_verifier_is_isolated_and_bound_to_exact_seed_relation( assert tree_digest(release) == before assert not (release / ".tlacache").exists() assert not (seed / ".tlacache").exists() + + +def test_network_airgap_executes_generated_companion_under_restricted_runtime( + tmp_path: Path, monkeypatch +) -> None: + from tools import alpha4_network_expression_airgap as airgap + + profiles = tmp_path / "profiles" + base = profiles / "base/seed/python/aset_seed_alpha4.py" + target = profiles / "python/aset_network_alpha4.py" + base.parent.mkdir(parents=True) + target.parent.mkdir(parents=True) + base.write_text( + "def state(subject, authority, recognition='UNKNOWN', evidence=()):\n" + " return {'subject': subject, 'authority': authority, " + "'recognition': recognition, 'evidence': tuple(evidence)}\n" + "def apply_component(current, component_id, *, evidence=None, " + "authority_recognition=frozenset()):\n" + " if component_id != 'ASET-COMPONENT-OBSERVE-UNKNOWN': " + "raise ValueError('component')\n" + " if current['recognition'] != 'UNKNOWN': raise ValueError('recognition')\n" + " result = dict(current); result['evidence'] = " + "tuple(sorted(set(current['evidence']) | {evidence})); return result\n", + encoding="utf-8", + ) + digest = "sha256:" + hashlib.sha256(base.read_bytes()).hexdigest() + write_python(target, digest, _manifest_records()) + binding = SeedBinding( + release_tag="seed-test", + release_tree="sha256:" + "0" * 64, + release_archive="sha256:" + "0" * 64, + profile_tree="sha256:" + "0" * 64, + profile_archive="sha256:" + "0" * 64, + sources={}, + assurance_bases={}, + companions={"PYTHON": ("base/seed/python/aset_seed_alpha4.py", digest)}, + ) + monkeypatch.setattr(airgap, "parse_seed_binding", lambda: binding) + evidence = airgap.check_airgap(profiles) + assert evidence["coverage"]["total_cases"] == 446 + assert evidence["coverage"]["sensitivity_cases"] == 26 + assert evidence["coverage"]["grand_total_cases"] == 472 + assert evidence["assurance_dependencies"]["companion_import_surface"] == "RESTRICTED" + assert ( + evidence["assurance_dependencies"]["companion_file_access"] + == "MATERIALIZED_PROFILE_TREE_READ_ONLY" + ) + assert evidence["status"] == "PASS" diff --git a/tests/test_alpha4_network_three_way_assurance.py b/tests/test_alpha4_network_three_way_assurance.py index 6b3cea0..238317d 100644 --- a/tests/test_alpha4_network_three_way_assurance.py +++ b/tests/test_alpha4_network_three_way_assurance.py @@ -134,7 +134,7 @@ def test_relational_source_derivation_and_sensitivity_are_first_class() -> None: evidence = check_triangulated_assurance() assert evidence["relational_source_derivations"] == 21 assert evidence["federation_identity_guard_derivations"] == 12 - assert evidence["interface_validator_cases"] == 4 + assert evidence["interface_validator_cases"] == 12 assert evidence["core_field_sensitivity"] == 5 assert evidence["dynamic_binding_sensitivity"] == 6 assert evidence["composition_identity_sensitivity"] == 16 @@ -186,7 +186,7 @@ def test_bound_federation_tla_guard_mutation_breaks_gate(tmp_path: Path) -> None path.write_text(text.replace(old, 'fs.members[context] = "ACTIVE"', 1), encoding="utf-8") status, output = _run_gate(repo) assert status != 0 - assert "three-way" in output.lower() or "mismatch" in output.lower() + assert "relational canonical scope drift" in output.lower() def test_manifest_duplicate_precedence_breaks_gate(tmp_path: Path) -> None: @@ -250,7 +250,7 @@ def test_federation_artifact_identity_guard_mutation_breaks_gate(tmp_path: Path) path.write_text(text.replace(old, "/\\ TRUE", 1), encoding="utf-8") status, output = _run_gate(repo) assert status != 0 - assert "artifact domain guard missing" in output or "identity/domain guard" in output + assert "relational canonical scope drift" in output.lower() def test_network_tlaps_runner_rejects_reduced_proof_scope(tmp_path: Path) -> None: @@ -289,3 +289,136 @@ def test_network_profile_tlaps_runner_rejects_reduced_proof_scope(tmp_path: Path ) assert result.returncode != 0 assert "SCOPE_DRIFT" in result.stdout + + +def test_relational_and_proof_tla_scopes_are_closed_world(tmp_path: Path) -> None: + from tools.alpha4_network_manifest import ManifestError, parse_network_manifests + + repo = _copy_repo(tmp_path / "relational") + relational = repo / "network/alpha4/formal/NetworkRelations.tla" + text = relational.read_text(encoding="utf-8") + marker = "/\\ o \\in ObservationUniverse" + assert marker in text + relational.write_text(text.replace(marker, marker + "\n /\\ FALSE", 1), encoding="utf-8") + with pytest.raises(ManifestError, match="relational canonical scope drift"): + parse_network_manifests(repo) + + repo = _copy_repo(tmp_path / "proof") + proof = repo / "network/alpha4/formal/OperationalRelationalPairingProofs.tla" + text = proof.read_text(encoding="utf-8") + theorem = "THEOREM AdmitFreshPairing ==" + assert theorem in text + proof.write_text(text.replace(theorem, theorem + "\n /\\ TRUE", 1), encoding="utf-8") + with pytest.raises(ManifestError, match="proof canonical scope drift"): + parse_network_manifests(repo) + + +def test_network_airgap_rejects_repository_semantic_import() -> None: + from tools.alpha4_network_expression_airgap import ( + NetworkExpressionAirgapError, + _validate_companion_ast, + ) + + source = "from tools.alpha4_network_relational_expression import derive_core_contract\n" + with pytest.raises(NetworkExpressionAirgapError, match="import forbidden"): + _validate_companion_ast( + source, + allowed_imports=frozenset({"hashlib", "pathlib", "typing"}), + allow_seed_loader=True, + ) + + +def test_network_airgap_rejects_import_smuggling_and_object_traversal() -> None: + from tools.alpha4_network_expression_airgap import ( + NetworkExpressionAirgapError, + _validate_companion_ast, + ) + + with pytest.raises(NetworkExpressionAirgapError, match="import forbidden"): + _validate_companion_ast( + "from pathlib import os\n", + allowed_imports=frozenset({"hashlib", "pathlib", "typing"}), + allow_seed_loader=True, + ) + + with pytest.raises(NetworkExpressionAirgapError, match="private attribute forbidden"): + _validate_companion_ast( + "value = ().__class__\n", + allowed_imports=frozenset({"hashlib", "pathlib", "typing"}), + allow_seed_loader=True, + ) + + +def test_network_release_sensitivity_rejects_known_bug_classes() -> None: + from tools.alpha4_network_expression_airgap import ( + NetworkExpressionAirgapError, + _check_composition_identity_sensitivity, + _check_core_identity_sensitivity, + ) + + def state(subject: str, authority: str) -> dict[str, object]: + return { + "subject": subject, + "authority": authority, + "recognition": "UNKNOWN", + "evidence": (), + } + + def first_only_admit(imports, observation, seed_state): + same_id = [item for item in imports[:1] if item["import_id"] == observation["import_id"]] + if observation in same_id: + code = "IDEMPOTENT_REPLAY" + next_imports = list(imports) + elif same_id: + code = "IDENTIFIER_CONFLICT" + next_imports = list(imports) + else: + code = "IMPORT_ADMITTED" + next_imports = [*imports, dict(observation)] + return ( + next_imports, + dict(seed_state), + { + "accepted": code != "IDENTIFIER_CONFLICT", + "code": code, + "state_changed": code == "IMPORT_ADMITTED", + }, + ) + + with pytest.raises(NetworkExpressionAirgapError, match="second-position replay"): + _check_core_identity_sensitivity({"admit_import": first_only_admit}, {"state": state}) + + buggy = { + "delivery_witness": (lambda exported, delivered, export: bool(exported) and bool(delivered)) + } + with pytest.raises(NetworkExpressionAirgapError, match="foreign-export identity"): + _check_composition_identity_sensitivity(buggy) + + +def test_network_formal_reflection_scope_is_closed_world(tmp_path: Path) -> None: + from tools.alpha4_network_manifest import ManifestError, parse_network_manifests + + repo = _copy_repo(tmp_path) + reflection = repo / "network/alpha4/formal/RestrictedOperationalSemantics.tla" + text = reflection.read_text(encoding="utf-8") + marker = "OperationalAdmitFresh(s, t, o, result) ==" + assert marker in text + reflection.write_text(text.replace(marker, marker + "\n /\\ TRUE", 1), encoding="utf-8") + with pytest.raises(ManifestError, match="formal reflection canonical scope drift"): + parse_network_manifests(repo) + + +def test_network_tla_scope_preserves_comment_tokens_inside_strings(tmp_path: Path) -> None: + from tools.alpha4_network_manifest import ManifestError, parse_network_manifests + + repo = _copy_repo(tmp_path) + relational = repo / "network/alpha4/formal/NetworkRelations.tla" + text = relational.read_text(encoding="utf-8") + marker = 'result = "IMPORT_ADMITTED"' + assert marker in text + relational.write_text( + text.replace(marker, 'result = "IMPORT_ADMITTED(*scope-drift*)"', 1), + encoding="utf-8", + ) + with pytest.raises(ManifestError, match="relational canonical scope drift"): + parse_network_manifests(repo) diff --git a/tools/alpha4_network_expression_airgap.py b/tools/alpha4_network_expression_airgap.py index efadcc1..8b75ff8 100644 --- a/tools/alpha4_network_expression_airgap.py +++ b/tools/alpha4_network_expression_airgap.py @@ -1,6 +1,9 @@ from __future__ import annotations import argparse +import ast +import builtins +import io import itertools import json from collections import deque @@ -13,6 +16,41 @@ ROOT = Path(__file__).resolve().parents[1] +_ALLOWED_DIRECT_IMPORTS = frozenset({"hashlib"}) +_ALLOWED_FROM_IMPORTS = { + "pathlib": frozenset({"Path"}), + "typing": frozenset({"Any"}), +} +_FILESYSTEM_INSPECTION_METHODS = frozenset( + { + "absolute", + "cwd", + "exists", + "expanduser", + "glob", + "group", + "home", + "is_block_device", + "is_char_device", + "is_dir", + "is_fifo", + "is_file", + "is_mount", + "is_socket", + "is_symlink", + "iterdir", + "lstat", + "owner", + "readlink", + "resolve", + "rglob", + "samefile", + "stat", + "walk", + } +) + + class NetworkExpressionAirgapError(RuntimeError): pass @@ -22,10 +60,193 @@ def require(condition: bool, message: str) -> None: raise NetworkExpressionAirgapError(message) -def execute(path: Path) -> dict[str, Any]: - namespace: dict[str, Any] = {"__file__": str(path)} +def _validate_companion_ast( + source: str, *, allowed_imports: frozenset[str], allow_seed_loader: bool +) -> None: + tree = ast.parse(source) + parents: dict[ast.AST, ast.AST] = {} + for parent in ast.walk(tree): + for child in ast.iter_child_nodes(parent): + parents[child] = parent + + def enclosing_function(node: ast.AST) -> str | None: + current = node + while current in parents: + current = parents[current] + if isinstance(current, (ast.FunctionDef, ast.AsyncFunctionDef)): + return current.name + return None + + for node in ast.walk(tree): + if isinstance(node, ast.Import): + for alias in node.names: + require( + alias.asname is None + and alias.name in allowed_imports + and alias.name in _ALLOWED_DIRECT_IMPORTS, + f"air-gap companion import forbidden: {alias.name}", + ) + elif isinstance(node, ast.ImportFrom): + module = node.module or "" + imported = {alias.name for alias in node.names} + require(node.level == 0, "air-gap companion relative import forbidden") + if module == "__future__": + require( + imported == {"annotations"} + and all(alias.asname is None for alias in node.names), + "air-gap companion future import drift", + ) + else: + require( + module in allowed_imports + and module in _ALLOWED_FROM_IMPORTS + and imported <= _ALLOWED_FROM_IMPORTS[module] + and all(alias.asname is None for alias in node.names), + f"air-gap companion import forbidden: {module}", + ) + elif isinstance(node, ast.Name) and node.id == "__builtins__": + raise NetworkExpressionAirgapError("air-gap companion accesses __builtins__") + elif isinstance(node, ast.Attribute) and node.attr.startswith("_"): + raise NetworkExpressionAirgapError( + f"air-gap companion private attribute forbidden: {node.attr}" + ) + elif isinstance(node, ast.Call) and isinstance(node.func, ast.Name): + if node.func.id in { + "__import__", + "breakpoint", + "delattr", + "dir", + "eval", + "getattr", + "globals", + "help", + "input", + "locals", + "setattr", + "type", + "vars", + }: + raise NetworkExpressionAirgapError( + f"air-gap companion dynamic capability forbidden: {node.func.id}" + ) + if node.func.id in {"exec", "compile"}: + require( + allow_seed_loader and enclosing_function(node) == "_load_seed_base", + f"air-gap companion {node.func.id} permitted only for exact Seed base loader", + ) + elif isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute): + require( + node.func.attr not in _FILESYSTEM_INSPECTION_METHODS, + f"air-gap companion filesystem inspection forbidden: {node.func.attr}", + ) + require( + node.func.attr + not in { + "write_text", + "write_bytes", + "unlink", + "rename", + "replace", + "mkdir", + "touch", + "chmod", + "symlink_to", + "hardlink_to", + }, + f"air-gap companion filesystem mutation forbidden: {node.func.attr}", + ) + elif isinstance(node, ast.Constant) and isinstance(node.value, str): + lowered = node.value.lower() + require( + not any( + marker in lowered for marker in ("tools.", "tools/", ".tla", ".forth", ".petri") + ), + "air-gap companion embeds repository semantic-source locator", + ) + + +def execute( + path: Path, + allowed_root: Path, + *, + allowed_imports: frozenset[str], + allow_seed_loader: bool, +) -> dict[str, Any]: source = path.read_text(encoding="utf-8") - exec(compile(source, str(path), "exec"), namespace) + _validate_companion_ast( + source, allowed_imports=allowed_imports, allow_seed_loader=allow_seed_loader + ) + allowed_root = allowed_root.resolve() + original_io_open = io.open + + def guarded_open(file: object, *args: object, **kwargs: object): + if isinstance(file, int): + return original_io_open(file, *args, **kwargs) + mode = kwargs.get("mode", args[0] if args else "r") + require( + isinstance(mode, str) and not any(flag in mode for flag in "wax+"), + "air-gap companion file access must be read-only", + ) + candidate = Path(file).resolve() # type: ignore[arg-type] + require( + candidate == allowed_root or allowed_root in candidate.parents, + f"air-gap companion file access escaped materialized profile tree: {candidate}", + ) + return original_io_open(file, *args, **kwargs) + + original_import = builtins.__import__ + + def guarded_import( + name: str, + globals: dict[str, Any] | None = None, + locals: dict[str, Any] | None = None, + fromlist: tuple[str, ...] = (), + level: int = 0, + ) -> Any: + requested = set(fromlist or ()) + if level != 0: + raise ImportError("air-gap companion relative import forbidden") + if name == "__future__": + if requested != {"annotations"}: + raise ImportError("air-gap companion future import drift") + elif name in _ALLOWED_DIRECT_IMPORTS: + if name not in allowed_imports or requested: + raise ImportError(f"air-gap companion import forbidden: {name}") + elif name in _ALLOWED_FROM_IMPORTS: + if ( + name not in allowed_imports + or not requested + or not requested <= _ALLOWED_FROM_IMPORTS[name] + ): + raise ImportError(f"air-gap companion import forbidden: {name}") + else: + raise ImportError(f"air-gap companion import forbidden: {name}") + return original_import(name, globals, locals, fromlist, level) + + safe_builtins = dict(vars(builtins)) + safe_builtins["__import__"] = guarded_import + safe_builtins["open"] = guarded_open + + def guarded_exec( + code: object, + globals_dict: dict[str, Any] | None = None, + locals_dict: dict[str, Any] | None = None, + ) -> None: + target_globals = {} if globals_dict is None else globals_dict + target_globals.setdefault("__builtins__", safe_builtins) + exec(code, target_globals, locals_dict) + + safe_builtins["exec"] = guarded_exec + namespace: dict[str, Any] = { + "__file__": str(path), + "__name__": "aset_network_alpha4_airgap_subject", + "__builtins__": safe_builtins, + } + io.open = guarded_open # type: ignore[assignment] + try: + exec(compile(source, str(path), "exec"), namespace) + finally: + io.open = original_io_open # type: ignore[assignment] return namespace @@ -117,6 +338,86 @@ def _check_core(network: dict[str, Any], seed: dict[str, Any]) -> int: return cases +def _check_core_identity_sensitivity(network: dict[str, Any], seed: dict[str, Any]) -> int: + admit = network["admit_import"] + make_seed_state = seed["state"] + base = { + "import_id": "i0", + "source_context": "s0", + "target_context": "t0", + "evidence_digest": "sha256:" + "0" * 64, + } + replacements = { + "import_id": "i1", + "source_context": "s1", + "target_context": "t1", + "evidence_digest": "sha256:" + "1" * 64, + } + checks = 0 + for field, replacement in replacements.items(): + candidate = {**base, field: replacement} + state = [dict(base)] + seed_before = make_seed_state("subject-1", "authority-1") + actual_state, _, actual_result = admit(state, candidate, seed_before) + expected_state, expected_result = _core_expected(state, candidate) + require( + actual_state == expected_state, + f"core identity sensitivity state mismatch: {field}", + ) + for key, value in expected_result.items(): + require( + actual_result.get(key) == value, + f"core identity sensitivity result mismatch: {field}:{key}", + ) + checks += 1 + + first = {**base, "import_id": "i1"} + state = [first, dict(base)] + seed_before = make_seed_state("subject-1", "authority-1") + actual_state, _, actual_result = admit(state, dict(base), seed_before) + require(actual_state == state, "core second-position replay changed state") + require(actual_result.get("code") == "IDEMPOTENT_REPLAY", "core second-position replay missed") + return checks + 1 + + +def _check_composition_identity_sensitivity(network: dict[str, Any]) -> int: + delivery = network["delivery_witness"] + export = "e0" + sets = (set(), {"e0"}, {"e1"}, {"e0", "e1"}) + checks = 0 + for exported, delivered in itertools.product(sets, repeat=2): + require( + delivery(exported, delivered, export) == (export in exported and export in delivered), + "composition foreign-export identity sensitivity mismatch", + ) + checks += 1 + return checks + + +def _check_federation_identity_sensitivity(network: dict[str, Any]) -> int: + checks = 0 + empty = network["federation_state"]() + created = network["federation_genesis"](empty, "f1", "e1") + require(created["federation_id"] == "f1", "federation id identity lost") + checks += 1 + require(created["federation_epoch"] == "e1", "federation epoch identity lost") + checks += 1 + joined = network["member_join"](created, "B") + require( + joined["members"].get("B") == "ACTIVE" and "A" not in joined["members"], + "member identity lost", + ) + checks += 1 + joined_a = network["member_join"](joined, "A") + granted = network["route_grant"](joined_a, "B", "A") + require(granted["routes"].get(("B", "A")) == "ACTIVE", "route endpoint identity lost") + checks += 1 + exported = network["export_artifact"](granted, "B", "A", "x1") + require(("B", "A", "x1") in exported["exports"], "artifact identity lost") + checks += 1 + return checks + + def _check_dynamic(network: dict[str, Any]) -> int: applicable = network.get("profile_applicable") stutter = network.get("profile_network_stutter") @@ -346,8 +647,13 @@ def check_airgap(profiles_root: Path) -> dict[str, Any]: require(network_path.is_file(), "Network Python companion missing") require(seed_path.is_file(), "exact Seed Python base missing") require(sha256(seed_path) == binding.companions["PYTHON"][1], "Seed Python base bytes mismatch") - seed = execute(seed_path) - network = execute(network_path) + seed = execute(seed_path, profiles_root, allowed_imports=frozenset(), allow_seed_loader=False) + network = execute( + network_path, + profiles_root, + allowed_imports=frozenset({"hashlib", "pathlib", "typing"}), + allow_seed_loader=True, + ) require( network.get("BASE_SEED_EXPRESSION_SHA256") == sha256(seed_path), "Network Python base binding mismatch", @@ -357,11 +663,19 @@ def check_airgap(profiles_root: Path) -> dict[str, Any]: federation_states, federation_edges = _check_federation(network) liveness = _check_liveness(network) composition = _check_composition(network) + core_identity = _check_core_identity_sensitivity(network, seed) + composition_identity = _check_composition_identity_sensitivity(network) + federation_identity = _check_federation_identity_sensitivity(network) total = core + dynamic + federation_edges + liveness + composition + sensitivity = core_identity + composition_identity + federation_identity require( (core, dynamic, federation_states, federation_edges, liveness, composition, total) == (272, 10, 20, 25, 51, 88, 446), - "Network Python air-gap coverage drift", + "Network Python air-gap structural coverage drift", + ) + require( + (core_identity, composition_identity, federation_identity, sensitivity) == (5, 16, 5, 26), + "Network Python air-gap identity sensitivity drift", ) after = tree_digest(profiles_root) require(after == before, "Network profile tree changed during air-gap verification") @@ -373,6 +687,8 @@ def check_airgap(profiles_root: Path) -> dict[str, Any]: "network_semantic_source": "NONE", "release_profile_generator": "NONE", "triangulated_expression_checker": "NONE", + "companion_import_surface": "RESTRICTED", + "companion_file_access": "MATERIALIZED_PROFILE_TREE_READ_ONLY", }, "profile_tree_digest": before, "inputs": { @@ -393,6 +709,11 @@ def check_airgap(profiles_root: Path) -> dict[str, Any]: "liveness_cases": liveness, "composition_cases": composition, "total_cases": total, + "core_identity_sensitivity_cases": core_identity, + "composition_identity_sensitivity_cases": composition_identity, + "federation_identity_sensitivity_cases": federation_identity, + "sensitivity_cases": sensitivity, + "grand_total_cases": total + sensitivity, }, "profile_tree_unchanged": True, "status": "PASS", @@ -424,6 +745,12 @@ def main() -> int: print(f"ALPHA4_NETWORK_PYTHON_AIRGAP_LIVENESS={coverage['liveness_cases']}/51 PASS") print(f"ALPHA4_NETWORK_PYTHON_AIRGAP_COMPOSITION={coverage['composition_cases']}/88 PASS") print(f"ALPHA4_NETWORK_PYTHON_AIRGAP_TOTAL={coverage['total_cases']}/446 PASS") + print( + "ALPHA4_NETWORK_PYTHON_AIRGAP_IDENTITY_SENSITIVITY=" + f"{coverage['sensitivity_cases']}/26 PASS" + ) + print(f"ALPHA4_NETWORK_PYTHON_AIRGAP_GRAND_TOTAL={coverage['grand_total_cases']}/472 PASS") + print("ALPHA4_NETWORK_PYTHON_COMPANION_RUNTIME_ISOLATION=PASS") print("ALPHA4_NETWORK_PYTHON_SEED_BASE=EXACT") print("ALPHA4_NETWORK_PYTHON_SEMANTIC_SOURCE_DEPENDENCY=NONE") print("ALPHA4_NETWORK_PYTHON_GENERATOR_DEPENDENCY=NONE") diff --git a/tools/alpha4_network_manifest.py b/tools/alpha4_network_manifest.py index 602aa78..47b9ee3 100644 --- a/tools/alpha4_network_manifest.py +++ b/tools/alpha4_network_manifest.py @@ -1,5 +1,6 @@ from __future__ import annotations +import hashlib from collections import Counter from dataclasses import dataclass from pathlib import Path @@ -561,6 +562,120 @@ def _proof(proof_id: str, module: str, theorem: str, obligations: int) -> ProofB ) +def _strip_tla_comments(source: str) -> str: + out: list[str] = [] + index = 0 + block_depth = 0 + in_string = False + while index < len(source): + if block_depth: + if source.startswith("(*", index): + block_depth += 1 + index += 2 + elif source.startswith("*)", index): + block_depth -= 1 + index += 2 + elif source[index] == "\n": + out.append("\n") + index += 1 + else: + index += 1 + continue + + if in_string: + char = source[index] + out.append(char) + if char == "\\" and index + 1 < len(source): + out.append(source[index + 1]) + index += 2 + else: + if char == '"': + in_string = False + index += 1 + continue + + if source.startswith("(*", index): + block_depth = 1 + index += 2 + continue + if source.startswith("\\*", index): + while index < len(source) and source[index] != "\n": + index += 1 + continue + char = source[index] + out.append(char) + if char == '"': + in_string = True + index += 1 + + if block_depth: + raise ManifestError("unterminated TLA block comment in canonical scope") + if in_string: + raise ManifestError("unterminated TLA string in canonical scope") + return "".join(out) + + +def _canonical_tla_scope_sha256(path: Path) -> str: + source = path.read_text(encoding="utf-8").replace("\r\n", "\n").replace("\r", "\n") + uncommented = _strip_tla_comments(source) + canonical = "\n".join(line.strip() for line in uncommented.splitlines() if line.strip()) + return "sha256:" + hashlib.sha256(canonical.encode("utf-8")).hexdigest() + + +EXPECTED_RELATIONAL_SCOPE_SHA256 = { + "network": ("sha256:b9131fcf4c369b721dd42f034513d162230660519bf6ed26f9407b4e32649cdb"), + "dynamic": ("sha256:b3092c0b7b9e7dd9fbae3c358e327318749026c4a6fc5d1272243204c7fa77ce"), + "federation": ("sha256:61c35ce0bb26e41291e2602a9831a59e4900c5fb437b0a89f40f83e617749b64"), + "liveness": ("sha256:7889274f51a93f273c014473e532ea853696479710774cd7b5394caf693e6165"), + "federation-liveness": ( + "sha256:5d28ff45e3919d656311586731855d06284bdfea14e2a37f5156c28f6fa5e036" + ), +} + +EXPECTED_FORMAL_REFLECTION_SCOPE_SHA256 = { + "network": ("sha256:bf1f919b0b38b8a5c2564de2e795bedcdd35efd7580e36b02e9845bb58da7675"), + "dynamic": ("sha256:896bdfde2f3d718f2ea2ca60c3c4878d0eaaa1f33920a0a14290dfb55d72c3f0"), + "federation": ("sha256:fd383161e69b7aeb935c43e1dde22047e150e816e8768f8c4c575855837b69cf"), + "liveness": ("sha256:df8b2743bd1917647ee88d8a445818226058bfdd239fc4258f4fa4721447acc4"), + "federation-liveness": ( + "sha256:a2928d06f7495c972748d9a812f9b6055e8f21a1374f4bf2f6f2112f677da13d" + ), +} + +EXPECTED_PROOF_SCOPE_SHA256 = { + ("network", "OPERATIONAL_RELATIONAL_PAIRING"): ( + "sha256:b41437ac70756fc117ba13c9e54260db56dd6042920bef98a82e7b46bcfa683a" + ), + ("network", "SEED_BOUNDARY"): ( + "sha256:12d5da90ac3d32eda384e34f3a5c3e508b2180366bd525d64a80cdaffcd941ec" + ), + ("dynamic", "OPERATIONAL_RELATIONAL_PAIRING"): ( + "sha256:7e1c168fae31f264c201d20c6d5a20718788dfbc3474d260b07e239f0a2c3bad" + ), + ("dynamic", "BOUNDARY"): ( + "sha256:2d7394fe8ba527a5866747a35d532a2f54df79c957c3d6c9e7cc49c30ee66ba6" + ), + ("federation", "OPERATIONAL_RELATIONAL_PAIRING"): ( + "sha256:e92229c0a75f01aee9c31131b97c8416e4585198e46081c04546b75a9f6dc4e2" + ), + ("federation", "NETWORK_STUTTER"): ( + "sha256:78c435a803bd20b0eb9d00e96c7849266f273af0d694dcdf1d7483bebb250885" + ), + ("liveness", "OPERATIONAL_RELATIONAL_PAIRING"): ( + "sha256:2cf561898d8e750cb8c395f2d4c1d6bf2626b2176b6accce982fe2cfc79c3a90" + ), + ("liveness", "BOUNDARY"): ( + "sha256:8b9c4b6cef4dc225937553d458c2a14a7ff800c23ba21890a0171044a5c491b6" + ), + ("federation-liveness", "OPERATIONAL_RELATIONAL_PAIRING"): ( + "sha256:6cd3405923872d261a4947810bbeed5b19e6e3e50748e3c5fcec26ec85164665" + ), + ("federation-liveness", "CONTRACT"): ( + "sha256:f4238ff97bf7cb4d37a1b910fd964ebcccaa05283f87afa5d6c6add3cf33894a" + ), +} + + def _read_tokens(path: Path) -> list[list[str]]: result: list[list[str]] = [] for raw in path.read_text(encoding="utf-8").splitlines(): @@ -664,11 +779,26 @@ def _parse_subject(root: Path, name: str, schema: SubjectSchema) -> SubjectBindi for relative in (*sources.values(), *(item.module for item in proofs)): bound = root / relative require(bound.is_file(), f"{name}: bound file missing: {relative}") + require( + _canonical_tla_scope_sha256(root / sources["RELATIONAL"]) + == EXPECTED_RELATIONAL_SCOPE_SHA256[name], + f"{name}: relational canonical scope drift", + ) + require( + _canonical_tla_scope_sha256(root / sources["FORMAL-REFLECTION"]) + == EXPECTED_FORMAL_REFLECTION_SCOPE_SHA256[name], + f"{name}: formal reflection canonical scope drift", + ) for proof in proofs: require( _theorem_present(root / proof.module, proof.final_theorem), f"{name}: final theorem missing: {proof.final_theorem}", ) + require( + _canonical_tla_scope_sha256(root / proof.module) + == EXPECTED_PROOF_SCOPE_SHA256[(name, proof.proof_id)], + f"{name}: proof canonical scope drift: {proof.proof_id}", + ) for pair in pairs: relational_text = (root / sources["RELATIONAL"]).read_text(encoding="utf-8") reflection_text = (root / sources["FORMAL-REFLECTION"]).read_text(encoding="utf-8") @@ -719,6 +849,19 @@ def main() -> int: print(f"ALPHA4_NETWORK_MANIFEST_SUBJECTS={len(plan.subjects)}/{len(plan.subjects)} PASS") print(f"ALPHA4_NETWORK_MANIFEST_PAIRS={pairs}/{pairs} PASS") print(f"ALPHA4_NETWORK_MANIFEST_PROOFS={proofs}/{proofs} PASS") + print( + "ALPHA4_NETWORK_RELATIONAL_CANONICAL_SCOPES=" + f"{len(EXPECTED_RELATIONAL_SCOPE_SHA256)}/{len(EXPECTED_RELATIONAL_SCOPE_SHA256)} PASS" + ) + print( + "ALPHA4_NETWORK_FORMAL_REFLECTION_CANONICAL_SCOPES=" + f"{len(EXPECTED_FORMAL_REFLECTION_SCOPE_SHA256)}/" + f"{len(EXPECTED_FORMAL_REFLECTION_SCOPE_SHA256)} PASS" + ) + print( + "ALPHA4_NETWORK_PROOF_CANONICAL_SCOPES=" + f"{len(EXPECTED_PROOF_SCOPE_SHA256)}/{len(EXPECTED_PROOF_SCOPE_SHA256)} PASS" + ) print(f"ALPHA4_NETWORK_MANIFEST_EXPECTED_TLAPS_OBLIGATIONS={obligations}") print("ALPHA4_NETWORK_BINDING_PLAN=PASS") return 0 diff --git a/tools/alpha4_network_paired_expression.py b/tools/alpha4_network_paired_expression.py index 2db1d7c..9910e2d 100644 --- a/tools/alpha4_network_paired_expression.py +++ b/tools/alpha4_network_paired_expression.py @@ -6,7 +6,10 @@ from pathlib import Path from typing import Any -from tools.alpha4_network_relational_expression import relational_admit_from_source +from tools.alpha4_network_relational_expression import ( + relational_admit_from_source, + relational_exact_observation_from_source, +) ROOT = Path(__file__).resolve().parents[1] FORTH = ROOT / "network/alpha4/operational/components.forth" @@ -97,7 +100,7 @@ def operational_admit( def relational_admit( imports: list[dict[str, Any]], observation: dict[str, Any] ) -> tuple[list[dict[str, Any]], dict[str, Any]]: - if not exact_observation(observation): + if not relational_exact_observation_from_source(observation): return deepcopy(imports), _result(False, "INVALID_IMPORT", False) return relational_admit_from_source(imports, observation) diff --git a/tools/alpha4_network_relational_expression.py b/tools/alpha4_network_relational_expression.py index 58cd39f..0d73d5f 100644 --- a/tools/alpha4_network_relational_expression.py +++ b/tools/alpha4_network_relational_expression.py @@ -157,13 +157,27 @@ def _result(accepted: bool, code: str, changed: bool) -> dict[str, Any]: } +def relational_exact_observation_from_source(value: dict[str, Any], root: Path = ROOT) -> bool: + contract = derive_core_contract(root) + fields = set(contract.observation_fields) + if set(value) != fields: + return False + if not all(isinstance(value[field], str) and value[field] for field in fields): + return False + digest_fields = [field for field in contract.observation_fields if field.endswith("_digest")] + if len(digest_fields) != 1: + raise RelationalExpressionError("relational observation digest field is not singular") + digest = value[digest_fields[0]] + return bool(re.fullmatch(r"sha256:[0-9a-f]{64}", digest)) + + def relational_admit_from_source( imports: list[dict[str, Any]], observation: dict[str, Any], root: Path = ROOT ) -> tuple[list[dict[str, Any]], dict[str, Any]]: contract = derive_core_contract(root) require( - set(observation) == set(contract.observation_fields), - "relational observation field surface mismatch", + relational_exact_observation_from_source(observation, root), + "relational observation field/type surface mismatch", ) identifier_exists = any( item[contract.identifier_field] == observation[contract.identifier_field] diff --git a/tools/alpha4_network_release_admission.py b/tools/alpha4_network_release_admission.py index e0f9c42..73a9eb0 100644 --- a/tools/alpha4_network_release_admission.py +++ b/tools/alpha4_network_release_admission.py @@ -200,9 +200,18 @@ def check_admission( "Network Python extension is not based on exact Seed Python companion", ) coverage = airgap.get("coverage") + require(isinstance(coverage, dict), "Network Python air-gap coverage missing") require( - isinstance(coverage, dict) and coverage.get("total_cases") == 446, - "Network Python air-gap coverage mismatch", + coverage.get("total_cases") == 446, + "Network Python air-gap structural coverage mismatch", + ) + require( + coverage.get("core_identity_sensitivity_cases") == 5 + and coverage.get("composition_identity_sensitivity_cases") == 16 + and coverage.get("federation_identity_sensitivity_cases") == 5 + and coverage.get("sensitivity_cases") == 26 + and coverage.get("grand_total_cases") == 472, + "Network Python air-gap identity sensitivity coverage mismatch", ) require(airgap.get("status") == "PASS", "Network Python air-gap is not PASS") dependencies = airgap.get("assurance_dependencies") @@ -210,7 +219,9 @@ def check_admission( isinstance(dependencies, dict) and dependencies.get("network_semantic_source") == "NONE" and dependencies.get("release_profile_generator") == "NONE" - and dependencies.get("triangulated_expression_checker") == "NONE", + and dependencies.get("triangulated_expression_checker") == "NONE" + and dependencies.get("companion_import_surface") == "RESTRICTED" + and dependencies.get("companion_file_access") == "MATERIALIZED_PROFILE_TREE_READ_ONLY", "Network Python air-gap independence boundary drift", ) @@ -301,7 +312,13 @@ def check_admission( "obligations_proved": proof_subject["obligations_proved"], "status": "PASS", }, - "python_airgap": {"cases": coverage["total_cases"], "status": "PASS"}, + "python_airgap": { + "structural_cases": coverage["total_cases"], + "identity_sensitivity_cases": coverage["sensitivity_cases"], + "grand_total_cases": coverage["grand_total_cases"], + "runtime_isolation": "PASS", + "status": "PASS", + }, "public_assurance": { "identity": "ASET_NETWORK", "representation": "0.1.0-alpha.4", @@ -345,6 +362,9 @@ def main() -> int: print("ALPHA4_NETWORK_RELEASE_ADMISSION_ENGLISH_SEED_BASE=EXACT") print("ALPHA4_NETWORK_RELEASE_ADMISSION_PYTHON_SEED_BASE=EXACT") print("ALPHA4_NETWORK_RELEASE_ADMISSION_PYTHON_AIRGAP=446/446 PASS") + print("ALPHA4_NETWORK_RELEASE_ADMISSION_PYTHON_AIRGAP_IDENTITY_SENSITIVITY=26/26 PASS") + print("ALPHA4_NETWORK_RELEASE_ADMISSION_PYTHON_AIRGAP_GRAND_TOTAL=472/472 PASS") + print("ALPHA4_NETWORK_RELEASE_ADMISSION_PYTHON_RUNTIME_ISOLATION=PASS") print("ALPHA4_NETWORK_RELEASE_ADMISSION_ARCHIVE_BINDING=EXACT") print("ALPHA4_NETWORK_PUBLIC_ASSURANCE_REPRESENTATIONS=OPERATIONAL,RELATIONAL,CAUSAL") print("ALPHA4_NETWORK_PUBLIC_POST_BUILD_FORMAL_ASSURANCE=PASS") diff --git a/tools/alpha4_network_release_profiles.py b/tools/alpha4_network_release_profiles.py index 3eaf3e5..8108083 100644 --- a/tools/alpha4_network_release_profiles.py +++ b/tools/alpha4_network_release_profiles.py @@ -141,7 +141,7 @@ def write_python(target: Path, seed_python_sha256: str, records: list[dict[str, BASE_SEED_EXPRESSION_SHA256 = {seed_python_sha256!r} BASE_SEED_EXPRESSION_PATH = ( - Path(__file__).resolve().parents[1] + Path(__file__).parent.parent / "base" / "seed" / "python" diff --git a/tools/alpha4_network_triangulated_expression.py b/tools/alpha4_network_triangulated_expression.py index 83b176e..407c98b 100644 --- a/tools/alpha4_network_triangulated_expression.py +++ b/tools/alpha4_network_triangulated_expression.py @@ -33,6 +33,7 @@ ) from tools.alpha4_network_relational_expression import ( federation_relational_edges_from_source, + relational_exact_observation_from_source, validate_all_relational_sources, validate_federation_identity_guards, ) @@ -191,23 +192,29 @@ def _interface_validator_independence() -> int: "target_context": "t0", "evidence_digest": "sha256:" + "0" * 64, } - invalid = [ - {**valid, "evidence_digest": "NOT-A-SHA256"}, - {key: value for key, value in valid.items() if key != "source_context"}, - {**valid, "extra": "x"}, - ] - require( - exact_observation(valid) and causal_exact_observation(valid), - "valid interface record rejected", + cases: list[tuple[dict[str, object], bool]] = [(valid, True)] + for field in tuple(valid): + cases.append(({key: value for key, value in valid.items() if key != field}, False)) + cases.extend( + [ + ({**valid, "extra": "x"}, False), + ({**valid, "import_id": ""}, False), + ({**valid, "source_context": 1}, False), + ({**valid, "target_context": None}, False), + ({**valid, "evidence_digest": "NOT-A-SHA256"}, False), + ({**valid, "evidence_digest": "sha256:" + "g" * 64}, False), + ({**valid, "evidence_digest": "sha256:" + "0" * 63}, False), + ] ) - checks = 1 - for value in invalid: + for value, expected in cases: + operational = exact_observation(value) + relational = relational_exact_observation_from_source(value) + causal = causal_exact_observation(value) require( - exact_observation(value) == causal_exact_observation(value) is False, - "operational/causal interface validators disagree", + operational == relational == causal == expected, + "operational/relational/causal interface validators disagree", ) - checks += 1 - return checks + return len(cases) def _core_triangulation(net: CausalNet) -> tuple[int, int]: