NK-WP-0058-T02 step 2: T15 and T19 exercises run in the contract CI

T19 runs the KeyCape binary from the pinned image instead of docker: tools/contract-tests/oci_extract.py
pulls the image anonymously by digest and verifies manifest and layer digests before extracting.
Both exercises record GITHUB_SHA when net-kingdom is a source tarball. PyJWT[crypto] chain pinned by
PyPI-checked hashes. 85 tests pass in a CI-like layout. rapp CI recorded (rapp 15df0e3).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 943911@bnt-lap001
Assistant-Session: fc322940-1c8e-4bed-ab39-86cec99f23a0

bounded-executor/exercise_approval_identity.py

@@ -80,7 +80,9 @@ def run(args, timeout=60):
 
 
 def commit(path: pathlib.Path) -> str:
-    return run(["git", "-C", str(path), "rev-parse", "HEAD"])
+    """The checkout's commit. In CI net-kingdom is a source tarball: record the commit CI built (GITHUB_SHA)."""
+    done = subprocess.run(["git", "-C", str(path), "rev-parse", "HEAD"], capture_output=True, text=True)
+    return done.stdout.strip() if done.returncode == 0 else os.environ.get("GITHUB_SHA", "not-a-git-checkout")
 
 
 def private_write(path: pathlib.Path, value: str) -> None:
@@ -145,6 +147,7 @@ def exercise(args) -> dict:
         "target": "disposable local processes; synthetic keys and credentials",
         "started_at": dt.datetime.now(dt.UTC).isoformat(timespec="seconds"),
         "keycape_image": KEYCAPE_IMAGE,
+        "keycape_runtime": "binary from the pinned image" if args.keycape_binary else "container",
         "issuer_contract_commit": commit(args.keycape_source),  # key-cape; a "key*" name trips the secret scanner
         "approval_engine_commit": commit(args.approval_engine_source),
         "net_kingdom_commit": commit(REPO),
@@ -176,6 +179,8 @@ def exercise(args) -> dict:
         operator = [c for c in examples if c["clientId"] == OPERATOR]
         assert len(operator) == 1
         config.update(issuer=issuer, clients=[client, operator[0]])
+        if args.keycape_binary:  # no container: the binary reads its files from the temporary folder directly
+            config.update(port=port, privateKeyPem=str(root / "key.pem"))
         config["authelia"]["issuer"] = "https://synthetic-upstream.invalid"
         private_write(root / "config.yaml", yaml.safe_dump(config, sort_keys=False))
         values = {c["clientId"]: secrets.token_urlsafe(48) for c in (client, operator[0])}
@@ -183,13 +188,20 @@ def exercise(args) -> dict:
             c["secretRef"].removeprefix("env:") + "=" + values[c["clientId"]] + "\n" for c in (client, operator[0])))
         os.chmod(root, 0o755)  # the container user reads the bind mount
         container = "nk56-t19-keycape-" + secrets.token_hex(5)
-        created = engine = api_server = None
+        created = engine = api_server = process = None
         try:
-            run(["docker", "run", "-d", "--rm", "--pull=never", "--name", container, "--user", str(os.getuid()),
-                 "--read-only", "--cap-drop=ALL", "--security-opt=no-new-privileges",
-                 "--publish", f"127.0.0.1:{port}:8080", "--env-file", str(root / "env"),
-                 "--mount", f"type=bind,source={temporary},target=/etc/keycape,readonly", KEYCAPE_IMAGE])
-            created = True
+            if args.keycape_binary:
+                # the binary extracted from KEYCAPE_IMAGE by tools/contract-tests/oci_extract.py (digest-verified)
+                env = {"PATH": "/usr/bin:/bin", "KEYCAPE_CONFIG": str(root / "config.yaml"),
+                       **{c["secretRef"].removeprefix("env:"): values[c["clientId"]] for c in (client, operator[0])}}
+                process = subprocess.Popen([str(args.keycape_binary)], env=env, cwd=temporary,
+                                           stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+            else:
+                run(["docker", "run", "-d", "--rm", "--pull=never", "--name", container, "--user", str(os.getuid()),
+                     "--read-only", "--cap-drop=ALL", "--security-opt=no-new-privileges",
+                     "--publish", f"127.0.0.1:{port}:8080", "--env-file", str(root / "env"),
+                     "--mount", f"type=bind,source={temporary},target=/etc/keycape,readonly", KEYCAPE_IMAGE])
+                created = True
             for _ in range(80):
                 try:
                     with urllib.request.urlopen(issuer + "/jwks", timeout=1) as response:
@@ -353,6 +365,9 @@ def exercise(args) -> dict:
                 engine.close()
             if created:
                 run(["docker", "stop", "--time=5", container])
+            if process:
+                process.terminate()
+                process.wait(timeout=10)
 
     receipt.update(status="passed", cleanup_complete=True,
                    finished_at=dt.datetime.now(dt.UTC).isoformat(timespec="seconds"))
@@ -368,6 +383,8 @@ def main(argv=None) -> int:
     parser.add_argument("--keycape-source", type=pathlib.Path, default=home / "key-cape")
     parser.add_argument("--approval-engine-source", type=pathlib.Path, default=home / "approval-engine")
     parser.add_argument("--receipt", required=True, type=pathlib.Path)
+    parser.add_argument("--keycape-binary", type=pathlib.Path,
+                        help="run this KeyCape binary (extracted from the pinned image) instead of docker (CI)")
     parser.add_argument("--trace", action="store_true", help="print the failing assertion's location (no values)")
     args = parser.parse_args(argv)
     if args.receipt.exists():

bounded-executor/exercise_audit_sender.py

@@ -59,8 +59,9 @@ C = _load("clients", HERE / "clients.py")
 
 
 def commit(path: pathlib.Path) -> str:
-    return subprocess.run(["git", "-C", str(path), "rev-parse", "HEAD"], capture_output=True, text=True,
-                          check=True).stdout.strip()
+    """The checkout's commit. In CI net-kingdom is a source tarball: record the commit CI built (GITHUB_SHA)."""
+    done = subprocess.run(["git", "-C", str(path), "rev-parse", "HEAD"], capture_output=True, text=True)
+    return done.stdout.strip() if done.returncode == 0 else os.environ.get("GITHUB_SHA", "not-a-git-checkout")
 
 
 def private_write(path: pathlib.Path, value: str) -> None:

docs/bounded-execution-contract-table.md

@@ -505,7 +505,7 @@ The agent's default context can read NetworkPolicies in `sso`, `approval-engine`
 | railiance-platform | `a2105e34` | open |
 | key-cape | `ee1e5bb5` | open |
 
-- **Open, founder:** rapp-bounded-executor is private, so net-kingdom CI cannot fetch it without a credential, and
-  the rapp has no CI of its own. Its packaging tests (E8 ceiling, E9 egress, R2 config and exit codes) run only
-  locally today. The options are a rapp CI workflow, which needs the runner to read a private repository, or
-  making the rapp public.
+- **rapp-bounded-executor (private):** it tests itself in its own CI since 2026-11-06 (founder choice, rapp
+  `15df0e3`, `.forgejo/workflows/packaging.yaml`). The run fetches its own tarball with the run's automatic
+  repository-scoped token, clones net-kingdom and railiance-platform anonymously at `main`, and fails on any skip.
+  That covers E8 (ceiling), E9 (egress) and R2 (config, exit codes), on push and nightly.

tools/contract-tests/oci_extract.py

@@ -0,0 +1,83 @@
+#!/usr/bin/env python3
+"""Extract one file from a digest-pinned public OCI image without docker (NK-WP-0058-T02). Read-only.
+
+  python3 tools/contract-tests/oci_extract.py --image forgejo.coulomb.social/coulomb/key-cape@sha256:<64 hex> \
+      --file keycape --out DIR/keycape
+
+The manifest must hash to the pinned digest, and every layer read must hash to its manifest digest, so the file
+is exactly the one in the pinned image. The pull token is the registry's anonymous one; no credential is used.
+The newest layer holding the file wins, as in the image's own filesystem. Exit 0 when the file is written.
+"""
+import argparse
+import hashlib
+import io
+import json
+import re
+import sys
+import tarfile
+import urllib.parse
+import urllib.request
+from pathlib import Path
+
+ACCEPT = ', '.join(('application/vnd.oci.image.manifest.v1+json',
+                   'application/vnd.docker.distribution.manifest.v2+json'))
+
+
+class ExtractError(Exception):
+    pass
+
+
+def get(url, token=None, accept=None, opener=urllib.request.urlopen):
+    headers = {**({'Authorization': 'Bearer ' + token} if token else {}), **({'Accept': accept} if accept else {})}
+    with opener(urllib.request.Request(url, headers=headers), timeout=120) as response:
+        return response.read()
+
+
+def pull_token(host, repository, opener=urllib.request.urlopen):
+    query = urllib.parse.urlencode({'service': 'container_registry', 'scope': f'repository:{repository}:pull'})
+    return json.loads(get(f'https://{host}/v2/token?{query}', opener=opener))['token']
+
+
+def extract(image, wanted, out, opener=urllib.request.urlopen):
+    match = re.fullmatch(r'([^/]+)/(.+)@(sha256:[0-9a-f]{64})', image)
+    if not match:
+        raise ExtractError('image must be host/repository@sha256:<64 hex>')
+    host, repository, digest = match.groups()
+    token = pull_token(host, repository, opener)
+    raw = get(f'https://{host}/v2/{repository}/manifests/{digest}', token, ACCEPT, opener)
+    if 'sha256:' + hashlib.sha256(raw).hexdigest() != digest:
+        raise ExtractError('manifest_digest_mismatch')
+    manifest = json.loads(raw)
+    if 'layers' not in manifest:
+        raise ExtractError('not_a_single_platform_manifest')
+    for layer in reversed(manifest['layers']):
+        blob = get(f"https://{host}/v2/{repository}/blobs/{layer['digest']}", token, opener=opener)
+        if 'sha256:' + hashlib.sha256(blob).hexdigest() != layer['digest']:
+            raise ExtractError('layer_digest_mismatch')
+        with tarfile.open(fileobj=io.BytesIO(blob), mode='r:*') as archive:
+            for member in archive.getmembers():
+                if member.isfile() and member.name.lstrip('./') == wanted.lstrip('/'):
+                    out.parent.mkdir(parents=True, exist_ok=True)
+                    out.write_bytes(archive.extractfile(member).read())
+                    out.chmod(0o755)
+                    return {'image': digest, 'layer': layer['digest'], 'file': wanted,
+                            'sha256': hashlib.sha256(out.read_bytes()).hexdigest()}
+    raise ExtractError('file_not_in_image')
+
+
+def main(argv=None):
+    p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    p.add_argument('--image', required=True)
+    p.add_argument('--file', required=True)
+    p.add_argument('--out', required=True, type=Path)
+    a = p.parse_args(argv)
+    try:
+        print(json.dumps(extract(a.image, a.file, a.out)))
+    except ExtractError as error:
+        print(json.dumps({'status': 'refused', 'reason': str(error)}))
+        return 1
+    return 0
+
+
+if __name__ == '__main__':
+    sys.exit(main())

tools/contract-tests/requirements.txt

@@ -8,3 +8,17 @@ pluggy==1.6.0 \
     --hash=sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746
 pytest==8.3.5 \
     --hash=sha256:c69214aa47deac29fad6c2a4f590b9c4a9fdb16a403176fe154b79c0b4d4d820
+# T19 exercise (approval-engine's JWT authenticator against KeyCape): PyJWT[crypto] and its chain, Linux x86_64 wheels.
+pyjwt==2.15.1 \
+    --hash=sha256:42d59d631f7768a1028a64c7ff581a9bf7519804daf91fc5b6c56e30eec5e193
+cryptography==50.0.2 \
+    --hash=sha256:630ebfea3bf689d075f82316324ff7433dc447fe6bc1bfc76524b74b4a9567d2 \
+    --hash=sha256:4061c0079120205fb760c58acab6443e217307dcf05e3702cf970e0689972856 \
+    --hash=sha256:9dab55f57c74c3cad24c323bacbbd04be4705ba6eb0d92e920b1fc4837ed5079 \
+    --hash=sha256:ee247f5c245c9a2fe7c8e2214e295918838e44e00a45a6718451e4004219e767 \
+    --hash=sha256:f21e8a22c8605750c7af886bab299a363721264061b4ac0a30efb73cfd58efc5 \
+    --hash=sha256:1981f1db4630889b9ef7803fadef12b056f428cb6b85c27ba57b774793b6093c
+cffi==2.1.1 \
+    --hash=sha256:c1453022f490d2459a11819d83ad1d586e9ff65a12ac3e705ffebd46d3685dcf
+pycparser==3.12 \
+    --hash=sha256:a076f3f489b39b43ed13563a1dafe42447445bec2e294154ec5c57fcdda4c85e

tools/contract-tests/run.sh

@@ -14,6 +14,7 @@ PYTHONDONTWRITEBYTECODE=1 python3 -m pytest -q -rs -p no:cacheprovider \
   tools/bounded-execution-run/tests \
   tools/execution-attribution-receipt/tests \
   sso-mfa/k8s/bounded-executor \
+  tools/contract-tests/tests \
   "$@" | tee "$out"
 if grep -q '^SKIPPED' "$out"; then
   echo "contract-test: skipped tests above; every contract must run (are the owner checkouts present?)" >&2

tools/contract-tests/tests/test_exercises.py

@@ -0,0 +1,61 @@
+"""The NK-WP-0056 exercises as contract tests (NK-WP-0058-T02): rows E2, E3 and E6 of the contract table.
+
+T15: the executor's audit sender against audit-core's real ingestion and sender scope.
+T19: the executor's KeyCape client and approval-engine carrier against approval-engine's real source and the KeyCape
+binary from the pinned image (extracted by digest, no docker). Both use synthetic credentials and leave nothing
+behind; locally the sibling checkouts, in CI the fetched owners. A missing sibling skips, which fails run.sh.
+"""
+import json
+import re
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+REPO = Path(__file__).resolve().parents[3]
+SIBLINGS = REPO.parent
+EXECUTOR = REPO / 'bounded-executor'
+
+
+def need(*owners):
+    missing = [o for o in owners if not (SIBLINGS / o).is_dir()]
+    if missing:
+        pytest.skip('owner checkout missing: ' + ', '.join(missing))
+
+
+def run(argv):
+    return subprocess.run([sys.executable, *map(str, argv)], capture_output=True, text=True, timeout=600)
+
+
+def test_t15_audit_sender_against_audit_core(tmp_path):
+    need('audit-core')
+    receipt = tmp_path / 't15.json'
+    done = run([EXECUTOR / 'exercise_audit_sender.py', '--receipt', receipt, '--audit-core-source',
+                SIBLINGS / 'audit-core'])
+    assert done.returncode == 0, done.stderr[-400:]
+    checks = json.loads(receipt.read_text())['checks']
+    assert checks and all(v is True for v in checks.values()), checks
+
+
+def keycape_image():
+    """The pinned image the exercise names (KEYCAPE_IMAGE), read from its source without importing it."""
+    text = (EXECUTOR / 'exercise_approval_identity.py').read_text()
+    parts = re.search(r'KEYCAPE_IMAGE = \("([^"]+)"\s*"([^"]+)"\)', text)
+    return parts.group(1) + parts.group(2)
+
+
+def test_t19_keycape_client_and_approval_engine(tmp_path):
+    need('approval-engine', 'key-cape')
+    binary = tmp_path / 'keycape'
+    extracted = run([REPO / 'tools/contract-tests/oci_extract.py', '--image', keycape_image(), '--file', 'keycape',
+                     '--out', binary])
+    assert extracted.returncode == 0, extracted.stdout
+    receipt = tmp_path / 't19.json'
+    done = run([EXECUTOR / 'exercise_approval_identity.py', '--receipt', receipt, '--keycape-binary', binary,
+                '--keycape-source', SIBLINGS / 'key-cape', '--approval-engine-source', SIBLINGS / 'approval-engine',
+                '--trace'])
+    assert done.returncode == 0, done.stderr[-400:]
+    r = json.loads(receipt.read_text())
+    assert r['status'] == 'passed' and r['keycape_runtime'] == 'binary from the pinned image'
+    assert r['checks'] and all(v is True for v in r['checks'].values()), r['checks']