diff --git a/sso-mfa/bootstrap/reconcile-keycape-client-secret.py b/sso-mfa/bootstrap/reconcile-keycape-client-secret.py new file mode 100755 index 0000000..ee29612 --- /dev/null +++ b/sso-mfa/bootstrap/reconcile-keycape-client-secret.py @@ -0,0 +1,81 @@ +#!/usr/bin/env python3 +"""Reconcile custody AUTHELIA_KEYCAPE_CLIENT_SECRET with the live value (NK-WP-0056-T04). + +Reads the plaintext KeyCape uses (sso/keycape-config, config.yaml authelia.clientSecret) +and accepts it only if it matches Authelia's live bcrypt hash (sso/authelia-secrets, +keycape_client_secret_hash). Without --apply it reports value-free whether custody +matches. With --apply it writes the live value into the decrypted custody copy through +rotation-env.py (stdin only) and re-checks custody against the hash. No cluster writes. + +Usage: reconcile-keycape-client-secret.py [--apply] +Needs KUBECONFIG with read access to Secrets in namespace sso. +""" +import base64 +import json +from pathlib import Path +import subprocess +import sys + +import bcrypt +import yaml + +HERE = Path(__file__).resolve().parent +KEY = 'AUTHELIA_KEYCAPE_CLIENT_SECRET' + + +class Refused(Exception): + pass + + +def secret_data(name, key): + raw = subprocess.run(['kubectl', 'get', 'secret', name, '-n', 'sso', '-o', 'json'], + check=True, capture_output=True).stdout + data = json.loads(raw).get('data', {}) + if key not in data: raise Refused() + return base64.b64decode(data[key], validate=True) + + +def custody_value(root): + sys.path.insert(0, str(HERE)) + import importlib.util + spec = importlib.util.spec_from_file_location('custody_io', HERE / 'apply-service-secrets.py') + module = importlib.util.module_from_spec(spec); spec.loader.exec_module(module) + with module.custody(root, ['authelia']) as dirs: + return module.literal(module.read_env(dirs['authelia']), KEY).encode() + + +def main(): + args = sys.argv[1:] + apply = '--apply' in args + roots = [a for a in args if a != '--apply'] + if len(roots) != 1: + print(__doc__.strip().splitlines()[-2], file=sys.stderr) + return 2 + root = roots[0] + phase = 'read' + try: + hashed = secret_data('authelia-secrets', 'keycape_client_secret_hash') + config = yaml.safe_load(secret_data('keycape-config', 'config.yaml')) + live = str(config['authelia']['clientSecret']).encode() + if not bcrypt.checkpw(live, hashed): raise Refused() # live pair must agree first + before = bcrypt.checkpw(custody_value(root), hashed) + report = {'live_keycape_matches_authelia_hash': True, 'custody_matched_before': before, + 'values_recorded': False} + if apply and not before: + phase = 'write' + subprocess.run([sys.executable, str(HERE / 'rotation-env.py'), 'update', root, 'authelia', KEY], + input=live + b'\n', check=True, capture_output=True) + phase = 'readback' + if not bcrypt.checkpw(custody_value(root), hashed): raise Refused() + report['custody_updated'] = True + report['custody_matches_live'] = bcrypt.checkpw(custody_value(root), hashed) + print(json.dumps(report, sort_keys=True)) + return 0 if report['custody_matches_live'] or not apply else 1 + except (Exception, KeyboardInterrupt): + print(json.dumps({'result': 'unverified', 'phase': phase, + 'partial_changes_possible': phase in ('write', 'readback')}), file=sys.stderr) + return 1 + + +if __name__ == '__main__': + raise SystemExit(main()) diff --git a/tests/test_reconcile_keycape_client_secret.py b/tests/test_reconcile_keycape_client_secret.py new file mode 100644 index 0000000..7dea1e4 --- /dev/null +++ b/tests/test_reconcile_keycape_client_secret.py @@ -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 diff --git a/workplans/NK-WP-0056-bounded-executor-path.md b/workplans/NK-WP-0056-bounded-executor-path.md index f42d697..cefafb3 100644 --- a/workplans/NK-WP-0056-bounded-executor-path.md +++ b/workplans/NK-WP-0056-bounded-executor-path.md @@ -135,6 +135,10 @@ agent handles the values in a rotation. First reconcile the stale custody **Done when:** the four service binds are ESO-delivered from OpenBao, and the custody copy no longer needs to be decrypted for a bind rotation. +**2026-10-30:** reconciliation tool prepared: `sso-mfa/bootstrap/reconcile-keycape-client-secret.py` +(value-free; accepts the live KeyCape value only if it matches Authelia's live hash, writes custody through +`rotation-env.py`, no cluster writes; `tests/test_reconcile_keycape_client_secret.py`). Founder run pending. + ## Synthetic verification instead of a founder sign-in ```task