🛡️ Sentinel: [MEDIUM] Fix log injection vulnerabilities

.jules/sentinel.md

@@ -53,3 +53,10 @@
 **Learning:** `ip.is_global` (available since Python 3.4) is the correct property for validating public Internet addresses. However, it considers Multicast addresses as "global" (technically true), so explicit `ip.is_multicast` checks are still needed if blocking them is desired.
 **Prevention:**
 1. Always use `if not ip.is_global or ip.is_multicast:` for strict SSRF filtering, rather than manual blacklists of private ranges.
+
+## 2026-05-18 - [Log Injection via Unsanitized Input]
+**Vulnerability:** User-controlled inputs (folder names from JSON, error messages) were logged using f-strings without sanitization. This allowed Terminal Escape Sequence Injection, potentially corrupting terminal output or spoofing log entries.
+**Learning:** `repr()` is a powerful, built-in mechanism for sanitizing strings for logs because it escapes control characters (like `\x1b`) by default.
+**Prevention:**
+1. Identify all log call sites that include user input.
+2. Wrap untrusted inputs in a sanitization function (e.g., `sanitize_for_log`) that uses `repr()` or similar escaping mechanisms.

main.py

@@ -100,10 +100,12 @@ def format(self, record):
 
 
 def sanitize_for_log(text: Any) -> str:
-    """Sanitize text for logging, ensuring TOKEN is redacted."""
+    """Sanitize text for logging, ensuring TOKEN is redacted and control chars are escaped."""
     s = str(text)
     if TOKEN and TOKEN in s:
         s = s.replace(TOKEN, "[REDACTED]")
+    # repr() safely escapes control characters (e.g., \n -> \\n, \x1b -> \\x1b)
+    # This prevents log injection and terminal hijacking.
     safe = repr(s)
     if len(safe) >= 2 and safe[0] == safe[-1] and safe[0] in ("'", '"'):
         return safe[1:-1]
@@ -513,7 +515,7 @@ def create_folder(client: httpx.Client, profile_id: str, name: str, do: int, sta
 
             if attempt < MAX_RETRIES:
                 wait_time = FOLDER_CREATION_DELAY * (attempt + 1)
-                log.info(f"Folder '{name}' not found yet. Retrying in {wait_time}s...")
+                log.info(f"Folder '{sanitize_for_log(name)}' not found yet. Retrying in {wait_time}s...")
                 time.sleep(wait_time)
 
         log.error(f"Folder {sanitize_for_log(name)} was not found after creation and retries.")
@@ -776,7 +778,7 @@ def sync_profile(
                         if future.result():
                             success_count += 1
                     except Exception as e:
-                        log.error(f"Failed to process folder '{folder_name}': {e}")
+                        log.error(f"Failed to process folder '{sanitize_for_log(folder_name)}': {sanitize_for_log(e)}")
 
         log.info(f"Sync complete: {success_count}/{len(folder_data_list)} folders processed successfully")
         return success_count == len(folder_data_list)

tests/test_log_sanitization.py

@@ -0,0 +1,71 @@
+import unittest
+from unittest.mock import MagicMock, patch
+import main
+
+class TestLogSanitization(unittest.TestCase):
+    def test_sanitize_for_log_escapes_ansi(self):
+        """Test that sanitize_for_log escapes ANSI characters."""
+        # ANSI Red color
+        malicious_input = "\x1b[31mMalicious"
+        sanitized = main.sanitize_for_log(malicious_input)
+
+        # repr() escapes \x1b as \x1b (4 chars: \, x, 1, b)
+        # So the output string should contain literal backslash
+        self.assertIn("\\x1b", sanitized)
+        # It should NOT contain the actual escape character
+        self.assertNotIn("\x1b", sanitized)
+
+    @patch('main.log')
+    @patch('main.time.sleep')
+    @patch('main._api_post')
+    @patch('main._api_get')
+    def test_create_folder_logs_unsafe_name(self, mock_get, mock_post, mock_sleep, mock_log):
+        """
+        Verify that create_folder logs the raw name if not sanitized.
+        We expect this to FAIL (or show raw usage) before the fix.
+        """
+        # Setup
+        main.MAX_RETRIES = 1
+        main.FOLDER_CREATION_DELAY = 0
+
+        # Mock POST to succeed (returns None, assuming polling needed if direct ID missing)
+        mock_post.return_value.json.return_value = {"body": {"group": {"something": "else"}}}
+
+        # Mock GET to return empty groups (fail to find)
+        mock_get.return_value.json.return_value = {"body": {"groups": []}}
+
+        unsafe_name = "\x1b[31mUNSAFE"
+
+        # Call
+        client = MagicMock()
+        main.create_folder(client, "pid", unsafe_name, 0, 1)
+
+        # Check logs: ensure we do not log the raw unsafe name, but do log the sanitized name.
+        # For this test file, I want it to PASS when the code is FIXED.
+        # So I should assert that I DO NOT find raw unsafe_name, but I DO find sanitized name.
+
+        sanitized_name = main.sanitize_for_log(unsafe_name)
+
+        found_sanitized = False
+        found_raw = False
+
+        for call in mock_log.info.call_args_list:
+            args = call[0]
+            # Since it is an f-string in the source, we can't easily check format args.
+            # We have to check the string content.
+            # But wait, if the source is f"Folder '{name}'...", logging receives the formatted string.
+            log_msg = args[0]
+            if unsafe_name in log_msg:
+                found_raw = True
+            if sanitized_name in log_msg:
+                found_sanitized = True
+
+        if found_raw:
+            print("VULNERABILITY DETECTED: Raw unsafe name found in logs.")
+
+        # This assertion will FAIL before fix, and PASS after fix.
+        self.assertTrue(found_sanitized, "Should find sanitized name in logs")
+        self.assertFalse(found_raw, "Should not find raw name in logs")
+
+if __name__ == '__main__':
+    unittest.main()