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..d1f0370c9c7 --- /dev/null +++ b/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 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..95f5b1ab964 100644 --- a/src/kiro_crew/platform/governance_parts/archetypes.py +++ b/src/kiro_crew/platform/governance_parts/archetypes.py @@ -391,6 +391,75 @@ 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 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 body and not _PLAIN_HOST_RE.fullmatch(body): + return "has a character no host reading holds ('@', '/', a backslash or a space)" + 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 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 + + 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, '/', '@', a backslash or a space)" + ) + 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 # ────────────────────────────────────────────────────────────────────────── @@ -418,11 +487,23 @@ 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() 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..7be83247de0 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,117 @@ 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 + 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. + """ + 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 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 + # 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 "" + + +#: 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/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..d3cbca3d634 --- /dev/null +++ b/test/test_governance_list_otlp_hosts.py @@ -0,0 +1,520 @@ +"""``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 + # 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 + ) -> 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 + ("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", + "hex", + "v4-mapped", + "ipv6-zone", + "long-ipv6", + "unicode", + "sharp-s", + "pct-sharp-s", + "odd-ascii", + "odd-ascii-glob", + "odd-ascii-exact", + ], + ) + 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 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.""" + + _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]", + "otel.corp.example:4318", + "fe80::1%eth*", + "evil.example:*", + "*@evil.example", + "evil.example/*", + "[ab]vil.example", + "ev il.example", + "ev\\il*", + ], + ids=[ + "unicode", + "unicode-glob", + "root-dot", + "long-ipv6", + "octal-ipv4", + "zoned-ipv6", + "bracketed-ipv6", + "port", + "glob-zone", + "glob-port", + "glob-userinfo", + "glob-path", + "glob-class", + "space", + "glob-backslash", + ], + ) + 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", + "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: + 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"),