feat: add sensitive content scanner for palace drawers

mempalace/cli.py

@@ -873,6 +873,55 @@ def cmd_repair(args):
     print(f"\n{'=' * 55}\n")
 
 
+def cmd_scan(args):
+    """Scan palace drawers for sensitive content."""
+    import chromadb
+    from .scanner import scan_content, format_warnings
+
+    palace_path = os.path.expanduser(args.palace) if args.palace else MempalaceConfig().palace_path
+    try:
+        client = chromadb.PersistentClient(path=palace_path)
+        col = client.get_collection("mempalace_drawers")
+    except Exception:
+        print(f"\n  No palace found at {palace_path}")
+        return
+
+    total_count = col.count()
+    flagged_drawers = 0
+    total_findings = 0
+    total_scanned = 0
+    batch_size = 500
+
+    print(f"\n  Scanning {total_count} drawers for sensitive content...\n")
+
+    while total_scanned < total_count:
+        kwargs = {
+            "include": ["documents", "metadatas"],
+            "limit": batch_size,
+            "offset": total_scanned,
+        }
+        if args.wing:
+            kwargs["where"] = {"wing": args.wing}
+        data = col.get(**kwargs)
+        if not data["ids"]:
+            break
+
+        for drawer_id, doc, meta in zip(data["ids"], data["documents"], data["metadatas"]):
+            findings = scan_content(doc)
+            if findings:
+                flagged_drawers += 1
+                total_findings += len(findings)
+                wing = meta.get("wing", "?")
+                room = meta.get("room", "?")
+                print(f"  {drawer_id}  ({wing}/{room})")
+                print(f"    {format_warnings(findings)}")
+
+        total_scanned += len(data["ids"])
+
+    print(f"\n  Scanned {total_scanned} drawers, "
+          f"found {total_findings} sensitive patterns in {flagged_drawers} drawers.\n")
+
+
 def cmd_hook(args):
     """Run hook logic: reads JSON from stdin, outputs JSON to stdout."""
     from .hooks_cli import run_hook
@@ -1289,6 +1338,10 @@ def main():
     for instr_name in ["init", "search", "mine", "help", "status"]:
         instructions_sub.add_parser(instr_name, help=f"Output {instr_name} instructions")
 
+    # scan
+    p_scan = sub.add_parser("scan", help="Scan palace drawers for sensitive content (API keys, tokens, passwords)")
+    p_scan.add_argument("--wing", default=None, help="Only scan drawers in this wing")
+
     # repair
     p_repair = sub.add_parser(
         "repair",
@@ -1425,6 +1478,7 @@ def main():
         "mcp": cmd_mcp,
         "compress": cmd_compress,
         "wake-up": cmd_wakeup,
+        "scan": cmd_scan,
         "repair": cmd_repair,
         "repair-status": cmd_repair_status,
         "migrate": cmd_migrate,

mempalace/mcp_server.py

@@ -82,6 +82,7 @@
 )
 
 from .knowledge_graph import KnowledgeGraph, DEFAULT_KG_PATH  # noqa: E402
+from .scanner import scan_content, format_warnings  # noqa: E402
 
 logging.basicConfig(level=logging.INFO, format="%(message)s", stream=sys.stderr)
 logger = logging.getLogger("mempalace_mcp")
@@ -931,6 +932,8 @@ def tool_add_drawer(
     except Exception:
         logger.debug("Idempotency pre-check failed for %s", drawer_id, exc_info=True)
 
+    findings = scan_content(content)
+
     try:
         col.upsert(
             ids=[drawer_id],
@@ -954,7 +957,10 @@ def tool_add_drawer(
             )
         _metadata_cache = None
         logger.info(f"Filed drawer: {drawer_id} → {wing}/{room}")
-        return {"success": True, "drawer_id": drawer_id, "wing": wing, "room": room}
+        result = {"success": True, "drawer_id": drawer_id, "wing": wing, "room": room}
+        if findings:
+            result["warnings"] = format_warnings(findings)
+        return result
     except Exception as e:
         return {"success": False, "error": str(e)}
 

mempalace/miner.py

@@ -33,6 +33,8 @@
 
 logger = logging.getLogger("mempalace_mcp")
 
+from .scanner import scan_content, format_warnings
+
 READABLE_EXTENSIONS = {
     ".txt",
     ".md",
@@ -830,6 +832,13 @@ def process_file(
     if len(content) < MIN_CHUNK_SIZE:
         return 0, "general"
 
+    findings = scan_content(content)
+    if findings:
+        print(
+            f"  ⚠ {filepath.name}: {format_warnings(findings)}",
+            file=sys.stderr,
+        )
+
     room = detect_room(filepath, content, rooms, project_path)
     chunks = chunk_text(content, source_file)
 

mempalace/scanner.py

@@ -0,0 +1,61 @@
+"""
+scanner.py — Sensitive content detection for MemPalace.
+
+Scans text for common secret patterns (API keys, tokens, passwords,
+private keys) and returns findings.  Advisory only — never blocks storage.
+"""
+
+import re
+
+PATTERNS = {
+    "api_key": re.compile(
+        r"(?:sk-(?:proj-|ant-|or-)|AKIA|ghp_|gho_|github_pat_|sk_live_|sk_test_|xoxb-|xoxp-|npm_)"
+        r"[A-Za-z0-9_-]{20,}"
+    ),
+    "bearer_token": re.compile(
+        r"Bearer\s+[A-Za-z0-9_-]{20,}", re.IGNORECASE
+    ),
+    "password_assignment": re.compile(
+        r"""(?:password|passwd|pwd)["']?\s*[=:]\s*['"][^'"$][^'"]*['"]""", re.IGNORECASE
+    ),
+    "private_key": re.compile(
+        r"-----BEGIN (?:RSA |EC )?PRIVATE KEY-----"
+    ),
+    "connection_string": re.compile(
+        r"(?:mongodb|postgres|mysql|redis)://[^\s'\"]{10,}", re.IGNORECASE
+    ),
+}
+
+
+def scan_content(content):
+    """Scan content for sensitive patterns.
+
+    Returns a list of dicts: {pattern_name, start, end}.
+    Accepts None gracefully (returns empty list).
+    """
+    if not content:
+        return []
+    findings = []
+    for name, pattern in PATTERNS.items():
+        for m in pattern.finditer(content):
+            findings.append({
+                "pattern_name": name,
+                "start": m.start(),
+                "end": m.end(),
+            })
+    return findings
+
+
+def format_warnings(findings):
+    """Format findings into a human-readable warning string.
+
+    Never includes secret content — only pattern names and positions.
+    """
+    if not findings:
+        return ""
+    lines = ["WARNING: Sensitive content detected:"]
+    for f in findings:
+        lines.append(
+            f"  - {f['pattern_name']} at chars {f['start']}-{f['end']}"
+        )
+    return "\n".join(lines)