271 lines
9.4 KiB
Python
271 lines
9.4 KiB
Python
"""AuthentikOidcProvider: PKCE, discovery, and real ID token signature/claim
|
|
validation against a JWKS (issuer, audience, expiry, nonce)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
import time
|
|
from collections.abc import Callable
|
|
from urllib.parse import parse_qs, urlsplit
|
|
|
|
import httpx
|
|
import jwt
|
|
import pytest
|
|
from cryptography.hazmat.primitives.asymmetric import rsa
|
|
from jwt.algorithms import RSAAlgorithm
|
|
|
|
from ragcore.application.auth.ports import (
|
|
IdTokenValidationError,
|
|
InvalidNonceError,
|
|
TokenExchangeError,
|
|
)
|
|
from ragcore.infrastructure.auth.oidc import AuthentikOidcProvider, OidcClientConfig
|
|
|
|
ISSUER = "https://auth.example.test/application/o/ragcore/"
|
|
CLIENT_ID = "ragcore"
|
|
REDIRECT_URI = "https://ragcore.example.test/auth/callback"
|
|
KID = "test-signing-key"
|
|
|
|
_private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
|
|
|
|
|
def _jwk_public() -> dict[str, object]:
|
|
jwk = RSAAlgorithm.to_jwk(_private_key.public_key(), as_dict=True)
|
|
jwk["kid"] = KID
|
|
jwk["use"] = "sig"
|
|
jwk["alg"] = "RS256"
|
|
return jwk
|
|
|
|
|
|
def _sign_id_token(claims: dict[str, object]) -> str:
|
|
return jwt.encode(claims, _private_key, algorithm="RS256", headers={"kid": KID})
|
|
|
|
|
|
def _base_claims(*, nonce: str, aud: str = CLIENT_ID, iss: str = ISSUER) -> dict[str, object]:
|
|
now = int(time.time())
|
|
return {
|
|
"iss": iss,
|
|
"aud": aud,
|
|
"sub": "user-123",
|
|
"exp": now + 300,
|
|
"iat": now,
|
|
"nonce": nonce,
|
|
"email": "user@example.test",
|
|
"groups": ["ragcore-editors"],
|
|
}
|
|
|
|
|
|
def _handler_factory(
|
|
id_token_claims_override: dict[str, object] | None = None,
|
|
) -> tuple[Callable[[httpx.Request], httpx.Response], dict[str, object]]:
|
|
state = {"token_calls": 0, "code_verifier": None}
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
if request.url.path.endswith("/.well-known/openid-configuration"):
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"issuer": ISSUER,
|
|
"authorization_endpoint": f"{ISSUER}authorize",
|
|
"token_endpoint": f"{ISSUER}token",
|
|
"jwks_uri": f"{ISSUER}jwks",
|
|
"end_session_endpoint": f"{ISSUER}logout",
|
|
},
|
|
)
|
|
if request.url.path.endswith("/jwks"):
|
|
return httpx.Response(200, json={"keys": [_jwk_public()]})
|
|
if request.url.path.endswith("/token"):
|
|
state["token_calls"] += 1
|
|
form = dict(
|
|
pair.split("=", 1) for pair in request.content.decode().split("&") if "=" in pair
|
|
)
|
|
state["code_verifier"] = form.get("code_verifier")
|
|
claims = id_token_claims_override or _base_claims(nonce="expected-nonce")
|
|
return httpx.Response(200, json={"id_token": _sign_id_token(claims)})
|
|
return httpx.Response(404)
|
|
|
|
return handler, state
|
|
|
|
|
|
def _provider(handler: Callable[[httpx.Request], httpx.Response]) -> AuthentikOidcProvider:
|
|
client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
|
|
return AuthentikOidcProvider(
|
|
config=OidcClientConfig(issuer=ISSUER, client_id=CLIENT_ID, client_secret="secret"),
|
|
client=client,
|
|
)
|
|
|
|
|
|
async def test_build_authorization_challenge_has_valid_pkce_s256_and_oidc_params() -> None:
|
|
handler, _ = _handler_factory()
|
|
provider = _provider(handler)
|
|
|
|
challenge = await provider.build_authorization_challenge(redirect_uri=REDIRECT_URI)
|
|
|
|
parsed = urlsplit(challenge.redirect_url)
|
|
query = parse_qs(parsed.query)
|
|
assert query["response_type"] == ["code"]
|
|
assert query["client_id"] == [CLIENT_ID]
|
|
assert query["redirect_uri"] == [REDIRECT_URI]
|
|
assert query["state"] == [challenge.state]
|
|
assert query["nonce"] == [challenge.nonce]
|
|
assert query["code_challenge_method"] == ["S256"]
|
|
expected_challenge = (
|
|
base64.urlsafe_b64encode(hashlib.sha256(challenge.code_verifier.encode()).digest())
|
|
.rstrip(b"=")
|
|
.decode()
|
|
)
|
|
assert query["code_challenge"] == [expected_challenge]
|
|
|
|
|
|
async def test_two_challenges_never_reuse_state_nonce_or_verifier() -> None:
|
|
handler, _ = _handler_factory()
|
|
provider = _provider(handler)
|
|
|
|
first = await provider.build_authorization_challenge(redirect_uri=REDIRECT_URI)
|
|
second = await provider.build_authorization_challenge(redirect_uri=REDIRECT_URI)
|
|
|
|
assert first.state != second.state
|
|
assert first.nonce != second.nonce
|
|
assert first.code_verifier != second.code_verifier
|
|
|
|
|
|
async def test_complete_login_validates_signature_issuer_audience_and_nonce() -> None:
|
|
handler, state = _handler_factory()
|
|
provider = _provider(handler)
|
|
|
|
claims = await provider.complete_login(
|
|
code="auth-code",
|
|
code_verifier="a-verifier",
|
|
nonce="expected-nonce",
|
|
redirect_uri=REDIRECT_URI,
|
|
)
|
|
|
|
assert claims.subject == "user-123"
|
|
assert claims.email == "user@example.test"
|
|
assert claims.groups == ("ragcore-editors",)
|
|
assert state["code_verifier"] == "a-verifier" # PKCE verifier reaches the token exchange
|
|
|
|
|
|
async def test_wrong_nonce_is_rejected() -> None:
|
|
handler, _ = _handler_factory(_base_claims(nonce="actual-nonce"))
|
|
provider = _provider(handler)
|
|
|
|
with pytest.raises(InvalidNonceError):
|
|
await provider.complete_login(
|
|
code="auth-code",
|
|
code_verifier="a-verifier",
|
|
nonce="a-different-nonce-entirely",
|
|
redirect_uri=REDIRECT_URI,
|
|
)
|
|
|
|
|
|
async def test_wrong_audience_is_rejected() -> None:
|
|
handler, _ = _handler_factory(_base_claims(nonce="n", aud="some-other-client"))
|
|
provider = _provider(handler)
|
|
|
|
with pytest.raises(IdTokenValidationError):
|
|
await provider.complete_login(
|
|
code="auth-code", code_verifier="v", nonce="n", redirect_uri=REDIRECT_URI
|
|
)
|
|
|
|
|
|
async def test_wrong_issuer_is_rejected() -> None:
|
|
handler, _ = _handler_factory(_base_claims(nonce="n", iss="https://not-the-real-idp.test/"))
|
|
provider = _provider(handler)
|
|
|
|
with pytest.raises(IdTokenValidationError):
|
|
await provider.complete_login(
|
|
code="auth-code", code_verifier="v", nonce="n", redirect_uri=REDIRECT_URI
|
|
)
|
|
|
|
|
|
async def test_expired_id_token_is_rejected() -> None:
|
|
now = int(time.time())
|
|
expired_claims = {
|
|
**_base_claims(nonce="n"),
|
|
"exp": now - 300,
|
|
"iat": now - 600,
|
|
}
|
|
handler, _ = _handler_factory(expired_claims)
|
|
provider = _provider(handler)
|
|
|
|
with pytest.raises(IdTokenValidationError):
|
|
await provider.complete_login(
|
|
code="auth-code", code_verifier="v", nonce="n", redirect_uri=REDIRECT_URI
|
|
)
|
|
|
|
|
|
async def test_signature_from_an_untrusted_key_is_rejected() -> None:
|
|
other_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
if request.url.path.endswith("/.well-known/openid-configuration"):
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"issuer": ISSUER,
|
|
"authorization_endpoint": f"{ISSUER}authorize",
|
|
"token_endpoint": f"{ISSUER}token",
|
|
"jwks_uri": f"{ISSUER}jwks",
|
|
},
|
|
)
|
|
if request.url.path.endswith("/jwks"):
|
|
return httpx.Response(200, json={"keys": [_jwk_public()]})
|
|
if request.url.path.endswith("/token"):
|
|
forged = jwt.encode(
|
|
_base_claims(nonce="n"), other_key, algorithm="RS256", headers={"kid": KID}
|
|
)
|
|
return httpx.Response(200, json={"id_token": forged})
|
|
return httpx.Response(404)
|
|
|
|
provider = _provider(handler)
|
|
with pytest.raises(IdTokenValidationError):
|
|
await provider.complete_login(
|
|
code="auth-code", code_verifier="v", nonce="n", redirect_uri=REDIRECT_URI
|
|
)
|
|
|
|
|
|
async def test_token_endpoint_failure_raises_a_stable_typed_error() -> None:
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
if request.url.path.endswith("/.well-known/openid-configuration"):
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"issuer": ISSUER,
|
|
"authorization_endpoint": f"{ISSUER}authorize",
|
|
"token_endpoint": f"{ISSUER}token",
|
|
"jwks_uri": f"{ISSUER}jwks",
|
|
},
|
|
)
|
|
if request.url.path.endswith("/token"):
|
|
return httpx.Response(400, json={"error": "invalid_grant"})
|
|
return httpx.Response(404)
|
|
|
|
provider = _provider(handler)
|
|
with pytest.raises(TokenExchangeError):
|
|
await provider.complete_login(
|
|
code="bad-code", code_verifier="v", nonce="n", redirect_uri=REDIRECT_URI
|
|
)
|
|
|
|
|
|
async def test_build_logout_url_uses_the_discovered_end_session_endpoint() -> None:
|
|
handler, _ = _handler_factory()
|
|
provider = _provider(handler)
|
|
|
|
url = await provider.build_logout_url(post_logout_redirect_uri="https://ragcore.example.test/")
|
|
|
|
assert url is not None
|
|
assert url.startswith(f"{ISSUER}logout")
|
|
assert "post_logout_redirect_uri" in url
|
|
|
|
|
|
def test_issuer_must_be_an_absolute_http_url() -> None:
|
|
with pytest.raises(ValueError, match="issuer"):
|
|
OidcClientConfig(issuer="not-a-url", client_id=CLIENT_ID, client_secret="secret")
|
|
|
|
|
|
def test_client_id_must_not_be_empty() -> None:
|
|
with pytest.raises(ValueError, match="client_id"):
|
|
OidcClientConfig(issuer=ISSUER, client_id="", client_secret="secret")
|