178 lines
8 KiB
Python
178 lines
8 KiB
Python
|
"""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()
|