Mint a separate MAS workload key in the attended gate instead of copying the supplier key.

The lane design forbids copying the founder-held supplier key onto the
workload path. The gate now mints a second key on the MAS application, proves
it, writes it CAS=0 over HTTP, and deletes it if a later step fails.

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

scripts/mas_evidence_iam_session.mjs

@@ -1,15 +1,26 @@
 #!/usr/bin/env node
-/** One attended Console login. The helper compares and corrects the MAS IAM rule. */
+/**
+ * One attended Console login.
+ *
+ * Default: the helper compares and corrects the MAS IAM rule.
+ * --mint-workload: start the attended OpenBao platform-admin login with the
+ * publisher gate as its child. The console session is read and sent only
+ * after the gate reports ready, so the OIDC login cannot age it out.
+ */
 import {spawn} from 'node:child_process';
-import {readFileSync} from 'node:fs';
+import {existsSync, readFileSync, rmSync} from 'node:fs';
+import {homedir} from 'node:os';
 import {dirname, join, resolve} from 'node:path';
 import {fileURLToPath, pathToFileURL} from 'node:url';
 import {parseArgs} from 'node:util';
 import {createInterface} from 'node:readline/promises';
 
-const {values} = parseArgs({options: {'playwright-module': {type: 'string'}}});
+const {values} = parseArgs({options: {'playwright-module': {type: 'string'}, 'mint-workload': {type: 'boolean'}}});
 const root = resolve(dirname(fileURLToPath(import.meta.url)), '..');
 const grant = join(root, 'scripts', 'mas_evidence_iam_grant.py');
+const gate = join(root, 'scripts', 'mas_evidence_publisher_gate.py');
+const ready = join(homedir(), '.local/state/railiance/receipts/mas-evidence-mint-ready.json');
+const READY_TIMEOUT_MS = 10 * 60 * 1000;
 let browser;
 let child;
 
@@ -45,6 +56,46 @@ function runGrant(auth) {
   });
 }
 
