setup fastapi auth using authentik instance

crawler/api/app.py

@@ -1,31 +1,27 @@
 from typing import Annotated
-from fastapi import Depends, FastAPI, HTTPException, status
-from fastapi.security import OAuth2PasswordBearer
-from models.user import User
+from api.auth import get_current_user
+from api.config import DEV_TIER_ORIGINS, PROD_TIER_ORIGINS
+from fastapi import Depends, FastAPI
+from api.auth import User
 from repositories.listing_repository import ListingRepository
 from repositories.listing_repository import ListingRepository
 from database import engine
-from repositories.user_repository import UserRepository
+from fastapi.middleware.cors import CORSMiddleware
 
 
 app = FastAPI()
-oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
 
-
-async def decode_token(token: Annotated[str, Depends(oauth2_scheme)]):
-    repository = UserRepository(engine)
-    user = await repository.get_user_from_token(token)
-    if not user:
-        raise HTTPException(
-            status_code=status.HTTP_401_UNAUTHORIZED,
-            detail="Invalid authentication credentials",
-            headers={"WWW-Authenticate": "Bearer"},
-        )
-    return user
+# Allow CORS (for React frontend)
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=[*DEV_TIER_ORIGINS, *PROD_TIER_ORIGINS],
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
 
 
 @app.get("/listing")
-async def get_listing(user: Annotated[User | None, Depends(decode_token)]):
+async def get_listing(user: Annotated[User, Depends(get_current_user)]):
     repository = ListingRepository(engine)
-    listings = await repository.get_listings()
+    listings = await repository.get_listings(limit=5)
     return {"listings": listings}

crawler/api/auth.py

@@ -0,0 +1,64 @@
+from datetime import timedelta
+from api.config import AUTHENTIK_URL, OIDC_CACHE_TTL, OIDC_CLIENT_ID, OIDC_METADATA_URL
+from cachetools import TTLCache
+from fastapi import Depends, HTTPException
+from fastapi.security import OAuth2AuthorizationCodeBearer
+from httpx import AsyncClient
+import jwt
+from pydantic import BaseModel
+
+
+# OAuth2 Scheme
+oauth2_scheme = OAuth2AuthorizationCodeBearer(
+    authorizationUrl=f"{AUTHENTIK_URL}/application/o/authorize/",
+    tokenUrl=f"{AUTHENTIK_URL}/application/o/token/",
+)
+
+JWKS_CACHE = TTLCache(maxsize=1, ttl=OIDC_CACHE_TTL)
+OIDC_METADATA_CACHE = TTLCache(maxsize=1, ttl=OIDC_CACHE_TTL)
+
+
+class User(BaseModel):
+    sub: str  # User ID
+    email: str
+    name: str
+
+
+async def get_oidc_metadata():
+    if "oidc_metadata" not in OIDC_METADATA_CACHE:
+        async with AsyncClient() as client:
+            resp = await client.get(OIDC_METADATA_URL, follow_redirects=True)
+        OIDC_METADATA_CACHE["oidc_metadata"] = resp.json()
+    return OIDC_METADATA_CACHE["oidc_metadata"]
+
+
+async def get_cached_jwks_client() -> jwt.PyJWKClient:
+    if "jwks_client" not in JWKS_CACHE:
+        metadata = await get_oidc_metadata()
+        jwks_url = metadata["jwks_uri"]
+        JWKS_CACHE["jwks_client"] = jwt.PyJWKClient(
+            jwks_url,
+            cache_keys=True,  # PyJWT's built-in key caching
+            max_cached_keys=5,
+        )
+    return JWKS_CACHE["jwks_client"]
+
+
+async def get_current_user(token: str = Depends(oauth2_scheme)) -> User:
+    try:
+        # Fetch JWKS keys from Authentik
+        metadata = await get_oidc_metadata()
+        signing_key = (await get_cached_jwks_client()).get_signing_key_from_jwt(token)
+
+        # Decode and verify JWT
+        payload = jwt.decode(
+            token,
+            signing_key,
+            algorithms=["RS256"],
+            audience=OIDC_CLIENT_ID,
+            issuer=metadata["issuer"],
+            options={"verify_exp": False},
+        )
+        return User(**payload)
+    except jwt.PyJWTError as e:
+        raise HTTPException(status_code=401, detail=f"Invalid token: {e}")

crawler/api/config.py

@@ -0,0 +1,16 @@
+from datetime import timedelta
+
+
+# Authentik OIDC Configuration
+AUTHENTIK_URL = "https://authentik.viktorbarzin.me/"
+OIDC_CLIENT_ID = "5AJKRgcdgVm1OyApBzFkadDFfStW9a555zwv2MOe"
+OIDC_METADATA_URL = (
+    f"{AUTHENTIK_URL}/application/o/wrongmove/.well-known/openid-configuration"
+)
+
+OIDC_CACHE_TTL = timedelta(
+    hours=1
+).total_seconds()  # Cache to avoid spamming authentik with requests
+
+DEV_TIER_ORIGINS = ["https://devvm.viktorbarzin.lan/"]
+PROD_TIER_ORIGINS = ["https://wrongmove.viktorbarzin.me/"]