tegwick
d9557ba90e
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a13eab-1e6a-7b33-9c37-372e90ada3a2
134 lines
6.1 KiB
Python
134 lines
6.1 KiB
Python
"""Final caller audit regressions; only disposable inputs and fake cluster tools."""
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
BOOT = ROOT / 'sso-mfa/bootstrap'
|
|
|
|
|
|
@pytest.fixture
|
|
def bundle(tmp_path):
|
|
key = tmp_path / 'fixture.key'
|
|
subprocess.run(['age-keygen', '-o', str(key)], capture_output=True, check=True)
|
|
recipient = subprocess.check_output(['age-keygen', '-y', str(key)], text=True).strip()
|
|
source = tmp_path / 'secrets'; source.mkdir(mode=0o700)
|
|
component = source / 'service'; component.mkdir(mode=0o700)
|
|
value = component / 'secrets.env'
|
|
value.write_text('VALUE=synthetic-recovery-payload\n'); value.chmod(0o600)
|
|
return source, recipient, key, tmp_path / 'bundle.tar.age'
|
|
|
|
|
|
def pack(bundle):
|
|
source, recipient, _, output = bundle
|
|
return subprocess.run(['bash', '-x', str(BOOT / 'pack-bundle.sh'), str(source), recipient, str(output)],
|
|
capture_output=True, timeout=10)
|
|
|
|
|
|
def test_bundle_does_not_replace_existing_output(bundle):
|
|
output = bundle[3]; output.write_bytes(b'previous-recovery-artifact')
|
|
r = pack(bundle)
|
|
assert r.returncode != 0
|
|
assert output.read_bytes() == b'previous-recovery-artifact'
|
|
|
|
|
|
def test_bundle_refuses_links_to_outside_custody(bundle, tmp_path):
|
|
source, _, _, output = bundle
|
|
outside = tmp_path / 'outside'; outside.write_text('synthetic-outside-marker')
|
|
(source / 'service/link').symlink_to(outside)
|
|
r = pack(bundle)
|
|
assert r.returncode != 0
|
|
assert not output.exists()
|
|
|
|
|
|
def test_bundle_ciphertext_is_private_and_has_no_plaintext_staging(bundle, tmp_path):
|
|
r = pack(bundle)
|
|
assert r.returncode == 0
|
|
assert bundle[3].stat().st_mode & 0o077 == 0
|
|
assert b'synthetic-recovery-payload' not in r.stdout + r.stderr
|
|
assert not list(tmp_path.glob('.netkingdom-bundle-*'))
|
|
clear = subprocess.check_output(['age', '-d', '-i', str(bundle[2]), str(bundle[3])])
|
|
assert b'synthetic-recovery-payload' in clear
|
|
|
|
|
|
def test_bundle_refuses_private_age_key_in_payload(bundle):
|
|
source, _, key, output = bundle
|
|
(source / 'service/custody.key').write_bytes(key.read_bytes())
|
|
(source / 'service/custody.key').chmod(0o600)
|
|
r = pack(bundle)
|
|
assert r.returncode != 0
|
|
assert not output.exists()
|
|
|
|
|
|
def test_legacy_delivery_does_not_contact_cluster_or_inspect_arguments(tmp_path):
|
|
binary = tmp_path / 'kubectl'; marker = tmp_path / 'called'
|
|
binary.write_text('#!/bin/sh\ntouch "$CALLED"\necho synthetic-provider-marker >&2\nexit 1\n')
|
|
binary.chmod(0o755)
|
|
env = {**os.environ, 'PATH': str(tmp_path) + ':' + os.environ['PATH'], 'CALLED': str(marker)}
|
|
r = subprocess.run(['bash', '-x', str(ROOT / 'tools/sync-user-engine-delivery-secret.sh'),
|
|
'synthetic-private-argument'], capture_output=True, env=env)
|
|
assert r.returncode == 2
|
|
assert not marker.exists()
|
|
assert b'synthetic-private-argument' not in r.stdout + r.stderr
|
|
|
|
|
|
def test_openbao_plan_is_nonmutating_and_does_not_contact_cluster(tmp_path):
|
|
binary = tmp_path / 'kubectl'; marker = tmp_path / 'called'
|
|
binary.write_text('#!/bin/sh\ntouch "$CALLED"\nexit 1\n'); binary.chmod(0o755)
|
|
env = {**os.environ, 'PATH': str(tmp_path) + ':' + os.environ['PATH'], 'CALLED': str(marker),
|
|
'OPENBAO_CONSOLE_METADATA': str(tmp_path / 'no-metadata')}
|
|
r = subprocess.run(['bash', '-x', str(BOOT / 'openbao-init-unseal.sh'), '--dry-run'],
|
|
capture_output=True, env=env)
|
|
assert r.returncode == 0
|
|
assert not marker.exists()
|
|
assert not (tmp_path / 'secrets').exists()
|
|
|
|
|
|
@pytest.mark.parametrize('relative', ['sso-mfa/k8s/keycape/configure-openbao-oidc.sh',
|
|
'sso-mfa/k8s/user-engine/verify-operability.sh'])
|
|
def test_retired_root_and_stale_rollout_paths_refuse_before_external_calls(tmp_path, relative):
|
|
marker = tmp_path / 'called'; binary = tmp_path / 'kubectl'
|
|
binary.write_text('#!/bin/sh\ntouch "$CALLED"\nexit 1\n'); binary.chmod(0o755)
|
|
env = {**os.environ, 'PATH': str(tmp_path) + ':' + os.environ['PATH'], 'CALLED': str(marker)}
|
|
r = subprocess.run(['bash', '-x', str(ROOT / relative), 'synthetic-private-argument'],
|
|
env=env, capture_output=True)
|
|
assert r.returncode == 2
|
|
assert not marker.exists()
|
|
assert b'synthetic-private-argument' not in r.stdout + r.stderr
|
|
|
|
|
|
def test_secret_presence_does_not_rewrite_historical_acceptance(tmp_path):
|
|
import shutil
|
|
script = tmp_path / 'creds-verify.sh'; shutil.copy(BOOT / script.name, script)
|
|
state = tmp_path / 'creds-state.yaml'
|
|
original = b'secrets_applied:\n keycape: false\nenckey_bootstrapped: false\n'
|
|
state.write_bytes(original)
|
|
binary = tmp_path / 'kubectl'
|
|
binary.write_text('#!/bin/sh\nif [ "$1" = get ]; then echo secret/fixture; fi\nexit 0\n')
|
|
binary.chmod(0o755)
|
|
env = {**os.environ, 'PATH': str(tmp_path) + ':' + os.environ['PATH']}
|
|
r = subprocess.run(['bash', str(script)], env=env, capture_output=True)
|
|
assert r.returncode == 0
|
|
assert state.read_bytes() == original
|
|
assert b'keycape-pi-token' not in r.stdout + r.stderr
|
|
|
|
|
|
@pytest.mark.parametrize('name', ['patch-openbao-client.sh', 'register-coulomb-social.sh'])
|
|
def test_legacy_registry_mutation_refuses_without_cluster_calls(tmp_path, name):
|
|
marker = tmp_path / 'called'; executable = tmp_path / 'kubectl'
|
|
executable.write_text('#!/bin/sh\ntouch "$CALLED"\nexit 1\n'); executable.chmod(0o755)
|
|
env = {**os.environ, 'PATH': str(tmp_path) + ':' + os.environ['PATH'], 'CALLED': str(marker)}
|
|
r = subprocess.run(['bash', '-x', str(ROOT / 'sso-mfa/k8s/keycape' / name)], env=env, capture_output=True)
|
|
assert r.returncode == 2
|
|
assert not marker.exists()
|
|
|
|
|
|
@pytest.mark.parametrize('name,args', [('register-coulomb-social.py', []), ('openbao-client-config.py', ['patch'])])
|
|
def test_legacy_python_registry_output_is_disabled(name, args):
|
|
r = subprocess.run(['python3', str(ROOT / 'sso-mfa/k8s/keycape' / name), *args],
|
|
input=b'synthetic-private-input', capture_output=True)
|
|
assert r.returncode != 0
|
|
assert b'synthetic-private-input' not in r.stdout + r.stderr
|
|
assert not r.stdout
|