feat: add authentik jwt verification and me endpoints
This commit is contained in:
46
backend/app/api/me.py
Normal file
46
backend/app/api/me.py
Normal file
@@ -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)
|
||||
Reference in New Issue
Block a user