import pytest from fastapi import HTTPException from app.auth import keycloak_dependencies from app.auth.keycloak_dependencies import ( _decode_keycloak_token, get_current_keycloak_username, ) @pytest.fixture def anyio_backend(): return "asyncio" @pytest.mark.anyio async def test_current_username_uses_preferred_username_only(): username = await get_current_keycloak_username( { "preferred_username": "tjwater", "username": "legacy-name", } ) assert username == "tjwater" @pytest.mark.anyio async def test_current_username_rejects_username_fallback(): with pytest.raises(HTTPException) as exc: await get_current_keycloak_username({"username": "legacy-name"}) assert exc.value.status_code == 401 assert exc.value.detail == "Missing preferred_username claim" def test_decode_keycloak_token_rejects_a_token_older_than_the_configured_limit( monkeypatch: pytest.MonkeyPatch, ): monkeypatch.setattr(keycloak_dependencies.settings, "KEYCLOAK_PUBLIC_KEY", "public-key") monkeypatch.setattr( keycloak_dependencies.settings, "KEYCLOAK_ACCESS_TOKEN_MAX_AGE_SECONDS", 900, ) monkeypatch.setattr(keycloak_dependencies.time, "time", lambda: 2_000) monkeypatch.setattr( keycloak_dependencies.jwt, "decode", lambda *args, **kwargs: {"iat": 1_000}, ) with pytest.raises(keycloak_dependencies.JWTError): _decode_keycloak_token("expired-by-policy") def test_decode_keycloak_token_accepts_a_recent_token( monkeypatch: pytest.MonkeyPatch, ): monkeypatch.setattr(keycloak_dependencies.settings, "KEYCLOAK_PUBLIC_KEY", "public-key") monkeypatch.setattr( keycloak_dependencies.settings, "KEYCLOAK_ACCESS_TOKEN_MAX_AGE_SECONDS", 900, ) monkeypatch.setattr(keycloak_dependencies.time, "time", lambda: 1_500) monkeypatch.setattr( keycloak_dependencies.jwt, "decode", lambda *args, **kwargs: {"iat": 1_000, "sub": "subject"}, ) assert _decode_keycloak_token("recent") == {"iat": 1_000, "sub": "subject"} @pytest.mark.parametrize("claims", [ {"aud": "tjwater-mcp"}, {"aud": ["account", "tjwater-mcp"]}, {"azp": "tjwater-agent-mcp"}, ]) @pytest.mark.parametrize("host_client_id", ["agent-mcp", "tjwater-agent-mcp"]) def test_product_rest_rejects_mcp_resource_tokens_even_without_api_audience(monkeypatch, claims, host_client_id): monkeypatch.setattr(keycloak_dependencies.mcp_resource_settings, "audience", "tjwater-mcp") monkeypatch.setattr(keycloak_dependencies.mcp_resource_settings, "host_client_id", host_client_id) if "azp" in claims: claims = {**claims, "azp": host_client_id} monkeypatch.setattr(keycloak_dependencies.settings, "KEYCLOAK_PUBLIC_KEY", "public-key") monkeypatch.setattr(keycloak_dependencies.settings, "KEYCLOAK_AUDIENCE", "") monkeypatch.setattr(keycloak_dependencies.settings, "KEYCLOAK_ACCESS_TOKEN_MAX_AGE_SECONDS", 0) monkeypatch.setattr(keycloak_dependencies.jwt, "decode", lambda *args, **kwargs: claims) with pytest.raises(keycloak_dependencies.JWTError): _decode_keycloak_token("resource-token")