Add the backup reader boundary probe and record founder approval of MAS node execution.

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

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 4541@bnt-lap001
Assistant-Session: 2099b3aa-d7d9-49a9-ab4e-fa8b7d0e39e8

WORK-RECORDS.md

@@ -237,7 +237,7 @@
 | task | RPF-WP-0069-T08 | done | — | workplans/RPF-WP-0069-bounded-executor-lanes.md |
 | task | RPF-WP-0069-T09 | wait | — | workplans/RPF-WP-0069-bounded-executor-lanes.md |
 | task | RPF-WP-0070-T01 | done | — | workplans/RPF-WP-0070-mas-evidence-recovery.md |
-| task | RPF-WP-0070-T02 | todo | — | workplans/RPF-WP-0070-mas-evidence-recovery.md |
+| task | RPF-WP-0070-T02 | progress | — | workplans/RPF-WP-0070-mas-evidence-recovery.md |
 | task | RPF-WP-0070-T03 | todo | — | workplans/RPF-WP-0070-mas-evidence-recovery.md |
 | task | RPF-WP-0070-T04 | todo | — | workplans/RPF-WP-0070-mas-evidence-recovery.md |
 | task | RPF-WP-0070-T05 | todo | — | workplans/RPF-WP-0070-mas-evidence-recovery.md |

scripts/backup_reader_boundary_probe.py

@@ -0,0 +1,307 @@
+#!/usr/bin/env python3
+"""Prove one backup reader's OpenBao boundary on railiance01 before delivery admission.
+
+Run as root on railiance01. Stdlib only and self-contained, so it can be piped:
+
+    ssh railiance01 'sudo python3 - --application mas --expected-kv-version 1' \
+        < scripts/backup_reader_boundary_probe.py
+
+Positive: the dedicated ServiceAccount logs in with audience openbao, holds only
+its one reader policy within the CCR bounds, and reads exactly the expected KV
+version with exactly the two pair fields.
+Negative: a wrong audience and a foreign role cannot log in; sibling data,
+metadata, parent LIST, a KV write and an administration endpoint are denied.
+The write probe sends CAS=0 against an existing path, so it can never replace
+the pair even if a policy were wrong.
+
+Prints one JSON receipt. Values, tokens and JWTs never leave process memory;
+the receipt holds field names, HTTP statuses, versions and bounds only.
+Exit 0 only when every check passes.
+"""
+
+from __future__ import annotations
+
+import argparse
+import base64
+import datetime as dt
+import ipaddress
+import json
+import re
+import subprocess
+import sys
+import time
+import urllib.error
+import urllib.request
+
+CLUSTER_UID = "a553c742-0115-43d4-99a4-a5ca56fe0786"
+SA = "backup-primary-delivery"
+FIELDS = ["AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY"]
+APPLICATIONS = ("coulomb-social", "email-connect", "mas")
+KUBE_ENV = {
+    "PATH": "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin",
+    "KUBECONFIG": "/etc/rancher/k3s/k3s.yaml",
+}
+
+
+class ProbeError(Exception):
+    """Fixed diagnostic codes only."""
+
+
+def guard(condition, code):
+    if not condition:
+        raise ProbeError(code)
+
+
+def policy(application):
+    return f"workload-kv-read-{application}-backup-primary"
+
+
+def role(application):
+    return f"{application}-backup-primary-workload-kv-read"
+
+
+def data_path(application):
+    return f"platform/data/workloads/{application}/backup-primary"
+
+
+class Native:
+    """kubectl and OpenBao access on the node. HTTP errors return their status."""
+
+    def __init__(self):
+        self.address = None
+        self.opener = urllib.request.build_opener(urllib.request.ProxyHandler({}), _NoRedirect())
+
+    def kube(self, args):
+        try:
+            result = subprocess.run(
+                ["kubectl", "--request-timeout=20s", *args],
+                capture_output=True, check=False, timeout=30, env=KUBE_ENV,
+            )
+        except (OSError, subprocess.SubprocessError):
+            raise ProbeError("kubernetes_command_failed") from None
+        guard(result.returncode == 0 and len(result.stdout) <= 1048576, "kubernetes_command_failed")
+        return result.stdout
+
+    def prepare(self, application):
+        ns = json.loads(self.kube(["get", "namespace", "kube-system", "-o", "json"]))
+        guard(ns.get("metadata", {}).get("uid") == CLUSTER_UID, "cluster_identity_mismatch")
+        svc = json.loads(self.kube(["-n", "openbao", "get", "service", "openbao", "-o", "json"]))
+        addr = ipaddress.ip_address(svc.get("spec", {}).get("clusterIP", ""))
+        guard(
+            addr in ipaddress.ip_network("10.43.0.0/16")
+            and any(p.get("port") == 8200 for p in svc["spec"].get("ports", [])),
+            "openbao_service_mismatch",
+        )
+        self.address = f"http://{addr}:8200/v1/"
+        sa = json.loads(self.kube(["-n", application, "get", "serviceaccount", SA, "-o", "json"]))
+        guard(sa.get("automountServiceAccountToken") is False, "non_mounting_service_account_required")
+        return sa.get("metadata", {}).get("uid")
+
+    def jwt(self, application, audience):
+        token = self.kube([
+            "-n", application, "create", "token", SA, f"--audience={audience}", "--duration=10m",
+        ]).decode().strip()
+        guard(len(token.split(".")) == 3 and len(token) < 16384, "bounded_jwt_required")
+        return token
+
+    def http(self, method, path, payload=None, token=None):
+        headers = {"Content-Type": "application/json"}
+        if token is not None:
+            headers["X-Vault-Token"] = token
+        body = None if payload is None else json.dumps(payload).encode()
+        request = urllib.request.Request(self.address + path, data=body, headers=headers, method=method)
+        try:
+            with self.opener.open(request, timeout=20) as response:
+                raw = response.read(1048577)
+                guard(len(raw) <= 1048576, "openbao_response_too_large")
+                return response.status, ({} if not raw else json.loads(raw))
+        except urllib.error.HTTPError as exc:
+            return exc.code, {}
+        except (urllib.error.URLError, OSError):
+            raise ProbeError("openbao_request_failed") from None
+
+
+class _NoRedirect(urllib.request.HTTPRedirectHandler):
+    def redirect_request(self, *args, **kwargs):
+        return None
+
+
+def jwt_claims(token):
+    piece = token.split(".")[1]
+    return json.loads(base64.urlsafe_b64decode(piece + "=" * (-len(piece) % 4)))
+
+
+def _login(native, application, login_role, audience="openbao"):
+    status, body = native.http(
+        "POST", "auth/kubernetes/login", {"role": login_role, "jwt": native.jwt(application, audience)}
+    )
+    auth = body.get("auth") if isinstance(body, dict) else None
+    token = auth.get("client_token") if isinstance(auth, dict) else None
+    return status, auth or {}, token if isinstance(token, str) and token else None
+
+
+def _revoke(native, token):
+    status, _ = native.http("POST", "auth/token/revoke-self", {}, token=token)
+    return status in (200, 204)
+
+
+def _denials(native, token, probes):
+    """Run up to three denied probes on one four-use token, then revoke it."""
+    statuses = {}
+    try:
+        for name, method, path, payload in probes:
+            statuses[name], _ = native.http(method, path, payload, token=token)
+    finally:
+        revoked = _revoke(native, token)
+    return statuses, revoked
+
+
+def probe(native, application, expected_version, sa_uid=None):
+    guard(application in APPLICATIONS, "application_refused")
+    guard(type(expected_version) is int and expected_version > 0, "expected_version_required")
+    expected_policy = policy(application)
+    receipt = {
+        "schema": "platform.backup-reader-boundary.v1",
+        "application": application,
+        "cluster_uid": CLUSTER_UID,
+        "service_account": f"{application}/{SA}",
+        "role": role(application),
+        "policy": expected_policy,
+        "kv_path": data_path(application),
+        "expected_kv_version": expected_version,
+        "values_recorded": False,
+    }
+    if sa_uid is None:
+        sa_uid = native.prepare(application)
+    claims = jwt_claims(native.jwt(application, "openbao"))
+    bound = claims.get("kubernetes.io", {}).get("serviceaccount", {})
+    receipt["jwt_binding"] = (
+        claims.get("sub") == f"system:serviceaccount:{application}:{SA}"
+        and claims.get("aud") == ["openbao"]
+        and bool(bound.get("uid"))
+        and bound.get("uid") == sa_uid
+        and 0 < claims.get("exp", 0) - time.time() <= 660
+    )
+
+    # Positive: one token for lookup, the read, and revoke.
+    status, auth, token = _login(native, application, role(application))
+    positive = {"login_status": status}
+    if token:
+        try:
+            positive["login_policies"] = auth.get("policies")
+            positive["lease_duration"] = auth.get("lease_duration")
+            _, info = native.http("GET", "auth/token/lookup-self", token=token)
+            data = info.get("data", {}) if isinstance(info, dict) else {}
+            positive["token"] = {
+                "policies": data.get("policies"),
+                "identity_policies": data.get("identity_policies") or [],
+                "ttl": data.get("ttl"),
+                "explicit_max_ttl": data.get("explicit_max_ttl"),
+                "period": data.get("period") or 0,
+                "num_uses": data.get("num_uses"),
+            }
+            read_status, record = native.http("GET", data_path(application), token=token)
+            payload = record.get("data", {}) if isinstance(record, dict) else {}
+            metadata = payload.get("metadata", {}) if isinstance(payload, dict) else {}
+            values = payload.get("data", {}) if isinstance(payload, dict) else {}
+            values = values if isinstance(values, dict) else {}
+            positive["read_status"] = read_status
+            positive["kv_version"] = metadata.get("version")
+            positive["kv_destroyed"] = metadata.get("destroyed")
+            positive["kv_deleted"] = bool(metadata.get("deletion_time"))
+            positive["field_names"] = sorted(values)
+            positive["encoding_ok"] = all(
+                isinstance(values.get(name), str)
+                and re.fullmatch(r"[A-Za-z0-9/+=_.-]{8,256}", values[name]) is not None
+                for name in FIELDS
+            )
+            positive["fields_distinct"] = values.get(FIELDS[0]) != values.get(FIELDS[1])
+            del values, payload, record
+        finally:
+            positive["token_revoked"] = _revoke(native, token)
+    token_view = positive.get("token", {})
+    positive["passed"] = (
+        status == 200
+        and positive.get("login_policies") == [expected_policy]
+        and 0 < (positive.get("lease_duration") or 0) <= 300
+        and token_view.get("policies") == [expected_policy]
+        and token_view.get("identity_policies") == []
+        and 0 < (token_view.get("ttl") or 0) <= 300
+        and 0 < (token_view.get("explicit_max_ttl") or 0) <= 300
+        and token_view.get("period") == 0
+        and 0 < (token_view.get("num_uses") or 0) <= 4
+        and positive.get("read_status") == 200
+        and positive.get("kv_version") == expected_version
+        and positive.get("kv_destroyed") is False
+        and positive.get("kv_deleted") is False
+        and positive.get("field_names") == FIELDS
+        and positive.get("encoding_ok") is True
+        and positive.get("fields_distinct") is True
+        and positive.get("token_revoked") is True
+    )
+    receipt["positive"] = positive
+
+    negative = {}
+    # Login boundary: a wrong audience, and this account against a sibling role.
+    status, _, stray = _login(native, application, role(application), audience="not-openbao")
+    negative["wrong_audience_login"] = status
+    if stray:
+        _revoke(native, stray)
+    sibling = next(item for item in APPLICATIONS if item != application)
+    status, _, stray = _login(native, application, role(sibling))
+    negative["foreign_role_login"] = status
+    if stray:
+        _revoke(native, stray)
+
+    others = [item for item in APPLICATIONS if item != application]
+    status, _, token = _login(native, application, role(application))
+    guard(status == 200 and token, "negative_login_failed")
+    first, revoked_first = _denials(native, token, [
+        ("sibling_data_read", "GET", data_path(others[0]), None),
+        ("metadata_read", "GET", f"platform/metadata/workloads/{application}/backup-primary", None),
+        ("parent_list", "LIST", f"platform/metadata/workloads/{application}", None),
+    ])
+    status, _, token = _login(native, application, role(application))
+    guard(status == 200 and token, "negative_login_failed")
+    second, revoked_second = _denials(native, token, [
+        # CAS=0 on an existing path cannot replace the pair even if allowed.
+        ("kv_write", "POST", data_path(application), {"options": {"cas": 0}, "data": {"probe": "x"}}),
+        ("second_sibling_data_read", "GET", data_path(others[1]), None),
+        ("admin_policy_list", "LIST", "sys/policies/acl", None),
+    ])
+    negative.update(first)
+    negative.update(second)
+    negative["tokens_revoked"] = revoked_first and revoked_second
+    login_denied = all(negative[name] in (400, 403) for name in ("wrong_audience_login", "foreign_role_login"))
+    api_denied = all(
+        negative[name] == 403
+        for name in (
+            "sibling_data_read", "metadata_read", "parent_list",
+            "kv_write", "second_sibling_data_read", "admin_policy_list",
+        )
+    )
+    negative["passed"] = login_denied and api_denied and negative["tokens_revoked"]
+    receipt["negative"] = negative
+    receipt["passed"] = receipt["jwt_binding"] and positive["passed"] and negative["passed"]
+    receipt["checked_at"] = dt.datetime.now(dt.timezone.utc).isoformat()
+    return receipt
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    parser.add_argument("--application", choices=APPLICATIONS, required=True)
+    parser.add_argument("--expected-kv-version", type=int, required=True)
+    args = parser.parse_args(argv)
+    try:
+        receipt = probe(Native(), args.application, args.expected_kv_version)
+    except ProbeError as exc:
+        receipt = {"status": "failed", "code": str(exc), "values_recorded": False}
+    except Exception as exc:  # noqa: BLE001 - never print backend bodies
+        receipt = {"status": "failed", "code": "contained_probe_failure",
+                   "error_type": type(exc).__name__, "values_recorded": False}
+    print(json.dumps(receipt, indent=2, sort_keys=True))
+    return 0 if receipt.get("passed") is True else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/test_backup_reader_boundary_probe.py

