CUST-WP-0075: tenant-engine sender seed helper; re-qualify RPF-WP-0035-T02/T08

- scripts/tenant_engine_audit_core_sender_seed.py: silent attended row-7 seed
  for MASON-WP-0007-T05 (wrap-migrate technique, CAS 0, never overwrites,
  metadata-only --dry-run, documented exit codes), with offline fake-bao tests.
- Lane review doc with the exact founder sequence and the read-only ESO
  token-create can-i check for MASON-WP-0007-T06.
- RPF-WP-0035-T02: SECRETS-WP-0007-T04 is done; real blocker is live KeyCape
  registration and client custody of secrets-engine-openbao (blocked_on
  key-cape). Design stays proposed.
- RPF-WP-0035-T08: receiver rollout with evidence_kind already accepted
  2026-09-11; moved the misplaced D1 correction note here from T02.

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

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 3775@bnt-lap001
Assistant-Session: cfacb6a5-ad5b-4f79-ac67-053e3b0d34a3

tests/test_tenant_engine_audit_core_sender_seed.py

@@ -0,0 +1,183 @@
+"""Offline tests for the tenant-engine sender seed helper against a fake ``bao``.
+
+The helper runs as a real subprocess with a fake ``bao`` first on PATH, so the
+silence contract (empty stdout and stderr) is tested end to end.
+"""
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+import textwrap
+import unittest
+
+ROOT = Path(__file__).resolve().parents[1]
+SCRIPT = ROOT / 'scripts/tenant_engine_audit_core_sender_seed.py'
+CONFIRM = 'SEED tenant-engine audit-core-sender'
+TOKEN = 'te-' + 'A1b2' * 12
+OTHER = 'op-' + 'Z9y8' * 12
+SRC = 'workloads/audit-core/senders'
+DST = 'workloads/tenant-engine/audit-core-sender'
+
+FAKE_BAO = textwrap.dedent('''\
+    #!/usr/bin/env python3
+    import json, os, sys
+    sf = os.environ['FAKE_BAO_STATE']
+    st = json.load(open(sf))
+    st['calls'].append(sys.argv[1:])
+    def save(): json.dump(st, open(sf, 'w'))
+    def out(obj): save(); print(json.dumps(obj)); sys.exit(0)
+    def missing(): save(); sys.stderr.write('No value found at x\\n'); sys.exit(2)
+    a = sys.argv[1:]
+    print('Success! noise', file=sys.stderr)
+    if a[0] == 'read':
+        p = a[-1]
+        if p == 'auth/token/lookup-self':
+            out({'data': st['identity']})
+        mount, kind, rest = p.split('/', 2)
+        e = st['kv'].get(rest)
+        if not e: missing()
+        cv = len(e['versions'])
+        if kind == 'metadata':
+            out({'data': {'current_version': cv, 'versions': {str(i + 1): v['meta'] for i, v in enumerate(e['versions'])}}})
+        v = e['versions'][-1]
+        if v['meta'].get('deletion_time'): missing()
+        out({'data': {'data': v['data'], 'metadata': {'version': cv}}})
+    if a[0] == 'write':
+        p = a[-2].split('/', 2)[2]
+        body = json.loads(sys.stdin.read())
+        st['stdin'].append(body)
+        if st.get('fail_write'): save(); sys.exit(2)
+        e = st['kv'].setdefault(p, {'versions': []})
+        if body['options']['cas'] != len(e['versions']): save(); sys.exit(2)
+        e['versions'].append({'data': body['data'], 'meta': {}})
+        if st.get('corrupt_after_write'): e['versions'][-1]['data'] = {'token': 'x'}
+        out({'data': {'version': len(e['versions'])}})
+    save(); sys.exit(9)
+''')
+
+
+def registry(rows):
+    return {'versions': [{'data': {'senders.json': json.dumps(rows), 'other': 'keep'}, 'meta': {}}] * 9}
+
+
+def sender(name, tokens, may_read=False):
+    return {'name': name, 'tokens': tokens, 'sources': [name], 'may_write': True, 'may_read': may_read}
+
+
+class SeedHelper(unittest.TestCase):
+    def setUp(self):
+        self.tmp = Path(tempfile.mkdtemp())
+        (self.tmp / 'bin').mkdir()
+        fake = self.tmp / 'bin/bao'
+        fake.write_text(FAKE_BAO)
+        fake.chmod(0o755)
+        self.home = self.tmp / '.warden-attended-login' / 'session'
+        self.home.mkdir(parents=True)
+        self.state = {'calls': [], 'stdin': [], 'identity': {'policies': ['default', 'platform-admin']},
+                      'kv': {SRC: registry([sender('operator', [OTHER], may_read=True),
+                                            sender('tenant-engine', [TOKEN])])}}
+
+    def run_helper(self, *extra, confirm=CONFIRM, home=None, env_extra=None, receipt=None):
+        sf = self.tmp / 'state.json'
+        sf.write_text(json.dumps(self.state))
+        preexisting = receipt is not None
+        receipt = receipt or self.tmp / f'receipt-{len(list(self.tmp.glob("receipt-*")))}.json'
+        env = {'PATH': f'{self.tmp / "bin"}:/usr/bin:/bin', 'HOME': str(home or self.home),
+               'FAKE_BAO_STATE': str(sf), **(env_extra or {})}
+        args = [sys.executable, str(SCRIPT), '--receipt', str(receipt), *extra]
+        if confirm is not None:
+            args += ['--confirm', confirm]
+        r = subprocess.run(args, capture_output=True, env=env, timeout=60)
+        self.assertEqual((r.stdout, r.stderr), (b'', b''), 'helper must be silent')
+        self.state = json.loads(sf.read_text())
+        text = receipt.read_text() if receipt.exists() and not preexisting else '{}'
+        self.assertNotIn(TOKEN, text)
+        self.assertNotIn(OTHER, text)
+        for call in self.state['calls']:
+            self.assertNotIn(TOKEN, ' '.join(call), 'value must never be in bao argv')
+        if receipt.exists() and not preexisting:
+            self.assertEqual(receipt.stat().st_mode & 0o777, 0o600)
+        return r.returncode, json.loads(text)
+
+    def test_seed_writes_cas0_and_is_idempotent(self):
+        code, rec = self.run_helper()
+        self.assertEqual((code, rec['outcome'], rec['target_version']), (0, 'written', 1))
+        self.assertEqual(self.state['stdin'], [{'options': {'cas': 0}, 'data': {'token': TOKEN}}])
+        self.assertEqual(self.state['kv'][DST]['versions'][0]['data'], {'token': TOKEN})
+        self.assertEqual(len(self.state['kv'][SRC]['versions']), 9, 'registry untouched')
+        code, rec = self.run_helper()
+        self.assertEqual((code, rec['outcome']), (0, 'already_identical'))
+        self.assertEqual(len(self.state['stdin']), 1, 'no second write')
+
+    def test_dry_run_reads_metadata_only(self):
+        code, rec = self.run_helper('--dry-run')
+        self.assertEqual((code, rec['outcome'], rec['value_reads']), (0, 'dry_run_would_write', 0))
+        paths = [c[-1] for c in self.state['calls']]
+        self.assertTrue(all(p.startswith('platform/metadata/') or p == 'auth/token/lookup-self' for p in paths), paths)
+        self.assertFalse(any(c[0] == 'write' for c in self.state['calls']))
+        self.state['kv'][DST] = {'versions': [{'data': {'token': TOKEN}, 'meta': {}}]}
+        code, rec = self.run_helper('--dry-run')
+        self.assertEqual((code, rec['outcome']), (6, 'dry_run_target_present'))
+
+    def test_existing_different_target_is_refused_not_overwritten(self):
+        for data in ({'token': OTHER}, {'token': TOKEN, 'extra': '1'}):
+            self.state['kv'][DST] = {'versions': [{'data': data, 'meta': {}}]}
+            code, rec = self.run_helper()
+            self.assertEqual((code, rec['outcome']), (3, 'target_differs'))
+            self.assertEqual(self.state['stdin'], [])
+
+    def test_deleted_current_version_is_refused(self):
+        self.state['kv'][DST] = {'versions': [{'data': {'token': TOKEN}, 'meta': {'deletion_time': '2026-10-01T00:00:00Z'}}]}
+        code, rec = self.run_helper()
+        self.assertEqual((code, rec['outcome']), (3, 'target_current_version_not_live'))
+
+    def test_preconditions_refuse_before_any_bao_call_or_write(self):
+        code, rec = self.run_helper(confirm='yes')
+        self.assertEqual((code, rec['outcome'], self.state['calls']), (2, 'exact_confirmation_required', []))
+        code, rec = self.run_helper(home=self.tmp)
+        self.assertEqual((code, rec['outcome']), (2, 'attended_warden_envelope_required'))
+        code, rec = self.run_helper(env_extra={'BAO_TOKEN': 'x'})
+        self.assertEqual((code, rec['outcome']), (2, 'attended_warden_envelope_required'))
+        for policies in (['default'], ['platform-admin', 'root']):
+            self.state['identity'] = {'policies': policies}
+            code, rec = self.run_helper()
+            self.assertEqual((code, rec['outcome']), (2, 'attended_platform_admin_required'))
+        self.assertFalse(any(c[0] == 'write' for c in self.state['calls']))
+
+    def test_sender_shape_refusals(self):
+        cases = {
+            'sender_missing_or_duplicate': [sender('operator', [OTHER], may_read=True)],
+            'sender_token_count_not_one': [sender('tenant-engine', [TOKEN, TOKEN + 'b'])],
+            'sender_not_write_only': [sender('tenant-engine', [TOKEN], may_read=True)],
+        }
+        for reason, rows in cases.items():
+            self.state['kv'][SRC] = registry(rows)
+            code, rec = self.run_helper()
+            self.assertEqual((code, rec['outcome']), (2, reason))
+            self.assertNotIn(DST, self.state['kv'])
+        del self.state['kv'][SRC]
+        code, rec = self.run_helper()
+        self.assertEqual((code, rec['outcome']), (2, 'registry_absent'))
+
+    def test_write_and_verify_failures_have_distinct_codes(self):
+        self.state['fail_write'] = True
+        code, rec = self.run_helper()
+        self.assertEqual((code, rec['outcome']), (4, 'write_failed'))
+        self.state['fail_write'] = False
+        self.state['corrupt_after_write'] = True
+        code, rec = self.run_helper()
+        self.assertEqual((code, rec['outcome']), (5, 'readback_mismatch'))
+
+    def test_existing_receipt_is_never_overwritten(self):
+        receipt = self.tmp / 'receipt-0.json'
+        receipt.write_text('keep')
+        code, _ = self.run_helper(receipt=receipt)
+        self.assertEqual(code, 2)
+        self.assertEqual(receipt.read_text(), 'keep')
+        self.assertEqual(self.state['calls'], [])
+
+
+if __name__ == '__main__':
+    unittest.main()