Lars
910b1d7e07
Some checks failed
Deploy Development / deploy (push) Successful in 33s
Test Suite / pytest-backend (push) Failing after 4s
Test Suite / k6 /api/health Baseline (push) Has been skipped
Test Suite / playwright-smoke (push) Has been skipped
Test Suite / lint-backend (push) Successful in 1s
Test Suite / compose-smoke (push) Has been skipped
Co-authored-by: Cursor <cursoragent@cursor.com>
68 lines
2.2 KiB
Python
68 lines
2.2 KiB
Python
"""Current user, tenant context and entitlements endpoints."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from auth import get_session, require_auth, set_session_active_tenant
|
|
from capabilities import require_capability
|
|
from entitlements import build_entitlements_snapshot
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from pydantic import BaseModel
|
|
from tenant_context import TenantContext, get_tenant_context, list_user_tenants, require_tenant_context, resolve_tenant_context
|
|
|
|
router = APIRouter(prefix="/api/me", tags=["me"])
|
|
|
|
|
|
class SwitchTenantRequest(BaseModel):
|
|
tenant_id: str
|
|
|
|
|
|
@router.get("")
|
|
def get_me(session: dict = Depends(require_auth)):
|
|
tenants = list_user_tenants(str(session["user_id"]))
|
|
return {
|
|
"id": str(session["user_id"]),
|
|
"email": session["email"],
|
|
"display_name": session["display_name"],
|
|
"portal_role": session["portal_role"],
|
|
"active_tenant_id": str(session["active_tenant_id"]) if session.get("active_tenant_id") else None,
|
|
"tenants": tenants,
|
|
}
|
|
|
|
|
|
@router.get("/context")
|
|
def get_my_context(ctx: TenantContext = Depends(get_tenant_context)):
|
|
return ctx.to_dict()
|
|
|
|
|
|
@router.get("/context/required")
|
|
def get_my_context_with_tenant(ctx: TenantContext = Depends(require_tenant_context)):
|
|
"""Endpoint requiring an active tenant."""
|
|
return ctx.to_dict()
|
|
|
|
|
|
@router.get("/entitlements")
|
|
def get_my_entitlements(ctx: TenantContext = Depends(get_tenant_context)):
|
|
return build_entitlements_snapshot(ctx)
|
|
|
|
|
|
@router.get("/admin/demo")
|
|
def admin_demo(ctx: TenantContext = Depends(require_capability("kairo.admin.access"))):
|
|
"""Example endpoint protected by require_capability."""
|
|
return {
|
|
"ok": True,
|
|
"message": "Portal-Admin Capability bestätigt",
|
|
"user_id": ctx.user_id,
|
|
"tenant_id": ctx.tenant_id,
|
|
}
|
|
|
|
|
|
@router.post("/tenant")
|
|
def switch_tenant(body: SwitchTenantRequest, session: dict = Depends(require_auth)):
|
|
ok = set_session_active_tenant(session["token"], body.tenant_id)
|
|
if not ok:
|
|
raise HTTPException(status_code=403, detail="Tenant nicht erlaubt")
|
|
fresh = get_session(session["token"])
|
|
if not fresh:
|
|
raise HTTPException(status_code=401, detail="Session ungültig")
|
|
return resolve_tenant_context(fresh).to_dict()
|