🛡️ Sentinel: [HIGH] DSN 파싱 우회를 통한 자격 증명 유출 취약점 수정

.jules/sentinel.md

@@ -57,3 +57,7 @@
 **Vulnerability:** Database driver exceptions can echo DSN fragments, query parameters, or assignment-style secrets after connection failures, leaking plaintext passwords through snapshot error messages and queue logs.
 **Learning:** Redacting only the literal DSN is not enough. Error messages may contain decoded, percent-encoded, query-string, or `password=`/`api_key=` style forms of the same secret.
 **Prevention:** Sanitize snapshot job errors before persisting or re-raising them, and raise sanitized exceptions with `from None` so Python exception chaining does not reattach the original secret-bearing exception.
+## 2025-02-28 - [DSN Credential Leakage via Underscore Schemes]
+**Vulnerability:** Python's `urllib.parse.urlsplit` fails to properly parse DSN strings where the scheme contains an underscore (e.g., `my_custom_db://`). This results in passwords and netloc information not being extracted, which causes the application's DSN redaction logic (`redact_dsn_error_message`) to bypass redaction, leaking plaintext credentials into logs and error messages.
+**Learning:** URL parsing libraries conform strictly to RFC 3986 (where schemes cannot contain underscores). To perform security redaction on non-standard DSNs, we must sanitize or workaround the scheme before parsing.
+**Prevention:** Temporarily swap non-compliant schemes (containing `_`) with standard ones (like `http://`) before utilizing `urlsplit` for password extraction to ensure robust credential redaction.

backend/app/dsn_redaction.py

@@ -18,7 +18,15 @@
 
 def _password_candidates_from_dsn(dsn: str) -> set[str]:
     candidates: set[str] = set()
-    parsed = urlsplit(dsn)
+    parse_dsn = dsn
+    # Workaround for Python's urllib.parse not extracting password/netloc
+    # when the URL scheme contains an underscore.
+    if "://" in dsn:
+        scheme, rest = dsn.split("://", 1)
+        if "_" in scheme:
+            parse_dsn = f"http://{rest}"
+
+    parsed = urlsplit(parse_dsn)
 
     if parsed.password:
         candidates.add(parsed.password)

backend/tests/test_dsn_redaction.py

@@ -0,0 +1,34 @@
+from app.dsn_redaction import _password_candidates_from_dsn, redact_dsn_error_message
+
+def test_dsn_redaction_underscore_scheme():
+    dsn = "my_custom_db://user:mysecretpassword@host/db"
+    candidates = _password_candidates_from_dsn(dsn)
+    assert "mysecretpassword" in candidates
+
+def test_dsn_redaction_query_params():
+    dsn = "snowflake_test://user@host/db?password=mysecretpassword"
+    candidates = _password_candidates_from_dsn(dsn)
+    assert "mysecretpassword" in candidates
+
+def test_dsn_redaction_message():
+    dsn = "my_custom_db://user:mysecretpassword@host/db"
+    msg = "Failed to connect to host with password mysecretpassword."
+    redacted = redact_dsn_error_message(msg, dsn)
+    assert "mysecretpassword" not in redacted
+    assert "***" in redacted
+
+def test_dsn_redaction_standard_scheme():
+    dsn = "postgresql://user:standardpass@host/db"
+    candidates = _password_candidates_from_dsn(dsn)
+    assert "standardpass" in candidates
+
+    # Add a query param test to hit line 47: continue if not _SECRET_KEY_PATTERN
+    dsn_with_safe_query = "postgresql://user:standardpass@host/db?timeout=30"
+    candidates_safe_query = _password_candidates_from_dsn(dsn_with_safe_query)
+    assert "standardpass" in candidates_safe_query
+    assert "30" not in candidates_safe_query
+
+    msg = "Connection failed for user, password standardpass"
+    redacted = redact_dsn_error_message(msg, dsn)
+    assert "standardpass" not in redacted
+    assert "***" in redacted