fix(release): bind RC06 apply authorization context

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"
   )]

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"
                 )

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
+            + "'\", '<pin-write>', '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