feat: add authentik jwt verification and me endpoints

.env

@@ -11,6 +11,9 @@ DB_PASSWORD=CHANGE_ME
 AUTHENTIK_BASE_URL=
 AUTHENTIK_ADMIN_TOKEN=
 AUTHENTIK_VERIFY_TLS=false
+AUTHENTIK_ISSUER=
+AUTHENTIK_JWKS_URL=
+AUTHENTIK_AUDIENCE=
 
 PUBLIC_FRONTEND_ORIGINS=https://member.ose.tw,https://mkt.ose.tw,https://admin.ose.tw
 INTERNAL_SHARED_SECRET=CHANGE_ME

.env.example

@@ -11,6 +11,9 @@ DB_PASSWORD=CHANGE_ME
 AUTHENTIK_BASE_URL=
 AUTHENTIK_ADMIN_TOKEN=
 AUTHENTIK_VERIFY_TLS=false
+AUTHENTIK_ISSUER=
+AUTHENTIK_JWKS_URL=
+AUTHENTIK_AUDIENCE=
 
 PUBLIC_FRONTEND_ORIGINS=https://member.ose.tw,https://mkt.ose.tw,https://admin.ose.tw
 INTERNAL_SHARED_SECRET=CHANGE_ME

.env.production.example

@@ -11,6 +11,9 @@ DB_PASSWORD=CHANGE_ME
 AUTHENTIK_BASE_URL=
 AUTHENTIK_ADMIN_TOKEN=
 AUTHENTIK_VERIFY_TLS=false
+AUTHENTIK_ISSUER=
+AUTHENTIK_JWKS_URL=
+AUTHENTIK_AUDIENCE=
 
 PUBLIC_FRONTEND_ORIGINS=https://member.ose.tw,https://mkt.ose.tw,https://admin.ose.tw
 INTERNAL_SHARED_SECRET=CHANGE_ME

README.md

@@ -15,10 +15,25 @@ uvicorn app.main:app --host 127.0.0.1 --port 8000 --reload
 
 1. Initialize API client whitelist table with `docs/API_CLIENTS_SQL.sql`.
 2. Initialize core tables with `backend/scripts/init_schema.sql`.
+3. Generate `api_key_hash` and update `api_clients` records, e.g.:
+
+```bash
+python scripts/generate_api_key_hash.py 'YOUR_PLAIN_KEY'
+```
+
+## Authentik JWT setup
+
+- Configure at least one of:
+  - `AUTHENTIK_JWKS_URL`
+  - `AUTHENTIK_ISSUER` (the service infers `<issuer>/jwks/`)
+- Optional:
+  - `AUTHENTIK_AUDIENCE` (enables audience claim validation)
 
 ## Main APIs
 
 - `GET /healthz`
+- `GET /me` (Bearer token required)
+- `GET /me/permissions/snapshot` (Bearer token required)
 - `POST /internal/users/upsert-by-sub`
 - `GET /internal/permissions/{authentik_sub}/snapshot`
 - `POST /admin/permissions/grant`

app/api/me.py

@@ -0,0 +1,46 @@
+from fastapi import APIRouter, Depends
+from sqlalchemy.orm import Session
+
+from app.db.session import get_db
+from app.repositories.permissions_repo import PermissionsRepository
+from app.repositories.users_repo import UsersRepository
+from app.schemas.auth import AuthentikPrincipal, MeSummaryResponse
+from app.schemas.permissions import PermissionSnapshotResponse
+from app.security.authentik_jwt import require_authenticated_principal
+from app.services.permission_service import PermissionService
+
+router = APIRouter(prefix="/me", tags=["me"])
+
+
+@router.get("", response_model=MeSummaryResponse)
+def get_me(
+    principal: AuthentikPrincipal = Depends(require_authenticated_principal),
+    db: Session = Depends(get_db),
+) -> MeSummaryResponse:
+    users_repo = UsersRepository(db)
+    user = users_repo.upsert_by_sub(
+        authentik_sub=principal.sub,
+        email=principal.email,
+        display_name=principal.name or principal.preferred_username,
+        is_active=True,
+    )
+    return MeSummaryResponse(sub=user.authentik_sub, email=user.email, display_name=user.display_name)
+
+
+@router.get("/permissions/snapshot", response_model=PermissionSnapshotResponse)
+def get_my_permission_snapshot(
+    principal: AuthentikPrincipal = Depends(require_authenticated_principal),
+    db: Session = Depends(get_db),
+) -> PermissionSnapshotResponse:
+    users_repo = UsersRepository(db)
+    perms_repo = PermissionsRepository(db)
+
+    user = users_repo.upsert_by_sub(
+        authentik_sub=principal.sub,
+        email=principal.email,
+        display_name=principal.name or principal.preferred_username,
+        is_active=True,
+    )
+    permissions = perms_repo.list_by_user_id(user.id)
+    tuples = [(p.scope_type, p.scope_id, p.module, p.action) for p in permissions]
+    return PermissionService.build_snapshot(authentik_sub=principal.sub, permissions=tuples)

app/core/config.py

@@ -20,6 +20,9 @@ class Settings(BaseSettings):
     authentik_base_url: str = ""
     authentik_admin_token: str = ""
     authentik_verify_tls: bool = False
+    authentik_issuer: str = ""
+    authentik_jwks_url: str = ""
+    authentik_audience: str = ""
 
     public_frontend_origins: Annotated[list[str], NoDecode] = ["https://member.ose.tw"]
     internal_shared_secret: str = ""

app/main.py

