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..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..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..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/test/test_governance_list_feature_video_hosts.py b/test/test_governance_list_feature_video_hosts.py new file mode 100644 index 00000000000..819be7e9840 --- /dev/null +++ b/test/test_governance_list_feature_video_hosts.py @@ -0,0 +1,287 @@ +"""``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 + + @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 + + 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