232 lines
11 KiB
Python
232 lines
11 KiB
Python
|
import base64
|
||
|
|
import hashlib
|
||
|
|
import json
|
||
|
|
import time
|
||
|
|
from dataclasses import replace
|
||
|
|
from urllib.parse import parse_qs, urlencode
|
||
|
|
|
||
|
|
import httpx
|
||
|
|
import jwt
|
||
|
|
import pytest
|
||
|
|
from cryptography.hazmat.primitives.asymmetric import rsa
|
||
|
|
from fastapi.testclient import TestClient
|
||
|
|
|
||
|
|
from hub_core.runtime.app import create_app
|
||
|
|
from hub_core.runtime.config import RuntimeSettings
|
||
|
|
from hub_core.security.browser import BrowserSettings, LOGIN_COOKIE, SESSION_COOKIE
|
||
|
|
from hub_core.security.identity import OIDCVerifier
|
||
|
|
from test_access_boundary import Owners
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture(scope='module')
|
||
|
|
def signing_key():
|
||
|
|
return rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def browser(tmp_path, signing_key):
|
||
|
|
secret = tmp_path / 'client-secret'
|
||
|
|
secret.write_text('test-client-secret')
|
||
|
|
owners = Owners()
|
||
|
|
state = {'exchanges': 0, 'id_changes': {}, 'access_changes': {}, 'id_type': 'JWT'}
|
||
|
|
jwk = json.loads(jwt.algorithms.RSAAlgorithm.to_jwk(signing_key.public_key()))
|
||
|
|
jwk.update(kid='browser-key', alg='RS256', use='sig')
|
||
|
|
|
||
|
|
def handle(request):
|
||
|
|
if request.url.path.endswith('openid-configuration'):
|
||
|
|
return httpx.Response(200, json={
|
||
|
|
'issuer': 'https://issuer.example', 'jwks_uri': 'https://issuer.example/keys',
|
||
|
|
'authorization_endpoint': 'https://issuer.example/authorize',
|
||
|
|
'token_endpoint': 'https://issuer.example/token',
|
||
|
|
'code_challenge_methods_supported': ['S256'], 'response_types_supported': ['code'],
|
||
|
|
'token_endpoint_auth_methods_supported': ['client_secret_basic']})
|
||
|
|
if request.url.path == '/keys':
|
||
|
|
return httpx.Response(200, json={'keys': [jwk]})
|
||
|
|
assert request.url.path == '/token'
|
||
|
|
state['exchanges'] += 1
|
||
|
|
form = parse_qs(request.content.decode())
|
||
|
|
assert form['code'] == ['one-time-code']
|
||
|
|
assert form['redirect_uri'] == ['https://hub.example/auth/callback']
|
||
|
|
assert form['grant_type'] == ['authorization_code']
|
||
|
|
assert request.headers['authorization'] == 'Basic ' + base64.b64encode(
|
||
|
|
b'hub-browser:test-client-secret').decode()
|
||
|
|
challenge = base64.urlsafe_b64encode(hashlib.sha256(form['code_verifier'][0].encode()).digest()).decode().rstrip('=')
|
||
|
|
assert challenge == state['login']['code_challenge'][0]
|
||
|
|
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(state['access_changes'])
|
||
|
|
access = jwt.encode(claims, signing_key, algorithm='RS256', headers={'kid': 'browser-key', 'typ': 'at+jwt'})
|
||
|
|
identity = dict(iss='https://issuer.example', sub='immutable-root', aud='hub-browser',
|
||
|
|
iat=now, exp=now+300, nonce=state['login']['nonce'][0], auth_time=now,
|
||
|
|
at_hash=base64.urlsafe_b64encode(hashlib.sha256(access.encode()).digest()[:16]).decode().rstrip('='))
|
||
|
|
identity.update(state['id_changes'])
|
||
|
|
identity = jwt.encode(identity, signing_key, algorithm='RS256', headers={'kid': 'browser-key', 'typ': state['id_type']})
|
||
|
|
state['access'] = access
|
||
|
|
return httpx.Response(200, json={'token_type': 'Bearer', 'access_token': access, 'id_token': identity})
|
||
|
|
|
||
|
|
upstream = httpx.AsyncClient(transport=httpx.MockTransport(handle))
|
||
|
|
controller = owners.controller()
|
||
|
|
controller.identity = OIDCVerifier(issuer='https://issuer.example', audience='hub-core', client=upstream)
|
||
|
|
app = create_app(settings=RuntimeSettings(environment='test', access_mode='enforce'),
|
||
|
|
access_controller=controller,
|
||
|
|
browser_settings=BrowserSettings('https://hub.example', 'hub-browser', secret))
|
||
|
|
with TestClient(app, base_url='https://hub.example', follow_redirects=False) as client:
|
||
|
|
yield client, app.state.browser_sessions, owners, state
|
||
|
|
import asyncio
|
||
|
|
asyncio.run(upstream.aclose())
|
||
|
|
|
||
|
|
|
||
|
|
def begin(client, state):
|
||
|
|
result = client.get('/auth/login')
|
||
|
|
assert result.status_code == 303
|
||
|
|
state['login'] = parse_qs(httpx.URL(result.headers['location']).query.decode())
|
||
|
|
assert state['login']['code_challenge_method'] == ['S256']
|
||
|
|
assert state['login']['redirect_uri'] == ['https://hub.example/auth/callback']
|
||
|
|
cookie = result.headers['set-cookie']
|
||
|
|
assert all(s in cookie for s in ['__Host-hub-login=', 'Secure', 'HttpOnly', 'SameSite=lax', 'Path=/'])
|
||
|
|
return '/auth/callback?' + urlencode({'state': state['login']['state'][0], 'code': 'one-time-code'})
|
||
|
|
|
||
|
|
|
||
|
|
def login(browser):
|
||
|
|
client, sessions, owners, state = browser
|
||
|
|
callback = begin(client, state)
|
||
|
|
result = client.get(callback)
|
||
|
|
assert result.status_code == 303, result.text
|
||
|
|
assert result.headers['location'] == '/docs'
|
||
|
|
assert owners.records[-1]['correlation_id'] == result.headers['x-correlation-id']
|
||
|
|
assert state['access'] not in str(result.headers)
|
||
|
|
assert client.cookies.get(SESSION_COOKIE) != state['access']
|
||
|
|
return callback
|
||
|
|
|
||
|
|
|
||
|
|
def test_login_cookie_session_revocation_and_logout(browser):
|
||
|
|
client, sessions, owners, state = browser
|
||
|
|
callback = login(browser)
|
||
|
|
assert client.get(callback).status_code == 401
|
||
|
|
assert state['exchanges'] == 1
|
||
|
|
assert client.get('/docs').status_code == 200
|
||
|
|
response = client.get('/auth/session')
|
||
|
|
assert response.status_code == 200
|
||
|
|
assert state['access'] not in response.text
|
||
|
|
assert response.headers['cache-control'] == 'no-store'
|
||
|
|
csrf = response.json()['csrf_token']
|
||
|
|
assert client.post('/auth/logout').status_code == 403
|
||
|
|
assert client.post('/auth/logout', headers={'origin': 'https://evil.example', 'x-hub-csrf': csrf}).status_code == 403
|
||
|
|
owners.facts = replace(owners.facts, root_entitled=False)
|
||
|
|
assert client.get('/docs').status_code == 403
|
||
|
|
assert client.get('/auth/session').status_code == 403
|
||
|
|
owners.audit_down = owners.policy_down = True
|
||
|
|
result = client.post('/auth/logout', headers={'origin': 'https://hub.example', 'x-hub-csrf': csrf})
|
||
|
|
assert result.status_code == 204
|
||
|
|
assert not sessions.sessions
|
||
|
|
assert client.cookies.get(SESSION_COOKIE) is None
|
||
|
|
|
||
|
|
|
||
|
|
def test_cookie_csrf_enforced_before_policy_and_mixed_credentials_refused(browser):
|
||
|
|
client, sessions, owners, state = browser
|
||
|
|
login(browser)
|
||
|
|
before = len(owners.requests)
|
||
|
|
assert client.post('/ports/messaging/messages', json={}).status_code == 403
|
||
|
|
assert len(owners.requests) == before
|
||
|
|
csrf = client.get('/auth/session').json()['csrf_token']
|
||
|
|
result = client.post('/ports/messaging/messages', json={}, headers={'origin': 'https://hub.example', 'x-hub-csrf': csrf})
|
||
|
|
# An invalid business request reaches content validation only with valid CSRF.
|
||
|
|
assert result.status_code == 422
|
||
|
|
body = {'schema_version': '0.1.0', 'correlation_id': 'f7cffcab-4c02-419e-89e5-0b463f5b433a',
|
||
|
|
'from_address': 'agent:root', 'to_addresses': ['agent:reader'], 'body': 'private browser text'}
|
||
|
|
headers = {'origin': 'https://hub.example', 'x-hub-csrf': csrf}
|
||
|
|
assert client.post('/ports/messaging/messages', json=body, headers=headers).status_code == 202
|
||
|
|
body['from_address'] = 'agent:someone-else'
|
||
|
|
assert client.post('/ports/messaging/messages', json=body, headers=headers).status_code == 403
|
||
|
|
assert 'private browser text' not in json.dumps(owners.records)
|
||
|
|
assert client.get('/docs', headers={'Authorization': 'Bearer ' + state['access']}).status_code == 400
|
||
|
|
assert client.get('/auth/session', headers={'Authorization': 'Bearer ' + state['access']}).status_code == 400
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize('changes', [
|
||
|
|
{'nonce': 'wrong'}, {'aud': 'wrong'}, {'azp': 'wrong'}, {'sub': 'different'},
|
||
|
|
{'aud': ['hub-browser', 'another']}, {'auth_time': 1}, {'iat': 1}, {'exp': 1},
|
||
|
|
{'auth_time': True}, {'at_hash': 'wrong'}, {'typ': 'Bearer'},
|
||
|
|
])
|
||
|
|
def test_invalid_id_token_never_creates_session(browser, changes):
|
||
|
|
client, sessions, owners, state = browser
|
||
|
|
state['id_changes'] = changes
|
||
|
|
assert client.get(begin(client, state)).status_code in {401, 403}
|
||
|
|
assert not sessions.sessions
|
||
|
|
assert client.cookies.get(SESSION_COOKIE) is None
|
||
|
|
|
||
|
|
|
||
|
|
def test_access_token_cannot_be_used_as_id_token(browser):
|
||
|
|
client, sessions, owners, state = browser
|
||
|
|
state['id_type'] = 'at+jwt'
|
||
|
|
assert client.get(begin(client, state)).status_code == 401
|
||
|
|
assert not sessions.sessions
|
||
|
|
|
||
|
|
|
||
|
|
def test_callback_bound_to_browser_and_one_time_state(browser):
|
||
|
|
client, sessions, owners, state = browser
|
||
|
|
callback = begin(client, state)
|
||
|
|
binding = client.cookies.get(LOGIN_COOKIE)
|
||
|
|
client.cookies.clear()
|
||
|
|
assert client.get(callback).status_code == 401
|
||
|
|
assert state['exchanges'] == 0
|
||
|
|
client.cookies.set(LOGIN_COOKIE, binding, domain='hub.example', path='/')
|
||
|
|
assert client.get(callback).status_code == 303
|
||
|
|
assert client.get(callback).status_code == 401
|
||
|
|
assert state['exchanges'] == 1
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.parametrize('changes', [{'sub': 'ordinary'}, {'tenant': 'tenant:other'},
|
||
|
|
{'assurance': dict(level='aal1', methods=['pwd'], mfa=False, source='key-cape', at=int(time.time()))}])
|
||
|
|
def test_login_requires_root_platform_and_mfa(browser, changes):
|
||
|
|
client, sessions, owners, state = browser
|
||
|
|
state['access_changes'] = changes
|
||
|
|
assert client.get(begin(client, state)).status_code == 403
|
||
|
|
assert not sessions.sessions
|
||
|
|
|
||
|
|
|
||
|
|
def test_session_expiry_and_restart_invalidate_cookies(browser):
|
||
|
|
client, sessions, owners, state = browser
|
||
|
|
login(browser)
|
||
|
|
sid = client.cookies.get(SESSION_COOKIE)
|
||
|
|
sessions.sessions[sid] = replace(sessions.sessions[sid], expires_at=1)
|
||
|
|
assert client.get('/docs').status_code == 401
|
||
|
|
login(browser)
|
||
|
|
sessions.clear()
|
||
|
|
assert client.get('/docs').status_code == 401
|
||
|
|
|
||
|
|
|
||
|
|
def test_logout_during_authority_check_prevents_handler(browser):
|
||
|
|
client, sessions, owners, state = browser
|
||
|
|
login(browser)
|
||
|
|
evaluate = owners.evaluate
|
||
|
|
async def revoke(request):
|
||
|
|
result = await evaluate(request)
|
||
|
|
sessions.sessions.clear()
|
||
|
|
return result
|
||
|
|
owners.evaluate = revoke
|
||
|
|
assert client.get('/docs').status_code == 401
|
||
|
|
|
||
|
|
|
||
|
|
def test_origin_redirect_capacity_and_owner_failures(browser):
|
||
|
|
client, sessions, owners, state = browser
|
||
|
|
assert client.get('/auth/login?next=https://evil.example').status_code == 400
|
||
|
|
assert client.get('/auth/login', headers={'host': 'evil.example'}).status_code == 403
|
||
|
|
callback = begin(client, state)
|
||
|
|
owners.audit_down = True
|
||
|
|
assert client.get(callback).status_code == 503
|
||
|
|
assert not sessions.sessions
|
||
|
|
sessions.settings = replace(sessions.settings, capacity=1)
|
||
|
|
begin(client, state)
|
||
|
|
assert client.get('/auth/login').status_code == 503
|
||
|
|
|
||
|
|
|
||
|
|
def test_browser_cannot_enable_legacy_runtime(tmp_path):
|
||
|
|
settings = BrowserSettings('https://hub.example', 'browser', tmp_path / 'secret')
|
||
|
|
with pytest.raises(ValueError, match='enforced access controller'):
|
||
|
|
create_app(browser_settings=settings)
|