🛡️ Sentinel: [MEDIUM] Fix DSN redaction credential leaks

.jules/sentinel.md

@@ -2,3 +2,7 @@
 **Vulnerability:** User-provided string fields (like project and connection names) lacked strict validation against control characters, only relying on length constraints.
 **Learning:** This could potentially lead to Log Injection (CRLF injection), Null Byte Injection, or terminal escape injection if these strings are subsequently logged or rendered directly.
 **Prevention:** Use explicit regex validation `pattern=r'^[^\x00-\x1F\x7F]+$'` on Pydantic string fields to strictly reject control characters.
+## 2024-05-18 - Fix over-redaction and edge-case leaks of DSN credentials in error messages
+**Vulnerability:** The DSN redaction code didn't fully decode passwords returned by `urlsplit` (missing `unquote_plus`), which could lead to partial leaks in driver error messages for passwords that contained URL-encoded symbols like `%20` or `+`. Also, the logic using word boundaries (`\b` implicitly via negative lookahead/lookbehind for alphanumerics) to prevent short passwords from over-redacting didn't properly handle the cases where the secret started or ended with a non-alphanumeric character (e.g. `=pass=`).
+**Learning:** URL decoding requires full `unquote_plus` rather than just `unquote`, and boundary logic on secrets requires checking `secret[0].isalnum()` and `secret[-1].isalnum()` when constructing the regex since the surrounding chars must only be verified as non-alphanumeric if the edge of the secret itself is alphanumeric.
+**Prevention:** Thoroughly check edge cases for both decoding and regex boundaries when doing find-and-replace for passwords in arbitrary log strings.

backend/app/dsn_redaction.py

@@ -57,15 +57,21 @@ def _password_candidates_from_dsn(dsn: str) -> set[str]:
         netloc, query = _split_dsn_best_effort(dsn)
 
     if password:
+        decoded_password = unquote_plus(password)
         candidates.add(password)
-        candidates.add(quote(password, safe=""))
+        candidates.add(decoded_password)
+        candidates.add(quote(decoded_password, safe=""))
+        candidates.add(quote_plus(decoded_password, safe=""))
 
     if "@" in netloc:
         userinfo = netloc.rsplit("@", 1)[0]
         if ":" in userinfo:
             raw_password = userinfo.split(":", 1)[1]
+            decoded_raw_password = unquote_plus(raw_password)
             candidates.add(raw_password)
-            candidates.add(unquote(raw_password))
+            candidates.add(decoded_raw_password)
+            candidates.add(quote(decoded_raw_password, safe=""))
+            candidates.add(quote_plus(decoded_raw_password, safe=""))
 
     for part in query.split("&"):
         key, sep, raw_value = part.partition("=")
@@ -86,7 +92,9 @@ def _redact_secret_occurrences(message: str, secret: str) -> str:
     if len(secret) > 4:
         return message.replace(secret, "***")
 
-    pattern = re.compile(rf"(?<![A-Za-z0-9]){re.escape(secret)}(?![A-Za-z0-9])")
+    prefix = r"(?<![A-Za-z0-9])" if secret and secret[0].isalnum() else ""
+    suffix = r"(?![A-Za-z0-9])" if secret and secret[-1].isalnum() else ""
+    pattern = re.compile(rf"{prefix}{re.escape(secret)}{suffix}")
     return pattern.sub("***", message)
 
 

backend/tests/test_dsn_redaction.py

@@ -43,3 +43,22 @@ def test_malformed_dsn_still_redacts_embedded_secrets() -> None:
     assert "s3cr3t" not in redacted
     assert "q/secret" not in redacted
     assert "password=***" in redacted
+
+
+def test_short_password_surrounded_by_non_alphanumerics() -> None:
+    dsn = "postgresql://user:=ab=@db.example.com/app"
+    error = "driver failed for =ab= with password==ab= while using postgresql://user:=ab=@db.example.com/app"
+
+    redacted = redact_dsn_error_message(error, dsn)
+
+    assert "=ab=" not in redacted
+    assert "postgresql://user:***@db.example.com/app" in redacted
+
+
+def test_short_password_is_alphanumeric_but_bounded_by_non_alphanumeric() -> None:
+    dsn = "postgresql://user:pass@db.example.com/app"
+    error = "driver failed for =pass= with password==pass="
+
+    redacted = redact_dsn_error_message(error, dsn)
+
+    assert "=pass=" not in redacted or "pass" not in redacted