@@ -0,0 +1,95 @@
+"""Reader boundary probe: passes only on an exact reader and never records values."""
+
+import base64
+import importlib.util
+import json
+import time
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+spec = importlib.util.spec_from_file_location("boundary", ROOT / "scripts/backup_reader_boundary_probe.py")
+probe = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(probe)
+
+SECRET = "supersecretvalue12345"
+ACCESS = "SCWACCESSKEY12345678"
+
+
+def _jwt(app, aud, uid="sa-uid"):
+    claims = {"sub": f"system:serviceaccount:{app}:{probe.SA}", "aud": [aud],
+              "exp": int(time.time()) + 600, "kubernetes.io": {"serviceaccount": {"uid": uid}}}
+    body = base64.urlsafe_b64encode(json.dumps(claims).encode()).decode().rstrip("=")
+    return f"h.{body}.s"
+
+
+class FakeBao:
+    def __init__(self, app="mas", version=1, allow_write=False, extra_policy=False, allow_foreign_role=False):
+        self.app, self.version = app, version
+        self.allow_write, self.extra_policy, self.allow_foreign_role = allow_write, extra_policy, allow_foreign_role
+        self.tokens, self.writes = {}, []
+
+    def jwt(self, application, audience):
+        return _jwt(application, audience)
+
+    def http(self, method, path, payload=None, token=None):
+        if path == "auth/kubernetes/login":
+            claims = probe.jwt_claims(payload["jwt"])
+            ok_role = payload["role"] == probe.role(self.app) or self.allow_foreign_role
+            if claims["aud"] != ["openbao"] or not ok_role:
+                return 403, {}
+            tok = f"t{len(self.tokens)}"
+            policies = [probe.policy(self.app)] + (["default"] if self.extra_policy else [])
+            self.tokens[tok] = {"uses": 4, "policies": policies}
+            return 200, {"auth": {"client_token": tok, "policies": policies, "lease_duration": 300}}
+        state = self.tokens.get(token)
+        if not state or state["uses"] <= 0:
+            return 403, {}
+        state["uses"] -= 1
+        if path == "auth/token/revoke-self":
+            del self.tokens[token]
+            return 204, {}
+        if path == "auth/token/lookup-self":
+            return 200, {"data": {"policies": state["policies"], "identity_policies": None, "ttl": 299,
+                                  "explicit_max_ttl": 300, "period": 0, "num_uses": state["uses"]}}
+        if method == "GET" and path == probe.data_path(self.app):
+            return 200, {"data": {"data": {"AWS_ACCESS_KEY_ID": ACCESS, "AWS_SECRET_ACCESS_KEY": SECRET},
+                                  "metadata": {"version": self.version, "destroyed": False, "deletion_time": ""}}}
+        if method == "POST" and path == probe.data_path(self.app) and self.allow_write:
+            self.writes.append(payload)
+            return 400, {}  # CAS mismatch: the path already exists
+        return 403, {}
+
+
+def run(fake, version=1):
+    return probe.probe(fake, "mas", version, sa_uid="sa-uid")
+
+
+def test_exact_reader_passes_and_records_no_values():
+    fake = FakeBao()
+    receipt = run(fake)
+    assert receipt["passed"] is True, receipt
+    assert receipt["positive"]["kv_version"] == 1
+    assert receipt["positive"]["field_names"] == probe.FIELDS
+    text = json.dumps(receipt)
+    assert SECRET not in text and ACCESS not in text and "t0" not in text
+    assert fake.tokens == {}  # every issued token revoked
+
+
+def test_version_mismatch_fails():
+    assert run(FakeBao(version=2))["passed"] is False
+
+
+def test_extra_policy_fails():
+    assert run(FakeBao(extra_policy=True))["positive"]["passed"] is False
+
+
+def test_write_not_denied_fails_and_cas0_payload_cannot_replace():
+    fake = FakeBao(allow_write=True)
+    receipt = run(fake)
+    assert receipt["negative"]["kv_write"] == 400 and receipt["passed"] is False
+    assert fake.writes == [{"options": {"cas": 0}, "data": {"probe": "x"}}]
+
+
+def test_foreign_role_login_accepted_fails():
+    receipt = run(FakeBao(allow_foreign_role=True))
+    assert receipt["negative"]["foreign_role_login"] == 200 and receipt["passed"] is False

