diff --git a/docs/openshell-support.md b/docs/openshell-support.md index f2117e94..48d6ed28 100644 --- a/docs/openshell-support.md +++ b/docs/openshell-support.md @@ -48,9 +48,53 @@ Defaults follow the pinned [OpenShell v0.1.2 authored schema](https://github.com and [conversion code](https://github.com/NVIDIA/OpenShell/blob/v0.1.2/crates/openshell-policy/src/lib.rs). The [upstream schema reference](https://docs.nvidia.com/openshell/how-it-works/policies/schema) describes runtime constraints beyond document inventory. This reader is not a -substitute for upstream policy validation. Semantic expansion/subset review, -Git dependency identity, gate integration, local composition and native proof -are separate implementation stages (#944–#948). +substitute for upstream policy validation. Git dependency identity, gate +integration, local composition and native proof are separate implementation +stages (#945–#948). + +## Conservative declared-authority comparison + +The shared host comparator reads normalized policy facts. It describes +declared authority, without establishing runtime enforcement or whether a +named file, executable or destination exists. JSON and Markdown use the same +before/after rows, expansion signals and explanations. `diff` publishes no +merge verdict. Mixed changes retain their proven expansions and narrowings; +their single overall direction remains unknown. + +| Surface | Supported direction proof | Unproven cases | +|---|---|---| +| Filesystem | Exact canonical absolute read/write path sets; read-write also grants read; duplicates are neutral | Omitted filesystem, changed workdir inclusion, ancestor overlap, noncanonical/wildcard paths, edits to runtime-baseline paths while network policy is present | +| Landlock | `hard_requirement` to explicit/omitted `best_effort` weakens the compatibility requirement; reverse strengthens it | Actual kernel support or applied rules | +| Process | Unchanged identity is neutral | Changed identity, image user/group resolution and driver defaults | +| Network L4 | Exact binary × hostname × port selection without address/transport/request options | Wildcards, DNS/IP reach, executable resolution, credential options and endpoint path routing | +| REST | Exact method/path matchers; method `*`; read-only/read-write/full presets; compatible overlapping allows with deny precedence | Request-path globs or omitted paths, query constraints, encoded-slash changes and protocol/credential options | +| Other protocols | Unchanged normalized facts are neutral | Changes to MCP, GraphQL, WebSocket, JSON-RPC or middleware semantics | + +For REST, a finite partition includes every literal method/path on either side +and an additional class for all other values. It evaluates the effective union +of matching allows minus matching denials, keeping each binary/destination +relationship. An unchanged covering grant makes an added grant redundant. +Each possible combination of up to eight exact binary selectors is evaluated +per destination, because a process may match both its own path and ancestor +paths. A denial or inspected rule can therefore affect grants from another +matching binary selector. +Removing one of two covering denials does not expand the allowed set. A matching +inspected rule suppresses request access from uninspected rules. Conflicting +enforcement modes in overlapping inspected endpoints are unproven. + +Removing `enforcement: enforce` restores upstream `audit`, which permits +well-formed requests that violate request rules. Malformed requests and runtime +transport checks are outside this comparison. A full-access endpoint without +denials already permits the compared request domain, so that transition alone +is neutral. Named rules, collection order, duplicates and explicit spelling of +an unchanged default are not authority. Adding/removing an entire selected +document remains unknown because it establishes no replacement runtime policy. + +Comparison stops with a named unknown result beyond eight exact binary selectors +per destination or 100,000 network reference +or request-partition cells. Unsupported semantics never become inferred safe +narrowing. An HTTP method or MCP tool name still supplies no business effect, +approval, argument restriction or deployed agent binding. ## Read limits and coverage diff --git a/src/agents_shipgate/core/capability_diff_rows.py b/src/agents_shipgate/core/capability_diff_rows.py index 04462c30..5473cd5a 100644 --- a/src/agents_shipgate/core/capability_diff_rows.py +++ b/src/agents_shipgate/core/capability_diff_rows.py @@ -50,6 +50,7 @@ step_action_key, ) from agents_shipgate.core.host_settings import rate_claude_setting, setting_value_text +from agents_shipgate.core.openshell_compare import compare_openshell_grants from agents_shipgate.core.permission_lattice import ( exec_equivalent_argument, permission_pairing_group, @@ -593,6 +594,11 @@ def _grant_value( if not grant: return ABSENT kind = str(grant.get("kind") or "") + if kind == "openshell_policy": + facts = grant["facts"] + policy = facts["policy"] + endpoints = sum(len(rule["endpoints"]) for rule in policy["network_policies"].values()) + return f"{facts['role']}, OpenShell {facts['runtime_version']}, {endpoints} endpoint(s), facts {grant['config_sha256']}" if kind == "permission_rule" and redact_permission_arguments: from agents_shipgate.core.host_boundary import _safe_rule @@ -1564,6 +1570,10 @@ def capability_diff_rows( agent_reasons=agent_reasons, ) kind = grant.get("kind") + if kind == "openshell_policy": + comparison = compare_openshell_grants(before_grant, after_grant) + direction = comparison.direction if comparison.direction in {"widened", "narrowed"} else CHANGED + why = comparison.explanation + "; declared policy only; runtime enforcement and freshness are unverified" if ( kind == "plugin_or_app" and after_grant is not None and not str(after_grant.get("name", "")).startswith("marketplace:") diff --git a/src/agents_shipgate/core/host_grants.py b/src/agents_shipgate/core/host_grants.py index e48d77e3..d9b65a09 100644 --- a/src/agents_shipgate/core/host_grants.py +++ b/src/agents_shipgate/core/host_grants.py @@ -78,6 +78,7 @@ parse_selection, policy_field_paths, ) +from agents_shipgate.core.openshell_compare import compare_openshell_grants from agents_shipgate.core.permission_lattice import ( exec_equivalent_argument, permission_pairing_group, @@ -6013,9 +6014,13 @@ def diff_host_grants(baseline: dict[str, Any], current: dict[str, Any]) -> list[ for grant_id in sorted(set(base_by_id) | set(current_by_id)): before = base_by_id.get(grant_id) after = current_by_id.get(grant_id) + same_openshell = bool( + before and after and before.get("kind") == after.get("kind") == "openshell_policy" + and compare_openshell_grants(before, after).direction == "equivalent" + ) if compared_grant(before) != compared_grant(after) and not _same_workflow_grant( before, after - ): + ) and not same_openshell: changes.append({"grant_id": grant_id, "baseline": before, "current": after}) return changes @@ -6351,6 +6356,8 @@ def host_grant_direction_unknown( """ before, after = change.get("baseline"), change.get("current") + if (after or before or {}).get("kind") == "openshell_policy": + return compare_openshell_grants(before, after).direction in {"unknown", "mixed"} if after is None or (before is not None and before.get("config_sha256") == after.get("config_sha256")): return False if host_grant_expansion_signals([change], comparison_changes=comparison_changes): @@ -6383,6 +6390,11 @@ def host_grant_expansion_signals( for change in changes: before = change.get("baseline") after = change.get("current") + if (after or before or {}).get("kind") == "openshell_policy": + for reason in compare_openshell_grants(before, after).widened: + grant = after or before + signals.append(f"openshell_authority_expanded: {grant['source']}: {reason}") + continue if after is None: if before and before.get("kind") == "permission_rule" and before.get("disposition") in {"deny", "ask"}: signals.append(f"{before['disposition']}_rule_removed: {before['host']}:{before['rule']}") diff --git a/src/agents_shipgate/core/openshell_compare.py b/src/agents_shipgate/core/openshell_compare.py new file mode 100644 index 00000000..330cd4fe --- /dev/null +++ b/src/agents_shipgate/core/openshell_compare.py @@ -0,0 +1,227 @@ +"""Conservative comparison of declared OpenShell authority, never runtime state. + +Finite partitions prove exact REST/L4 comparisons while retaining each +binary × host × port × request relationship. Unsupported shapes are unknown. +""" +from __future__ import annotations + +import json +import re +from dataclasses import dataclass +from pathlib import PurePosixPath +from typing import Any + +MAX_COMPARISON_CELLS = 100_000 +_OTHER = object() +_READ = {"GET", "HEAD", "OPTIONS"} +_WRITE = _READ | {"POST", "PUT", "PATCH"} +_BASELINE_PATHS = {"/bin", "/usr", "/lib", "/proc", "/dev/urandom", "/etc", "/var/log", "/tmp", "/dev/null"} + + +@dataclass(frozen=True) +class OpenShellComparison: + widened: tuple[str, ...] = () + narrowed: tuple[str, ...] = () + limits: tuple[str, ...] = () + + @property + def direction(self) -> str: + if self.limits: + return "unknown" + if self.widened and self.narrowed: + return "mixed" + return "widened" if self.widened else "narrowed" if self.narrowed else "equivalent" + + @property + def explanation(self) -> str: + parts = [*(f"widens {item}" for item in self.widened), + *(f"narrows {item}" for item in self.narrowed), *self.limits] + if self.direction in {"mixed", "unknown"}: + parts.append("a single authority direction is unknown") + return "; ".join(parts) or "equivalent declared authority in the supported static comparison" + + +def _canonical(value: Any) -> str: + """Declaration collections are sets; labels and ordering grant no authority.""" + def normalize(item: Any) -> Any: + if isinstance(item, dict): + return {key: normalize(child) for key, child in sorted(item.items())} + if isinstance(item, list): + return sorted({json.dumps(normalize(child), sort_keys=True) for child in item}) + return item + return json.dumps(normalize(value), sort_keys=True) + + +def _network(policy: dict) -> list[dict]: + return [{key: value for key, value in rule.items() if key != "name"} + for rule in policy.get("network_policies", {}).values()] + + +def _literal_path(path: str) -> bool: + return bool(path.startswith("/") and PurePosixPath(path).as_posix() == path + and not set(path.split("/")) & {".", ".."} + and not any(char in path for char in "*?[]{}\\")) + + +def compare_openshell_grants(before: dict | None, after: dict | None) -> OpenShellComparison: + if not before or not after: + return OpenShellComparison(limits=("selected document added or removed; replacement runtime policy is unestablished",)) + left, right = before["facts"], after["facts"] + if any(left.get(key) != right.get(key) for key in ("runtime_version", "role", "policy_schema_version")): + return OpenShellComparison(limits=("policy version or input role changed",)) + old, new = left["policy"], right["policy"] + widened: list[str] = [] + narrowed: list[str] = [] + limits: list[str] = [] + + def record(label: str, grows: bool, shrinks: bool) -> None: + if grows: + widened.append(label) + if shrinks: + narrowed.append(label) + + old_fs, new_fs = old.get("filesystem_policy"), new.get("filesystem_policy") + if _canonical(old_fs) != _canonical(new_fs): + if old_fs is None or new_fs is None: + limits.append("filesystem omission selects runtime defaults") + elif old_fs["include_workdir"] != new_fs["include_workdir"]: + limits.append("workdir identity and filesystem reach are runtime-dependent") + else: + old_read = set(old_fs["read_only"]) | set(old_fs["read_write"]) + new_read = set(new_fs["read_only"]) | set(new_fs["read_write"]) + old_write, new_write = set(old_fs["read_write"]), set(new_fs["read_write"]) + changed = (old_read ^ new_read) | (old_write ^ new_write) + all_paths = old_read | new_read + # Whether a path is an existing file, directory or symlink is not + # established. Ancestor/descendant replacements need that context. + nested = any(a != b and b.startswith(a.rstrip("/") + "/") + for a in all_paths for b in all_paths) + baseline_active = bool(old.get("network_policies")) or bool(new.get("network_policies")) + if any(not _literal_path(path) for path in all_paths) or nested: + limits.append("filesystem path resolution or ancestor overlap is unproven") + elif baseline_active and changed & _BASELINE_PATHS: + limits.append("changed filesystem path overlaps the runtime-added baseline") + else: + record("filesystem read paths", bool(new_read - old_read), bool(old_read - new_read)) + record("filesystem write paths", bool(new_write - old_write), bool(old_write - new_write)) + if bool(old.get("network_policies")) != bool(new.get("network_policies")): + limits.append("network policy presence changes the runtime-added filesystem baseline") + old_landlock = (old.get("landlock") or {}).get("compatibility", "best_effort") + new_landlock = (new.get("landlock") or {}).get("compatibility", "best_effort") + record("Landlock compatibility", old_landlock == "hard_requirement" and new_landlock == "best_effort", + old_landlock == "best_effort" and new_landlock == "hard_requirement") + if old.get("process") != new.get("process"): + limits.append("process identity comparison requires image and driver context") + if _canonical(_network(old)) != _canonical(_network(new)): + try: + grows, shrinks = _compare_network(old, new) + record("exact binary/destination/request grants", grows, shrinks) + except _Unsupported as exc: + limits.append(str(exc)) + return OpenShellComparison(tuple(widened), tuple(narrowed), tuple(limits)) + + +class _Unsupported(ValueError): + pass + + +def _groups(policy: dict) -> dict[tuple[str, str, int], list[dict]]: + groups: dict[tuple[str, str, int], list[dict]] = {} + references = 0 + for rule in policy.get("network_policies", {}).values(): + for binary in rule["binaries"]: + if not _literal_path(binary["path"]): + raise _Unsupported("wildcard or unresolved executable selector is outside exact comparison") + for endpoint in rule["endpoints"]: + host = endpoint["host"] + if not host or not re.fullmatch(r"[A-Za-z0-9.-]+", host) or endpoint["allowed_ips"]: + raise _Unsupported("host wildcard, address constraint or unresolved destination is outside exact comparison") + if endpoint["path"] or endpoint["protocol"] not in {"", "tcp", "rest"}: + raise _Unsupported("endpoint path routing or protocol is outside exact REST/L4 comparison") + if endpoint["tls"] or endpoint["allow_encoded_slash"] or any( + endpoint[key] for key in ("credential_signing", "signing_service", "signing_region", + "credential_binding", "allow_uninspected_credentials", + "websocket_credential_rewrite", "request_body_credential_rewrite") + ): + raise _Unsupported("transport or credential semantics are outside exact comparison") + if (endpoint["mcp"] or endpoint["json_rpc"] or endpoint["graphql_persisted_queries"] + or endpoint["persisted_queries"] not in {"", "deny"} + or endpoint["graphql_max_body_bytes"] != 65536): + raise _Unsupported("protocol-specific options are outside exact comparison") + if endpoint["protocol"] != "rest" and (endpoint["rules"] or endpoint["deny_rules"] or endpoint["access"]): + raise _Unsupported("request fields on an uninspected endpoint are outside exact comparison") + for matcher in [*(item["allow"] for item in endpoint["rules"]), *endpoint["deny_rules"]]: + if any(matcher[key] for key in ("command", "query", "operation_type", "operation_name", "fields", "tool", "params")): + raise _Unsupported("request constraints are outside exact method/path comparison") + if not matcher["method"] or (matcher["method"] != "*" and not re.fullmatch(r"[A-Z]+", matcher["method"])): + raise _Unsupported("method matcher is outside exact comparison") + if not _literal_path(matcher["path"]): + raise _Unsupported("request path wildcard or omission is outside exact comparison") + for port in set(endpoint["ports"] or [endpoint["port"]]): + groups.setdefault((binary["path"], host.lower(), port), []).append(endpoint) + references += 1 + if references > MAX_COMPARISON_CELLS: + raise _Unsupported("network comparison exceeds its cell limit") + return groups + + +def _matches(matcher: dict, method: Any, path: Any) -> bool: + return matcher["method"] in {"*", method} and matcher["path"] == path + + +def _allows(endpoints: list[dict], method: Any, path: Any) -> bool: + if not endpoints: + return False + inspected = [endpoint for endpoint in endpoints if endpoint["protocol"] == "rest"] + if not inspected: + return True + modes = {endpoint["enforcement"] or "audit" for endpoint in inspected} + if len(modes) != 1: + raise _Unsupported("overlapping inspected endpoints disagree on enforcement") + if modes == {"audit"}: + return True # Well-formed requests only; malformed requests remain denied. + if any(_matches(deny, method, path) for endpoint in inspected for deny in endpoint["deny_rules"]): + return False + return any( + endpoint["access"] == "full" + or method in (_READ if endpoint["access"] == "read-only" else _WRITE if endpoint["access"] == "read-write" else set()) + or any(_matches(item["allow"], method, path) for item in endpoint["rules"]) + for endpoint in inspected + ) + + +def _compare_network(old: dict, new: dict) -> tuple[bool, bool]: + before, after = _groups(old), _groups(new) + grows = shrinks = False + cells = 0 + for host, port in sorted({key[1:] for key in before.keys() | after.keys()}): + binaries = sorted({key[0] for key in before.keys() | after.keys() if key[1:] == (host, port)}) + if len(binaries) > 8: + raise _Unsupported("network comparison exceeds eight exact binary selectors per destination") + endpoints = [endpoint for groups in (before, after) + for key, entries in groups.items() if key[1:] == (host, port) + for endpoint in entries] + methods: set[Any] = _WRITE | {"DELETE", "TRACE", "CONNECT", _OTHER} + paths: set[Any] = {_OTHER} + for endpoint in endpoints: + for matcher in [*(item["allow"] for item in endpoint["rules"]), *endpoint["deny_rules"]]: + if matcher["method"] != "*": + methods.add(matcher["method"]) + paths.add(matcher["path"]) + combinations = (1 << len(binaries)) - 1 + cells += combinations * len(methods) * len(paths) + if cells > MAX_COMPARISON_CELLS: + raise _Unsupported("network comparison exceeds its cell limit") + # A process and any ancestor may independently match a binary path. + # Evaluate every nonempty matching set: denial and inspection apply + # across all those rules, not just to each binary in isolation. + for mask in range(1, combinations + 1): + matched = [binary for index, binary in enumerate(binaries) if mask & (1 << index)] + left = [endpoint for binary in matched for endpoint in before.get((binary, host, port), [])] + right = [endpoint for binary in matched for endpoint in after.get((binary, host, port), [])] + for method in methods: + for path in paths: + was, now = _allows(left, method, path), _allows(right, method, path) + grows |= now and not was + shrinks |= was and not now + return grows, shrinks diff --git a/tests/test_openshell_compare.py b/tests/test_openshell_compare.py new file mode 100644 index 00000000..3d1d846f --- /dev/null +++ b/tests/test_openshell_compare.py @@ -0,0 +1,237 @@ +from __future__ import annotations + +import copy +import hashlib +import json + +import pytest + +from agents_shipgate.core.capability_diff_rows import capability_diff_rows +from agents_shipgate.core.host_grants import ( + diff_host_grants, + host_grant_direction_unknown, + host_grant_expansion_signals, +) +from agents_shipgate.core.openshell import parse_policy +from agents_shipgate.core.openshell_compare import compare_openshell_grants + + +def policy(*, access="read-only", enforcement="enforce", host="api.example.com", binary="/usr/bin/client"): + return {"version": 1, "filesystem_policy": {"read_only": ["/data"], "read_write": []}, + "network_policies": {"client": {"binaries": [{"path": binary}], + "endpoints": [{"host": host, "port": 443, + "protocol": "rest", "access": access, + "enforcement": enforcement}]}}} + + +def endpoint(value): + return value["network_policies"]["client"]["endpoints"][0] + + +def grant(value, *, role="authored"): + facts = {"runtime_version": "0.1.2", "role": role, "policy_schema_version": 1, + "policy": parse_policy(json.dumps(value)).model_dump(mode="json")} + return {"grant_id": "one", "kind": "openshell_policy", "host": "openshell", "source": "arbitrary.rules", + "scope": "repository", "risk": "unknown", "config_sha256": hashlib.sha256(json.dumps(facts).encode()).hexdigest(), + "facts": facts} + + +def compare(old, new): + return compare_openshell_grants(grant(old), grant(new)) + + +def test_enforcement_removal_uses_audit_default_and_publishes_same_direction(): + old, new = policy(), policy() + del endpoint(new)["enforcement"] + before, after = grant(old), grant(new) + result = compare_openshell_grants(before, after) + assert result.direction == "widened" + change = {"baseline": before, "current": after} + signals = host_grant_expansion_signals([change]) + row, = capability_diff_rows({"changes": [change], "expansion_signals": signals}) + assert row.direction == "widened" and row.expands + assert result.explanation in row.why + assert "runtime enforcement and freshness are unverified" in row.why + assert "facts" in row.before and "facts" in row.after + + +@pytest.mark.parametrize("old,new,direction", [("full", "read-write", "narrowed"), + ("read-only", "read-write", "widened"), + ("read-write", "full", "widened")]) +def test_rest_access_presets(old, new, direction): + assert compare(policy(access=old), policy(access=new)).direction == direction + + +def test_audit_access_edit_does_not_change_allowed_requests(): + assert compare(policy(enforcement="audit"), policy(access="full", enforcement="audit")).direction == "equivalent" + + +def test_full_enforce_to_audit_is_equivalent_in_the_supported_well_formed_domain(): + assert compare(policy(access="full"), policy(access="full", enforcement="audit")).direction == "equivalent" + + +def test_redundant_allow_and_renaming_reordering_are_quiet(): + old, new = policy(), policy() + new["network_policies"]["renamed"] = new["network_policies"].pop("client") + new["network_policies"]["redundant"] = copy.deepcopy(new["network_policies"]["renamed"]) + new["network_policies"]["redundant"]["name"] = "display name" + assert compare(old, new).direction == "equivalent" + assert diff_host_grants({"grants": [grant(old)]}, {"grants": [grant(new)]}) == [] + + +def test_duplicate_denial_removal_does_not_expand(): + old = policy(access="full") + endpoint(old)["deny_rules"] = [{"method": "DELETE", "path": "/records"}] + old["network_policies"]["second"] = copy.deepcopy(old["network_policies"]["client"]) + new = copy.deepcopy(old) + endpoint(new)["deny_rules"] = [] + assert compare(old, new).direction == "equivalent" + new["network_policies"]["second"]["endpoints"][0]["deny_rules"] = [] + assert compare(old, new).direction == "widened" + + +def test_global_denial_overrides_an_allow_in_another_rule(): + old = policy(access="full") + endpoint(old)["deny_rules"] = [{"method": "*", "path": "/records"}] + new = copy.deepcopy(old) + new["network_policies"]["extra"] = copy.deepcopy(old["network_policies"]["client"]) + extra = new["network_policies"]["extra"]["endpoints"][0] + extra.pop("access") + extra["rules"] = [{"allow": {"method": "POST", "path": "/records"}}] + extra["deny_rules"] = [] + assert compare(old, new).direction == "equivalent" + + +def test_uninspected_overlap_adds_no_request_access(): + old, new = policy(), policy() + new["network_policies"]["l4"] = {"binaries": [{"path": "/usr/bin/client"}], + "endpoints": [{"host": "api.example.com", "port": 443}]} + assert compare(old, new).direction == "equivalent" + + +def test_correlation_is_not_flattened(): + old = policy(host="one.example.com", binary="/bin/one") + old["network_policies"]["second"] = policy(host="two.example.com", binary="/bin/two")["network_policies"]["client"] + new = copy.deepcopy(old) + endpoint(new)["host"] = "two.example.com" + new["network_policies"]["second"]["endpoints"][0]["host"] = "one.example.com" + result = compare(old, new) + assert result.direction == "mixed" and result.widened and result.narrowed + assert host_grant_direction_unknown({"baseline": grant(old), "current": grant(new)}) + + +@pytest.mark.parametrize("field,value", [("host", "*.example.com"), ("path", "/v1/**"), + ("credential_binding", {"provider": "github"}), + ("protocol", "graphql"), ("allow_encoded_slash", True)]) +def test_unsupported_endpoint_semantics_never_become_safe_narrowing(field, value): + old, new = policy(access="full"), policy() + endpoint(new)[field] = value + result = compare(old, new) + assert result.direction == "unknown" and result.limits + assert not result.narrowed and not result.widened + + +def test_request_glob_is_named_unknown(): + old, new = policy(access="full"), policy() + endpoint(new).pop("access") + endpoint(new)["rules"] = [{"allow": {"method": "GET", "path": "/v1/**"}}] + assert compare(old, new).direction == "unknown" + + +def test_exact_request_set_addition_and_removal(): + old = policy() + endpoint(old).pop("access") + endpoint(old)["rules"] = [{"allow": {"method": "GET", "path": "/records"}}] + new = copy.deepcopy(old) + endpoint(new)["rules"].append({"allow": {"method": "POST", "path": "/records"}}) + assert compare(old, new).direction == "widened" + assert compare(new, old).direction == "narrowed" + + +@pytest.mark.parametrize("field,old,new,direction", [ + ("read_write", [], ["/data"], "widened"), + ("read_write", ["/data"], [], "narrowed"), + ("read_only", ["/data"], ["/data", "/logs"], "widened"), +]) +def test_exact_filesystem_direction(field, old, new, direction): + left, right = policy(), policy() + left["filesystem_policy"][field], right["filesystem_policy"][field] = old, new + assert compare(left, right).direction == direction + + +def test_filesystem_ancestor_or_runtime_baseline_changes_are_unknown(): + left, right = policy(), policy() + right["filesystem_policy"]["read_only"].append("/data/nested") + assert compare(left, right).direction == "unknown" + right = policy() + right["filesystem_policy"]["read_only"].append("/usr") + assert compare(left, right).direction == "unknown" + + +def test_landlock_omission_restores_best_effort(): + left, right = policy(), policy() + left["landlock"] = {"compatibility": "hard_requirement"} + assert compare(left, right).direction == "widened" + assert compare(right, left).direction == "narrowed" + + +def test_process_workdir_role_and_missing_replacement_are_unproven(): + left, right = policy(), policy() + right["process"] = {"run_as_user": "1001"} + assert compare(left, right).direction == "unknown" + right = policy() + right["filesystem_policy"]["include_workdir"] = True + assert compare(left, right).direction == "unknown" + assert compare_openshell_grants(grant(left), grant(left, role="effective_snapshot")).direction == "unknown" + assert compare_openshell_grants(grant(left), None).direction == "unknown" + + +def test_credential_change_is_unknown_even_if_display_hash_is_equal(): + left, right = grant(policy()), grant(policy()) + endpoint(right["facts"]["policy"])["credential_binding"] = {"provider": "github"} + assert left["config_sha256"] == right["config_sha256"] + assert host_grant_direction_unknown({"baseline": left, "current": right}) + + +def test_mcp_tool_name_does_not_assert_business_effects(): + old, new = policy(), policy() + target = endpoint(new) + target.pop("access") + target["protocol"] = "mcp" + target["rules"] = [{"allow": {"tool": "delete_resource"}}] + result = compare(old, new) + assert result.direction == "unknown" and "protocol" in result.explanation + assert not any(word in result.explanation for word in ("approval", "binding", "business effect")) + + +def test_comparison_work_is_bounded(monkeypatch): + from agents_shipgate.core import openshell_compare + monkeypatch.setattr(openshell_compare, "MAX_COMPARISON_CELLS", 1) + assert compare(policy(), policy(access="full")).direction == "unknown" + + +def test_ancestor_denial_applies_to_a_child_binary_allow(): + old = policy(access="read-only", binary="/bin/ancestor") + old["network_policies"]["child"] = policy(access="full", binary="/bin/child")["network_policies"]["client"] + new = copy.deepcopy(old) + endpoint(new)["deny_rules"] = [{"method": "POST", "path": "/records"}] + # POST was already disallowed by the ancestor's own allow, but its new + # denial also blocks a child allow when both selectors match the chain. + assert compare(old, new).direction == "narrowed" + + +def test_inspection_suppresses_an_ancestor_l4_grant_as_well_as_adding_child_access(): + old = policy(binary="/bin/ancestor") + endpoint(old).pop("access") + endpoint(old).pop("protocol") + new = copy.deepcopy(old) + new["network_policies"]["child"] = policy(binary="/bin/child")["network_policies"]["client"] + result = compare(old, new) + assert result.direction == "mixed" and result.widened and result.narrowed + + +def test_overlapping_ancestor_modes_cannot_be_proven_safe(): + old = policy(binary="/bin/ancestor") + new = copy.deepcopy(old) + new["network_policies"]["child"] = policy(binary="/bin/child", enforcement="audit")["network_policies"]["client"] + assert compare(old, new).direction == "unknown"