fix(identity): execute reviewed checker from sealed attempt snapshot

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a1468b-63fa-7710-85dc-8f2e16b3ebfa

sso-mfa/k8s/keycape/keycape-client.py

@@ -198,12 +198,14 @@ def deployment_bound(deployment, client):
     return bool(matches)
 
 
-def check_config(binary, config_text, env_names, policy_store):
+def check_config(binary, config_text, env_names, policy_store, *, executable_fd=None):
     """KeyCape's startup checks on one registry: names and messages only, no values."""
-    command = [str(Path(binary).resolve()), 'check-config', '--config', '/dev/stdin', '--env-names', ','.join(env_names)]
+    executable = str(Path(binary).resolve()) if executable_fd is None else f'/proc/self/fd/{executable_fd}'
+    command = [executable, 'check-config', '--config', '/dev/stdin', '--env-names', ','.join(env_names)]
     if policy_store:
         command.append('--policy-store')
-    result = subprocess.run(command, input=config_text.encode(), capture_output=True, timeout=60, env={})
+    result = subprocess.run(command, input=config_text.encode(), capture_output=True, timeout=60, env={},
+                            pass_fds=() if executable_fd is None else (executable_fd,))
     report = json.loads(result.stdout)
     require(isinstance(report, dict) and isinstance(report.get('valid'), bool)
             and report['valid'] == (result.returncode == 0), 'check_config_output_shape')
@@ -233,7 +235,9 @@ def run(args, client):
     if args.check_config:
         names = deployment_env_names(deployment)
         policy_store = 'KEYCAPE_POLICY_PATH' in names
