"""No real keys, credentials, cluster or HOME are used by these regressions.""" import os from pathlib import Path import shutil import subprocess import pytest ROOT = Path(__file__).resolve().parents[1] BOOTSTRAP = ROOT / 'sso-mfa/bootstrap' @pytest.fixture def private_env(tmp_path): home = tmp_path / 'home' home.mkdir(mode=0o700) return {'PATH': os.environ['PATH'], 'HOME': str(home), 'TMPDIR': str(tmp_path), 'KUBECONFIG': str(tmp_path / 'no-cluster-config')} @pytest.mark.parametrize('flag', [[], ['--reprint']]) def test_emergency_guide_never_exports_the_default_root(private_env, flag): key = Path(private_env['HOME']) / '.config/sops/age/keys.txt' key.parent.mkdir(parents=True) key.write_text('synthetic-private-custody-marker\n') result = subprocess.run(['bash', str(BOOTSTRAP / 'emergency-bundle.sh'), *flag], env=private_env, input='\n', capture_output=True, text=True, timeout=5) assert result.returncode == 0 assert 'synthetic-private-custody-marker' not in result.stdout + result.stderr assert 'non-secret' in result.stdout assert not list(Path(private_env['TMPDIR']).glob('*emergency-bundle*')) def test_legacy_root_export_arguments_fail_without_echo(private_env): result = subprocess.run(['bash', str(BOOTSTRAP / 'emergency-bundle.sh'), '--age-key', 'synthetic-private-argument'], env=private_env, input='\n', capture_output=True, text=True, timeout=5) assert result.returncode != 0 assert 'synthetic-private-argument' not in result.stdout + result.stderr @pytest.mark.parametrize('trace', [False, True]) def test_generation_logs_no_secret_prefixes_and_files_are_private(tmp_path, private_env, trace): out = tmp_path / 'generated' result = subprocess.run(['bash', *(['-x'] if trace else []), str(BOOTSTRAP / 'gen-secrets.sh'), str(out)], env=private_env, capture_output=True, text=True, timeout=15) assert result.returncode == 0 transcript = result.stdout + result.stderr prefixes = [] for path in out.glob('*/secrets.env'): assert path.stat().st_mode & 0o077 == 0 for line in path.read_text().splitlines(): if line and not line.startswith('#') and '=' in line: value = line.split('=', 1)[1] if len(value) >= 8: prefixes.append(value[:8]) assert len(prefixes) >= 10 # Boolean only: test failure must not display generated material. assert not any(prefix in transcript for prefix in prefixes) def test_dry_run_does_not_read_root_contact_cluster_or_mutate_state(tmp_path, private_env): checkout = tmp_path / 'checkout' bootstrap = checkout / 'sso-mfa/bootstrap' bootstrap.mkdir(parents=True) for name in ['creds-bootstrap-agent.sh', 'creds-state.yaml']: shutil.copyfile(BOOTSTRAP / name, bootstrap / name) state = bootstrap / 'creds-state.yaml' before = state.read_bytes() bindir = tmp_path / 'bin'; bindir.mkdir() marker = tmp_path / 'unexpected-call' for name in ['kubectl', 'age', 'age-keygen', 'git', 'openssl']: exe = bindir / name exe.write_text('#!/bin/sh\nprintf called >> "$UNEXPECTED_CALL"\nexit 1\n') exe.chmod(0o755) env = {**private_env, 'PATH': str(bindir) + ':' + os.environ['PATH'], 'UNEXPECTED_CALL': str(marker)} result = subprocess.run(['bash', str(bootstrap / 'creds-bootstrap-agent.sh'), '--dry-run'], env=env, capture_output=True, text=True, timeout=5) assert result.returncode == 0 assert state.read_bytes() == before assert not marker.exists() assert not (Path(env['HOME']) / '.config/sops/age').exists() def test_unattended_live_bootstrap_refuses_before_effects(tmp_path, private_env): result = subprocess.run(['bash', str(BOOTSTRAP / 'creds-bootstrap-agent.sh')], env=private_env, stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=5) assert result.returncode != 0 assert 'attended terminal' in result.stderr assert not (Path(private_env['HOME']) / '.config/sops/age').exists() def run_attended(script, env, answer=b'STORED AND VERIFIED\n'): import pty import select import time master, slave = pty.openpty() child = subprocess.Popen(['bash', str(script)], env=env, stdin=slave, stdout=slave, stderr=slave, start_new_session=True) os.close(slave) transcript = bytearray() sent = False deadline = time.monotonic() + 15 try: while time.monotonic() < deadline: if select.select([master], [], [], 0.05)[0]: try: chunk = os.read(master, 65536) except OSError: break if not chunk: break transcript.extend(chunk) if b'type STORED AND VERIFIED:' in transcript and not sent: os.write(master, answer) sent = True if child.poll() is not None: # Drain the pty on the following select until EOF. continue return child.wait(timeout=2), bytes(transcript) finally: if child.poll() is None: child.kill() child.wait() os.close(master) @pytest.fixture def bootstrap_fixture(tmp_path, private_env): checkout = tmp_path / 'checkout' bootstrap = checkout / 'sso-mfa/bootstrap' bootstrap.mkdir(parents=True) for name in ['creds-bootstrap-agent.sh', 'gen-secrets.sh', 'encrypt-secrets.sh', 'pack-bundle.sh', 'emergency-bundle.sh', 'decrypt-secrets.sh']: shutil.copyfile(BOOTSTRAP / name, bootstrap / name) (bootstrap / 'creds-verify.sh').write_text('#!/bin/bash\nexit 0\n') state = bootstrap / 'creds-state.yaml' state.write_text('''schema_version: 2 secrets_generated: false ops_bundle_created: false ops_bundle_location: null emergency_bundle_delivered: true emergency_bundle_delivered_at: "historical-test-receipt" bootstrap_complete: true secrets_applied: postgres: true lldap: true authelia: true privacyidea: true keycape: true enckey_bootstrapped: true pi_admin_created: true openbao_post_unseal_verified: false ''') key = tmp_path / 'disposable-custody.key' subprocess.run(['age-keygen', '-o', str(key)], check=True, capture_output=True) recipient = subprocess.check_output(['age-keygen', '-y', str(key)], text=True).strip() (checkout / 'keys').mkdir() (checkout / 'keys/age.pub').write_text(recipient + '\n') bindir = tmp_path / 'bin'; bindir.mkdir() (bindir / 'kubectl').write_text('''#!/bin/bash printf call >> "$CLUSTER_CALLS" if [[ "$*" == 'get namespace openbao' ]]; then exit 1; fi if [[ "$*" == 'config current-context' ]]; then echo disposable-test; fi if [[ "$1 $2" == 'get secret' ]]; then echo secret/disposable-test; fi exit 0 ''') (bindir / 'git').write_text('#!/bin/bash\nexit 0\n') for path in bindir.iterdir(): path.chmod(0o755) calls = tmp_path / 'cluster-calls' env = {**private_env, 'PATH': str(bindir) + ':' + os.environ['PATH'], 'CLUSTER_CALLS': str(calls)} return bootstrap, state, key, env, calls def test_public_only_bootstrap_encrypts_recovery_without_exporting_root(bootstrap_fixture): import io import tarfile bootstrap, state, key, env, calls = bootstrap_fixture code, transcript = run_attended(bootstrap / 'creds-bootstrap-agent.sh', env) assert code == 0 assert calls.exists() assert not (Path(env['HOME']) / '.config/sops/age').exists() assert not (bootstrap / 'secrets').exists() bundles = list(bootstrap.parents[1].glob('ops-bundle-*.tar.age')) assert len(bundles) == 1 clear = subprocess.check_output(['age', '-d', '-i', str(key), str(bundles[0])]) with tarfile.open(fileobj=io.BytesIO(clear)) as archive: files = [m for m in archive.getmembers() if m.isfile()] assert len(files) == 6 for member in files: body = archive.extractfile(member).read() for line in body.splitlines(): if line and not line.startswith(b'#') and b'=' in line: value = line.split(b'=', 1)[1] if len(value) >= 8: assert bool(value[:8] in transcript) is False private_line = next(line for line in key.read_bytes().splitlines() if line.startswith(b'AGE-SECRET-KEY-')) assert bool(private_line in clear) is False recorded = state.read_text() assert 'recovery_handoff_confirmed: true' in recorded assert 'recovery_handoff_bundle_sha256:' in recorded assert 'bootstrap_complete: true' in recorded assert 'historical-test-receipt' in recorded def test_missing_custody_resume_refuses_before_cluster_call(bootstrap_fixture): bootstrap, state, _, env, calls = bootstrap_fixture state.write_text(state.read_text().replace('secrets_generated: false', 'secrets_generated: true')) before = state.read_bytes() code, transcript = run_attended(bootstrap / 'creds-bootstrap-agent.sh', env) assert code != 0 assert b'attended custody unlock helper' in transcript assert not calls.exists() assert state.read_bytes() == before def test_declined_handoff_is_not_marked_complete_or_cleaned(bootstrap_fixture): bootstrap, state, _, env, _ = bootstrap_fixture code, _ = run_attended(bootstrap / 'creds-bootstrap-agent.sh', env, answer=b'NO\n') assert code != 0 assert 'recovery_handoff_confirmed: false' in state.read_text() assert 'bootstrap_complete: false' in state.read_text() assert (bootstrap / 'secrets').is_dir() def test_explicit_matching_custody_resume_preserves_prior_bundle(bootstrap_fixture): bootstrap, state, key, env, _ = bootstrap_fixture code, _ = run_attended(bootstrap / 'creds-bootstrap-agent.sh', env) assert code == 0 first = next(bootstrap.parents[1].glob('ops-bundle-*.tar.age')) before = first.read_bytes() env['NETKINGDOM_CUSTODY_AGE_KEY_FILE'] = str(key) code, transcript = run_attended(bootstrap / 'creds-bootstrap-agent.sh', env) assert code == 0 assert first.read_bytes() == before assert len(list(bootstrap.parents[1].glob('ops-bundle-*.tar.age'))) == 2 private_line = next(line for line in key.read_bytes().splitlines() if line.startswith(b'AGE-SECRET-KEY-')) assert bool(private_line in transcript) is False assert not (bootstrap / 'secrets').exists() def test_mismatched_explicit_custody_key_refuses_before_cluster(bootstrap_fixture, tmp_path): bootstrap, state, _, env, calls = bootstrap_fixture wrong = tmp_path / 'other-disposable.key' subprocess.run(['age-keygen', '-o', str(wrong)], check=True, capture_output=True) state.write_text(state.read_text().replace('secrets_generated: false', 'secrets_generated: true')) env['NETKINGDOM_CUSTODY_AGE_KEY_FILE'] = str(wrong) code, transcript = run_attended(bootstrap / 'creds-bootstrap-agent.sh', env) assert code != 0 assert b'does not match' in transcript assert not calls.exists() def test_status_keeps_historical_receipt_distinct_from_current_handoff(): result = subprocess.run(['bash', str(BOOTSTRAP / 'creds-status.sh'), '--v2'], capture_output=True, text=True, timeout=5) assert result.returncode == 0 assert 'historical emergency delivery' in result.stdout assert 'current recovery acknowledgment' in result.stdout assert 'does not establish native recovery readiness' in result.stdout def test_interrupted_generation_never_erases_existing_custody(bootstrap_fixture): bootstrap, _, _, env, _ = bootstrap_fixture source=bootstrap/'secrets';source.mkdir(mode=0o700) retained=source/'retained';retained.write_bytes(b'synthetic-retained-recovery-material') code, transcript=run_attended(bootstrap/'creds-bootstrap-agent.sh', env) assert code != 0 assert retained.read_bytes()==b'synthetic-retained-recovery-material' assert b'synthetic-retained-recovery-material' not in transcript def test_failed_openbao_phase_preserves_private_files_and_refuses_completion(bootstrap_fixture): bootstrap, state, _, env, _ = bootstrap_fixture binary=Path(env['PATH'].split(':')[0])/'kubectl' binary.write_text(binary.read_text().replace("'get namespace openbao' ]]; then exit 1", "'get namespace openbao' ]]; then exit 0")) # Existing component booleans make the fixture skip creation; the optional # owner helper returns failure. No real OpenBao or cluster is involved. (bootstrap/'openbao-init-unseal.sh').write_text('#!/bin/bash\nexit 1\n') code, _=run_attended(bootstrap/'creds-bootstrap-agent.sh', env) assert code != 0 assert (bootstrap/'secrets').is_dir() assert 'bootstrap_complete: false' in state.read_text() assert not list(bootstrap.parents[1].glob('ops-bundle-*.tar.age'))