Lars
5dc372c043
Some checks failed
Deploy Development / deploy (push) Failing after 42s
Test Suite / pytest-backend (push) Failing after 1s
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 3s
Test Suite / compose-smoke (push) Has been skipped
Agenten authentifizieren per X-Actor-Token; /api/operational/ Fassade mit context, next-action, status, evidence; Token-Verwaltung unter /api/actors/{id}/service-tokens.
Co-authored-by: Cursor <cursoragent@cursor.com>
93 lines
3.2 KiB
Python
93 lines
3.2 KiB
Python
"""Operational API auth — Session or Actor Service Token (AP1.7c)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Callable, Optional
|
|
|
|
from fastapi import Depends, Header, HTTPException
|
|
|
|
from auth import AUTH_HEADER, get_session
|
|
from capabilities import _enforce_capability
|
|
from services.actor_service_tokens import validate_service_token
|
|
from tenant_context import TenantContext, resolve_tenant_context
|
|
|
|
ACTOR_TOKEN_HEADER = "X-Actor-Token"
|
|
|
|
|
|
def _extract_actor_token(
|
|
x_actor_token: Optional[str],
|
|
authorization: Optional[str],
|
|
) -> Optional[str]:
|
|
if x_actor_token and x_actor_token.strip():
|
|
return x_actor_token.strip()
|
|
if authorization:
|
|
scheme, _, value = authorization.partition(" ")
|
|
if scheme.lower() == "bearer" and value.strip():
|
|
return value.strip()
|
|
return None
|
|
|
|
|
|
def resolve_service_token_context(raw_token: str) -> TenantContext:
|
|
resolved = validate_service_token(raw_token)
|
|
if not resolved:
|
|
raise HTTPException(status_code=401, detail="Ungültiger oder abgelaufener Actor-Token")
|
|
|
|
return TenantContext(
|
|
user_id="",
|
|
email="",
|
|
display_name=resolved["actor_name"],
|
|
portal_role="",
|
|
tenant_id=resolved["tenant_id"],
|
|
tenant_slug=resolved["tenant_slug"],
|
|
tenant_name=resolved["tenant_name"],
|
|
tenant_role=None,
|
|
actor_id=resolved["actor_id"],
|
|
actor_type=resolved["actor_type"],
|
|
session_token="",
|
|
capabilities=resolved["capabilities"],
|
|
auth_source="service_token",
|
|
service_token_id=resolved["token_id"],
|
|
)
|
|
|
|
|
|
def get_operational_context(
|
|
x_actor_token: Optional[str] = Header(default=None, alias=ACTOR_TOKEN_HEADER),
|
|
authorization: Optional[str] = Header(default=None),
|
|
x_auth_token: Optional[str] = Header(default=None, alias=AUTH_HEADER),
|
|
) -> TenantContext:
|
|
actor_token = _extract_actor_token(x_actor_token, authorization)
|
|
if actor_token:
|
|
return resolve_service_token_context(actor_token)
|
|
|
|
session_token = (x_auth_token or "").strip()
|
|
if not session_token:
|
|
raise HTTPException(
|
|
status_code=401,
|
|
detail="Authentifizierung erforderlich (X-Actor-Token oder X-Auth-Token)",
|
|
)
|
|
session = get_session(session_token)
|
|
if not session:
|
|
raise HTTPException(status_code=401, detail="Ungültige Session")
|
|
ctx = resolve_tenant_context(session)
|
|
if not ctx.tenant_id:
|
|
raise HTTPException(status_code=403, detail="Kein aktiver Tenant — Tenant wählen")
|
|
return ctx
|
|
|
|
|
|
def require_operational_capability(capability_key: str) -> Callable[..., TenantContext]:
|
|
"""Service tokens: always enforce. Sessions: probe/enforce per CAPABILITY_ENFORCE."""
|
|
|
|
def _dependency(
|
|
ctx: TenantContext = Depends(get_operational_context),
|
|
) -> TenantContext:
|
|
if ctx.auth_source == "service_token":
|
|
if capability_key not in ctx.capabilities:
|
|
raise HTTPException(
|
|
status_code=403,
|
|
detail=f"Capability fehlt auf Service Token: {capability_key}",
|
|
)
|
|
return ctx
|
|
return _enforce_capability(ctx, capability_key)
|
|
|
|
return _dependency
|