fix(security): reconcile KeyCape bootstrap with scoped factor custody

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a1346b-72a4-7d40-b8bd-8b8ca346cfa2

tests/test_keycape_bootstrap.py

@@ -0,0 +1,142 @@
+"""Synthetic custody tests; no real cluster or credential access."""
+import base64
+import json
+import os
+from pathlib import Path
+import subprocess
+
+import pytest
+import yaml
+
+ROOT = Path(__file__).resolve().parents[1]
+SCRIPTS = ROOT / 'sso-mfa/k8s/keycape'
+PASSWORD = 'SyntheticOnlyCredential12345678901234567890'
+
+
+@pytest.fixture
+def setup(tmp_path):
+    root = tmp_path / 'private'; root.mkdir(mode=0o700)
+    for component, key in [('lldap', 'LLDAP_LDAP_USER_PASS'), ('authelia', 'AUTHELIA_KEYCAPE_CLIENT_SECRET'), ('keycape', None)]:
+        folder = root / component; folder.mkdir(mode=0o700)
+        file = folder / 'secrets.env'; file.write_text(f'{key}={PASSWORD}\n' if key else '# placeholder\n'); file.chmod(0o600)
+    binary = tmp_path / 'bin'; binary.mkdir()
+    (binary / 'kubectl').write_text('''#!/usr/bin/python3
+import json,os,sys
+args=sys.argv[1:]
+raw=sys.stdin.read() if args[-1:]==['-'] else ''
+with open(os.environ['CALLS'],'a') as f:f.write(json.dumps({'args':args,'stdin':raw})+'\\n')
+if os.environ.get('FAIL')==args[0]:
+ print(os.environ['SYNTHETIC']);print(os.environ['SYNTHETIC'],file=sys.stderr);sys.exit(1)
+if args[:3]==['get','secret','keycape-config']:
+ print(os.environ.get('EXISTING',''),end='')
+elif args[:2]==['get','externalsecret']:
+ print(os.environ.get('READY','True'),end='')
+elif args[:3]==['get','secret','keycape-factor-read']:
+ print(os.environ.get('FACTOR','secret/keycape-factor-read'),end='')
+elif args[0]=='create':
+ print('secret/keycape-config',end='')
+else: sys.exit(3)
+''')
+    (binary / 'kubectl').chmod(0o700)
+    for name in ('curl', 'sleep'):
+        (binary / name).write_text('#!/bin/sh\nexit 97\n'); (binary / name).chmod(0o700)
+    env={'PATH':str(binary)+':'+os.environ['PATH'],'HOME':str(tmp_path),'CALLS':str(tmp_path/'calls'),'SYNTHETIC':PASSWORD}
+    return root, env
+
+
+def invoke(setup, script='create-secrets.sh', **extra):
+    root, env=setup
+    return subprocess.run(['bash','-x',str(SCRIPTS/script),str(root)],input=PASSWORD+'\n',
+                          env={**env,**extra},capture_output=True,text=True,timeout=15)
+
+
+def calls(setup):
+    file=Path(setup[1]['CALLS'])
+    return [json.loads(line) for line in file.read_text().splitlines()] if file.exists() else []
+
+
+@pytest.mark.parametrize('script',['create-pi-token.sh','refresh-pi-token-live.sh'])
+def test_retired_admin_token_paths_refuse_before_read_or_cluster_access(setup,script):
+    result=invoke(setup,script)
+    assert result.returncode!=0
+    assert 'RPF-WP-0040' in result.stdout+result.stderr
+    assert PASSWORD not in result.stdout+result.stderr
+    assert not calls(setup)
+    assert not (setup[0]/'keycape/pi_admin_token').exists()
+
+
+def test_new_config_uses_mounted_scoped_credential_without_secret_argv_or_output(setup):
+    result=invoke(setup)
+    assert result.returncode==0, result.stderr
+    assert PASSWORD not in result.stdout+result.stderr
+    observed=calls(setup)
+    assert PASSWORD not in json.dumps([c['args'] for c in observed])
+    writes=[c for c in observed if c['args'][0] not in ['get']]
+    assert len(writes)==1 and writes[0]['args']==['create','-o','name','-f','-']
+    secret=json.loads(writes[0]['stdin'])
+    assert 'annotations' not in secret['metadata']
+    config=yaml.safe_load(base64.b64decode(secret['data']['config.yaml']))
+    assert config['privacyidea'].get('adminToken','')==''
+    assert config['privacyidea']['adminTokenFile']=='/etc/keycape-factor/admin-token'
+    assert config['privacyidea']['realm']=='coulomb'
+    assert config['privacyidea']['requireForAll'] is True
+    assert config['lldap']['bindPW']==PASSWORD
+    assert config['authelia']['clientSecret']==PASSWORD
+    key=setup[0]/'keycape/key.pem'
+    assert key.stat().st_mode & 0o077==0
+    assert base64.b64decode(secret['data']['key.pem'])==key.read_bytes()
+    checked=subprocess.run(['openssl','rsa','-check','-noout'],input=key.read_bytes(),capture_output=True)
+    assert checked.returncode==0
+
+
+@pytest.mark.parametrize('extra',[{'EXISTING':'secret/keycape-config'},{'READY':'False'},{'READY':''},{'FACTOR':''},{'FAIL':'get'}])
+def test_existing_config_or_missing_delivery_refuses_without_local_or_remote_writes(setup,extra):
+    result=invoke(setup,**extra)
+    assert result.returncode!=0
+    assert not (setup[0]/'keycape/key.pem').exists()
+    assert all(c['args'][0]=='get' for c in calls(setup))
+    assert PASSWORD not in result.stdout+result.stderr
+
+
+@pytest.mark.parametrize('defect',['public','public-parent','public-root','symlink','parent-symlink','hardlink','duplicate','shell','missing'])
+def test_bad_custody_refused_before_cluster_access(setup,defect):
+    root,_=setup; file=root/'lldap/secrets.env'; marker=root/'executed'
+    if defect=='public':file.chmod(0o644)
+    elif defect=='public-parent':file.parent.chmod(0o755)
+    elif defect=='public-root':root.chmod(0o755)
+    elif defect=='symlink':
+        target=file.with_name('other');file.rename(target);file.symlink_to(target)
+    elif defect=='parent-symlink':
+        target=root/'other';file.parent.rename(target);(root/'lldap').symlink_to(target)
+    elif defect=='hardlink':os.link(file,file.with_name('alias'))
+    elif defect=='duplicate':file.write_text(file.read_text()*2)
+    elif defect=='shell':file.write_text(f'LLDAP_LDAP_USER_PASS=$(touch {marker})\n')
+    else:file.unlink()
+    result=invoke(setup)
+    assert result.returncode!=0
+    assert not marker.exists()
+    assert not calls(setup)
+    assert PASSWORD not in result.stdout+result.stderr
+
+
+def test_creation_failure_retains_private_key_for_retry_without_leaking(setup):
+    result=invoke(setup,FAIL='create')
+    assert result.returncode!=0
+    assert PASSWORD not in result.stdout+result.stderr
+    key=setup[0]/'keycape/key.pem'; before=key.read_bytes()
+    assert key.stat().st_mode & 0o077==0
+    assert invoke(setup).returncode==0
+    assert key.read_bytes()==before
+
+
+@pytest.mark.parametrize('defect',['symlink','hardlink','public','invalid'])
+def test_existing_invalid_signing_key_is_not_replaced(setup,defect):
+    key=setup[0]/'keycape/key.pem'; key.write_text('invalid');key.chmod(0o600)
+    if defect=='symlink':
+        target=key.with_name('other');key.rename(target);key.symlink_to(target)
+    elif defect=='hardlink':os.link(key,key.with_name('alias'))
+    elif defect=='public':key.chmod(0o644)
+    result=invoke(setup)
+    assert result.returncode!=0
+    assert key.read_text()=='invalid'
+    assert not any(c['args'][0]=='create' for c in calls(setup))

