188 lines
7 KiB
Python
188 lines
7 KiB
Python
import json
|
|
import secrets
|
|
from base64 import b64encode
|
|
from datetime import UTC, datetime
|
|
from urllib.parse import parse_qs, urlparse
|
|
|
|
from argon2 import PasswordHasher
|
|
from cryptojwt.jwk.jwk import key_from_jwk_dict
|
|
from cryptojwt.jws.jws import JWS
|
|
from httpx import AsyncClient
|
|
|
|
from porchlight.authn.password import PasswordService
|
|
from porchlight.models import PasswordCredential, User
|
|
from tests.conftest import get_csrf_token
|
|
|
|
|
|
async def test_full_authorization_code_flow(client: AsyncClient) -> None:
|
|
"""End-to-end test of the complete OIDC Authorization Code flow.
|
|
|
|
Exercises: discovery, client registration, authorization request,
|
|
password login, authorization completion, token exchange, ID token
|
|
validation (JWKS + JWT signature), and userinfo.
|
|
"""
|
|
# -- Setup: register RP client and create user --
|
|
app = client._transport.app # type: ignore[union-attr]
|
|
oidc_server = app.state.oidc_server
|
|
user_repo = app.state.user_repo
|
|
cred_repo = app.state.credential_repo
|
|
|
|
client_id = "e2e-rp"
|
|
client_secret = "e2e-secret-0123456789abcdef" # 30+ chars
|
|
redirect_uri = "http://localhost:9000/callback"
|
|
state = secrets.token_urlsafe(16)
|
|
nonce = secrets.token_urlsafe(16)
|
|
|
|
oidc_server.context.cdb[client_id] = {
|
|
"client_id": client_id,
|
|
"client_secret": client_secret,
|
|
"redirect_uris": [(redirect_uri, {})],
|
|
"response_types_supported": ["code"],
|
|
"token_endpoint_auth_method": "client_secret_basic",
|
|
"scope": ["openid", "profile", "email"],
|
|
"allowed_scopes": ["openid", "profile", "email"],
|
|
"client_salt": secrets.token_hex(8),
|
|
}
|
|
oidc_server.keyjar.add_symmetric(client_id, client_secret)
|
|
|
|
user = User(
|
|
userid="lusab-bansen",
|
|
username="alice",
|
|
email="alice@example.com",
|
|
email_verified=True,
|
|
given_name="Alice",
|
|
family_name="Wonderland",
|
|
created_at=datetime.now(UTC),
|
|
updated_at=datetime.now(UTC),
|
|
)
|
|
await user_repo.create(user)
|
|
|
|
svc = PasswordService(hasher=PasswordHasher(time_cost=1, memory_cost=8192))
|
|
await cred_repo.create_password(PasswordCredential(user_id=user.userid, password_hash=svc.hash("testpass")))
|
|
|
|
# -- Step 1: Authorization request (unauthenticated) → redirect to /login --
|
|
auth_res = await client.get(
|
|
"/authorization",
|
|
params={
|
|
"response_type": "code",
|
|
"client_id": client_id,
|
|
"redirect_uri": redirect_uri,
|
|
"scope": "openid profile email",
|
|
"state": state,
|
|
"nonce": nonce,
|
|
},
|
|
follow_redirects=False,
|
|
)
|
|
assert auth_res.status_code in (302, 303), f"Expected redirect to /login, got {auth_res.status_code}"
|
|
assert "/login" in auth_res.headers["location"]
|
|
|
|
# -- Step 2: Password login → HX-Redirect to /authorization/complete --
|
|
token = await get_csrf_token(client)
|
|
login_res = await client.post(
|
|
"/login/password",
|
|
data={"username": "alice", "password": "testpass"},
|
|
headers={"HX-Request": "true", "X-CSRF-Token": token},
|
|
)
|
|
assert login_res.status_code == 200
|
|
hx_redirect = login_res.headers.get("HX-Redirect", "")
|
|
assert "/authorization/complete" in hx_redirect, (
|
|
f"Expected HX-Redirect to /authorization/complete, got '{hx_redirect}'"
|
|
)
|
|
|
|
# -- Step 3: Complete authorization → redirect to consent --
|
|
complete_res = await client.get("/authorization/complete", follow_redirects=False)
|
|
assert complete_res.status_code in (302, 303), (
|
|
f"Expected redirect to /consent, got {complete_res.status_code}: {complete_res.text}"
|
|
)
|
|
assert "/consent" in complete_res.headers["location"]
|
|
|
|
# -- Step 3b: Approve consent → redirect to callback with code + state --
|
|
token = await get_csrf_token(client)
|
|
consent_res = await client.post(
|
|
"/consent",
|
|
data={"action": "allow", "scope": ["openid", "profile", "email"]},
|
|
headers={"X-CSRF-Token": token},
|
|
follow_redirects=False,
|
|
)
|
|
assert consent_res.status_code in (302, 303)
|
|
|
|
location = consent_res.headers["location"]
|
|
parsed = urlparse(location)
|
|
assert parsed.netloc == "localhost:9000"
|
|
assert parsed.path == "/callback"
|
|
|
|
callback_params = parse_qs(parsed.query)
|
|
assert "code" in callback_params, f"No code in redirect: {location}"
|
|
assert callback_params["state"] == [state], f"State mismatch: expected {state}, got {callback_params.get('state')}"
|
|
|
|
code = callback_params["code"][0]
|
|
assert len(code) > 0
|
|
|
|
# -- Step 4: Token exchange → access_token, id_token, token_type --
|
|
auth_header = b64encode(f"{client_id}:{client_secret}".encode()).decode()
|
|
token_res = await client.post(
|
|
"/token",
|
|
data={
|
|
"grant_type": "authorization_code",
|
|
"code": code,
|
|
"redirect_uri": redirect_uri,
|
|
},
|
|
headers={
|
|
"Authorization": f"Basic {auth_header}",
|
|
"Content-Type": "application/x-www-form-urlencoded",
|
|
},
|
|
)
|
|
assert token_res.status_code == 200, f"Token endpoint failed: {token_res.text}"
|
|
|
|
token_data = token_res.json()
|
|
assert "access_token" in token_data
|
|
assert "id_token" in token_data
|
|
assert token_data["token_type"].lower() == "bearer"
|
|
|
|
access_token = token_data["access_token"]
|
|
id_token_jwt = token_data["id_token"]
|
|
|
|
# -- Step 5: Validate ID token — fetch JWKS, verify JWT signature --
|
|
jwks_res = await client.get("/jwks")
|
|
assert jwks_res.status_code == 200
|
|
jwks = jwks_res.json()
|
|
assert "keys" in jwks
|
|
assert len(jwks["keys"]) > 0
|
|
|
|
# Decode the ID token header to find the key ID
|
|
id_token_header = json.loads(JWS()._decode(id_token_jwt.split(".")[0]))
|
|
kid = id_token_header.get("kid")
|
|
|
|
# Find the matching key in JWKS
|
|
matching_keys = [k for k in jwks["keys"] if k.get("kid") == kid]
|
|
assert len(matching_keys) == 1, f"Expected 1 matching key for kid={kid}, found {len(matching_keys)}"
|
|
|
|
key = key_from_jwk_dict(matching_keys[0])
|
|
|
|
# Verify JWT signature and decode payload
|
|
verifier = JWS()
|
|
payload = verifier.verify_compact(id_token_jwt, keys=[key])
|
|
id_token_payload = json.loads(payload) if isinstance(payload, (str, bytes)) else payload
|
|
|
|
assert id_token_payload["iss"] == "http://localhost:8000"
|
|
assert client_id in id_token_payload["aud"]
|
|
assert id_token_payload["nonce"] == nonce
|
|
|
|
# sub is a pairwise identifier — just verify it exists and is non-empty
|
|
id_token_sub = id_token_payload["sub"]
|
|
assert isinstance(id_token_sub, str)
|
|
assert len(id_token_sub) > 0
|
|
|
|
# -- Step 6: UserInfo — GET /userinfo with Bearer token --
|
|
userinfo_res = await client.get(
|
|
"/userinfo",
|
|
headers={"Authorization": f"Bearer {access_token}"},
|
|
)
|
|
assert userinfo_res.status_code == 200, f"UserInfo failed: {userinfo_res.status_code} {userinfo_res.text}"
|
|
|
|
userinfo_data = userinfo_res.json()
|
|
assert userinfo_data["email"] == "alice@example.com"
|
|
assert userinfo_data["email_verified"] is True
|
|
|
|
# sub should match between id_token and userinfo
|
|
assert userinfo_data["sub"] == id_token_sub
|