app/security/context.py

from __future__ import annotations

import contextvars
from dataclasses import dataclass
from datetime import datetime


@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
    expires_at: datetime | 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

    def allows(self, required_scope: str) -> bool:
        if required_scope in self.scopes or "admin" in self.scopes:
            return True
        return False


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
)