mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-03 07:23:33 +00:00
50 lines
2 KiB
Python
50 lines
2 KiB
Python
import functools
|
|
import os
|
|
import re
|
|
from collections.abc import Callable
|
|
from typing import Final
|
|
from urllib.parse import unquote, urlsplit
|
|
|
|
_REDACTED: Final = "REDACTED"
|
|
_PASSWORD_QUERY_KEYS: Final = frozenset(("password", "sslpassword"))
|
|
|
|
|
|
@functools.cache
|
|
def _secret_shape_redactor() -> Callable[[str], str]:
|
|
try:
|
|
from litellm._logging import redact_secrets
|
|
except ImportError:
|
|
return lambda text: text
|
|
return redact_secrets
|
|
|
|
|
|
def _url_passwords(url: str) -> frozenset[str]:
|
|
try:
|
|
parts: Final = urlsplit(url)
|
|
except ValueError:
|
|
return frozenset()
|
|
query_pairs: Final = tuple(pair.partition("=") for pair in parts.query.split("&"))
|
|
raw_query_passwords: Final = tuple(
|
|
value for key, separator, value in query_pairs if separator and key.lower() in _PASSWORD_QUERY_KEYS
|
|
)
|
|
raw_passwords: Final = ((parts.password,) if parts.password else ()) + raw_query_passwords
|
|
return frozenset(password for password in raw_passwords + tuple(map(unquote, raw_passwords)) if password)
|
|
|
|
|
|
def _configured_database_passwords() -> frozenset[str]:
|
|
database_url: Final = os.getenv("DATABASE_URL")
|
|
direct_url: Final = os.getenv("DIRECT_URL")
|
|
database_passwords: Final = _url_passwords(database_url) if database_url else frozenset()
|
|
direct_passwords: Final = _url_passwords(direct_url) if direct_url else frozenset()
|
|
return database_passwords | direct_passwords
|
|
|
|
|
|
def _redact_credentials(text: str) -> str:
|
|
"""Mask configured database passwords before passing the text to LiteLLM redaction."""
|
|
passwords: Final = sorted(_configured_database_passwords(), key=len, reverse=True)
|
|
alternation: Final = "|".join(re.escape(password) for password in passwords)
|
|
password_pattern: Final = (
|
|
re.compile(rf"(?<![A-Za-z0-9_])(?:{alternation})(?![A-Za-z0-9_])", re.IGNORECASE) if passwords else None
|
|
)
|
|
result: Final = password_pattern.sub(_REDACTED, text) if password_pattern is not None else text
|
|
return _secret_shape_redactor()(result)
|