# pyright: reportAny=false, reportExplicitAny=false, reportUnusedParameter=false

"""API middleware for auth, error handling, CORS, and CQRS dispatch helpers.

Provides:
- AuthEnforcementMiddleware: shared HTTP/WebSocket auth enforcement
- create_auth_enforced_app: reusable auth wrapper for arbitrary ASGI apps
- error_handler: Converts domain exceptions to JSON error envelopes
- dispatch_query: Async wrapper for query bus calls via run_in_threadpool
- dispatch_command: Async wrapper for command bus calls via run_in_threadpool
- get_cors_config: Validated CORS configuration from environment
"""

import logging
import os
import re
import math
from typing import Any
from urllib.parse import parse_qs
from urllib.parse import urlparse

from starlette.concurrency import run_in_threadpool
from starlette.datastructures import Headers, State
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
from starlette.requests import Request
from starlette.responses import JSONResponse, Response
from starlette.types import ASGIApp, Receive, Scope, Send

from ...domain.contracts.authentication import AuthRequest
from ...domain.contracts.authorization import GrantScope
from ...application.ports.bus import HandlerNotRegisteredError
from ...domain.exceptions import (
    AccessDeniedError,
    AuthenticationError,
    AuthorizationError,
    MCPError,
    McpServerDegradedError,
    McpServerNotFoundError,
    ConfigurationRestartRequiredError,
    ConfigurationUnavailableError,
    McpServerNotHereError,
    McpServerNotReadyError,
    MissingCredentialsError,
    RateLimitExceeded,
    RateLimitExceededError,
    ToolNotFoundError,
    ToolTimeoutError,
    ValidationError,
)
from ...infrastructure.identity.trusted_proxy import TrustedProxyResolver, headers_from_asgi_scope, resolve_source_ip
from ..context import get_context
from .route_permissions import resolve_rule
from .serializers import HangarJSONResponse
from .tenant_scope import record_grant_scope

logger = logging.getLogger(__name__)

_ALLOWED_METHODS = ["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"]
_ALLOWED_HEADERS = [
    "Authorization",
    "Content-Type",
    "X-API-Key",
    "X-Correlation-ID",
    "X-Requested-With",
]
_ORIGIN_RE = re.compile(r"^https?://[^*\s]+$")
_CSRF_PROTECTED_METHODS = frozenset({"POST", "PUT", "PATCH", "DELETE"})
_CSRF_BYPASS_AUTH_SCHEME = "bearer "
_BROWSER_HINT_HEADERS = ("origin", "referer", "cookie")
_SESSION_SUSPEND_PATH_RE = re.compile(r"^/sessions/(?P<session_id>[^/]+)/suspend/?$")
_VALID_SESSION_ID_RE = re.compile(r"^[a-zA-Z0-9_-]{1,128}$")
_DEFAULT_AUTH_SKIP_PATHS = frozenset(
    {
        "/health/live",
        "/health/ready",
        "/health/startup",
        "/metrics",
        # RFC 9728 discovery — must be reachable before the client has a token.
        "/.well-known/oauth-protected-resource",
    }
)


class _AuthLoggerAdapter:
    @staticmethod
    def warning(event: str, **kwargs: Any) -> None:
        logger.warning("%s %s", event, kwargs)


