test: unit tests for core/auth against a mock jwks

"""Unit tests for core/auth: a local RSA keypair signs test tokens, and a real

local HTTP server serves the public half as a JWKS response -- no live Keycloak,

no mocking of PyJWT/urllib internals."""

import json

import threading

import time

from collections.abc import Generator

from http.server import BaseHTTPRequestHandler, HTTPServer

from typing import Any, Optional

import jwt

import pytest

from cryptography.hazmat.primitives.asymmetric import rsa

from cryptography.hazmat.primitives.asymmetric.rsa import RSAPrivateKey

from jwt.algorithms import RSAAlgorithm

from open_exposure_gateway.core.auth.jwks_cache import JWKSCache

from open_exposure_gateway.core.auth.tokens import TokenFlow, validate_token

from open_exposure_gateway.core.config import KeycloakSettings

from open_exposure_gateway.core.exceptions import UnauthorizedException

_ISSUER = "https://keycloak.test/realms/oop"

_AUDIENCE = "oeg"

_KID = "test-kid"

def _settings() -> KeycloakSettings:

    return KeycloakSettings(issuer=_ISSUER, jwks_url="unused", audience=_AUDIENCE)

def _make_token(

    private_key: RSAPrivateKey,

    kid: Optional[str] = _KID,

    omit: tuple[str, ...] = (),

    **overrides: Any,

) -> str:

    now = int(time.time())

    payload: dict[str, Any] = {

        "iss": _ISSUER,

        "aud": _AUDIENCE,

        "sub": "service-account-test-client",

        "iat": now,

        "exp": now + 300,

        "realm_access": {"roles": ["app-provider"]},

        "organization": "acme",

        **overrides,

    }

    for key in omit:

        payload.pop(key, None)

    headers = {"kid": kid} if kid else None

    return jwt.encode(payload, private_key, algorithm="RS256", headers=headers)

def _jwks_handler(body: bytes) -> type[BaseHTTPRequestHandler]:

    class Handler(BaseHTTPRequestHandler):

        def do_GET(self) -> None:

            self.send_response(200)

            self.send_header("Content-Type", "application/json")

            self.send_header("Content-Length", str(len(body)))

            self.end_headers()

            self.wfile.write(body)

        def log_message(self, format: str, *args: Any) -> None:

            pass

    return Handler

@pytest.fixture()

def rsa_private_key() -> RSAPrivateKey:

    return rsa.generate_private_key(public_exponent=65537, key_size=2048)

@pytest.fixture()

def jwks_url(rsa_private_key: RSAPrivateKey) -> Generator[str, None, None]:

    jwk = json.loads(RSAAlgorithm.to_jwk(rsa_private_key.public_key()))

    jwk["kid"] = _KID

    jwk["use"] = "sig"

    jwk["alg"] = "RS256"

    body = json.dumps({"keys": [jwk]}).encode()

    server = HTTPServer(("localhost", 0), _jwks_handler(body))

    thread = threading.Thread(target=server.serve_forever, daemon=True)

    thread.start()

    try:

        yield f"http://localhost:{server.server_port}/certs"

    finally:

        server.shutdown()

        thread.join()

class TestValidateToken:

    async def test_valid_token_is_accepted(

        self, rsa_private_key: RSAPrivateKey, jwks_url: str

    ) -> None:

        decoded = await validate_token(

            _make_token(rsa_private_key), JWKSCache(jwks_url), _settings()

        )

        assert decoded.claims["sub"] == "service-account-test-client"

    async def test_wrong_issuer_is_rejected(

        self, rsa_private_key: RSAPrivateKey, jwks_url: str

    ) -> None:

        token = _make_token(rsa_private_key, iss="https://evil.example/realms/oop")

        with pytest.raises(UnauthorizedException):

            await validate_token(token, JWKSCache(jwks_url), _settings())

    async def test_wrong_audience_is_rejected(

        self, rsa_private_key: RSAPrivateKey, jwks_url: str

    ) -> None:

        token = _make_token(rsa_private_key, aud="someone-else")

        with pytest.raises(UnauthorizedException):

            await validate_token(token, JWKSCache(jwks_url), _settings())

    async def test_missing_audience_is_rejected(

        self, rsa_private_key: RSAPrivateKey, jwks_url: str

    ) -> None:

        token = _make_token(rsa_private_key, omit=("aud",))

        with pytest.raises(UnauthorizedException):

            await validate_token(token, JWKSCache(jwks_url), _settings())

    async def test_expired_token_is_rejected(

        self, rsa_private_key: RSAPrivateKey, jwks_url: str

    ) -> None:

        token = _make_token(rsa_private_key, exp=int(time.time()) - 10)

        with pytest.raises(UnauthorizedException):

            await validate_token(token, JWKSCache(jwks_url), _settings())

    async def test_malformed_token_is_rejected(self, jwks_url: str) -> None:

        with pytest.raises(UnauthorizedException):

            await validate_token("not-a-jwt", JWKSCache(jwks_url), _settings())

    async def test_unsigned_token_is_rejected(self, jwks_url: str) -> None:

        # alg=none: pinning algorithms=["RS256"] in validate_token must reject

        # this regardless of signature, or it's an algorithm-confusion hole.

        token = jwt.encode({"iss": _ISSUER, "aud": _AUDIENCE, "sub": "x"}, key="", algorithm="none")

        with pytest.raises(UnauthorizedException):

            await validate_token(token, JWKSCache(jwks_url), _settings())

    async def test_signature_from_a_different_keypair_is_rejected(self, jwks_url: str) -> None:

        other_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)

        token = _make_token(other_key)

        with pytest.raises(UnauthorizedException):

            await validate_token(token, JWKSCache(jwks_url), _settings())

class TestTokenFlowClassification:

    async def test_token_with_auth_time_is_three_legged(

        self, rsa_private_key: RSAPrivateKey, jwks_url: str

    ) -> None:

        token = _make_token(rsa_private_key, auth_time=int(time.time()))

        decoded = await validate_token(token, JWKSCache(jwks_url), _settings())

        assert decoded.flow == TokenFlow.THREE_LEGGED

    async def test_token_without_auth_time_is_two_legged(

        self, rsa_private_key: RSAPrivateKey, jwks_url: str

    ) -> None:

        token = _make_token(rsa_private_key)

        decoded = await validate_token(token, JWKSCache(jwks_url), _settings())

        assert decoded.flow == TokenFlow.TWO_LEGGED

    async def test_flow_is_cached_after_first_access(

        self, rsa_private_key: RSAPrivateKey, jwks_url: str

    ) -> None:

        decoded = await validate_token(

            _make_token(rsa_private_key), JWKSCache(jwks_url), _settings()

        )

        assert decoded.flow == TokenFlow.TWO_LEGGED

        # Mutating claims after the first access must not change the result --

        # `flow` is derived once (cached_property), not recomputed per read.

        decoded.claims["auth_time"] = int(time.time())

        assert decoded.flow == TokenFlow.TWO_LEGGED