tegwick
3e386147fd
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a0e747-8f27-7242-8df8-8bc44f88c929
102 lines
4.3 KiB
Python
102 lines
4.3 KiB
Python
import asyncio
|
|
import json
|
|
import time
|
|
|
|
import httpx
|
|
import jwt
|
|
import pytest
|
|
from cryptography.hazmat.primitives.asymmetric import rsa
|
|
|
|
from hub_core.security.identity import AccessFailure, OIDCVerifier
|
|
|
|
|
|
@pytest.fixture(scope='module')
|
|
def signing_key():
|
|
return rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
|
|
|
|
|
def setup(signing_key, changes=None, header_changes=None):
|
|
now = int(time.time())
|
|
claims = dict(iss='https://issuer.example', sub='immutable-root', aud='hub-core',
|
|
iat=now, exp=now+300, nbf=now, tenant='tenant:platform',
|
|
principal_type='human', groups=[], roles=[], scope='openid',
|
|
assurance=dict(level='aal2', methods=['pwd', 'otp'], mfa=True,
|
|
source='key-cape', at=now))
|
|
claims.update(changes or {})
|
|
headers = {'kid': 'key-1', 'typ': 'at+jwt', **(header_changes or {})}
|
|
token = jwt.encode(claims, signing_key, algorithm='RS256', headers=headers)
|
|
jwk = json.loads(jwt.algorithms.RSAAlgorithm.to_jwk(signing_key.public_key()))
|
|
jwk.update(kid='key-1', use='sig', alg='RS256')
|
|
responses = {'discovery': 200, 'keys': [jwk]}
|
|
|
|
def handle(request):
|
|
if request.url.path.endswith('openid-configuration'):
|
|
return httpx.Response(responses['discovery'], json={
|
|
'issuer': 'https://issuer.example', 'jwks_uri': 'https://issuer.example/keys',
|
|
})
|
|
return httpx.Response(200, json={'keys': responses['keys']})
|
|
|
|
client = httpx.AsyncClient(transport=httpx.MockTransport(handle))
|
|
verifier = OIDCVerifier(issuer='https://issuer.example', audience='hub-core', client=client)
|
|
return token, verifier, responses, client
|
|
|
|
|
|
def test_accepts_valid_iam_access_token(signing_key):
|
|
token, verifier, _, client = setup(signing_key)
|
|
async def run():
|
|
async with client:
|
|
actor = await verifier.authenticate(token)
|
|
assert actor.subject == 'immutable-root'
|
|
assert actor.tenant == 'tenant:platform'
|
|
asyncio.run(run())
|
|
|
|
|
|
@pytest.mark.parametrize('claims,headers', [
|
|
({'iss': 'https://evil.example'}, {}), ({'aud': 'different'}, {}),
|
|
({'exp': 1}, {}), ({'nbf': int(time.time())+3600}, {}),
|
|
({'iat': int(time.time())+3600}, {}), ({'sub': ''}, {}),
|
|
({'roles': 'platform-root'}, {}), ({'groups': {}}, {}),
|
|
({'tenant': None}, {}), ({'scope': None}, {}),
|
|
({'assurance': {'level': 'aal2'}}, {}), ({'principal_type': 'root'}, {}),
|
|
({'exp': int(time.time())+3600}, {}), ({}, {'typ': 'JWT'}),
|
|
({}, {'kid': 'unknown'}),
|
|
({'principal_type': 'agent', 'agent': {'id': 'a', 'mode': 'delegated'}}, {}),
|
|
])
|
|
def test_invalid_tokens_are_401(signing_key, claims, headers):
|
|
token, verifier, _, client = setup(signing_key, claims, headers)
|
|
async def run():
|
|
async with client:
|
|
with pytest.raises(AccessFailure) as result:
|
|
await verifier.authenticate(token)
|
|
assert result.value.status == 401
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_key_rotation_removes_old_trust_and_outage_fails_closed(signing_key):
|
|
token, verifier, responses, client = setup(signing_key)
|
|
second = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
|
next_jwk = json.loads(jwt.algorithms.RSAAlgorithm.to_jwk(second.public_key()))
|
|
next_jwk.update(kid='key-2', alg='RS256', use='sig')
|
|
claims = jwt.decode(token, options={'verify_signature': False})
|
|
rotated = jwt.encode(claims, second, algorithm='RS256', headers={'kid': 'key-2', 'typ': 'at+jwt'})
|
|
async def run():
|
|
async with client:
|
|
await verifier.authenticate(token)
|
|
responses['keys'] = [next_jwk]
|
|
verifier._loaded -= 2
|
|
await verifier.authenticate(rotated)
|
|
with pytest.raises(AccessFailure) as result:
|
|
await verifier.authenticate(token)
|
|
assert result.value.status == 401
|
|
responses['discovery'] = 503
|
|
verifier._loaded = 0
|
|
with pytest.raises(AccessFailure) as result:
|
|
await verifier.authenticate(rotated)
|
|
assert result.value.status == 503
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_untrusted_issuer_configuration_rejected():
|
|
for issuer in ['http://issuer.example', 'https://localhost', 'https://u:p@example.com']:
|
|
with pytest.raises(ValueError):
|
|
OIDCVerifier(issuer=issuer, audience='hub-core', client=None)
|