def mount_privacy_routes(router, *, privacy, resolver, admin_target_authorizer=None):
"""Mount privacy routes; a cross-tenant export needs an explicit grant.
The injected authorizer receives the verified caller and root target scope;
only a literal True grants access. It may be synchronous or asynchronous.
"""
from fastapi import Header, HTTPException, Request
from symfonic.kernel.contracts.tenancy import SubjectScope, SubjectScopeError
from symfonic.platform.audit import AuditSinkError
from symfonic.platform.errors import (
AuthenticationError,
AuthorizationError,
DestructiveConfirmationError,
)
from symfonic.platform.values import AuthenticatedPrincipal, RequestCredentials
from symfonic.services.privacy.errors import ReadSuppressedError, SubjectErasedError
async def principal(tenant, authorization, request):
if resolver is None or not tenant or not tenant.strip():
raise HTTPException(401, "privacy requires a verified identity")
headers = dict(request.headers)
headers["x-tenant-id"] = tenant.strip()
if authorization:
headers["authorization"] = authorization
from symfonic.platform.transport_production_gate import gate_native_http_assembly
gate_native_http_assembly(resolver)
try:
caller = resolver.resolve(RequestCredentials(
headers=headers,
peer=request.client.host if request.client else None,
user_agent=request.headers.get("user-agent"),
route=request.url.path,
))
if inspect.isawaitable(caller):
caller = await caller
except AuthenticationError as failed:
raise HTTPException(401, "privacy authentication failed") from failed
except AuthorizationError as failed:
raise HTTPException(403, "privacy scope denied") from failed
except Exception as failed:
raise HTTPException(503, "privacy identity unavailable") from failed
if not isinstance(caller, AuthenticatedPrincipal):
raise HTTPException(401, "privacy requires a verified identity")
if caller.derivation.get("verified") is not True:
raise HTTPException(401, "privacy requires a verified identity")
if caller.scope.tenant_id != tenant.strip():
raise HTTPException(403, "privacy scope denied")
return caller
async def export_principal(caller, target_tenant_id):
if target_tenant_id is None:
return caller
target_name = target_tenant_id.strip()
if not target_name:
raise HTTPException(400, "invalid export target")
try:
target = SubjectScope(tenant_id=target_name)
except SubjectScopeError as failed:
raise HTTPException(400, "invalid export target") from failed
if target == caller.scope:
return caller
if caller.is_admin is not True or admin_target_authorizer is None:
raise HTTPException(403, "privacy scope denied")
try:
granted = admin_target_authorizer(caller, target)
if inspect.isawaitable(granted):
granted = await granted
except AuthorizationError as failed:
raise HTTPException(403, "privacy scope denied") from failed
except Exception as failed:
raise HTTPException(503, "privacy authorization unavailable") from failed
if granted is not True:
raise HTTPException(403, "privacy scope denied")
return AuthenticatedPrincipal(
principal_id=caller.principal_id, scope=target,
is_admin=caller.is_admin, derivation=caller.derivation,
)
async def invoke(caller, *, confirmation=None):
try:
if confirmation is None:
return await privacy.export(caller)
outcome = await privacy.erase(caller, confirmation=confirmation)
# Saga diagnostics stay internal; partial results can contain raw
# participant exceptions even though no exception escaped erase().
fields = ("scope_key", "complete", "counts", "unconfirmed", "generation", "erased_at")
return {field: outcome[field] for field in fields}
except DestructiveConfirmationError as failed:
raise HTTPException(400, "confirmation must be DELETE-<resolved tenant_id>") from failed
except SubjectErasedError as failed:
raise HTTPException(410, "subject data erased") from failed
except ReadSuppressedError as failed:
raise HTTPException(409, "subject data temporarily unavailable") from failed
except AuditSinkError as failed:
raise HTTPException(503, "privacy audit unavailable") from failed
except Exception as failed:
raise HTTPException(503, "privacy operation unavailable") from failed
@router.get("/tenants/me/export")
async def export(
request: Request,
x_tenant_id: str | None = Header(default=None), # noqa: B008
authorization: str | None = Header(default=None), # noqa: B008
target_tenant_id: str | None = None,
):
caller = await principal(x_tenant_id, authorization, request)
return await invoke(await export_principal(caller, target_tenant_id))
@router.delete("/tenants/me/data")
async def erase(
request: Request,
confirmation: str = "",
x_tenant_id: str | None = Header(default=None), # noqa: B008
authorization: str | None = Header(default=None), # noqa: B008
):
return await invoke(
await principal(x_tenant_id, authorization, request), confirmation=confirmation,
)