feat: add authentik admin user sync endpoint

README.md

@@ -29,6 +29,13 @@ python scripts/generate_api_key_hash.py 'YOUR_PLAIN_KEY'
 - Optional:
   - `AUTHENTIK_AUDIENCE` (enables audience claim validation)
 
+## Authentik Admin API setup
+
+- Required for `/internal/authentik/users/ensure`:
+  - `AUTHENTIK_BASE_URL`
+  - `AUTHENTIK_ADMIN_TOKEN`
+  - `AUTHENTIK_VERIFY_TLS`
+
 ## Main APIs
 
 - `GET /healthz`
@@ -36,5 +43,6 @@ python scripts/generate_api_key_hash.py 'YOUR_PLAIN_KEY'
 - `GET /me/permissions/snapshot` (Bearer token required)
 - `POST /internal/users/upsert-by-sub`
 - `GET /internal/permissions/{authentik_sub}/snapshot`
+- `POST /internal/authentik/users/ensure`
 - `POST /admin/permissions/grant`
 - `POST /admin/permissions/revoke`

app/api/internal.py

@@ -5,8 +5,10 @@ from app.core.config import get_settings
 from app.db.session import get_db
 from app.repositories.permissions_repo import PermissionsRepository
 from app.repositories.users_repo import UsersRepository
+from app.schemas.authentik_admin import AuthentikEnsureUserRequest, AuthentikEnsureUserResponse
 from app.schemas.permissions import PermissionSnapshotResponse
 from app.schemas.users import UserUpsertBySubRequest
+from app.services.authentik_admin_service import AuthentikAdminService
 from app.services.permission_service import PermissionService
 
 router = APIRouter(prefix="/internal", tags=["internal"])
@@ -36,6 +38,7 @@ def upsert_user_by_sub(
     return {
         "id": user.id,
         "sub": user.authentik_sub,
+        "authentik_user_id": user.authentik_user_id,
         "email": user.email,
         "display_name": user.display_name,
         "is_active": user.is_active,
@@ -58,3 +61,29 @@ def get_permission_snapshot(
     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=authentik_sub, permissions=tuples)
+
+
+@router.post("/authentik/users/ensure", response_model=AuthentikEnsureUserResponse)
+def ensure_authentik_user(
+    payload: AuthentikEnsureUserRequest,
+    _: None = Depends(verify_internal_secret),
+    db: Session = Depends(get_db),
+) -> AuthentikEnsureUserResponse:
+    settings = get_settings()
+    authentik_service = AuthentikAdminService(settings=settings)
+    sync_result = authentik_service.ensure_user(
+        sub=payload.sub,
+        email=payload.email,
+        display_name=payload.display_name,
+        is_active=payload.is_active,
+    )
+
+    users_repo = UsersRepository(db)
+    users_repo.upsert_by_sub(
+        authentik_sub=payload.sub,
+        email=payload.email,
+        display_name=payload.display_name,
+        is_active=payload.is_active,
+        authentik_user_id=sync_result.user_id,
+    )
+    return AuthentikEnsureUserResponse(authentik_user_id=sync_result.user_id, action=sync_result.action)

app/models/user.py

@@ -1,7 +1,7 @@
 from datetime import datetime
 from uuid import uuid4
 
-from sqlalchemy import Boolean, DateTime, String, func
+from sqlalchemy import Boolean, DateTime, Integer, String, func
 from sqlalchemy.dialects.postgresql import UUID
 from sqlalchemy.orm import Mapped, mapped_column
 
@@ -13,6 +13,7 @@ class User(Base):
 
     id: Mapped[str] = mapped_column(UUID(as_uuid=False), primary_key=True, default=lambda: str(uuid4()))
     authentik_sub: Mapped[str] = mapped_column(String(255), unique=True, nullable=False, index=True)
+    authentik_user_id: Mapped[int | None] = mapped_column(Integer)
     email: Mapped[str | None] = mapped_column(String(320))
     display_name: Mapped[str | None] = mapped_column(String(255))
     is_active: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)

app/repositories/users_repo.py

@@ -18,17 +18,21 @@ class UsersRepository:
         email: str | None,
         display_name: str | None,
         is_active: bool,
+        authentik_user_id: int | None = None,
     ) -> User:
         user = self.get_by_sub(authentik_sub)
         if user is None:
             user = User(
                 authentik_sub=authentik_sub,
+                authentik_user_id=authentik_user_id,
                 email=email,
                 display_name=display_name,
                 is_active=is_active,
             )
             self.db.add(user)
         else:
+            if authentik_user_id is not None:
+                user.authentik_user_id = authentik_user_id
             user.email = email
             user.display_name = display_name
             user.is_active = is_active

app/schemas/authentik_admin.py

