Fix SQL injection via untrusted dictionary keys in metadata generator (#241)

Adds strict schema validation using PRAGMA table_info to filter
untrusted dictionary keys before dynamically constructing the SQL
INSERT statement.

Co-authored-by: google-labs-jules[bot] <161369871+google-labs-jules[bot]@users.noreply.github.com>
Co-authored-by: thebearwithabite <216692431+thebearwithabite@users.noreply.github.com>

Author: thebearwithabite

.jules/sentinel.md

@@ -19,3 +19,13 @@
 **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 - SQL Injection via Dictionary Keys in Dynamic Queries
+
+**Vulnerability:** The `save_file_metadata` function in `metadata_generator.py` accepted an untrusted dictionary of metadata and dynamically constructed an `INSERT` statement using the dictionary's keys as column names (`column_names = ', '.join(columns)`). This allowed SQL injection if an attacker supplied a malformed key (e.g., `invalid_column) VALUES (?); DROP TABLE files; --`).
+
+**Learning:** While parameterization (`?`) protects against SQL injection in values, it does not protect against injection in table or column names. When dynamically building queries where column names are derived from user input or external dictionaries, the keys must be strictly validated against an explicit schema allowlist.
+
+**Prevention:**
+1. Never use untrusted input directly as column names or table names in SQL queries.
+2. Fetch valid columns dynamically using `PRAGMA table_info(table_name)` in SQLite (or hardcode an allowlist).
+3. Pre-filter dictionaries to only include keys that match the explicitly validated schema allowlist before building dynamic queries.

metadata_generator.py

@@ -463,13 +463,24 @@ def _analyze_multimedia(self, file_path: Path) -> Dict[str, Any]:
         return multimedia_data
 
     def save_file_metadata(self, metadata: Dict[str, Any]) -> bool:
-        """Save file metadata to database"""
+        """Save file metadata to database safely, preventing SQL injection"""
 
         try:
             with sqlite3.connect(self.db_path) as conn:
+                # Validate 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:
+                    print("Warning: No valid metadata fields to save")
+                    return False
+
                 # Convert to database format
-                columns = list(metadata.keys())
-                values = list(metadata.values())
+                columns = list(safe_metadata.keys())
+                values = list(safe_metadata.values())
                 placeholders = ', '.join(['?' for _ in values])
                 column_names = ', '.join(columns)