@@ -2,6 +2,7 @@ from fastapi import FastAPI
 
 from app.api.admin import router as admin_router
 from app.api.internal import router as internal_router
+from app.api.me import router as me_router
 
 app = FastAPI(title="memberapi.ose.tw", version="0.1.0")
 
@@ -13,3 +14,4 @@ def healthz() -> dict[str, str]:
 
 app.include_router(internal_router)
 app.include_router(admin_router)
+app.include_router(me_router)

app/schemas/auth.py

@@ -0,0 +1,14 @@
+from pydantic import BaseModel
+
+
+class AuthentikPrincipal(BaseModel):
+    sub: str
+    email: str | None = None
+    name: str | None = None
+    preferred_username: str | None = None
+
+
+class MeSummaryResponse(BaseModel):
+    sub: str
+    email: str | None = None
+    display_name: str | None = None

app/security/authentik_jwt.py

@@ -0,0 +1,84 @@
+from __future__ import annotations
+
+from functools import lru_cache
+
+import jwt
+from fastapi import Depends, HTTPException, status
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+
+from app.core.config import get_settings
+from app.schemas.auth import AuthentikPrincipal
+
+bearer_scheme = HTTPBearer(auto_error=False)
+
+
+class AuthentikTokenVerifier:
+    def __init__(self, issuer: str | None, jwks_url: str | None, audience: str | None) -> None:
+        self.issuer = issuer.strip() if issuer else None
+        self.jwks_url = jwks_url.strip() if jwks_url else self._infer_jwks_url(self.issuer)
+        self.audience = audience.strip() if audience else None
+
+        if not self.jwks_url:
+            raise ValueError("AUTHENTIK_JWKS_URL or AUTHENTIK_ISSUER is required")
+
+        self._jwk_client = jwt.PyJWKClient(self.jwks_url)
+
+    @staticmethod
+    def _infer_jwks_url(issuer: str | None) -> str | None:
+        if not issuer:
+            return None
+        normalized = issuer.rstrip("/") + "/"
+        if normalized.endswith("/jwks/"):
+            return normalized
+        return normalized + "jwks/"
+
+    def verify_access_token(self, token: str) -> AuthentikPrincipal:
+        try:
+            signing_key = self._jwk_client.get_signing_key_from_jwt(token)
+            options = {
+                "verify_signature": True,
+                "verify_exp": True,
+                "verify_aud": bool(self.audience),
+                "verify_iss": bool(self.issuer),
+            }
+            claims = jwt.decode(
+                token,
+                signing_key.key,
+                algorithms=["RS256", "RS384", "RS512"],
+                audience=self.audience,
+                issuer=self.issuer,
+                options=options,
+            )
+        except Exception as exc:
+            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="invalid_bearer_token") from exc
+
+        sub = claims.get("sub")
+        if not sub:
+            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="token_missing_sub")
+
+        return AuthentikPrincipal(
+            sub=sub,
+            email=claims.get("email"),
+            name=claims.get("name"),
+            preferred_username=claims.get("preferred_username"),
+        )
+
+
+@lru_cache
+def _get_verifier() -> AuthentikTokenVerifier:
+    settings = get_settings()
+    return AuthentikTokenVerifier(
+        issuer=settings.authentik_issuer,
+        jwks_url=settings.authentik_jwks_url,
+        audience=settings.authentik_audience,
+    )
+
+
+def require_authenticated_principal(
+    credentials: HTTPAuthorizationCredentials | None = Depends(bearer_scheme),
+) -> AuthentikPrincipal:
+    if credentials is None or credentials.scheme.lower() != "bearer":
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="missing_bearer_token")
+
+    verifier = _get_verifier()
+    return verifier.verify_access_token(credentials.credentials)

pyproject.toml

@@ -11,6 +11,7 @@ dependencies = [
   "pydantic-settings>=2.11.0",
   "python-dotenv>=1.1.1",
   "passlib[bcrypt]>=1.7.4",
+  "pyjwt[crypto]>=2.10.1",
 ]
 
 [project.optional-dependencies]

scripts/generate_api_key_hash.py

@@ -0,0 +1,29 @@
+#!/usr/bin/env python3
+import argparse
+import hashlib
+
+from passlib.context import CryptContext
+
+pwd_context = CryptContext(schemes=["argon2", "bcrypt"], deprecated="auto")
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Generate API key hash for api_clients table")
+    parser.add_argument("api_key", help="Plain API key")
+    parser.add_argument(
+        "--algo",
+        choices=["argon2", "bcrypt", "sha256"],
+        default="argon2",
+        help="Hash algorithm (default: argon2)",
+    )
+    args = parser.parse_args()
+
+    if args.algo == "sha256":
+        print("sha256:" + hashlib.sha256(args.api_key.encode("utf-8")).hexdigest())
+        return
+
+    print(pwd_context.hash(args.api_key, scheme=args.algo))
+
+
+if __name__ == "__main__":
+    main()

tests/test_authentik_jwt.py

@@ -0,0 +1,17 @@
+from fastapi.testclient import TestClient
+
+from app.main import app
+from app.security.authentik_jwt import AuthentikTokenVerifier
+
+
+def test_infer_jwks_url() -> None:
+    assert AuthentikTokenVerifier._infer_jwks_url("https://auth.ose.tw/application/o/member/") == (
+        "https://auth.ose.tw/application/o/member/jwks/"
+    )
+
+
+def test_me_requires_bearer_token() -> None:
+    client = TestClient(app)
+    resp = client.get("/me")
+    assert resp.status_code == 401
+    assert resp.json()["detail"] == "missing_bearer_token"