🛡️ Sentinel: [CRITICAL] Fix SQL injection in metadata_generator

🚨 Severity: CRITICAL
💡 Vulnerability: SQL injection vulnerability in `metadata_generator.py` where dictionary keys from untrusted input were used directly to construct an `INSERT OR REPLACE` query.
🎯 Impact: Attackers could inject arbitrary SQL commands by passing malicious dictionary keys.
🔧 Fix: Fetched valid column names using `PRAGMA table_info(file_metadata)` and filtered the metadata keys against this allowlist before building the query.
✅ Verification: Verified syntax, ran test suite, and manually verified db insert.

Co-authored-by: thebearwithabite <216692431+thebearwithabite@users.noreply.github.com>

Author: google-labs-jules[bot]

.jules/sentinel.md

@@ -19,3 +19,7 @@
 **Prevention:**
 1. Always convert `pathlib.Path` objects to absolute strings using `str(path.absolute())` before passing them as arguments to `subprocess.run`.
 2. Absolute paths always begin with a directory separator (`/` on Unix) or a drive letter (`C:\` on Windows), guaranteeing the command-line tool parses them as file paths rather than flags or options.
+## 2024-05-30 - Fix SQL Injection in dynamic column updates
+**Vulnerability:** SQL injection vulnerability in `metadata_generator.py` where dictionary keys from untrusted input were used directly to construct an `INSERT OR REPLACE` query.
+**Learning:** Parameterization (`?`) only protects values, not column names. When dynamically building queries from dictionaries, the keys must be validated against a strict schema allowlist.
+**Prevention:** Always use `PRAGMA table_info(table_name)` or a hardcoded list of allowed column names to validate keys before building dynamic queries.

metadata_generator.py

@@ -467,9 +467,18 @@ def save_file_metadata(self, metadata: Dict[str, Any]) -> bool:
 
         try:
             with sqlite3.connect(self.db_path) as conn:
-                # Convert to database format
-                columns = list(metadata.keys())
-                values = list(metadata.values())
+                # Fetch valid column names to prevent SQL injection
+                cursor = conn.execute("PRAGMA table_info(file_metadata)")
+                valid_columns = {row[1] for row in cursor.fetchall()}
+
+                # Filter metadata to only include valid columns
+                safe_metadata = {k: v for k, v in metadata.items() if k in valid_columns}
+
+                if not safe_metadata:
+                    return False
+
+                columns = list(safe_metadata.keys())
+                values = list(safe_metadata.values())
                 placeholders = ', '.join(['?' for _ in values])
                 column_names = ', '.join(columns)
 

test_mock.py

@@ -0,0 +1,7 @@
+import sqlite3
+
+def run_test():
+    with sqlite3.connect('/tmp/metadata.db') as conn:
+        cursor = conn.execute("SELECT name FROM sqlite_master WHERE type='table';")
+        print(cursor.fetchall())
+run_test()

test_wrap.py

@@ -0,0 +1,36 @@
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock
+
+# Mock dependencies
+sys.modules['pandas'] = MagicMock()
+sys.modules['google'] = MagicMock()
+sys.modules['google.generativeai'] = MagicMock()
+
+# Mock gdrive_integration properly
+gdrive_mock = MagicMock()
+gdrive_mock.ensure_safe_local_path = lambda x: Path(x)
+gdrive_mock.get_ai_organizer_root = lambda: Path('/tmp')
+gdrive_mock.get_metadata_root = lambda: Path('/tmp')
+sys.modules['gdrive_integration'] = gdrive_mock
+
+# Mock path_manager
+pm_mock = MagicMock()
+pm_mock.get_path = lambda *args, **kwargs: Path('/tmp/metadata.db')
+sys.modules['path_manager'] = pm_mock
+
+import test_metadata
+import metadata_generator
+import sqlite3
+
+# Initialize generator to ensure DB is created
+generator = metadata_generator.MetadataGenerator()
+generator._init_tracking_db()
+
+# Provide metadata manually since the mocked dependencies might not generate it
+test_data = {"file_path": "/tmp/test.txt", "file_name": "test.txt"}
+generator.save_file_metadata(test_data)
+
+# Check db
+with sqlite3.connect(generator.db_path) as conn:
+    print(conn.execute("SELECT * FROM file_metadata;").fetchall())