"""Fail-closed guard for ``ui://`` (MCP Apps / SEP-1865) resources. MCP Apps lets a tool return a ``ui://`` resource that a client renders in a webview / sandboxed iframe -- an XSS / data-exfiltration surface. This guard is the enforcement point that decides whether a ``ui://`` resource may be delivered, and under what constraints. It enforces three fail-closed controls: 1. **Per-tenant allowlist.** A ``ui://`` resource is DENIED unless it matches the tenant's :class:`~mcp_hangar.domain.value_objects.ui_resource.UiResourcePolicy` allowlist. **The default policy has an empty allowlist, so every ``ui://`` resource is denied by default.** An unknown tenant likewise gets the empty default -> denied. 2. **Restrictive CSP.** An allowed ``ui://`` resource carries a restrictive Content-Security-Policy (:data:`~mcp_hangar.domain.value_objects.ui_resource.DEFAULT_UI_CSP`) in its decision, to be attached to the delivered resource's metadata. 3. **Mandatory consent.** An allowed ``ui://`` resource additionally requires a consent decision from the existing approval gate before delivery. No consent gate wired, or consent not granted (denied / error / timeout) -> DENIED. Non-``ui://`` resources are completely unaffected: :meth:`UiResourceGuard.evaluate` returns a pass-through decision and :meth:`UiResourceGuard.enforce` never invokes the consent gate for them. Wiring (#1048). The front door relays resources since #1031, and ``resource_link_read_through`` calls :meth:`UiResourceGuard.evaluate` on every catalogue entry and :meth:`UiResourceGuard.enforce` before delivering a read. The process guard is the one :func:`get_ui_resource_guard` returns: configuration fills its per-tenant policies (``ui_resources.tenants`` in the config file) and bootstrap attaches the consent gate once the approval service exists. Both halves have to be present for a ``ui://`` resource to be delivered at all, and they fail closed independently: no allowlist entry denies before consent is ever asked for, and an allowlisted resource with no consent gate attached is denied too. That is deliberate -- ADR-024 records why this consent is the one human decision that belongs on a fetch, and it is only a decision if there is somebody to make it. """ from __future__ import annotations from dataclasses import dataclass from typing import Protocol, runtime_checkable from mcp_hangar.domain.value_objects.ui_resource import ( UiResourcePolicy, is_ui_scheme, ) from mcp_hangar.logging_config import get_logger logger = get_logger(__name__) @dataclass(frozen=True) class UiResourceDecision: """Outcome of evaluating a single resource against the guard. Attributes: uri: The resource URI that was evaluated. is_ui: Whether the URI targets the ``ui://`` scheme. allowed: Whether the resource may be delivered. For non-``ui://`` resources this is always True (pass-through). For ``ui://`` resources it is True only when allowlisted AND (if required) consent was granted. csp: The Content-Security-Policy to attach to the delivered resource, or None when not applicable (non-``ui://`` or denied). requires_consent: Whether an allowlisted ``ui://`` resource still needs a consent decision before delivery. Set on the pure :meth:`UiResourceGuard.evaluate` result; cleared once consent has been resolved by :meth:`UiResourceGuard.enforce`. reason: Human-readable explanation, primarily for denials / audit. """ uri: str is_ui: bool allowed: bool csp: str | None = None requires_consent: bool = False reason: str | None = None @runtime_checkable class UiConsentGate(Protocol): """Consent provider for ``ui://`` delivery. Implemented in production by an adapter over the existing ``ApprovalGateService`` (see :class:`ApprovalConsentGate`). Returns True only when a human has affirmatively consented to delivering the ``ui://`` resource; returns False on denial, timeout, or error (fail-closed). """ async def request_consent( self, uri: str, tenant_id: str | None, mcp_server_id: str, correlation_id: str, ) -> bool: """Request consent to deliver a ``ui://`` resource. Fail-closed.""" ... class UiResourceGuard: """Fail-closed enforcement point for ``ui://`` resources. Holds per-tenant :class:`UiResourcePolicy` objects plus a default policy used for tenants with no explicit policy (empty allowlist -> deny all ``ui://``). Optionally holds a :class:`UiConsentGate` used to mandate consent on allowed ``ui://`` resources. ``evaluate`` is a pure, synchronous allowlist + CSP decision (no consent). ``enforce`` is the full async enforcement: allowlist + CSP + mandatory consent. Both leave non-``ui://`` resources untouched. """ def __init__( self, policies: dict[str, UiResourcePolicy] | None = None, *, default_policy: UiResourcePolicy | None = None, consent_gate: UiConsentGate | None = None, ) -> None: # Per-tenant policies. Absent tenant -> default_policy (fail-closed). self._policies: dict[str, UiResourcePolicy] = dict(policies or {}) # The default is an empty-allowlist policy: deny all ui:// unless a # tenant explicitly allowlists a resource. self._default_policy = default_policy or UiResourcePolicy() self._consent_gate = consent_gate def attach_consent_gate(self, consent_gate: UiConsentGate | None) -> None: """Attach the consent provider after construction (#1048). The guard is configured in two steps because its two inputs become available at different times: the per-tenant policies come from the config file, and the gate is an adapter over the approval service, which does not exist until components are built. Until this is called an allowlisted ``ui://`` resource is denied -- consent is mandated, and a mandate with nobody to ask is a denial. """ self._consent_gate = consent_gate def replace_policies(self, policies: dict[str, UiResourcePolicy]) -> None: """Replace the per-tenant policies, and keep the consent gate (#1424). The file's ``ui_resources`` block is read at startup and again on every reload. Building a new guard for it would drop the consent gate bootstrap attached, and every allowlisted resource would then be denied for want of anyone to ask. One assignment, so a concurrent decision sees the previous policies or the new ones. """ self._policies = dict(policies) def policy_for(self, tenant_id: str | None) -> UiResourcePolicy: """Return the effective policy for ``tenant_id`` (fail-closed default). Unknown or None tenant -> the empty-allowlist default policy, so every ``ui://`` resource is denied. """ if tenant_id is None: return self._default_policy return self._policies.get(tenant_id, self._default_policy) def evaluate(self, uri: str, tenant_id: str | None) -> UiResourceDecision: """Pure allowlist + CSP decision for a single resource (no consent). - Non-``ui://`` resource -> pass through unchanged (allowed, no CSP). - ``ui://`` resource not on the tenant allowlist -> denied (fail-closed). - ``ui://`` resource on the allowlist -> allowed here, carrying the CSP and flagged ``requires_consent`` per policy. Final delivery still requires :meth:`enforce` to resolve consent. """ if not is_ui_scheme(uri): return UiResourceDecision( uri=uri, is_ui=False, allowed=True, reason="non-ui-scheme: not governed by the ui:// guard", ) policy = self.policy_for(tenant_id) if not policy.is_allowed(uri): return UiResourceDecision( uri=uri, is_ui=True, allowed=False, reason="ui:// resource not on the tenant allowlist (fail-closed)", ) return UiResourceDecision( uri=uri, is_ui=True, allowed=True, csp=policy.csp, requires_consent=policy.require_consent, reason="ui:// resource allowlisted; consent required before delivery" if policy.require_consent else "ui:// resource allowlisted", ) async def enforce( self, uri: str, tenant_id: str | None, mcp_server_id: str, correlation_id: str = "", ) -> UiResourceDecision: """Full fail-closed enforcement for a single resource before delivery. Runs :meth:`evaluate`, then -- for an allowlisted ``ui://`` resource that requires consent -- mandates consent via the wired :class:`UiConsentGate`. Fail-closed on every consent edge: - allowlisted + requires consent but **no consent gate wired** -> DENIED. - consent gate returns False (denied / timeout) -> DENIED. - consent gate raises -> DENIED (error is swallowed, not propagated). Non-``ui://`` resources and denied ``ui://`` resources return the ``evaluate`` decision unchanged (the consent gate is never consulted). """ decision = self.evaluate(uri, tenant_id) if not decision.is_ui or not decision.allowed: return decision if not decision.requires_consent: return decision if self._consent_gate is None: logger.warning( "ui_resource_consent_gate_missing", uri=uri, tenant_id=tenant_id, mcp_server_id=mcp_server_id, ) return UiResourceDecision( uri=uri, is_ui=True, allowed=False, reason="ui:// consent mandated but no consent gate wired (fail-closed)", ) try: consented = await self._consent_gate.request_consent( uri=uri, tenant_id=tenant_id, mcp_server_id=mcp_server_id, correlation_id=correlation_id, ) except Exception: # noqa: BLE001 -- fail-closed: any error denies delivery logger.warning( "ui_resource_consent_gate_error", uri=uri, tenant_id=tenant_id, mcp_server_id=mcp_server_id, exc_info=True, ) return UiResourceDecision( uri=uri, is_ui=True, allowed=False, reason="ui:// consent gate error (fail-closed)", ) if not consented: return UiResourceDecision( uri=uri, is_ui=True, allowed=False, reason="ui:// resource consent not granted (fail-closed)", ) # Consent granted: deliver with CSP; consent is now resolved. return UiResourceDecision( uri=uri, is_ui=True, allowed=True, csp=decision.csp, requires_consent=False, reason="ui:// resource allowlisted and consent granted", ) #: The process-wide guard. Configuration fills its policies, bootstrap attaches #: its consent gate, and the resources projection asks it about every entry. _guard: UiResourceGuard | None = None def get_ui_resource_guard() -> UiResourceGuard: """Return the process guard, creating the fail-closed default on first use. A guard nobody configured denies every ``ui://`` resource, which is the right answer for a deployment that never opted in. """ global _guard if _guard is None: _guard = UiResourceGuard() return _guard def set_ui_resource_guard(guard: UiResourceGuard) -> None: """Replace the process guard -- config load builds it from the file.""" global _guard _guard = guard def reset_ui_resource_guard() -> None: """Drop the process guard, so the next read builds the default again.""" global _guard _guard = None