railiance-platform/tests/test_keycape_client_secret_generate.py

"""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 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)
''')

CASE = textwrap.dedent('''\
    apiVersion: netkingdom.io/security-case/v0.1
    kind: SecurityCase
    spec:
      id: forgejo-netkingdom-oidc
      type: oidc-browser-relying-party@1.0
      status: materialized
      parameters:
        client_id: forgejo
        client_type: confidential
''')


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'
        env = {k: v for k, v in os.environ.items() if k not in ('BAO_TOKEN', 'VAULT_TOKEN', 'WARDEN_AGENT_ID')}
        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_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()