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
87 lines
3.6 KiB
Python
Executable file
87 lines
3.6 KiB
Python
Executable file
#!/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 <custody-root> [--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 = 'custody'
|
|
try:
|
|
custody_value(root) # fail first, and distinctly, if the decrypted copy is absent
|
|
phase = 'cluster'
|
|
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()
|
|
phase = 'live-pair'
|
|
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) as error:
|
|
# Class and exit code only: messages and kubectl stderr may carry values.
|
|
detail = {'error': type(error).__name__}
|
|
if isinstance(error, subprocess.CalledProcessError): detail['exit_code'] = error.returncode
|
|
print(json.dumps({'result': 'unverified', 'phase': phase,
|
|
'partial_changes_possible': phase in ('write', 'readback'), **detail}), file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
if __name__ == '__main__':
|
|
raise SystemExit(main())
|