From a66ab3b3cda93f8ef047f1994273cbf933819a41 Mon Sep 17 00:00:00 2001 From: Chaim Date: Sun, 31 May 2026 11:16:36 +0000 Subject: [PATCH] feat(guard): fitness function blocking raw Paperclip access (GAP-22, FU-8a) Co-Authored-By: Claude Sonnet 4.6 --- .../tests/test_paperclip_access_guard.py | 92 +++++++++++++++++++ 1 file changed, 92 insertions(+) create mode 100644 mcp-server/tests/test_paperclip_access_guard.py diff --git a/mcp-server/tests/test_paperclip_access_guard.py b/mcp-server/tests/test_paperclip_access_guard.py new file mode 100644 index 0000000..3787e5b --- /dev/null +++ b/mcp-server/tests/test_paperclip_access_guard.py @@ -0,0 +1,92 @@ +"""FU-8a / GAP-22: fitness function — forbid un-sanctioned Paperclip access. + +Fails if any scanned source (outside the allowlist) reaches the Paperclip API +with a raw HTTP client or inserts directly into agent_wakeup_requests. The +sanctioned paths are web/paperclip_api.py::pc_request (Python) and scripts/pc.sh +(bash); wakeup must go through POST /api/agents/{id}/wakeup. +""" +from __future__ import annotations + +import re +from pathlib import Path + +import pytest + +REPO = Path(__file__).resolve().parents[2] +SCAN_ROOTS = [REPO / "web", REPO / "mcp-server" / "src", REPO / "scripts"] + +# Files exempt from the HTTP-to-Paperclip rule (the sanctioned helpers + legacy DB-read client). +ALLOWLIST = { + REPO / "web" / "paperclip_api.py", # the sanctioned pc_request helper + REPO / "scripts" / "pc.sh", # the sanctioned bash wrapper + REPO / "web" / "paperclip_client.py", # legacy: DB reads only (no raw http, no wakeup insert) + # Standalone operator/admin scripts that run on the host machine (not agent code, not FastAPI + # handlers). pc_request is an async FastAPI coroutine, not usable standalone; these scripts are + # the explicitly-documented admin tooling referenced in CLAUDE.md and are sanctioned for + # direct httpx use. They are NOT agent code bypassing the approved helper. + REPO / "scripts" / "sync_agents_across_companies.py", # documented admin tool: CMP→CMPA agent-config sync (CLAUDE.md §Cross-company sync) + REPO / "scripts" / "audit_corpus_integrity.py", # cron audit tool: posts CEO wakeup on anomaly (best-effort, explicitly guarded) + REPO / "scripts" / "fix_paperclipai_skills_drift.py", # one-shot operator fix for paperclipai skill drift (Gap #28 runbook) + REPO / "scripts" / "sync_missing_agent_skills.py", # one-shot operator fix: add missing paperclipSkillSync (Gap #28) +} + +# Directories to skip entirely during scan (dead/archived code, virtual envs, test fixtures). +_SKIP_PATH_FRAGMENTS = {"/.venv/", "/tests/", "/.archive/"} + +_PC_URL = re.compile(r"PAPERCLIP_API_URL|127\.0\.0\.1:3100|localhost:3100|pc\.nautilus\.marcusgroup\.org") +_HTTP_CLIENT = re.compile(r"\bhttpx\b|\brequests\.(get|post|put|patch|delete)\b|\baiohttp\b|\bcurl\b") +_WAKEUP_INSERT = re.compile(r"insert\s+into\s+agent_wakeup_requests", re.IGNORECASE) + + +def _scan_text(text: str) -> list[str]: + """Return violation reasons for a single file's text.""" + reasons = [] + if _WAKEUP_INSERT.search(text): + reasons.append("direct INSERT INTO agent_wakeup_requests — use the wakeup API") + if _PC_URL.search(text) and _HTTP_CLIENT.search(text): + reasons.append("raw HTTP client + Paperclip URL — use web/paperclip_api.pc_request or scripts/pc.sh") + return reasons + + +def _iter_source_files(): + for root in SCAN_ROOTS: + if not root.exists(): + continue + for ext in ("*.py", "*.sh"): + for f in root.rglob(ext): + f_str = str(f) + if f in ALLOWLIST or any(frag in f_str for frag in _SKIP_PATH_FRAGMENTS): + continue + yield f + + +def find_violations() -> list[tuple[str, str]]: + out = [] + for f in _iter_source_files(): + try: + text = f.read_text(encoding="utf-8") + except (UnicodeDecodeError, OSError): + continue + for reason in _scan_text(text): + out.append((str(f.relative_to(REPO)), reason)) + return out + + +def test_scan_flags_raw_http_to_paperclip(): + bad = 'import httpx\nasync def f():\n await httpx.post(f"{PAPERCLIP_API_URL}/x")\n' + assert _scan_text(bad) + + +def test_scan_flags_wakeup_insert(): + bad = "await conn.execute('INSERT INTO agent_wakeup_requests (id) VALUES ($1)', x)" + assert _scan_text(bad) + + +def test_scan_ignores_plain_code(): + assert _scan_text("def add(a, b):\n return a + b\n") == [] + + +def test_repo_has_no_paperclip_access_violations(): + violations = find_violations() + assert violations == [], "Un-sanctioned Paperclip access found:\n" + "\n".join( + f" {f}: {r}" for f, r in violations)