tegwick
bae9f85160
T01 done: railiance-platform agreed the case list. The packet adds an otherwise-valid expiry client and pins audience on the claim probes so each token fails only the claim under test. The script prints one refused boolean and refusal class per case, revokes any accepted token, and never counts a transport failure as a refusal. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 779409@bnt-lap001 Assistant-Session: c658ca06-19b5-468d-be64-2985dc20bfd3
99 lines
4.1 KiB
Python
99 lines
4.1 KiB
Python
import base64
|
|
import json
|
|
import os
|
|
import sys
|
|
import unittest
|
|
|
|
sys.path.insert(0, os.path.dirname(__file__))
|
|
import openbao_negative_probe as probe # noqa: E402
|
|
|
|
TOKEN_URL = 'https://kc.example/token'
|
|
BAO = 'https://bao.example'
|
|
SECRET = 'not-a-real-secret'
|
|
SECRETS = {cid: SECRET for _, cid, _ in probe.CLAIM_PROBES + [probe.EXPIRY_PROBE, probe.CODING_AGENT]}
|
|
|
|
|
|
def fake_jwt(client_id, exp=1000):
|
|
payload = base64.urlsafe_b64encode(json.dumps({'exp': exp, 'cid': client_id}).encode()).decode().rstrip('=')
|
|
return f'h.{payload}.s'
|
|
|
|
|
|
class FakeBao:
|
|
"""KeyCape and OpenBao in one: refuses every token unless told to accept it."""
|
|
|
|
def __init__(self, accept=(), login_status=None):
|
|
self.accept = set(accept)
|
|
self.login_status = login_status
|
|
self.revoked = 0
|
|
|
|
def __call__(self, url, form=None, body=None, headers=None, timeout=15):
|
|
if url == TOKEN_URL:
|
|
cid = base64.b64decode(headers['Authorization'].split()[1]).decode().split(':')[0]
|
|
return 200, json.dumps({'access_token': fake_jwt(cid)}).encode()
|
|
if url.endswith('/revoke-self'):
|
|
self.revoked += 1
|
|
return 204, b''
|
|
cid = json.loads(base64.urlsafe_b64decode(body['jwt'].split('.')[1] + '==')).get('cid', 'human')
|
|
if self.login_status:
|
|
return self.login_status, b''
|
|
if cid in self.accept:
|
|
return 200, json.dumps({'auth': {'client_token': 'bao-token'}}).encode()
|
|
reason = {'openbao-probe-expiry': 'token is expired',
|
|
'human': 'claim "principal_type" does not match'}.get(cid, 'claim "sub" does not match any bound value')
|
|
return 400, json.dumps({'errors': [reason]}).encode()
|
|
|
|
|
|
def run(fake, **kw):
|
|
kw.setdefault('human_token', fake_jwt('human'))
|
|
return probe.run(fake, SECRETS, TOKEN_URL, BAO, 'keycape-services', 'secrets-engine-openbao',
|
|
sleep=lambda s: None, now=lambda: 0, **kw)
|
|
|
|
|
|
class ProbeTest(unittest.TestCase):
|
|
def test_all_refused_with_classes(self):
|
|
receipt = run(FakeBao())
|
|
self.assertTrue(receipt['all_refused'])
|
|
classes = {c['case']: c['class'] for c in receipt['cases']}
|
|
self.assertEqual(classes['expired'], 'expired')
|
|
self.assertEqual(classes['principal_type'], 'not_service_principal')
|
|
self.assertEqual(classes['sub'], 'claim_mismatch')
|
|
self.assertEqual(len(receipt['cases']), 8)
|
|
|
|
def test_accepted_token_fails_and_is_revoked(self):
|
|
fake = FakeBao(accept={'openbao-probe-tenant'})
|
|
receipt = run(fake)
|
|
self.assertFalse(receipt['all_refused'])
|
|
tenant = next(c for c in receipt['cases'] if c['case'] == 'tenant')
|
|
self.assertEqual((tenant['refused'], tenant['class'], tenant['revoked']), (False, 'accepted', True))
|
|
self.assertEqual(fake.revoked, 1)
|
|
|
|
def test_transport_failure_is_never_a_refusal(self):
|
|
receipt = run(FakeBao(login_status=503))
|
|
self.assertFalse(receipt['all_refused'])
|
|
self.assertTrue(all(c['class'] == 'inconclusive' for c in receipt['cases']))
|
|
|
|
def test_missing_inputs_are_inconclusive(self):
|
|
receipt = run(FakeBao(), human_token=None)
|
|
pt = next(c for c in receipt['cases'] if c['case'] == 'principal_type')
|
|
self.assertEqual(pt['class'], 'inconclusive')
|
|
self.assertFalse(receipt['all_refused'])
|
|
|
|
def test_receipt_carries_no_values(self):
|
|
text = json.dumps(run(FakeBao(accept={'openbao-probe-aud'})))
|
|
for value in (SECRET, 'bao-token', 'does not match', 'h.'):
|
|
self.assertNotIn(value, text)
|
|
|
|
def test_expiry_waits_past_leeway(self):
|
|
waits = []
|
|
probe.run(FakeBao(), SECRETS, TOKEN_URL, BAO, 'm', 'r', human_token=fake_jwt('human'),
|
|
leeway=150, sleep=waits.append, now=lambda: 900)
|
|
self.assertEqual(waits, [1000 + 150 + 10 - 900])
|
|
|
|
def test_classify(self):
|
|
self.assertEqual(probe.classify('error validating token: token is expired'), 'expired')
|
|
self.assertEqual(probe.classify('aud claim does not match any bound audience'), 'claim_mismatch')
|
|
self.assertEqual(probe.classify('permission denied'), 'refused_other')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|