tegwick
b19f7f6493
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 673304@bnt-lap001 Assistant-Session: 959b12c7-cf2e-440b-a60d-91229cc9e96e
105 lines
4.2 KiB
Python
105 lines
4.2 KiB
Python
"""Custody reconciliation of the KeyCape client secret against synthetic files and a read-only kubectl stub."""
|
|
import base64
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
|
|
import bcrypt
|
|
import pytest
|
|
import yaml
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
TOOL = ROOT / 'sso-mfa/bootstrap/reconcile-keycape-client-secret.py'
|
|
LIVE = 'LiveSyntheticClientCredential1234567890AB'
|
|
STALE = 'StaleSyntheticClientCredential1234567890A'
|
|
|
|
|
|
@pytest.fixture
|
|
def case(tmp_path):
|
|
private = tmp_path / 'private'; private.mkdir(mode=0o700)
|
|
authelia = private / 'authelia'; authelia.mkdir(mode=0o700)
|
|
env = authelia / 'secrets.env'
|
|
env.write_text(f'AUTHELIA_JWT_SECRET={STALE}\nAUTHELIA_KEYCAPE_CLIENT_SECRET={STALE}\n'); env.chmod(0o600)
|
|
state = tmp_path / 'state'; state.mkdir()
|
|
hashed = bcrypt.hashpw(LIVE.encode(), bcrypt.gensalt(4))
|
|
(state / 'authelia-secrets').write_text(json.dumps(
|
|
{'data': {'keycape_client_secret_hash': base64.b64encode(hashed).decode()}}))
|
|
write_keycape(state, LIVE)
|
|
binary = tmp_path / 'bin'; binary.mkdir()
|
|
(binary / 'kubectl').write_text(f'''#!{sys.executable}
|
|
import os,sys
|
|
from pathlib import Path
|
|
args=sys.argv[1:]
|
|
with open(os.environ['CALLS'],'a') as f:f.write(' '.join(args)+'\\n')
|
|
if args[:2]!=['get','secret']:sys.exit(9)
|
|
sys.stdout.write((Path(os.environ['STATE'])/args[2]).read_text())
|
|
''')
|
|
(binary / 'kubectl').chmod(0o755)
|
|
environ = dict(os.environ, PATH=f'{binary}:{os.environ["PATH"]}', STATE=str(state), CALLS=str(tmp_path / 'calls'))
|
|
return private, env, state, environ, tmp_path / 'calls'
|
|
|
|
|
|
def write_keycape(state, value):
|
|
config = yaml.safe_dump({'authelia': {'clientSecret': value}}).encode()
|
|
(state / 'keycape-config').write_text(json.dumps({'data': {'config.yaml': base64.b64encode(config).decode()}}))
|
|
|
|
|
|
def run(case, *extra):
|
|
private, _, _, environ, _ = case
|
|
return subprocess.run([sys.executable, str(TOOL), str(private), *extra], env=environ, capture_output=True, text=True)
|
|
|
|
|
|
def test_preview_reports_stale_custody_without_writing(case):
|
|
_, env, *_ = case
|
|
before = env.read_bytes()
|
|
result = run(case)
|
|
assert result.returncode == 0
|
|
report = json.loads(result.stdout)
|
|
assert report == {'custody_matched_before': False, 'custody_matches_live': False,
|
|
'live_keycape_matches_authelia_hash': True, 'values_recorded': False}
|
|
assert env.read_bytes() == before
|
|
assert LIVE not in result.stdout + result.stderr and STALE not in result.stdout + result.stderr
|
|
|
|
|
|
def test_apply_writes_live_value_only_and_is_idempotent(case):
|
|
_, env, _, _, calls = case
|
|
result = run(case, '--apply')
|
|
assert result.returncode == 0, result.stderr
|
|
assert json.loads(result.stdout)['custody_updated'] is True
|
|
assert env.read_text() == f'AUTHELIA_JWT_SECRET={STALE}\nAUTHELIA_KEYCAPE_CLIENT_SECRET={LIVE}\n'
|
|
assert oct(env.stat().st_mode & 0o777) == '0o600'
|
|
again = run(case, '--apply')
|
|
assert again.returncode == 0
|
|
assert json.loads(again.stdout)['custody_matched_before'] is True and 'custody_updated' not in again.stdout
|
|
assert LIVE not in result.stdout + result.stderr + again.stdout
|
|
assert all(line.startswith('get secret ') for line in calls.read_text().splitlines())
|
|
|
|
|
|
def test_refuses_when_live_pair_disagrees(case):
|
|
_, env, state, _, _ = case
|
|
write_keycape(state, STALE)
|
|
before = env.read_bytes()
|
|
result = run(case, '--apply')
|
|
assert result.returncode == 1 and result.stdout == ''
|
|
assert json.loads(result.stderr) == {'result': 'unverified', 'phase': 'live-pair', 'partial_changes_possible': False,
|
|
'error': 'Refused'}
|
|
assert env.read_bytes() == before
|
|
|
|
|
|
def test_missing_custody_copy_is_reported_as_custody(case):
|
|
private, *_ = case
|
|
import shutil; shutil.rmtree(private)
|
|
result = run(case)
|
|
assert result.returncode == 1
|
|
assert json.loads(result.stderr)['phase'] == 'custody'
|
|
|
|
|
|
def test_cluster_failure_reports_exit_code_only(case):
|
|
*_, environ, _ = case
|
|
(Path(environ['STATE']) / 'keycape-config').unlink()
|
|
result = run(case)
|
|
report = json.loads(result.stderr)
|
|
assert report['phase'] == 'cluster' and report['error'] == 'CalledProcessError' and report['exit_code'] == 1
|
|
assert LIVE not in result.stderr
|