# Mapping of exception types to HTTP status codes.
# More specific types must come before their base classes.
_EXCEPTION_STATUS_MAP: list[tuple[type, int]] = [
    # Ahead of the domain errors: the caller asked for something this process
    # does not serve, which is a statement about the deployment rather than
    # about the request. `GET /api/auth/keys` with auth disabled reached a
    # mounted route whose handlers are registered only when auth is enabled,
    # and answered 500 -- telling the caller the server had broken when the
    # feature was simply off.
    (HandlerNotRegisteredError, 503),
    # ONLY the narrow "the filesystem said no" case is a 503: the rotating
    # backup could not be written because the config directory is not writable
    # by the gateway process. That is a genuine transient/unavailable condition
    # the caller can retry once the environment is fixed. A *generic*
    # ConfigurationError is NOT mapped here: it is an operator-input problem
    # (a bad capabilities block, the reload fault-barrier wrapping an internal
    # failure) and must fall through to 500 -- mapping it to 503 turned every
    # such error into a faked retryable outage and leaked the wrapped internal
    # text to the caller (#823 regression). Being a subclass of
    # ConfigurationError, this entry is matched before the base MCPError->500
    # below, so the specific case wins and the generic case does not.
    (ConfigurationUnavailableError, 503),
    # A reload that would change what only a restart can (#1424): nothing is
    # wrong with the file and nothing was changed, so neither 500 nor 503.
    (ConfigurationRestartRequiredError, 409),
    (McpServerNotFoundError, 404),
    (ToolNotFoundError, 404),
    # Not "this broke" but "not here": a follower asked to start a server that
    # belongs to the lease holder. It travelled as a 500 until a two-replica
    # deployment showed a correct refusal reported as a server fault.
    (McpServerNotHereError, 409),
    (McpServerNotReadyError, 409),
    (ValidationError, 422),
    (RateLimitExceededError, 429),
    (RateLimitExceeded, 429),
    (AuthenticationError, 401),
    (AccessDeniedError, 403),
    (AuthorizationError, 403),
    (McpServerDegradedError, 503),
    (ToolTimeoutError, 504),
    (MCPError, 500),
]


class CSRFMiddleware(BaseHTTPMiddleware):
    """Require X-Requested-With on mutating browser-style requests."""

    async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
        method = request.method.upper()
        headers = Headers(scope=request.scope)
        if (
            method not in _CSRF_PROTECTED_METHODS
            or not self._is_browser_csrf_path(request.url.path)
            or not self._is_browser_request(headers)
            or self._should_skip_csrf(headers)
            or headers.get("x-requested-with", "").strip()
        ):
            return await call_next(request)

        return HangarJSONResponse(
            {
                "error": "csrf_header_required",
                "message": "X-Requested-With header is required for mutating requests",
            },
            status_code=403,
        )

    @staticmethod
    def _should_skip_csrf(headers: Headers) -> bool:
        if headers.get("x-api-key", "").strip():
            return True

        authorization = headers.get("authorization", "")
        return authorization.lower().startswith(_CSRF_BYPASS_AUTH_SCHEME)

    @staticmethod
    def _is_browser_request(headers: Headers) -> bool:
        return any(headers.get(header, "").strip() for header in _BROWSER_HINT_HEADERS)

    @staticmethod
    def _is_browser_csrf_path(path: str) -> bool:
        match = _SESSION_SUSPEND_PATH_RE.match(path)
        if match is None:
            return False

        return _VALID_SESSION_ID_RE.match(match.group("session_id")) is not None


def _should_skip_auth_path(path: str, skip_paths: frozenset[str]) -> bool:
    return path in skip_paths or path.startswith("/health/")


def _headers_from_scope(scope: Scope) -> dict[str, str]:
    headers = headers_from_asgi_scope(scope.get("headers"))
    if scope["type"] == "websocket":
        raw_query_string = scope.get("query_string", b"")
        query_params = parse_qs(raw_query_string.decode("latin-1"))
        token = query_params.get("token", [None])[0]
        if token and "authorization" not in headers and "x-api-key" not in headers:
            headers["authorization"] = token if token.lower().startswith("bearer ") else f"Bearer {token}"
    return headers


def _build_auth_request(scope: Scope, trusted_proxies: TrustedProxyResolver) -> tuple[AuthRequest, str]:
    headers = _headers_from_scope(scope)
    client = scope.get("client")
    client_host = client[0] if client else None
    source_ip = (
        resolve_source_ip(headers=headers, client_host=client_host, trusted_proxies=trusted_proxies) or "unknown"
    )
    method = scope.get("method", "GET" if scope["type"] == "websocket" else "")
    path = scope.get("path", "")
    return AuthRequest(headers=headers, source_ip=source_ip, method=method, path=path), source_ip


