"""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()