"""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_SVC_KEYCAPE_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_SVC_KEYCAPE_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))