def _store_auth_context(scope: Scope, auth_context: Any) -> None:
    """Store auth context as Starlette State.auth on the ASGI scope.

    Uses starlette.datastructures.State so downstream Starlette handlers can
    read it via request.state.auth. A bare dict would not work -- Starlette's
    Request.state casts scope["state"] as State, and dict lacks __getattr__.
    """
    state = scope.get("state")
    if not isinstance(state, State):
        state = State()
        scope["state"] = state
    state.auth = auth_context


async def _send_auth_failure(
    scope: Scope,
    receive: Receive,
    send: Send,
    exc: AuthenticationError | AccessDeniedError,
    source_ip: str,
    www_authenticate: str = "Bearer, ApiKey",
) -> None:
    path = scope.get("path", "")
    if scope["type"] == "websocket":
        event_name = "ws_authentication_failed" if isinstance(exc, AuthenticationError) else "ws_access_denied"
        message = exc.message if isinstance(exc, AuthenticationError) else str(exc)
        _AuthLoggerAdapter.warning(event_name, path=path, source_ip=source_ip, message=message)
        await send({"type": "websocket.close", "code": 1008, "reason": message})
        return

    if isinstance(exc, AuthenticationError):
        response = JSONResponse(
            status_code=401,
            content={
                "error": "authentication_failed",
                "message": exc.message,
                "details": exc.details,
            },
            headers={"WWW-Authenticate": www_authenticate},
        )
    else:
        response = JSONResponse(
            status_code=403,
            content={
                "error": "access_denied",
                "message": str(exc),
                "principal_id": exc.principal_id,
                "action": exc.action,
                "resource": exc.resource,
            },
        )

    await response(scope, receive, send)


class AuthEnforcementMiddleware:
    """Shared HTTP/WebSocket auth enforcement middleware."""

    app: ASGIApp
    _authn: Any
    _skip_paths: frozenset[str]
    _trusted_proxies: TrustedProxyResolver
    _oidc_issuers: list[str]
    _oidc_resource_uri: str

    def __init__(
        self,
        app: ASGIApp,
        authn: Any,
        skip_paths: frozenset[str] | None = None,
        trusted_proxies: TrustedProxyResolver | None = None,
        oidc_issuers: list[str] | None = None,
        oidc_resource_uri: str = "",
    ) -> None:
        self.app = app
        self._authn = authn
        self._skip_paths = skip_paths or _DEFAULT_AUTH_SKIP_PATHS
        self._trusted_proxies = trusted_proxies or TrustedProxyResolver()
        self._oidc_issuers = oidc_issuers if oidc_issuers is not None else []
        self._oidc_resource_uri = oidc_resource_uri

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] not in ("http", "websocket"):
            await self.app(scope, receive, send)
            return

        path = scope.get("path", "")
        if _should_skip_auth_path(path, self._skip_paths):
            await self.app(scope, receive, send)
            return

        # CORS preflight carries no credentials by design (the same contract
        # the authorization chokepoint below already honors). Authenticating
        # OPTIONS made every browser preflight 401 before the CORS layer could
        # answer, with no Access-Control-Allow-Origin on the refusal -- so a
        # browser OAuth client could not call /mcp or /api at all (#993).
        # OPTIONS is a safe method: nothing behind this skip mutates on it.
        if scope["type"] == "http" and str(scope.get("method", "")).upper() == "OPTIONS":
            await self.app(scope, receive, send)
            return

        auth_request, source_ip = _build_auth_request(scope, self._trusted_proxies)
        try:
            auth_context = self._authn.authenticate(auth_request)
            _store_auth_context(scope, auth_context)
            await self.app(scope, receive, send)
        except (AuthenticationError, AccessDeniedError) as exc:
            # RFC 9728: advertise resource_metadata in Bearer challenge when OIDC active.
            if self._oidc_issuers and isinstance(exc, AuthenticationError):
                from ...auth.prm import build_resource_base_url, build_www_authenticate

                resource_base = self._oidc_resource_uri or build_resource_base_url(scope)
                www_auth = build_www_authenticate(resource_base)
            else:
                www_auth = "Bearer, ApiKey"
            await _send_auth_failure(scope, receive, send, exc, source_ip, www_authenticate=www_auth)