tests/test_rotation_output.py

@@ -12,7 +12,7 @@ BOOT = ROOT / 'sso-mfa/bootstrap'
 CASES = {
     'PI_SECRET_KEY': 'privacyidea', 'PI_DB_PASSWORD': 'privacyidea',
     'LLDAP_JWT_SECRET': 'lldap', 'AUTHELIA_SESSION_SECRET': 'authelia',
-    'AUTHELIA_KEYCAPE_CLIENT_SECRET': 'authelia', 'BREAKGLASS_PASSWORD': 'breakglass',
+    'BREAKGLASS_PASSWORD': 'breakglass',
 }
 SYNTHETIC = 'A' * 64
 
@@ -183,3 +183,19 @@ def test_declining_rotation_preserves_custody_file_and_cluster(rotation):
     assert file.read_bytes() == before
     assert not calls(rotation)
     assert bool(SYNTHETIC[:40] in result.stdout + result.stderr) is False
+
+
+@pytest.mark.parametrize('key',['AUTHELIA_KEYCAPE_CLIENT_SECRET','KEYCAPE_RSA_KEY'])
+@pytest.mark.parametrize('attended',[False,True])
+def test_retired_keycape_replacement_refuses_before_any_credential_change(rotation,key,attended):
+    repo,boot,secrets,env=rotation
+    signing=secrets/'keycape/key.pem';signing.write_text('synthetic-existing-private-key');signing.chmod(0o600)
+    before={p:p.read_bytes() for p in secrets.rglob('*') if p.is_file()}
+    args=['bash','-x',str(boot/'creds-rotate.sh')]
+    if not attended:args.append('--non-interactive')
+    result=subprocess.run([*args,str(secrets)],input='yes\n',env={**env,'SECRET':key},capture_output=True,text=True,timeout=10)
+    assert result.returncode!=0
+    assert not calls(rotation)
+    assert {p:p.read_bytes() for p in secrets.rglob('*') if p.is_file()}==before
+    assert 'last_rotated_' not in (boot/'creds-state.yaml').read_text()
+    assert 'NK-WP-0049-T02' in result.stderr