From 9ea03e35f72428f2450b2466592e5332d1626028 Mon Sep 17 00:00:00 2001 From: EddieMurphy <142191175+EddieMurphy89@users.noreply.github.com> Date: Mon, 9 Nov 2026 17:21:25 +0800 Subject: [PATCH] fix(permissions): prevent a corrupt rules file from running permanently blocked commands --- apodex/agent_tools.py | 9 +++ apodex/permissions.py | 80 ++++++++++++++++++-- apodex/tests/test_permissions.py | 125 +++++++++++++++++++++++++++++++ 3 files changed, 207 insertions(+), 7 deletions(-) create mode 100644 apodex/tests/test_permissions.py diff --git a/apodex/agent_tools.py b/apodex/agent_tools.py index df342816..e748f421 100644 --- a/apodex/agent_tools.py +++ b/apodex/agent_tools.py @@ -438,6 +438,15 @@ def assess_with_rules( downgrades: the typed-confirmation gate must still fire. """ base = assess_tool_risk(name, args, cwd) + if rules is not None and getattr(rules, "load_error", None): + # The file exists but could not be read or parsed. The deny list is + # unknown, so the only safe decision is to block rather than fall back + # to an empty (permissive) policy. + return ToolRisk( + RISK_DENY, + "saved permission rules could not be read; refusing to run with an unknown policy", + base.target, + ) if rules is not None and rules.denies(name, args): return ToolRisk(RISK_DENY, "denied by a saved rule", base.target) if base.level == RISK_DENY or base.must_ask: diff --git a/apodex/permissions.py b/apodex/permissions.py index 71688787..6853ade7 100644 --- a/apodex/permissions.py +++ b/apodex/permissions.py @@ -24,11 +24,15 @@ from __future__ import annotations import json +import logging import os import re import shlex +import threading from dataclasses import dataclass, field +logger = logging.getLogger(__name__) + _DEFAULT_PATH = os.path.expanduser("~/.config/apodex/permissions.json") # Commands whose first token alone is too coarse — keep two words so # "always allow npm test" doesn't also allow "npm publish". @@ -293,33 +297,93 @@ def rule_for(name: str, args: dict) -> str: return name +def _string_list(value: object) -> list[str] | None: + """Return a JSON rule list, or ``None`` when the shape is invalid. + + ``None`` (the key is absent) is a valid empty list; a string or a list + containing non-strings is not, because ``set()`` over it would silently + produce a policy different from the one on disk. + """ + if value is None: + return [] + if not isinstance(value, list) or not all(isinstance(item, str) for item in value): + return None + return value + + @dataclass class PermissionStore: - """Allow/deny rules, persisted to a JSON file.""" + """Allow/deny rules, persisted to a JSON file. + + ``load_error`` records that the file exists but could not be read or + parsed. In that state the policy is unknown, so ``denies`` fails closed: + an unreadable deny list must never turn into "no deny rules". + """ allow: set[str] = field(default_factory=set) deny: set[str] = field(default_factory=set) path: str = _DEFAULT_PATH + load_error: str | None = None @classmethod def load(cls, path: str = _DEFAULT_PATH) -> PermissionStore: try: with open(path, encoding="utf-8") as f: d = json.load(f) - return cls(set(d.get("allow") or []), set(d.get("deny") or []), path) - except Exception: + except FileNotFoundError: + # No file is a valid first run: there is no policy to lose yet. + # A broken symlink also raises FileNotFoundError, but the path + # exists; treat that as an unknown policy rather than a first run. + if os.path.lexists(path): + logger.warning("permission rules path %s exists but cannot be opened", path) + return cls(path=path, load_error="permission rules path exists but cannot be opened") return cls(path=path) + except Exception as exc: + # The file is present but unreadable or invalid. Treating this as + # an empty policy would silently drop every saved deny rule. + logger.warning("could not read permission rules from %s: %s", path, exc) + return cls(path=path, load_error=str(exc)) + if not isinstance(d, dict): + logger.warning("permission rules in %s are not a JSON object", path) + return cls(path=path, load_error="permission rules must be a JSON object") + allow = _string_list(d.get("allow")) + deny = _string_list(d.get("deny")) + if allow is None or deny is None: + logger.warning("permission rules in %s must be string lists", path) + return cls(path=path, load_error="permission rules must be string lists") + return cls(set(allow), set(deny), path) def save(self) -> None: + """Write the rules atomically. + + A failed write leaves the previous file untouched instead of + truncating it first; callers still see a warning in the log rather + than a silent success. + """ + directory = os.path.dirname(self.path) or "." + tmp_path = f"{self.path}.{os.getpid()}.{threading.get_ident()}.tmp" try: - os.makedirs(os.path.dirname(self.path) or ".", exist_ok=True) - with open(self.path, "w", encoding="utf-8") as f: - json.dump({"allow": sorted(self.allow), "deny": sorted(self.deny)}, f, indent=2) + os.makedirs(directory, exist_ok=True) + with open(tmp_path, "w", encoding="utf-8") as f: + json.dump( + {"allow": sorted(self.allow), "deny": sorted(self.deny)}, + f, + indent=2, + ) + f.flush() + os.fsync(f.fileno()) + os.replace(tmp_path, self.path) except Exception: - pass + logger.warning("could not save permission rules to %s", self.path, exc_info=True) + try: + os.unlink(tmp_path) + except OSError: + pass def allows(self, name: str, args: dict) -> bool: """True only when every segment and every nested payload matches.""" + if self.load_error is not None: + return False if _name_matches(self.allow, name): return True prefixes = _bash_rule_prefixes(self.allow) if name == "bash" else set() @@ -346,6 +410,8 @@ def allows(self, name: str, args: dict) -> bool: def denies(self, name: str, args: dict) -> bool: """True when any segment, top-level or nested, matches a deny rule.""" + if self.load_error is not None: + return True if _name_matches(self.deny, name): return True prefixes = _bash_rule_prefixes(self.deny) if name == "bash" else set() diff --git a/apodex/tests/test_permissions.py b/apodex/tests/test_permissions.py new file mode 100644 index 00000000..7c1ca931 --- /dev/null +++ b/apodex/tests/test_permissions.py @@ -0,0 +1,125 @@ +"""Regression tests for the persistent permission store.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path + +import pytest + +from apodex import permissions +from apodex.agent_tools import RISK_DENY, assess_with_rules +from apodex.permissions import PermissionStore + + +def test_missing_file_is_an_empty_policy(tmp_path: Path) -> None: + store = PermissionStore.load(str(tmp_path / "permissions.json")) + + assert store.load_error is None + assert not store.denies("bash", {"command": "git push"}) + + +def test_unparseable_file_fails_closed(tmp_path: Path) -> None: + path = tmp_path / "permissions.json" + path.write_text('{"allow": [], "deny": ["Bash(git push)"]', encoding="utf-8") + + store = PermissionStore.load(str(path)) + + assert store.load_error is not None + assert store.denies("bash", {"command": "git push"}) + assert not store.allows("bash", {"command": "git push"}) + + +def test_wrong_shape_fails_closed(tmp_path: Path) -> None: + path = tmp_path / "permissions.json" + path.write_text('{"allow": "Bash(git push)", "deny": []}', encoding="utf-8") + + store = PermissionStore.load(str(path)) + + assert store.load_error is not None + assert store.denies("bash", {"command": "git push"}) + + +def test_unreadable_path_fails_closed(tmp_path: Path) -> None: + path = tmp_path / "permissions.json" + path.mkdir() + + store = PermissionStore.load(str(path)) + + assert store.load_error is not None + assert store.denies("bash", {"command": "git push"}) + + +@pytest.mark.skipif(os.name == "nt", reason="symlink creation may require privileges on Windows") +def test_broken_symlink_fails_closed(tmp_path: Path) -> None: + link = tmp_path / "permissions.json" + link.symlink_to(tmp_path / "missing.json") + + store = PermissionStore.load(str(link)) + + assert store.load_error is not None + assert store.denies("bash", {"command": "git push"}) + + +def test_load_reads_a_valid_file(tmp_path: Path) -> None: + path = tmp_path / "permissions.json" + path.write_text( + json.dumps({"allow": ["Bash(npm test)"], "deny": ["Bash(git push)"]}), + encoding="utf-8", + ) + + store = PermissionStore.load(str(path)) + + assert store.load_error is None + assert store.denies("bash", {"command": "git push"}) + assert store.allows("bash", {"command": "npm test"}) + + +def test_gate_blocks_when_rules_could_not_be_loaded(tmp_path: Path) -> None: + path = tmp_path / "permissions.json" + path.write_text("{", encoding="utf-8") + store = PermissionStore.load(str(path)) + + risk = assess_with_rules( + "write_file", + {"path": str(tmp_path / "created.txt")}, + str(tmp_path), + store, + ) + + assert risk.level == RISK_DENY + assert "could not be read" in risk.reason + + +def test_failed_save_keeps_the_previous_file( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + path = tmp_path / "permissions.json" + store = PermissionStore({"Bash(npm test)"}, {"Bash(git push)"}, str(path)) + store.save() + before = path.read_text(encoding="utf-8") + + def fail(*args: object, **kwargs: object) -> None: + raise OSError("disk full") + + monkeypatch.setattr(permissions.json, "dump", fail) + store.save() + + assert path.read_text(encoding="utf-8") == before + assert sorted(p.name for p in tmp_path.iterdir()) == ["permissions.json"] + + +def test_save_replaces_the_file_atomically(tmp_path: Path) -> None: + path = tmp_path / "permissions.json" + store = PermissionStore({"Bash(npm test)"}, {"Bash(git push)"}, str(path)) + + store.save() + store.deny.add("write_file") + store.save() + + reloaded = PermissionStore.load(str(path)) + assert reloaded.load_error is None + assert reloaded.denies("write_file", {}) + assert reloaded.allows("bash", {"command": "npm test"}) + assert sorted(p.name for p in tmp_path.iterdir()) == ["permissions.json"]