workplans/RPF-WP-0070-mas-evidence-recovery.md

@@ -140,11 +140,11 @@ was not read or copied. Delivery admission stays false, and
 
 ```task
 id: RPF-WP-0070-T02
-status: todo
+status: progress
 priority: medium
 needs_human: true
 depends_on: [RPF-WP-0070-T01]
-intervention_note: "Attended: set delivery admission for CCR-2026-0062 and run the node-file delivery after T01 records the pair."
+intervention_note: "Founder approved node execution for CCR-2026-0062 on 2026-11-03. Next: run scripts/backup_reader_boundary_probe.py --application mas --expected-kv-version 1 as root on railiance01 and record the receipt. Then set the admission flags, export the CCR and pin its SHA256, and run the node delivery."
 state_hub_task_id: "c8a9ba31-9fb6-5412-90b7-dea658943053"
 ```
 
@@ -156,6 +156,18 @@ pair, through ServiceAccount `backup-primary-delivery`. Record a
 metadata-only node check. Never copy the supplier key onto the workload
 path, and give no TEM or SMTP credential to the node file.
 
+2026-11-03: the founder approved node execution for CCR-2026-0062 in an
+attended session. `scripts/backup_reader_boundary_probe.py` is the reader
+boundary check. It is a self-contained stdlib script that runs as root on
+railiance01. The positive check: the ServiceAccount logs in with audience
+`openbao`, holds only the one reader policy within the CCR bounds, and
+reads exactly the expected KV version with exactly the two fields. The
+negative checks: login fails with a wrong audience and with a sibling role,
+and these are refused with 403: sibling data reads, a metadata read, the
+parent LIST, a KV write, and `sys/policies/acl`. The write probe uses CAS=0
+on the existing path, so it cannot replace the pair. Every token is revoked,
+and the receipt holds statuses and names only.
+
 ## Install the worker and publish the first offsite copy
 
 ```task