From d4a9ead76ed8c9c6053a282e6f374261084f032a Mon Sep 17 00:00:00 2001 From: yuqei Date: Sun, 27 Sep 2026 16:59:15 +0800 Subject: [PATCH] fix(release): bind RC06 apply authorization context --- .../scripts/Launch-ReleaseAuthority.ps1 | 2 + .../scripts/rc06-release-trust-bootstrap.py | 444 ++++++++++++++++-- .../test_rc06_release_trust_bootstrap.py | 261 +++++++++- 3 files changed, 667 insertions(+), 40 deletions(-) diff --git a/yuqei-infra/scripts/Launch-ReleaseAuthority.ps1 b/yuqei-infra/scripts/Launch-ReleaseAuthority.ps1 index a389c304..2efdb849 100644 --- a/yuqei-infra/scripts/Launch-ReleaseAuthority.ps1 +++ b/yuqei-infra/scripts/Launch-ReleaseAuthority.ps1 @@ -7,7 +7,9 @@ param( "status", "build-recovery-keyrings", "pin-source", + "recover-pin-source", "sign-rc06-keyring", + "issue-apply-challenge", "sign-apply-payloads", "provision-apply-private-key" )] diff --git a/yuqei-infra/scripts/rc06-release-trust-bootstrap.py b/yuqei-infra/scripts/rc06-release-trust-bootstrap.py index 20752088..fd7c9c27 100644 --- a/yuqei-infra/scripts/rc06-release-trust-bootstrap.py +++ b/yuqei-infra/scripts/rc06-release-trust-bootstrap.py @@ -95,6 +95,7 @@ RECOVERY_BUNDLE_SCHEMA = "yuqei.release-recovery-keyring-bundle.v2" RC06_KEYRING_BUNDLE_SCHEMA = "yuqei.rc06-candidate-keyring-bundle.v2" APPLY_RECEIPT_BUNDLE_SCHEMA = "yuqei.rc06-apply-receipt-bundle.v3" APPLY_CHALLENGE_CONSUMPTION_SCHEMA = "yuqei.rc06-apply-challenge-consumption.v1" +APPLY_CHALLENGE_CONTEXT_SCHEMA = "yuqei.rc06-apply-challenge-context.v1" PIN_QUARANTINE_SCHEMA = "yuqei.release-authority-pin-quarantine.v1" PIN_PLAN_SCHEMA = "yuqei.release-authority-pin-plan.v1" PIN_TRANSACTION_SCHEMA = "yuqei.release-authority-pin-transaction.v2" @@ -1296,6 +1297,105 @@ def _read_expected_json( return payload, raw +def _apply_challenge_context_identity( + *, + manifest_sha256: str, + approval_id: str, + approval_sha256: str, + environment_id: str, +) -> dict[str, str]: + return { + "manifest_raw_sha256": manifest_sha256, + "approval_id": approval_id, + "approval_raw_sha256": approval_sha256, + "environment_id": environment_id, + "operation": batch.APPLY_OPERATION, + } + + +def _apply_challenge_context_path( + state_directory: Path, + identity: Mapping[str, str], +) -> tuple[Path, str]: + directory = state_directory / "apply-challenge-contexts" + try: + batch._assert_no_symlink_components(directory, "apply challenge context directory") + except batch.PublicationBatchError as exc: + raise AuthorityError(str(exc)) from exc + directory.mkdir(mode=0o700, exist_ok=True) + try: + batch._assert_no_symlink_components(directory, "apply challenge context directory") + except batch.PublicationBatchError as exc: + raise AuthorityError(str(exc)) from exc + context_id = "rc06-apply-context-" + batch._sha256_json(identity) + return directory / f"{context_id}.json", context_id + + +def _load_apply_challenge_context( + path: Path, + context_id: str, + identity: Mapping[str, str], +) -> tuple[dict[str, Any], dict[str, Any], bytes]: + ledger = _read_json(path, "apply challenge context ledger") + expected_fields = { + "schema_version", + "context_id", + "context", + "challenge", + "challenge_raw_sha256", + } + challenge = ledger.get("challenge") + if ( + set(ledger) != expected_fields + or ledger.get("schema_version") != APPLY_CHALLENGE_CONTEXT_SCHEMA + or ledger.get("context_id") != context_id + or ledger.get("context") != dict(identity) + or not isinstance(challenge, dict) + ): + raise AuthorityError("apply challenge context ledger identity drift") + challenge_raw = _canonical(challenge) + if ledger.get("challenge_raw_sha256") != batch._sha256_bytes(challenge_raw): + raise AuthorityError("apply challenge context ledger challenge drift") + return ledger, challenge, challenge_raw + + +def _publish_persisted_apply_challenge( + state_directory: Path, + challenge: Mapping[str, Any], + challenge_raw: bytes, +) -> Path: + challenge_directory = state_directory / "apply-challenges" + try: + batch._assert_no_symlink_components( + challenge_directory, "apply challenge directory" + ) + except batch.PublicationBatchError as exc: + raise AuthorityError(str(exc)) from exc + challenge_directory.mkdir(mode=0o700, exist_ok=True) + try: + batch._assert_no_symlink_components( + challenge_directory, "apply challenge directory" + ) + except batch.PublicationBatchError as exc: + raise AuthorityError(str(exc)) from exc + challenge_path = challenge_directory / f"{challenge['challenge_id']}.json" + if challenge_path.exists() or challenge_path.is_symlink(): + existing = batch.secure_read_bytes( + challenge_path, + "persisted apply challenge", + maximum_bytes=MAX_REGISTRATION_BYTES, + ) + if existing != challenge_raw: + raise AuthorityError("persisted apply challenge identity drift") + else: + _publish_private_no_replace( + challenge_path, + challenge_raw, + "apply challenge", + ) + return challenge_path + + def issue_apply_challenge( state_directory: Path, *, @@ -1368,6 +1468,49 @@ def issue_apply_challenge( ) except (ValueError, batch.PublicationBatchError) as exc: raise AuthorityError(f"RC-06 approval chain is invalid: {exc}") from exc + context_identity = _apply_challenge_context_identity( + manifest_sha256=manifest_sha256, + approval_id=approval["approval_id"], + approval_sha256=approval_sha256, + environment_id=environment_id, + ) + context_path, context_id = _apply_challenge_context_path( + state_directory, + context_identity, + ) + if context_path.exists() or context_path.is_symlink(): + _ledger, challenge, challenge_raw = _load_apply_challenge_context( + context_path, + context_id, + context_identity, + ) + try: + batch.validate_apply_challenge( + validated_manifest, + manifest_sha256, + approval_sha256, + apply_service.public_key, + challenge, + ) + except batch.PublicationBatchError as exc: + raise AuthorityError( + f"persisted apply challenge failed verification: {exc}" + ) from exc + challenge_path = _publish_persisted_apply_challenge( + state_directory, + challenge, + challenge_raw, + ) + return { + "status": "apply_challenge_already_issued_not_replaced", + "authorization_context_id": context_id, + "challenge_id": challenge["challenge_id"], + "challenge_file": os.fspath(challenge_path), + "expires_at": challenge["payload"]["expires_at"], + "environment_id": environment_id, + "operation": batch.APPLY_OPERATION, + } + approved_at = _parse_time(approval["payload"]["approved_at"], "approval approved_at") approval_expires = _parse_time( approval["payload"]["expires_at"], "approval expires_at" @@ -1423,28 +1566,27 @@ def issue_apply_challenge( ) except batch.PublicationBatchError as exc: raise AuthorityError(f"apply challenge self-verification failed: {exc}") from exc - challenge_directory = state_directory / "apply-challenges" - try: - batch._assert_no_symlink_components( - challenge_directory, "apply challenge directory" - ) - except batch.PublicationBatchError as exc: - raise AuthorityError(str(exc)) from exc - challenge_directory.mkdir(mode=0o700, exist_ok=True) - try: - batch._assert_no_symlink_components( - challenge_directory, "apply challenge directory" - ) - except batch.PublicationBatchError as exc: - raise AuthorityError(str(exc)) from exc - challenge_path = challenge_directory / f"{challenge['challenge_id']}.json" + challenge_raw = _canonical(challenge) + context_ledger = { + "schema_version": APPLY_CHALLENGE_CONTEXT_SCHEMA, + "context_id": context_id, + "context": context_identity, + "challenge": challenge, + "challenge_raw_sha256": batch._sha256_bytes(challenge_raw), + } _publish_private_no_replace( - challenge_path, - _canonical(challenge), - "apply challenge", + context_path, + _canonical(context_ledger), + "apply challenge context ledger", + ) + challenge_path = _publish_persisted_apply_challenge( + state_directory, + challenge, + challenge_raw, ) return { "status": "apply_challenge_issued_not_consumed", + "authorization_context_id": context_id, "challenge_id": challenge["challenge_id"], "challenge_file": os.fspath(challenge_path), "expires_at": payload["expires_at"], @@ -1525,7 +1667,11 @@ def _build_apply_receipt( def _load_persisted_apply_challenge( state_directory: Path, online_receipt: Mapping[str, Any], -) -> tuple[dict[str, Any], bytes]: + *, + manifest_sha256: str, + approval: Mapping[str, Any], + approval_sha256: str, +) -> tuple[dict[str, Any], bytes, str]: challenge = online_receipt.get("payload", {}).get("authority_challenge") if not isinstance(challenge, dict): raise AuthorityError("online transaction receipt has no authority challenge") @@ -1534,6 +1680,28 @@ def _load_persisted_apply_challenge( r"rc06-apply-challenge-[0-9a-f]{24}", challenge_id ): raise AuthorityError("online transaction authority challenge_id is invalid") + challenge_payload = challenge.get("payload") + if not isinstance(challenge_payload, dict): + raise AuthorityError("online transaction authority challenge payload is invalid") + context_identity = _apply_challenge_context_identity( + manifest_sha256=manifest_sha256, + approval_id=str(approval.get("approval_id", "")), + approval_sha256=approval_sha256, + environment_id=str(challenge_payload.get("environment_id", "")), + ) + context_path, context_id = _apply_challenge_context_path( + state_directory, + context_identity, + ) + _ledger, context_challenge, context_challenge_raw = _load_apply_challenge_context( + context_path, + context_id, + context_identity, + ) + if context_challenge != challenge: + raise AuthorityError( + "online transaction challenge is not the unique authorization-context challenge" + ) challenge_path = state_directory / "apply-challenges" / f"{challenge_id}.json" persisted = _read_json(challenge_path, "persisted apply challenge") persisted_raw = batch.secure_read_bytes( @@ -1541,9 +1709,13 @@ def _load_persisted_apply_challenge( "persisted apply challenge", maximum_bytes=MAX_REGISTRATION_BYTES, ) - if _canonical(persisted) != persisted_raw or persisted != challenge: + if ( + _canonical(persisted) != persisted_raw + or persisted_raw != context_challenge_raw + or persisted != challenge + ): raise AuthorityError("online transaction challenge is not the persisted authority challenge") - return persisted, persisted_raw + return persisted, persisted_raw, context_id def _apply_consumption_identity( @@ -1554,8 +1726,10 @@ def _apply_consumption_identity( output_directory: Path, manifest_sha256: str, approval_sha256: str, + authorization_context_id: str, ) -> dict[str, Any]: return { + "authorization_context_id": authorization_context_id, "challenge_id": challenge["challenge_id"], "challenge_raw_sha256": batch._sha256_bytes(challenge_raw), "online_receipt_id": online_receipt["receipt_id"], @@ -1800,9 +1974,16 @@ def sign_apply_payloads( ) from exc verified_payloads.append(_apply_payload_from_online_receipt(online_receipt)) - persisted_challenge, persisted_challenge_raw = _load_persisted_apply_challenge( + ( + persisted_challenge, + persisted_challenge_raw, + authorization_context_id, + ) = _load_persisted_apply_challenge( state_directory, online_receipts[0], + manifest_sha256=manifest_sha256, + approval=approval, + approval_sha256=approval_sha256, ) consumption_identity = _apply_consumption_identity( challenge=persisted_challenge, @@ -1811,6 +1992,7 @@ def sign_apply_payloads( output_directory=output_directory, manifest_sha256=manifest_sha256, approval_sha256=approval_sha256, + authorization_context_id=authorization_context_id, ) if output_directory.exists() or output_directory.is_symlink(): @@ -1894,6 +2076,7 @@ def sign_apply_payloads( ), "apply_service_public_key_sha256": apply_service.public_sha256, "apply_challenge_id": persisted_challenge["challenge_id"], + "authorization_context_id": authorization_context_id, "apply_challenge_raw_sha256": batch._sha256_bytes( persisted_challenge_raw ), @@ -2017,6 +2200,7 @@ def sign_apply_payloads( ), "apply_service_public_key_sha256": apply_service.public_sha256, "apply_challenge_id": persisted_challenge["challenge_id"], + "authorization_context_id": authorization_context_id, "apply_challenge_raw_sha256": batch._sha256_bytes( persisted_challenge_raw ), @@ -2059,22 +2243,31 @@ def sign_apply_payloads( } +_VERIFIED_MODULE_EXEC_TEMPLATE = """ +def _exec_verified_module(name: str, path: Path, raw: bytes) -> Any: + module = types.ModuleType(name) + module.__file__ = str(path) + module.__package__ = "" + sys.modules[name] = module + try: + exec(compile(raw, str(path), "exec"), module.__dict__) + except Exception: + sys.modules.pop(name, None) + raise + return module +""" + + def _safe_verified_module_exec(call: ast.Call, parents: Mapping[ast.AST, ast.AST]) -> bool: parent: ast.AST | None = call while parent is not None and not isinstance(parent, (ast.FunctionDef, ast.AsyncFunctionDef)): parent = parents.get(parent) - return bool( - isinstance(parent, ast.FunctionDef) - and parent.name == "_exec_verified_module" - and len(call.args) == 2 - and not call.keywords - and isinstance(call.args[0], ast.Call) - and isinstance(call.args[0].func, ast.Name) - and call.args[0].func.id == "compile" - and isinstance(call.args[1], ast.Attribute) - and isinstance(call.args[1].value, ast.Name) - and call.args[1].value.id == "module" - and call.args[1].attr == "__dict__" + if not isinstance(parent, ast.FunctionDef): + return False + expected = ast.parse(_VERIFIED_MODULE_EXEC_TEMPLATE).body[0] + return ast.dump(parent, include_attributes=False) == ast.dump( + expected, + include_attributes=False, ) @@ -2085,6 +2278,7 @@ def _validate_isolated_runtime_constants( expected: Mapping[str, Any], ) -> None: marker = "YUQEI_PIN_RUNTIME_VALUES=" + module_name = "_" + os.urandom(32).hex() verifier = r""" import json import pathlib @@ -2093,7 +2287,7 @@ import types source = sys.stdin.buffer.read() sys.path.insert(0, str(pathlib.Path(sys.argv[1]).resolve().parent)) -name = "_yuqei_pin_runtime_validation" +name = sys.argv[3] module = types.ModuleType(name) module.__file__ = sys.argv[1] module.__package__ = "" @@ -2112,6 +2306,7 @@ sys.__stdout__.write("YUQEI_PIN_RUNTIME_VALUES=" + json.dumps(values, sort_keys= verifier, os.fspath(path), json.dumps(sorted(expected)), + module_name, ], cwd=repo, input=raw, @@ -2180,14 +2375,187 @@ def _validate_source_constant_contract( ): top_level[statement.target.id].append(statement) stores = {name: 0 for name in names} + builtins_bindings = {"__builtins__"} + for candidate in ast.walk(tree): + if isinstance(candidate, ast.Import): + builtins_bindings.update( + alias.asname or alias.name + for alias in candidate.names + if alias.name == "builtins" + ) + forbidden_dynamic_calls = { + "globals", + "locals", + "vars", + "eval", + "setattr", + "delattr", + "__setattr__", + "__delattr__", + "setitem", + "delitem", + "__setitem__", + "__delitem__", + "__import__", + } + dangerous_builtins = forbidden_dynamic_calls | {"exec"} + + def called_name(call: ast.Call) -> str | None: + target = call.func + if isinstance(target, ast.Name): + return target.id + if isinstance(target, ast.Attribute): + return target.attr + return None + + def literal_value(node: ast.AST) -> Any | None: + try: + return ast.literal_eval(node) + except (TypeError, ValueError): + return None + + def is_module_namespace(node: ast.AST) -> bool: + if isinstance(node, ast.Attribute) and node.attr == "__dict__": + return True + if isinstance(node, ast.Call): + dynamic = called_name(node) + if dynamic in {"globals", "locals", "vars"}: + return True + if ( + dynamic == "getattr" + and len(node.args) >= 2 + and literal_value(node.args[1]) == "__dict__" + ): + return True + return False + + def protected_literal(node: ast.AST) -> str | None: + value = literal_value(node) + if isinstance(value, str) and value in stores: + return value + if isinstance(value, dict): + for key in value: + if isinstance(key, str) and key in stores: + return key + return None + + def dangerous_alias_value(node: ast.AST | None) -> str | None: + if isinstance(node, ast.Name) and node.id in dangerous_builtins: + return node.id + if isinstance(node, ast.Attribute) and node.attr in dangerous_builtins: + return node.attr + return None + for node in ast.walk(tree): if isinstance(node, ast.Name) and isinstance(node.ctx, ast.Store) and node.id in stores: stores[node.id] += 1 + if ( + isinstance(node, ast.Attribute) + and isinstance(node.ctx, (ast.Store, ast.Del)) + ): + if node.attr == "__dict__": + raise AuthorityError( + f"{path.name} mutates a module namespace beside frozen pins" + ) + if node.attr in stores: + raise AuthorityError( + f"{path.name} mutates frozen pin {node.attr} through an attribute" + ) + if isinstance(node, ast.Subscript) and isinstance(node.ctx, (ast.Store, ast.Del)): + if is_module_namespace(node.value): + raise AuthorityError( + f"{path.name} uses dynamic namespace mutation beside frozen pins" + ) + indexed_name = protected_literal(node.slice) + if indexed_name in stores: + raise AuthorityError( + f"{path.name} mutates frozen pin {indexed_name} through a subscript" + ) if isinstance(node, ast.ImportFrom) and any(alias.name == "*" for alias in node.names): raise AuthorityError(f"{path.name} uses a star import beside frozen pins") - if isinstance(node, ast.Call) and isinstance(node.func, ast.Name): - dynamic = node.func.id - if dynamic in {"globals", "locals", "eval", "setattr", "delattr", "__import__"}: + if isinstance(node, (ast.Import, ast.ImportFrom)): + bound_names = { + alias.asname or alias.name.split(".", 1)[0] + for alias in node.names + } + overlap = bound_names.intersection(stores) + if overlap: + raise AuthorityError( + f"{path.name} imports over frozen pin {sorted(overlap)[0]}" + ) + if isinstance(node, ast.ImportFrom) and node.module == "builtins": + dangerous_imports = { + alias.name for alias in node.names if alias.name in dangerous_builtins + } + if dangerous_imports: + raise AuthorityError( + f"{path.name} aliases dangerous built-in {sorted(dangerous_imports)[0]}" + ) + if isinstance(node, (ast.Assign, ast.AnnAssign, ast.NamedExpr)): + dangerous_alias = dangerous_alias_value(node.value) + if dangerous_alias is not None: + raise AuthorityError( + f"{path.name} aliases dangerous built-in {dangerous_alias}" + ) + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)) and node.name in stores: + raise AuthorityError(f"{path.name} defines over frozen pin {node.name}") + if isinstance(node, ast.Call): + dynamic = called_name(node) + if dynamic == "getattr": + attribute_name = ( + literal_value(node.args[1]) if len(node.args) >= 2 else None + ) + target = node.args[0] if node.args else None + if ( + not isinstance(attribute_name, str) + and isinstance(target, ast.Name) + and target.id in builtins_bindings + ): + raise AuthorityError( + f"{path.name} uses non-literal built-in lookup beside frozen pins" + ) + if isinstance(attribute_name, str) and ( + attribute_name in dangerous_builtins + or attribute_name in stores + or attribute_name == "__dict__" + ): + raise AuthorityError( + f"{path.name} looks up dangerous attribute {attribute_name} beside frozen pins" + ) + if isinstance(node.func, ast.Attribute) and is_module_namespace(node.func.value): + if node.func.attr in { + "update", + "setdefault", + "pop", + "popitem", + "clear", + "__setitem__", + "__delitem__", + }: + raise AuthorityError( + f"{path.name} uses dynamic module namespace mutation beside frozen pins" + ) + if dynamic in {"setattr", "delattr", "__setattr__", "__delattr__"}: + # Builtins and unbound descriptors receive the target before the + # attribute name, while a bound module method receives the name + # as its first positional argument. Check both layouts so a + # caller cannot hide a protected write behind binding semantics. + name_arguments = node.args[:2] + if any(protected_literal(argument) for argument in name_arguments): + raise AuthorityError( + f"{path.name} dynamically mutates a frozen pin" + ) + if dynamic in {"setitem", "__setitem__", "delitem", "__delitem__"}: + namespace_argument = node.args[0] if node.args else None + name_argument = node.args[1] if len(node.args) > 1 else None + if ( + (namespace_argument is not None and is_module_namespace(namespace_argument)) + or (name_argument is not None and protected_literal(name_argument)) + ): + raise AuthorityError( + f"{path.name} dynamically mutates a frozen pin" + ) + if dynamic in forbidden_dynamic_calls: raise AuthorityError( f"{path.name} uses dynamic namespace mutation beside frozen pins" ) diff --git a/yuqei-infra/tests/test_rc06_release_trust_bootstrap.py b/yuqei-infra/tests/test_rc06_release_trust_bootstrap.py index 40154e86..b5bd1ec4 100644 --- a/yuqei-infra/tests/test_rc06_release_trust_bootstrap.py +++ b/yuqei-infra/tests/test_rc06_release_trust_bootstrap.py @@ -595,10 +595,116 @@ def test_pin_planner_refuses_unknown_existing_pin_without_overwriting(tmp_path: '\nexec("TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256 = \\\"' + "f" * 64 + '\\\"")\n', "untrusted dynamic exec", ), + ( + "\nexec(compile(" + "\"TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256 = '" + + "f" * 64 + + "'\", '', 'exec'))\n", + "untrusted dynamic exec", + ), ( "\nfrom attacker_pin_side_effect import *\n", "star import", ), + ( + "\nfrom builtins import exec as _danger\n" + "_danger(\"TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256 = '" + + "f" * 64 + + "'\")\n", + "aliases dangerous built-in exec", + ), + ( + "\nimport sys as _shadow_sys\n" + "from builtins import setattr as _danger\n" + "_danger(_shadow_sys.modules[__name__], " + "'TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256', '" + + "f" * 64 + + "')\n", + "aliases dangerous built-in setattr", + ), + ( + "\n_danger = exec\n" + "_danger(\"TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256 = '" + + "f" * 64 + + "'\")\n", + "aliases dangerous built-in exec", + ), + ( + "\nimport builtins as _builtins\nimport sys as _shadow_sys\n" + "_danger = getattr(_builtins, 'exec')\n" + "if sys.argv[0] != '-c':\n" + " _danger(\"TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256 = '" + + "f" * 64 + + "'\", _shadow_sys.modules[__name__].__dict__)\n", + "looks up dangerous attribute exec", + ), + ( + "\nimport sys as _shadow_sys\n" + "if __name__ != '_yuqei_pin_runtime_validation':\n" + " _shadow_sys.modules[__name__].TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256 = '" + + "f" * 64 + + "'\n", + "mutates frozen pin.*attribute", + ), + ( + "\n_shadow = {}\n" + "_shadow['TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256'] = '" + + "f" * 64 + + "'\n", + "mutates frozen pin.*subscript", + ), + ( + "\nsetattr(object(), 'TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256', '" + + "f" * 64 + + "')\n", + "dynamically mutates a frozen pin", + ), + ( + "\nimport sys as _shadow_sys\n" + "_shadow_sys.modules[__name__].__dict__.update({" + "'TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256': '" + + "f" * 64 + + "'})\n", + "dynamic module namespace mutation", + ), + ( + "\nimport sys as _shadow_sys\n" + "getattr(_shadow_sys.modules[__name__], '__dict__').update({" + "'TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256': '" + + "f" * 64 + + "'})\n", + "dynamic module namespace mutation", + ), + ( + "\nimport sys as _shadow_sys\n" + "object.__setattr__(_shadow_sys.modules[__name__], " + "'TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256', '" + + "f" * 64 + + "')\n", + "dynamically mutates a frozen pin", + ), + ( + "\nimport sys as _shadow_sys\n" + "_shadow_sys.modules[__name__].__setattr__(" + "'TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256', '" + + "f" * 64 + + "')\n", + "dynamically mutates a frozen pin", + ), + ( + "\nimport sys as _shadow_sys\n" + "_shadow_sys.modules[__name__].__delattr__(" + "'TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256')\n", + "dynamically mutates a frozen pin", + ), + ( + "\nimport operator as _operator\nimport sys as _shadow_sys\n" + "_operator.setitem(_shadow_sys.modules[__name__].__dict__, " + "'TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256', '" + + "f" * 64 + + "')\n", + "dynamically mutates a frozen pin", + ), ), ) def test_pin_planner_rejects_duplicate_dynamic_and_import_side_effect_assignments( @@ -626,9 +732,14 @@ def test_pin_planner_isolated_runtime_detects_import_side_effect_override( rc06 = repo / "yuqei-infra/scripts/build-rc06-publication-batch.py" side_effect = rc06.parent / "attacker_pin_side_effect.py" side_effect.write_text( + "import pathlib\n" "import sys\n" - "module = sys.modules['_yuqei_pin_runtime_validation']\n" - "module.TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256 = '" + "f" * 64 + "'\n", + "target = pathlib.Path(__file__).with_name('build-rc06-publication-batch.py').resolve()\n" + "for module in tuple(sys.modules.values()):\n" + " module_file = getattr(module, '__file__', None)\n" + " if module_file and pathlib.Path(module_file).resolve() == target:\n" + " module.TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256 = '" + "f" * 64 + "'\n" + " break\n", encoding="utf-8", ) rc06.write_text( @@ -645,6 +756,58 @@ def test_pin_planner_isolated_runtime_detects_import_side_effect_override( ) +def test_pin_planner_rejects_modified_verified_module_exec_loader( + tmp_path: Path, +) -> None: + state, repo, _plan = _pin_plan_fixture(tmp_path, include_online=False) + rollback = repo / "yuqei-infra/deploy/plan-rollback-rehearsal.py" + original = rollback.read_text(encoding="utf-8") + needle = ' module = types.ModuleType(name)\n' + replacement = ( + " if sys.argv[0] != '-c':\n" + " module = sys.modules[__name__]\n" + " raw = b\"TRUSTED_APPROVAL_ROOT_PUBLIC_KEY_SHA256 = '" + + "f" * 64 + + "'\"\n" + " else:\n" + " module = types.ModuleType(name)\n" + ) + assert needle in original + rollback.write_text(original.replace(needle, replacement, 1), encoding="utf-8") + + with pytest.raises(authority.AuthorityError, match="untrusted dynamic exec"): + authority.plan_pin_updates( + repo, + state, + state / "keyrings" / "recovery-epoch-1", + epoch=1, + unprotect=_unprotect, + ) + + +def test_pin_runtime_validation_uses_a_fresh_unpredictable_module_name( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + _state, repo, _plan = _pin_plan_fixture(tmp_path, include_online=False) + rc06 = repo / "yuqei-infra/scripts/build-rc06-publication-batch.py" + raw = rc06.read_bytes() + expected = authority._validate_source_constant_contract(rc06, repo, raw) + real_run = authority.subprocess.run + module_names: list[str] = [] + + def observe_run(command, *args, **kwargs): + module_names.append(command[-1]) + return real_run(command, *args, **kwargs) + + monkeypatch.setattr(authority.subprocess, "run", observe_run) + authority._validate_isolated_runtime_constants(rc06, repo, raw, expected) + authority._validate_isolated_runtime_constants(rc06, repo, raw, expected) + assert len(module_names) == 2 + assert module_names[0] != module_names[1] + assert all(re.fullmatch(r"_[0-9a-f]{64}", name) for name in module_names) + + def test_pin_apply_rolls_back_completed_files_when_later_replace_fails( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, @@ -852,6 +1015,98 @@ def test_apply_payload_signing_is_bound_verified_atomic_and_idempotent( ) +def test_apply_challenge_is_unique_per_authorization_context_and_recovers_publish_interrupt( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + materials = _formal_apply_materials(tmp_path, monkeypatch) + common = { + "manifest_file": materials["manifest_file"], + "manifest_sha256": materials["manifest_sha"], + "approval_file": materials["approval_file"], + "approval_sha256": materials["approval_sha"], + "approval_keyring_file": materials["keyring_file"], + "approval_keyring_sha256": materials["keyring_sha"], + "approval_root_public_key_file": materials["root_file"], + "environment_id": "isolated-uat-fixture", + "unprotect": _unprotect, + } + repeated = authority.issue_apply_challenge( + materials["state"], + now=datetime(2026, 9, 27, 3, 56, tzinfo=UTC), + **common, + ) + assert repeated["status"] == "apply_challenge_already_issued_not_replaced" + assert repeated["challenge_id"] == materials["challenge"]["challenge_id"] + assert len(list((materials["state"] / "apply-challenge-contexts").glob("*.json"))) == 1 + assert len(list((materials["state"] / "apply-challenges").glob("*.json"))) == 1 + + second_payload = dict(materials["approval"]["payload"]) + second_payload["approved_at"] = "2026-09-27T03:11:00Z" + second_signature = materials["reviewer_private"].sign( + authority.batch.APPROVAL_SIGNATURE_DOMAIN + authority._canonical(second_payload) + ) + second_approval = { + "schema_version": authority.batch.APPROVAL_SCHEMA_VERSION, + "approval_id": ( + "rc06-batch-approval-" + + authority.batch._sha256_bytes( + authority._canonical(second_payload) + second_signature + )[:24] + ), + "payload": second_payload, + "signature": base64.b64encode(second_signature).decode("ascii"), + } + second_file = tmp_path / "second-context-approval.json" + second_raw = authority._canonical(second_approval) + second_file.write_bytes(second_raw) + second_common = { + **common, + "approval_file": second_file, + "approval_sha256": authority.batch._sha256_bytes(second_raw), + } + + real_publish = authority._publish_persisted_apply_challenge + interrupted = False + + def interrupt_after_context(*args, **kwargs): + nonlocal interrupted + if not interrupted: + interrupted = True + raise KeyboardInterrupt() + return real_publish(*args, **kwargs) + + monkeypatch.setattr( + authority, + "_publish_persisted_apply_challenge", + interrupt_after_context, + ) + with pytest.raises(KeyboardInterrupt): + authority.issue_apply_challenge( + materials["state"], + now=datetime(2026, 9, 27, 3, 57, tzinfo=UTC), + **second_common, + ) + assert len(list((materials["state"] / "apply-challenge-contexts").glob("*.json"))) == 2 + assert len(list((materials["state"] / "apply-challenges").glob("*.json"))) == 1 + + recovered = authority.issue_apply_challenge( + materials["state"], + now=datetime(2026, 9, 27, 3, 58, tzinfo=UTC), + **second_common, + ) + recovered_again = authority.issue_apply_challenge( + materials["state"], + now=datetime(2026, 9, 27, 3, 59, tzinfo=UTC), + **second_common, + ) + assert recovered["challenge_id"] == recovered_again["challenge_id"] + assert recovered["authorization_context_id"] == recovered_again[ + "authorization_context_id" + ] + assert len(list((materials["state"] / "apply-challenges").glob("*.json"))) == 2 + + def test_apply_payload_signing_rejects_tamper_cross_approval_and_partial_bundle( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, @@ -1690,6 +1945,8 @@ def test_pin_planner_rejects_hardlinked_source_and_recovery_refuses_unknown_byte def test_authority_launcher_uses_authority_script_and_forwards_non_secret_arguments() -> None: text = AUTHORITY_LAUNCHER.read_text(encoding="utf-8-sig") assert '"rc06-release-trust-bootstrap.py"' in text + assert '"issue-apply-challenge"' in text + assert '"recover-pin-source"' in text assert "ValueFromRemainingArguments = $true" in text assert "$authorityScript $Command @CommandArguments" in text assert "rc06-single-reviewer-signing.py" not in text