security+perf: harden inputs, fix command injection, optimize DB access

hooks/mempal_save_hook.sh

@@ -64,13 +64,20 @@ MEMPAL_DIR=""
 # Read JSON input from stdin
 INPUT=$(cat)
 
-# Parse fields from Claude Code's JSON
-SESSION_ID=$(echo "$INPUT" | python3 -c "import sys,json; print(json.load(sys.stdin).get('session_id','unknown'))" 2>/dev/null)
-# Sanitize SESSION_ID to prevent path traversal (only allow alnum, dash, underscore)
-SESSION_ID=$(echo "$SESSION_ID" | tr -cd 'a-zA-Z0-9_-')
-[ -z "$SESSION_ID" ] && SESSION_ID="unknown"
-STOP_HOOK_ACTIVE=$(echo "$INPUT" | python3 -c "import sys,json; print(json.load(sys.stdin).get('stop_hook_active', False))" 2>/dev/null)
-TRANSCRIPT_PATH=$(echo "$INPUT" | python3 -c "import sys,json; print(json.load(sys.stdin).get('transcript_path',''))" 2>/dev/null)
+# Parse all fields in a single Python call (3x faster than separate invocations)
+eval $(echo "$INPUT" | python3 -c "
+import sys, json
+data = json.load(sys.stdin)
+sid = data.get('session_id', 'unknown')
+sha = data.get('stop_hook_active', False)
+tp = data.get('transcript_path', '')
+# Shell-safe output — only allow alphanumeric, underscore, hyphen, slash, dot, tilde
+import re
+safe = lambda s: re.sub(r'[^a-zA-Z0-9_/.\-~]', '', str(s))
+print(f'SESSION_ID=\"{safe(sid)}\"')
+print(f'STOP_HOOK_ACTIVE=\"{sha}\"')
+print(f'TRANSCRIPT_PATH=\"{safe(tp)}\"')
+" 2>/dev/null)
 
 # Expand ~ in path
 TRANSCRIPT_PATH="${TRANSCRIPT_PATH/#\~/$HOME}"
@@ -83,6 +90,7 @@ if [ "$STOP_HOOK_ACTIVE" = "True" ] || [ "$STOP_HOOK_ACTIVE" = "true" ]; then
 fi
 
 # Count human messages in the JSONL transcript
+# SECURITY: Pass transcript path as sys.argv to avoid shell injection via crafted paths
 if [ -f "$TRANSCRIPT_PATH" ]; then
     EXCHANGE_COUNT=$(python3 - "$TRANSCRIPT_PATH" <<'PYEOF'
 import json, sys
@@ -94,7 +102,6 @@ with open(sys.argv[1]) as f:
             msg = entry.get('message', {})
             if isinstance(msg, dict) and msg.get('role') == 'user':
                 content = msg.get('content', '')
-                # Skip system/command messages — only count real human input
                 if isinstance(content, str) and '<command-message>' in content:
                     continue
                 count += 1

mempalace/config.py

@@ -6,8 +6,53 @@
 
 import json
 import os
+import re
 from pathlib import Path
 
+
+# ── Input validation ──────────────────────────────────────────────────────────
+# Shared sanitizers for wing/room/entity names. Prevents path traversal,
+# excessively long strings, and special characters that could cause issues
+# in file paths, SQLite, or ChromaDB metadata.
+
+MAX_NAME_LENGTH = 128
+_SAFE_NAME_RE = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9_ .'-]{0,126}[a-zA-Z0-9]?$")
+
+
+def sanitize_name(value: str, field_name: str = "name") -> str:
+    """Validate and sanitize a wing/room/entity name.
+
+    Raises ValueError if the name is invalid.
+    """
+    if not isinstance(value, str) or not value.strip():
+        raise ValueError(f"{field_name} must be a non-empty string")
+
+    value = value.strip()
+
+    if len(value) > MAX_NAME_LENGTH:
+        raise ValueError(f"{field_name} exceeds maximum length of {MAX_NAME_LENGTH} characters")
+
+    # Block path traversal
+    if ".." in value or "/" in value or "\\" in value:
+        raise ValueError(f"{field_name} contains invalid path characters")
+
+    # Block null bytes
+    if "\x00" in value:
+        raise ValueError(f"{field_name} contains null bytes")
+
+    return value
+
+
+def sanitize_content(value: str, max_length: int = 100_000) -> str:
+    """Validate drawer/diary content length."""
+    if not isinstance(value, str) or not value.strip():
+        raise ValueError("content must be a non-empty string")
+    if len(value) > max_length:
+        raise ValueError(f"content exceeds maximum length of {max_length} characters")
+    if "\x00" in value:
+        raise ValueError("content contains null bytes")
+    return value
+
 DEFAULT_PALACE_PATH = os.path.expanduser("~/.mempalace/palace")
 DEFAULT_COLLECTION_NAME = "mempalace_drawers"
 
@@ -126,6 +171,11 @@ def hall_keywords(self):
     def init(self):
         """Create config directory and write default config.json if it doesn't exist."""
         self._config_dir.mkdir(parents=True, exist_ok=True)
+        # Restrict directory permissions to owner only (Unix)
+        try:
+            self._config_dir.chmod(0o700)
+        except (OSError, NotImplementedError):
+            pass  # Windows doesn't support Unix permissions
         if not self._config_file.exists():
             default_config = {
                 "palace_path": DEFAULT_PALACE_PATH,
@@ -135,6 +185,11 @@ def init(self):
             }
             with open(self._config_file, "w") as f:
                 json.dump(default_config, f, indent=2)
+            # Restrict config file to owner read/write only
+            try:
+                self._config_file.chmod(0o600)
+            except (OSError, NotImplementedError):
+                pass
         return self._config_file
 
     def save_people_map(self, people_map):

mempalace/convo_miner.py

@@ -18,6 +18,7 @@
 import chromadb
 
 from .normalize import normalize
+from .palace import SKIP_DIRS, get_collection, file_already_mined
 
 
 # File types that might contain conversations
@@ -28,22 +29,8 @@
     ".jsonl",
 }
 
-SKIP_DIRS = {
-    ".git",
-    "node_modules",
-    "__pycache__",
-    ".venv",
-    "venv",
-    "env",
-    "dist",
-    "build",
-    ".next",
-    ".mempalace",
-    "tool-results",
-    "memory",
-}
-
 MIN_CHUNK_SIZE = 30
+MAX_FILE_SIZE = 10 * 1024 * 1024  # 10 MB — skip files larger than this
 
 
 # =============================================================================
@@ -211,23 +198,6 @@ def detect_convo_room(content: str) -> str:
 # =============================================================================
 
 
-def get_collection(palace_path: str):
-    os.makedirs(palace_path, exist_ok=True)
-    client = chromadb.PersistentClient(path=palace_path)
-    try:
-        return client.get_collection("mempalace_drawers")
-    except Exception:
-        return client.create_collection("mempalace_drawers")
-
-
-def file_already_mined(collection, source_file: str) -> bool:
-    try:
-        results = collection.get(where={"source_file": source_file}, limit=1)
-        return len(results.get("ids", [])) > 0
-    except Exception:
-        return False
-
-
 # =============================================================================
 # SCAN FOR CONVERSATION FILES
 # =============================================================================
@@ -244,6 +214,14 @@ def scan_convos(convo_dir: str) -> list:
                 continue
             filepath = Path(root) / filename
             if filepath.suffix.lower() in CONVO_EXTENSIONS:
+                # Skip symlinks and oversized files
+                if filepath.is_symlink():
+                    continue
+                try:
+                    if filepath.stat().st_size > MAX_FILE_SIZE:
+                        continue
+                except OSError:
+                    continue
                 files.append(filepath)
     return files
 
@@ -356,7 +334,7 @@ def mine_convos(
             chunk_room = chunk.get("memory_type", room) if extract_mode == "general" else room
             if extract_mode == "general":
                 room_counts[chunk_room] += 1
-            drawer_id = f"drawer_{wing}_{chunk_room}_{hashlib.md5((source_file + str(chunk['chunk_index'])).encode(), usedforsecurity=False).hexdigest()[:16]}"
+            drawer_id = f"drawer_{wing}_{chunk_room}_{hashlib.sha256((source_file + str(chunk['chunk_index'])).encode()).hexdigest()[:24]}"
             try:
                 collection.add(
                     documents=[chunk["content"]],