async def _send_authz_failure(
    scope: Scope,
    receive: Receive,
    send: Send,
    exc: AuthenticationError | AccessDeniedError,
) -> None:
    """Reject a request from the authorization chokepoint, in the API's envelope.

    The REST API's error shape is the one ``error_handler`` produces for every
    other domain exception::

        {"error": {"code": "AccessDeniedError", "message": ..., "details": ...}}

    Before the route-driven chokepoint existed, an authorization failure was an
    ``AccessDeniedError`` raised inside a handler, so it reached that handler and
    got that shape. Denying in middleware means the exception never reaches it,
    and reusing ``_send_auth_failure`` -- written for the authentication layer,
    which has always emitted a flatter body -- silently changed the 403 contract
    for every REST client. A nightly live test caught it: it asserts
    ``error["code"] == "AccessDeniedError"`` and got the string ``"access_denied"``.

    WebSocket rejections still go through ``_send_auth_failure``: there is no
    body to shape, only a close frame.
    """
    if scope["type"] == "websocket":
        await _send_auth_failure(scope, receive, send, exc, "unknown")
        return

    status_code = _get_status_code(exc)
    body: dict[str, Any] = {
        "error": {
            "code": type(exc).__name__,
            "message": exc.message if isinstance(exc, MCPError) else str(exc),
            "details": (exc.details or None) if isinstance(exc, MCPError) else None,
        }
    }
    headers = {"WWW-Authenticate": "Bearer, ApiKey"} if isinstance(exc, AuthenticationError) else None
    await JSONResponse(status_code=status_code, content=body, headers=headers)(scope, receive, send)


