tegwick
4cf26a9bc8
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a12d13-27f2-7163-b8c0-f682af02772c
178 lines
8.2 KiB
Python
178 lines
8.2 KiB
Python
"""Disposable age keys only; subprocess logs are never asserted with raw values."""
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
BOOT = ROOT / 'sso-mfa/bootstrap'
|
|
|
|
|
|
@pytest.fixture
|
|
def custody(tmp_path):
|
|
home = tmp_path / 'home'; home.mkdir()
|
|
key = tmp_path / 'disposable.key'
|
|
subprocess.run(['age-keygen', '-o', str(key)], capture_output=True, check=True)
|
|
public = subprocess.check_output(['age-keygen', '-y', str(key)], text=True).strip()
|
|
env = {'PATH': os.environ['PATH'], 'HOME': str(home), 'TMPDIR': str(tmp_path)}
|
|
return key, public, env
|
|
|
|
|
|
def unlock(custody, *args):
|
|
_, public, env = custody
|
|
return subprocess.run(['bash', str(BOOT / 'sops-custody-unlock.sh'), '--expected-recipient', public, *args],
|
|
env=env, capture_output=True, text=True, timeout=10)
|
|
|
|
|
|
def test_custody_unknown_arguments_are_not_echoed(custody):
|
|
result = unlock(custody, 'synthetic-secret-argument')
|
|
assert result.returncode != 0
|
|
assert 'synthetic-secret-argument' not in result.stdout + result.stderr
|
|
|
|
|
|
def test_signal_preserves_failure_status_and_removes_temporary_key(custody):
|
|
key, _, env = custody
|
|
result = unlock(custody, '--from-file', str(key), '--', 'sh', '-c', 'kill -TERM "$PPID"; exit 0')
|
|
assert result.returncode == 143
|
|
assert not list(Path(env['TMPDIR']).glob('netkingdom-sops-age.*'))
|
|
|
|
|
|
def test_ambiguous_custody_sources_are_refused_before_child(custody, tmp_path):
|
|
key, _, _ = custody
|
|
marker = tmp_path / 'child-called'
|
|
result = unlock(custody, '--from-file', str(key), '--from-file', str(key), '--', 'touch', str(marker))
|
|
assert result.returncode != 0
|
|
assert not marker.exists()
|
|
|
|
|
|
def test_explicit_custody_check_cleans_key(custody):
|
|
key, _, env = custody
|
|
result = unlock(custody, '--from-file', str(key), '--check-only')
|
|
assert result.returncode == 0
|
|
assert not list(Path(env['TMPDIR']).glob('netkingdom-sops-age.*'))
|
|
|
|
|
|
def test_prompt_does_not_leak_when_shell_tracing_is_requested(custody):
|
|
import pty, select, time
|
|
key, public, env = custody
|
|
private_line = next(line for line in key.read_bytes().splitlines() if line.startswith(b'AGE-SECRET-KEY-'))
|
|
master, slave = pty.openpty()
|
|
child = subprocess.Popen(['bash', '-x', str(BOOT / 'sops-custody-unlock.sh'),
|
|
'--expected-recipient', public, '--prompt', '--check-only'],
|
|
stdin=slave, stdout=slave, stderr=slave, env=env)
|
|
os.close(slave)
|
|
captured = bytearray(); sent = False
|
|
deadline = time.monotonic() + 10
|
|
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
|
|
captured.extend(chunk)
|
|
if b'input hidden): ' in captured and not sent:
|
|
os.write(master, private_line + b'\n'); sent = True
|
|
assert child.wait(timeout=2) == 0
|
|
assert bool(private_line in captured) is False
|
|
assert not list(Path(env['TMPDIR']).glob('netkingdom-sops-age.*'))
|
|
finally:
|
|
if child.poll() is None: child.kill(); child.wait()
|
|
os.close(master)
|
|
|
|
|
|
def encrypt_fixture(tmp_path, public):
|
|
secrets = tmp_path / 'secrets'; (secrets / 'service').mkdir(parents=True)
|
|
(secrets / 'service/one.env').write_text('TEST_VALUE=synthetic-private-payload\n')
|
|
out = tmp_path / 'secrets.enc/service'; out.mkdir(parents=True)
|
|
subprocess.run(['age', '-r', public, '-o', str(out / 'one.env.age'), str(secrets / 'service/one.env')],
|
|
capture_output=True, check=True)
|
|
return secrets, out
|
|
|
|
|
|
def test_encryption_refuses_private_key_as_recipient(custody, tmp_path):
|
|
key, public, env = custody
|
|
secrets, _ = encrypt_fixture(tmp_path, public)
|
|
result = subprocess.run(['bash', str(BOOT / 'encrypt-secrets.sh'), str(secrets), str(key), '--no-shred'],
|
|
env=env, capture_output=True, text=True, timeout=10)
|
|
assert result.returncode != 0
|
|
assert 'public recipient' in result.stderr
|
|
assert (secrets / 'service/one.env').exists()
|
|
|
|
|
|
def test_decryption_does_not_fall_back_to_home_key(custody, tmp_path):
|
|
key, public, env = custody
|
|
encrypt_fixture(tmp_path, public)
|
|
home_key = Path(env['HOME']) / '.config/net-kingdom/age.key'
|
|
home_key.parent.mkdir(parents=True); home_key.write_bytes(key.read_bytes())
|
|
output = tmp_path / 'restored'
|
|
result = subprocess.run(['bash', str(BOOT / 'decrypt-secrets.sh'), str(output)],
|
|
env=env, capture_output=True, text=True, timeout=10)
|
|
assert result.returncode != 0
|
|
assert not output.exists()
|
|
|
|
|
|
def test_decryption_failure_removes_partial_plaintext(custody, tmp_path):
|
|
key, public, env = custody
|
|
_, encrypted = encrypt_fixture(tmp_path, public)
|
|
(encrypted / 'two.env.age').write_bytes(b'not-a-valid-ciphertext')
|
|
output = tmp_path / 'restored'
|
|
result = subprocess.run(['bash', str(BOOT / 'decrypt-secrets.sh'), str(output), str(key)],
|
|
env=env, capture_output=True, text=True, timeout=10)
|
|
assert result.returncode != 0
|
|
assert not output.exists()
|
|
assert not list(tmp_path.glob('.netkingdom-decrypt.*'))
|
|
|
|
|
|
def test_explicit_decryption_has_private_modes_and_no_output(custody, tmp_path):
|
|
key, public, env = custody
|
|
original, _ = encrypt_fixture(tmp_path, public)
|
|
output = tmp_path / 'restored'
|
|
result = subprocess.run(['bash', str(BOOT / 'decrypt-secrets.sh'), str(output), str(key)],
|
|
env=env, capture_output=True, text=True, timeout=10)
|
|
assert result.returncode == 0
|
|
restored = output / 'service/one.env'
|
|
assert restored.read_bytes() == (original / 'service/one.env').read_bytes()
|
|
assert output.stat().st_mode & 0o077 == 0
|
|
assert restored.stat().st_mode & 0o077 == 0
|
|
assert 'synthetic-private-payload' not in result.stdout + result.stderr
|
|
|
|
|
|
def test_default_encryption_uses_registered_public_file(custody, tmp_path):
|
|
import shutil
|
|
_, public, env = custody
|
|
checkout = tmp_path / 'fixture-repo'
|
|
script_dir = checkout / 'sso-mfa/bootstrap'; script_dir.mkdir(parents=True)
|
|
shutil.copyfile(BOOT / 'encrypt-secrets.sh', script_dir / 'encrypt-secrets.sh')
|
|
(checkout / 'keys').mkdir(); (checkout / 'keys/age.pub').write_text(public + '\n')
|
|
secrets = tmp_path / 'plaintext'; (secrets / 'service').mkdir(parents=True)
|
|
(secrets / 'service/test.env').write_text('TEST=synthetic-private-payload\n')
|
|
result = subprocess.run(['bash', str(script_dir / 'encrypt-secrets.sh'), str(secrets)],
|
|
env=env, capture_output=True, text=True, timeout=10)
|
|
assert result.returncode == 0
|
|
assert not secrets.exists()
|
|
assert (tmp_path / 'secrets.enc/service/test.env.age').is_file()
|
|
assert not (Path(env['HOME']) / '.config').exists()
|
|
|
|
|
|
def test_custody_child_does_not_inherit_competing_inline_key(custody):
|
|
key, public, env = custody
|
|
env = {**env, 'SOPS_AGE_KEY': 'synthetic-competing-key', 'SOPS_AGE_KEY_CMD': 'false'}
|
|
result = unlock((key, public, env), '--from-file', str(key), '--', 'sh', '-c',
|
|
'test -z "${SOPS_AGE_KEY+x}" && test -z "${SOPS_AGE_KEY_CMD+x}" && test -f "$SOPS_AGE_KEY_FILE"')
|
|
assert result.returncode == 0
|
|
|
|
|
|
@pytest.mark.parametrize('kind', ['file', 'directory', 'dangling-symlink'])
|
|
def test_decryption_never_overwrites_existing_output(custody, tmp_path, kind):
|
|
key, public, env = custody
|
|
encrypt_fixture(tmp_path, public)
|
|
output = tmp_path / 'restored'
|
|
if kind == 'file': output.write_bytes(b'existing-output')
|
|
elif kind == 'directory': output.mkdir(); (output / 'existing').write_bytes(b'existing-output')
|
|
else: output.symlink_to(tmp_path / 'uncreated')
|
|
result = subprocess.run(['bash', str(BOOT / 'decrypt-secrets.sh'), str(output), str(key)],
|
|
env=env, capture_output=True, text=True, timeout=10)
|
|
assert result.returncode != 0
|
|
if kind == 'file': assert output.read_bytes() == b'existing-output'
|
|
elif kind == 'directory': assert (output / 'existing').read_bytes() == b'existing-output'
|
|
else: assert output.is_symlink(); assert not (tmp_path / 'uncreated').exists()
|