+function runMint(page) {
+  rmSync(ready, {force: true});
+  return new Promise((resolveResult, reject) => {
+    const env = {...process.env};
+    env.BAO_ADDR = env.BAO_ADDR || 'http://127.0.0.1:18200';
+    env.VAULT_ADDR = env.VAULT_ADDR || env.BAO_ADDR;
+    child = spawn('warden', [
+      'access', 'openbao-platform-admin-login', '--exec', '--', 'python3', gate, '--mint-workload',
+    ], {stdio: ['pipe', 'inherit', 'inherit'], env});
+    let sent = false;
+    const started = Date.now();
+    const timer = setInterval(async () => {
+      if (sent) return;
+      if (Date.now() - started > READY_TIMEOUT_MS) {
+        sent = true;
+        clearInterval(timer);
+        child?.stdin.end('');
+        return;
+      }
+      if (!existsSync(ready)) return;
+      sent = true;
+      clearInterval(timer);
+      try {
+        child?.stdin.end(JSON.stringify(await sessionAuth(page)));
+      } catch {
+        child?.stdin.end('');
+      }
+    }, 1000);
+    child.on('error', () => {
+      clearInterval(timer);
+      reject(new Error('warden_child_failed'));
+    });
+    child.on('close', (code) => {
+      clearInterval(timer);
+      child = null;
+      resolveResult({exit_code: code, output: `warden exit ${code}; receipts under ~/.local/state/railiance/receipts\n`});
+    });
+  });
+}
+
 try {
   const modulePath = resolve(values['playwright-module'] || '');
   const pkg = JSON.parse(readFileSync(join(dirname(modulePath), 'package.json'), 'utf8'));
@@ -61,7 +112,7 @@ try {
   console.log('Console opened. Sign in, then press Enter. The session token stays in this process.');
   await reader.question('');
   reader.close();
-  const outcome = await runGrant(await sessionAuth(page));
+  const outcome = values['mint-workload'] ? await runMint(page) : await runGrant(await sessionAuth(page));
   process.stdout.write(outcome.output.endsWith('\n') ? outcome.output : `${outcome.output}\n`);
   process.exitCode = outcome.exit_code === 0 ? 0 : 1;
   try {

scripts/mas_evidence_publisher_gate.py

@@ -1,8 +1,15 @@
 #!/usr/bin/env python3
-"""Apply the MAS publisher boundary and write its one CAS=0 pair.
+"""Mint the MAS workload key, prove its boundary, and write its one CAS=0 pair.
+
+Run only as `--mint-workload` inside the attended OpenBao platform-admin login,
+started by `mas_evidence_iam_session.mjs --mint-workload`. The founder-held
+supplier key at SUPPLIER is not read or copied: the workload pair is a second
+API key on the same IAM application, so it can be revoked on its own. A
+temporary Scaleway Console session arrives on stdin after the gate signals
+readiness. If anything fails after the mint, the minted key is deleted.
 
 Stdout stays empty. Receipts record statement ids, HTTP statuses, versions,
-and field names. Values are never written to a receipt or to stderr.
+and field names. Values are never written to a receipt, a file, or stderr.
 """
 
 from __future__ import annotations
@@ -13,7 +20,7 @@ import pwd
 import re
 import subprocess
 import sys
-import tempfile
+import time
 import urllib.request
 from pathlib import Path
 
@@ -45,6 +52,13 @@ STATEMENT_LIMIT = 10
 SUPPLIER = "platform/suppliers/scaleway/mas-evidence-backup-primary"
 WORKLOAD = "platform/workloads/mas/backup-primary"
 PREFERRED_PROJECT = "reef-storage"
+PROJECT_ID = "e1a0dd0e-04b8-4ea9-8b30-d53f8c35c688"
+POLICY_NAME = "mas-evidence-backup-primary"
+WORKLOAD_KEY_DESCRIPTION = "MAS evidence workload pair; RPF-WP-0070; node delivery"
+READY_NAME = "mas-evidence-mint-ready.json"
+MINT_NAME = "mas-evidence-workload-mint.json"
+PROOF_ATTEMPTS = 7
+PROOF_DELAY_SECONDS = 10
 BAO_ADDR = "http://127.0.0.1:18200"
 _PROJECT_ID = re.compile(
     r"\A[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}\Z"
@@ -540,24 +554,27 @@ class _Transport:
         return None
 
     def put_cas0(self, path, body):
-        handle = tempfile.NamedTemporaryFile(
-            "w", prefix="mas-pair-", dir=RECEIPT_DIR, delete=False
-        )
-        name = Path(handle.name)
-        try:
-            json.dump(body["data"], handle)
-            handle.flush()
-            os.fchmod(handle.fileno(), 0o600)
-            handle.close()
-            proc = _bao(
-                ["kv", "put", "-format=json", "-cas=0", logical_path(path), f"@{name}"],
-                token=self.token,
-            )
-        finally:
-            name.unlink(missing_ok=True)
-        if proc.returncode:
+        """Write over the HTTP API. The value never touches a file."""
+        if logical_path(path) != WORKLOAD or body.get("options") != {"cas": 0}:
             raise GateError("pair_write_failed")
-        return version_from_put(json.loads(proc.stdout))
+        request = urllib.request.Request(
+            BAO_ADDR + "/v1/" + path,
+            data=json.dumps(body).encode(),
+            method="POST",
+            headers={"X-Vault-Token": self.token, "Content-Type": "application/json"},
+        )
+        opener = urllib.request.build_opener(_NoRedirect)
+        try:
+            with opener.open(request, timeout=20) as response:
+                raw = response.read(65537)
+        except GateError:
+            raise
+        except Exception:
+            raise GateError("pair_write_failed") from None
+        try:
+            return version_from_put(json.loads(raw))
+        except json.JSONDecodeError:
+            raise GateError("pair_write_failed") from None
 
     def revoke(self):
         _bao(["token", "revoke", "-self"], token=self.token)
@@ -627,9 +644,6 @@ def write_pair(pair):
     existing = _bao(["kv", "metadata", "get", "-format=json", WORKLOAD])
     if existing.returncode == 0:
         raise GateError("existing_custody_requires_rotation_review")
-    written = _bao(["policy", "write", "backup-pair-writer", str(POLICY_FILE)])
-    if written.returncode:
-        raise GateError("writer_policy_install_failed")
     auth = _request_writer_auth()
     token = auth["client_token"]
     try:
@@ -657,31 +671,148 @@ def write_pair(pair):
     return summary
 
 
-def main():
+def _app_keys(api):
+    value = api(
+        "GET",
+        "/iam/v1alpha1/api-keys?application_id=" + APPLICATION_ID + "&page_size=100",
+    )
+    keys = value.get("api_keys") if isinstance(value, dict) else None
+    if not isinstance(keys, list) or value.get("total_count") != len(keys):
+        raise GateError("key_inventory_rejected")
+    return [item for item in keys if item.get("application_id") == APPLICATION_ID]
+
+
+def _key_view(keys):
+    return {
+        "count": len(keys),
+        "descriptions": sorted(str(item.get("description") or "") for item in keys),
+        "default_project_ids": sorted({str(item.get("default_project_id")) for item in keys}),
+    }
+
+
+def mint_workload_pair(api, prove, write, sleep=time.sleep):
+    """Mint one more key on the MAS application, prove it, then store it CAS=0.
+
+    `api` is the console transport. `prove(pair, project)` and `write(pair)`
+    are the S3 proof and the OpenBao write. The minted key is deleted when a
+    later step fails, and the receipt says whether that deletion happened.
+    """
+    receipt = {"application_id": APPLICATION_ID, "values_recorded": False}
+    app = api("GET", "/iam/v1alpha1/applications/" + APPLICATION_ID)
+    if app.get("name") != POLICY_NAME:
+        raise GateError("application_binding_refused")
+    before = _app_keys(api)
+    receipt["keys_before"] = _key_view(before)
+    if len(before) != 1 or before[0].get("default_project_id") != PROJECT_ID:
+        raise GateError("supplier_key_state_requires_review", receipt["keys_before"])
+    issued = api("POST", "/iam/v1alpha1/api-keys", {
+        "application_id": APPLICATION_ID,
+        "default_project_id": PROJECT_ID,
+        "description": WORKLOAD_KEY_DESCRIPTION,
+    })
+    access = issued.get("access_key") if isinstance(issued, dict) else None
+    try:
+        if issued.get("application_id") != APPLICATION_ID:
+            raise GateError("issued_key_binding_refused")
+        pair = {
+            "AWS_ACCESS_KEY_ID": access,
+            "AWS_SECRET_ACCESS_KEY": issued.get("secret_key"),
+        }
+        if any(not isinstance(value, str) or not value for value in pair.values()):
+            raise GateError("issued_key_incomplete")
+        receipt["minted_at"] = issued.get("created_at")
+        attempts = 0
+        while True:
+            attempts += 1
+            try:
+                prove(pair, PROJECT_ID)
+                break
+            except GateError as exc:
+                if exc.code != "publisher_denials_failed" or attempts >= PROOF_ATTEMPTS:
+                    raise
+                sleep(PROOF_DELAY_SECONDS)
+        receipt["proof_attempts"] = attempts
+        receipt["pair"] = write(pair)
+    except Exception as exc:
+        deleted = False
+        if isinstance(access, str) and access:
+            try:
+                api("DELETE", "/iam/v1alpha1/api-keys/" + access)
+                deleted = True
+            except Exception:
+                deleted = False
+        code = exc.code if isinstance(exc, GateError) else "contained_gate_failure"
+        detail = dict(exc.detail or {}) if isinstance(exc, GateError) and isinstance(exc.detail, dict) else {}
+        detail["minted_key_deleted"] = deleted
+        raise GateError(code, detail) from None
+    receipt["keys_after"] = _key_view(_app_keys(api))
+    receipt["status"] = "passed"
+    return receipt
+
+
+def _console_api(raw):
+    import importlib.util
+
+    if len(raw) > 65536:
+        raise GateError("console_auth_rejected")
+    try:
+        auth = json.loads(raw)
+    except json.JSONDecodeError:
+        raise GateError("console_auth_rejected") from None
+    spec = importlib.util.spec_from_file_location(
+        "application_recovery_identities", ROOT / "scripts/application_recovery_identities.py"
+    )
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    try:
+        transport = module.Transport(auth)
+    except module.Refused:
+        raise GateError("console_auth_rejected") from None
+
+    def api(method, path, data=None):
+        try:
+            return transport(method, path, data)
+        except module.ProviderError as exc:
+            raise GateError("provider_http_error", {"method": method, "http_status": exc.status}) from None
+        except module.Refused as exc:
+            raise GateError("provider_refused", {"reason": str(exc)}) from None
+
+    return api
+
+
+def _preflight():
+    existing = _bao(["kv", "metadata", "get", "-format=json", WORKLOAD])
+    if existing.returncode == 0:
+        raise GateError("existing_custody_requires_rotation_review")
+    if _bao(["policy", "write", "backup-pair-writer", str(POLICY_FILE)]).returncode:
+        raise GateError("writer_policy_install_failed")
+
+
+def main(argv=None):
+    args = list(sys.argv[1:] if argv is None else argv)
     sys.stdout = open(os.devnull, "w")
     sys.stderr = open(os.devnull, "w")
+    receipt_path = RECEIPT_DIR / "mas-evidence-publisher-gate.json"
     try:
+        if args != ["--mint-workload"]:
+            raise GateError("usage_refused")
         _ensure_boto()
         RECEIPT_DIR.mkdir(mode=0o700, exist_ok=True)
-        supplier = supplier_pair(_kv_data(SUPPLIER), SUPPLIER)
-        backup_path = "platform/workloads/railiance/backup/object-storage"
-        bootstrap_path = "platform/workloads/railiance/scaleway/bootstrap"
-        backup = _kv_data(backup_path)
-        bootstrap = _kv_data(bootstrap_path)
-        reader = supplier_pair(backup, backup_path)
-        writer = supplier_pair(bootstrap, bootstrap_path)
-        apply_live_policy(reader, writer)
-        prove_denials(supplier, bootstrap.get("DEFAULT_PROJECT_ID", ""))
-        write_pair(supplier)
+        (RECEIPT_DIR / MINT_NAME).unlink(missing_ok=True)
+        _preflight()
+        _write_json(RECEIPT_DIR / READY_NAME, {"status": "waiting_for_console", "values_recorded": False})
+        api = _console_api(sys.stdin.read(65537))
+        receipt = mint_workload_pair(api, prove_denials, write_pair)
+        _write_json(RECEIPT_DIR / MINT_NAME, receipt)
     except GateError as exc:
         receipt = {"status": "failed", "code": exc.code, "values_recorded": False}
         if exc.detail:
             receipt["detail"] = exc.detail
-        _write_json(RECEIPT_DIR / "mas-evidence-publisher-gate.json", receipt)
+        _write_json(receipt_path, receipt)
         return 71
     except Exception as exc:
         _write_json(
-            RECEIPT_DIR / "mas-evidence-publisher-gate.json",
+            receipt_path,
             {
                 "status": "failed",
                 "code": "contained_gate_failure",
@@ -690,7 +821,9 @@ def main():
             },
         )
         return 72
-    _write_json(RECEIPT_DIR / "mas-evidence-publisher-gate.json", {"status": "passed", "values_recorded": False})
+    finally:
+        (RECEIPT_DIR / READY_NAME).unlink(missing_ok=True)
+    _write_json(receipt_path, {"status": "passed", "values_recorded": False})
     return 0