NK-WP-0056-T04: value-free custody reconciliation for the KeyCape client secret
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
This commit is contained in:
parent
14e7056469
commit
05d63e05bd
3 changed files with 172 additions and 0 deletions
87
tests/test_reconcile_keycape_client_secret.py
Normal file
87
tests/test_reconcile_keycape_client_secret.py
Normal file
|
|
@ -0,0 +1,87 @@
|
|||
"""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': 'read', 'partial_changes_possible': False}
|
||||
assert env.read_bytes() == before
|
||||
Loading…
Add table
Add a link
Reference in a new issue