coulomb/key-cape
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Bind JWT negative probes to exact claim and cleanup evidence

Browse source 
Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade
This commit is contained in:
tegwick 2026-10-11 02:21:03 +02:00
parent f97f006ede
commit aef56bcba1
4 changed files with 322 additions and 22 deletions
Download patch file Download diff file Expand all files Collapse all files

98
scripts/test_openbao_negative_probe.py
View file

@ -14,7 +14,15 @@ SECRETS = {cid: SECRET for _, cid, _ in probe.CLAIM_PROBES + [probe.EXPIRY_PROBE
def fake_jwt(client_id, exp=1000):
payload = base64.urlsafe_b64encode(json.dumps({'exp': exp, 'cid': client_id}).encode()).decode().rstrip('=')
claims = dict({'aud': 'secrets-engine-openbao', 'sub': 'service:secrets-engine', 'tenant': 'tenant:platform', 'roles': ['secrets-engine'], 'scope': 'openbao:login', 'principal_type': 'service'}, exp=exp, iat=exp-60, cid=client_id, iss='https://kc.coulomb.social')
field = client_id.removeprefix('openbao-probe-')
if field in {'aud', 'sub', 'tenant', 'roles', 'scope'}:
claims[field] = ['probe'] if field == 'roles' else 'probe'
if client_id == 'human':
claims['principal_type'] = 'human'
if client_id == 'codex-railiance-platform':
claims['roles'] = ['coding-agent']
payload = base64.urlsafe_b64encode(json.dumps(claims).encode()).decode().rstrip('=')
return f'h.{payload}.s'
@ -25,6 +33,7 @@ class FakeBao:
self.accept = set(accept)
self.login_status = login_status
self.revoked = 0
self.expiry_presentations = 0
def __call__(self, url, form=None, body=None, headers=None, timeout=15):
if url == TOKEN_URL:
@ -36,7 +45,9 @@ class FakeBao:
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:
if cid == 'openbao-probe-expiry':
self.expiry_presentations += 1
if cid in self.accept or (cid == 'openbao-probe-expiry' and self.expiry_presentations == 1):
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')
@ -65,7 +76,7 @@ class ProbeTest(unittest.TestCase):
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)
self.assertEqual(fake.revoked, 2)
def test_transport_failure_is_never_a_refusal(self):
receipt = run(FakeBao(login_status=503))
@ -89,6 +100,87 @@ class ProbeTest(unittest.TestCase):
leeway=150, sleep=waits.append, now=lambda: 900)
self.assertEqual(waits, [1000 + 150 + 10 - 900])
def test_wrong_expiry_reason_is_not_evidence(self):
fake = FakeBao()
def wrong(url, **kw):
result = fake(url, **kw)
if url.endswith('/login') and fake.expiry_presentations > 1:
return 400, json.dumps({'errors': ['claim sub does not match']}).encode()
return result
receipt = run(wrong)
self.assertFalse(receipt['all_refused'])
expired = receipt['cases'][-1]
self.assertTrue(expired['refused'])
self.assertFalse(expired['verified'])
def test_generic_denial_is_not_evidence(self):
fake = FakeBao()
def deny(url, **kw):
result = fake(url, **kw)
if result[0] == 400:
return 403, b'{"errors":["permission denied"]}'
return result
self.assertFalse(run(deny)['all_refused'])
def test_expiry_requires_valid_control(self):
fake = FakeBao()
def reject_control(url, **kw):
result = fake(url, **kw)
if url.endswith('/login') and fake.expiry_presentations == 1:
return 400, b'{"errors":["claim sub mismatch"]}'
return result
receipt = run(reject_control)
self.assertFalse(receipt['all_refused'])
self.assertFalse(receipt['expiry_positive_control_verified'])
def test_unconfirmed_control_revoke_fails(self):
fake = FakeBao()
def bad_revoke(url, **kw):
if url.endswith('/revoke-self'):
return 500, b''
return fake(url, **kw)
self.assertFalse(run(bad_revoke)['all_refused'])
def test_each_probe_has_exactly_one_claim_defect(self):
for case, cid, _ in probe.CLAIM_PROBES:
self.assertEqual(probe.check_claims(fake_jwt(cid), case, lambda: 0), 1000)
with self.assertRaises(probe.Inconclusive):
probe.check_claims(fake_jwt(cid), 'expired', lambda: 0)
self.assertEqual(probe.check_claims(fake_jwt('openbao-probe-expiry'), 'expired', lambda: 0), 1000)
def test_audience_alias_breaks_probe_isolation(self):
token = fake_jwt('openbao-probe-sub')
claims = json.loads(base64.urlsafe_b64decode(token.split('.')[1] + '=='))
claims['aud'] = 'openbao-probe-sub'
payload = base64.urlsafe_b64encode(json.dumps(claims).encode()).decode().rstrip('=')
with self.assertRaises(probe.Inconclusive):
probe.check_claims(f'h.{payload}.s', 'sub', lambda: 0)
def test_redirects_are_not_followed(self):
self.assertIsNone(probe.NoRedirect().redirect_request(None, None, 302, '', {}, 'https://other.example'))
def test_not_yet_valid_does_not_count_as_expired(self):
self.assertNotEqual(probe.classify('token not valid yet: nbf'), 'expired')
def test_malformed_refusal_is_inconclusive(self):
fake = FakeBao()
def malformed(url, **kw):
result = fake(url, **kw)
if result[0] == 400:
return 400, b'{"errors":null}'
return result
receipt = run(malformed)
self.assertFalse(receipt['all_refused'])
self.assertTrue(all(c['class'] == 'inconclusive' for c in receipt['cases'] if c['case'] != 'expired'))
def test_expiry_client_ttl_is_bounded(self):
token = fake_jwt('openbao-probe-expiry')
claims = json.loads(base64.urlsafe_b64decode(token.split('.')[1] + '=='))
claims['iat'] = claims['exp'] - 900
payload = base64.urlsafe_b64encode(json.dumps(claims).encode()).decode().rstrip('=')
with self.assertRaises(probe.Inconclusive):
probe.check_claims(f'h.{payload}.s', 'expired', lambda: 0)
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')