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_PUBLIC_KEY="-----BEGIN PUBLIC KEY-----\n...\n-----END PUBLIC KEY-----"
|
||||||
KEYCLOAK_ALGORITHM=RS256
|
KEYCLOAK_ALGORITHM=RS256
|
||||||
KEYCLOAK_AUDIENCE="account"
|
KEYCLOAK_AUDIENCE="account"
|
||||||
|
KEYCLOAK_ACCESS_TOKEN_MAX_AGE_SECONDS=900
|
||||||
|
|
||||||
|
|
||||||
# ============================================
|
# ============================================
|
||||||
|
|||||||
@@ -1,4 +1,4 @@
|
|||||||
# import logging
|
import time
|
||||||
from uuid import UUID
|
from uuid import UUID
|
||||||
|
|
||||||
from fastapi import Depends, HTTPException, status
|
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")
|
key = settings.KEYCLOAK_PUBLIC_KEY.replace("\\n", "\n")
|
||||||
|
|
||||||
return jwt.decode(
|
payload = jwt.decode(
|
||||||
token,
|
token,
|
||||||
key,
|
key,
|
||||||
algorithms=[settings.KEYCLOAK_ALGORITHM],
|
algorithms=[settings.KEYCLOAK_ALGORITHM],
|
||||||
audience=settings.KEYCLOAK_AUDIENCE or None,
|
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(
|
async def get_current_keycloak_payload(
|
||||||
|
|||||||
@@ -48,6 +48,7 @@ class Settings(BaseSettings):
|
|||||||
KEYCLOAK_PUBLIC_KEY: str = ""
|
KEYCLOAK_PUBLIC_KEY: str = ""
|
||||||
KEYCLOAK_ALGORITHM: str = "RS256"
|
KEYCLOAK_ALGORITHM: str = "RS256"
|
||||||
KEYCLOAK_AUDIENCE: str = ""
|
KEYCLOAK_AUDIENCE: str = ""
|
||||||
|
KEYCLOAK_ACCESS_TOKEN_MAX_AGE_SECONDS: int = 900
|
||||||
|
|
||||||
# Bocha Web Search API
|
# Bocha Web Search API
|
||||||
BOCHA_API_KEY: str = ""
|
BOCHA_API_KEY: str = ""
|
||||||
|
|||||||
@@ -1,7 +1,11 @@
|
|||||||
import pytest
|
import pytest
|
||||||
from fastapi import HTTPException
|
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
|
@pytest.fixture
|
||||||
@@ -28,3 +32,42 @@ async def test_current_username_rejects_username_fallback():
|
|||||||
|
|
||||||
assert exc.value.status_code == 401
|
assert exc.value.status_code == 401
|
||||||
assert exc.value.detail == "Missing preferred_username claim"
|
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