@@ -0,0 +1,13 @@
+from pydantic import BaseModel
+
+
+class AuthentikEnsureUserRequest(BaseModel):
+    sub: str
+    email: str
+    display_name: str | None = None
+    is_active: bool = True
+
+
+class AuthentikEnsureUserResponse(BaseModel):
+    authentik_user_id: int
+    action: str

app/services/authentik_admin_service.py

@@ -0,0 +1,75 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+import httpx
+from fastapi import HTTPException, status
+
+from app.core.config import Settings
+
+
+@dataclass
+class AuthentikSyncResult:
+    user_id: int
+    action: str
+
+
+class AuthentikAdminService:
+    def __init__(self, settings: Settings) -> None:
+        self.base_url = settings.authentik_base_url.rstrip("/")
+        self.admin_token = settings.authentik_admin_token
+        self.verify_tls = settings.authentik_verify_tls
+
+        if not self.base_url or not self.admin_token:
+            raise HTTPException(
+                status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+                detail="authentik_admin_not_configured",
+            )
+
+    def _client(self) -> httpx.Client:
+        return httpx.Client(
+            base_url=self.base_url,
+            headers={
+                "Authorization": f"Bearer {self.admin_token}",
+                "Accept": "application/json",
+                "Content-Type": "application/json",
+            },
+            timeout=10,
+            verify=self.verify_tls,
+        )
+
+    @staticmethod
+    def _safe_username(sub: str, email: str) -> str:
+        if email and "@" in email:
+            return email.split("@", 1)[0]
+        return sub.replace("|", "_")[:150]
+
+    def ensure_user(self, sub: str, email: str, display_name: str | None, is_active: bool = True) -> AuthentikSyncResult:
+        payload = {
+            "username": self._safe_username(sub=sub, email=email),
+            "name": display_name or email,
+            "email": email,
+            "is_active": is_active,
+        }
+
+        with self._client() as client:
+            resp = client.get("/api/v3/core/users/", params={"email": email})
+            if resp.status_code >= 400:
+                raise HTTPException(status_code=502, detail="authentik_lookup_failed")
+
+            data = resp.json()
+            results = data.get("results") if isinstance(data, dict) else None
+            existing = results[0] if isinstance(results, list) and results else None
+
+            if existing and existing.get("pk") is not None:
+                user_pk = int(existing["pk"])
+                patch_resp = client.patch(f"/api/v3/core/users/{user_pk}/", json=payload)
+                if patch_resp.status_code >= 400:
+                    raise HTTPException(status_code=502, detail="authentik_update_failed")
+                return AuthentikSyncResult(user_id=user_pk, action="updated")
+
+            create_resp = client.post("/api/v3/core/users/", json=payload)
+            if create_resp.status_code >= 400:
+                raise HTTPException(status_code=502, detail="authentik_create_failed")
+            created = create_resp.json()
+            return AuthentikSyncResult(user_id=int(created["pk"]), action="created")

pyproject.toml

@@ -12,12 +12,12 @@ dependencies = [
   "python-dotenv>=1.1.1",
   "passlib[bcrypt]>=1.7.4",
   "pyjwt[crypto]>=2.10.1",
+  "httpx>=0.28.1",
 ]
 
 [project.optional-dependencies]
 dev = [
   "pytest>=8.4.2",
-  "httpx>=0.28.1",
   "ruff>=0.13.0",
 ]
 

scripts/init_schema.sql

@@ -5,6 +5,7 @@ CREATE EXTENSION IF NOT EXISTS pgcrypto;
 CREATE TABLE IF NOT EXISTS users (
   id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
   authentik_sub VARCHAR(255) NOT NULL UNIQUE,
+  authentik_user_id INTEGER,
   email VARCHAR(320),
   display_name VARCHAR(255),
   is_active BOOLEAN NOT NULL DEFAULT TRUE,

scripts/migrate_add_authentik_user_id.sql

@@ -0,0 +1,2 @@
+ALTER TABLE users
+  ADD COLUMN IF NOT EXISTS authentik_user_id INTEGER;

tests/test_internal_authentik_sync.py

@@ -0,0 +1,19 @@
+from fastapi.testclient import TestClient
+
+from app.main import app
+
+
+def test_internal_authentik_ensure_requires_config() -> None:
+    client = TestClient(app)
+    resp = client.post(
+        "/internal/authentik/users/ensure",
+        headers={"X-Internal-Secret": "CHANGE_ME"},
+        json={
+            "sub": "authentik-sub-1",
+            "email": "user@example.com",
+            "display_name": "User Example",
+            "is_active": True,
+        },
+    )
+    assert resp.status_code == 503
+    assert resp.json()["detail"] == "authentik_admin_not_configured"