class AuthorizationEnforcementMiddleware:
    """Route-driven authorization for the REST/WebSocket API.

    Runs innermost -- after authentication has attached the principal to the
    scope -- and resolves the required permission from
    :mod:`.route_permissions` rather than from the handler. A route absent from
    that table is DENIED: adding an endpoint without deciding who may call it
    now fails closed instead of shipping it public.

    Auth off must mean off. ``NullAuthComponents`` reports ``enabled=False``
    while still exposing an ``authz_middleware``, so testing only ``is None``
    would arm the guard with nobody able to satisfy it -- the failure #590/#600
    fixed on the invoke and REST paths respectively. Both conditions are checked
    here for the same reason.

    Critically, the components are taken from the SAME object the router was
    built with rather than re-read from the application context. Those two can
    disagree -- a router built without auth mounts no authentication middleware,
    so no principal is ever attached -- and an authorizer armed against an
    unauthenticated app answers 401 to every caller with no credential that
    could ever satisfy it. Binding both to one object makes the invariant
    structural: authorization is armed if and only if authentication is mounted.

    A grant held only at tenant scope passes a rule only when the rule is
    ``tenant_aware``; everywhere else it is refused, because the permission was
    checked against every resource and the grant covers one tenant's. The
    reach of the grant that did pass is recorded on the scope
    (:func:`.tenant_scope.record_grant_scope`) so a tenant-aware handler confines
    itself by the same decision rather than by a second one.
    """

    app: ASGIApp
    _authz: Any
    _skip_paths: frozenset[str]

    def __init__(
        self,
        app: ASGIApp,
        auth_components: Any = None,
        skip_paths: frozenset[str] | None = None,
    ) -> None:
        self.app = app
        self._skip_paths = skip_paths or _DEFAULT_AUTH_SKIP_PATHS
        enabled = auth_components is not None and bool(getattr(auth_components, "enabled", False))
        self._authz = getattr(auth_components, "authz_middleware", None) if enabled else None

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if scope["type"] not in ("http", "websocket"):
            await self.app(scope, receive, send)
            return

        path = self._router_relative_path(scope)
        if _should_skip_auth_path(path, self._skip_paths):
            await self.app(scope, receive, send)
            return

        # CORS preflight carries no credentials by design; CORSMiddleware sits
        # outside this and answers it, but a direct OPTIONS must not 403 here.
        method = scope.get("method", "GET" if scope["type"] == "websocket" else "")
        if method.upper() == "OPTIONS":
            await self.app(scope, receive, send)
            return

        authz = self._authz
        if authz is None:
            # Auth off narrows nothing. Recording that spares a tenant-aware
            # handler from having to infer it.
            record_grant_scope(scope, GrantScope())
            await self.app(scope, receive, send)
            return

        principal = self._principal(scope)
        if principal is None or principal.is_anonymous():
            await _send_authz_failure(scope, receive, send, MissingCredentialsError("Authentication required"))
            return

        rule = resolve_rule(method, path)
        if rule is None:
            _AuthLoggerAdapter.warning(
                "authz_route_not_in_permission_table",
                path=path,
                method=method,
                principal_id=str(getattr(principal, "id", "unknown")),
            )
            await _send_authz_failure(
                scope,
                receive,
                send,
                AccessDeniedError(
                    principal_id=str(getattr(principal, "id", "unknown")),
                    action=method.lower(),
                    resource=path,
                    reason="route has no permission mapping",
                ),
            )
            return

        if rule.permission is None:
            await self.app(scope, receive, send)
            return

        resource_type, action = rule.permission
        try:
            decision = authz.authorize(
                principal=principal,
                action=action,
                resource_type=resource_type,
                resource_id="*",
            )
        except AccessDeniedError as exc:
            await _send_authz_failure(scope, receive, send, exc)
            return

        grant = GrantScope.of(decision)
        if grant.confined and (not rule.tenant_aware or grant.tenant_id is None):
            principal_id = str(getattr(principal, "id", "unknown"))
            _AuthLoggerAdapter.warning(
                "authz_tenant_scoped_grant_refused",
                path=path,
                method=method,
                principal_id=principal_id,
                grant_tenant=grant.tenant_id,
                rule=rule.template,
            )
            await _send_authz_failure(
                scope,
                receive,
                send,
                AccessDeniedError(
                    principal_id=principal_id,
                    action=action,
                    resource=f"{resource_type}:*",
                    reason="tenant-scoped grant; route requires a global grant",
                ),
            )
            return

        record_grant_scope(scope, grant)
        await self.app(scope, receive, send)

    @staticmethod
    def _router_relative_path(scope: Scope) -> str:
        """Return the path relative to this router, stripping any mount prefix.

        Starlette (>=0.35, and 1.x) does NOT rewrite ``scope["path"]`` when it
        dispatches into a mounted sub-application: it leaves the full path in
        place and records the consumed prefix in ``scope["root_path"]``. The
        served application mounts this router at ``/api`` (lifecycle.run_http),
        so a table keyed on the raw path would match ``/api/groups`` against a
        rule written as ``/groups`` -- match nothing, and therefore deny
        everything, because the table's default is deny.

        Unit tests that build the router directly never see this: root_path is
        empty there, so the raw path already is the relative one. Only the
        served-app assembly exposes it, which is why
        tests/integration/test_rest_authz_on_served_app.py exists.
        """
        path: str = scope.get("path", "")
        root_path: str = scope.get("root_path", "")
        if root_path and path.startswith(root_path):
            return path[len(root_path) :] or "/"
        return path

    @staticmethod
    def _principal(scope: Scope) -> Any:
        """Read the authenticated principal off the scope.

        The scope carries auth context in one of two shapes, because the two
        authentication middlewares write it differently:
        ``AuthEnforcementMiddleware`` installs a ``State`` object via
        ``_store_auth_context``, while ``AuthMiddlewareHTTP`` assigns
        ``request.state.auth``, and Starlette's ``Request.state`` wraps a plain
        ``dict`` it leaves in ``scope["state"]``. Reading only one shape makes
        every request look unauthenticated -- a 401 that no credential can fix.
        """
        state = scope.get("state")
        if state is None:
            return None
        auth_context = state.get("auth") if isinstance(state, dict) else getattr(state, "auth", None)
        return getattr(auth_context, "principal", None)


