diff --git a/CHANGELOG.md b/CHANGELOG.md index 0fff21de..44bad7e0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -63,6 +63,25 @@ PyPI version — not the changelog header. ### Fixed +- **Owner-capability PART 4 — devpulse's `watchdog` + `feedback` now gate on the + sealed-registry owner, and cross-project (issue #681).** Closes the + owner-capability model (#678): the last two owner-only tools were still gated + by a hardcoded `cwd.name == "devpulse"` check — which, it turns out, was a + **no-op through drone**: drone runs a routed module with `cwd=`, + so the module's own `Path.cwd()` is *always* the devpulse tree and can't + identify the caller (a `@flow` caller sailed straight through). A new shared + `handlers/owner/guard.py` resolves the *real* caller from the env drone sets + (`AIPASS_CALLER_BRANCH` / `AIPASS_CALLER_CWD`) and checks it against the sealed + owner via the frozen `is_owner(email, start_path)` contract — so it works in + any project (devpulse in AIPass, whoever owns elsewhere), not a hardcoded name. + `feedback send` stays open (it's the inbound channel any agent uses to drop + feedback to the owner); every mailbox read/manage verb is owner-only. Fail-safe: + if no owner is sealed yet (old/partial install) or the resolver can't import, + it falls back to the legacy devpulse-path heuristic so existing installs never + hard-break. Live-verified end-to-end: owner allowed, `@flow` denied on both + tools, `send` open. 18 new tests (15 guard + 3 gate), branch audit 100%. + (built + verified by devpulse) + - **seedgo `json_structure` now sanctions `custom_config/` for operator-editable config (issue #643).** The standard said "`{branch}_json/` root, one directory, no splits" and the checker ignored subdirs, so `custom_config/` (home of diff --git a/src/aipass/devpulse/.seedgo/bypass.json b/src/aipass/devpulse/.seedgo/bypass.json index bcd62676..9ed1f7f9 100644 --- a/src/aipass/devpulse/.seedgo/bypass.json +++ b/src/aipass/devpulse/.seedgo/bypass.json @@ -10,6 +10,16 @@ "file": "devpulse_json/compass", "reason": "compass/ is the devpulse-owned Compass decision store (SQLite/FTS5 — db + wal + shm) which needs its own directory. Legitimate data subdir, not operator-config (custom_config/) nor auto-gen root data. Sanctioned exception per #643." }, + { + "standard": "handlers", + "file": "apps/handlers/owner/guard.py", + "reason": "Lazy-imports is_owner/get_owner from spawn.apps.handlers.registry inside _owner_decision() — is_owner is the frozen shared owner-capability contract (#191, TDPLAN-0012); spawn is its sole home, no modules re-export. Same authorized cross-branch primitive pattern as ai_mail dispatch_monitor. Import failure falls back to the legacy heuristic. #681." + }, + { + "standard": "encapsulation", + "file": "apps/handlers/owner/guard.py", + "reason": "Lazy cross-branch import of the is_owner/get_owner resolver from spawn.apps.handlers.registry — the frozen owner-capability contract consumed identically by hooks + ai_mail. No spawn modules-level re-export exists; this is the sanctioned entry point. #681." + }, { "standard": "architecture", "reason": "No 'manager' citizen_class template in spawn. Devpulse is the only manager branch." @@ -50,6 +60,16 @@ "file": "tests/test_git_gate.py", "reason": "Test imports git_gate.py from ~/.claude/hooks/ via importlib — external hook, not a branch module." }, + { + "standard": "encapsulation", + "file": "tests/test_feedback_module.py", + "reason": "Router test imports the feedback storage handler directly to arrange/assert inbox state (save_inbox/load_inbox) — standard test-fixture access, same pattern as test_feedback_storage.py / test_compass_store.py. #681." + }, + { + "standard": "encapsulation", + "file": "tests/test_owner_guard.py", + "reason": "Unit test imports the owner guard handler (its SUT) and patches spawn.registry's get_owner/is_owner — the guard is a shared handler primitive with no apps/modules/ command entry point. Same direct-SUT pattern as test_compass_store.py. #681." + }, { "standard": "encapsulation", "file": "tools/spot_check.py", diff --git a/src/aipass/devpulse/apps/handlers/owner/__init__.py b/src/aipass/devpulse/apps/handlers/owner/__init__.py new file mode 100644 index 00000000..caeb032b --- /dev/null +++ b/src/aipass/devpulse/apps/handlers/owner/__init__.py @@ -0,0 +1,7 @@ +# =================== AIPass ==================== +# Name: __init__.py +# Description: Owner-capability guard handlers package +# Version: 1.0.0 +# Created: 2026-07-10 +# Modified: 2026-07-10 +# ============================================= diff --git a/src/aipass/devpulse/apps/handlers/owner/guard.py b/src/aipass/devpulse/apps/handlers/owner/guard.py new file mode 100644 index 00000000..f0a718a9 --- /dev/null +++ b/src/aipass/devpulse/apps/handlers/owner/guard.py @@ -0,0 +1,115 @@ +# =================== AIPass ==================== +# Name: guard.py +# Description: Owner-capability caller guard for devpulse owner-only tools +# Version: 1.0.0 +# Created: 2026-07-10 +# Modified: 2026-07-10 +# ============================================= + +""" +Owner-capability guard for devpulse's owner-only tools. + +watchdog and feedback's mailbox-management verbs are the project OWNER's +tools. The catch: drone runs a routed module with ``cwd=`` (see +drone router_handler), so the module's own ``Path.cwd()`` is ALWAYS the +devpulse tree and can't identify who called. The real caller lives in the env +drone sets — ``AIPASS_CALLER_BRANCH`` / ``AIPASS_CALLER_CWD``. This resolves +that caller and checks it against the sealed-registry owner via ``is_owner``. + +Portable by construction: ``is_owner`` reads each project's OWN sealed +registry, so "owner" is devpulse in AIPass and whoever owns elsewhere (e.g. +@vera in Vera Studio) — no hardcoded name, no per-project scaffolding. + +Fail-safe: if the owner resolver is unavailable (import fails, or a project +has no sealed owner yet — an old/partial install), it falls back to the legacy +devpulse-path heuristic so existing installs never hard-break. Concretely #681. + +Returns a plain bool — the calling MODULE owns user-facing output (handlers +don't print). Denials are audit-logged here. +""" + +import os +from pathlib import Path + +from aipass.prax import logger +from aipass.devpulse.apps.handlers.json import json_handler + + +def _resolve_caller() -> tuple[str, Path]: + """Resolve ``(caller_email, caller_cwd)`` from the env drone sets. + + ``caller_email`` is ``@`` — a branch's address is ``@`` + its + directory name (matches ai_mail's identity resolution). Prefers the + ``AIPASS_CALLER_BRANCH`` env var; otherwise walks up ``AIPASS_CALLER_CWD`` + for a ``.trinity/passport.json`` and uses that directory's name. Falls back + to the process cwd when no caller env is set (direct, non-drone invocation). + + Returns: + tuple: (caller_email_or_empty, caller_cwd_path) + """ + caller_cwd_env = os.environ.get("AIPASS_CALLER_CWD", "") + caller_cwd = Path(caller_cwd_env) if caller_cwd_env else Path.cwd() + + branch = os.environ.get("AIPASS_CALLER_BRANCH", "") + if not branch: + for candidate in [caller_cwd, *caller_cwd.parents]: + if (candidate / ".trinity" / "passport.json").exists(): + branch = candidate.name + break + + email = f"@{branch.lstrip('@').lower()}" if branch else "" + return email, caller_cwd + + +def _legacy_devpulse_heuristic(caller_cwd: Path) -> bool: + """Pre-owner behavior: allow only a caller standing in the devpulse tree. + + Used solely as the fail-safe when the owner resolver can't decide, so + existing AIPass installs keep working before the sealed owner is present. + """ + return caller_cwd.name == "devpulse" or any(p.name == "devpulse" for p in caller_cwd.parents) + + +def _owner_decision(email: str, caller_cwd: Path) -> bool: + """Decide whether ``email`` is the owner of the project at ``caller_cwd``. + + Owner check runs against the caller's OWN project registry (start_path = + caller_cwd), so cross-project calls resolve the caller's owner, not + AIPass's. Falls back to the legacy heuristic when the resolver is + unavailable or the project has no sealed owner yet. + """ + try: + # is_owner is the frozen shared owner-capability contract (spawn is its + # sole home; no modules re-export). Lazy import keeps cold start fast and + # enables the fail-safe fallback below. + from aipass.spawn.apps.handlers.registry import get_owner, is_owner + except ImportError as exc: + logger.warning("[owner_guard] owner resolver unavailable (%s) — legacy heuristic", exc) + return _legacy_devpulse_heuristic(caller_cwd) + + if get_owner(start_path=caller_cwd) is None: + # No sealed owner in this project -> resolver can't decide -> legacy path check. + logger.info("[owner_guard] no sealed owner at %s — legacy heuristic", caller_cwd) + return _legacy_devpulse_heuristic(caller_cwd) + + return bool(email) and is_owner(email, start_path=caller_cwd) + + +def guard_owner_caller(tool: str) -> bool: + """Gate an owner-only tool. + + Args: + tool: Name of the calling tool (e.g. 'watchdog', 'feedback') for the + audit line. The caller is responsible for any user-facing message. + + Returns: + bool: True to allow the call; False (after audit-logging the denial) to + reject a non-owner caller. + """ + email, caller_cwd = _resolve_caller() + if _owner_decision(email, caller_cwd): + return True + + json_handler.log_operation("owner_guard_denied", {"tool": tool, "caller": email or "unknown"}) + logger.info("[owner_guard] %s denied non-owner caller=%s", tool, email or "unknown") + return False diff --git a/src/aipass/devpulse/apps/modules/feedback.py b/src/aipass/devpulse/apps/modules/feedback.py index 876c3b9f..79b2192f 100644 --- a/src/aipass/devpulse/apps/modules/feedback.py +++ b/src/aipass/devpulse/apps/modules/feedback.py @@ -27,7 +27,7 @@ from aipass.devpulse.apps.handlers.feedback.compose import ( ) from aipass.prax import logger -from aipass.cli.apps.modules import err_console, error +from aipass.cli.apps.modules import err_console, error, warning from aipass.devpulse.apps.handlers.json import json_handler console = err_console @@ -47,6 +47,21 @@ HELP_TEXT = """\ """ +def _guard_caller() -> bool: + """Owner-only gate for mailbox reads/management (see handlers.owner.guard). + + The mailbox belongs to the project owner. `send` and `--help` stay open + (send is the inbound channel any agent uses to drop feedback here); every + other verb reads or mutates the owner's mail and is owner-gated. #681. + """ + from aipass.devpulse.apps.handlers.owner.guard import guard_owner_caller + + if guard_owner_caller("feedback"): + return True + warning("feedback mailbox management is owner-only — refusing non-owner call") + return False + + def print_introspection() -> None: """Display module introspection info.""" console.print() @@ -73,6 +88,20 @@ def handle_command(command: str, args: list[str]) -> bool: if command != "feedback": return False + # Open verbs (no owner gate): help + `send`. `send` is the inbound channel + # any agent uses to drop feedback into the owner's mailbox. Everything else + # reads or manages that mailbox -> owner-only (#681). + if args and args[0] in ("--help", "-h", "help"): + console.print(HELP_TEXT) + return True + + if args and args[0] == "send": + json_handler.log_operation("feedback_command", {"subcommand": "send"}) + return _handle_send(args[1:]) + + if not _guard_caller(): + return True + if not args: print_introspection() summary = get_summary() @@ -83,10 +112,6 @@ def handle_command(command: str, args: list[str]) -> bool: sub_args = args[1:] json_handler.log_operation("feedback_command", {"subcommand": subcommand}) - if subcommand in ("--help", "-h", "help"): - console.print(HELP_TEXT) - return True - if subcommand == "inbox": list_messages() return True @@ -107,9 +132,6 @@ def handle_command(command: str, args: list[str]) -> bool: reply_to(msg_id, body) return True - if subcommand == "send": - return _handle_send(sub_args) - if subcommand == "clear": if not sub_args: error("Usage: feedback clear | feedback clear --all") diff --git a/src/aipass/devpulse/apps/modules/watchdog.py b/src/aipass/devpulse/apps/modules/watchdog.py index cb169eb8..58141481 100644 --- a/src/aipass/devpulse/apps/modules/watchdog.py +++ b/src/aipass/devpulse/apps/modules/watchdog.py @@ -25,7 +25,6 @@ See FPLAN-0186 for the build plan and DPLAN-0130 for the design record. """ import importlib -from pathlib import Path from typing import List from aipass.prax.apps.modules.logger import system_logger as logger @@ -113,11 +112,18 @@ def print_introspection() -> None: def _guard_caller() -> bool: - """Reject cross-branch invocation. Devpulse-only tool.""" - cwd = Path.cwd() - if cwd.name == "devpulse" or any(p.name == "devpulse" for p in cwd.parents): + """Reject non-owner invocation. Owner-only tool. + + Gates on the PROJECT OWNER (sealed registry) via the shared owner guard, so + it works across projects — not a hardcoded 'devpulse' name. (Drone runs a + routed module with cwd=, so Path.cwd() can't identify the real + caller; the guard reads the AIPASS_CALLER_* env drone sets.) #681. + """ + from aipass.devpulse.apps.handlers.owner.guard import guard_owner_caller + + if guard_owner_caller("watchdog"): return True - warning("watchdog is a devpulse-only module — refusing cross-branch call") + warning("watchdog is an owner-only module — refusing non-owner call") return False diff --git a/src/aipass/devpulse/tests/test_feedback_module.py b/src/aipass/devpulse/tests/test_feedback_module.py index 607803b1..646461c5 100644 --- a/src/aipass/devpulse/tests/test_feedback_module.py +++ b/src/aipass/devpulse/tests/test_feedback_module.py @@ -1,7 +1,10 @@ -# META -# module: devpulse.feedback -# description: Tests for feedback module command routing -# END META +# =================== AIPass ==================== +# Name: test_feedback_module.py +# Description: Tests for feedback module command routing +# Version: 1.0.0 +# Created: 2026-04-11 +# Modified: 2026-07-10 +# ============================================= """Tests for feedback module — command routing via handle_command().""" @@ -13,6 +16,13 @@ from aipass.devpulse.apps.handlers.feedback import storage from aipass.devpulse.apps.modules import feedback as feedback_module +@pytest.fixture(autouse=True) +def _bypass_caller_guard(): + """Force _guard_caller to pass so routing tests don't depend on owner env.""" + with patch.object(feedback_module, "_guard_caller", return_value=True): + yield + + @pytest.fixture def mock_feedback_dir(tmp_path): """Patch FEEDBACK_DIR to use tmp_path for isolation.""" @@ -183,6 +193,35 @@ class TestCommandRouting: assert result is True # Handled (shows error + hint) +class TestOwnerGate: + """Owner gate wraps mailbox management; send + help stay open (#681).""" + + def test_management_blocked_for_non_owner(self, populated_inbox): + """A denied guard blocks a management verb — view does not mark read.""" + with patch.object(feedback_module, "_guard_caller", return_value=False): + result = feedback_module.handle_command("feedback", ["view", "aaa11111"]) + assert result is True # command still "handled" (clean refusal) + + data = storage.load_inbox() + msg = next(m for m in data["messages"] if m["id"] == "aaa11111") + assert msg["read"] is False # action was gated out + + def test_send_open_for_non_owner(self, empty_inbox): + """send bypasses the owner gate — any agent can drop feedback.""" + with patch.object(feedback_module, "_guard_caller", return_value=False): + result = feedback_module.handle_command("feedback", ["send", "seedgo", "Bug report", "Found an issue"]) + assert result is True + + data = storage.load_inbox() + assert data["total_messages"] == 1 # send bypassed the gate + + def test_help_open_for_non_owner(self, empty_inbox): + """--help bypasses the owner gate.""" + with patch.object(feedback_module, "_guard_caller", return_value=False): + result = feedback_module.handle_command("feedback", ["--help"]) + assert result is True + + class TestHandleCommandHasCorrectSignature: """Verify handle_command meets auto-discovery requirements.""" diff --git a/src/aipass/devpulse/tests/test_owner_guard.py b/src/aipass/devpulse/tests/test_owner_guard.py new file mode 100644 index 00000000..7f901ec9 --- /dev/null +++ b/src/aipass/devpulse/tests/test_owner_guard.py @@ -0,0 +1,178 @@ +# =================== AIPass ==================== +# Name: test_owner_guard.py +# Description: Tests for the shared owner-capability caller guard (#681) +# Version: 1.0.0 +# Created: 2026-07-10 +# Modified: 2026-07-10 +# ============================================= + +"""Tests for the shared owner-capability caller guard (handlers/owner/guard.py). + +Covers caller resolution from the drone env, the owner decision against a +(patched) sealed registry — including cross-project ownership — and the +legacy fail-safe fallback when no owner is sealed. +""" + +from pathlib import Path + +import pytest + +from aipass.devpulse.apps.handlers.owner import guard as guard_mod + + +@pytest.fixture(autouse=True) +def _clear_caller_env(monkeypatch): + """Start each test with no caller env; tests set exactly what they need.""" + monkeypatch.delenv("AIPASS_CALLER_BRANCH", raising=False) + monkeypatch.delenv("AIPASS_CALLER_CWD", raising=False) + + +def _patch_registry(monkeypatch, owner_email): + """Patch spawn's owner resolver. owner_email=None => no sealed owner.""" + import aipass.spawn.apps.handlers.registry as reg + + def fake_get_owner(start_path=None): + """Stand-in for get_owner: owner entry dict, or None when unsealed.""" + return {"email": owner_email} if owner_email else None + + def _norm(email): + """Normalize an email to lowercase with a leading '@'.""" + return (email if email.startswith("@") else f"@{email}").lower() + + def fake_is_owner(email, start_path=None): + """Stand-in for is_owner: True iff email matches the sealed owner.""" + if not owner_email or not email: + return False + return _norm(email) == _norm(owner_email) + + monkeypatch.setattr(reg, "get_owner", fake_get_owner) + monkeypatch.setattr(reg, "is_owner", fake_is_owner) + + +# ------------------------------------------------------------------ +# _resolve_caller +# ------------------------------------------------------------------ + + +class TestResolveCaller: + """Caller identity resolution from the env drone sets.""" + + def test_uses_branch_env(self, monkeypatch, tmp_path): + """AIPASS_CALLER_BRANCH becomes the '@branch' email directly.""" + monkeypatch.setenv("AIPASS_CALLER_BRANCH", "flow") + monkeypatch.setenv("AIPASS_CALLER_CWD", str(tmp_path)) + email, cwd = guard_mod._resolve_caller() + assert email == "@flow" + assert cwd == tmp_path + + def test_strips_and_lowercases(self, monkeypatch, tmp_path): + """A '@Mixed' branch env normalizes to lowercase, single leading '@'.""" + monkeypatch.setenv("AIPASS_CALLER_BRANCH", "@DevPulse") + monkeypatch.setenv("AIPASS_CALLER_CWD", str(tmp_path)) + email, _ = guard_mod._resolve_caller() + assert email == "@devpulse" + + def test_walks_up_for_passport(self, monkeypatch, tmp_path): + """With no branch env, walk up the caller cwd to the passport dir name.""" + branch = tmp_path / "mybranch" + (branch / ".trinity").mkdir(parents=True) + (branch / ".trinity" / "passport.json").write_text("{}", encoding="utf-8") + sub = branch / "apps" / "modules" + sub.mkdir(parents=True) + monkeypatch.setenv("AIPASS_CALLER_CWD", str(sub)) + email, cwd = guard_mod._resolve_caller() + assert email == "@mybranch" + assert cwd == sub + + def test_empty_when_no_branch_and_no_passport(self, monkeypatch, tmp_path): + """No branch env and no passport anywhere above => empty email.""" + monkeypatch.setenv("AIPASS_CALLER_CWD", str(tmp_path)) + email, _ = guard_mod._resolve_caller() + assert email == "" + + +# ------------------------------------------------------------------ +# _legacy_devpulse_heuristic +# ------------------------------------------------------------------ + + +class TestLegacyHeuristic: + """The pre-owner fail-safe: allow only a caller in the devpulse tree.""" + + def test_allows_devpulse_subtree(self, tmp_path): + """A path with a 'devpulse' ancestor is allowed.""" + assert guard_mod._legacy_devpulse_heuristic(tmp_path / "devpulse" / "apps") is True + + def test_allows_devpulse_leaf(self, tmp_path): + """A path whose leaf dir is 'devpulse' is allowed.""" + assert guard_mod._legacy_devpulse_heuristic(tmp_path / "devpulse") is True + + def test_rejects_other_branch(self, tmp_path): + """A path with no 'devpulse' component is rejected.""" + assert guard_mod._legacy_devpulse_heuristic(tmp_path / "flow" / "apps") is False + + +# ------------------------------------------------------------------ +# _owner_decision (patched resolver) +# ------------------------------------------------------------------ + + +class TestOwnerDecision: + """The core owner check against a (patched) sealed registry.""" + + def test_allows_owner(self, monkeypatch, tmp_path): + """The sealed owner's email is allowed.""" + _patch_registry(monkeypatch, "@devpulse") + assert guard_mod._owner_decision("@devpulse", tmp_path) is True + + def test_rejects_non_owner(self, monkeypatch, tmp_path): + """A non-owner email is rejected when an owner is sealed.""" + _patch_registry(monkeypatch, "@devpulse") + assert guard_mod._owner_decision("@flow", tmp_path) is False + + def test_cross_project_owner(self, monkeypatch, tmp_path): + """Owner is per-project: @vera owns elsewhere, devpulse does not.""" + _patch_registry(monkeypatch, "@vera") + assert guard_mod._owner_decision("@vera", tmp_path) is True + assert guard_mod._owner_decision("@devpulse", tmp_path) is False + + def test_no_sealed_owner_falls_back_to_heuristic(self, monkeypatch): + """No sealed owner => legacy devpulse-path heuristic decides.""" + _patch_registry(monkeypatch, None) # get_owner -> None + assert guard_mod._owner_decision("@anyone", Path("/x/devpulse/y")) is True + assert guard_mod._owner_decision("@anyone", Path("/x/flow/y")) is False + + def test_empty_email_rejected_when_owner_sealed(self, monkeypatch, tmp_path): + """An unresolved caller (empty email) is rejected when owner is sealed.""" + _patch_registry(monkeypatch, "@devpulse") + assert guard_mod._owner_decision("", tmp_path) is False + + +# ------------------------------------------------------------------ +# guard_owner_caller / is_owner_caller (end to end via env) +# ------------------------------------------------------------------ + + +class TestGuardOwnerCaller: + """End-to-end gate behavior driven by the drone caller env.""" + + def test_allows_owner(self, monkeypatch, tmp_path): + """Owner caller => guard allows.""" + _patch_registry(monkeypatch, "@devpulse") + monkeypatch.setenv("AIPASS_CALLER_BRANCH", "devpulse") + monkeypatch.setenv("AIPASS_CALLER_CWD", str(tmp_path)) + assert guard_mod.guard_owner_caller("watchdog") is True + + def test_denies_non_owner(self, monkeypatch, tmp_path): + """Non-owner caller => guard denies (and audit-logs).""" + _patch_registry(monkeypatch, "@devpulse") + monkeypatch.setenv("AIPASS_CALLER_BRANCH", "flow") + monkeypatch.setenv("AIPASS_CALLER_CWD", str(tmp_path)) + assert guard_mod.guard_owner_caller("feedback") is False + + def test_cross_project_owner_end_to_end(self, monkeypatch, tmp_path): + """A non-devpulse owner (@vera) is allowed in its own project.""" + _patch_registry(monkeypatch, "@vera") + monkeypatch.setenv("AIPASS_CALLER_BRANCH", "vera") + monkeypatch.setenv("AIPASS_CALLER_CWD", str(tmp_path)) + assert guard_mod.guard_owner_caller("watchdog") is True