🛡️ Sentinel: [CRITICAL] Fix SQL injection risk in dynamic kwargs

.jules/sentinel.md

@@ -0,0 +1,5 @@
+
+## 2024-05-18 - Prevent SQL Injection via **kwargs Unpacking Bypass
+**Vulnerability:** In `streamrip/db.py`, the `contains` and `remove` methods accepted `**kwargs` that were unpacked directly into dynamic SQL query strings. While `contains` validated the dictionary keys against allowed database schema columns, it used `assert` for this validation. If Python were run with the `-O` (optimize) flag, assertions would be disabled, allowing arbitrary keyword arguments to inject SQL directly into the column names of `SELECT` or `DELETE` statements.
+**Learning:** Dictionary unpacking (`**kwargs`) bypasses standard Python identifier restrictions (allowing characters normally forbidden in variable names). Validating these keys using `assert` creates a vulnerability that relies on interpreter configuration, as `assert` statements should never be used for security-critical validation or control flow.
+**Prevention:** Explicitly validate all dynamic keys used in SQL query construction using `if not in ... raise ValueError` to ensure validation is never optimized away, regardless of execution environment.

streamrip/db.py

@@ -113,9 +113,12 @@ def contains(self, **items) -> bool:
         :rtype: bool
         """
         allowed_keys = set(self.structure.keys())
-        assert all(
-            key in allowed_keys for key in items.keys()
-        ), f"Invalid key. Valid keys: {allowed_keys}"
+
+        # 🛡️ Sentinel: Prevent SQL injection via arbitrary column names when unpacking kwargs.
+        # Use ValueError instead of assert, which can be optimized away (e.g., python -O).
+        for key in items.keys():
+            if key not in allowed_keys:
+                raise ValueError(f"Invalid key: {key}. Valid keys: {allowed_keys}")
 
         items = {k: str(v) for k, v in items.items()}
 
@@ -155,6 +158,13 @@ def remove(self, **items):
 
         :param items:
         """
+        allowed_keys = set(self.structure.keys())
+
+        # 🛡️ Sentinel: Prevent SQL injection via arbitrary column names when unpacking kwargs.
+        for key in items.keys():
+            if key not in allowed_keys:
+                raise ValueError(f"Invalid key: {key}. Valid keys: {allowed_keys}")
+
         conditions = " AND ".join(f"{key}=?" for key in items.keys())
         command = f"DELETE FROM {self.name} WHERE {conditions}"