Merge pull request #375 from AIOSAI/citizen/seedgo-track-e-single-path
feat(hooks+seedgo): DPLAN-0139 Track E — single-path enforcement
This commit is contained in:
AIPass
@@ -0,0 +1,128 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
PreToolUse Gate — Blocks unsafe edits at the hook layer.
|
||||
|
||||
Rules (checked in order):
|
||||
1. Inbox lock — any write targeting *.ai_mail.local/inbox.json is BLOCKED.
|
||||
Use `drone @ai_mail email` instead.
|
||||
2. Cross-branch — writes to src/aipass/X/** from a CWD inside src/aipass/Y/**
|
||||
are BLOCKED unless the calling branch is in TRUSTED_CROSS_WRITERS.
|
||||
3. State-file — edits to OTHER .py files while the current branch has unresolved
|
||||
type errors are BLOCKED. (original v1.2.0 logic)
|
||||
|
||||
Track E additions: rules 1 + 2 (DPLAN-0139).
|
||||
Version: 1.3.0
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
STATE_FILE = Path(__file__).parent / ".diagnostics_state.json"
|
||||
EDIT_TOOLS = {"Edit", "Write", "MultiEdit", "NotebookEdit"}
|
||||
|
||||
# Single source of truth lives in permissions.py — inline here as fallback
|
||||
# so the hook works even when aipass package is not on sys.path.
|
||||
TRUSTED_CROSS_WRITERS: tuple[str, ...] = ("devpulse", "seedgo", "spawn")
|
||||
|
||||
|
||||
def _get_branch(file_path: str) -> str:
|
||||
"""Extract AIPass branch name from a file path (src/aipass/{branch}/ pattern)."""
|
||||
parts = Path(file_path).parts
|
||||
for i, part in enumerate(parts):
|
||||
if part == "aipass" and i > 0 and parts[i - 1] == "src" and i + 1 < len(parts):
|
||||
return parts[i + 1]
|
||||
return ""
|
||||
|
||||
|
||||
def _block(reason: str) -> None:
|
||||
print(json.dumps({"decision": "block", "reason": reason}))
|
||||
sys.exit(2)
|
||||
|
||||
|
||||
def main():
|
||||
try:
|
||||
input_data = json.load(sys.stdin)
|
||||
tool_name = input_data.get("tool_name", "")
|
||||
tool_input = input_data.get("tool_input", {})
|
||||
file_path = tool_input.get("file_path", "")
|
||||
|
||||
if tool_name not in EDIT_TOOLS:
|
||||
return
|
||||
|
||||
if not file_path:
|
||||
return
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Rule 1: Inbox lock — block all writes to *.ai_mail.local/inbox.json
|
||||
# ------------------------------------------------------------------
|
||||
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\""
|
||||
)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# 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:
|
||||
if cwd_branch not in TRUSTED_CROSS_WRITERS:
|
||||
_block(
|
||||
f"Cross-branch write blocked: '{cwd_branch}' cannot write to '{target_branch}'.\n"
|
||||
f"Trusted cross-writers: {', '.join(TRUSTED_CROSS_WRITERS)}"
|
||||
)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Rule 3: State-file (original v1.2.0) — .py files only
|
||||
# ------------------------------------------------------------------
|
||||
if not file_path.endswith(".py"):
|
||||
return
|
||||
|
||||
if not STATE_FILE.exists():
|
||||
return
|
||||
|
||||
try:
|
||||
state = json.loads(STATE_FILE.read_text(encoding="utf-8"))
|
||||
except (json.JSONDecodeError, IOError):
|
||||
return
|
||||
|
||||
errored_file = state.get("file", "")
|
||||
errors = state.get("errors", [])
|
||||
|
||||
if not errors:
|
||||
return
|
||||
|
||||
try:
|
||||
current = str(Path(file_path).resolve())
|
||||
errored = str(Path(errored_file).resolve())
|
||||
except (OSError, ValueError):
|
||||
return
|
||||
|
||||
if current == errored:
|
||||
return
|
||||
|
||||
current_branch = _get_branch(current)
|
||||
errored_branch = _get_branch(errored)
|
||||
if not errored_branch:
|
||||
return
|
||||
if current_branch and errored_branch and current_branch != errored_branch:
|
||||
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}"
|
||||
)
|
||||
|
||||
except Exception:
|
||||
pass # Silent fail → allow
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -378,6 +378,69 @@ def deliver_email_to_branch(
|
||||
return True, ""
|
||||
|
||||
|
||||
def deliver_to_inbox_file(inbox_file: Path, email_data: Dict) -> Tuple[bool, str, str]:
|
||||
"""Write *email_data* to an inbox.json file and fire a desktop notification.
|
||||
|
||||
Single canonical path for direct-path delivery (used by cross-project
|
||||
reply.py to replace the raw-write backdoor). Always fires notify-send.
|
||||
|
||||
Args:
|
||||
inbox_file: Absolute path to the target inbox.json.
|
||||
email_data: Dict with at minimum ``from``, ``to``, ``subject``,
|
||||
``message``, ``timestamp``. An ``id`` key is assigned
|
||||
internally if absent.
|
||||
|
||||
Returns:
|
||||
``(success, error_msg, reply_id)`` — ``reply_id`` is the 8-char hex
|
||||
string assigned to the message (empty string on failure).
|
||||
"""
|
||||
if not inbox_file.exists():
|
||||
return False, f"inbox not found: {inbox_file}", ""
|
||||
|
||||
try:
|
||||
with _get_inbox_lock()(inbox_file):
|
||||
try:
|
||||
with open(inbox_file, "r", encoding="utf-8") as fh:
|
||||
inbox_data = json.load(fh)
|
||||
except Exception as exc:
|
||||
logger.warning("[delivery] deliver_to_inbox_file read failed %s: %s", inbox_file, exc)
|
||||
return False, f"Failed to read inbox: {exc}", ""
|
||||
|
||||
inbox_data = _migrate_inbox_format(inbox_data, inbox_file)
|
||||
|
||||
reply_id = str(uuid.uuid4())[:8]
|
||||
email_data = dict(email_data)
|
||||
email_data.setdefault("id", reply_id)
|
||||
reply_id = email_data["id"]
|
||||
|
||||
inbox_data.setdefault("messages", []).insert(0, email_data)
|
||||
inbox_data["total_messages"] = len(inbox_data["messages"])
|
||||
inbox_data["unread_count"] = sum(
|
||||
1
|
||||
for m in inbox_data["messages"]
|
||||
if m.get("status") == "new" or (m.get("status") is None and not m.get("read", False))
|
||||
)
|
||||
|
||||
try:
|
||||
with open(inbox_file, "w", encoding="utf-8") as fh:
|
||||
json.dump(inbox_data, fh, indent=2, ensure_ascii=False)
|
||||
except Exception as exc:
|
||||
logger.warning("[delivery] deliver_to_inbox_file write failed %s: %s", inbox_file, exc)
|
||||
return False, f"Failed to write inbox: {exc}", ""
|
||||
|
||||
except OSError as exc:
|
||||
logger.warning("[delivery] deliver_to_inbox_file lock failed %s: %s", inbox_file, exc)
|
||||
return False, f"Failed to acquire inbox lock: {exc}", ""
|
||||
|
||||
_send_desktop_notification(
|
||||
email_data.get("from", "@unknown"),
|
||||
email_data.get("to", str(inbox_file)),
|
||||
email_data.get("subject", ""),
|
||||
email_data.get("message", ""),
|
||||
)
|
||||
return True, "", reply_id
|
||||
|
||||
|
||||
_NOTIFICATION_TIMESTAMPS: Dict[str, List[float]] = {}
|
||||
|
||||
# Rate limit: max notifications per recipient within time window
|
||||
|
||||
@@ -20,6 +20,7 @@ from datetime import datetime
|
||||
|
||||
from aipass.prax.apps.modules.logger import system_logger as logger
|
||||
from aipass.ai_mail.apps.handlers.json import json_handler
|
||||
from aipass.ai_mail.apps.handlers.email.delivery import deliver_to_inbox_file
|
||||
|
||||
# Services imported in __main__ only (handlers should not display)
|
||||
|
||||
@@ -185,31 +186,12 @@ def _deliver_via_reply_path(
|
||||
Tuple of (success, message, reply_id or None)
|
||||
"""
|
||||
inbox_file = Path(reply_path)
|
||||
if not inbox_file.exists():
|
||||
return False, f"reply_path inbox not found: {reply_path}", None
|
||||
success, error_msg, reply_id = deliver_to_inbox_file(inbox_file, reply_email_data)
|
||||
if not success:
|
||||
logger.warning("[reply] _deliver_via_reply_path failed for %s: %s", reply_path, error_msg)
|
||||
return False, f"Failed to deliver to reply_path: {error_msg}", None
|
||||
|
||||
try:
|
||||
with open(inbox_file, "r", encoding="utf-8") as f:
|
||||
inbox_data = json.load(f)
|
||||
except Exception as e:
|
||||
logger.warning("[reply] _deliver_via_reply_path read failed %s: %s", reply_path, e)
|
||||
return False, f"Failed to read target inbox: {e}", None
|
||||
|
||||
reply_id = str(uuid.uuid4())[:8]
|
||||
reply_email_data["id"] = reply_id
|
||||
|
||||
inbox_data.setdefault("messages", []).insert(0, reply_email_data)
|
||||
inbox_data["total_messages"] = len(inbox_data["messages"])
|
||||
new_count = sum(1 for m in inbox_data["messages"] if m.get("status") == "new" or not m.get("read", False))
|
||||
inbox_data["unread_count"] = new_count
|
||||
|
||||
try:
|
||||
with open(inbox_file, "w", encoding="utf-8") as f:
|
||||
json.dump(inbox_data, f, indent=2, ensure_ascii=False)
|
||||
except Exception as e:
|
||||
logger.warning("[reply] _deliver_via_reply_path write failed %s: %s", reply_path, e)
|
||||
return False, f"Failed to write to target inbox: {e}", None
|
||||
|
||||
logger.info("[reply] Cross-project reply delivered to %s", reply_path)
|
||||
|
||||
# Save to sender's sent folder
|
||||
|
||||
@@ -20,8 +20,9 @@ from pathlib import Path
|
||||
|
||||
from aipass.prax import logger
|
||||
from aipass.drone.apps.handlers.json import json_handler
|
||||
from aipass.seedgo.apps.modules.permissions import TRUSTED_CROSS_WRITERS
|
||||
|
||||
ALLOWED_CALLERS: list[str] = ["devpulse"]
|
||||
ALLOWED_CALLERS: list[str] = list(TRUSTED_CROSS_WRITERS)
|
||||
|
||||
|
||||
def verify_caller() -> str:
|
||||
@@ -52,7 +53,7 @@ def verify_caller() -> str:
|
||||
logger.error(msg)
|
||||
raise PermissionError(msg)
|
||||
if name not in ALLOWED_CALLERS:
|
||||
msg = f"Branch '{name}' is not authorized for system-pr. Allowed callers: {ALLOWED_CALLERS}"
|
||||
msg = f"Branch '{name}' is not authorized for system-pr. Trusted cross-writers: {ALLOWED_CALLERS}"
|
||||
logger.error(msg)
|
||||
raise PermissionError(msg)
|
||||
json_handler.log_operation(
|
||||
|
||||
@@ -219,6 +219,21 @@
|
||||
"file": "templates/",
|
||||
"standard": "unused_function",
|
||||
"reason": "Template files are reference implementations for other branches to copy. They contain function definitions that are not called within seedgo itself."
|
||||
},
|
||||
{
|
||||
"file": "tests/test_hooks_track_e.py",
|
||||
"standard": "architecture",
|
||||
"reason": "Test file lives in tests/ by convention — outside the 3-layer apps/ structure by design."
|
||||
},
|
||||
{
|
||||
"file": "tests/test_hooks_track_e.py",
|
||||
"standard": "encapsulation",
|
||||
"reason": "Unit tests must import handlers directly to test them in isolation. Same pattern as test_checkers_batch5.py."
|
||||
},
|
||||
{
|
||||
"file": "apps/modules/permissions.py",
|
||||
"standard": "unused_function",
|
||||
"reason": "is_trusted_caller() and identify_caller() are public API consumed by pre_edit_gate.py (hook layer) and drone auth.py. Checker cannot trace cross-file dynamic dispatch to the hook scripts."
|
||||
}
|
||||
],
|
||||
"notes": {
|
||||
|
||||
@@ -0,0 +1,120 @@
|
||||
# =================== AIPass ====================
|
||||
# Name: inbox_audit.py
|
||||
# Description: Inbox ID validator — scans all inbox.json files for non-8-hex ids
|
||||
# Version: 1.0.0
|
||||
# Created: 2026-04-21
|
||||
# Modified: 2026-04-21
|
||||
# =============================================
|
||||
|
||||
"""Inbox ID validator for the drone @seedgo audit inbox-ids command.
|
||||
|
||||
Walks all .ai_mail.local/inbox.json files in the AIPass repo and flags any
|
||||
message ids that are not 8-character lowercase hex strings. Alerts devpulse
|
||||
when violations are found.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import List
|
||||
|
||||
from aipass.prax import logger
|
||||
from aipass.cli import console, header
|
||||
from aipass.seedgo.apps.handlers.json import json_handler
|
||||
|
||||
_HEX8_RE = re.compile(r"^[0-9a-f]{8}$")
|
||||
|
||||
|
||||
def _find_repo_root() -> Path:
|
||||
current = Path(__file__).resolve().parent
|
||||
for parent in (current, *current.parents):
|
||||
if (parent / ".git").exists():
|
||||
return parent
|
||||
return current
|
||||
|
||||
|
||||
def _scan_inbox(inbox_path: Path) -> List[dict]:
|
||||
"""Return a list of violation dicts for messages with bad ids in *inbox_path*."""
|
||||
violations: List[dict] = []
|
||||
try:
|
||||
data = json.loads(inbox_path.read_text(encoding="utf-8"))
|
||||
except Exception as exc:
|
||||
logger.warning("[inbox_audit] could not read %s: %s", inbox_path, exc)
|
||||
return violations
|
||||
|
||||
for msg in data.get("messages", []):
|
||||
msg_id = msg.get("id", "")
|
||||
if not _HEX8_RE.match(str(msg_id)):
|
||||
violations.append(
|
||||
{
|
||||
"inbox": str(inbox_path),
|
||||
"id": msg_id,
|
||||
"subject": msg.get("subject", ""),
|
||||
"from": msg.get("from", ""),
|
||||
}
|
||||
)
|
||||
return violations
|
||||
|
||||
|
||||
def _run_inbox_id_scan() -> int:
|
||||
"""Scan all inbox.json files; return number of violations found."""
|
||||
json_handler.log_operation("inbox_audit_scan", {})
|
||||
repo_root = _find_repo_root()
|
||||
inbox_files = list(repo_root.rglob(".ai_mail.local/inbox.json"))
|
||||
|
||||
console.print()
|
||||
header("SEEDGO — Inbox ID Validator")
|
||||
console.print(f"[dim]Scanning {len(inbox_files)} inbox file(s) for non-8-hex message ids...[/dim]")
|
||||
console.print()
|
||||
|
||||
all_violations: List[dict] = []
|
||||
for inbox_path in sorted(inbox_files):
|
||||
violations = _scan_inbox(inbox_path)
|
||||
all_violations.extend(violations)
|
||||
|
||||
if not all_violations:
|
||||
console.print("[green]✓[/green] All message ids are valid 8-char hex strings.")
|
||||
console.print()
|
||||
return 0
|
||||
|
||||
console.print(f"[red]✗[/red] Found [bold]{len(all_violations)}[/bold] id violation(s):\n")
|
||||
for v in all_violations:
|
||||
rel = Path(v["inbox"]).relative_to(repo_root) if Path(v["inbox"]).is_absolute() else v["inbox"]
|
||||
console.print(
|
||||
f" [red]•[/red] [bold]{rel}[/bold] id=[yellow]{v['id']!r}[/yellow] from={v['from']} subject={v['subject']!r}"
|
||||
)
|
||||
|
||||
console.print()
|
||||
console.print("[yellow]Action:[/yellow] Alert devpulse — run:")
|
||||
console.print(
|
||||
f' [green]drone @ai_mail email @devpulse "inbox-id violations" '
|
||||
f'"Found {len(all_violations)} bad message id(s) — run drone @seedgo audit inbox-ids for details"[/green]'
|
||||
)
|
||||
console.print()
|
||||
return len(all_violations)
|
||||
|
||||
|
||||
def print_introspection() -> None:
|
||||
"""Show inbox_audit module structure."""
|
||||
console.print("[bold cyan]inbox_audit[/bold cyan] — Inbox ID validator")
|
||||
console.print(" Connected Handlers: none (uses stdlib + pathlib only)")
|
||||
console.print(" Command: drone @seedgo audit inbox-ids")
|
||||
|
||||
|
||||
def handle_command(command: str, args: List[str]) -> bool:
|
||||
"""Handle `audit inbox-ids` — return True only for that exact subcommand."""
|
||||
if command not in ("audit", "standards_audit"):
|
||||
return False
|
||||
if not args:
|
||||
print_introspection()
|
||||
return True
|
||||
if args[0] in ("--help", "-h", "help"):
|
||||
console.print("Usage: drone @seedgo audit inbox-ids")
|
||||
console.print(" Scans all .ai_mail.local/inbox.json files for non-8-hex message ids.")
|
||||
return True
|
||||
if args[0] != "inbox-ids":
|
||||
return False
|
||||
_run_inbox_id_scan()
|
||||
return True
|
||||
@@ -0,0 +1,77 @@
|
||||
# =================== AIPass ====================
|
||||
# Name: permissions.py
|
||||
# Description: Shared trust list for hook layer and drone authorization
|
||||
# Version: 1.0.0
|
||||
# Created: 2026-04-21
|
||||
# Modified: 2026-04-21
|
||||
# =============================================
|
||||
|
||||
"""Shared permission definitions for cross-branch write authorization.
|
||||
|
||||
Single source of truth for which branches may write outside their own
|
||||
directory. Consumed by pre_edit_gate.py (hook layer) and
|
||||
drone/apps/plugins/devpulse_ops/auth.py (drone layer).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
from aipass.prax import logger
|
||||
from aipass.seedgo.apps.handlers.json import json_handler
|
||||
|
||||
TRUSTED_CROSS_WRITERS: tuple[str, ...] = ("devpulse", "seedgo", "spawn")
|
||||
|
||||
|
||||
def is_trusted_caller(name: str) -> bool:
|
||||
"""Return True if *name* is in TRUSTED_CROSS_WRITERS."""
|
||||
json_handler.log_operation("is_trusted_caller", {"name": name})
|
||||
return name.lower() in TRUSTED_CROSS_WRITERS
|
||||
|
||||
|
||||
def identify_caller(cwd: str | None = None) -> str:
|
||||
"""Walk up from *cwd* (default: CWD) to find passport.json, return branch_name.
|
||||
|
||||
Returns the branch name string, or empty string if no passport is found
|
||||
or the file cannot be parsed.
|
||||
"""
|
||||
start = Path(cwd).resolve() if cwd else Path.cwd().resolve()
|
||||
current = start
|
||||
for _ in range(10):
|
||||
passport = current / ".trinity" / "passport.json"
|
||||
if passport.exists():
|
||||
try:
|
||||
data = json.loads(passport.read_text(encoding="utf-8"))
|
||||
name = data.get("branch_info", {}).get("branch_name")
|
||||
if not name:
|
||||
name = data.get("identity", {}).get("name")
|
||||
return name or ""
|
||||
except Exception as exc:
|
||||
logger.warning("[permissions] identify_caller: failed to parse passport at %s: %s", passport, exc)
|
||||
return ""
|
||||
parent = current.parent
|
||||
if parent == current:
|
||||
break
|
||||
current = parent
|
||||
return ""
|
||||
|
||||
|
||||
def print_introspection() -> None:
|
||||
"""Display permissions module info."""
|
||||
from aipass.cli import console
|
||||
|
||||
console.print("[bold cyan]permissions[/bold cyan] — shared trust list for hook + drone layers")
|
||||
console.print(f" TRUSTED_CROSS_WRITERS: {TRUSTED_CROSS_WRITERS}")
|
||||
console.print(" Functions: is_trusted_caller(name), identify_caller(cwd)")
|
||||
|
||||
|
||||
def handle_command(command: str, args: list) -> bool:
|
||||
"""Library module — not a command handler. Returns False for all commands."""
|
||||
if not args:
|
||||
print_introspection()
|
||||
return False
|
||||
if args[0] in ("--help", "-h", "help"):
|
||||
print_introspection()
|
||||
return False
|
||||
return False
|
||||
@@ -0,0 +1,402 @@
|
||||
# =================== AIPass ====================
|
||||
# Name: test_hooks_track_e.py
|
||||
# Description: DPLAN-0139 Track E — single-path enforcement tests
|
||||
# Version: 1.0.0
|
||||
# Created: 2026-04-21
|
||||
# Modified: 2026-04-21
|
||||
# =============================================
|
||||
"""Tests for DPLAN-0139 Track E — single-path enforcement.
|
||||
|
||||
Covers:
|
||||
- permissions.py: TRUSTED_CROSS_WRITERS, is_trusted_caller(), identify_caller()
|
||||
- pre_edit_gate.py v1.3.0: inbox lock rule + cross-branch write rule
|
||||
- drone auth.py: ALLOWED_CALLERS derived from TRUSTED_CROSS_WRITERS
|
||||
- inbox_audit.py: handle_command routing + _scan_inbox validation
|
||||
- delivery.py: deliver_to_inbox_file single-path helper
|
||||
"""
|
||||
|
||||
import importlib.util
|
||||
import io
|
||||
import json
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _find_repo_root() -> Path:
|
||||
"""Walk up from this file to find the git repo root."""
|
||||
current = Path(__file__).resolve().parent
|
||||
for parent in (current, *current.parents):
|
||||
if (parent / ".git").exists():
|
||||
return parent
|
||||
return Path(__file__).resolve().parents[4]
|
||||
|
||||
|
||||
REPO_ROOT = _find_repo_root()
|
||||
HOOKS_DIR = REPO_ROOT / ".claude" / "hooks"
|
||||
|
||||
|
||||
def _load_hook(name: str):
|
||||
"""Import a hook script by filename via importlib (outside package)."""
|
||||
path = HOOKS_DIR / name
|
||||
if not path.exists():
|
||||
pytest.skip(f"Hook script not found: {path}")
|
||||
spec = importlib.util.spec_from_file_location(name.replace(".py", ""), path)
|
||||
assert spec is not None and spec.loader is not None
|
||||
mod = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(mod) # type: ignore[union-attr]
|
||||
return mod
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# permissions.py
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_trusted_cross_writers_contains_expected_members():
|
||||
"""TRUSTED_CROSS_WRITERS must include devpulse, seedgo, and spawn."""
|
||||
from aipass.seedgo.apps.modules.permissions import TRUSTED_CROSS_WRITERS
|
||||
|
||||
assert "devpulse" in TRUSTED_CROSS_WRITERS
|
||||
assert "seedgo" in TRUSTED_CROSS_WRITERS
|
||||
assert "spawn" in TRUSTED_CROSS_WRITERS
|
||||
|
||||
|
||||
def test_is_trusted_caller_returns_true_for_devpulse():
|
||||
"""devpulse is a trusted cross-writer."""
|
||||
from aipass.seedgo.apps.modules.permissions import is_trusted_caller
|
||||
|
||||
assert is_trusted_caller("devpulse") is True
|
||||
|
||||
|
||||
def test_is_trusted_caller_returns_true_for_seedgo():
|
||||
"""seedgo is a trusted cross-writer."""
|
||||
from aipass.seedgo.apps.modules.permissions import is_trusted_caller
|
||||
|
||||
assert is_trusted_caller("seedgo") is True
|
||||
|
||||
|
||||
def test_is_trusted_caller_returns_true_for_spawn():
|
||||
"""spawn is a trusted cross-writer."""
|
||||
from aipass.seedgo.apps.modules.permissions import is_trusted_caller
|
||||
|
||||
assert is_trusted_caller("spawn") is True
|
||||
|
||||
|
||||
def test_is_trusted_caller_returns_false_for_unknown():
|
||||
"""Regular branches are not trusted cross-writers."""
|
||||
from aipass.seedgo.apps.modules.permissions import is_trusted_caller
|
||||
|
||||
assert is_trusted_caller("flow") is False
|
||||
assert is_trusted_caller("memory") is False
|
||||
assert is_trusted_caller("random_branch") is False
|
||||
|
||||
|
||||
def test_identify_caller_returns_empty_when_no_passport(tmp_path):
|
||||
"""identify_caller returns empty string when no passport.json is found."""
|
||||
from aipass.seedgo.apps.modules.permissions import identify_caller
|
||||
|
||||
result = identify_caller(str(tmp_path))
|
||||
assert result == ""
|
||||
|
||||
|
||||
def test_identify_caller_reads_branch_name_from_passport(tmp_path):
|
||||
"""identify_caller reads branch_name from branch_info section."""
|
||||
from aipass.seedgo.apps.modules.permissions import identify_caller
|
||||
|
||||
trinity = tmp_path / ".trinity"
|
||||
trinity.mkdir()
|
||||
passport = trinity / "passport.json"
|
||||
passport.write_text(json.dumps({"branch_info": {"branch_name": "testbranch"}}), encoding="utf-8")
|
||||
result = identify_caller(str(tmp_path))
|
||||
assert result == "testbranch"
|
||||
|
||||
|
||||
def test_identify_caller_falls_back_to_identity_name(tmp_path):
|
||||
"""identify_caller falls back to identity.name when branch_info absent."""
|
||||
from aipass.seedgo.apps.modules.permissions import identify_caller
|
||||
|
||||
trinity = tmp_path / ".trinity"
|
||||
trinity.mkdir()
|
||||
passport = trinity / "passport.json"
|
||||
passport.write_text(json.dumps({"identity": {"name": "fallback_branch"}}), encoding="utf-8")
|
||||
result = identify_caller(str(tmp_path))
|
||||
assert result == "fallback_branch"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# pre_edit_gate.py v1.3.0 — Track E rules
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_gate_allows_non_edit_tool():
|
||||
"""Non-edit tools (Read) must pass through without blocking."""
|
||||
mod = _load_hook("pre_edit_gate.py")
|
||||
payload = json.dumps({"tool_name": "Read", "tool_input": {"file_path": "/tmp/foo.py"}})
|
||||
with patch("sys.stdin", io.StringIO(payload)):
|
||||
mod.main()
|
||||
|
||||
|
||||
def test_gate_blocks_inbox_json_write(capsys):
|
||||
"""Any write targeting .ai_mail.local/inbox.json must be blocked."""
|
||||
mod = _load_hook("pre_edit_gate.py")
|
||||
inbox_path = "/home/user/Projects/AIPass/src/aipass/flow/.ai_mail.local/inbox.json"
|
||||
payload = json.dumps({"tool_name": "Write", "tool_input": {"file_path": inbox_path}})
|
||||
with patch("sys.stdin", io.StringIO(payload)):
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
mod.main()
|
||||
assert exc_info.value.code == 2
|
||||
captured = capsys.readouterr()
|
||||
result = json.loads(captured.out)
|
||||
assert result["decision"] == "block"
|
||||
assert "inbox.json" in result["reason"].lower() or "drone" in result["reason"].lower()
|
||||
|
||||
|
||||
def test_gate_blocks_cross_branch_write_from_untrusted(capsys):
|
||||
"""Untrusted branch writing to a different branch must be blocked."""
|
||||
mod = _load_hook("pre_edit_gate.py")
|
||||
payload = json.dumps(
|
||||
{
|
||||
"tool_name": "Edit",
|
||||
"tool_input": {"file_path": "/repo/src/aipass/flow/apps/modules/foo.py"},
|
||||
"cwd": "/repo/src/aipass/memory",
|
||||
}
|
||||
)
|
||||
with patch("sys.stdin", io.StringIO(payload)):
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
mod.main()
|
||||
assert exc_info.value.code == 2
|
||||
captured = capsys.readouterr()
|
||||
result = json.loads(captured.out)
|
||||
assert result["decision"] == "block"
|
||||
|
||||
|
||||
def test_gate_allows_cross_branch_write_from_devpulse(capsys):
|
||||
"""devpulse may write to any branch without being blocked."""
|
||||
mod = _load_hook("pre_edit_gate.py")
|
||||
payload = json.dumps(
|
||||
{
|
||||
"tool_name": "Edit",
|
||||
"tool_input": {"file_path": "/repo/src/aipass/flow/apps/modules/foo.py"},
|
||||
"cwd": "/repo/src/aipass/devpulse",
|
||||
}
|
||||
)
|
||||
with patch("sys.stdin", io.StringIO(payload)):
|
||||
mod.main()
|
||||
captured = capsys.readouterr()
|
||||
assert captured.out == ""
|
||||
|
||||
|
||||
def test_gate_allows_cross_branch_write_from_seedgo(capsys):
|
||||
"""seedgo may write to any branch without being blocked."""
|
||||
mod = _load_hook("pre_edit_gate.py")
|
||||
payload = json.dumps(
|
||||
{
|
||||
"tool_name": "Edit",
|
||||
"tool_input": {"file_path": "/repo/src/aipass/flow/apps/modules/foo.py"},
|
||||
"cwd": "/repo/src/aipass/seedgo",
|
||||
}
|
||||
)
|
||||
with patch("sys.stdin", io.StringIO(payload)):
|
||||
mod.main()
|
||||
captured = capsys.readouterr()
|
||||
assert captured.out == ""
|
||||
|
||||
|
||||
def test_gate_allows_cross_branch_write_from_spawn(capsys):
|
||||
"""spawn may write to any branch without being blocked."""
|
||||
mod = _load_hook("pre_edit_gate.py")
|
||||
payload = json.dumps(
|
||||
{
|
||||
"tool_name": "Write",
|
||||
"tool_input": {"file_path": "/repo/src/aipass/prax/apps/modules/bar.py"},
|
||||
"cwd": "/repo/src/aipass/spawn",
|
||||
}
|
||||
)
|
||||
with patch("sys.stdin", io.StringIO(payload)):
|
||||
mod.main()
|
||||
captured = capsys.readouterr()
|
||||
assert captured.out == ""
|
||||
|
||||
|
||||
def test_gate_allows_same_branch_write(capsys):
|
||||
"""Writes within the same branch must not be blocked by the cross-branch rule."""
|
||||
mod = _load_hook("pre_edit_gate.py")
|
||||
payload = json.dumps(
|
||||
{
|
||||
"tool_name": "Edit",
|
||||
"tool_input": {"file_path": "/repo/src/aipass/flow/apps/modules/foo.py"},
|
||||
"cwd": "/repo/src/aipass/flow",
|
||||
}
|
||||
)
|
||||
with patch("sys.stdin", io.StringIO(payload)):
|
||||
mod.main()
|
||||
captured = capsys.readouterr()
|
||||
assert captured.out == ""
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# drone auth.py — ALLOWED_CALLERS derived from permissions
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_drone_auth_allowed_callers_matches_permissions():
|
||||
"""Hook and drone must reach the same decision for the same caller."""
|
||||
from aipass.drone.apps.plugins.devpulse_ops.auth import ALLOWED_CALLERS
|
||||
from aipass.seedgo.apps.modules.permissions import TRUSTED_CROSS_WRITERS
|
||||
|
||||
for branch in TRUSTED_CROSS_WRITERS:
|
||||
assert branch in ALLOWED_CALLERS, f"'{branch}' in TRUSTED_CROSS_WRITERS but missing from drone ALLOWED_CALLERS"
|
||||
|
||||
|
||||
def test_drone_auth_allowed_callers_includes_devpulse():
|
||||
"""devpulse must remain in drone ALLOWED_CALLERS."""
|
||||
from aipass.drone.apps.plugins.devpulse_ops.auth import ALLOWED_CALLERS
|
||||
|
||||
assert "devpulse" in ALLOWED_CALLERS
|
||||
|
||||
|
||||
def test_drone_auth_allowed_callers_includes_seedgo():
|
||||
"""seedgo must be in drone ALLOWED_CALLERS."""
|
||||
from aipass.drone.apps.plugins.devpulse_ops.auth import ALLOWED_CALLERS
|
||||
|
||||
assert "seedgo" in ALLOWED_CALLERS
|
||||
|
||||
|
||||
def test_drone_auth_allowed_callers_includes_spawn():
|
||||
"""spawn must be in drone ALLOWED_CALLERS."""
|
||||
from aipass.drone.apps.plugins.devpulse_ops.auth import ALLOWED_CALLERS
|
||||
|
||||
assert "spawn" in ALLOWED_CALLERS
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# inbox_audit.py — handle_command routing + _scan_inbox
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_inbox_audit_ignores_non_audit_command():
|
||||
"""handle_command returns False for non-audit command names."""
|
||||
from aipass.seedgo.apps.modules.inbox_audit import handle_command
|
||||
|
||||
assert handle_command("standards_query", ["inbox-ids"]) is False
|
||||
assert handle_command("checklist", ["inbox-ids"]) is False
|
||||
|
||||
|
||||
def test_inbox_audit_handles_inbox_ids_subcommand():
|
||||
"""handle_command returns True and runs scan for `audit inbox-ids`."""
|
||||
from aipass.seedgo.apps.modules.inbox_audit import handle_command
|
||||
|
||||
with patch("aipass.seedgo.apps.modules.inbox_audit._run_inbox_id_scan", return_value=0):
|
||||
result = handle_command("audit", ["inbox-ids"])
|
||||
assert result is True
|
||||
|
||||
|
||||
def test_inbox_audit_ignores_other_audit_subcommands():
|
||||
"""handle_command returns False for audit subcommands other than inbox-ids."""
|
||||
from aipass.seedgo.apps.modules.inbox_audit import handle_command
|
||||
|
||||
assert handle_command("audit", ["aipass"]) is False
|
||||
assert handle_command("audit", ["flow"]) is False
|
||||
|
||||
|
||||
def test_inbox_audit_scan_detects_bad_id(tmp_path):
|
||||
"""_scan_inbox flags message ids that are not 8-char lowercase hex."""
|
||||
from aipass.seedgo.apps.modules.inbox_audit import _scan_inbox
|
||||
|
||||
inbox = tmp_path / "inbox.json"
|
||||
inbox.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"messages": [
|
||||
{"id": "not-hex!", "subject": "bad", "from": "@test", "status": "new"},
|
||||
{"id": "a1b2c3d4", "subject": "ok", "from": "@test", "status": "new"},
|
||||
]
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
violations = _scan_inbox(inbox)
|
||||
assert len(violations) == 1
|
||||
assert violations[0]["id"] == "not-hex!"
|
||||
|
||||
|
||||
def test_inbox_audit_scan_passes_valid_ids(tmp_path):
|
||||
"""_scan_inbox returns empty list when all message ids are valid 8-hex."""
|
||||
from aipass.seedgo.apps.modules.inbox_audit import _scan_inbox
|
||||
|
||||
inbox = tmp_path / "inbox.json"
|
||||
inbox.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"messages": [
|
||||
{"id": "a1b2c3d4", "subject": "ok1", "from": "@x", "status": "new"},
|
||||
{"id": "deadbeef", "subject": "ok2", "from": "@y", "status": "new"},
|
||||
]
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
violations = _scan_inbox(inbox)
|
||||
assert violations == []
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# delivery.py — deliver_to_inbox_file single-path helper
|
||||
# Load by file path to avoid cross-branch package import restriction.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _load_delivery():
|
||||
"""Load ai_mail delivery.py by file path (bypasses cross-branch import check)."""
|
||||
delivery_path = REPO_ROOT / "src" / "aipass" / "ai_mail" / "apps" / "handlers" / "email" / "delivery.py"
|
||||
if not delivery_path.exists():
|
||||
pytest.skip(f"delivery.py not found: {delivery_path}")
|
||||
spec = importlib.util.spec_from_file_location("delivery", delivery_path)
|
||||
assert spec is not None and spec.loader is not None
|
||||
mod = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(mod) # type: ignore[union-attr]
|
||||
return mod
|
||||
|
||||
|
||||
def test_deliver_to_inbox_file_returns_false_for_missing_inbox(tmp_path):
|
||||
"""deliver_to_inbox_file returns (False, error, '') when inbox does not exist."""
|
||||
delivery = _load_delivery()
|
||||
missing = tmp_path / "inbox.json"
|
||||
success, error_msg, reply_id = delivery.deliver_to_inbox_file(
|
||||
missing,
|
||||
{"from": "@x", "to": "@y", "subject": "s", "message": "m", "timestamp": "t"},
|
||||
)
|
||||
assert success is False
|
||||
assert reply_id == ""
|
||||
|
||||
|
||||
def test_deliver_to_inbox_file_writes_message_and_returns_id(tmp_path):
|
||||
"""deliver_to_inbox_file writes to inbox and returns the assigned 8-char id."""
|
||||
delivery = _load_delivery()
|
||||
inbox = tmp_path / "inbox.json"
|
||||
inbox.write_text(
|
||||
json.dumps({"mailbox": "inbox", "total_messages": 0, "unread_count": 0, "messages": []}),
|
||||
encoding="utf-8",
|
||||
)
|
||||
email_data = {
|
||||
"from": "@sender",
|
||||
"to": "@recv",
|
||||
"subject": "Hello",
|
||||
"message": "body",
|
||||
"timestamp": "2026-04-21 00:00:00",
|
||||
}
|
||||
with patch.object(delivery, "_send_desktop_notification"):
|
||||
success, error_msg, reply_id = delivery.deliver_to_inbox_file(inbox, email_data)
|
||||
|
||||
assert success is True
|
||||
assert len(reply_id) == 8
|
||||
data = json.loads(inbox.read_text())
|
||||
assert len(data["messages"]) == 1
|
||||
assert data["messages"][0]["id"] == reply_id
|
||||
Reference in New Issue
Block a user