class AuthMiddlewareHTTP(BaseHTTPMiddleware):
    """Starlette HTTP middleware adapter over shared auth enforcement."""

    _authn: Any
    _skip_paths: frozenset[str]
    _trusted_proxies: TrustedProxyResolver
    _oidc_issuers: list[str]
    _oidc_resource_uri: str

    def __init__(
        self,
        app: ASGIApp,
        authn: Any,
        skip_paths: frozenset[str] | None = None,
        oidc_issuers: list[str] | None = None,
        oidc_resource_uri: str = "",
    ) -> None:
        super().__init__(app)
        self._authn = authn
        self._skip_paths = skip_paths or _DEFAULT_AUTH_SKIP_PATHS
        self._trusted_proxies = TrustedProxyResolver()
        self._oidc_issuers = oidc_issuers if oidc_issuers is not None else []
        self._oidc_resource_uri = oidc_resource_uri

    async def dispatch(self, request: Request, call_next: RequestResponseEndpoint) -> Response:
        path = request.url.path
        if _should_skip_auth_path(path, self._skip_paths):
            return await call_next(request)

        auth_request, _source_ip = _build_auth_request(request.scope, self._trusted_proxies)
        try:
            auth_context = self._authn.authenticate(auth_request)
            request.state.auth = auth_context
            return await call_next(request)
        except AuthenticationError as exc:
            # RFC 9728: advertise resource_metadata in Bearer challenge when OIDC active.
            if self._oidc_issuers:
                from ...auth.prm import build_resource_base_url, build_www_authenticate

                resource_base = self._oidc_resource_uri or build_resource_base_url(request.scope)
                www_auth = build_www_authenticate(resource_base)
            else:
                www_auth = "Bearer, ApiKey"
            return JSONResponse(
                status_code=401,
                content={
                    "error": "authentication_failed",
                    "message": exc.message,
                    "details": exc.details,
                },
                headers={"WWW-Authenticate": www_auth},
            )
        except AccessDeniedError as exc:
            return JSONResponse(
                status_code=403,
                content={
                    "error": "access_denied",
                    "message": str(exc),
                    "principal_id": exc.principal_id,
                    "action": exc.action,
                    "resource": exc.resource,
                },
            )


def create_auth_enforced_app(
    inner_app: ASGIApp,
    auth_components: Any,
    *,
    skip_paths: frozenset[str] | None = None,
) -> ASGIApp:
    """Wrap an ASGI app with shared auth enforcement when available."""
    authn = getattr(auth_components, "authn_middleware", None)
    if authn is None:
        return inner_app
    return AuthEnforcementMiddleware(
        inner_app,
        authn=authn,
        skip_paths=skip_paths,
        oidc_issuers=getattr(auth_components, "oidc_issuers", []),
        oidc_resource_uri=getattr(auth_components, "oidc_resource_uri", ""),
    )


def _get_status_code(exc: Exception) -> int:
    """Determine HTTP status code for an exception.

    Args:
        exc: The exception to map.

    Returns:
        HTTP status code integer.
    """
    for exc_type, status_code in _EXCEPTION_STATUS_MAP:
        if isinstance(exc, exc_type):
            return status_code
    return 500


