GitHub／ChatGPT 须先绑定已有 MSG 账号；相同邮箱不会自动合并账号。
' + if any(provider in providers for provider in ('github', 'chatgpt', 'agentid')): + body += ( + 'GitHub／ChatGPT／AgentID 须先绑定已有 MSG 账号；相同邮箱不会自动合并账号。
' + ) if email: body += email_form(token, mode='register' if register else 'login', handle=register) if passkey and not register: diff --git a/src/msg/transports/login_providers.py b/src/msg/transports/login_providers.py index 87146688..805b981e 100644 --- a/src/msg/transports/login_providers.py +++ b/src/msg/transports/login_providers.py @@ -23,17 +23,21 @@ GOOGLE_ISSUER = 'https://accounts.google.com' GITHUB_ISSUER = 'https://github.com' CHATGPT_ISSUER = 'https://auth.openai.com' +AGENTID_ISSUER = 'https://auth.agentid.com' _AUTHORIZATION = { + 'agentid': AGENTID_ISSUER + '/v0/authorize', 'google': 'https://accounts.google.com/o/oauth2/v2/auth', 'github': 'https://github.com/login/oauth/authorize', 'chatgpt': 'https://auth.openai.com/api/accounts/authorize', } _TOKEN = { + 'agentid': AGENTID_ISSUER + '/v0/token', 'google': 'https://oauth2.googleapis.com/token', 'github': 'https://github.com/login/oauth/access_token', 'chatgpt': 'https://auth.openai.com/api/accounts/oauth/token', } _JWKS = { + 'agentid': AGENTID_ISSUER + '/v0/jwks.json', 'google': 'https://www.googleapis.com/oauth2/v3/certs', 'chatgpt': 'https://auth.openai.com/.well-known/jwks.json', } @@ -66,7 +70,7 @@ class _JwksEntry: expires_at: float -# Only the two pinned JWKS URLs can enter these maps. No provider token is cached. +# Only the pinned JWKS URLs can enter these maps. No provider token is cached. _jwks_cache: dict[str, _JwksEntry] = {} _jwks_locks: dict[str, asyncio.Lock] = {} _discovery_expires_at = 0.0 @@ -107,8 +111,20 @@ def _settings(provider, config): client_id = _text(_get(config, 'client_id'), limit=1024, code='login_provider_configuration') method = _alias(config, 'token_auth_method', 'client_auth_method') if method is None: - method = 'none' if provider == 'chatgpt' else 'client_secret_post' - allowed = {'none', 'client_secret_basic'} if provider == 'chatgpt' else {'client_secret_post'} + method = ( + 'none' + if provider == 'chatgpt' + else 'client_secret_basic' + if provider == 'agentid' + else 'client_secret_post' + ) + allowed = ( + {'none', 'client_secret_basic'} + if provider == 'chatgpt' + else {'client_secret_basic', 'client_secret_post'} + if provider == 'agentid' + else {'client_secret_post'} + ) if ( type(method) is not str or method not in allowed @@ -143,7 +159,9 @@ def _callback_inputs(verifier, nonce, redirect_uri): return verifier, nonce, redirect_uri -def authorization_url(provider, config, state, verifier, nonce, redirect_uri) -> str: +def authorization_url( + provider, config, state, verifier, nonce, redirect_uri, *, login_hint=None +) -> str: """Build a pinned authorization URL; the HTTP boundary owns one-use state.""" client_id, _ = _settings(provider, config) state = _text(state, limit=1024) @@ -161,6 +179,10 @@ def authorization_url(provider, config, state, verifier, nonce, redirect_uri) -> parameters['allow_signup'] = 'false' else: parameters['nonce'] = nonce + if login_hint is not None: + if provider != 'agentid': + raise Failure('login_provider_invalid_request') + parameters['login_hint'] = _text(login_hint, limit=254) return _AUTHORIZATION[provider] + '?' + urlencode(parameters) @@ -259,6 +281,20 @@ def _cache_seconds(headers) -> int: return min(int(match[1][:10]), 3600) if match else 300 +def _check_discovery(document, provider, issuer, algorithm): + required = { + 'issuer': issuer, + 'authorization_endpoint': _AUTHORIZATION[provider], + 'token_endpoint': _TOKEN[provider], + 'jwks_uri': _JWKS[provider], + } + if any(document.get(key) != value for key, value in required.items()): + raise Failure('login_provider_invalid_discovery') + algorithms = document.get('id_token_signing_alg_values_supported') + if type(algorithms) is not list or algorithm not in algorithms: + raise Failure('login_provider_invalid_discovery') + + async def _openai_discovery(client, instant): global _discovery_expires_at, _discovery_lock if instant < _discovery_expires_at: @@ -269,17 +305,7 @@ async def _openai_discovery(client, instant): if instant < _discovery_expires_at: return document, headers = await _json_request(client, 'GET', _DISCOVERY) - required = { - 'issuer': CHATGPT_ISSUER, - 'authorization_endpoint': _AUTHORIZATION['chatgpt'], - 'token_endpoint': _TOKEN['chatgpt'], - 'jwks_uri': _JWKS['chatgpt'], - } - if any(document.get(key) != value for key, value in required.items()): - raise Failure('login_provider_invalid_discovery') - algorithms = document.get('id_token_signing_alg_values_supported') - if type(algorithms) is not list or 'RS256' not in algorithms: - raise Failure('login_provider_invalid_discovery') + _check_discovery(document, 'chatgpt', CHATGPT_ISSUER, 'RS256') _discovery_expires_at = instant + _cache_seconds(headers) @@ -326,6 +352,13 @@ async def exchange_identity( ) if provider == 'chatgpt': await _openai_discovery(client, instant) + elif provider == 'agentid': + document, _ = await _json_request( + client, 'GET', AGENTID_ISSUER + '/.well-known/openid-configuration' + ) + _check_discovery(document, provider, AGENTID_ISSUER, 'ES256') + if document.get('authorization_response_iss_parameter_supported') is not True: + raise Failure('login_provider_invalid_discovery') data = { 'grant_type': 'authorization_code', 'code': code, @@ -365,17 +398,21 @@ async def exchange_identity( raise Failure('login_provider_invalid_identity') return VerifiedIdentity(provider, GITHUB_ISSUER, str(subject)) id_token = tokens.get('id_token') - kid = token_key_id(id_token) + algorithm = 'ES256' if provider == 'agentid' else 'RS256' + kid = token_key_id(id_token, algorithm=algorithm) jwks = await _identity_jwks(provider, client, instant, kid) identity = verify_identity_token( id_token, jwks, issuers={GOOGLE_ISSUER, 'accounts.google.com'} if provider == 'google' + else {AGENTID_ISSUER} + if provider == 'agentid' else {CHATGPT_ISSUER}, audience=client_id, nonce=nonce, now=now, + algorithm=algorithm, ) issuer = GOOGLE_ISSUER if provider == 'google' else identity.issuer return VerifiedIdentity( diff --git a/tests/test_agentid_account_login.py b/tests/test_agentid_account_login.py new file mode 100644 index 00000000..cf0b4269 --- /dev/null +++ b/tests/test_agentid_account_login.py @@ -0,0 +1,210 @@ +"""AgentID browser flows through PostgreSQL and the existing account executor.""" + +from dataclasses import replace +from datetime import timedelta +from urllib.parse import parse_qs, urlsplit + +import pytest +from test_account_login import ( + account, # noqa: F401 + approve_if_pending, + post, + session_source, + token, +) +from test_oauth import browser_login, oauth # noqa: F401 + +from msg.login_config import LoginProvider +from msg.transports.login_providers import AGENTID_ISSUER, authorization_url +from msg.transports.oauth_http import csrf + + +@pytest.fixture +async def agentid_account(account, monkeypatch): # noqa: F811 + app, _, _, _, _ = account + credential = app.settings.config_dir / 'agentid.fixture' + credential.write_text('fixture-only-secret') + credential.chmod(0o600) + provider = LoginProvider( + 'agentid', + enabled=True, + client_id='agentid-fixture', + credential_file=str(credential), + token_auth_method='client_secret_basic', + ) + app.settings = replace( + app.settings, + login=replace( + app.settings.login, + providers=tuple( + provider if item.name == 'agentid' else item + for item in app.settings.login.providers + ), + ), + ) + + def authorize(name, config, state, verifier, nonce, redirect_uri, **options): + assert redirect_uri == app.settings.service_url + '/-/login/callback/' + name + return authorization_url( + name, + config, + state, + verifier, + nonce, + redirect_uri.replace('http://testserver/', 'http://localhost/'), + **options, + ) + + monkeypatch.setattr('msg.transports.account_login.authorization_url', authorize) + return account + + +async def begin(case, *, mode='login'): + _, _, _, http, _ = case + response = await post( + case, + '/-/login/start', + { + 'csrf': await token(http), + 'provider': 'agentid', + 'mode': mode, + }, + ) + assert response.status_code == 303, response.text + state = parse_qs(urlsplit(response.headers['location']).query)['state'][0] + return state, csrf(http.cookies.get('msg_account_login')) + + +async def callback(case, state, **overrides): + return await case[3].get( + '/-/login/callback/agentid', + params={ + 'state': state, + 'code': 'fixture-code', + 'iss': AGENTID_ISSUER, + **overrides, + }, + ) + + +async def test_public_initiate_url_preserves_hint_and_is_login_only(agentid_account): + app, _, _, http, _ = agentid_account + async with app.metadata.transaction(write=False) as tx: + before = tx.one('SELECT COUNT(*) FROM identities')[0] + response = await http.get( + '/login/agentid', + params={ + 'iss': AGENTID_ISSUER, + 'login_hint': 'agent+one@example.org', + }, + ) + assert response.status_code == 303 and 'set-cookie' not in response.headers + assert response.headers['location'].startswith('/-/login/start?') + response = await http.get(response.headers['location']) + assert response.status_code == 303, response.text + target = urlsplit(response.headers['location']) + assert target._replace(query='').geturl() == AGENTID_ISSUER + '/v0/authorize' + params = parse_qs(target.query) + assert params['login_hint'] == ['agent+one@example.org'] + assert params['client_id'] == ['agentid-fixture'] + assert params['code_challenge_method'] == ['S256'] + assert len(params['state'][0]) == 43 + assert 'Max-Age=600' in response.headers['set-cookie'] + assert 'HttpOnly' in response.headers['set-cookie'] + async with app.metadata.transaction(write=False) as tx: + assert tx.one('SELECT COUNT(*) FROM identities')[0] == before + assert 'Continue with AgentID' in (await http.get('/login')).text + assert 'Continue with AgentID' not in (await http.get('/register')).text + + +@pytest.mark.parametrize( + 'query', + [ + 'provider=agentid&iss=https%3A%2F%2Fattacker.example', + 'provider=agentid&mode=bind', + 'provider=agentid&mode=register', + 'provider=agentid&callbackURL=https%3A%2F%2Fattacker.example', + 'provider=agentid&provider=google', + 'provider=google', + 'provider=agentid&login_hint=a&login_hint=b', + ], +) +async def test_initiation_rejects_issuer_or_mode_injection(agentid_account, query): + response = await agentid_account[3].get('/-/login/start?' + query) + assert response.status_code == 400, response.text + assert 'location' not in response.headers + + +@pytest.mark.parametrize('issuer', [None, 'https://attacker.example']) +async def test_callback_requires_exact_issuer(agentid_account, issuer): + _, _, _, http, _ = agentid_account + state, _ = await begin(agentid_account) + params = {'state': state, 'code': 'fixture-code'} + if issuer is not None: + params['iss'] = issuer + denied = await http.get('/-/login/callback/agentid', params=params) + assert denied.status_code == 400 and 'login_provider_invalid_issuer' in denied.text + assert (await callback(agentid_account, state)).status_code == 303 + replay = await callback(agentid_account, state) + assert replay.status_code == 400 and 'invalid_login_transaction' in replay.text + + +async def test_waiting_callback_and_unbound_identity_do_not_create_account(agentid_account): + app, _, _, http, _ = agentid_account + state, value = await begin(agentid_account) + app._oauth_clock[0] += timedelta(seconds=330) + returned = await callback(agentid_account, state) + assert returned.status_code == 303, returned.text + async with app.metadata.transaction(write=False) as tx: + before = tx.one('SELECT COUNT(*) FROM identities')[0] + result = await post(agentid_account, '/-/login/complete', {'csrf': value, 'state': state}) + assert result.status_code == 400 and 'login_binding_required' in result.text + async with app.metadata.transaction(write=False) as tx: + assert tx.one('SELECT COUNT(*) FROM identities')[0] == before + replay = await post(agentid_account, '/-/login/complete', {'csrf': value, 'state': state}) + assert replay.status_code == 400 + + +async def test_agentid_cannot_register_or_bypass_csrf(agentid_account): + _, _, _, http, _ = agentid_account + form = {'provider': 'agentid', 'mode': 'register', 'handle': 'new-agent', 'custody': 'yes'} + denied = await post(agentid_account, '/-/login/start', form) + assert denied.status_code == 403 + denied = await post(agentid_account, '/-/login/start', {**form, 'csrf': await token(http)}) + assert denied.status_code == 400 and 'login_registration_forbidden' in denied.text + + +async def test_bound_agent_returns_to_same_self_held_identity(agentid_account): + app, key, subject, http, _ = agentid_account + await browser_login((app, key, subject, http)) + async with app.metadata.transaction(write=False) as tx: + before = tx.rows('SELECT id,body FROM credentials WHERE subject=?', (subject,)) + vaults = tx.one('SELECT COUNT(*) FROM custodial_vault WHERE subject=?', (subject,))[0] + state, value = await begin(agentid_account, mode='bind') + assert (await callback(agentid_account, state)).status_code == 303 + result = await post(agentid_account, '/-/login/complete', {'csrf': value, 'state': state}) + result = await approve_if_pending(agentid_account, result) + assert result.status_code == 303, result.text + assert (await session_source(app, http))['subject'] == subject + async with app.metadata.transaction(write=False) as tx: + assert (await tx.subject(subject)).kind == 'registered' + after = dict(tx.rows('SELECT id,body FROM credentials WHERE subject=?', (subject,))) + assert all(after[key_id] == body for key_id, body in before) + assert ( + tx.one('SELECT COUNT(*) FROM custodial_vault WHERE subject=?', (subject,))[0] == vaults + ) + http.cookies.clear() + state, value = await begin(agentid_account) + assert (await callback(agentid_account, state)).status_code == 303 + result = await post(agentid_account, '/-/login/complete', {'csrf': value, 'state': state}) + assert result.status_code == 303, result.text + assert (await session_source(app, http))['subject'] == subject + + +async def test_callback_cannot_move_to_another_browser(agentid_account): + _, _, _, http, _ = agentid_account + state, _ = await begin(agentid_account) + http.cookies.clear() + await token(http) + denied = await callback(agentid_account, state) + assert denied.status_code == 400 and 'invalid_login_transaction' in denied.text diff --git a/tests/test_agentid_providers.py b/tests/test_agentid_providers.py new file mode 100644 index 00000000..ddf71a6b --- /dev/null +++ b/tests/test_agentid_providers.py @@ -0,0 +1,343 @@ +"""AgentID verification with real ES256 signatures and isolated provider HTTP.""" + +import base64 +import hashlib +import json +from collections import Counter +from dataclasses import asdict +from datetime import UTC, datetime, timedelta +from urllib.parse import parse_qs, urlsplit + +import httpx +import jwt +import pytest +from cryptography.hazmat.primitives.asymmetric import ec, rsa + +from msg.core.codec import b64 +from msg.core.errors import Failure +from msg.login_config import LoginConfig, load_login +from msg.security.login_jwt import token_key_id +from msg.transports import login_providers as providers + +ISSUER = 'https://auth.agentid.com' +DISCOVERY = ISSUER + '/.well-known/openid-configuration' +AUTHORIZE = ISSUER + '/v0/authorize' +TOKEN = ISSUER + '/v0/token' +JWKS = ISSUER + '/v0/jwks.json' +CALLBACK = 'https://msg.example/-/login/callback/agentid' +NOW = datetime(2026, 10, 9, tzinfo=UTC) +NONCE = 'original-agentid-browser-nonce' +VERIFIER = 'v' * 43 +CODE = 'fixture-authorization-code' +CLIENT = 'fixture-agentid-client' + + +@pytest.fixture(autouse=True) +def clear_caches(): + providers._jwks_cache.clear() + providers._jwks_locks.clear() + yield + providers._jwks_cache.clear() + providers._jwks_locks.clear() + + +@pytest.fixture(scope='module') +def signing_keys(): + return [ec.generate_private_key(ec.SECP256R1()) for _ in range(2)] + + +def public_jwk(key, kid='current'): + return json.loads(jwt.algorithms.ECAlgorithm.to_jwk(key.public_key())) | { + 'kid': kid, + 'alg': 'ES256', + 'use': 'sig', + 'key_ops': ['verify'], + } + + +class AgentScenario: + def __init__(self, keys, credential): + self.keys = keys + self.key = keys[0] + self.kid = 'current' + self.config = {'client_id': CLIENT, 'credential_file': str(credential)} + self.claims = { + 'iss': ISSUER, + 'aud': CLIENT, + 'sub': 's' * 43, + 'iat': int(NOW.timestamp()) - 1, + 'exp': int(NOW.timestamp()) + 599, + 'nonce': NONCE, + 'actor_type': 'agent', + 'scope': 'openid email profile', + 'email': 'agent@example.org', + 'email_verified': True, + } + self.metadata = { + 'issuer': ISSUER, + 'authorization_endpoint': AUTHORIZE, + 'token_endpoint': TOKEN, + 'jwks_uri': JWKS, + 'id_token_signing_alg_values_supported': ['ES256'], + 'authorization_response_iss_parameter_supported': True, + } + self.jwks = {'keys': [public_jwk(self.key)]} + self.requests = [] + self.responses = {} + self.token_override = None + + def handle(self, request): + self.requests.append(request) + url = str(request.url) + if url in self.responses: + return self.responses[url] + if url == DISCOVERY: + return httpx.Response(200, json=self.metadata) + if url == TOKEN: + token = self.token_override or jwt.encode( + self.claims, self.key, algorithm='ES256', headers={'kid': self.kid} + ) + return httpx.Response( + 200, json={'id_token': token, 'access_token': 'fixture-not-retained'} + ) + if url == JWKS: + return httpx.Response(200, json=self.jwks, headers={'Cache-Control': 'max-age=3600'}) + pytest.fail('Unexpected provider endpoint: ' + url) + + async def exchange(self, now=NOW): + async with httpx.AsyncClient(transport=httpx.MockTransport(self.handle)) as client: + return await providers.exchange_identity( + 'agentid', self.config, CODE, VERIFIER, NONCE, CALLBACK, now, client=client + ) + + def count(self, url): + return Counter(str(request.url) for request in self.requests)[url] + + +@pytest.fixture +def flow(signing_keys, tmp_path): + credential = tmp_path / 'client.secret' + credential.write_text('fixture-only-secret\n') + credential.chmod(0o600) + return AgentScenario(signing_keys, credential) + + +def test_agentid_config_is_opt_in_and_uses_registered_basic(flow): + assert not LoginConfig().provider('agentid').enabled + config = load_login( + {'enabled': True, 'providers': {'agentid': flow.config}}, 'https://msg.example' + ) + provider = config.provider('agentid') + assert provider.enabled + assert provider.token_auth_method == 'client_secret_basic' + assert provider.credential_file == flow.config['credential_file'] + + +@pytest.mark.parametrize( + 'override', + [ + {'client_id': ''}, + {'client_id': '客户端'}, + {'credential_file': ''}, + {'credential_file': 'relative.secret'}, + {'token_auth_method': 'none'}, + {'token_auth_method': 'private_key_jwt'}, + {'client_secret': 'do-not-store-in-config'}, + ], +) +def test_agentid_config_rejects_missing_or_unsafe_credentials(flow, override): + with pytest.raises(Failure): + load_login( + {'enabled': True, 'providers': {'agentid': flow.config | override}}, + 'https://msg.example', + ) + + +def test_authorize_is_pinned_has_pkce_nonce_and_minimal_scopes(flow): + url = providers.authorization_url( + 'agentid', + flow.config, + 'original-state', + VERIFIER, + NONCE, + CALLBACK, + login_hint='agent+test@example.org', + ) + parsed = urlsplit(url) + params = parse_qs(parsed.query) + assert parsed._replace(query='').geturl() == AUTHORIZE + assert params == { + 'client_id': [CLIENT], + 'response_type': ['code'], + 'redirect_uri': [CALLBACK], + 'state': ['original-state'], + 'nonce': [NONCE], + 'code_challenge_method': ['S256'], + 'code_challenge': [b64(hashlib.sha256(VERIFIER.encode()).digest())], + 'scope': ['openid profile email'], + 'login_hint': ['agent+test@example.org'], + } + assert VERIFIER not in url and 'fixture-only-secret' not in url + + +@pytest.mark.parametrize('hint', ['', 'x\r\nInjected:yes', 'a' * 255, ['agent@example.org']]) +def test_rejects_unsafe_login_hint(flow, hint): + with pytest.raises(Failure, match='^login_provider_invalid_request$'): + providers.authorization_url( + 'agentid', flow.config, 'state', VERIFIER, NONCE, CALLBACK, login_hint=hint + ) + + +@pytest.mark.parametrize('method', ['client_secret_basic', 'client_secret_post']) +async def test_real_es256_identity_and_registered_authentication(flow, method): + flow.config['token_auth_method'] = method + identity = await flow.exchange() + assert (identity.provider, identity.issuer, identity.subject) == ('agentid', ISSUER, 's' * 43) + assert set(asdict(identity)) == {'provider', 'issuer', 'subject', 'client_id'} + assert identity.client_id is None + request = next(item for item in flow.requests if str(item.url) == TOKEN) + data = parse_qs(request.content.decode()) + assert data['code'] == [CODE] and data['code_verifier'] == [VERIFIER] + assert data['redirect_uri'] == [CALLBACK] + assert data['grant_type'] == ['authorization_code'] + if method == 'client_secret_basic': + header = request.headers['authorization'].removeprefix('Basic ') + assert base64.b64decode(header).decode() == CLIENT + ':fixture-only-secret' + assert 'client_secret' not in data + else: + assert 'authorization' not in request.headers + assert data['client_secret'] == ['fixture-only-secret'] + assert {str(item.url) for item in flow.requests} == {DISCOVERY, TOKEN, JWKS} + + +async def test_profile_changes_do_not_change_identity_or_merge_inboxes(flow): + original = await flow.exchange() + flow.claims.update( + email='renamed@example.org', name='Renamed agent', owner_email='owner@example.org' + ) + assert await flow.exchange() == original + flow.claims['sub'] = 'other-inbox' + assert (await flow.exchange()).stable_id != original.stable_id + assert flow.count(JWKS) == 1 + + +@pytest.mark.parametrize( + ('claim', 'value'), + [ + ('iss', 'https://attacker.example'), + ('aud', 'other-client'), + ('nonce', 'another-browser'), + ('exp', int(NOW.timestamp())), + ('exp', True), + ('iat', int(NOW.timestamp()) + 6), + ('iat', True), + ('sub', ''), + ('sub', 42), + ('aud', [CLIENT, 'other-client']), + ('azp', 'other-client'), + ('nbf', int(NOW.timestamp()) + 1), + ], +) +async def test_rejects_signed_wrong_claims(flow, claim, value): + flow.claims[claim] = value + with pytest.raises(Failure, match='^login_provider_invalid_token$') as error: + await flow.exchange() + assert error.value.details is None + assert CODE not in str(error.value) + + +@pytest.mark.parametrize('claim', ['iss', 'aud', 'sub', 'exp', 'iat', 'nonce']) +async def test_rejects_missing_claims(flow, claim): + del flow.claims[claim] + with pytest.raises(Failure, match='^login_provider_invalid_token$'): + await flow.exchange() + + +@pytest.mark.parametrize('kind', ['forged', 'none', 'HS256', 'RS256']) +async def test_rejects_forged_or_wrong_algorithm(flow, kind): + if kind == 'forged': + flow.key = flow.keys[1] + else: + key = ( + rsa.generate_private_key(public_exponent=65537, key_size=2048) + if kind == 'RS256' + else 'x' * 32 + if kind == 'HS256' + else '' + ) + flow.token_override = jwt.encode( + flow.claims, key, algorithm=kind, headers={'kid': 'current'} + ) + with pytest.raises(Failure, match='^login_provider_invalid_token$'): + await flow.exchange() + if kind != 'forged': + assert flow.count(JWKS) == 0 + + +@pytest.mark.parametrize( + ('field', 'value'), + [ + ('kty', 'RSA'), + ('crv', 'P-384'), + ('alg', 'RS256'), + ('use', 'enc'), + ('key_ops', ['sign']), + ('d', 'a' * 43), + ('x', 'a' * 42), + ('y', '!' * 43), + ], +) +async def test_rejects_invalid_or_private_jwks(flow, field, value): + flow.jwks['keys'][0][field] = value + with pytest.raises(Failure, match='^login_provider_invalid_token$'): + await flow.exchange() + + +async def test_rejects_duplicate_key_ids(flow): + flow.jwks['keys'].append(dict(flow.jwks['keys'][0])) + with pytest.raises(Failure, match='^login_provider_invalid_token$'): + await flow.exchange() + + +async def test_rotation_refresh_is_bounded(flow): + await flow.exchange() + flow.key, flow.kid = flow.keys[1], 'next' + flow.jwks = {'keys': [public_jwk(flow.key, 'next')]} + with pytest.raises(Failure, match='^login_provider_invalid_token$'): + await flow.exchange(NOW + timedelta(seconds=29)) + assert flow.count(JWKS) == 1 + assert (await flow.exchange(NOW + timedelta(seconds=30))).subject == 's' * 43 + assert flow.count(JWKS) == 2 + + +@pytest.mark.parametrize( + ('field', 'value'), + [ + ('issuer', 'https://attacker.example'), + ('token_endpoint', 'https://attacker.example/token'), + ('authorization_endpoint', 'https://attacker.example/auth'), + ('jwks_uri', 'https://attacker.example/keys'), + ('id_token_signing_alg_values_supported', ['RS256']), + ('authorization_response_iss_parameter_supported', False), + ], +) +async def test_discovery_mismatch_fails_before_sending_credentials(flow, field, value): + flow.metadata[field] = value + with pytest.raises(Failure, match='^login_provider_invalid_discovery$'): + await flow.exchange() + assert flow.count(TOKEN) == 0 + + +@pytest.mark.parametrize('endpoint', [DISCOVERY, TOKEN, JWKS]) +async def test_provider_redirects_are_not_followed(flow, endpoint): + flow.responses[endpoint] = httpx.Response(302, headers={'Location': 'https://attacker.example'}) + with pytest.raises(Failure, match='^login_provider_invalid_response$'): + await flow.exchange() + assert all(str(request.url) in {DISCOVERY, TOKEN, JWKS} for request in flow.requests) + + +def test_es256_does_not_relax_existing_provider_algorithm(flow): + token = jwt.encode(flow.claims, flow.key, algorithm='ES256', headers={'kid': 'current'}) + assert token_key_id(token, algorithm='ES256') == 'current' + with pytest.raises(Failure, match='^login_provider_invalid_token$'): + token_key_id(token)