"""The OIDC adapter against a stand-in provider: real signed tokens, canned HTTP answers.""" import json import time from urllib.parse import parse_qs, urlsplit import pytest import requests from joserfc import jwt from joserfc.jwk import KeySet, OctKey, RSAKey from portal.adapters.idp import IdpUnavailable, LoginRejected, OidcIdp from portal.config.env import OidcConfig ISSUER = "https://idp.example.test/application/o/portal/" CLIENT_ID = "portal-client" SECRET = "client-secret-value-0123456789abcdef" BASE = "https://idp.example.test/application/o" DISCOVERY_URL = ISSUER + ".well-known/openid-configuration" JWKS_URL = f"{BASE}/jwks/" TOKEN_URL = f"{BASE}/token/" CONFIG = OidcConfig(ISSUER, CLIENT_ID, SECRET, "https://portal.example.test") KEY = RSAKey.generate_key(2048, parameters={"kid": "k1", "use": "sig"}) OTHER_KEY = RSAKey.generate_key(2048, parameters={"kid": "k1", "use": "sig"}) ROTATED_KEY = RSAKey.generate_key(2048, parameters={"kid": "k2", "use": "sig"}) class Provider(requests.adapters.BaseAdapter): """Answers requests from a table of address -> (status, json) and records them.""" def __init__(self): super().__init__() self.calls = [] self.error = None self.routes = { DISCOVERY_URL: ( 200, { "issuer": ISSUER, "authorization_endpoint": f"{BASE}/authorize/", "token_endpoint": TOKEN_URL, "jwks_uri": JWKS_URL, }, ), JWKS_URL: (200, KeySet([KEY]).as_dict(private=False)), TOKEN_URL: (200, {"access_token": "at", "token_type": "Bearer", "id_token": ""}), } def send(self, request, **kwargs): self.calls.append(request) if self.error: raise self.error status, body = self.routes[request.url.split("?")[0]] response = requests.Response() response.status_code = status response._content = json.dumps(body).encode() response.headers["Content-Type"] = "application/json" response.url = request.url response.request = request response.encoding = "utf-8" return response def close(self): pass def count(self, url): return sum(1 for r in self.calls if r.url.split("?")[0] == url) @pytest.fixture def provider(): return Provider() @pytest.fixture def idp(provider): instance = OidcIdp(CONFIG) def factory(): session = instance._default_session() session.mount("https://", provider) return session instance._factory = factory return instance def mint(flow, *, key=KEY, alg="RS256", header=None, **overrides): claims = { "iss": ISSUER, "aud": CLIENT_ID, "sub": "user-1", "exp": int(time.time()) + 300, "iat": int(time.time()), "nonce": flow.nonce, "preferred_username": "alice", } claims.update(overrides) claims = {k: v for k, v in claims.items() if v is not None} return jwt.encode({"alg": alg, **(header or {"kid": "k1"})}, claims, key) def sign_in(idp, provider, flow, token, **params): provider.routes[TOKEN_URL] = ( 200, {"access_token": "at", "token_type": "Bearer", "id_token": token}, ) return idp.complete({"state": flow.state, "code": "the-code", **params}, flow) def test_begin_builds_a_pkce_authorization_request(idp): url, flow = idp.begin() parts = urlsplit(url) query = {k: v[0] for k, v in parse_qs(parts.query).items()} assert f"{parts.scheme}://{parts.netloc}{parts.path}" == f"{BASE}/authorize/" assert query["response_type"] == "code" assert query["client_id"] == CLIENT_ID assert query["redirect_uri"] == "https://portal.example.test/auth/callback" assert query["scope"] == "openid profile email" assert query["code_challenge_method"] == "S256" assert query["state"] == flow.state and query["nonce"] == flow.nonce assert "code_challenge" in query assert flow.code_verifier not in url # only its hash is sent def test_each_flow_has_fresh_random_values(idp): _, first = idp.begin() _, second = idp.begin() assert len({first.state, second.state, first.nonce, second.nonce}) == 4 def test_discovery_is_fetched_once(idp, provider): idp.begin() idp.begin() assert provider.count(DISCOVERY_URL) == 1 def test_a_valid_sign_in_returns_the_identity(idp, provider): _, flow = idp.begin() identity = sign_in(idp, provider, flow, mint(flow)) assert (identity.username, identity.subject) == ("alice", "user-1") def test_the_code_is_exchanged_with_the_verifier_and_client_authentication(idp, provider): _, flow = idp.begin() sign_in(idp, provider, flow, mint(flow)) request = [r for r in provider.calls if r.url == TOKEN_URL][0] form = parse_qs(request.body) assert form["grant_type"] == ["authorization_code"] assert form["code"] == ["the-code"] assert form["code_verifier"] == [flow.code_verifier] assert form["redirect_uri"] == ["https://portal.example.test/auth/callback"] assert request.headers["Authorization"].startswith("Basic ") def test_an_audience_list_that_contains_the_client_is_accepted(idp, provider): _, flow = idp.begin() assert sign_in(idp, provider, flow, mint(flow, aud=[CLIENT_ID, "other"])).username == "alice" @pytest.mark.parametrize( "params", [{"state": "wrong"}, {"state": ""}, {"code": ""}, {"error": "access_denied"}] ) def test_a_bad_callback_is_rejected_before_any_token_request(idp, provider, params): _, flow = idp.begin() with pytest.raises(LoginRejected): idp.complete({"state": flow.state, "code": "c", **params}, flow) assert provider.count(TOKEN_URL) == 0 def test_a_refused_code_is_rejected(idp, provider): _, flow = idp.begin() provider.routes[TOKEN_URL] = (400, {"error": "invalid_grant"}) with pytest.raises(LoginRejected): idp.complete({"state": flow.state, "code": "c"}, flow) def test_an_unreachable_token_endpoint_is_an_outage(idp, provider): _, flow = idp.begin() provider.error = requests.ConnectionError("down") with pytest.raises(IdpUnavailable): idp.complete({"state": flow.state, "code": "c"}, flow) def test_a_missing_id_token_is_rejected(idp, provider): _, flow = idp.begin() provider.routes[TOKEN_URL] = (200, {"access_token": "at", "token_type": "Bearer"}) with pytest.raises(LoginRejected): idp.complete({"state": flow.state, "code": "c"}, flow) @pytest.mark.parametrize( "overrides", [ {"nonce": "somebody-elses"}, {"nonce": None}, {"iss": "https://evil.example.test/"}, {"aud": "another-client"}, {"exp": int(time.time()) - 3600}, {"exp": None}, {"sub": None}, {"preferred_username": None}, {"preferred_username": " "}, {"preferred_username": 42}, ], ) def test_invalid_claims_are_rejected(idp, provider, overrides): _, flow = idp.begin() with pytest.raises(LoginRejected): sign_in(idp, provider, flow, mint(flow, **overrides)) def test_a_token_signed_by_another_key_is_rejected(idp, provider): _, flow = idp.begin() with pytest.raises(LoginRejected): sign_in(idp, provider, flow, mint(flow, key=OTHER_KEY)) def test_a_token_signed_with_the_client_secret_is_rejected(idp, provider): _, flow = idp.begin() forged = mint(flow, key=OctKey.import_key(SECRET), alg="HS256", header={}) with pytest.raises(LoginRejected): sign_in(idp, provider, flow, forged) def test_an_unsigned_token_is_rejected(idp, provider): _, flow = idp.begin() unsigned = "eyJhbGciOiJub25lIn0.eyJzdWIiOiJ4In0." with pytest.raises(LoginRejected): sign_in(idp, provider, flow, unsigned) def test_a_rotated_key_is_picked_up_with_one_refetch(idp, provider): _, flow = idp.begin() sign_in(idp, provider, flow, mint(flow)) # caches the key set with k1 only provider.routes[JWKS_URL] = (200, KeySet([KEY, ROTATED_KEY]).as_dict(private=False)) _, flow = idp.begin() token = mint(flow, key=ROTATED_KEY, header={"kid": "k2"}) assert sign_in(idp, provider, flow, token).username == "alice" assert provider.count(JWKS_URL) == 2 def test_an_unknown_key_is_rejected_after_one_refetch(idp, provider): _, flow = idp.begin() token = mint(flow, key=ROTATED_KEY, header={"kid": "k2"}) with pytest.raises(LoginRejected): sign_in(idp, provider, flow, token) assert provider.count(JWKS_URL) == 2 def test_a_different_issuer_in_discovery_is_an_outage(idp, provider): provider.routes[DISCOVERY_URL] = ( 200, {**provider.routes[DISCOVERY_URL][1], "issuer": "https://evil.example.test/"}, ) with pytest.raises(IdpUnavailable): idp.begin() def test_a_discovery_document_with_a_plain_http_endpoint_is_an_outage(idp, provider): document = {**provider.routes[DISCOVERY_URL][1], "token_endpoint": "http://x.test/token"} provider.routes[DISCOVERY_URL] = (200, document) with pytest.raises(IdpUnavailable): idp.begin() def test_an_unreachable_provider_is_an_outage(idp, provider): provider.error = requests.ConnectionError("down") with pytest.raises(IdpUnavailable): idp.begin()