diff --git a/mempalace/cli.py b/mempalace/cli.py index 964fa84dbf..a77b4f1633 100644 --- a/mempalace/cli.py +++ b/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, diff --git a/mempalace/mcp_server.py b/mempalace/mcp_server.py index 521cb078d2..db022bde19 100644 --- a/mempalace/mcp_server.py +++ b/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)} diff --git a/mempalace/miner.py b/mempalace/miner.py index e919c58d41..5363780a4d 100644 --- a/mempalace/miner.py +++ b/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) diff --git a/mempalace/scanner.py b/mempalace/scanner.py new file mode 100644 index 0000000000..5877926068 --- /dev/null +++ b/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) diff --git a/tests/test_scanner.py b/tests/test_scanner.py new file mode 100644 index 0000000000..2036323f8f --- /dev/null +++ b/tests/test_scanner.py @@ -0,0 +1,227 @@ +""" +test_scanner.py — Tests for sensitive content detection. +""" + +from mempalace.scanner import scan_content, format_warnings + + +class TestScanContent: + # ── API key patterns ────────────────────────────────────────────── + + def test_detects_openai_api_key(self): + content = "Use this key: sk-proj-abc123def456ghi789jkl012mno345" + findings = scan_content(content) + assert len(findings) == 1 + assert findings[0]["pattern_name"] == "api_key" + + def test_detects_anthropic_api_key(self): + content = "ANTHROPIC_KEY=sk-ant-abc123def456ghi789jkl012mno345" + findings = scan_content(content) + assert len(findings) == 1 + assert findings[0]["pattern_name"] == "api_key" + + def test_detects_aws_access_key(self): + content = "AWS key is AKIAIOSFODNN7EXAMPLE1234" + findings = scan_content(content) + assert len(findings) == 1 + assert findings[0]["pattern_name"] == "api_key" + + def test_detects_github_personal_token(self): + content = "export TOKEN=ghp_ABCDEFGHIJKLMNOPQRSTuvwx" + findings = scan_content(content) + assert len(findings) == 1 + assert findings[0]["pattern_name"] == "api_key" + + def test_detects_github_oauth_token(self): + content = "token: gho_abcdefghijklmnopqrstuv" + findings = scan_content(content) + assert len(findings) == 1 + assert findings[0]["pattern_name"] == "api_key" + + def test_detects_github_pat(self): + content = "pat = github_pat_ABCDEFGHIJKLMNOPQRST1234567890" + findings = scan_content(content) + assert len(findings) == 1 + assert findings[0]["pattern_name"] == "api_key" + + def test_detects_stripe_live_key(self): + prefix = "sk_live_" + content = f"STRIPE_KEY={prefix}{'x' * 24}" + findings = scan_content(content) + assert len(findings) == 1 + assert findings[0]["pattern_name"] == "api_key" + + def test_detects_stripe_test_key(self): + prefix = "sk_test_" + content = f"key = {prefix}{'x' * 24}" + findings = scan_content(content) + assert len(findings) == 1 + assert findings[0]["pattern_name"] == "api_key" + + def test_detects_slack_bot_token(self): + content = "SLACK_TOKEN=xoxb-123456789012-abcdefghijkl" + findings = scan_content(content) + assert len(findings) == 1 + assert findings[0]["pattern_name"] == "api_key" + + def test_detects_slack_user_token(self): + content = "token: xoxp-123456789012-abcdefghijkl" + findings = scan_content(content) + assert len(findings) == 1 + assert findings[0]["pattern_name"] == "api_key" + + def test_detects_npm_token(self): + content = "//registry.npmjs.org/:_authToken=npm_abcdefghijklmnopqrst" + findings = scan_content(content) + assert len(findings) == 1 + assert findings[0]["pattern_name"] == "api_key" + + # ── Bearer token ────────────────────────────────────────────────── + + def test_detects_bearer_token(self): + content = "Authorization: Bearer eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9" + findings = scan_content(content) + assert len(findings) == 1 + assert findings[0]["pattern_name"] == "bearer_token" + + def test_detects_bearer_token_case_insensitive(self): + content = "authorization: bearer eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9" + findings = scan_content(content) + assert len(findings) == 1 + assert findings[0]["pattern_name"] == "bearer_token" + + # ── Password assignment ─────────────────────────────────────────── + + def test_detects_password_assignment(self): + content = 'db_config = {"password": "s3cret_passw0rd!"}' + findings = scan_content(content) + assert len(findings) == 1 + assert findings[0]["pattern_name"] == "password_assignment" + + def test_detects_password_with_equals(self): + content = "PASSWORD = 'hunter2_is_not_secure'" + findings = scan_content(content) + assert len(findings) == 1 + assert findings[0]["pattern_name"] == "password_assignment" + + # ── Private key ─────────────────────────────────────────────────── + + def test_detects_private_key(self): + content = "-----BEGIN RSA PRIVATE KEY-----\nMIIEpAIBAAKCAQEA..." + findings = scan_content(content) + assert len(findings) == 1 + assert findings[0]["pattern_name"] == "private_key" + + def test_detects_ec_private_key(self): + content = "-----BEGIN EC PRIVATE KEY-----\nMHQCAQEEI..." + findings = scan_content(content) + assert len(findings) == 1 + assert findings[0]["pattern_name"] == "private_key" + + def test_detects_generic_private_key(self): + content = "-----BEGIN PRIVATE KEY-----\nMIIEvgIBADANBg..." + findings = scan_content(content) + assert len(findings) == 1 + assert findings[0]["pattern_name"] == "private_key" + + # ── Connection strings ──────────────────────────────────────────── + + def test_detects_postgres_connection_string(self): + content = "DATABASE_URL=postgres://user:pass@host:5432/mydb" + findings = scan_content(content) + assert len(findings) == 1 + assert findings[0]["pattern_name"] == "connection_string" + + def test_detects_mongodb_connection_string(self): + content = "MONGO_URI=mongodb://admin:secret@cluster0.example.net/db" + findings = scan_content(content) + assert len(findings) == 1 + assert findings[0]["pattern_name"] == "connection_string" + + def test_detects_redis_connection_string(self): + content = "REDIS_URL=redis://default:mypassword@redis.example.com:6379" + findings = scan_content(content) + assert len(findings) == 1 + assert findings[0]["pattern_name"] == "connection_string" + + # ── False positives ─────────────────────────────────────────────── + + def test_no_false_positives_on_normal_text(self): + content = ( + "The authentication module uses JWT tokens for session management. " + "Tokens expire after 24 hours. We discussed the password policy " + "at the team meeting. The bearer of bad news arrived late." + ) + findings = scan_content(content) + assert len(findings) == 0 + + def test_no_false_positives_on_code(self): + content = ( + "def get_password_hash(password: str) -> str:\n" + " return bcrypt.hashpw(password.encode(), bcrypt.gensalt())\n" + ) + findings = scan_content(content) + assert len(findings) == 0 + + def test_no_false_positive_on_bare_sk_prefix(self): + content = "The variable sk-session-key-manager-helper is used for routing." + findings = scan_content(content) + assert len(findings) == 0 + + def test_no_false_positive_on_env_var_password(self): + content = 'password: "${DB_PASSWORD}"' + findings = scan_content(content) + assert len(findings) == 0 + + def test_no_false_positive_on_env_ref_password(self): + content = "password = '${MYSQL_ROOT_PASSWORD}'" + findings = scan_content(content) + assert len(findings) == 0 + + # ── Edge cases ──────────────────────────────────────────────────── + + def test_empty_content(self): + assert scan_content("") == [] + + def test_none_content(self): + assert scan_content(None) == [] + + def test_multiple_findings(self): + prefix = "sk-proj-" + content = ( + "Keys:\n" + f" OPENAI: {prefix}abcdefghijklmnopqrst1234\n" + " AWS: AKIAIOSFODNN7EXAMPLE1234\n" + " password = 'admin123_secret'\n" + ) + findings = scan_content(content) + names = [f["pattern_name"] for f in findings] + assert "api_key" in names + assert "password_assignment" in names + assert len(findings) >= 3 + + def test_findings_have_position_not_content(self): + content = "secret: sk-proj-abc123def456ghi789jkl012mno345" + findings = scan_content(content) + assert "start" in findings[0] + assert "end" in findings[0] + assert "match" not in findings[0] + + +class TestFormatWarnings: + def test_empty_findings(self): + assert format_warnings([]) == "" + + def test_single_finding(self): + findings = [{"pattern_name": "api_key", "start": 10, "end": 45}] + result = format_warnings(findings) + assert "WARNING" in result + assert "api_key" in result + assert "chars 10-45" in result + + def test_no_secret_content_in_output(self): + findings = [{"pattern_name": "api_key", "start": 0, "end": 40}] + result = format_warnings(findings) + # Should only contain pattern name and position, no actual secret + assert "sk-" not in result + assert "AKIA" not in result