2026-07-22 21:21:05 +02:00
|
|
|
from __future__ import annotations
|
|
|
|
|
|
2026-07-22 21:48:09 +02:00
|
|
|
from collections.abc import Mapping
|
|
|
|
|
|
2026-07-22 21:21:05 +02:00
|
|
|
from fastapi import Request
|
|
|
|
|
|
|
|
|
|
from qonto_assistant.config import Settings
|
|
|
|
|
from qonto_assistant.contracts import ActorClaims
|
|
|
|
|
|
|
|
|
|
|
2026-07-22 21:48:09 +02:00
|
|
|
def actor_claims_from_headers(headers: Mapping[str, str], settings: Settings) -> ActorClaims:
|
|
|
|
|
"""Shared REST/MCP claims parsing so both transports enforce identical actor identity.
|
|
|
|
|
|
|
|
|
|
Real workload/OIDC auth for the MCP transport is QONTO-WP-0003-T03; until
|
|
|
|
|
then MCP callers use the same X-Actor-* header convention as REST.
|
|
|
|
|
"""
|
|
|
|
|
actor_id = headers.get("x-actor-id", "anonymous")
|
|
|
|
|
tenant_id = headers.get("x-tenant-id", settings.default_tenant_id)
|
|
|
|
|
lane = headers.get("x-actor-lane", settings.default_actor_lane)
|
|
|
|
|
raw_scopes = headers.get("x-actor-scopes", "")
|
2026-07-22 21:21:05 +02:00
|
|
|
scopes = frozenset(scope.strip() for scope in raw_scopes.split(",") if scope.strip())
|
|
|
|
|
return ActorClaims(actor_id=actor_id, tenant_id=tenant_id, lane=lane, scopes=scopes)
|
2026-07-22 21:48:09 +02:00
|
|
|
|
|
|
|
|
|
|
|
|
|
def actor_claims_from_request(request: Request, settings: Settings) -> ActorClaims:
|
|
|
|
|
return actor_claims_from_headers(request.headers, settings)
|