coulomb/railiance-platform
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
railiance-platform/tests/test_keycape_client_secret_generate.py
codex d35c0f9d4b
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
Details
CI Smoke / container-smoke (push) Successful in 2s
Details
Bind mint broker validation to exact reviewed sources and track successor obligations 
Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a1468b-63fa-7710-85dc-8f2e16b3ebfa
2026-10-17 22:55:51 +02:00

302 lines
17 KiB
Python
Raw Blame History

"""Offline tests for credential.py generate (RPF-WP-0054-T03) against a fake ``bao``.
Runs ``credential.py generate`` as a real subprocess, so the silence contract and
the delegation from credential.py are tested end to end. The fake records which
token each call used: KV calls must use the minted grant token, never the
caller's session.
"""
import hashlib
import importlib.util
import shutil
from unittest import mock
import yaml
import json
import os
from pathlib import Path
import subprocess
import sys
import tempfile
import textwrap
import unittest
ROOT = Path(__file__).resolve().parents[1]
CLI = ROOT / 'scripts/credential.py'
GRANT_TOKEN = 'hvs.grant-' + 'G7h8' * 6
FAKE_BAO = textwrap.dedent(f'''\
#!/usr/bin/env python3
import json, os, sys
sf = os.environ['FAKE_BAO_STATE']
st = json.load(open(sf))
a = sys.argv[1:]
tok = os.environ.get('BAO_TOKEN', 'caller-session')
st['calls'].append({{'args': a, 'token': tok}})
def save(): json.dump(st, open(sf, 'w'))
def out(obj): save(); print(json.dumps(obj)); sys.exit(0)
def missing(): save(); sys.stderr.write('No value found at x\\n'); sys.exit(2)
if a[:2] == ['token', 'lookup']: out({{'data': {{'policies': st['caller_policies']}}}})
if a[:2] == ['token', 'create']:
if st.get('fail_create'): save(); sys.exit(2)
out({{'auth': {{'client_token': '{GRANT_TOKEN}', 'accessor': 'acc'}}}})
if a[:2] == ['token', 'revoke']: st['revoked'] = tok; save(); sys.exit(0)
if a[0] == 'read':
rest = a[-1].split('/', 2)[2]
if rest not in st['kv']: missing()
n = len(st['kv'][rest])
out({{'data': {{'current_version': n, 'versions': {{str(i + 1): {{}} for i in range(n)}}}}}})
if a[0] == 'write':
rest = a[-2].split('/', 2)[2]
body = json.loads(sys.stdin.read())
st['written'] = body
if st.get('race'): st['kv'][rest] = [{{'x': 1}}]; save(); sys.exit(2)
if st.get('fail_write'): save(); sys.exit(2)
e = st['kv'].setdefault(rest, [])
if body['options']['cas'] != len(e): save(); sys.exit(2)
e.append(body['data'])
out({{'data': {{'version': len(e)}}}})
save(); sys.exit(9)
''')
# Exact accepted Forgejo @1.1 shape; materialized only in this fake test.
CASE = '# First KnownSecurityCase of oidc-browser-relying-party 1.1 (NK-WP-0048-T06).\n# Parameters are the request the founder accepted on 2026-10-08\n# (key-cape docs/forgejo-oidc-client-request.yaml, KEY-WP-0038-T01); that\n# acceptance carries over as this case\'s acceptance. One difference: custody is\n# derived by the type (platform/workloads/keycape/forgejo), not the request\'s\n# proposed platform/workloads/forgejo/netkingdom-oidc-client. The founder\n# acknowledged that derivation with the type (memo r4, 2026-10-09).\n# Re-pinned from 1.0 to 1.1 on 2026-10-10 on the founder\'s go, before any\n# apply, so no 1.0 run overlaps (minor change: consumer-enforces-groups).\napiVersion: netkingdom.io/security-case/v0.1\nkind: SecurityCase\nspec:\n id: forgejo-netkingdom-oidc\n type: oidc-browser-relying-party@1.1\n status: materialized\n acceptance:\n at: "2026-10-08"\n record: "key-cape docs/forgejo-oidc-client-request.yaml (status accepted-2026-10-08, KEY-WP-0038-T01); type acknowledged via informed-decision memo nk-wp-0048-t05-oidc-browser-relying-party-r4, hub decision e418263e"\n by:\n - {role: owner, name: "founder (for railiance-apps)"}\n - {role: founder, name: founder}\n owner:\n repo: railiance-apps\n person: founder\n parameters:\n client_id: forgejo\n display_name: Forgejo source forge\n redirect_uris: https://forgejo.coulomb.social/user/oauth2/netkingdom/callback\n allowed_scopes: openid,profile,email,groups\n client_type: confidential\n mfa_required: "true"\n access_groups: forge-admins,forge-coulomb-owners,forge-coulomb-readers\n initial_members: forge-admins=tegwick;forge-coulomb-owners=tegwick\n consuming_service: forgejo\n consumer_owner_repo: railiance-apps\n'
class Generate(unittest.TestCase):
def setUp(self):
self.tmp = Path(tempfile.mkdtemp())
(self.tmp / 'bin').mkdir()
(self.tmp / 'bin/bao').write_text(FAKE_BAO)
(self.tmp / 'bin/bao').chmod(0o755)
self.state = {'calls': [], 'kv': {},
'caller_policies': ['default', 'credential-broker-keycape-client-secret-generate-issuer']}
self.write_case(CASE)
def write_case(self, text):
self.case = self.tmp / 'case.yaml'
self.case.write_text(text)
self.ref = 'forgejo-netkingdom-oidc@sha256:' + hashlib.sha256(text.encode()).hexdigest()
def run_cli(self, *extra, client_id='forgejo', ref=None, env_extra=None):
(self.tmp / 'state.json').write_text(json.dumps(self.state))
receipt = self.tmp / f'r{len(list(self.tmp.glob("r*.json")))}.json'
empty_home = self.tmp / 'empty-home'
empty_home.mkdir(exist_ok=True)
env = {'PATH': os.defpath, 'HOME': str(empty_home)}
env.update(PATH=f'{self.tmp / "bin"}:{env["PATH"]}', FAKE_BAO_STATE=str(self.tmp / 'state.json'),
**(env_extra or {}))
proc = subprocess.run([sys.executable, str(CLI), 'generate', '--grant', 'keycape/client-secret-generate',
'--client-id', client_id, '--case-file', str(self.case),
'--case-ref', ref or self.ref, '--receipt', str(receipt), *extra],
env=env, capture_output=True, text=True)
self.assertEqual((proc.stdout, proc.stderr), ('', ''))
self.state = json.loads((self.tmp / 'state.json').read_text())
text = receipt.read_text()
self.assertNotIn(GRANT_TOKEN, text)
written = self.state.get('written')
if written:
self.assertNotIn(written['data']['CLIENT_SECRET'], text)
return proc.returncode, json.loads(text)
def kv_calls(self):
return [c for c in self.state['calls'] if c['args'][0] in ('read', 'write')]
def test_generates_once_with_grant_token_and_revokes_it(self):
code, receipt = self.run_cli()
self.assertEqual((code, receipt['outcome']), (0, 'generated'))
self.assertEqual(receipt['kv_version'], 1)
self.assertTrue(receipt['grant_token_revoked'])
self.assertEqual(self.state['revoked'], GRANT_TOKEN)
self.assertTrue(all(c['token'] == GRANT_TOKEN for c in self.kv_calls()))
self.assertFalse(any(c['args'][0] == 'read' and '/data/' in c['args'][-1] for c in self.kv_calls()))
value = self.state['written']['data']['CLIENT_SECRET']
self.assertEqual(len(value), 64)
self.assertEqual(self.state['written']['options'], {'cas': 0})
def test_existing_entry_is_never_touched(self):
self.state['kv']['workloads/keycape/forgejo'] = [{'CLIENT_SECRET': 'old'}]
code, receipt = self.run_cli()
self.assertEqual((code, receipt['outcome']), (3, 'already_present'))
self.assertNotIn('written', self.state)
self.assertTrue(receipt['grant_token_revoked'])
def test_lost_cas_race_reports_already_present(self):
self.state['race'] = True
code, receipt = self.run_cli()
self.assertEqual((code, receipt['outcome']), (3, 'already_present_lost_race'))
def test_dry_run_writes_nothing(self):
code, receipt = self.run_cli('--dry-run')
self.assertEqual((code, receipt['outcome']), (0, 'dry_run_would_generate'))
self.assertNotIn('written', self.state)
def test_case_checks_refuse_before_any_bao_call(self):
cases = [
('case_ref_mismatch', {'ref': 'forgejo-netkingdom-oidc@sha256:' + '0' * 64}),
('case_client_id_mismatch', {'client_id': 'gitea'}),
('client_id_invalid_or_reserved', {'client_id': 'secrets-engine-openbao'}),
('client_id_invalid_or_reserved', {'client_id': '../forgejo'}),
('case_ref_malformed', {'ref': 'forgejo'}),
]
for outcome, kwargs in cases:
with self.subTest(outcome=outcome, **kwargs):
self.state['calls'] = []
code, receipt = self.run_cli(**kwargs)
self.assertEqual((code, receipt['outcome']), (2, outcome))
self.assertEqual(self.state['calls'], [])
def test_unmaterialized_or_public_case_refused(self):
for text, outcome in ((CASE.replace('materialized', 'accepted'), 'case_not_materialized'),
(CASE.replace('confidential', 'public'), 'case_client_not_confidential')):
with self.subTest(outcome=outcome):
self.write_case(text)
code, receipt = self.run_cli()
self.assertEqual((code, receipt['outcome']), (2, outcome))
def test_only_exact_acknowledged_type_version_is_admitted(self):
for type_ref in ('oidc-browser-relying-party@1.0',
'oidc-browser-relying-party@1.2',
'oidc-browser-relying-party@1.3',
'oidc-browser-relying-party@99.0',
'confidential-service-client@1.1'):
with self.subTest(type_ref=type_ref):
self.write_case(CASE.replace('oidc-browser-relying-party@1.1', type_ref))
self.state['calls'] = []
code, receipt = self.run_cli()
self.assertEqual((code, receipt['outcome']), (2, 'case_type_unacknowledged'))
self.assertEqual(self.state['calls'], [])
def test_source_acknowledgment_alone_does_not_admit_12(self):
self.write_case(CASE.replace('oidc-browser-relying-party@1.1',
'oidc-browser-relying-party@1.2')
+ ' consumer_namespace: forgejo\n')
code, receipt = self.run_cli()
self.assertEqual((code, receipt['outcome']), (2, 'case_type_unacknowledged'))
self.assertEqual(self.state['calls'], [])
def test_caller_environment_cannot_select_type_or_source_root(self):
code, receipt = self.run_cli(env_extra={
'NETKINGDOM_ROOT': '/nonexistent',
'WARDEN_NETKINGDOM_ROOT': '/nonexistent',
'PYTHONPATH': '/nonexistent',
})
self.assertEqual((code, receipt['outcome']), (0, 'generated'))
def test_acknowledged_11_still_requires_its_exact_parameters(self):
self.write_case(CASE + ' unexpected_target: forgejo\n')
code, receipt = self.run_cli()
self.assertEqual((code, receipt['outcome']), (2, 'case_source_validation_failed'))
self.assertEqual(self.state['calls'], [])
def test_future_revision_requires_explicit_owner_binding_and_acknowledgment(self):
# Synthetic owner configuration exercises the extension mechanism only.
# This is not the actual proposed 1.3 type, acknowledgment or parameter pin.
spec = importlib.util.spec_from_file_location(
'future_fixture_generator', ROOT / 'scripts/keycape_client_secret_generate.py')
broker = importlib.util.module_from_spec(spec)
spec.loader.exec_module(broker)
source = self.tmp / 'synthetic-owner-root'
source.mkdir()
relative, digest = broker.ACKNOWLEDGED_TYPES['oidc-browser-relying-party@1.1']
for rel in (relative, *broker.SOURCE_PINS):
dest = source / rel
dest.parent.mkdir(parents=True, exist_ok=True)
shutil.copyfile(broker.NETKINGDOM_ROOT / rel, dest)
typ = source / relative
original = typ.read_bytes()
doc = yaml.safe_load(CASE)
doc['spec']['type'] = 'oidc-browser-relying-party@1.3'
typ.write_bytes(original.replace(b'version: "1.1"', b'version: "1.3"'))
# YAML numeric spelling is owner-authored, never an inferred revision.
type_doc = yaml.safe_load(original)
type_doc['spec']['version'] = '1.3'
typ.write_text(yaml.safe_dump(type_doc))
binding = {doc['spec']['type']: (relative, hashlib.sha256(typ.read_bytes()).hexdigest())}
with mock.patch.object(broker, 'NETKINGDOM_ROOT', source):
with self.assertRaises(broker.Exit):
broker.check_acknowledged_case(doc) # no explicit binding
with mock.patch.dict(broker.ACKNOWLEDGED_TYPES, binding):
with self.assertRaises(broker.Exit):
broker.check_acknowledged_case(doc) # no parameter admission
with mock.patch.dict(broker.PARAMETER_PINS, {'1.3': broker.PARAMETER_PINS['1.1']}):
broker.check_acknowledged_case(doc)
type_doc['spec']['status'] = 'proposed'
typ.write_text(yaml.safe_dump(type_doc))
proposed = {doc['spec']['type']: (relative, hashlib.sha256(typ.read_bytes()).hexdigest())}
with mock.patch.dict(broker.ACKNOWLEDGED_TYPES, proposed):
with self.assertRaises(broker.Exit) as caught:
broker.check_acknowledged_case(doc)
self.assertEqual(caught.exception.reason, 'case_source_validation_failed')
def test_wrong_sources_proposed_unknown_versions_and_symlinks_fail_closed(self):
spec = importlib.util.spec_from_file_location(
'candidate_generator', ROOT / 'scripts/keycape_client_secret_generate.py')
broker = importlib.util.module_from_spec(spec)
spec.loader.exec_module(broker)
source = self.tmp / 'net-kingdom'
source.mkdir()
relative, digest = broker.ACKNOWLEDGED_TYPES['oidc-browser-relying-party@1.1']
for rel in (relative, *broker.SOURCE_PINS):
dest = source / rel
dest.parent.mkdir(parents=True, exist_ok=True)
shutil.copyfile(broker.NETKINGDOM_ROOT / rel, dest)
doc = yaml.safe_load(CASE)
with mock.patch.object(broker, 'NETKINGDOM_ROOT', source):
broker.check_acknowledged_case(doc)
producer = source / next(iter(broker.SOURCE_PINS))
raw = producer.read_bytes()
producer.write_bytes(raw + b'\n# drift\n')
with self.assertRaises(broker.Exit) as caught:
broker.check_acknowledged_case(doc)
self.assertEqual(caught.exception.reason, 'case_source_validation_failed')
producer.write_bytes(raw)
typ = source / relative
raw = typ.read_bytes()
typ.write_bytes(raw.replace(b'status: acknowledged', b'status: proposed'))
# Even a hypothetical owner pin update cannot admit proposed status.
with mock.patch.dict(broker.ACKNOWLEDGED_TYPES, {
doc['spec']['type']: (relative, hashlib.sha256(typ.read_bytes()).hexdigest())}):
with self.assertRaises(broker.Exit) as caught:
broker.check_acknowledged_case(doc)
self.assertEqual(caught.exception.reason, 'case_source_validation_failed')
typ.write_bytes(raw + b'\n# source drift\n')
with self.assertRaises(broker.Exit) as caught:
broker.check_acknowledged_case(doc)
self.assertEqual(caught.exception.reason, 'case_source_validation_failed')
typ.unlink()
typ.symlink_to(broker.__file__)
with self.assertRaises(broker.Exit) as caught:
broker.check_acknowledged_case(doc)
self.assertEqual(caught.exception.reason, 'case_source_validation_failed')
for ref in ('oidc-browser-relying-party@1.3', 'oidc-browser-relying-party@99.0'):
doc['spec']['type'] = ref
with self.assertRaises(broker.Exit) as caught:
broker.check_acknowledged_case(doc)
self.assertEqual(caught.exception.reason, 'case_type_unacknowledged')
def test_exact_accepted_case_still_requires_materialization(self):
self.write_case(CASE.replace('status: materialized', 'status: accepted'))
code, receipt = self.run_cli()
self.assertEqual((code, receipt['outcome']), (2, 'case_not_materialized'))
self.assertEqual(self.state['calls'], [])
def test_agents_root_and_unbound_callers_refused(self):
code, receipt = self.run_cli(env_extra={'WARDEN_AGENT_ID': 'codex'})
self.assertEqual((code, receipt['outcome']), (2, 'agent_callers_refused'))
self.state['caller_policies'] = ['root']
code, receipt = self.run_cli()
self.assertEqual((code, receipt['outcome']), (2, 'root_refused'))
self.state['caller_policies'] = ['default']
code, receipt = self.run_cli()
self.assertEqual((code, receipt['outcome']), (2, 'caller_lacks_issuer_policy'))
self.assertFalse(any(c['args'][:2] == ['token', 'create'] for c in self.state['calls']))
def test_write_failure_is_reported_and_token_revoked(self):
self.state['fail_write'] = True
code, receipt = self.run_cli()
self.assertEqual((code, receipt['outcome']), (4, 'write_failed'))
self.assertTrue(receipt['grant_token_revoked'])
if __name__ == '__main__':
unittest.main()
Reference in a new issue View git blame Copy permalink