From 84937570c55118720279cd08e3a1191f495b07ad Mon Sep 17 00:00:00 2001 From: Joe Guo Date: Sat, 24 Oct 2026 08:50:59 +0000 Subject: [PATCH 1/2] feat(governance): let a policy name the feature content hosts --- docs/system-specs/modules/governance.md | 2 + src/kiro_crew/asset_downloader.py | 43 ++- src/kiro_crew/docs/feature-videos.md | 3 + src/kiro_crew/feature_videos_cache.py | 17 +- src/kiro_crew/feature_videos_manifest.py | 48 +++- src/kiro_crew/platform/governance.py | 5 + .../platform/governance_parts/archetypes.py | 68 +++++ .../platform/governance_parts/catalog.py | 9 +- .../platform/governance_parts/matching.py | 110 ++++++++ ...est_governance_list_feature_video_hosts.py | 264 ++++++++++++++++++ 10 files changed, 561 insertions(+), 8 deletions(-) create mode 100644 test/test_governance_list_feature_video_hosts.py diff --git a/docs/system-specs/modules/governance.md b/docs/system-specs/modules/governance.md index 063cfa788e2..3d73a8133fb 100644 --- a/docs/system-specs/modules/governance.md +++ b/docs/system-specs/modules/governance.md @@ -3155,6 +3155,8 @@ is served from this origin — a bundled asset or a downloaded, verified one — policy needs no off-origin media host, and a header that admitted one would be admitting a fetch the server never offers. +**Which hosts — the `hosts` list.** An inner `hosts` ruleset (`host` matcher) names the hosts the manifest and media may come from, a mirror named by `KIROCREW_FEATURE_VIDEOS_MANIFEST_URL` included (RFC [crew-enterprise-edition](../../request-for-change/rfc-crew-enterprise-edition.md), appendix row "Feature content hosts"). `feature_videos_manifest.host_denied(url)` asks `hosts:` for every reading of the url'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) on the same pinned surface, audited and fail-closed, permitting only when every reading is permitted, before each manifest request in the walk and before each poster and each clip request in the download pass (`feature_videos_cache.HOST_DENIED` fails that entry only). The manifest fetch also holds every redirect hop to the list (`asset_downloader.build_opener(redirect_allowed=...)`), since an operator mirror may redirect across hosts; clip and poster transfers already refuse any cross-host hop. It does not touch the `download_denied` memo: one refused host does not report the feature off. A policy without the ruleset permits every host. + ### The Jev decision seam — `capabilities.decisions` An enabled, sampled session sends message excerpts and skill descriptions to an diff --git a/src/kiro_crew/asset_downloader.py b/src/kiro_crew/asset_downloader.py index b6a710a6cb7..183f21c06c1 100644 --- a/src/kiro_crew/asset_downloader.py +++ b/src/kiro_crew/asset_downloader.py @@ -189,9 +189,30 @@ def __init__(self, origin: str, target: str, policy: str) -> None: " redirects across hosts is honoured only when set through the environment override)" ) _HTTPS_ONLY_POLICY = "the transfer stays on https" +_CALLER_POLICY = "the caller's policy does not permit the redirect target" -class _SameHostRedirectHandler(urllib.request.HTTPRedirectHandler): +class _GuardedRedirect(urllib.request.HTTPRedirectHandler): + """A redirect handler that also asks the caller's own predicate about each hop. + + ``redirect_allowed`` is ``None`` (no extra rule) unless :func:`build_opener` + was given one; a hop it refuses raises :class:`RedirectRefused`. + """ + + redirect_allowed: "Callable[[str], bool] | None" = None + + def _check_caller_guard(self, req, newurl: str) -> None: # noqa: ANN001 + guard = self.redirect_allowed + if guard is not None and not guard(newurl): + logger.warning( + "refusing a redirect the caller's policy does not permit from %s to %s", + redact_url(req.full_url), + redact_url(newurl), + ) + raise RedirectRefused(req.full_url, newurl, _CALLER_POLICY) + + +class _SameHostRedirectHandler(_GuardedRedirect): """Refuse any redirect that leaves https or changes the host. The authorization to make a request is granted per HOST — a manifest's signed @@ -229,10 +250,11 @@ def redirect_request(self, req, fp, code, msg, headers, newurl): # noqa: ANN001 redact_url(newurl), ) raise RedirectRefused(req.full_url, newurl, _SAME_HOST_POLICY) + self._check_caller_guard(req, newurl) return super().redirect_request(req, fp, code, msg, headers, newurl) -class _HttpsOnlyRedirectHandler(urllib.request.HTTPRedirectHandler): +class _HttpsOnlyRedirectHandler(_GuardedRedirect): """Refuse a redirect that leaves https; allow one that changes the host. The relaxed policy, reachable only through ``allow_cross_host_redirects``, @@ -262,11 +284,15 @@ def redirect_request(self, req, fp, code, msg, headers, newurl): # noqa: ANN001 redact_url(newurl), ) raise RedirectRefused(req.full_url, newurl, _HTTPS_ONLY_POLICY) + self._check_caller_guard(req, newurl) return super().redirect_request(req, fp, code, msg, headers, newurl) def build_opener( - context: "ssl.SSLContext | None" = None, *, allow_cross_host_redirects: bool = False + context: "ssl.SSLContext | None" = None, + *, + allow_cross_host_redirects: bool = False, + redirect_allowed: "Callable[[str], bool] | None" = None, ) -> urllib.request.OpenerDirector: """An opener that verifies TLS and pins redirects. @@ -280,8 +306,17 @@ def build_opener( place of :class:`_SameHostRedirectHandler`. It is for a url the operator set in the process environment, or one written into the source as a constant, and nothing else — see the module docstring. + + *redirect_allowed*, when given, is asked about every redirect target the + handler would otherwise follow; ``False`` refuses that hop. It only narrows. """ - handler = _HttpsOnlyRedirectHandler if allow_cross_host_redirects else _SameHostRedirectHandler + handler_cls = ( + _HttpsOnlyRedirectHandler if allow_cross_host_redirects else _SameHostRedirectHandler + ) + handler: "type[_GuardedRedirect] | _GuardedRedirect" = handler_cls + if redirect_allowed is not None: + handler = handler_cls() + handler.redirect_allowed = redirect_allowed return urllib.request.build_opener( urllib.request.HTTPSHandler(context=context or make_ssl_context()), handler ) diff --git a/src/kiro_crew/docs/feature-videos.md b/src/kiro_crew/docs/feature-videos.md index 9d1e7aad325..5a42dd55817 100644 --- a/src/kiro_crew/docs/feature-videos.md +++ b/src/kiro_crew/docs/feature-videos.md @@ -102,6 +102,9 @@ Two independent switches: no clip is downloaded, and nothing streams from the CDN — clips already on disk still play. The settings panel reads the answer from `GET /api/feature-videos/status` (`download_enabled`), so the controls reflect it rather than failing when pressed. +- The same scope can instead name the hosts the catalog and clips may come from + (`"scopes": {"hosts": {"mode": "allow", "allow": ["mirror.example.com"]}}`). + A request to any other host is not made. To fetch the catalog from somewhere else, set `KIROCREW_FEATURE_VIDEOS_MANIFEST_URL` in the gateway's environment to point at your own copy of the manifest. It must be diff --git a/src/kiro_crew/feature_videos_cache.py b/src/kiro_crew/feature_videos_cache.py index dd89de88dfd..541bf962efa 100644 --- a/src/kiro_crew/feature_videos_cache.py +++ b/src/kiro_crew/feature_videos_cache.py @@ -299,6 +299,10 @@ def cache_ceiling_bytes(max_mb: float) -> int: #: ``_download_entry`` keeps its ``(ok, error)`` shape. DENIED_MID_ENTRY = "download denied by the governance ceiling" +#: The error for an entry whose poster or clip host the policy's ``hosts`` list +#: does not permit. The pass goes on to the next entry, which may name another host. +HOST_DENIED = "host not permitted by the governance ceiling" + #: Failure reason for an entry whose media landed but whose receipt could not be #: written. The media stays on disk; without the receipt it reads as not cached, #: so the next pass re-fetches it and tries the receipt again. @@ -602,10 +606,16 @@ def _download_entry_into( *folder* is held open for the whole entry, so the directory the poster landed in is the directory the clip lands in and the receipts are written to — no step between them re-resolves the folder's name. + + Each request is first held to the policy's ``hosts`` list + (:func:`manifest_mod.host_denied`); a refused host fails this entry only. """ + poster_url = current.asset_url(entry.poster) + if manifest_mod.host_denied(poster_url): + return False, HOST_DENIED poster_ok, poster_err = asset_downloader.download_to( folder.describe(entry.poster), - current.asset_url(entry.poster), + poster_url, sha256=entry.poster_sha256, # A bound, not a declared length: the manifest carries a poster sha but # no poster byte count, and max_bytes caps the transfer without @@ -628,9 +638,12 @@ def _download_entry_into( # the poster's permit. if manifest_mod.download_denied(): return False, DENIED_MID_ENTRY + clip_url = current.asset_url(entry.file) + if manifest_mod.host_denied(clip_url): + return False, HOST_DENIED clip_ok, clip_err = asset_downloader.download_to( folder.describe(entry.file), - current.asset_url(entry.file), + clip_url, sha256=entry.sha256, size=entry.bytes, resume=True, diff --git a/src/kiro_crew/feature_videos_manifest.py b/src/kiro_crew/feature_videos_manifest.py index f49b8ef2145..e943fcffc29 100644 --- a/src/kiro_crew/feature_videos_manifest.py +++ b/src/kiro_crew/feature_videos_manifest.py @@ -274,6 +274,42 @@ def download_permitted_cached() -> bool: return not download_denied() +def host_denied(url: str) -> bool: + """Whether the ceiling forbids fetching *url*'s host. Blocking, audited, fail-closed. + + Asks :data:`DOWNLOAD_SCOPE` for ``hosts:``, so a policy can name the exact + CDN or mirror hosts. A gate pinned off denies every host; a policy with no + ``hosts`` ruleset permits every one, as before the list existed. Called next to + :func:`download_denied` before each manifest, poster and clip request, and does + not touch its memo: one refused host does not turn the whole feature off. + + Every host a client could read *url* as is asked (``url_policy_hosts``), and + the request is permitted only when all are, so a parser disagreement cannot + reach a host the list refused. + """ + from kiro_crew.platform.governance_parts.matching import url_policy_hosts + + try: + decisions = [ + _governance_profiles.vet_and_audit( + DOWNLOAD_SCOPE, + f"hosts:{host}", + session_key=DASHBOARD_SURFACE_KEY, + tool_name=AUDIT_TOOL, + log_warning=False, + fail_closed=True, + ) + for host in url_policy_hosts(url) + ] + except Exception as exc: + # The type only: the url can carry a credential, and a traceback would + # render the exception's own text, which may quote it. + logger.debug("feature-video host governance probe failed (%s); denying", type(exc).__name__) + _audit_unevaluable() + return True + return not all(getattr(d, "permitted", False) for d in decisions) + + def _audit_unevaluable() -> None: """Best-effort SEL record for the path ``vet_and_audit`` never reached.""" try: @@ -1028,7 +1064,12 @@ def _fetch_json(url: str, *, operator_url: bool = False) -> "dict | None": # build_opener, never urlopen: the url was authorized against ONE host, and # urlopen's default handler would follow a cross-host redirect and spend # that authorization on a destination the CDN chose. - opener = asset_downloader.build_opener(allow_cross_host_redirects=operator_url) + # Every hop is held to the policy's ``hosts`` list as the first url was, so + # an operator mirror that redirects cannot reach a host the list omits. + opener = asset_downloader.build_opener( + allow_cross_host_redirects=operator_url, + redirect_allowed=lambda target: not host_denied(target), + ) # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected -- manifest_url enforces https://, redirects are pinned (host-pinned unless the operator's own env url), and the document is signature-verified before use with opener.open(request, timeout=_MANIFEST_TIMEOUT_SECS) as resp: body = resp.read(_MANIFEST_MAX_BYTES + 1) @@ -1164,6 +1205,10 @@ def fetch_manifest(version: str) -> "tuple[VideoManifest | None, dict]": if download_denied(): logger.info("feature-video manifest fetch stopped: downloading withdrawn mid-walk") return None, {} + if host_denied(url): + # Every candidate shares one host, so the walk ends here too. + logger.info("feature-video manifest fetch stopped: the policy does not permit its host") + return None, {} raw = _fetch_json(url, operator_url=overridden) if raw is None: continue @@ -1199,6 +1244,7 @@ def fetch_manifest(version: str) -> "tuple[VideoManifest | None, dict]": "checked_cache_root", "download_denied", "download_permitted_cached", + "host_denied", "ensure_cache_dir", "fetch_manifest", "is_safe_media_name", 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..1592477eafb 100644 --- a/src/kiro_crew/platform/governance_parts/catalog.py +++ b/src/kiro_crew/platform/governance_parts/catalog.py @@ -369,7 +369,14 @@ class ScopeSpec: # on disk is a separate decision this row does not make. # Data row only — CONTRACT_VERSION and the evaluator are untouched (mirrors # social_share). - "capabilities.feature_videos_download": ScopeSpec(CAPABILITY, capability_default=True), + # + # Inner ``hosts`` ruleset (``host`` matcher): which hosts the manifest and + # its media may be fetched from (a mirror named by the URL override included). + # Queried as ``hosts:`` before each manifest request and each clip or + # poster request (``feature_videos_manifest.host_denied``). + "capabilities.feature_videos_download": ScopeSpec( + CAPABILITY, capability_default=True, scope_matchers={"hosts": "host"} + ), # The Jev decision seam: an enabled, sampled session sends message excerpts and # skill descriptions to an external, PAID provider endpoint and spends the # operator's account to do it. The keystone ``decisions_consent.json`` is 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/test/test_governance_list_feature_video_hosts.py b/test/test_governance_list_feature_video_hosts.py new file mode 100644 index 00000000000..f91703919c2 --- /dev/null +++ b/test/test_governance_list_feature_video_hosts.py @@ -0,0 +1,264 @@ +"""``capabilities.feature_videos_download`` ``hosts``: where feature content may come from. + +RFC crew-enterprise-edition, appendix row "Feature content hosts". 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 manifest walk, a mirror's redirect hops, and each clip request. +""" + +from __future__ import annotations + +import urllib.request +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.feature_videos_download", + "hosts", + "host", + "mirror.corp.example", + "d3j0sthz5doyui.cloudfront.net", + ), +) +_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 + + +# ── The call sites ────────────────────────────────────────────────────── + + +class TestFeatureVideoHosts: + _SCOPE = "capabilities.feature_videos_download" + + def test_host_denied_follows_the_list(self, isolated_home, fake_sel, monkeypatch) -> None: + from kiro_crew import feature_videos_manifest as fvm + + _install_policy(monkeypatch, _allow(self._SCOPE, "hosts", "mirror.corp.example")) + assert fvm.host_denied("https://mirror.corp.example/fv/manifest.json") is False + assert fvm.host_denied(f"{fvm.DEFAULT_CDN_BASE}/0.9.0/manifest.json") is True + items = [c[1]["item"] for c in fake_sel.log_governance_decision.call_args_list] + assert items == [ + "hosts:mirror.corp.example", + "hosts:d3j0sthz5doyui.cloudfront.net", + ] + + def test_host_denied_permits_without_a_policy(self, isolated_home, fake_sel, monkeypatch): + from kiro_crew import feature_videos_manifest as fvm + + _install_policy(monkeypatch, None) + assert fvm.host_denied(f"{fvm.DEFAULT_CDN_BASE}/0.9.0/manifest.json") is False + + def test_the_manifest_walk_makes_no_request_to_an_unlisted_host( + self, isolated_home, fake_sel, monkeypatch + ) -> None: + from kiro_crew import feature_videos_manifest as fvm + + monkeypatch.delenv(fvm.MANIFEST_URL_ENV, raising=False) + _install_policy(monkeypatch, _allow(self._SCOPE, "hosts", "mirror.corp.example")) + fetched: list[str] = [] + monkeypatch.setattr(fvm, "_fetch_json", lambda url, **_k: fetched.append(url)) + assert fvm.fetch_manifest("0.9.3") == (None, {}) + assert fetched == [] + + def test_a_mirror_redirect_to_an_unlisted_host_is_refused( + self, isolated_home, fake_sel, monkeypatch + ) -> None: + """The operator-mirror fetch may follow a cross-host redirect; each hop is + held to the list like the first url.""" + from kiro_crew import asset_downloader + from kiro_crew import feature_videos_manifest as fvm + + _install_policy(monkeypatch, _allow(self._SCOPE, "hosts", "mirror.corp.example")) + seen: dict = {} + real = asset_downloader.build_opener + + def _capture(*a, **kw): + seen.update(kw) + raise OSError("no network in tests") + + monkeypatch.setattr(asset_downloader, "build_opener", _capture) + assert fvm._fetch_json("https://mirror.corp.example/m.json", operator_url=True) is None + assert seen["allow_cross_host_redirects"] is True + guard = seen["redirect_allowed"] + assert guard("https://mirror.corp.example/b/m.json") is True + assert guard("https://bucket.storage.example/m.json") is False + + handler = next( + h + for h in real(allow_cross_host_redirects=True, redirect_allowed=guard).handlers + if isinstance(h, asset_downloader._GuardedRedirect) + ) + req = urllib.request.Request("https://mirror.corp.example/m.json") + with pytest.raises(asset_downloader.RedirectRefused): + handler.redirect_request( + req, None, 302, "Found", {}, "https://bucket.storage.example/x" + ) + assert ( + handler.redirect_request(req, None, 302, "Found", {}, "https://mirror.corp.example/y") + is not None + ) + + def test_a_clip_on_an_unlisted_host_is_not_downloaded( + self, isolated_home, fake_sel, monkeypatch + ) -> None: + from kiro_crew import asset_downloader + from kiro_crew import feature_videos_cache as fvc + from kiro_crew import feature_videos_manifest as fvm + + _install_policy(monkeypatch, _allow(self._SCOPE, "hosts", "mirror.corp.example")) + + def _never(*_a, **_k): + raise AssertionError("an unlisted host must not be fetched") + + monkeypatch.setattr(asset_downloader, "download_to", _never) + entry = SimpleNamespace(id="a", poster="a.jpg", file="a.mp4") + current = fvm.VideoManifest( + release="0.9.0", cdn_base=fvm.DEFAULT_CDN_BASE, generated_at="", entries=() + ) + ok, err = fvc.FeatureVideoCache()._download_entry_into(MagicMock(), entry, current, 0) + assert (ok, err) == (False, fvc.HOST_DENIED) + + +class TestUrlReadings: + """Every reading of a url's host is asked, so parser disagreements and + root-dot spellings cannot reach a host the list refused.""" + + _SCOPE = "capabilities.feature_videos_download" + + def test_a_backslash_authority_is_refused(self, isolated_home, fake_sel, monkeypatch) -> None: + from kiro_crew import feature_videos_manifest as fvm + + _install_policy(monkeypatch, _allow(self._SCOPE, "hosts", "mirror.corp.example")) + assert fvm.host_denied("https://evil.example\\@mirror.corp.example/m.json") is True + + def test_a_root_dot_cannot_dodge_a_deny(self, isolated_home, fake_sel, monkeypatch) -> None: + from kiro_crew import feature_videos_manifest as fvm + + doc = _policy( + self._SCOPE, {"scopes": {"hosts": {"mode": "deny", "deny": ["evil.example"]}}} + ) + _install_policy(monkeypatch, doc) + assert fvm.host_denied("https://evil.example./m.json") is True + assert fvm.host_denied("https://mirror.corp.example/m.json") is False + + def test_a_percent_encoded_redirect_target_is_refused( + self, isolated_home, fake_sel, monkeypatch + ) -> None: + """urllib percent-decodes the authority, so the decoded host is asked too.""" + from kiro_crew import feature_videos_manifest as fvm + + doc = _policy( + self._SCOPE, {"scopes": {"hosts": {"mode": "deny", "deny": ["evil.example"]}}} + ) + _install_policy(monkeypatch, doc) + assert fvm.host_denied("https://%65vil.example/manifest.json") is True + + def test_a_probe_error_logs_no_url(self, isolated_home, fake_sel, monkeypatch, caplog) -> None: + import logging + + from kiro_crew import feature_videos_manifest as fvm + + def _boom(*_a, **_k): + raise ValueError("https://user:s3cret@mirror.corp.example/") + + monkeypatch.setattr(fvm._governance_profiles, "vet_and_audit", _boom) + with caplog.at_level(logging.DEBUG, logger=fvm.logger.name): + assert fvm.host_denied("https://user:s3cret@mirror.corp.example/m.json") is True + assert "s3cret" not in caplog.text + + +class TestUrllibIdna: + """The manifest and media transfers use urllib, which IDNA-encodes a non-ASCII + host with Python's own codec, so that is the host a deny names.""" + + _SCOPE = "capabilities.feature_videos_download" + + @pytest.mark.parametrize( + "url", + ["https://stra\u00dfe.example/m.json", "https://stra%C3%9Fe.example/m.json"], + ids=["sharp-s", "percent-encoded"], + ) + def test_a_deny_names_the_host_urllib_connects_to( + self, isolated_home, fake_sel, monkeypatch, url + ) -> None: + from kiro_crew import feature_videos_manifest as fvm + + doc = _policy( + self._SCOPE, {"scopes": {"hosts": {"mode": "deny", "deny": ["strasse.example"]}}} + ) + _install_policy(monkeypatch, doc) + assert fvm.host_denied(url) is True From 18f7598c49a37249d131c119ed2d8c1fbea30901 Mon Sep 17 00:00:00 2001 From: Joe Guo Date: Sun, 25 Oct 2026 13:55:14 +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. - 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 --- .../platform/governance_parts/archetypes.py | 31 +++++++++++++------ .../platform/governance_parts/matching.py | 23 ++++++++------ ...est_governance_list_feature_video_hosts.py | 23 ++++++++++++++ 3 files changed, 59 insertions(+), 18 deletions(-) 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_feature_video_hosts.py b/test/test_governance_list_feature_video_hosts.py index f91703919c2..819be7e9840 100644 --- a/test/test_governance_list_feature_video_hosts.py +++ b/test/test_governance_list_feature_video_hosts.py @@ -227,6 +227,29 @@ def test_a_percent_encoded_redirect_target_is_refused( _install_policy(monkeypatch, doc) assert fvm.host_denied("https://%65vil.example/manifest.json") is True + @pytest.mark.parametrize("value", [["mirror.example.com"], "mirror.example.com", 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 + from kiro_crew.platform.governance import parse_policy + + with pytest.raises(PlatformCompositionError, match="scopes.hosts must be an object"): + parse_policy(_policy(self._SCOPE, {"scopes": {"hosts": value}})) + + def test_an_odd_ascii_redirect_target_is_refused_by_a_deny_glob( + self, isolated_home, fake_sel, monkeypatch + ) -> None: + """A ``+`` is no DNS name character, but a resolver may still answer for it, + so the host is asked as read rather than dropped.""" + from kiro_crew import feature_videos_manifest as fvm + + doc = _policy( + self._SCOPE, {"scopes": {"hosts": {"mode": "deny", "deny": ["*.blocked.example"]}}} + ) + _install_policy(monkeypatch, doc) + assert fvm.host_denied("https://clip+one.blocked.example/manifest.json") is True + assert fvm.host_denied("https://mirror.corp.example/manifest.json") is False + def test_a_probe_error_logs_no_url(self, isolated_home, fake_sel, monkeypatch, caplog) -> None: import logging