From 7af588cd04da48de4fc7ed5542b0850e10b4cda7 Mon Sep 17 00:00:00 2001 From: Joe Guo Date: Sat, 24 Oct 2026 08:50:57 +0000 Subject: [PATCH 1/2] feat(governance): let a policy name the OTLP collector hosts --- docs/system-specs/modules/governance.md | 10 +- docs/system-specs/modules/metrics.md | 5 +- src/kiro_crew/metrics/otlp_governance.py | 176 +++++++ src/kiro_crew/metrics/provider.py | 64 ++- src/kiro_crew/platform/bootstrap.py | 32 +- src/kiro_crew/platform/governance.py | 5 + .../platform/governance_parts/archetypes.py | 68 +++ .../platform/governance_parts/catalog.py | 12 +- .../platform/governance_parts/matching.py | 110 ++++ src/kiro_crew/security_posture.py | 5 + src/kiro_crew/sel.py | 13 +- test/metrics/test_mcp_gateway_gauges.py | 3 +- .../test_provider_otlp_destinations.py | 6 + test/test_governance_list_otlp_hosts.py | 473 ++++++++++++++++++ test/test_sel_bindings.py | 3 +- 15 files changed, 973 insertions(+), 12 deletions(-) create mode 100644 src/kiro_crew/metrics/otlp_governance.py create mode 100644 test/test_governance_list_otlp_hosts.py diff --git a/docs/system-specs/modules/governance.md b/docs/system-specs/modules/governance.md index 063cfa788e2..553817c1f34 100644 --- a/docs/system-specs/modules/governance.md +++ b/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:` 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` diff --git a/docs/system-specs/modules/metrics.md b/docs/system-specs/modules/metrics.md index 2c35f4dff75..45a0eaed156 100644 --- a/docs/system-specs/modules/metrics.md +++ b/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 diff --git a/src/kiro_crew/metrics/otlp_governance.py b/src/kiro_crew/metrics/otlp_governance.py new file mode 100644 index 00000000000..67fd96375a3 --- /dev/null +++ b/src/kiro_crew/metrics/otlp_governance.py @@ -0,0 +1,176 @@ +"""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, or 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. + """ + kept = list(supplied) + if ceiling is None: + return tuple(kept) + allowed = [] + for dest in kept: + 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: + current = ceiling() + permitted, host, rule, reason = ( + (True, hosts[0] if hosts else "", "", "") + if current is None + else _decide(current, 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 diff --git a/src/kiro_crew/metrics/provider.py b/src/kiro_crew/metrics/provider.py index e32babceef4..5175b039e61 100644 --- a/src/kiro_crew/metrics/provider.py +++ b/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, diff --git a/src/kiro_crew/platform/bootstrap.py b/src/kiro_crew/platform/bootstrap.py index 1550a7632d7..7bad31a0b09 100644 --- a/src/kiro_crew/platform/bootstrap.py +++ b/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( diff --git a/src/kiro_crew/platform/governance.py b/src/kiro_crew/platform/governance.py index 126c38c52be..5ed34905972 100644 --- a/src/kiro_crew/platform/governance.py +++ b/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, diff --git a/src/kiro_crew/platform/governance_parts/archetypes.py b/src/kiro_crew/platform/governance_parts/archetypes.py index a88679a8d61..91cc9090f2e 100644 --- a/src/kiro_crew/platform/governance_parts/archetypes.py +++ b/src/kiro_crew/platform/governance_parts/archetypes.py @@ -391,6 +391,71 @@ def is_at_least_as_strict_as(self, ceiling: "OrdinalControl") -> bool: return self.rank() >= ceiling.rank() +def _unmatchable_glob(glob: str) -> str: + """Why a host glob can match no reading, or ``""`` when it might. + + A character check only: a reading is a lower-case ASCII name or a canonical IP + literal, so a glob holding a ``[`` class, ``%``, ``@``, ``/``, a zone, or a + ``:`` outside a hex IPv6 glob can never match. How an IP glob spells its + numbers is not checked: it matches canonical readings as written (an + alternate numeric spelling is tracked in the deferral issue for these lists). + """ + import re + + if "[" in glob: + return "uses a [ ] character class; host lists take * and ? only" + body = glob.replace("*", "").replace("?", "") + if ":" in body and not re.fullmatch(r"[0-9a-f:]*", body): + return "is not an IPv6 glob in canonical form (hex groups only, no zone or port)" + if not re.fullmatch(r"[a-z0-9._:-]*", body): + return "has a character no host is asked with ('%', '@', '/' or similar)" + return "" + + +def _refuse_unmatchable_host_entries(scope_name: str, ruleset: "ScopedRuleset") -> None: + """Refuse a ``host`` entry in a capability's list that no host reading can match. + + A capability's ``host`` lists are asked about hosts as clients send them + (``matching.url_policy_hosts``): ASCII, no root dot, an IP literal in canonical + form. An entry in another spelling would match nothing, so a deny would refuse + nothing and an allow would admit nothing, silently. It fails to load instead, + and the message says which form to write. Top-level ``network.egress`` keeps its + own parsing; this applies only to lists inside a capability. + """ + from kiro_crew.platform.governance_parts.matching import _PLAIN_HOST_RE, _canonical_ip + + for listed, entries in (("allow", ruleset.allow), ("deny", ruleset.deny)): + for index, raw in enumerate(entries): + entry = raw.strip() + spelled = entry.lower() + why = "" + if not entry.isascii(): + why = "is not ASCII; write the name in its xn-- (IDNA) form" + elif entry.endswith("."): + why = "ends in a root dot; write it without the dot" + elif entry.startswith("[") and ":" in entry: + why = "is a bracketed IPv6 literal; write the address without brackets" + elif any(c in entry for c in "*?["): + why = _unmatchable_glob(spelled) + else: + # Exactly the form a host is asked in, or the entry can never match. + ip = _canonical_ip(spelled) + if ip: + asked = ip.split("%", 1)[0] + if asked != spelled: + why = f"is an IP literal not in the form hosts are asked in; write {asked}" + elif ":" in spelled or not _PLAIN_HOST_RE.fullmatch(spelled): + why = ( + "is not a host as clients send it " + "(brackets, a zone, a port, '%', '/' or '@')" + ) + if why: + # The position, never the value: an entry can be a pasted URL. + raise PlatformCompositionError( + f"capability scope {scope_name!r} {listed}[{index}] {why}" + ) + + # ────────────────────────────────────────────────────────────────────────── # Archetype 3 — CapabilityGate # ────────────────────────────────────────────────────────────────────────── @@ -423,6 +488,9 @@ def from_dict( for k, v in raw_scopes.items() if isinstance(v, dict) } + for name, ruleset in scopes.items(): + if isinstance(ruleset, ScopedRuleset) and ruleset.matcher == "host": + _refuse_unmatchable_host_entries(name, ruleset) if "enabled" not in d: enabled_flag = default_enabled else: diff --git a/src/kiro_crew/platform/governance_parts/catalog.py b/src/kiro_crew/platform/governance_parts/catalog.py index fad5ed42d72..ac3f878fbdb 100644 --- a/src/kiro_crew/platform/governance_parts/catalog.py +++ b/src/kiro_crew/platform/governance_parts/catalog.py @@ -258,7 +258,17 @@ class ScopeSpec: # per-surface profile is not the question it answers. # Data row only — CONTRACT_VERSION and the evaluator are untouched (mirrors # theme_install). - "capabilities.telemetry": ScopeSpec(CAPABILITY, capability_default=True), + # + # Inner ``otlp_hosts`` ruleset (``host`` matcher): which OTLP collector hosts + # the metrics exporter may send to (RFC crew-enterprise-edition, appendix row + # "Diagnostic telemetry target"). Queried as ``otlp_hosts:`` for every + # destination before its reader is built and before each export + # (``metrics/otlp_governance.py``), policy layer only like the beacon probe. A policy that names the row with + # ``enabled: false`` therefore stops OTLP export too; an unnamed ruleset + # leaves every host permitted, as before. + "capabilities.telemetry": ScopeSpec( + CAPABILITY, capability_default=True, scope_matchers={"otlp_hosts": "host"} + ), # Auto-deriving the dashboard's OWN tailnet origin (``dashboard/tailnet.py``, # RFC §4). When ``dashboard.tailscale.enabled`` is on, the gateway shells out # to the local ``tailscale`` daemon at startup and, on success, puts the diff --git a/src/kiro_crew/platform/governance_parts/matching.py b/src/kiro_crew/platform/governance_parts/matching.py index b035b0f26a5..c121097fed8 100644 --- a/src/kiro_crew/platform/governance_parts/matching.py +++ b/src/kiro_crew/platform/governance_parts/matching.py @@ -14,6 +14,7 @@ import fnmatch import os +import re from typing import Callable, Dict, Tuple from kiro_crew.platform.guest_paths import GuestPath, guest_norm @@ -268,3 +269,112 @@ def register_matcher(name: str, fn: Callable[[str, str], bool]) -> None: if existing is not None and existing is not fn: raise ValueError(f"matcher {name!r} is already registered") _MATCHERS[name] = fn + + +def url_policy_hosts(url: object) -> Tuple[str, ...]: + """Every host a client could connect to for *url*, for a ``host``-matcher list item. + + A policy that names hosts is only as strong as the host it is asked about, and + the HTTP clients in use read one URL differently: ``urllib`` percent-decodes the + authority, ``requests``/``urllib3`` and ``aiohttp``/``yarl`` treat a backslash + as the start of the path and IDNA-encode a non-ASCII name, and ``yarl`` reads + an ideographic full stop as a dot. Rather than chase spellings, each client's + own parser is asked (when installed), plus the standard library's plain and + percent-decoded readings. A caller permits the request only when EVERY reading + returned here is permitted. + + Each reading is lower-cased and loses any IPv6 zone and terminal DNS root + dot, so entries are written without the dot. An IP literal is asked in canonical form only + (dotted IPv4, compressed IPv6), so an IP entry names a host in that form. A + reading that is not a plain ASCII name is not asked, because no client sends + it as written; the clients that encode it supply the ``xn--`` or decoded form + they do send. A URL with no plain reading at all is asked as ``""``, which + matches no entry: an allow list refuses it and a deny list does not. So, as + for ``network.egress``, an allow list is the hard bound, entries name ASCII + (``xn--``) hosts, and a deny list refuses the spellings it can name. + """ + from urllib.parse import unquote, urlsplit + + text = str(url or "").strip() + raw: list = [] + + def _add(read: "Callable[[], object]") -> None: + try: + host = read() + except Exception: + return + if isinstance(host, str) and host: + raw.append(host.strip("[]")) + + _add(lambda: urlsplit(text).hostname) + # The host part only: decoding the userinfo too would cut the authority at an + # encoded ``@`` or ``/`` in a password and invent a host from it. + _add(lambda: urlsplit("//" + unquote(urlsplit(text).netloc.rpartition("@")[2])).hostname) + # What ``urllib``/``http.client`` connect to for a non-ASCII name: Python's own + # ``idna`` codec, which maps ``ß`` to ``ss`` where ``urllib3``/``yarl`` keep it. + for read in list(raw): # the two standard-library readings so far + if not read.isascii(): + + def _python_idna(host: str = read) -> str: + return host.encode("idna").decode("ascii") + + _add(_python_idna) + _add(lambda: urlsplit(text.replace("\\", "/")).hostname) + try: + from urllib3.util import parse_url + except ImportError: + pass + else: + _add(lambda: parse_url(text).host) + try: + import yarl + except ImportError: + pass + else: + _add(lambda: yarl.URL(text).raw_host) + + hosts: list = [] + + def _keep(host: str) -> None: + if host not in hosts: + hosts.append(host) + + for host in raw: + spelled = host.lower() + # An IPv6 zone (``fe80::1%eth0``) names an interface, not another host. + if ":" in spelled and "%" in spelled: + spelled = spelled.split("%", 1)[0] + bare = spelled.rstrip(".") + ip = _canonical_ip(bare) + if ip: + # An IP literal is asked in canonical form only, whatever spelling the + # resolver accepts (``167772165``, ``0x0a000005``, ``::ffff:10.0.0.5``). + _keep(ip) + continue + if not _PLAIN_HOST_RE.fullmatch(bare): + # Not a plain ASCII name: no client sends it as written. The clients + # that IDNA-encode or percent-decode it contribute their own reading. + continue + _keep(bare) + if not hosts: + # No reading is a host a client would send: ``""``, which an allow list + # refuses. + hosts.append("") + return tuple(hosts) + + +def _canonical_ip(host: str) -> str: + """The canonical text of *host* when it is an IP literal in any form, else ``""``.""" + import ipaddress + + from kiro_crew.security import canonicalize_ip + + try: + return str(ipaddress.ip_address(canonicalize_ip(host))) + except ValueError: + return "" + + +#: A host as a client sends it: ASCII letters, digits, hyphens and dots, or an IP +#: literal's hex digits and colons. +_PLAIN_HOST_RE = re.compile(r"[a-z0-9._:-]+") diff --git a/src/kiro_crew/security_posture.py b/src/kiro_crew/security_posture.py index 06e899b0bd7..e1292f6423e 100644 --- a/src/kiro_crew/security_posture.py +++ b/src/kiro_crew/security_posture.py @@ -68,6 +68,11 @@ # only to REFUSE a file whose rules, crons or carried files hold a # credential-shaped string. It returns no redacted text. "crewmate_bundle.py", + # Selects which redaction the OTLP host audit rows go through in a + # process with no platform context (`_policy_redactor` returns the + # companion's), and calls none itself. The rows are written by the SEL, + # which is the sink. + "metrics/provider.py", # The redaction cards' owner-only route (allowed hosts). It applies no # redactor; the call-site scan matches its # handler names (`api_redaction_allow_host`, ...). diff --git a/src/kiro_crew/sel.py b/src/kiro_crew/sel.py index 805c71f16ec..34b588ed03a 100644 --- a/src/kiro_crew/sel.py +++ b/src/kiro_crew/sel.py @@ -832,6 +832,7 @@ def log_governance_decision( layer: str = "", reason: str = "", critical: bool = False, + redact: "Callable[[str], str] | None" = None, ) -> None: """Convenience: log a governance (Level 1 ∩ Level 2) decision. @@ -856,15 +857,21 @@ def log_governance_decision( proceed unaudited. Without it the write is enqueued to the background writer, which swallows persistence failures — fine for best-effort audits (e.g. an ungoverned allow) but NOT for audit-or-deny. + + *redact* replaces ``redact_via_context`` for a process that installs no + platform context (the MCP gateway daemon), where that shim raises: the + caller passes the redaction of the companion it composed itself, so the + broader pass still applies. Omitted everywhere else. """ # Lazy import: sel.py is imported very early (security.py imports it), and # platform.context imports security indirectly — keep this off the # module-load path to avoid a cycle (documented pattern). from kiro_crew.platform.context import redact_via_context - safe_operation = redact_via_context(tool_name) - safe_item = redact_via_context(item) if item else "" - safe_reason = redact_via_context(reason) if reason else "" + _redact = redact or redact_via_context + safe_operation = _redact(tool_name) + safe_item = _redact(item) if item else "" + safe_reason = _redact(reason) if reason else "" self.log( SecurityEvent( event_id=uuid.uuid4().hex[:16], diff --git a/test/metrics/test_mcp_gateway_gauges.py b/test/metrics/test_mcp_gateway_gauges.py index 6523cbc3d6a..e3f5e0ac8de 100644 --- a/test/metrics/test_mcp_gateway_gauges.py +++ b/test/metrics/test_mcp_gateway_gauges.py @@ -409,7 +409,8 @@ def test_compose_telemetry_provider_applies_the_boot_security_floor(gatewayd_env gatewayd, edition, composed, readers, set_config, _ = gatewayd_env set_config(True) cfg = KiroCrewConfig.load() - assert bootstrap.compose_telemetry_provider(cfg) is edition + # Wrapped only to carry the companion's ceiling; the provider is the edition's. + assert bootstrap.compose_telemetry_provider(cfg)._inner is edition def _weak(profile, cfg, *, entry_points=None, policy=None): base = build_default_context(cfg) diff --git a/test/metrics/test_provider_otlp_destinations.py b/test/metrics/test_provider_otlp_destinations.py index 2c5430f5eb4..8878598c1a8 100644 --- a/test/metrics/test_provider_otlp_destinations.py +++ b/test/metrics/test_provider_otlp_destinations.py @@ -170,6 +170,9 @@ class _StubExporter: def __init__(self, **kwargs): captured.update(kwargs) + def export(self, *args, **kwargs): # the core wraps it with the policy check + return None + mod = types.ModuleType("opentelemetry.exporter.otlp.proto.http.metric_exporter") mod.OTLPMetricExporter = _StubExporter # type: ignore[attr-defined] monkeypatch.setitem( @@ -217,6 +220,9 @@ class _StubExporter: def __init__(self, **kwargs): captured.update(kwargs) + def export(self, *args, **kwargs): # the core wraps it with the policy check + return None + mod = types.ModuleType("opentelemetry.exporter.otlp.proto.http.metric_exporter") mod.OTLPMetricExporter = _StubExporter # type: ignore[attr-defined] monkeypatch.setitem( diff --git a/test/test_governance_list_otlp_hosts.py b/test/test_governance_list_otlp_hosts.py new file mode 100644 index 00000000000..badd0096c1b --- /dev/null +++ b/test/test_governance_list_otlp_hosts.py @@ -0,0 +1,473 @@ +"""``capabilities.telemetry`` ``otlp_hosts``: which OTLP collectors metrics may reach. + +RFC crew-enterprise-edition, appendix row "Diagnostic telemetry target". The key +parses with the catalog's matcher; a list that names X permits X and denies Y; a +gate pinned off denies every host; no policy, or no list, permits as before. Then +the call sites: the reader build (audited), the per-export gate, the local-only +enable gate (unfiltered on purpose) and the context-less MCP gateway daemon. +""" + +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest + +from kiro_crew.platform.governance import CapabilityGate, parse_policy, resolve + +#: (scope, inner key, matcher, an item the list allows, an item it does not) +_CASES = ( + ( + "capabilities.telemetry", + "otlp_hosts", + "host", + "otel.corp.example", + "evil.example", + ), +) +_IDS = [f"{scope}.{key}" for scope, key, *_ in _CASES] + + +def _policy(scope: str, body: dict) -> dict: + return { + "version": 1, + "boot": {"fail_closed": True}, + "capabilities": {scope.split(".", 1)[1]: body}, + } + + +def _allow(scope: str, key: str, allowed: str) -> dict: + return _policy(scope, {"scopes": {key: {"mode": "allow", "allow": [allowed]}}}) + + +@pytest.fixture +def isolated_home(tmp_path, monkeypatch): + monkeypatch.setenv("KIROCREW_HOME", str(tmp_path)) + return tmp_path + + +@pytest.fixture +def fake_sel(monkeypatch): + import kiro_crew.sel as sel_mod + + fake = MagicMock() + monkeypatch.setattr(sel_mod, "sel", lambda: fake) + return fake + + +def _install_policy(monkeypatch, doc: dict | None) -> None: + from kiro_crew.metrics import provider + from kiro_crew.platform import context as pc + + ceiling = parse_policy(doc) if doc is not None else None + ctx = SimpleNamespace(governance=ceiling) + monkeypatch.setattr(pc, "current_context", lambda: ctx) + # ``metrics.provider`` binds the name at import, so it is patched there too. + monkeypatch.setattr(provider, "current_context", lambda: ctx) + + +# ── The keys, through the real evaluator ──────────────────────────────── + + +class TestListKeys: + @pytest.mark.parametrize(("scope", "key", "matcher", "yes", "no"), _CASES, ids=_IDS) + def test_each_key_parses_with_its_catalog_matcher(self, scope, key, matcher, yes, no) -> None: + gate = parse_policy(_allow(scope, key, yes)).get(scope) + assert isinstance(gate, CapabilityGate) + assert gate.enabled is True # every one of these rows defaults on + assert gate.scopes[key].matcher == matcher + + @pytest.mark.parametrize(("scope", "key", "matcher", "yes", "no"), _CASES, ids=_IDS) + def test_the_list_permits_what_it_names_and_denies_the_rest( + self, scope, key, matcher, yes, no + ) -> None: + ceiling = parse_policy(_allow(scope, key, yes)) + assert resolve(ceiling, None, scope, f"{key}:{yes}").permitted is True + assert resolve(ceiling, None, scope, f"{key}:{no}").permitted is False + + @pytest.mark.parametrize(("scope", "key", "matcher", "yes", "no"), _CASES, ids=_IDS) + def test_a_gate_pinned_off_denies_every_item(self, scope, key, matcher, yes, no) -> None: + ceiling = parse_policy(_policy(scope, {"enabled": False})) + assert resolve(ceiling, None, scope, f"{key}:{yes}").permitted is False + + @pytest.mark.parametrize(("scope", "key", "matcher", "yes", "no"), _CASES, ids=_IDS) + def test_no_policy_or_no_list_permits_as_before(self, scope, key, matcher, yes, no) -> None: + assert resolve(None, None, scope, f"{key}:{no}").permitted is True + ceiling = parse_policy(_policy(scope, {"enabled": True})) + assert resolve(ceiling, None, scope, f"{key}:{no}").permitted is True + + def test_a_host_glob_binds_subdomains(self) -> None: + ceiling = parse_policy(_allow("capabilities.telemetry", "otlp_hosts", "*.corp.example")) + scope = "capabilities.telemetry" + assert resolve(ceiling, None, scope, "otlp_hosts:otel.corp.example").permitted is True + assert resolve(ceiling, None, scope, "otlp_hosts:corp.example.evil").permitted is False + + +# ── The call sites ────────────────────────────────────────────────────── + + +def _dest(name: str, endpoint: str): + return SimpleNamespace(name=name, endpoint=endpoint, signals=frozenset({"metrics"})) + + +#: A telemetry config with no local endpoint, as the provider path reads it. +_CFG = SimpleNamespace(otlp_endpoint="") + + +def _provider_with(monkeypatch, dests, **extra): + """Point ``metrics.provider`` at a telemetry provider that supplies *dests*.""" + from kiro_crew.metrics import provider + + prov = SimpleNamespace(otlp_destinations=lambda cfg: list(dests), **extra) + monkeypatch.setattr(provider, "_telemetry_provider", lambda: prov) + return provider + + +class TestTelemetryDestinations: + _SCOPE = "capabilities.telemetry" + _DESTS = ( + _dest("corp", "https://user:pw@otel.corp.example:4318/v1/metrics"), + _dest("other", "https://evil.example/v1/metrics"), + ) + + def test_only_listed_collector_hosts_get_a_reader_and_each_is_audited( + self, fake_sel, monkeypatch + ) -> None: + provider = _provider_with(monkeypatch, self._DESTS) + _install_policy(monkeypatch, _allow(self._SCOPE, "otlp_hosts", "otel.corp.example")) + assert [d.name for d in provider._otlp_destinations(_CFG)] == ["corp"] + rows = [c[1] for c in fake_sel.log_governance_decision.call_args_list] + assert [(r["item"], r["outcome"], r["layer"]) for r in rows] == [ + ("otlp_hosts:otel.corp.example", "allowed", "policy"), + ("otlp_hosts:evil.example", "denied", "policy"), + ] + + def test_a_gate_pinned_off_builds_no_reader(self, fake_sel, monkeypatch) -> None: + provider = _provider_with(monkeypatch, self._DESTS) + _install_policy(monkeypatch, _policy(self._SCOPE, {"enabled": False})) + assert provider._otlp_destinations(_CFG) == () + + def test_no_policy_keeps_every_destination(self, fake_sel, monkeypatch) -> None: + provider = _provider_with(monkeypatch, self._DESTS) + _install_policy(monkeypatch, None) + assert provider._otlp_destinations(_CFG) == self._DESTS + assert fake_sel.log_governance_decision.call_count == 0 + + def test_the_enable_gate_still_reports_egress_for_a_refused_host( + self, fake_sel, monkeypatch + ) -> None: + """``otlp_egress_active`` answers whether enabling collection could ever + send metrics away; a host refused today may be permitted by a later policy, + so the local-only enable switch must still be refused.""" + provider = _provider_with(monkeypatch, self._DESTS[1:]) + _install_policy(monkeypatch, _policy(self._SCOPE, {"enabled": False})) + assert provider.otlp_egress_active(_CFG) is True + assert fake_sel.log_governance_decision.call_count == 0 + + def test_a_process_without_a_context_uses_the_companion_ceiling( + self, fake_sel, monkeypatch + ) -> None: + """The MCP gateway daemon installs no context; the provider the bootstrap + composes for it carries the companion's ceiling, for the build and for + every export.""" + from kiro_crew.platform import context as pc + from kiro_crew.platform.bootstrap import _PolicyBoundTelemetry + + def _raise(): + raise pc.PlatformCompositionError("no context") + + provider = _provider_with(monkeypatch, ()) + monkeypatch.setattr(provider, "current_context", _raise) + inner = SimpleNamespace(otlp_destinations=lambda cfg: list(self._DESTS)) + ceiling = parse_policy(_allow(self._SCOPE, "otlp_hosts", "otel.corp.example")) + bound = _PolicyBoundTelemetry(inner, ceiling) + monkeypatch.setattr(provider, "_telemetry_provider", lambda: bound) + assert provider._policy_ceiling() is ceiling + assert [d.name for d in provider._otlp_destinations(_CFG)] == ["corp"] + # The inspection read is the companion's unfiltered set. + assert list(bound.otlp_destinations(_CFG)) == list(self._DESTS) + + def test_a_process_without_a_context_still_writes_the_audit_row(self, monkeypatch) -> None: + """With no context the SEL's own redaction shim raises; the row goes + through the composed companion's redaction instead of being dropped.""" + from kiro_crew import sel as sel_mod + from kiro_crew.platform import context as pc + from kiro_crew.platform.bootstrap import _PolicyBoundTelemetry + + def _raise(): + raise pc.PlatformCompositionError("no context") + + provider = _provider_with(monkeypatch, ()) + monkeypatch.setattr(provider, "current_context", _raise) + monkeypatch.setattr(pc, "current_context", _raise) + redacted: list[str] = [] + + def _redact(text: str) -> str: + redacted.append(text) + return text + + ceiling = parse_policy(_allow(self._SCOPE, "otlp_hosts", "otel.corp.example")) + bound = _PolicyBoundTelemetry( + SimpleNamespace(otlp_destinations=lambda cfg: list(self._DESTS)), + ceiling, + _redact, + ) + monkeypatch.setattr(provider, "_telemetry_provider", lambda: bound) + log = MagicMock() + real = sel_mod.SecurityEventLog.log_governance_decision + monkeypatch.setattr( + sel_mod, + "sel", + lambda: SimpleNamespace( + log_governance_decision=lambda **kw: real(SimpleNamespace(log=log), **kw) + ), + ) + assert [d.name for d in provider._otlp_destinations(_CFG)] == ["corp"] + events = [c[0][0] for c in log.call_args_list] + assert [(e.resources, e.outcome) for e in events] == [ + ("otlp_hosts:otel.corp.example", "allowed"), + ("otlp_hosts:evil.example", "denied"), + ] + assert "otlp_hosts:evil.example" in redacted + + def test_a_running_exporter_stops_when_a_new_policy_denies_its_host( + self, fake_sel, monkeypatch + ) -> None: + """A policy installed at runtime binds the next export, the shutdown flush + included, without a rebuild; the change is audited once.""" + pytest.importorskip("opentelemetry.sdk.metrics.export") + from opentelemetry.sdk.metrics.export import MetricExportResult + + from kiro_crew.metrics import otlp_governance + + sent: list[object] = [] + exporter = SimpleNamespace(export=lambda data, **kw: sent.append(data) or "sent") + ceiling = {"now": parse_policy(_allow(self._SCOPE, "otlp_hosts", "otel.corp.example"))} + otlp_governance.gate_exporter( + exporter, "https://otel.corp.example/v1/metrics", lambda: ceiling["now"] + ) + + assert exporter.export("batch-1") == "sent" + ceiling["now"] = parse_policy(_allow(self._SCOPE, "otlp_hosts", "other.example")) + assert exporter.export("batch-2") is MetricExportResult.FAILURE + assert exporter.export("batch-3") is MetricExportResult.FAILURE + assert sent == ["batch-1"] + rows = [c[1] for c in fake_sel.log_governance_decision.call_args_list] + assert [(r["item"], r["outcome"]) for r in rows] == [ + ("otlp_hosts:otel.corp.example", "denied") + ] + + +class TestEndpointReadings: + """The host is asked in every reading a client could take, so a parser + disagreement cannot reach a host the list refused.""" + + _SCOPE = "capabilities.telemetry" + + @pytest.mark.parametrize( + "endpoint", + [ + "http://evil.example\\@otel.corp.example/v1/metrics", + "https://outside.example\\@otel.corp.example/v1/metrics", + ], + ) + def test_a_backslash_authority_is_refused(self, fake_sel, monkeypatch, endpoint) -> None: + provider = _provider_with(monkeypatch, [_dest("trick", endpoint)]) + _install_policy(monkeypatch, _allow(self._SCOPE, "otlp_hosts", "otel.corp.example")) + assert provider._otlp_destinations(_CFG) == () + + def test_a_root_dot_cannot_dodge_a_deny(self, fake_sel, monkeypatch) -> None: + provider = _provider_with(monkeypatch, [_dest("dot", "https://evil.example./v1/metrics")]) + _install_policy( + monkeypatch, + _policy( + self._SCOPE, {"scopes": {"otlp_hosts": {"mode": "deny", "deny": ["evil.example"]}}} + ), + ) + assert provider._otlp_destinations(_CFG) == () + + @pytest.mark.parametrize( + "endpoint", + [ + "https://stra\u00dfe.example/v1/metrics", # requests IDNA-encodes it + "https://%6Ftel.corp.example.evil/v1/metrics", # urllib percent-decodes it + ], + ) + def test_a_host_the_transport_reads_differently_is_refused( + self, fake_sel, monkeypatch, endpoint + ) -> None: + provider = _provider_with(monkeypatch, [_dest("trick", endpoint)]) + _install_policy(monkeypatch, _allow(self._SCOPE, "otlp_hosts", "strasse.example")) + assert provider._otlp_destinations(_CFG) == () + + def test_the_ordinary_endpoint_is_still_permitted(self, fake_sel, monkeypatch) -> None: + provider = _provider_with( + monkeypatch, [_dest("corp", "https://user:pw@OTEL.corp.example.:4318/v1/metrics")] + ) + _install_policy(monkeypatch, _allow(self._SCOPE, "otlp_hosts", "otel.corp.example")) + assert [d.name for d in provider._otlp_destinations(_CFG)] == ["corp"] + + @pytest.mark.parametrize( + ("endpoint", "deny"), + [ + ("http://167772165:4318/v1/metrics", "10.0.0.5"), # decimal IPv4 + ("http://0x0a000005:4318/v1/metrics", "10.0.0.5"), # hex IPv4 + ("http://[::ffff:10.0.0.5]:4318/v1/metrics", "10.0.0.5"), # v4-mapped + ("http://[fe80::1%25eth0]:4318/v1/metrics", "fe80::*"), # scoped IPv6 + ("http://[2001:db8:0:0:0:0:0:1]:4318/v1/metrics", "2001:db8::1"), # long IPv6 + ("https://b\u00fccher.example/v1/metrics", "xn--bcher-*"), # sent as xn-- + ("https://stra\u00dfe.example/v1/metrics", "strasse.example"), # urllib's idna + ("https://fa%C3%9F.example/v1/metrics", "fass.example"), # decoded, then idna + ], + ids=[ + "decimal", + "hex", + "v4-mapped", + "ipv6-zone", + "long-ipv6", + "unicode", + "sharp-s", + "pct-sharp-s", + ], + ) + def test_a_deny_names_every_spelling_of_its_host( + self, fake_sel, monkeypatch, endpoint, deny + ) -> None: + provider = _provider_with(monkeypatch, [_dest("d", endpoint)]) + doc = _policy(self._SCOPE, {"scopes": {"otlp_hosts": {"mode": "deny", "deny": [deny]}}}) + _install_policy(monkeypatch, doc) + assert provider._otlp_destinations(_CFG) == () + + def test_the_shared_host_matcher_is_unchanged(self) -> None: + """``network.egress`` shares the matcher, so these lists change nothing there: + a Unicode entry still matches only the Unicode spelling.""" + from kiro_crew.platform.governance import _match_host + + assert _match_host("b\u00fccher.example", "b\u00fc*.example") + assert not _match_host("xn--bcher-kva.example", "b\u00fccher.example") + + def test_userinfo_is_never_read_as_a_host(self, fake_sel, monkeypatch) -> None: + """An encoded ``#`` or ``/`` in a password must not cut the authority.""" + provider = _provider_with( + monkeypatch, + [ + _dest("a", "https://tok%23en@otel.corp.example/v1/metrics"), + _dest("b", "https://user:p%2Fsecret@otel.corp.example:4318/v1/metrics"), + ], + ) + _install_policy(monkeypatch, _allow(self._SCOPE, "otlp_hosts", "otel.corp.example")) + assert [d.name for d in provider._otlp_destinations(_CFG)] == ["a", "b"] + items = {c[1]["item"] for c in fake_sel.log_governance_decision.call_args_list} + assert items == {"otlp_hosts:otel.corp.example"} + + def test_an_idna_name_is_asked_in_its_ascii_form(self, fake_sel, monkeypatch) -> None: + provider = _provider_with(monkeypatch, [_dest("b", "https://b\u00fccher.example/v1")]) + _install_policy(monkeypatch, _allow(self._SCOPE, "otlp_hosts", "xn--bcher-kva.example")) + assert [d.name for d in provider._otlp_destinations(_CFG)] == ["b"] + + +class TestUnevaluableCeiling: + def test_no_context_and_no_carried_ceiling_refuses_every_host( + self, fake_sel, monkeypatch + ) -> None: + """A daemon whose policy could not be loaded must not export as if ungoverned.""" + from kiro_crew.platform import context as pc + + def _raise(): + raise pc.PlatformCompositionError("policy unreadable") + + provider = _provider_with(monkeypatch, [_dest("a", "https://otel.corp.example/v1")]) + monkeypatch.setattr(provider, "current_context", _raise) + assert provider._otlp_destinations(_CFG) == () + rows = [c[1] for c in fake_sel.log_governance_decision.call_args_list] + assert [r["outcome"] for r in rows] == ["denied"] + assert "fail-closed" in rows[0]["reason"] + + +class TestUnmatchableEntriesAreRefused: + """An entry no host reading can match fails the policy load, instead of a + deny that refuses nothing or an allow that admits nothing.""" + + _SCOPE = "capabilities.telemetry" + + @pytest.mark.parametrize("mode", ["allow", "deny"]) + @pytest.mark.parametrize( + "entry", + [ + "b\u00fccher.example", + "b\u00fc*.example", + "otel.corp.example.", + "2001:db8:0:0:0:0:0:1", + "010.0.0.5", + "fe80::1%eth0", + "[2001:db8::1]", + "ev%69l.example", + "otel.corp.example:4318", + "ev%69l*", + "*.ev%69l.example", + "fe80::1%eth*", + "evil.example:*", + "*@evil.example", + "evil.example/*", + "[ab]vil.example", + ], + ids=[ + "unicode", + "unicode-glob", + "root-dot", + "long-ipv6", + "octal-ipv4", + "zoned-ipv6", + "bracketed-ipv6", + "percent", + "port", + "glob-percent", + "glob-percent-label", + "glob-zone", + "glob-port", + "glob-userinfo", + "glob-path", + "glob-class", + ], + ) + def test_the_policy_fails_to_load(self, mode, entry) -> None: + from kiro_crew.platform.context import PlatformCompositionError + + doc = _policy(self._SCOPE, {"scopes": {"otlp_hosts": {"mode": mode, mode: [entry]}}}) + with pytest.raises(PlatformCompositionError, match=r"otlp_hosts"): + parse_policy(doc) + + @pytest.mark.parametrize( + "entry", + [ + "xn--bcher-kva.example", + "*.corp.example", + "otel-?.corp.example", + "2001:db8::1", + "10.0.0.5", + "10.0.0.*", + "fe80::*", + "fd00:0:1::*", + "2001:db8:10:0:*", + "01-collector*.corp.example", + "*.0x.org", + "*.007agent.example", + "*_svc.corp.example", + "*", + "10*", + "*:1:0:0:1", + ], + ) + def test_a_matchable_entry_loads(self, entry) -> None: + doc = _policy(self._SCOPE, {"scopes": {"otlp_hosts": {"mode": "deny", "deny": [entry]}}}) + assert parse_policy(doc).get(self._SCOPE) is not None + + def test_network_egress_entries_are_not_checked(self) -> None: + """Top-level ``network.egress`` keeps its own parsing.""" + doc = { + "version": 1, + "boot": {"fail_closed": True}, + "network": {"egress": {"mode": "deny", "deny": ["b\u00fccher.example"]}}, + } + assert parse_policy(doc).get("network.egress") is not None diff --git a/test/test_sel_bindings.py b/test/test_sel_bindings.py index 650edabb0e6..89c528310c1 100644 --- a/test/test_sel_bindings.py +++ b/test/test_sel_bindings.py @@ -95,7 +95,8 @@ "flush": ("function", False, "a9c45e0c08a5", "sel"), "log": ("function", False, "105f9e23f712", "sel"), "log_api_access": ("function", False, "205f9b11a348", "sel"), - "log_governance_decision": ("function", False, "416a40eec8a0", "sel"), + # `redact` added for the context-less MCP gateway daemon (OTLP host audit). + "log_governance_decision": ("function", False, "f3c3785395e6", "sel"), "log_governance_degraded": ("function", False, "9b129b9270d1", "sel"), "log_tool_invocation": ("function", False, "28f3ece4bc30", "sel"), "prune": ("function", False, "254f3f4fcb91", "sel_rotation"), From ee3eec40f82e29dfdc86397f621c0a5947aa6d07 Mon Sep 17 00:00:00 2001 From: Joe Guo Date: Sun, 25 Oct 2026 13:37:34 +0000 Subject: [PATCH 2/2] fix(governance): close host-list bypasses found in review - url_policy_hosts asks every ASCII reading as read, so a host with a character a DNS name does not use (clip+one.blocked.example) is still refused by a deny glob; only non-ASCII readings are dropped. - CapabilityGate.from_dict refuses a declared scope list in the wrong shape (a bare array or string) with PlatformCompositionError instead of dropping it and leaving that list unbounded. - otlp_governance asks resolve even with no ceiling, at build and at each export, so an evaluator answer with no policy is not skipped. The build audits every decision, a no-policy grant included. - The load-time entry check uses the characters a host reading can hold (matching._PLAIN_HOST_RE), so an entry such as clip+*.blocked.example loads; '@', '/', a backslash or a space still fail the load. Refs #23208 --- src/kiro_crew/metrics/otlp_governance.py | 19 +++--- .../platform/governance_parts/archetypes.py | 31 +++++++--- .../platform/governance_parts/matching.py | 23 ++++--- test/test_governance_list_otlp_hosts.py | 61 ++++++++++++++++--- 4 files changed, 97 insertions(+), 37 deletions(-) diff --git a/src/kiro_crew/metrics/otlp_governance.py b/src/kiro_crew/metrics/otlp_governance.py index 67fd96375a3..d1f0370c9c7 100644 --- a/src/kiro_crew/metrics/otlp_governance.py +++ b/src/kiro_crew/metrics/otlp_governance.py @@ -14,8 +14,9 @@ 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, or a ceiling without the row or -the ruleset, permits every host. An evaluation error denies (egress fails closed; +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. @@ -112,11 +113,10 @@ def governed_destinations( *audit* is for the build that acts on the answer. """ - kept = list(supplied) - if ceiling is None: - return tuple(kept) allowed = [] - for dest in kept: + 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) @@ -153,12 +153,7 @@ def gate_exporter( last = {"permitted": True} def export(metrics_data: Any, *args: Any, **kwargs: Any) -> Any: - current = ceiling() - permitted, host, rule, reason = ( - (True, hosts[0] if hosts else "", "", "") - if current is None - else _decide(current, hosts) - ) + permitted, host, rule, reason = _decide(ceiling(), hosts) if permitted != last["permitted"]: last["permitted"] = permitted _audit(host, permitted, rule, reason, redact) diff --git a/src/kiro_crew/platform/governance_parts/archetypes.py b/src/kiro_crew/platform/governance_parts/archetypes.py index 91cc9090f2e..95f5b1ab964 100644 --- a/src/kiro_crew/platform/governance_parts/archetypes.py +++ b/src/kiro_crew/platform/governance_parts/archetypes.py @@ -394,21 +394,23 @@ def is_at_least_as_strict_as(self, ceiling: "OrdinalControl") -> bool: def _unmatchable_glob(glob: str) -> str: """Why a host glob can match no reading, or ``""`` when it might. - A character check only: a reading is a lower-case ASCII name or a canonical IP - literal, so a glob holding a ``[`` class, ``%``, ``@``, ``/``, a zone, or a - ``:`` outside a hex IPv6 glob can never match. How an IP glob spells its + A character check only: a reading holds only a URL host's characters + (``matching._PLAIN_HOST_RE``), so a glob holding a ``[`` class, ``@``, ``/``, + ``\\``, whitespace, a zone, or a ``:`` outside a hex IPv6 glob can never match. How an IP glob spells its numbers is not checked: it matches canonical readings as written (an alternate numeric spelling is tracked in the deferral issue for these lists). """ import re + from kiro_crew.platform.governance_parts.matching import _PLAIN_HOST_RE + if "[" in glob: return "uses a [ ] character class; host lists take * and ? only" body = glob.replace("*", "").replace("?", "") if ":" in body and not re.fullmatch(r"[0-9a-f:]*", body): return "is not an IPv6 glob in canonical form (hex groups only, no zone or port)" - if not re.fullmatch(r"[a-z0-9._:-]*", body): - return "has a character no host is asked with ('%', '@', '/' or similar)" + if body and not _PLAIN_HOST_RE.fullmatch(body): + return "has a character no host reading holds ('@', '/', a backslash or a space)" return "" @@ -417,9 +419,11 @@ def _refuse_unmatchable_host_entries(scope_name: str, ruleset: "ScopedRuleset") A capability's ``host`` lists are asked about hosts as clients send them (``matching.url_policy_hosts``): ASCII, no root dot, an IP literal in canonical - form. An entry in another spelling would match nothing, so a deny would refuse - nothing and an allow would admit nothing, silently. It fails to load instead, - and the message says which form to write. Top-level ``network.egress`` keeps its + form. An exact entry in another spelling would match nothing, so a deny would + refuse nothing and an allow would admit nothing, silently. It fails to load + instead, and the message says which form to write. A glob gets a character + check only (``_unmatchable_glob``); how it spells IP numbers is not checked. + Top-level ``network.egress`` keeps its own parsing; this applies only to lists inside a capability. """ from kiro_crew.platform.governance_parts.matching import _PLAIN_HOST_RE, _canonical_ip @@ -447,7 +451,7 @@ def _refuse_unmatchable_host_entries(scope_name: str, ruleset: "ScopedRuleset") elif ":" in spelled or not _PLAIN_HOST_RE.fullmatch(spelled): why = ( "is not a host as clients send it " - "(brackets, a zone, a port, '%', '/' or '@')" + "(brackets, a zone, a port, '/', '@', a backslash or a space)" ) if why: # The position, never the value: an entry can be a pasted URL. @@ -483,6 +487,15 @@ def from_dict( if not isinstance(raw_scopes, dict): raise PlatformCompositionError("CapabilityGate.scopes must be an object") matchers = scope_matchers or {} + for k, v in raw_scopes.items(): + # A declared list in the wrong shape would otherwise be dropped below, + # leaving the capability with no bound on that list at all. + if str(k) in matchers and not isinstance(v, dict): + raise PlatformCompositionError( + f"CapabilityGate.scopes.{k} must be an object such as " + '{"mode": "allow", "allow": [...]}, ' + f"got {type(v).__name__}" + ) scopes: Dict[str, RulesetLike] = { str(k): ScopedRuleset.from_dict(v, matcher=matchers.get(str(k), _DEFAULT_MATCHER)) for k, v in raw_scopes.items() diff --git a/src/kiro_crew/platform/governance_parts/matching.py b/src/kiro_crew/platform/governance_parts/matching.py index c121097fed8..7be83247de0 100644 --- a/src/kiro_crew/platform/governance_parts/matching.py +++ b/src/kiro_crew/platform/governance_parts/matching.py @@ -286,9 +286,10 @@ def url_policy_hosts(url: object) -> Tuple[str, ...]: Each reading is lower-cased and loses any IPv6 zone and terminal DNS root dot, so entries are written without the dot. An IP literal is asked in canonical form only (dotted IPv4, compressed IPv6), so an IP entry names a host in that form. A - reading that is not a plain ASCII name is not asked, because no client sends - it as written; the clients that encode it supply the ``xn--`` or decoded form - they do send. A URL with no plain reading at all is asked as ``""``, which + non-ASCII reading is not asked, because no client sends it as written; the + clients that encode it supply the ``xn--`` form they do send. Every other + ASCII reading is asked as read, even one with characters a DNS name does not + use, so a deny glob still sees it. A URL with no ASCII reading is asked as ``""``, which matches no entry: an allow list refuses it and a deny list does not. So, as for ``network.egress``, an allow list is the hard bound, entries name ASCII (``xn--``) hosts, and a deny list refuses the spellings it can name. @@ -351,10 +352,13 @@ def _keep(host: str) -> None: # resolver accepts (``167772165``, ``0x0a000005``, ``::ffff:10.0.0.5``). _keep(ip) continue - if not _PLAIN_HOST_RE.fullmatch(bare): - # Not a plain ASCII name: no client sends it as written. The clients - # that IDNA-encode or percent-decode it contribute their own reading. + if not bare.isascii(): + # Not ASCII: no client sends it as written. The clients that + # IDNA-encode it contribute their own reading. continue + # Any other ASCII reading is asked as read, even an odd one + # (``clip+one.blocked.example``): a resolver may still answer for it, so + # a deny glob must see it, and an allow list refuses it. _keep(bare) if not hosts: # No reading is a host a client would send: ``""``, which an allow list @@ -375,6 +379,7 @@ def _canonical_ip(host: str) -> str: return "" -#: A host as a client sends it: ASCII letters, digits, hyphens and dots, or an IP -#: literal's hex digits and colons. -_PLAIN_HOST_RE = re.compile(r"[a-z0-9._:-]+") +#: The characters a host reading can hold: a URL host's own characters (RFC 3986 +#: unreserved, sub-delims and ``%``), lower-cased, plus an IP literal's colons. +#: ``url_policy_hosts`` asks every ASCII reading, so a list entry is held to these. +_PLAIN_HOST_RE = re.compile(r"[a-z0-9._~!$&'()*+,;=%:-]+") diff --git a/test/test_governance_list_otlp_hosts.py b/test/test_governance_list_otlp_hosts.py index badd0096c1b..d3cbca3d634 100644 --- a/test/test_governance_list_otlp_hosts.py +++ b/test/test_governance_list_otlp_hosts.py @@ -152,7 +152,12 @@ def test_no_policy_keeps_every_destination(self, fake_sel, monkeypatch) -> None: provider = _provider_with(monkeypatch, self._DESTS) _install_policy(monkeypatch, None) assert provider._otlp_destinations(_CFG) == self._DESTS - assert fake_sel.log_governance_decision.call_count == 0 + # A no-policy grant is still an egress grant, so the build audits it. + rows = [c[1] for c in fake_sel.log_governance_decision.call_args_list] + assert [(r["item"], r["outcome"]) for r in rows] == [ + ("otlp_hosts:otel.corp.example", "allowed"), + ("otlp_hosts:evil.example", "allowed"), + ] def test_the_enable_gate_still_reports_egress_for_a_refused_host( self, fake_sel, monkeypatch @@ -319,6 +324,9 @@ def test_the_ordinary_endpoint_is_still_permitted(self, fake_sel, monkeypatch) - ("https://b\u00fccher.example/v1/metrics", "xn--bcher-*"), # sent as xn-- ("https://stra\u00dfe.example/v1/metrics", "strasse.example"), # urllib's idna ("https://fa%C3%9F.example/v1/metrics", "fass.example"), # decoded, then idna + ("https://clip+one.blocked.example/v1/metrics", "*.blocked.example"), # odd ASCII + ("https://clip+one.blocked.example/v1/metrics", "clip+*.blocked.example"), + ("https://clip+one.blocked.example/v1/metrics", "clip+one.blocked.example"), ], ids=[ "decimal", @@ -329,6 +337,9 @@ def test_the_ordinary_endpoint_is_still_permitted(self, fake_sel, monkeypatch) - "unicode", "sharp-s", "pct-sharp-s", + "odd-ascii", + "odd-ascii-glob", + "odd-ascii-exact", ], ) def test_a_deny_names_every_spelling_of_its_host( @@ -385,6 +396,39 @@ def _raise(): assert "fail-closed" in rows[0]["reason"] +class TestWrongShape: + @pytest.mark.parametrize("value", [["otel.corp.example"], "otel.corp.example", 1, None]) + def test_a_host_list_in_the_wrong_shape_fails_to_load(self, value) -> None: + """A malformed host list must not load as no list, which permits every host.""" + from kiro_crew.platform.context import PlatformCompositionError + + with pytest.raises(PlatformCompositionError, match="scopes.otlp_hosts must be an object"): + parse_policy(_policy("capabilities.telemetry", {"scopes": {"otlp_hosts": value}})) + + +class TestNoCeilingAsksTheEvaluator: + def test_no_ceiling_follows_what_resolve_answers(self, fake_sel, monkeypatch) -> None: + """With no policy, the build and each export take ``resolve``'s answer.""" + from types import SimpleNamespace + + from kiro_crew.metrics import otlp_governance as og + + dest = SimpleNamespace(name="d", endpoint="https://otel.corp.example/v1/metrics") + assert og.governed_destinations([dest], None) == (dest,) + + def _deny(*_a, **_k): + return SimpleNamespace(permitted=False, rule="baseline", reason="denied") + + monkeypatch.setattr("kiro_crew.platform.governance.resolve", _deny) + assert og.governed_destinations([dest], None) == () + + sent = [] + exporter = SimpleNamespace(export=lambda data, *a, **k: sent.append(data) or "OK") + og.gate_exporter(exporter, dest.endpoint, lambda: None) + exporter.export("batch") + assert sent == [] + + class TestUnmatchableEntriesAreRefused: """An entry no host reading can match fails the policy load, instead of a deny that refuses nothing or an allow that admits nothing.""" @@ -402,15 +446,14 @@ class TestUnmatchableEntriesAreRefused: "010.0.0.5", "fe80::1%eth0", "[2001:db8::1]", - "ev%69l.example", "otel.corp.example:4318", - "ev%69l*", - "*.ev%69l.example", "fe80::1%eth*", "evil.example:*", "*@evil.example", "evil.example/*", "[ab]vil.example", + "ev il.example", + "ev\\il*", ], ids=[ "unicode", @@ -420,15 +463,14 @@ class TestUnmatchableEntriesAreRefused: "octal-ipv4", "zoned-ipv6", "bracketed-ipv6", - "percent", "port", - "glob-percent", - "glob-percent-label", "glob-zone", "glob-port", "glob-userinfo", "glob-path", "glob-class", + "space", + "glob-backslash", ], ) def test_the_policy_fails_to_load(self, mode, entry) -> None: @@ -457,6 +499,11 @@ def test_the_policy_fails_to_load(self, mode, entry) -> None: "*", "10*", "*:1:0:0:1", + "clip+one.blocked.example", + "clip+*.blocked.example", + "ev%69l.example", + "*.ev%69l.example", + "a!*.blocked.example", ], ) def test_a_matchable_entry_loads(self, entry) -> None: