tegwick
35880bbce4
keycape-client.py --check-config runs KeyCape's startup checks (key-cape c528daa) on the live and proposed registry via stdin; apply refuses a registry KeyCape would refuse at start. Type 1.1 patch names the option. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 736651@bnt-lap001 Assistant-Session: f6189624-fd64-4109-b53e-f8a59ca8a2d8
250 lines
11 KiB
Python
250 lines
11 KiB
Python
import argparse
|
|
import base64
|
|
import contextlib
|
|
import copy
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import unittest
|
|
|
|
import yaml
|
|
|
|
spec = importlib.util.spec_from_file_location('keycape_client', Path(__file__).with_name('keycape-client.py'))
|
|
kc = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(kc)
|
|
|
|
vspec = importlib.util.spec_from_file_location('vergabe_rollout', Path(__file__).with_name('vergabe-client-rollout.py'))
|
|
|
|
UID = '00000000-0000-4000-8000-000000000001'
|
|
EXISTING = [
|
|
{'clientId': 'openbao-admin', 'clientType': 'public',
|
|
'redirectUris': ['http://localhost:8250/oidc/callback']},
|
|
{'clientId': 'secrets-engine-approval', 'clientType': 'confidential',
|
|
'secretRef': 'env:KEYCAPE_SECRETS_ENGINE_APPROVAL_CLIENT_SECRET'},
|
|
]
|
|
FORGEJO_PARAMS = {
|
|
'client_id': 'forgejo', 'display_name': 'Forgejo source forge',
|
|
'redirect_uris': 'https://forgejo.coulomb.social/user/oauth2/netkingdom/callback',
|
|
'allowed_scopes': 'openid,profile,email,groups', 'client_type': 'confidential',
|
|
'mfa_required': 'true',
|
|
}
|
|
FORGEJO = kc.client_from_params(FORGEJO_PARAMS)
|
|
|
|
|
|
def fixture(clients=None, tail='tokenLifetime: 15m\n'):
|
|
# Synthetic marker values only; no live Secret is loaded by this suite.
|
|
raw = ('# preserve this comment\nissuer: https://kc.coulomb.social\n'
|
|
'authelia:\n issuer: https://auth.coulomb.social\n clientSecret: synthetic-marker\n'
|
|
'clients:\n' + yaml.safe_dump(json.loads(json.dumps(EXISTING if clients is None else clients)),
|
|
sort_keys=False) + tail)
|
|
secret = {'metadata': {'uid': UID, 'resourceVersion': '7'},
|
|
'data': {'config.yaml': base64.b64encode(raw.encode()).decode(),
|
|
'signing-key.pem': 'c3ludGhldGljLXNpZ25pbmcta2V5'}}
|
|
return secret, raw
|
|
|
|
|
|
def decode(encoded):
|
|
return base64.b64decode(encoded).decode()
|
|
|
|
|
|
class ClientRecordTests(unittest.TestCase):
|
|
def test_confidential_record_binds_env_secret_and_pins_mfa(self):
|
|
self.assertEqual(FORGEJO['secretRef'], 'env:KEYCAPE_FORGEJO_CLIENT_SECRET')
|
|
self.assertIs(FORGEJO['mfaRequired'], True)
|
|
self.assertEqual(FORGEJO['grantTypes'], ['authorization_code'])
|
|
self.assertEqual(kc.env_binding(FORGEJO)['valueFrom']['secretKeyRef']['name'], 'keycape-forgejo-client')
|
|
|
|
def test_public_record_has_no_secret(self):
|
|
params = dict(FORGEJO_PARAMS, client_type='public', mfa_required='false')
|
|
client = kc.client_from_params(params)
|
|
self.assertNotIn('secretRef', client)
|
|
self.assertNotIn('mfaRequired', client)
|
|
|
|
def test_invalid_params_refused(self):
|
|
for change in [dict(redirect_uris='https://x.example/*'), dict(redirect_uris='http://x.example/cb'),
|
|
dict(allowed_scopes='profile,openid'), dict(allowed_scopes='openid,admin'),
|
|
dict(client_type='weird'), dict(secret_ref='bao:platform/workloads/forgejo/x'), dict(client_id='Bad_Id')]:
|
|
with self.subTest(change=change), self.assertRaises(kc.Refused):
|
|
kc.client_from_params(dict(FORGEJO_PARAMS, **change))
|
|
|
|
def test_vergabe_case_reproduces_the_hand_built_registration(self):
|
|
vergabe = importlib.util.module_from_spec(vspec)
|
|
vspec.loader.exec_module(vergabe)
|
|
derived = kc.client_from_params({
|
|
'client_id': 'vergabe-demo-company', 'display_name': 'Vergabe Demo Company',
|
|
'redirect_uris': vergabe.CLIENT['redirectUris'][0],
|
|
'allowed_scopes': ','.join(vergabe.CLIENT['allowedScopes']),
|
|
'client_type': 'public', 'mfa_required': 'false'})
|
|
self.assertEqual(derived, vergabe.CLIENT)
|
|
|
|
def test_client_from_case_requires_the_type(self):
|
|
case = {'spec': {'type': 'oidc-browser-relying-party@1.0', 'parameters': FORGEJO_PARAMS}}
|
|
self.assertEqual(kc.client_from_case(case), FORGEJO)
|
|
with self.assertRaises(kc.Refused):
|
|
kc.client_from_case({'spec': {'type': 'other@1.0', 'parameters': FORGEJO_PARAMS}})
|
|
|
|
def test_redirects_keycape_would_refuse_at_startup_are_refused(self):
|
|
# key-cape type review R1: a refused entry stops the single replica.
|
|
for bad in ['https://forgejo.example/cb?x=1', 'https://forgejo.example/cb#frag']:
|
|
with self.subTest(redirect=bad), self.assertRaises(kc.Refused):
|
|
kc.client_from_params(dict(FORGEJO_PARAMS, redirect_uris=bad))
|
|
|
|
def test_access_groups_require_groups_scope(self):
|
|
# key-cape type review R2: UserInfo returns groups only for that scope.
|
|
with self.assertRaises(kc.Refused):
|
|
kc.client_from_params(dict(FORGEJO_PARAMS, access_groups='forge-admins',
|
|
allowed_scopes='openid,profile,email'))
|
|
|
|
|
|
class MergeTests(unittest.TestCase):
|
|
def test_add_appends_only_this_client_and_preserves_bytes(self):
|
|
secret, raw = fixture()
|
|
before = copy.deepcopy(secret)
|
|
encoded, changed = kc.replacement(secret, FORGEJO)
|
|
updated = decode(encoded)
|
|
self.assertTrue(changed)
|
|
head = raw[:raw.index('tokenLifetime:')]
|
|
self.assertTrue(updated.startswith(head))
|
|
expected = yaml.safe_load(raw)
|
|
expected['clients'].append(FORGEJO)
|
|
self.assertEqual(yaml.safe_load(updated), expected)
|
|
self.assertEqual(secret, before)
|
|
|
|
def test_identical_is_noop_and_drift_refused(self):
|
|
secret, _ = fixture(EXISTING + [FORGEJO])
|
|
self.assertEqual(kc.replacement(secret, FORGEJO), (secret['data']['config.yaml'], False))
|
|
for change in [dict(redirectUris=['https://evil.example/cb']), dict(clientType='public'),
|
|
dict(allowedScopes=['openid'])]:
|
|
with self.subTest(change=change), self.assertRaises(kc.Refused):
|
|
kc.replacement(fixture(EXISTING + [dict(FORGEJO, **change)])[0], FORGEJO)
|
|
|
|
def test_shared_secret_ref_and_duplicates_refused(self):
|
|
other = dict(EXISTING[1], secretRef=FORGEJO['secretRef'])
|
|
with self.assertRaises(kc.Refused):
|
|
kc.replacement(fixture([EXISTING[0], other])[0], FORGEJO)
|
|
with self.assertRaises(kc.Refused):
|
|
kc.replacement(fixture(EXISTING + [EXISTING[0]])[0], FORGEJO)
|
|
|
|
def test_removal_removes_exactly_this_client(self):
|
|
secret, raw = fixture(EXISTING + [FORGEJO])
|
|
encoded, changed = kc.removal(secret, FORGEJO)
|
|
self.assertTrue(changed)
|
|
self.assertEqual(yaml.safe_load(decode(encoded))['clients'], EXISTING)
|
|
self.assertEqual(kc.removal(fixture()[0], FORGEJO), (fixture()[0]['data']['config.yaml'], False))
|
|
|
|
|
|
class FakeKube:
|
|
def __init__(self, secret, bound=True):
|
|
self.secret = secret
|
|
self.bound = bound
|
|
self.patches = []
|
|
|
|
def __call__(self, arguments, payload=None):
|
|
if arguments[:2] == ['get', 'namespace']:
|
|
return {'metadata': {'uid': kc.lane.CLUSTER_UID}}
|
|
if 'deployment' in arguments:
|
|
env = [kc.env_binding(FORGEJO)] if self.bound else []
|
|
return {'spec': {'template': {'spec': {'containers': [{'name': 'keycape', 'env': env}]}}}}
|
|
if 'patch' in arguments:
|
|
self.patches.append(arguments)
|
|
data = dict(self.secret['data'], **{'config.yaml': payload[2]['value']})
|
|
if '--dry-run=server' not in arguments:
|
|
self.secret = dict(self.secret, data=data)
|
|
return {'data': data}
|
|
return copy.deepcopy(self.secret)
|
|
|
|
|
|
def args(mode, **kw):
|
|
base = dict(mode=mode, operation='add', expected_uid=UID, expected_resource_version='7', case_acceptance=None,
|
|
check_config='keycape')
|
|
base.update(kw)
|
|
return argparse.Namespace(**base)
|
|
|
|
|
|
class FakeCheck:
|
|
"""Stands in for keycape check-config; refuses registries naming a client in `refuse`."""
|
|
def __init__(self, refuse=()):
|
|
self.refuse = refuse
|
|
self.calls = []
|
|
|
|
def __call__(self, binary, config_text, env_names, policy_store):
|
|
self.calls.append((binary, env_names, policy_store))
|
|
ids = [c['clientId'] for c in yaml.safe_load(config_text)['clients']]
|
|
errors = [f'client {i!r}: refused' for i in ids if i in self.refuse]
|
|
return {'valid': not errors, 'clients': len(ids), 'errors': errors}
|
|
|
|
|
|
class RunTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.original = kc.lane.kubectl, kc.check_config
|
|
kc.check_config = FakeCheck()
|
|
|
|
def tearDown(self):
|
|
kc.lane.kubectl, kc.check_config = self.original
|
|
|
|
def test_apply_needs_keycape_check_config(self):
|
|
kc.lane.kubectl = FakeKube(fixture()[0])
|
|
with self.assertRaises(kc.Refused):
|
|
kc.run(args('apply', case_acceptance='decision:x', check_config=None), FORGEJO)
|
|
|
|
def test_apply_refuses_a_registry_keycape_would_refuse(self):
|
|
fake = FakeKube(fixture()[0])
|
|
kc.lane.kubectl = fake
|
|
kc.check_config = FakeCheck(refuse=('forgejo',))
|
|
with self.assertRaises(kc.Refused):
|
|
kc.run(args('apply', case_acceptance='decision:x'), FORGEJO)
|
|
self.assertEqual([], fake.patches)
|
|
|
|
def test_inspect_reports_live_and_proposed_checks_with_deployment_env(self):
|
|
kc.lane.kubectl = FakeKube(fixture()[0])
|
|
kc.check_config = FakeCheck(refuse=('forgejo',))
|
|
receipt = kc.run(args('inspect'), FORGEJO)
|
|
self.assertTrue(receipt['check_config']['live']['valid'])
|
|
self.assertFalse(receipt['check_config']['proposed']['valid'])
|
|
self.assertEqual([kc.env_name('forgejo')], kc.check_config.calls[0][1])
|
|
|
|
def test_inspect_changes_nothing_and_emits_no_values(self):
|
|
fake = FakeKube(fixture()[0])
|
|
kc.lane.kubectl = fake
|
|
receipt = kc.run(args('inspect'), FORGEJO)
|
|
self.assertTrue(receipt['change_needed'])
|
|
self.assertFalse(receipt['values_emitted'])
|
|
self.assertEqual([], fake.patches)
|
|
self.assertNotIn('synthetic-marker', json.dumps(receipt))
|
|
|
|
def test_apply_needs_case_acceptance(self):
|
|
kc.lane.kubectl = FakeKube(fixture()[0])
|
|
with self.assertRaises(kc.Refused):
|
|
kc.run(args('apply'), FORGEJO)
|
|
|
|
def test_apply_needs_env_binding_for_confidential_clients(self):
|
|
kc.lane.kubectl = FakeKube(fixture()[0], bound=False)
|
|
with self.assertRaises(kc.Refused):
|
|
kc.run(args('apply', case_acceptance='decision:x'), FORGEJO)
|
|
|
|
def test_apply_refuses_a_changed_revision(self):
|
|
kc.lane.kubectl = FakeKube(fixture()[0])
|
|
with self.assertRaises(kc.Refused):
|
|
kc.run(args('apply', case_acceptance='decision:x', expected_resource_version='6'), FORGEJO)
|
|
|
|
def test_apply_patches_reads_back_and_records_acceptance(self):
|
|
fake = FakeKube(fixture()[0])
|
|
kc.lane.kubectl = fake
|
|
receipt = kc.run(args('apply', case_acceptance='decision:x'), FORGEJO)
|
|
self.assertTrue(receipt['changed'])
|
|
self.assertTrue(receipt['restart_required'])
|
|
self.assertEqual('decision:x', receipt['case_acceptance'])
|
|
self.assertEqual(1, len(fake.patches))
|
|
self.assertIn(FORGEJO, yaml.safe_load(decode(fake.secret['data']['config.yaml']))['clients'])
|
|
|
|
def test_main_contains_failures(self):
|
|
out = io.StringIO()
|
|
with contextlib.redirect_stdout(out):
|
|
code = kc.main(['inspect'])
|
|
self.assertEqual(1, code)
|
|
self.assertEqual('refused', json.loads(out.getvalue())['status'])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|