fix(permissions): prevent a corrupt rules file from running permanently blocked commands

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:

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()

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"]