net-kingdom/tests/test_keycape_bootstrap.py

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