feat(governance): let a policy name the OTLP collector hosts

docs/system-specs/modules/governance.md

@@ -3025,8 +3025,14 @@ session-bearing chokepoint, not this probe.
 
 The Security panel picks the row up automatically — `api_governance_policy` iterates
 `SCOPE_CATALOG` — and labels it **"Anonymous telemetry"** rather than the leaf's
-bare "Telemetry", because this scope governs only the outbound heartbeat and NOT the
-unrelated local-only `telemetry.enabled` OTEL collection.
+bare "Telemetry", because this scope governs the outbound heartbeat and OTLP export
+(below) but NOT the local-only `telemetry.enabled` OTEL collection.
+
+**Which OTLP collectors — the `otlp_hosts` list.** The row also carries an inner `otlp_hosts` ruleset (`host` matcher), so a policy can name the exact collector hosts metrics export may reach (RFC [crew-enterprise-edition](../../request-for-change/rfc-crew-enterprise-edition.md), appendix row "Diagnostic telemetry target"). `metrics/otlp_governance.py` holds metrics to it twice. `governed_destinations` asks `otlp_hosts:<host>` for every reading of each OTLP destination's host (`governance_parts.matching.url_policy_hosts`: the host as the standard library, `urllib3` and `yarl` each read it, plain and with the host part percent-decoded, lower-cased, without a root dot or an IPv6 zone; an IP literal in canonical form; a non-ASCII name only in the forms the clients send (Python's `idna` codec for `urllib`, UTS #46 `xn--` for `urllib3` and `yarl`). A URL with no such reading is asked as `""`, which matches no entry: as for `network.egress`, an allow list is the hard bound, entries name ASCII hosts without a root dot and IPs in canonical form (an exact entry in any other form, or a glob with a character no reading holds, fails the policy load, `archetypes._refuse_unmatchable_host_entries`; an IP glob in a non-canonical numeric spelling is a known gap, #23734), and a deny list refuses the spellings it can name), permits it only if every reading is permitted, and drops a refused one before its reader is built (once-per-host WARNING naming the host only). The Privacy panel's `otlp_egress_active`, which also gates enabling collection from the local-only switch, deliberately reads the UNFILTERED set: a host refused today may be permitted by a later policy, so it still counts as egress. `gate_exporter` then asks again before every export of a running reader, so in a process with a platform context a policy installed at runtime (`policy_distribution.apply_ceiling`) stops the next export, the shutdown flush included, with no rebuild. The MCP gateway daemon has no context and receives no runtime policy install: it holds the ceiling its companion was composed with until it restarts. Build decisions and each change of a running reader's answer write a SEL `governance_decision` row (tool `telemetry_otlp_export`); the per-interval exports do not. Policy layer only, for the beacon's reason: the build is process-wide and carries no session. Because an item query checks the gate first, a policy that pins `capabilities.telemetry` off now stops OTLP metrics export as well as the beacon. A policy without the ruleset, or no policy, keeps every destination. The MCP gateway daemon boots no context, so `bootstrap.compose_telemetry_provider` wraps the companion's provider in `_PolicyBoundTelemetry`, which carries the companion's ceiling as `governance_ceiling` and its credential redaction as `governance_redact`; `provider._policy_ceiling` reads the first for the build and each export there, and the SEL rows go through the second (`log_governance_decision(redact=...)`), because the SEL's own `redact_via_context` raises in a process with no context. Before composition the daemon has only the no-destination stand-in. A process with no context whose provider carries no ceiling (a standalone daemon whose policy could not be read) refuses every host: a failure to load the policy is not the absence of one.
+
+```json
+{"version": 1, "capabilities": {"telemetry": {"scopes": {"otlp_hosts": {"mode": "allow", "allow": ["otel.corp.example"]}}}}}
+```
 
 ### Tailnet origin derivation — `capabilities.tailnet_origin`
 

docs/system-specs/modules/metrics.md

@@ -202,7 +202,10 @@ change to them applies at the next start.
 **pluggable**. `_build_recorder` asks the active `TelemetryProvider` for them via
 `otlp_destinations(cfg)` and attaches one `PeriodicExportingMetricReader` per
 returned destination that names the `"metrics"` signal (`provider._otlp_destinations`
-resolves, `provider._build_otlp_reader` builds one reader). The public
+resolves, `provider._build_otlp_reader` builds one reader). Each destination's
+host must also pass the security policy's `capabilities.telemetry` `otlp_hosts`
+list (`otlp_governance.governed_destinations` before the reader is built and `otlp_governance.gate_exporter` before each export; see [governance.md](governance.md)), and a
+policy that pins `capabilities.telemetry` off drops every destination. The public
 `DefaultTelemetryProvider` turns a non-empty `otlp_endpoint` into exactly one
 destination and returns none when it is empty, so a standalone build behaves as it
 did when this module constructed that exporter itself. Readers are appended AFTER

src/kiro_crew/metrics/otlp_governance.py

@@ -0,0 +1,171 @@
+"""Which OTLP collectors metrics may reach: ``capabilities.telemetry`` ``otlp_hosts``.
+
+A policy names collector hosts in the row's inner ``otlp_hosts`` ruleset (RFC
+crew-enterprise-edition, appendix row "Diagnostic telemetry target"). Two places
+hold metrics to it:
+
+* :func:`governed_destinations` -- before a reader is built, a destination whose
+  host the policy does not permit gets no reader;
+* :func:`gate_exporter` -- before every export of a reader already running, so in
+  a process with a platform context a policy installed at runtime
+  (``policy_distribution.apply_ceiling``) stops the next export, the shutdown
+  flush included, without waiting for a rebuild. The MCP gateway daemon has no
+  context and gets no runtime install; it keeps its composed companion's ceiling
+  until it restarts.
+
+Policy layer only, like the beacon probe: both run process-wide with no session,
+so a per-surface profile is not asked. No ceiling is still asked through
+``resolve``, so the answer it gives with no policy holds; a ceiling without the row
+or the ruleset permits every host. An evaluation error denies (egress fails closed;
+local collection carries on). Enforcement decisions are SEL-audited: each build
+decision, and each change of a running reader's answer -- not every export, which
+would write a row per export interval.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any, Callable, Iterable
+
+logger = logging.getLogger(__name__)
+
+#: Governed scope whose inner ``otlp_hosts`` ruleset names the permitted collectors.
+TELEMETRY_SCOPE = "capabilities.telemetry"
+
+#: A credential redaction function, as ``CredentialPolicy.redact``.
+Redactor = Callable[[str], str]
+
+#: Tool name on the SEL ``governance_decision`` row for these decisions.
+AUDIT_TOOL = "telemetry_otlp_export"
+
+#: Stands for a ceiling that could not be established. Every host is refused under
+#: it: a failure to load the policy must not read as "no policy".
+UNEVALUABLE = object()
+
+#: Hosts already reported as dropped, so a pinned list warns once, not per read.
+_dropped_hosts_warned: set[str] = set()
+
+
+def hosts_of(endpoint: object) -> "tuple[str, ...]":
+    """Every host a client could read an OTLP endpoint as (``url_policy_hosts``)."""
+    from kiro_crew.platform.governance_parts.matching import url_policy_hosts
+
+    return url_policy_hosts(endpoint)
+
+
+def _decide(ceiling: Any, hosts: "tuple[str, ...]") -> "tuple[bool, str, str, str]":
+    """``(permitted, host, rule, reason)``: permitted only if EVERY reading is.
+
+    *host* is the reading the answer rests on: the first refused one, else the
+    first. Fails closed.
+    """
+    from kiro_crew.platform.governance import resolve
+
+    first = hosts[0] if hosts else ""
+    if ceiling is UNEVALUABLE:
+        return False, first, "default", "governance unavailable (fail-closed)"
+    try:
+        for host in hosts or ("",):
+            decision = resolve(ceiling, None, TELEMETRY_SCOPE, f"otlp_hosts:{host}")
+            if not decision.permitted:
+                return False, host, str(decision.rule), str(decision.reason)
+    except Exception:
+        logger.debug("OTLP host governance evaluation failed; denying")
+        return False, first, "default", "governance unavailable (fail-closed)"
+    return True, first, str(decision.rule), str(decision.reason)
+
+
+def _audit(
+    host: str, permitted: bool, rule: str, reason: str, redact: "Redactor | None" = None
+) -> None:
+    """Best-effort SEL row for one enforcement decision. Never raises.
+
+    *redact* is the companion's redaction in a process with no platform context,
+    where the SEL's default shim would raise; ``None`` everywhere else.
+    """
+    try:
+        # Looked up at call time: the suite patches ``kiro_crew.sel.sel``.
+        from kiro_crew.sel import sel
+
+        sel().log_governance_decision(
+            session_key="",
+            tool_name=AUDIT_TOOL,
+            scope=TELEMETRY_SCOPE,
+            item=f"otlp_hosts:{host}",
+            outcome="allowed" if permitted else "denied",
+            rule=rule,
+            layer="policy",
+            reason=reason,
+            redact=redact,
+        )
+    except Exception:
+        pass
+
+
+def governed_destinations(
+    supplied: "Iterable[Any]",
+    ceiling: Any,
+    *,
+    audit: bool = False,
+    redact: "Redactor | None" = None,
+) -> tuple:
+    """Keep the destinations whose host *ceiling* permits.
+
+    *audit* is for the build that acts on the answer.
+    """
+    allowed = []
+    for dest in supplied:
+        # No ceiling is still asked through ``resolve``, so whatever the evaluator
+        # answers with no policy holds here too, and is audited like any other.
+        permitted, host, rule, reason = _decide(ceiling, hosts_of(getattr(dest, "endpoint", "")))
+        if audit:
+            _audit(host, permitted, rule, reason, redact)
+        if permitted:
+            allowed.append(dest)
+        elif host not in _dropped_hosts_warned:
+            # The host only, never the endpoint: it can carry userinfo or a signed query.
+            _dropped_hosts_warned.add(host)
+            logger.warning(
+                "OTLP destination %r (host %r) dropped: the security policy does not "
+                "permit it (capabilities.telemetry otlp_hosts)",
+                getattr(dest, "name", ""),
+                host,
+            )
+    return tuple(allowed)
+
+
+def gate_exporter(
+    exporter: Any,
+    endpoint: object,
+    ceiling: "Callable[[], Any]",
+    *,
+    redact: "Redactor | None" = None,
+) -> None:
+    """Hold every ``export`` of *exporter* to the policy *ceiling()* returns now.
+
+    The reader calls ``exporter.export`` on each interval and once more on
+    shutdown; a refused call drops that batch and returns ``FAILURE`` without
+    contacting the collector. The reader was built under a permit, so the first
+    refusal (and a later return to permitted) is audited and logged once.
+    """
+    hosts = hosts_of(endpoint)
+    inner = exporter.export
+    last = {"permitted": True}
+
+    def export(metrics_data: Any, *args: Any, **kwargs: Any) -> Any:
+        permitted, host, rule, reason = _decide(ceiling(), hosts)
+        if permitted != last["permitted"]:
+            last["permitted"] = permitted
+            _audit(host, permitted, rule, reason, redact)
+            logger.warning(
+                "OTLP export to host %r %s by the security policy (capabilities.telemetry)",
+                host,
+                "resumed" if permitted else "stopped",
+            )
+        if not permitted:
+            from opentelemetry.sdk.metrics.export import MetricExportResult
+
+            return MetricExportResult.FAILURE
+        return inner(metrics_data, *args, **kwargs)
+
+    exporter.export = export

src/kiro_crew/metrics/provider.py

@@ -46,6 +46,7 @@
 from kiro_crew import __version__, beacon
 from kiro_crew.config.loader import KiroCrewConfig
 from kiro_crew.config.paths import config_dir
+from kiro_crew.metrics import otlp_governance
 from kiro_crew.metrics.recorder import MetricsRecorder
 from kiro_crew.platform.context import (
     PlatformCompositionError,
@@ -1138,6 +1139,51 @@ def _filter_metric_destinations(
     return tuple(kept)
 
 
+def _policy_redactor() -> "Callable[[str], str] | None":
+    """The redaction OTLP SEL rows go through when this process has no context.
+
+    ``None`` -- the SEL's own ``redact_via_context`` -- whenever a context exists.
+    Without one, the composed companion provider's ``governance_redact``.
+    """
+    try:
+        current_context()
+        return None
+    except PlatformCompositionError:
+        pass
+    try:
+        redact = getattr(_telemetry_provider(), "governance_redact", None)
+    except Exception:
+        return None
+    return redact if callable(redact) else None
+
+
+def _policy_ceiling() -> object | None:
+    """The policy ceiling OTLP export is held to in this process, or ``None``.
+
+    The platform context's, when the process has one. A process that never boots
+    the platform on a non-standalone profile (the MCP gateway daemon) has none; its
+    composed companion provider carries the companion's ceiling instead
+    (``bootstrap.compose_telemetry_provider``), and before composition it has no
+    destinations at all.
+
+    ``otlp_governance.UNEVALUABLE`` when neither answers: the context failed to
+    compose and the provider carries no ceiling of its own (for instance a
+    standalone daemon whose policy could not be read). Every host is refused then;
+    a failure to load the policy is not the absence of one.
+    """
+    try:
+        return getattr(current_context(), "governance", None)
+    except PlatformCompositionError:
+        pass
+    try:
+        provider = _telemetry_provider()
+    except Exception:
+        return otlp_governance.UNEVALUABLE
+    if not hasattr(provider, "governance_ceiling"):
+        return otlp_governance.UNEVALUABLE
+    return getattr(provider, "governance_ceiling", None)
+
+
 def _otlp_destinations(cfg: "TelemetryConfig") -> "tuple[OtlpDestination, ...]":
     """Resolve the metric egress destinations the active edition supplies.
 
