slaguru666/AIPass
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(system): security: DPLAN-0155 Phases 2-4 — prompt injection defense, daemon path confinement, JSONL writes removed

Browse Source 
Co-Authored-By: @devpulse <devpulse@aipass>
This commit is contained in:
AIOSAI
2026-04-26 17:08:22 -07:00
co-authored by @devpulse
parent d84e56344c
commit b6be4a9c68
5 changed files with 33 additions and 303 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.claude/hooks/pre_edit_gate.py
+30 -10
View File
@@ -59,16 +59,39 @@ def main():
# ------------------------------------------------------------------
fp = Path(file_path)
if fp.name == "inbox.json" and ".ai_mail.local" in fp.parts:
_block(
"Direct writes to inbox.json are blocked.\n"
"Use: drone @ai_mail email @<branch> \"Subject\" \"Body\""
)
_block('Direct writes to inbox.json are blocked.\nUse: drone @ai_mail email @<branch> "Subject" "Body"')
# ------------------------------------------------------------------
# Rule 1.5: Dispatched-agent path confinement (DPLAN-0155 M3)
# Daemon-spawned agents can only write inside their own branch dir.
# Breaks the prompt-injection amplifier chain — even if injected,
# a dispatched agent cannot write to other agents' inboxes or code.
# ------------------------------------------------------------------
cwd = input_data.get("cwd", "") or os.getcwd()
cwd_branch = _get_branch(cwd)
session_type = os.environ.get("AIPASS_SESSION_TYPE", "interactive")
if session_type == "daemon" and cwd_branch:
target_branch = _get_branch(str(fp.resolve()) if not fp.is_absolute() else str(fp))
if target_branch and target_branch != cwd_branch:
_block(
f"Dispatched agent confined to own branch: '{cwd_branch}' "
f"cannot write to '{target_branch}' in daemon mode."
)
repo_root = None
for parent in Path(cwd).parents:
if (parent / ".git").exists():
repo_root = parent
break
if repo_root and not target_branch:
allowed_prefix = str(repo_root / "src" / "aipass" / cwd_branch)
resolved = str(fp.resolve()) if not fp.is_absolute() else str(fp)
if not resolved.startswith(allowed_prefix):
_block(f"Dispatched agent restricted to {allowed_prefix}. Cannot write to: {file_path}")
# ------------------------------------------------------------------
# Rule 2: Cross-branch write enforcement
# ------------------------------------------------------------------
cwd = input_data.get("cwd", "") or os.getcwd()
cwd_branch = _get_branch(cwd)
target_branch = _get_branch(str(fp.resolve()) if not fp.is_absolute() else str(fp))
if cwd_branch and target_branch and cwd_branch != target_branch:
@@ -115,10 +138,7 @@ def main():
return
error_summary = "\n".join(f" L{e['line']}: {e['message']}" for e in errors[:5])
_block(
f"Fix {len(errors)} error(s) in {Path(errored_file).name} before editing other files:\n"
f"{error_summary}"
)
_block(f"Fix {len(errors)} error(s) in {Path(errored_file).name} before editing other files:\n{error_summary}")
except Exception:
pass # Silent fail → allow
src/aipass/ai_mail/apps/handlers/dispatch/daemon.py
+3 -32
View File
@@ -117,31 +117,6 @@ def _write_json(filepath: Path, data: Dict[str, Any]) -> bool:
return False
def _set_session_name(branch_path: Path, name: str) -> bool:
"""Write custom-title to the most recent Claude session JSONL for a branch.
Claude stores sessions at ~/.claude/projects/{encoded-cwd}/*.jsonl.
Writing a custom-title entry makes the session identifiable in /resume picker.
"""
encoded_cwd = str(branch_path).replace("/", "-")
projects_dir = Path("~/.claude/projects").expanduser() / encoded_cwd
if not projects_dir.exists():
return False
jsonl_files = sorted(projects_dir.glob("*.jsonl"), key=lambda f: f.stat().st_mtime, reverse=True)
if not jsonl_files:
return False
latest = jsonl_files[0]
session_id = latest.stem
entry = json.dumps({"type": "custom-title", "customTitle": name, "sessionId": session_id})
try:
with open(latest, "a", encoding="utf-8") as f:
f.write(entry + "\n")
return True
except OSError as e:
logger.warning("[daemon] Failed to write session name for %s: %s", branch_path, e)
return False
def _check_lock(branch_path: Path) -> Optional[Dict[str, Any]]:
"""Check if branch has an active dispatch lock. Returns lock data or None."""
lock_file = branch_path / ".ai_mail.local" / ".dispatch.lock"
@@ -362,14 +337,14 @@ def spawn_agent(
True if monitor was spawned successfully
"""
sender = message.get("from", "unknown")
msg_id = message.get("id", "unknown")
subject = message.get("subject", "")
max_turns = config.get("max_turns_per_wake", 100)
lock_file_path = str(branch_path / ".ai_mail.local" / ".dispatch.lock")
# Prompt — no lock cleanup instruction (dispatch_monitor handles it)
prompt = f"Hi. Check inbox for task from {sender} (message ID: {msg_id}). Execute it. Send confirmation when done."
# Prompt — never interpolate sender-controlled content into the prompt.
# The agent reads inbox.json as data, not as instructions (DPLAN-0155 M1).
prompt = "Hi. Check your inbox for new dispatch emails and execute the task. Send confirmation when done."
claude_cmd = [
"claude",
@@ -409,10 +384,6 @@ def spawn_agent(
if key.startswith("CLAUDE") or key == "AIPASS_BOT_ID":
spawn_env.pop(key)
# Set session name for /resume picker (daemon always uses -c resume)
spawn_branch_name = branch_email.lstrip("@").upper()
_set_session_name(branch_path, f"{spawn_branch_name}-daemon")
try:
process = subprocess.Popen(
monitor_cmd,
src/aipass/ai_mail/apps/handlers/dispatch/wake.py
-27
View File
@@ -210,27 +210,6 @@ def _load_config() -> dict:
return config
def _set_session_name(branch_path: Path, name: str) -> bool:
"""Write custom-title to the most recent Claude session JSONL for a branch."""
encoded_cwd = str(branch_path).replace("/", "-")
projects_dir = Path("~/.claude/projects").expanduser() / encoded_cwd
if not projects_dir.exists():
return False
jsonl_files = sorted(projects_dir.glob("*.jsonl"), key=lambda f: f.stat().st_mtime, reverse=True)
if not jsonl_files:
return False
latest = jsonl_files[0]
session_id = latest.stem
entry = json.dumps({"type": "custom-title", "customTitle": name, "sessionId": session_id})
try:
with open(latest, "a", encoding="utf-8") as f:
f.write(entry + "\n")
return True
except OSError as e:
logger.warning("[wake] Failed to write session name for %s: %s", branch_path, e)
return False
def _read_session_type(pid_str: str) -> str:
"""Read AIPASS_SESSION_TYPE from /proc/{pid}/environ. Returns 'interactive' if unset."""
if sys.platform != "linux":
@@ -476,12 +455,6 @@ def wake_branch(
"json",
]
# Set session name for /resume picker
branch_name = email.lstrip("@").upper()
session_label = f"{branch_name}-dispatched"
if not fresh:
_set_session_name(branch_path, session_label)
# Step 7: Spawn via dispatch_monitor
log_dir = branch_path / "logs"
log_dir.mkdir(parents=True, exist_ok=True)
src/aipass/ai_mail/tests/test_daemon.py
-155
View File
@@ -767,13 +767,11 @@ def test_poll_cycle_absolute_path_unchanged(tmp_path, monkeypatch):
import os
import sys
from pathlib import Path
from unittest.mock import MagicMock, mock_open
from aipass.ai_mail.apps.handlers.dispatch.daemon import (
_notify_telegram,
_handle_signal,
_set_session_name,
_check_lock,
_acquire_lock,
poll_cycle,
@@ -872,106 +870,6 @@ def test_handle_signal_sets_shutdown(monkeypatch):
assert daemon_mod.SHUTDOWN is True
# ---- _set_session_name tests ------------------------------------
def test_set_session_name_success(tmp_path, monkeypatch):
"""Writes custom-title entry to most recent JSONL file."""
branch_path = tmp_path / "branch"
branch_path.mkdir()
# Redirect ~/.claude/projects to tmp_path so no real filesystem side effects
fake_home = tmp_path / "fakehome"
encoded_cwd = str(branch_path).replace("/", "-")
projects_dir = fake_home / ".claude" / "projects" / encoded_cwd
projects_dir.mkdir(parents=True)
jsonl_file = projects_dir / "session123.jsonl"
jsonl_file.write_text('{"type":"init"}\n', encoding="utf-8")
_orig_expanduser = Path.expanduser
def _fake_expanduser(self):
if str(self).startswith("~"):
return fake_home / str(self)[2:]
return _orig_expanduser(self)
monkeypatch.setattr(Path, "expanduser", _fake_expanduser)
result = _set_session_name(branch_path, "TEST-daemon")
assert result is True
content = jsonl_file.read_text(encoding="utf-8")
assert "custom-title" in content
assert "TEST-daemon" in content
def test_set_session_name_no_projects_dir(tmp_path, monkeypatch):
"""Returns False when projects dir does not exist."""
branch_path = tmp_path / "nonexistent_branch_xyz_test"
fake_home = tmp_path / "fakehome"
_orig_expanduser = Path.expanduser
def _fake_expanduser(self):
if str(self).startswith("~"):
return fake_home / str(self)[2:]
return _orig_expanduser(self)
monkeypatch.setattr(Path, "expanduser", _fake_expanduser)
result = _set_session_name(branch_path, "TEST-daemon")
assert result is False
def test_set_session_name_no_jsonl_files(tmp_path, monkeypatch):
"""Returns False when projects dir exists but has no JSONL files."""
branch_path = tmp_path / "branch"
branch_path.mkdir()
fake_home = tmp_path / "fakehome"
encoded_cwd = str(branch_path).replace("/", "-")
projects_dir = fake_home / ".claude" / "projects" / encoded_cwd
projects_dir.mkdir(parents=True)
_orig_expanduser = Path.expanduser
def _fake_expanduser(self):
if str(self).startswith("~"):
return fake_home / str(self)[2:]
return _orig_expanduser(self)
monkeypatch.setattr(Path, "expanduser", _fake_expanduser)
result = _set_session_name(branch_path, "TEST-daemon")
assert result is False
def test_set_session_name_oserror_on_write(tmp_path, monkeypatch):
"""Returns False on OSError when writing to JSONL file."""
branch_path = tmp_path / "branch"
branch_path.mkdir()
fake_home = tmp_path / "fakehome"
encoded_cwd = str(branch_path).replace("/", "-")
projects_dir = fake_home / ".claude" / "projects" / encoded_cwd
projects_dir.mkdir(parents=True)
jsonl_file = projects_dir / "session456.jsonl"
jsonl_file.write_text('{"type":"init"}\n', encoding="utf-8")
_orig_expanduser = Path.expanduser
def _fake_expanduser(self):
if str(self).startswith("~"):
return fake_home / str(self)[2:]
return _orig_expanduser(self)
monkeypatch.setattr(Path, "expanduser", _fake_expanduser)
with patch("builtins.open", side_effect=OSError("disk full")):
result = _set_session_name(branch_path, "TEST-daemon")
assert result is False
# ---- _check_lock tests -----------------------------------------
@@ -1430,10 +1328,6 @@ def test_spawn_agent_success(tmp_path):
"aipass.ai_mail.apps.handlers.dispatch.daemon._acquire_lock",
return_value=(True, "Lock acquired"),
),
patch(
"aipass.ai_mail.apps.handlers.dispatch.daemon._set_session_name",
return_value=True,
),
patch("aipass.ai_mail.apps.handlers.dispatch.daemon.log_dispatch"),
patch(
"aipass.ai_mail.apps.handlers.dispatch.daemon._notify_telegram",
@@ -1466,10 +1360,6 @@ def test_spawn_agent_exception(tmp_path):
"aipass.ai_mail.apps.handlers.dispatch.daemon.subprocess.Popen",
side_effect=OSError("command not found"),
),
patch(
"aipass.ai_mail.apps.handlers.dispatch.daemon._set_session_name",
return_value=True,
),
patch("aipass.ai_mail.apps.handlers.dispatch.daemon.log_dispatch"),
patch(
"aipass.ai_mail.apps.handlers.dispatch.daemon._notify_telegram",
@@ -1512,10 +1402,6 @@ def test_spawn_agent_strips_claude_env_vars(tmp_path, monkeypatch):
"aipass.ai_mail.apps.handlers.dispatch.daemon._acquire_lock",
return_value=(True, "Lock acquired"),
),
patch(
"aipass.ai_mail.apps.handlers.dispatch.daemon._set_session_name",
return_value=True,
),
patch("aipass.ai_mail.apps.handlers.dispatch.daemon.log_dispatch"),
patch(
"aipass.ai_mail.apps.handlers.dispatch.daemon._notify_telegram",
@@ -1536,47 +1422,6 @@ def test_spawn_agent_strips_claude_env_vars(tmp_path, monkeypatch):
assert captured_env.get("AIPASS_SESSION_TYPE") == "daemon"
def test_spawn_agent_sets_session_name(tmp_path):
"""Spawn calls _set_session_name with correct branch name."""
branch_path = tmp_path / "branch"
branch_path.mkdir()
(branch_path / "logs").mkdir()
message = {"from": "@devpulse", "id": "msg1", "subject": "Test task"}
config = {"max_turns_per_wake": 50}
state = {"daily_counts": {}, "session_cycles": {}}
mock_process = MagicMock()
mock_process.pid = 54321
with (
patch(
"aipass.ai_mail.apps.handlers.dispatch.daemon.subprocess.Popen",
return_value=mock_process,
),
patch(
"aipass.ai_mail.apps.handlers.dispatch.daemon._acquire_lock",
return_value=(True, "Lock acquired"),
),
patch(
"aipass.ai_mail.apps.handlers.dispatch.daemon._set_session_name",
return_value=True,
) as mock_ssn,
patch("aipass.ai_mail.apps.handlers.dispatch.daemon.log_dispatch"),
patch(
"aipass.ai_mail.apps.handlers.dispatch.daemon._notify_telegram",
return_value=True,
),
patch(
"aipass.ai_mail.apps.handlers.dispatch.daemon.send_notification",
create=True,
),
):
spawn_agent(branch_path, "@testbranch", message, config, state)
mock_ssn.assert_called_once_with(branch_path, "TESTBRANCH-daemon")
# ---- run_daemon tests -------------------------------------------
src/aipass/ai_mail/tests/test_wake.py
-79
View File
@@ -29,7 +29,6 @@ from aipass.ai_mail.apps.handlers.dispatch.wake import (
DEFAULT_MODEL,
_acquire_lock,
_load_config,
_set_session_name,
_is_branch_occupied,
wake_branch,
)
@@ -710,84 +709,6 @@ class TestLoadConfig:
assert result["max_turns_per_wake"] == 50
# --- _set_session_name tests --------------------------------------------
class TestSetSessionName:
"""Tests for _set_session_name() — writes custom-title to Claude session JSONL."""
def test_success_appends_entry(self, tmp_path, monkeypatch):
"""Creates expected JSON entry in the most recent session JSONL."""
encoded_cwd = str(tmp_path).replace("/", "-")
projects_dir = tmp_path / ".claude" / "projects" / encoded_cwd
projects_dir.mkdir(parents=True)
session_file = projects_dir / "abc123.jsonl"
session_file.write_text("", encoding="utf-8")
monkeypatch.setattr(
"aipass.ai_mail.apps.handlers.dispatch.wake.Path.expanduser",
lambda self: tmp_path / ".claude" / "projects" if str(self).endswith("projects") else self,
)
# We need to mock expanduser properly — override the whole projects_dir lookup
monkeypatch.setattr(
_Path,
"expanduser",
lambda self: tmp_path / str(self).lstrip("~/"),
)
result = _set_session_name(tmp_path, "TEST-dispatched")
assert result is True
content = session_file.read_text(encoding="utf-8")
entry = json.loads(content.strip())
assert entry["type"] == "custom-title"
assert entry["customTitle"] == "TEST-dispatched"
assert entry["sessionId"] == "abc123"
def test_no_projects_dir_returns_false(self, tmp_path, monkeypatch):
"""Returns False when ~/.claude/projects/{encoded} does not exist."""
monkeypatch.setattr(
_Path,
"expanduser",
lambda self: tmp_path / str(self).lstrip("~/"),
)
result = _set_session_name(tmp_path, "TEST-dispatched")
assert result is False
def test_no_jsonl_files_returns_false(self, tmp_path, monkeypatch):
"""Returns False when projects dir exists but has no .jsonl files."""
encoded_cwd = str(tmp_path).replace("/", "-")
projects_dir = tmp_path / ".claude" / "projects" / encoded_cwd
projects_dir.mkdir(parents=True)
monkeypatch.setattr(
_Path,
"expanduser",
lambda self: tmp_path / str(self).lstrip("~/"),
)
result = _set_session_name(tmp_path, "TEST-dispatched")
assert result is False
def test_os_error_on_write_returns_false(self, tmp_path, monkeypatch):
"""Returns False when write to JSONL file raises OSError."""
encoded_cwd = str(tmp_path).replace("/", "-")
projects_dir = tmp_path / ".claude" / "projects" / encoded_cwd
projects_dir.mkdir(parents=True)
session_file = projects_dir / "abc123.jsonl"
session_file.write_text("", encoding="utf-8")
monkeypatch.setattr(
_Path,
"expanduser",
lambda self: tmp_path / str(self).lstrip("~/"),
)
def _fail_open(path, *args, **kwargs):
path_str = str(path)
if path_str.endswith(".jsonl") and "a" in args:
raise OSError("permission denied")
return _REAL_OPEN(path, *args, **kwargs)
monkeypatch.setattr("builtins.open", _fail_open)
result = _set_session_name(tmp_path, "TEST-dispatched")
assert result is False
# --- _is_branch_occupied tests ------------------------------------------