diff --git a/docs/system-specs/modules/governance.md b/docs/system-specs/modules/governance.md index 063cfa788e2..8eef08a3222 100644 --- a/docs/system-specs/modules/governance.md +++ b/docs/system-specs/modules/governance.md @@ -3251,6 +3251,12 @@ seam. A policy pins the seam off MACHINE-WIDE with one object -- every surface, {"version": 1, "capabilities": {"decisions": {"enabled": false}}} ``` +**Which models, points and endpoints — three lists.** The row carries inner `models` (`identifier`), `points` (`identifier`) and `endpoints` (`host`) rulesets (RFC [crew-enterprise-edition](../../request-for-change/rfc-crew-enterprise-edition.md), appendix rows "Decision models", "Decision kinds" and "Decision endpoints"). `decide` asks `capability.targets_denied` once the lane and model are known and after the scrub (`scrub_reason`), so a model id the scrub refuses never reaches the SEL row: `points:` always, and for the Jev lane, the one that sends to a decision provider, `models:` and `endpoints:` for every reading of the provider endpoint'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). The judge's LLM lane sends to the session's own model provider through the session runner, which picks the served model itself, so it is asked about its point only; `models` and `endpoints` do not bind it. Each item is evaluated and audited, never short-circuited; a refusal returns `None` and writes no decision row, like the capability refusal. The lists bind local presets too, as `capabilities.decisions` already does. A policy without these rulesets permits as before. + +```json +{"version": 1, "capabilities": {"decisions": {"scopes": {"models": {"mode": "allow", "allow": ["jev-latest"]}, "points": {"mode": "deny", "deny": ["tool.risk"]}}}}} +``` + ### "Share as image" — `capabilities.social_share` The chat's *More actions → Share as image* turns an assistant reply into a branded diff --git a/src/kiro_crew/decisions/capability.py b/src/kiro_crew/decisions/capability.py index 1808850db50..71037c9458a 100644 --- a/src/kiro_crew/decisions/capability.py +++ b/src/kiro_crew/decisions/capability.py @@ -220,12 +220,13 @@ def _answer_generation() -> int | None: return None -def _row_denied(surface_key: str, scope: str) -> bool: - """Evaluate and audit ONE capability row. Fail-closed; see :func:`is_decisions_denied`.""" +def _row_denied(surface_key: str, scope: str, item: str = "") -> bool: + """Evaluate and audit ONE capability row or list item. Fail-closed; see + :func:`is_decisions_denied`.""" from kiro_crew.edition_locks import UNNAMED_REASON, follows_policy, locked_off try: - if scope == DECISIONS_SCOPE and locked_off(): + if scope == DECISIONS_SCOPE and not item and locked_off(): # Enterprise: content sent to a decision provider follows this hosted # row and is off by default (RFC enterprise edition, Red "Content sent # to external decisions"). Evaluated without the seam's own audit, so @@ -240,7 +241,7 @@ def _row_denied(surface_key: str, scope: str) -> bool: return denied decision = vet_and_audit( scope, - "", + item, session_key=surface_key, tool_name=AUDIT_TOOL, log_warning=False, @@ -257,6 +258,45 @@ def _row_denied(surface_key: str, scope: str) -> bool: return not getattr(decision, "permitted", False) +def targets_denied( + surface_key: str = DASHBOARD_SURFACE_KEY, + *, + point: str, + model: str | None, + endpoint: str | None, +) -> bool: + """Whether the ceiling's lists refuse this request's point, model or endpoint. + + Asks :data:`DECISIONS_SCOPE` for ``points:`` and, when the lane sends a + *model* to a decision provider at an *endpoint*, ``models:`` and + ``endpoints:`` for every reading of its host. A ruleset the + policy does not name permits, so a ceiling without these lists answers exactly + as before. Every item is evaluated and audited, never short-circuited, so which + rows appear does not depend on another item's answer. Fail-closed, like + :func:`is_decisions_denied`. Blocking: call off the event loop. + """ + items = [f"points:{point}"] + if model is not None: + items.append(f"models:{model}") + if endpoint is not None: + # Every host a client could read the endpoint as, so a parser + # disagreement or a root-dot spelling cannot reach a refused host. + from kiro_crew.platform.governance_parts.matching import url_policy_hosts + + items.extend(f"endpoints:{host}" for host in url_policy_hosts(endpoint)) + # One request's items must be answered by ONE policy: a swap between two of + # them could pair one policy's permit for the model with another's permit for + # the endpoint. Bracketed as in :func:`denied_sides`, failing closed if it moved. + _prime_profiles() + before = _answer_generation() + denied = [_row_denied(surface_key, DECISIONS_SCOPE, item) for item in items] + if not any(denied) and (before is None or _answer_generation() != before): + logger.debug("decisions governance answer moved mid-evaluation; denying the request") + _audit_unevaluable(surface_key, DECISIONS_SCOPE) + return True + return any(denied) + + def _audit_unevaluable(surface_key: str, scope: str = DECISIONS_SCOPE) -> None: """Best-effort SEL record for the path ``vet_and_audit`` never reached. Never raises.""" _audit_row(surface_key, scope, "denied", _UNEVALUABLE_REASON) diff --git a/src/kiro_crew/decisions/gate.py b/src/kiro_crew/decisions/gate.py index 607e39bded1..68b9b9bf542 100644 --- a/src/kiro_crew/decisions/gate.py +++ b/src/kiro_crew/decisions/gate.py @@ -338,6 +338,32 @@ def _capability_denied(session_key: str | None, *, local: bool = False) -> bool: return True +def _targets_denied( + config: Any | None, session_key: str | None, point: str, model: str, lane: str +) -> bool: + """Whether the ceiling's ``points`` / ``models`` / ``endpoints`` lists refuse this + request. Filesystem IO; never raises (an error denies). + + The ``models`` and ``endpoints`` lists bind the decision PROVIDER, which only + the Jev lane uses. The judge's LLM lane sends to the session's own model + provider through the session runner, which picks the served model itself, so + it is asked about its point only. + """ + from kiro_crew.decisions.capability import DASHBOARD_SURFACE_KEY, targets_denied + + try: + jev = lane == LANE_JEV + return targets_denied( + session_key or DASHBOARD_SURFACE_KEY, + point=point, + model=model if jev else None, + endpoint=configured_endpoint(config) if jev else None, + ) + except Exception as exc: + logger.debug("decisions: %s target check failed (%s)", point, type(exc).__name__) + return True + + def _consented_for( config: Any | None, session_key: str | None = None, @@ -1137,10 +1163,18 @@ async def _write( # For every lane but the judge's LLM one this is ``provider.model`` with the # same fallback ``JevOracle`` applies, unchanged. model = lane_model(cfg, lane=lane) + # The scrub runs first: the list check below writes the model id to the SEL, + # so an id the scrub refuses (a pasted credential) never reaches that row. refusal = scrub_reason(state, questions, model=model) if refusal is not None: await _write(latency_ms=0, answers=None, error=refusal) return None + # The fleet's lists: which points, models and endpoints the seam may use. Like + # the capability row in ``_consented_for``, a refusal here writes no decision + # row (its audit is the SEL ``governance_decision`` row) and costs nothing on + # an unconsented install, which never reaches this line. + if await asyncio.to_thread(_targets_denied, cfg, session_key, point, model, lane): + return None started = time.monotonic() oracle: Any = None 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..a7c69576345 100644 --- a/src/kiro_crew/platform/governance_parts/catalog.py +++ b/src/kiro_crew/platform/governance_parts/catalog.py @@ -391,7 +391,21 @@ class ScopeSpec: # ``decisions_enabled`` so the Feature Previews card is not drawn at all. # Data row only -- CONTRACT_VERSION and the evaluator are untouched (mirrors # social_share). - "capabilities.decisions": ScopeSpec(CAPABILITY, capability_default=True), + # + # Three inner rulesets bound WHAT the seam may use once it is on, each queried + # by ``decide`` before a request is sent (``capability.targets_denied``): + # ``models`` (``identifier``: the model id the request names), ``points`` + # (``identifier``: the decision point, e.g. ``skills.select``) and + # ``endpoints`` (``host``: the provider endpoint's host). ``models`` and + # ``endpoints`` bind the decision provider, so the judge's LLM lane, which + # runs on the session's own model provider, is asked about its point only. + # They bind local presets too, because a ``capabilities.decisions`` answer + # already does. + "capabilities.decisions": ScopeSpec( + CAPABILITY, + capability_default=True, + scope_matchers={"models": "identifier", "points": "identifier", "endpoints": "host"}, + ), # The same seam answered by a LOCAL PRESET model on this machine # (``decisions/local_models.py``) instead of hosted Jev. Nothing leaves the # machine and nothing is spent, so a fleet that permits hosted Jev can still 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_decisions_gate.py b/test/test_decisions_gate.py index a13abac9e79..8887afff83d 100644 --- a/test/test_decisions_gate.py +++ b/test/test_decisions_gate.py @@ -1531,3 +1531,166 @@ async def test_metadata_does_not_ride_a_failure_row(self, install_impl, log_home row = log_home()[0] assert row["error"] == gate_mod.ERROR_PROVIDER assert "provider_model" not in row + + +class TestDecisionLists: + """``capabilities.decisions`` ``models`` / ``points`` / ``endpoints`` lists, at + ``decide`` (RFC crew-enterprise-edition, appendix: decision models, decision + kinds, decision endpoints). Real policy, real evaluator; only SEL is faked.""" + + @pytest.fixture(autouse=True) + def quiet(self, _floor_monkeypatch): + from unittest.mock import MagicMock + + import kiro_crew.sel as sel_mod + + _floor_monkeypatch.setattr(gate_mod, "_capability_denied_warned", False) + self.sel = MagicMock() + _floor_monkeypatch.setattr(sel_mod, "sel", lambda: self.sel) + + @staticmethod + def _policy(monkeypatch, body: dict | None) -> None: + from kiro_crew.platform import context as pc + from kiro_crew.platform.governance import parse_policy + + doc = None + if body is not None: + doc = {"version": 1, "boot": {"fail_closed": True}, "capabilities": {"decisions": body}} + ceiling = parse_policy(doc) if doc is not None else None + monkeypatch.setattr(pc, "current_context", lambda: SimpleNamespace(governance=ceiling)) + + @staticmethod + def _allow(**lists: list[str]) -> dict: + return {"scopes": {k: {"mode": "allow", "allow": v} for k, v in lists.items()}} + + def _send(self, install_impl) -> tuple[object, _RecordingOracle]: + oracle = install_impl(_RecordingOracle()) + return asyncio.run(decide(POINT, "hi", QUESTIONS, config=_config())), oracle + + def test_no_policy_sends_as_before(self, install_impl, log_home, monkeypatch): + self._policy(monkeypatch, None) + answers, oracle = self._send(install_impl) + assert answers is not None and len(oracle.calls) == 1 + + @pytest.mark.parametrize( + "lists", + [ + {"points": [POINT]}, + {"models": [gate_mod._DEFAULT_MODEL]}, + {"endpoints": ["api.typesafe.ai"]}, + ], + ids=["points", "models", "endpoints"], + ) + def test_a_list_naming_this_request_sends(self, install_impl, log_home, monkeypatch, lists): + self._policy(monkeypatch, self._allow(**lists)) + answers, oracle = self._send(install_impl) + assert answers is not None and len(oracle.calls) == 1 + + @pytest.mark.parametrize( + "lists", + [ + {"points": ["tool.risk"]}, + {"models": ["other-model"]}, + {"endpoints": ["jev.corp.example"]}, + ], + ids=["points", "models", "endpoints"], + ) + def test_a_list_not_naming_it_sends_nothing_and_writes_no_row( + self, install_impl, log_home, monkeypatch, lists + ): + self._policy(monkeypatch, self._allow(**lists)) + oracle = install_impl(_ExplodingOracle()) + assert asyncio.run(decide(POINT, "hi", QUESTIONS, config=_config())) is None + assert oracle.entered is False + assert log_home() == [] + items = [c[1]["item"] for c in self.sel.log_governance_decision.call_args_list] + assert {f"points:{POINT}", f"models:{gate_mod._DEFAULT_MODEL}"} <= set(items) + assert "endpoints:api.typesafe.ai" in items + + def test_a_credential_shaped_model_never_reaches_the_audit( + self, install_impl, log_home, monkeypatch + ): + """The scrub refuses the model id before the list check writes it to SEL.""" + model = "sk-" + "0123456789abcdef" * 2 + self._policy(monkeypatch, self._allow(models=["other-model"])) + oracle = install_impl(_ExplodingOracle()) + assert asyncio.run(decide(POINT, "hi", QUESTIONS, config=_config(model=model))) is None + assert oracle.entered is False + for call in self.sel.mock_calls: + assert model not in repr(call) + + def test_a_gate_pinned_off_sends_nothing(self, install_impl, log_home, monkeypatch): + self._policy(monkeypatch, {"enabled": False}) + oracle = install_impl(_ExplodingOracle()) + assert asyncio.run(decide(POINT, "hi", QUESTIONS, config=_config())) is None + assert oracle.entered is False + + def test_the_llm_lane_is_asked_about_its_point_only(self, monkeypatch): + from kiro_crew.decisions import capability + + seen: list[dict] = [] + + def _spy(surface_key=capability.DASHBOARD_SURFACE_KEY, **kw): + seen.append(kw) + return False + + monkeypatch.setattr(capability, "targets_denied", _spy) + assert gate_mod._targets_denied(_config(), None, POINT, "m", gate_mod.LANE_LLM) is False + assert seen == [{"point": POINT, "model": None, "endpoint": None}] + + def test_an_error_in_the_check_denies(self, monkeypatch): + from kiro_crew.decisions import capability + + def _boom(*_a, **_k): + raise RuntimeError("boom") + + monkeypatch.setattr(capability, "targets_denied", _boom) + assert gate_mod._targets_denied(_config(), None, POINT, "m", gate_mod.LANE_JEV) is True + + def test_a_policy_swap_between_items_denies(self, monkeypatch): + """Each item permits under some policy, but the answers span two, so the + request is refused and the refusal is audited.""" + from kiro_crew.decisions import capability + + self._policy(monkeypatch, None) + generations = iter([1, 2]) + monkeypatch.setattr(capability, "_answer_generation", lambda: next(generations)) + assert ( + capability.targets_denied(point=POINT, model="m", endpoint="https://api.typesafe.ai/x") + is True + ) + rows = [c[1] for c in self.sel.log_governance_decision.call_args_list] + assert rows[-1]["outcome"] == "denied" and "fail-closed" in rows[-1]["reason"] + + @pytest.mark.parametrize( + "endpoint", + [ + "https://jev.corp.example./x", + "https://jev.corp.example\\@api.typesafe.ai/x", + "https://jev\u3002corp.example/x", # aiohttp/yarl reads U+3002 as a dot + "https://%6aev.corp.example/x", # urllib percent-decodes the authority + "https://j\u00e9v.corp.example/x", # sent as its xn-- form, which the list names + ], + ids=["root-dot", "backslash", "ideographic-dot", "percent", "unicode"], + ) + def test_an_endpoint_spelling_cannot_dodge_a_deny(self, monkeypatch, endpoint): + from kiro_crew.decisions import capability + + self._policy( + monkeypatch, + { + "scopes": { + "endpoints": { + "mode": "deny", + "deny": ["jev.corp.example", "xn--jv-bja.corp.example", "10.0.0.5"], + } + } + }, + ) + assert capability.targets_denied(point=POINT, model="m", endpoint=endpoint) is True + # An IP literal is refused in every spelling the resolver accepts. + assert capability.targets_denied(point=POINT, model="m", endpoint="http://167772165/v1") + assert ( + capability.targets_denied(point=POINT, model="m", endpoint="https://api.typesafe.ai/x") + is False + ) diff --git a/test/test_governance_list_decisions.py b/test/test_governance_list_decisions.py new file mode 100644 index 00000000000..5f759a50946 --- /dev/null +++ b/test/test_governance_list_decisions.py @@ -0,0 +1,128 @@ +"""``capabilities.decisions`` ``models`` / ``points`` / ``endpoints``. + +RFC crew-enterprise-edition, appendix rows "Decision models", "Decision kinds" and +"Decision endpoints". Each key parses with the catalog's matcher; a list that names +X permits X and denies Y; a gate pinned off denies every item; no policy, or no +list, permits as before. The call site, ``decide``, is pinned in +``test_decisions_gate.py`` (``TestDecisionLists``). +""" + +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.decisions", "models", "identifier", "jev-1", "other-model"), + ("capabilities.decisions", "points", "identifier", "skills.select", "tool.risk"), + ("capabilities.decisions", "endpoints", "host", "jev.corp.example", "api.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 + + +class TestHostReadings: + """An ASCII host with a character a DNS name does not use is still asked as + read, so a deny glob over its domain refuses it.""" + + def test_an_odd_ascii_host_is_asked_as_read(self) -> None: + from kiro_crew.platform.governance import url_policy_hosts + + hosts = url_policy_hosts("https://clip+one.blocked.example/v1") + assert "clip+one.blocked.example" in hosts + assert "" not in hosts + + def test_a_deny_glob_refuses_an_odd_ascii_host(self) -> None: + from kiro_crew.platform.governance import url_policy_hosts + + scope = "capabilities.decisions" + doc = _policy( + scope, {"scopes": {"endpoints": {"mode": "deny", "deny": ["*.blocked.example"]}}} + ) + ceiling = parse_policy(doc) + hosts = url_policy_hosts("https://clip+one.blocked.example/v1") + assert any(not resolve(ceiling, None, scope, f"endpoints:{h}").permitted for h in hosts) + + +class TestWrongShape: + @pytest.mark.parametrize("key", ["models", "points", "endpoints"]) + @pytest.mark.parametrize("value", [["jev-1"], "jev-1", 1, None]) + def test_a_list_in_the_wrong_shape_fails_to_load(self, key, value) -> None: + """A malformed list must not load as no list, which permits every item.""" + from kiro_crew.platform.context import PlatformCompositionError + + with pytest.raises(PlatformCompositionError, match=f"scopes.{key} must be an object"): + parse_policy(_policy("capabilities.decisions", {"scopes": {key: value}}))