@@ -1183,7 +1229,18 @@ def _otlp_destinations(cfg: "TelemetryConfig") -> "tuple[OtlpDestination, ...]":
         logger.warning("platform context unavailable; OTLP egress disabled (local-only)")
         return ()
 
-    return _filter_metric_destinations(supplied)
+    # The policy's ``otlp_hosts`` list narrows what is BUILT, audited here, and
+    # ``gate_exporter`` re-checks each export. Deliberately not applied in
+    # ``otlp_egress_active``: that answers whether enabling collection could ever
+    # send metrics off the machine, and a host refused today may be permitted later.
+    kept = _filter_metric_destinations(supplied)
+    if not kept:
+        # Nothing to hold to a policy; and resolving the ceiling could compose a
+        # companion the provider read above just failed to compose.
+        return ()
+    return otlp_governance.governed_destinations(
+        kept, _policy_ceiling(), audit=True, redact=_policy_redactor()
+    )
 
 
 _FOREIGN_SIGNAL_ROUTES = (("/v1/traces", "traces"), ("/v1/logs", "logs"))
@@ -1289,6 +1346,11 @@ def _build_otlp_reader(dest: "OtlpDestination", cfg: object) -> Optional["_Reade
             # separate field on the descriptor.
             kwargs["session"] = dest.session
         exporter = OTLPMetricExporter(**kwargs)
+        # Every later export is held to the policy in force THEN, so a runtime
+        # policy install stops a running reader without waiting for a rebuild.
+        otlp_governance.gate_exporter(
+            exporter, endpoint, _policy_ceiling, redact=_policy_redactor()
+        )
         return PeriodicExportingMetricReader(
             exporter,
             export_interval_millis=float(getattr(cfg, "export_interval_seconds", 60)) * 1000.0,

src/kiro_crew/platform/bootstrap.py

@@ -12,7 +12,7 @@
 
 import logging
 import threading
-from typing import TYPE_CHECKING, Any, Sequence
+from typing import TYPE_CHECKING, Any, Sequence, cast
 
 from kiro_crew.platform.context import (
     CONTRACT_VERSION,
@@ -136,7 +136,35 @@ def compose_telemetry_provider(cfg: "KiroCrewConfig") -> "TelemetryProvider":
         return DefaultTelemetryProvider()
     companion = _checked_companion(profile, cfg, eps)
     assert_security_floor(companion.security)
-    return companion.telemetry
+    # The calling process installs no context, so the core cannot read a ceiling
+    # for these destinations: the provider carries the companion's own.
+    return cast(
+        "TelemetryProvider",
+        _PolicyBoundTelemetry(
+            companion.telemetry,
+            getattr(companion, "governance", None),
+            getattr(getattr(companion, "credentials", None), "redact", None),
+        ),
+    )
+
+
+class _PolicyBoundTelemetry:
+    """The companion's telemetry provider, plus the ceiling its OTLP export obeys.
+
+    ``governance_ceiling`` is what ``metrics.provider`` holds destinations and each
+    export to in a process with no context, and ``governance_redact`` is the
+    companion's credential redaction its SEL rows go through there. Every other
+    attribute, including ``otlp_destinations`` (or its absence), is the wrapped
+    provider's own.
+    """
+
+    def __init__(self, inner: object, ceiling: object | None, redact: object = None) -> None:
+        self._inner = inner
+        self.governance_ceiling = ceiling
+        self.governance_redact = redact
+
+    def __getattr__(self, name: str) -> object:
+        return getattr(self._inner, name)
 
 
 def _checked_companion(

src/kiro_crew/platform/governance.py

@@ -1099,8 +1099,10 @@ def assert_governance_paths_protected() -> None:
         _compose_controls,
         _compose_posture,
         _dedup,
+        _refuse_unmatchable_host_entries,
         _reject_unknown_keys,
         _str_tuple,
+        _unmatchable_glob,
     )
     from kiro_crew.platform.governance_parts.baseline import (  # noqa: F401
         _CAPABILITY_OFF,
@@ -1169,6 +1171,8 @@ def assert_governance_paths_protected() -> None:
         _MATCHERS,
         _NETWORK_SCHEMES,
         _ORDINAL_SCALES,
+        _PLAIN_HOST_RE,
+        _canonical_ip,
         _expand,
         _looks_like_host,
         _match_command,
@@ -1179,6 +1183,7 @@ def assert_governance_paths_protected() -> None:
         _norm_item,
         _url_host,
         register_matcher,
+        url_policy_hosts,
     )
     from kiro_crew.platform.governance_parts.parsing import (  # noqa: F401
         _KEY_OPEN_NAMESPACES,