async def error_handler(request: Request, exc: Exception) -> HangarJSONResponse:
    """Convert exceptions to JSON error envelopes.

    Maps domain exceptions to appropriate HTTP status codes.
    Unhandled exceptions get 500 with a generic message (internals not leaked).

    Args:
        request: The Starlette request.
        exc: The exception that was raised.

    Returns:
        HangarJSONResponse with error envelope body.
    """
    status_code = _get_status_code(exc)

    if isinstance(exc, MCPError):
        error_body: dict[str, Any] = {
            "error": {
                "code": type(exc).__name__,
                "message": exc.message,
                "details": exc.details or None,
            }
        }
    else:
        # Generic exception -- do NOT expose internal message
        logger.exception("Unhandled exception in API request", exc_info=exc)
        error_body = {
            "error": {
                "code": "InternalServerError",
                "message": "An internal server error occurred.",
                "details": None,
            }
        }

    # A rate-limit refusal that knows when its budget refills says so the way
    # HTTP does, as well as in `details` (#1471).
    retry_after = exc.retry_after if isinstance(exc, RateLimitExceeded) else None
    headers = {"Retry-After": str(math.ceil(retry_after))} if retry_after is not None else None
    return HangarJSONResponse(error_body, status_code=status_code, headers=headers)


async def dispatch_query(query: Any) -> Any:
    """Dispatch a query to the query bus using run_in_threadpool.

    The backend is thread-based, so all CQRS calls must be executed
    via run_in_threadpool to avoid blocking the async event loop.

    Args:
        query: The query to execute.

    Returns:
        Result from query_bus.execute(query).
    """
    ctx = get_context()
    return await run_in_threadpool(ctx.query_bus.execute, query)


async def dispatch_command(command: Any) -> Any:
    """Dispatch a command to the command bus using run_in_threadpool.

    The backend is thread-based, so all CQRS calls must be executed
    via run_in_threadpool to avoid blocking the async event loop.

    Args:
        command: The command to send.

    Returns:
        Result from command_bus.send(command).
    """
    ctx = get_context()
    return await run_in_threadpool(ctx.command_bus.send, command)


def _validate_origin(origin: str) -> str | None:
    """Return the origin if valid, or None otherwise.

    Logs a warning on rejection.
    """
    origin = origin.strip()
    if not _ORIGIN_RE.match(origin):
        logger.warning("cors_origin_rejected origin=%s reason=%s", origin, "invalid format or wildcard")
        return None
    parsed = urlparse(origin)
    if not parsed.hostname:
        logger.warning("cors_origin_rejected origin=%s reason=%s", origin, "no hostname")
        return None
    return origin


def get_cors_config() -> dict[str, Any]:
    """Get CORS configuration from environment variables.

    Reads MCP_CORS_ORIGINS (comma-separated) from the environment.
    Each origin must have a scheme (http:// or https://) and no wildcards.
    Defaults to http://localhost:5173 when MCP_CORS_ORIGINS is not set.

    allow_credentials is False by default. Set MCP_CORS_CREDENTIALS=true
    to enable (requires explicit non-wildcard origins).

    Returns:
        Dict of CORSMiddleware kwargs.
    """
    cors_origins_env = os.environ.get("MCP_CORS_ORIGINS", "")
    if cors_origins_env.strip():
        raw = [o.strip() for o in cors_origins_env.split(",") if o.strip()]
        allow_origins = [o for o in (_validate_origin(o) for o in raw) if o is not None]
        if not allow_origins:
            logger.warning("cors_all_origins_rejected fallback=%s", "http://localhost:5173")
            allow_origins = ["http://localhost:5173"]
    else:
        allow_origins = ["http://localhost:5173"]

    allow_credentials = os.environ.get("MCP_CORS_CREDENTIALS", "false").lower() == "true"

    return {
        "allow_origins": allow_origins,
        "allow_methods": _ALLOWED_METHODS,
        "allow_headers": _ALLOWED_HEADERS,
        "allow_credentials": allow_credentials,
    }