chore(utils): mask credentials embedded in connection URLs

Add mask_url_credentials() to redact the password in a connection-URL's
userinfo (e.g. redis://:pw@host -> redis://:****@host) while preserving
scheme, host, and port. Field-name masking misses passwords smuggled into
URL values; this closes that gap for reuse across the proxy read endpoints.
This commit is contained in:
user 2026-05-30 06:59:49 +00:00
parent 70d2748d80
commit 8aa941b62a
No known key found for this signature in database
2 changed files with 70 additions and 0 deletions

View file

@ -1,5 +1,7 @@
import re
from collections.abc import Mapping
from typing import Any, Dict, List, Optional, Set
from urllib.parse import urlsplit, urlunsplit
from litellm.constants import DEFAULT_MAX_RECURSE_DEPTH_SENSITIVE_DATA_MASKER
@ -177,6 +179,38 @@ def mask_sensitive_keys(
return masked
# A connection URL whose authority contains userinfo, e.g. ``redis://u:p@host``.
# Requires the ``@`` to appear before any path/query/fragment so plain URLs
# without credentials (``https://host/path?x=@y``) are left untouched.
_URL_WITH_USERINFO = re.compile(r"^[a-zA-Z][a-zA-Z0-9+.\-]*://[^/?#\s]*@")
def mask_url_credentials(value: Any) -> Any:
"""Redact a password embedded in a connection URL's userinfo while keeping
the scheme, host, and port intact so the field stays diagnostically useful.
``redis://user:pass@host:6379`` -> ``redis://user:****@host:6379``
``rediss://:pass@host`` -> ``rediss://:****@host``
Strings without credentialed userinfo, and non-strings, are returned
unchanged.
"""
if not isinstance(value, str) or not _URL_WITH_USERINFO.match(value):
return value
try:
parts = urlsplit(value)
except ValueError:
return value
if parts.password is None:
return value
netloc = (
f"{parts.username or ''}:{_default_masker.mask_char * 4}@{parts.hostname or ''}"
)
if parts.port is not None:
netloc += f":{parts.port}"
return urlunsplit((parts.scheme, netloc, parts.path, parts.query, parts.fragment))
# Usage example:
"""
masker = SensitiveDataMasker()

View file

@ -154,3 +154,39 @@ def test_cost_per_token_fields_not_masked():
# Actual secrets must still be masked
assert "*" in masked["api_key"]
assert "*" in masked["access_token"]
from litellm.litellm_core_utils.sensitive_data_masker import mask_url_credentials
def test_mask_url_credentials_redacts_password_only():
"""A password embedded in a connection URL is redacted while scheme/host/
port stay intact, so the value is still diagnostically useful."""
assert (
mask_url_credentials("redis://:supersecretpw@redis.internal:6379")
== "redis://:****@redis.internal:6379"
)
assert (
mask_url_credentials("rediss://user:supersecretpw@host.example:6380/0")
== "rediss://user:****@host.example:6380/0"
)
# The plaintext password must never survive.
assert "supersecretpw" not in mask_url_credentials(
"redis://user:supersecretpw@host:6379"
)
def test_mask_url_credentials_leaves_non_credentialed_values_untouched():
"""URLs without a password, non-URL strings, and the '@' appearing only in
a path/query are all returned verbatim."""
assert (
mask_url_credentials("redis://redis.internal:6379")
== "redis://redis.internal:6379"
)
assert mask_url_credentials("redis://user@host:6379") == "redis://user@host:6379"
assert mask_url_credentials("https://api.example.com/v1?to=a@b.com") == (
"https://api.example.com/v1?to=a@b.com"
)
assert mask_url_credentials("plain-secret-value") == "plain-secret-value"
assert mask_url_credentials(None) is None
assert mask_url_credentials(1234) == 1234