fix(auth): enforce Keycloak access token age
This commit is contained in:
@@ -45,6 +45,7 @@ METADATA_DB_PASSWORD="password"
|
||||
KEYCLOAK_PUBLIC_KEY="-----BEGIN PUBLIC KEY-----\n...\n-----END PUBLIC KEY-----"
|
||||
KEYCLOAK_ALGORITHM=RS256
|
||||
KEYCLOAK_AUDIENCE="account"
|
||||
KEYCLOAK_ACCESS_TOKEN_MAX_AGE_SECONDS=900
|
||||
|
||||
|
||||
# ============================================
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
# import logging
|
||||
import time
|
||||
from uuid import UUID
|
||||
|
||||
from fastapi import Depends, HTTPException, status
|
||||
@@ -23,12 +23,22 @@ def _decode_keycloak_token(token: str) -> dict:
|
||||
|
||||
key = settings.KEYCLOAK_PUBLIC_KEY.replace("\\n", "\n")
|
||||
|
||||
return jwt.decode(
|
||||
payload = jwt.decode(
|
||||
token,
|
||||
key,
|
||||
algorithms=[settings.KEYCLOAK_ALGORITHM],
|
||||
audience=settings.KEYCLOAK_AUDIENCE or None,
|
||||
)
|
||||
if settings.KEYCLOAK_ACCESS_TOKEN_MAX_AGE_SECONDS <= 0:
|
||||
return payload
|
||||
|
||||
issued_at = payload.get("iat")
|
||||
if not isinstance(issued_at, (int, float)) or (
|
||||
time.time() >= issued_at + settings.KEYCLOAK_ACCESS_TOKEN_MAX_AGE_SECONDS
|
||||
):
|
||||
raise JWTError("Keycloak access token is older than the allowed maximum age")
|
||||
|
||||
return payload
|
||||
|
||||
|
||||
async def get_current_keycloak_payload(
|
||||
|
||||
@@ -53,6 +53,7 @@ class Settings(BaseSettings):
|
||||
KEYCLOAK_PUBLIC_KEY: str = ""
|
||||
KEYCLOAK_ALGORITHM: str = "RS256"
|
||||
KEYCLOAK_AUDIENCE: str = ""
|
||||
KEYCLOAK_ACCESS_TOKEN_MAX_AGE_SECONDS: int = 900
|
||||
|
||||
# Bocha Web Search API
|
||||
BOCHA_API_KEY: str = ""
|
||||
|
||||
@@ -1,7 +1,11 @@
|
||||
import pytest
|
||||
from fastapi import HTTPException
|
||||
|
||||
from app.auth.keycloak_dependencies import get_current_keycloak_username
|
||||
from app.auth import keycloak_dependencies
|
||||
from app.auth.keycloak_dependencies import (
|
||||
_decode_keycloak_token,
|
||||
get_current_keycloak_username,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
@@ -28,3 +32,42 @@ async def test_current_username_rejects_username_fallback():
|
||||
|
||||
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"}
|
||||
|
||||
Reference in New Issue
Block a user