-        checks = {stage: check_config(args.check_config, base64.b64decode(config).decode(), names, policy_store)
+        checker_options = ({'executable_fd': args.check_config_fd}
+                           if getattr(args, 'check_config_fd', None) is not None else {})
+        checks = {stage: check_config(args.check_config, base64.b64decode(config).decode(), names, policy_store, **checker_options)
                   for stage, config in (('live', before['data']['config.yaml']), ('proposed', encoded))}
     receipt = {'schema': 'netkingdom.keycape-client.v1', 'target': 'sso/keycape-config',
                'before': metadata, 'operation': args.operation, 'client_id': client['clientId'],

sso-mfa/k8s/keycape/keycape-registration-adapter.py

@@ -7,6 +7,10 @@ injected capabilities. The kernel must independently admit both evidence kinds.
 from __future__ import annotations
 
 import argparse
+from contextlib import contextmanager
+import fcntl
+import os
+import stat
 from dataclasses import asdict, replace
 import json
 import importlib.util
@@ -29,6 +33,39 @@ helper = importlib.util.module_from_spec(_spec)
 _spec.loader.exec_module(helper)
 
 
+@contextmanager
+def checker_snapshot(path, expected_digest):
+    """Execute only verified immutable bytes, independent of later path changes.
+
+    Linux sealed memfd and procfs are required. This is a per-attempt executable
+    snapshot, not native owner storage custody or process/credential isolation.
+    """
+    if not isinstance(expected_digest, str) or not DIGEST.fullmatch(expected_digest):
+        raise Refused('exact reviewed checker digest required')
+    source_fd = os.open(path, os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW | os.O_NONBLOCK)
+    with os.fdopen(source_fd, 'rb') as source:
+        metadata = os.fstat(source.fileno())
+        if not stat.S_ISREG(metadata.st_mode):
+            raise Refused('regular checker artifact required')
+        if metadata.st_size > 64 * 1024 * 1024:
+            raise Refused('checker artifact exceeds snapshot limit')
+        data = source.read(64 * 1024 * 1024 + 1)
+    if len(data) > 64 * 1024 * 1024 or digest(data) != expected_digest:
+        raise Refused('check-config binary changed or exceeds snapshot limit')
+    fd = os.memfd_create('keycape-reviewed-checker', os.MFD_CLOEXEC | os.MFD_ALLOW_SEALING)
+    try:
+        with os.fdopen(os.dup(fd), 'wb') as output:
+            output.write(data)
+        os.fchmod(fd, 0o500)
+        seals = fcntl.F_SEAL_WRITE | fcntl.F_SEAL_GROW | fcntl.F_SEAL_SHRINK | fcntl.F_SEAL_SEAL
+        fcntl.fcntl(fd, fcntl.F_ADD_SEALS, seals)
+        if fcntl.fcntl(fd, fcntl.F_GET_SEALS) != seals:
+            raise Refused('immutable checker snapshot required')
+        yield fd
+    finally:
+        os.close(fd)
+
+
 def source_digest(check_config_digest):
     """Pin adapter plus the complete local helper import closure, in filename order."""
     if not isinstance(check_config_digest, str) or not DIGEST.fullmatch(check_config_digest):
@@ -103,8 +140,6 @@ def build(*, admitted: Invocation, case_source: bytes, expected_uid: str,
             raise Refused('attempt binding differs')
         if source_digest(check_config_digest) != reviewed_source:
             raise Refused('owner source changed')
-        if digest(check_config.read_bytes()) != check_config_digest:
-            raise Refused('check-config binary changed')
 
     def execute(inv, operation, created=()):
         nonlocal creation
@@ -121,7 +156,9 @@ def build(*, admitted: Invocation, case_source: bytes, expected_uid: str,
                 args = argparse.Namespace(mode='apply', operation=operation,
                     expected_uid=uid, expected_resource_version=revision,
                     case_acceptance=case_acceptance, check_config=str(check_config))
-                receipt = helper.run(args, client)
+                with checker_snapshot(check_config, check_config_digest) as checker_fd:
+                    args.check_config_fd = checker_fd
+                    receipt = helper.run(args, client)
                 # The helper has completed CAS, whole-data readback and preservation.
                 # Never retain check-config diagnostics or arbitrary helper fields.
                 if not (receipt['schema'] == 'netkingdom.keycape-client.v1'

sso-mfa/k8s/keycape/registration-adapter.md

@@ -13,12 +13,23 @@ catalog/acceptance/run-sheet digests and target. The descriptor pins this adapte
 and its complete local helper import closure together with the exact reviewed
 `check-config` executable digest. Changing that digest changes the descriptor
 source pin and descriptor bytes, requiring independently reviewed admission.
-The fixed executable is rehashed before each attempt. The helper resolves the
+The selected artifact path is rehashed before each attempt. Before helper access,
+the adapter also reads that regular file with final-component symlinks refused,
+checks the descriptor's exact digest, and copies at most 64 MiB into a Linux
+memfd sealed against write, shrink, growth and seal removal. Both live and
+proposed checks execute this same snapshot through `/proc/self/fd`, inherited
+only by the checker subprocess with an empty environment. The descriptor still
+binds the original artifact digest and adapter/helper composition, never an
+ephemeral fd pathname. Changing the selected file after snapshot creation cannot
+change executed bytes. Snapshot/seal/exec failures close the fd, revoke the parent
+session through the existing boundary and cannot create a success receipt.
+Linux memfd seals, executable memfds and accessible procfs are required; unsupported
+runtime policy fails closed. This does not prove trusted native storage,
+protect against a compromised interpreter/kernel or supply process isolation. The helper resolves the
 selected file to an absolute executable path and supplies an empty checker
 environment, so ambient PATH and KeyCape URL overrides cannot select another
-executable or alter validation of supplied bytes. Path resolution does not
-establish immutable storage or eliminate replacement races; executable custody
-and native runtime admission remain separate requirements. The caller constructs the
+executable or alter validation of supplied bytes. Standalone helper path selection still does not establish immutable storage.
+Native executable custody and runtime admission remain separate requirements. The caller constructs the
 proposed invocation from exact source inputs, installs the fixed adapter once,
 and calls normal `Executor.admit`; independently derived admission must match.
 No callable replacement after admission is required or supported by this recipe.

sso-mfa/k8s/keycape/test_keycape_checker_snapshot.py

@@ -0,0 +1,122 @@
+"""Linux executable-byte custody fixtures; no credentials or native authority."""
+import fcntl
+import hashlib
+import mmap
+import os
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from test_keycape_registration_adapter import a
+from test_keycape_checker_environment import VALID
+
+
+class SnapshotTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.source = Path(self.temp.name) / 'checker'
+        self.source.write_bytes(b'synthetic artifact')
+        self.expected = a.digest(self.source.read_bytes())
+
+    def test_snapshot_bytes_sealed_and_descriptor_closed_on_exit(self):
+        with a.checker_snapshot(self.source, self.expected) as fd:
+            self.assertEqual(Path(f'/proc/self/fd/{fd}').read_bytes(), self.source.read_bytes())
+            for action in (lambda: os.write(fd, b'x'), lambda: os.ftruncate(fd, 0),
+                           lambda: os.ftruncate(fd, 128)):
+                with self.assertRaises(OSError):
+                    action()
+            # Even if this process restores write permission, seals forbid writes.
+            os.fchmod(fd, 0o700)
+            reopened = os.open(f'/proc/self/fd/{fd}', os.O_RDWR)
+            try:
+                with self.assertRaises(OSError):
+                    os.write(reopened, b'x')
+                with self.assertRaises(OSError):
+                    mmap.mmap(reopened, len(b'synthetic artifact'), access=mmap.ACCESS_WRITE)
+            finally:
+                os.close(reopened)
+        with self.assertRaises(OSError):
+            os.fstat(fd)
+
+    def test_hash_mismatch_refused_before_memfd(self):
+        self.source.write_bytes(b'changed')
+        with patch.object(a.os, 'memfd_create') as create:
+            with self.assertRaises(a.Refused):
+                with a.checker_snapshot(self.source, self.expected):
+                    self.fail('unexpected snapshot')
+            create.assert_not_called()
+
+    def test_source_symlink_refused(self):
+        link = self.source.with_name('link')
+        link.symlink_to(self.source)
+        with self.assertRaises(OSError):
+            with a.checker_snapshot(link, self.expected):
+                self.fail('unexpected snapshot')
+
+    def test_fifo_refused_without_blocking(self):
+        fifo = self.source.with_name('fifo')
+        os.mkfifo(fifo)
+        with self.assertRaises(a.Refused):
+            with a.checker_snapshot(fifo, self.expected):
+                self.fail('unexpected snapshot')
+
+    def test_sealing_failure_closes_snapshot(self):
+        original = a.os.memfd_create
+        opened = []
+        def create(*args):
+            fd = original(*args)
+            opened.append(fd)
+            return fd
+        with patch.object(a.os, 'memfd_create', side_effect=create), patch.object(a.fcntl, 'fcntl', side_effect=OSError('synthetic failure')):
+            with self.assertRaises(OSError):
+                with a.checker_snapshot(self.source, self.expected):
+                    self.fail('unexpected snapshot')
+        with self.assertRaises(OSError):
+            os.fstat(opened[0])
+
+    def test_seal_readback_mismatch_refused(self):
+        original = a.fcntl.fcntl
+        def invoke(fd, command, *args):
+            return 0 if command == fcntl.F_GET_SEALS else original(fd, command, *args)
+        with patch.object(a.fcntl, 'fcntl', side_effect=invoke):
+            with self.assertRaises(a.Refused):
+                with a.checker_snapshot(self.source, self.expected):
+                    self.fail('unexpected snapshot')
+
+
+class RealSnapshotTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        source = os.environ.get('KEYCAPE_TEST_CHECKER_BINARY')
+        if not source:
+            raise unittest.SkipTest('exact real checker artifact not supplied')
+        cls.data = Path(source).read_bytes()
+        cls.expected = os.environ.get('KEYCAPE_TEST_CHECKER_SHA256')
+        if cls.expected != 'sha256:' + hashlib.sha256(cls.data).hexdigest():
+            raise ValueError('real checker artifact digest differs')
+
+    def test_real_checker_executes_pinned_snapshot_after_selected_path_replaced(self):
+        with tempfile.TemporaryDirectory() as directory:
+            source = Path(directory) / 'keycape'
+            source.write_bytes(self.data)
+            source.chmod(0o700)
+            with a.checker_snapshot(source, self.expected) as fd:
+                replacement = source.with_name('replacement')
+                replacement.write_text('#!/bin/sh\nexit 27\n')
+                replacement.chmod(0o700)
+                replacement.replace(source)
+                result = a.helper.check_config(source, VALID, [], False, executable_fd=fd)
+                self.assertTrue(result['valid'])
+                invalid = VALID + 'accountPortalURL: http://invalid.example.invalid\n'
+                self.assertFalse(a.helper.check_config(source, invalid, [], False, executable_fd=fd)['valid'])
+
+    def test_real_checker_ignores_in_place_source_overwrite_after_snapshot(self):
+        with tempfile.TemporaryDirectory() as directory:
+            source = Path(directory) / 'keycape'
+            source.write_bytes(self.data)
+            with a.checker_snapshot(source, self.expected) as fd:
+                source.write_bytes(b'no executable remains at the selected path')
+                with patch.dict(os.environ, {'KEYCAPE_ACCOUNT_PORTAL_URL': 'http://polluted.example.invalid'}):
+                    self.assertTrue(a.helper.check_config(source, VALID, [], False, executable_fd=fd)['valid'])

sso-mfa/k8s/keycape/test_keycape_registration_adapter.py

@@ -3,6 +3,7 @@ from dataclasses import asdict, replace
 import copy
 import importlib.util
 import json
+import os
 from pathlib import Path
 import tempfile
 import unittest
@@ -96,6 +97,33 @@ class RegistrationTests(unittest.TestCase):
             self.assertNotIn(b'config.yaml', payload)
             self.assertNotIn(b'signing-key', payload)
 
+    def test_adapter_refuses_symlink_fifo_and_oversized_source_before_helper(self):
+        symlink = self.binary.with_name('symlink')
+        symlink.symlink_to(self.binary)
+        fifo = self.binary.with_name('fifo')
+        os.mkfifo(fifo)
+        oversized = self.binary.with_name('oversized')
+        with oversized.open('wb') as output:
+            output.truncate(64 * 1024 * 1024 + 1)
+        for source in (symlink, fifo, oversized):
+            with self.subTest(source=source.name), patch.object(a.helper, 'run') as run:
+                adapter = a.build(**dict(self.kw, check_config=source))
+                with self.assertRaises(Refused):
+                    adapter.invoke(self.inv)
+                run.assert_not_called()
+        self.assertEqual(self.revocations, [self.inv] * 3)
+        self.assertFalse(self.effects)
+        self.assertFalse(self.evidence)
+
+    def test_snapshot_unavailable_refuses_before_tooling_and_revokes_session(self):
+        adapter = a.build(**self.kw)
+        with patch.object(a.os, 'memfd_create', side_effect=OSError('synthetic policy denial')):
+            with self.assertRaises(Refused):
+                adapter.invoke(self.inv)
+        self.assertFalse(self.effects)
+        self.assertFalse(self.evidence)
+        self.assertEqual(self.revocations, [self.inv])
+
     def test_identical_client_is_no_change_and_cannot_be_removed(self):
         self.secret, _ = f.fixture(f.EXISTING + [f.FORGEJO])
         adapter = a.build(**self.kw)

workplans/NK-WP-0048-security-case-types.md

@@ -748,6 +748,14 @@ change. Platform owns structural/mint/broker/delivery evidence; Apps owns actual
 consumer login/group enforcement. Neither their assent nor native acceptance is
 supplied by this task.
 
+2026-10-16 source checkpoint: T19 below extracts the completed per-attempt
+executable snapshot implementation and reviewed fixture proof. T16 retains
+native executable storage/custody, exact owner evidence/session callbacks,
+complete descriptor composition and ordinary type/version admission. The
+successor descriptor/source pins are proposed only; the earlier descriptor
+history is preserved. T14 group provenance and T15 durable restart/recovery
+remain independent open requirements. No native acceptance is counted here.
+
 ## T17 — Build and exercise the exact offline config-checker artifact
 
 ```task
@@ -825,3 +833,39 @@ The exact prior staged/working brief bytes were restored. Final source/publicati
 projection evidence is retained in Custodian browser-checker-environment-r2/
 canonical-integration.json. These source receipts do not admit native executable
 custody, owner callbacks, acknowledged type changes or browser identities.
+
+## T19 — Bind checker execution to verified immutable attempt bytes
+
+```task
+id: NK-WP-0048-T19
+status: done
+priority: high
+```
+
+Completed 2026-10-16, independently reviewed source implementation and synthetic
+fixture proof only. The registration adapter now opens a bounded regular checker
+artifact without following its terminal symlink, verifies the reviewed digest,
+and snapshots the exact bytes into a write/grow/shrink/seal-removal-sealed Linux
+memfd before helper access. Live and proposed config checks execute that same fd
+through procfs with the existing empty environment; the fd closes on completion
+or failure. Unsupported snapshot/seal/exec policy refuses through the existing
+parent-session revocation and uncertainty boundaries. The earlier unbounded
+pre-snapshot path read was removed during independent review; actual adapter
+symlink/FIFO/oversize refusal tests verify helper access never starts.
+
+Evidence: Custodian `docs/changes/netkingdom-target-2026-10-16/browser-resume-r1/`,
+including exact base/preimages/postimages/patch, proposed descriptor, coordinator
+source review and canonical integration receipt. Independent Custodian checks
+passed 49 affected source tests and 12 actual Go checker/helper/kernel scenarios;
+the sets overlap and are not separate acceptance counts. Existing published
+Warden d97652a producer pins remain unchanged. The exact KeyCape artifact remains
+`sha256:be37738fe1c08259fb4f9ca0daf0f9301bfffc237e5bb38435525dfbf11ccff5`.
+The predecessor path-replacement refusal and the first expanded reopened-fd test
+failure are retained; the latter fixture was corrected to restore write mode
+before independently proving seals block reopened-fd writes and shared mmap.
+
+This source task adds no native owner custody, callback, evidence trust, type
+acknowledgment, credential access, consumer sign-in or rollback acceptance.
+Executable memfds, Linux seals and usable procfs are explicit runtime dependencies
+and still require native admission. This is an extracted source outcome from
+T16, not completion of T16; T12/T14/T15 and all external-owner gates remain open.