from __future__ import annotations import contextvars from dataclasses import dataclass from datetime import datetime from app.security.scopes import membership_role_scopes @dataclass(frozen=True, slots=True) class AuthContext: api_key_id: str key_name: str key_prefix: str environment: str role: str | None scopes: frozenset[str] requests_per_minute: int concurrent_jobs: int uploads_per_hour: int processing_bytes_per_day: int # API keys may be configured without an expiry; callers that omit the # field therefore receive the same explicit non-expiring semantics. expires_at: datetime | None = None # Resolved exclusively by the server from APIKeyPrincipal. These are not # accepted from REST, MCP, SDK, or n8n request payloads. workspace_id: str | None = None user_id: str | None = None membership_id: str | None = None membership_role: str | None = None membership_scopes: frozenset[str] | None = None def allows(self, required_scope: str) -> bool: key_allows = required_scope in self.scopes or "admin" in self.scopes if not key_allows: return False # API-key scopes are a credential ceiling; workspace membership is an # independent tenant ceiling. A powerful key must never elevate a # viewer/member beyond the role granted by the workspace owner. if self.membership_role is None: return True membership_scopes = self.membership_scopes if membership_scopes is None: membership_scopes = membership_role_scopes(self.membership_role) return required_scope in membership_scopes or "admin" in membership_scopes auth_context: contextvars.ContextVar[AuthContext | None] = contextvars.ContextVar( "auth_context", default=None ) # HTTP authentication already applies request and concurrency limits before the # MCP ASGI application runs. Standalone stdio MCP calls use this marker to apply # the same limiter exactly once inside the transport-neutral registry. http_auth_applied: contextvars.ContextVar[bool] = contextvars.ContextVar( "http_auth_applied", default=False )