Security hardening: 7 fixes from security evaluation + integration tests (#24)

Author: iamontheinet

plugins/cortex-code/README.md

@@ -92,6 +92,29 @@ Edit the config to change approval mode, allowed envelopes, audit settings, and
 
 Skill discovery runs automatically on session start. To force a re-discovery, start a new Claude Code session.
 
+## Testing
+
+Tests live in `tests/run-tests.sh` at the repo root. Two tiers:
+
+```bash
+# Structural + unit tests (no network, runs in CI)
+bash tests/run-tests.sh
+
+# Include integration tests (requires cortex CLI + Snowflake connection)
+bash tests/run-tests.sh --integration
+```
+
+**Structural tests** (always run): file existence checks, config validation, Python syntax, and unit tests for `envelope_policy.py`, `prompt_filter.py`, and plugin hooks.
+
+**Integration tests** (`--integration` flag): spawn real Cortex CLI sessions against a live Snowflake connection. Located at `scripts/router/test_integration.py`. Verifies:
+
+- Credential path blocking (prompts referencing `.ssh/`, `.env`, etc. are rejected pre-flight)
+- End-to-end query flow (RO envelope, permission protocol, result event)
+- Envelope enforcement (RO blocks DDL — via hard gate denial or LLM self-policing)
+- Process cleanup (no orphaned `cortex` processes after execution)
+
+Set `CORTEX_TEST_CONNECTION` env var to test against a specific Snowflake connection (defaults to your CLI default).
+
 ## License
 
 Copyright (c) Snowflake Inc. All rights reserved.

plugins/cortex-code/scripts/router/config.yaml.example

@@ -38,12 +38,13 @@ security:
     - "**/.npmrc"
     - "**/.pypirc"
 
-  # Which envelopes are allowed (RO, RW, RESEARCH, DEPLOY)
+  # Which envelopes are allowed (RO, RW, RESEARCH)
+  # Note: DEPLOY grants full access — only enable if you understand the blast radius.
   allowed_envelopes:
     - "RO"
     - "RW"
     - "RESEARCH"
-    - "DEPLOY"
+    # - "DEPLOY"  # Uncomment to enable full-access mode
 
 # --- Deployment Profiles (uncomment one) ---
 

plugins/cortex-code/scripts/router/discover_cortex.py

@@ -21,11 +21,11 @@
 
 
 def run_command(cmd):
-    """Run shell command and return output."""
+    """Run command and return output."""
     try:
         result = subprocess.run(
-            cmd,
-            shell=True,
+            cmd.split(),
+            shell=False,
             capture_output=True,
             text=True,
             timeout=10

plugins/cortex-code/scripts/router/execute_cortex.py

@@ -323,6 +323,15 @@ def execute_cortex_streaming(prompt: str, connection: Optional[str] = None,
         return results
 
     except Exception as e:
+        # Prevent orphaned cortex processes on unexpected exceptions
+        try:
+            process.terminate()
+            process.wait(timeout=2)
+        except Exception:
+            try:
+                process.kill()
+            except Exception:
+                pass
         return {
             "session_id": None,
             "events": [],

plugins/cortex-code/scripts/router/predict_tools.py

@@ -39,15 +39,18 @@
 
 
 def load_capabilities():
-    """Load cached Cortex capabilities."""
-    cache_path = Path("/tmp/cortex-capabilities.json")
-
-    if not cache_path.exists():
+    """Load cached Cortex capabilities via CacheManager."""
+    try:
+        sys.path.insert(0, str(Path(__file__).parent.parent))
+        from security.config_manager import ConfigManager
+        from security.cache_manager import CacheManager
+        config_manager = ConfigManager()
+        cache_dir = Path(config_manager.get("security.cache_dir")).expanduser()
+        cache_manager = CacheManager(cache_dir)
+        return cache_manager.read("cortex-capabilities") or {}
+    except Exception:
         return {}
 
-    with open(cache_path, 'r') as f:
-        return json.load(f)
-
 
 def predict_tools(prompt, envelope=None):
     """

plugins/cortex-code/scripts/router/read_cortex_sessions.py

@@ -43,6 +43,12 @@ def parse_session_file(session_path, sanitize=True):
         Dictionary with session data, or None on error
     """
     try:
+        # Guard against pathologically large session files (10MB limit)
+        file_size = session_path.stat().st_size
+        if file_size > 10 * 1024 * 1024:
+            print(f"Skipping oversized session file ({file_size} bytes): {session_path}", file=sys.stderr)
+            return None
+
         with open(session_path, 'r') as f:
             lines = f.readlines()
 

plugins/cortex-code/scripts/router/session_state.py

@@ -57,6 +57,7 @@ def save_active_session(session_id: str) -> None:
         with os.fdopen(fd, "w") as f:
             json.dump(payload, f)
         os.replace(tmp_name, path)
+        os.chmod(path, 0o600)
     except Exception:
         try:
             os.unlink(tmp_name)