tegwick
70f408ca23
Some checks failed
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a1468b-63fa-7710-85dc-8f2e16b3ebfa
261 lines
11 KiB
Python
261 lines
11 KiB
Python
"""Use real SOPS with disposable age custody; no repository secret inputs."""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import subprocess
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
SOPS = shutil.which('sops')
|
|
pytestmark = pytest.mark.skipif(SOPS is None, reason='SOPS required for real crypto regression')
|
|
|
|
|
|
@pytest.mark.parametrize('wrong_recipient', [False, True])
|
|
def test_real_sops_decrypt_with_only_sealed_descriptor(case, wrong_recipient):
|
|
import fcntl
|
|
original = case[1].read_bytes()
|
|
material = (subprocess.run(['age-keygen'], capture_output=True, check=True).stdout
|
|
if wrong_recipient else original)
|
|
fd = os.memfd_create('generated-sops-custody', os.MFD_ALLOW_SEALING)
|
|
try:
|
|
os.fchmod(fd, 0o600); os.write(fd, material); os.lseek(fd, 0, os.SEEK_SET)
|
|
fcntl.fcntl(fd, fcntl.F_ADD_SEALS,
|
|
fcntl.F_SEAL_WRITE | fcntl.F_SEAL_GROW | fcntl.F_SEAL_SHRINK | fcntl.F_SEAL_SEAL)
|
|
case[1].unlink()
|
|
env = {**case[2], 'NETKINGDOM_CUSTODY_AGE_KEY_FD': str(fd)}
|
|
env.pop('NETKINGDOM_CUSTODY_AGE_KEY_FILE')
|
|
if wrong_recipient:
|
|
# A correct ambient key must never rescue the wrong explicit FD.
|
|
env['SOPS_AGE_KEY'] = original.decode()
|
|
result = subprocess.run(['make', '-s', 'sops-decrypt', 'FILE=' + str(case[3]),
|
|
'OUTPUT=' + str(case[4])], cwd=case[0], env=env,
|
|
pass_fds=(fd,), capture_output=True, text=True, timeout=15)
|
|
assert not case[1].exists()
|
|
assert 'synthetic-secret-payload' not in result.stdout + result.stderr
|
|
if wrong_recipient:
|
|
assert result.returncode != 0
|
|
assert not case[4].exists()
|
|
else:
|
|
assert result.returncode == 0
|
|
assert json.loads(case[4].read_text()) == {'credential': 'synthetic-secret-payload'}
|
|
assert case[4].stat().st_mode & 0o077 == 0
|
|
assert not list(case[4].parent.glob('.netkingdom-sops-*'))
|
|
finally:
|
|
os.close(fd)
|
|
|
|
|
|
@pytest.fixture
|
|
def case(tmp_path):
|
|
repo = tmp_path / 'repo'; (repo / 'sso-mfa/bootstrap').mkdir(parents=True)
|
|
shutil.copyfile(ROOT / 'Makefile', repo / 'Makefile')
|
|
shutil.copyfile(ROOT / 'sso-mfa/bootstrap/sops-private.py', repo / 'sso-mfa/bootstrap/sops-private.py')
|
|
(repo / 'keys').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()
|
|
(repo / 'keys/age.pub').write_text(public + '\n')
|
|
home = tmp_path / 'home'; home.mkdir(mode=0o700)
|
|
env = {'HOME': str(home), 'PATH': os.environ['PATH'], 'NETKINGDOM_CUSTODY_AGE_KEY_FILE': str(key)}
|
|
clear = tmp_path / 'input.json'; clear.write_text(json.dumps({'credential': 'synthetic-secret-payload'}))
|
|
cipher = repo / 'cipher.json'
|
|
encrypted = subprocess.run([SOPS, '--encrypt', '--age', public, str(clear)], cwd=tmp_path,
|
|
env=env, capture_output=True, check=True)
|
|
cipher.write_bytes(encrypted.stdout)
|
|
output_dir = tmp_path / 'private-output'; output_dir.mkdir(mode=0o700)
|
|
return repo, key, env, cipher, output_dir / 'restored.json'
|
|
|
|
|
|
def make(case, target='sops-decrypt', *, env=None, source=None, output=None):
|
|
repo, _, default_env, cipher, destination = case
|
|
return subprocess.run(['make', '-s', target, 'FILE=' + str(source or cipher),
|
|
'OUTPUT=' + str(output or destination)],
|
|
env=env or default_env, cwd=repo, capture_output=True, text=True, timeout=15)
|
|
|
|
|
|
def ambient_key(case):
|
|
_, key, env, _, _ = case
|
|
path = Path(env['HOME']) / '.config/sops/age/keys.txt'
|
|
path.parent.mkdir(parents=True); path.write_bytes(key.read_bytes()); path.chmod(0o600)
|
|
|
|
|
|
def test_setup_never_installs_private_key(case):
|
|
_, key, env, _, _ = case
|
|
legacy = Path(env['HOME']) / '.config/age/key.txt'
|
|
legacy.parent.mkdir(parents=True); legacy.write_bytes(key.read_bytes())
|
|
result = make(case, 'sops-setup')
|
|
assert result.returncode == 0
|
|
assert not (Path(env['HOME']) / '.config/sops').exists()
|
|
|
|
|
|
def test_decrypt_real_sops_to_private_output_without_transcript(case):
|
|
ambient_key(case) # Old target succeeds through ambient discovery and leaks.
|
|
result = make(case)
|
|
assert result.returncode == 0
|
|
assert 'synthetic-secret-payload' not in result.stdout + result.stderr
|
|
assert json.loads(case[4].read_text()) == {'credential': 'synthetic-secret-payload'}
|
|
assert case[4].stat().st_mode & 0o077 == 0
|
|
assert not list(case[4].parent.glob('.netkingdom-sops-*'))
|
|
|
|
|
|
def test_wrong_explicit_key_never_falls_back_to_home_or_inline_keys(case, tmp_path):
|
|
ambient_key(case)
|
|
wrong = tmp_path / 'wrong.key'
|
|
subprocess.run(['age-keygen', '-o', str(wrong)], capture_output=True, check=True)
|
|
env = {**case[2], 'NETKINGDOM_CUSTODY_AGE_KEY_FILE': str(wrong), 'SOPS_AGE_KEY': case[1].read_text()}
|
|
result = make(case, env=env)
|
|
assert result.returncode != 0
|
|
assert 'synthetic-secret-payload' not in result.stdout + result.stderr
|
|
assert not case[4].exists()
|
|
|
|
|
|
def test_absent_explicit_custody_refused_even_if_home_has_key(case):
|
|
ambient_key(case)
|
|
env = dict(case[2]); del env['NETKINGDOM_CUSTODY_AGE_KEY_FILE']
|
|
assert make(case, env=env).returncode != 0
|
|
assert not case[4].exists()
|
|
|
|
|
|
@pytest.mark.parametrize('kind', ['file', 'symlink'])
|
|
def test_existing_output_is_not_replaced(case, kind):
|
|
ambient_key(case)
|
|
if kind == 'file': case[4].write_bytes(b'original-output')
|
|
else: case[4].symlink_to(case[4].parent / 'absent')
|
|
assert make(case).returncode != 0
|
|
if kind == 'file': assert case[4].read_bytes() == b'original-output'
|
|
else: assert case[4].is_symlink()
|
|
|
|
|
|
def test_git_worktree_output_is_refused(case):
|
|
ambient_key(case)
|
|
subprocess.run(['git', 'init', str(case[0])], env=case[2], capture_output=True, check=True)
|
|
assert make(case, output=case[0] / 'plaintext.json').returncode != 0
|
|
assert not (case[0] / 'plaintext.json').exists()
|
|
|
|
|
|
def test_shared_output_directory_is_refused(case):
|
|
ambient_key(case)
|
|
case[4].parent.chmod(0o755)
|
|
assert make(case).returncode != 0
|
|
assert not case[4].exists()
|
|
|
|
|
|
def noisy_tool(case, tmp_path):
|
|
bin_dir = tmp_path / 'bin'; bin_dir.mkdir()
|
|
tool = bin_dir / 'sops'
|
|
tool.write_text('#!/bin/sh\necho synthetic-tool-secret\necho synthetic-error-secret >&2\nexit 23\n')
|
|
tool.chmod(0o755)
|
|
return {**case[2], 'PATH': str(bin_dir) + ':' + case[2]['PATH']}
|
|
|
|
|
|
def test_child_output_and_errors_are_not_exposed_on_failure(case, tmp_path):
|
|
result = make(case, env=noisy_tool(case, tmp_path))
|
|
assert result.returncode != 0
|
|
assert 'synthetic-tool-secret' not in result.stdout + result.stderr
|
|
assert 'synthetic-error-secret' not in result.stdout + result.stderr
|
|
assert not case[4].exists()
|
|
assert not list(case[4].parent.glob('.netkingdom-sops-*'))
|
|
|
|
|
|
def test_non_age_metadata_refuses_before_tool(case, tmp_path):
|
|
data = json.loads(case[3].read_text()); data['sops']['kms'] = [{'arn': 'synthetic'}]
|
|
case[3].write_text(json.dumps(data))
|
|
result = make(case, env=noisy_tool(case, tmp_path))
|
|
assert result.returncode != 0
|
|
assert 'age-only metadata' in result.stderr
|
|
assert 'synthetic-tool-secret' not in result.stdout
|
|
|
|
|
|
def test_filename_is_not_shell_code(case):
|
|
ambient_key(case)
|
|
odd = case[0] / 'cipher; touch SHELL_EXECUTED; #.json'
|
|
shutil.copyfile(case[3], odd)
|
|
result = make(case, source=odd)
|
|
assert result.returncode == 0
|
|
assert not (case[0] / 'SHELL_EXECUTED').exists()
|
|
assert case[4].exists()
|
|
|
|
|
|
def test_yaml_roundtrip(case, tmp_path):
|
|
public = (case[0] / 'keys/age.pub').read_text().strip()
|
|
clear = tmp_path / 'input.yaml'; clear.write_text('credential: synthetic-yaml-secret\n')
|
|
cipher = case[0] / 'cipher.yaml'
|
|
result = subprocess.run([SOPS, '--encrypt', '--age', public, str(clear)], cwd=tmp_path,
|
|
env=case[2], capture_output=True, check=True)
|
|
cipher.write_bytes(result.stdout)
|
|
result = make(case, source=cipher)
|
|
assert result.returncode == 0
|
|
assert 'synthetic-yaml-secret' not in result.stdout + result.stderr
|
|
assert 'synthetic-yaml-secret' in case[4].read_text()
|
|
|
|
|
|
def test_parser_error_never_echoes_input(case):
|
|
case[3].write_text('synthetic-malformed-secret')
|
|
result = make(case)
|
|
assert result.returncode != 0
|
|
assert 'synthetic-malformed-secret' not in result.stdout + result.stderr
|
|
assert not case[4].exists()
|
|
|
|
|
|
def test_shared_custody_key_is_refused(case):
|
|
case[1].chmod(0o644)
|
|
result = make(case)
|
|
assert result.returncode != 0
|
|
assert not case[4].exists()
|
|
|
|
|
|
def test_creds_init_has_no_permanent_key_requirement(case):
|
|
subprocess.run(['git', 'init', str(case[0])], env=case[2], capture_output=True, check=True)
|
|
hooks = case[0] / '.githooks'; hooks.mkdir()
|
|
(hooks / 'pre-commit').write_text('#!/bin/sh\nexit 0\n')
|
|
result = make(case, 'creds-init')
|
|
assert result.returncode == 0
|
|
assert not (Path(case[2]['HOME']) / '.config/sops').exists()
|
|
|
|
|
|
def test_real_sops_does_not_execute_ambient_key_commands(case, tmp_path):
|
|
marker = tmp_path / 'ambient-command-executed'
|
|
command = tmp_path / 'ambient-key-command'
|
|
command.write_text('#!/bin/sh\ntouch ' + str(marker) + '\nexit 1\n')
|
|
command.chmod(0o755)
|
|
env = {**case[2], 'SOPS_AGE_KEY_CMD': str(command), 'SOPS_AGE_SSH_PRIVATE_KEY_CMD': str(command)}
|
|
result = make(case, env=env)
|
|
assert result.returncode == 0
|
|
assert not marker.exists()
|
|
|
|
|
|
def test_yaml_parser_error_is_sanitized(case):
|
|
bad = case[0] / 'broken.yaml'
|
|
bad.write_text('secret: [synthetic-malformed-yaml-secret\n')
|
|
result = make(case, source=bad)
|
|
assert result.returncode != 0
|
|
assert 'synthetic-malformed-yaml-secret' not in result.stdout + result.stderr
|
|
assert 'Traceback' not in result.stderr
|
|
|
|
|
|
@pytest.mark.parametrize('kind', ['json', 'yaml'])
|
|
def test_duplicate_metadata_refused_before_sops(case, kind):
|
|
source = case[0] / ('duplicate.' + kind)
|
|
source.write_text('{"sops": {}, "sops": {}}' if kind == 'json' else 'sops: {}\nsops: {}\n')
|
|
result = make(case, source=source)
|
|
assert result.returncode != 0
|
|
assert 'duplicate' in result.stderr
|
|
|
|
|
|
@pytest.mark.parametrize('kind', ['bare', 'linked'])
|
|
def test_bare_and_linked_git_outputs_are_refused(case, kind):
|
|
location = case[4].parent / 'git-output'
|
|
if kind == 'bare':
|
|
subprocess.run(['git', 'init', '--bare', str(location)], env=case[2], capture_output=True, check=True)
|
|
else:
|
|
subprocess.run(['git', 'init', str(case[0])], env=case[2], capture_output=True, check=True)
|
|
subprocess.run(['git', '-c', 'user.name=Test', '-c', 'user.email=test@example.invalid',
|
|
'commit', '--allow-empty', '-m', 'fixture'], cwd=case[0], env=case[2],
|
|
capture_output=True, check=True)
|
|
subprocess.run(['git', 'worktree', 'add', '--detach', str(location)], cwd=case[0], env=case[2],
|
|
capture_output=True, check=True)
|
|
location.chmod(0o700)
|
|
result = make(case, output=location / 'plaintext.json')
|
|
assert result.returncode != 0
|
|
assert not (location / 'plaintext.json').exists()
|