tegwick
b97f9f616f
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a12d13-27f2-7163-b8c0-f682af02772c
259 lines
11 KiB
Python
259 lines
11 KiB
Python
"""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
|