mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-08 08:08:37 +00:00
ab78459097
Response transformation crashed with TypeError when counting tokens for multimodal (dict-shaped) embedding inputs, since encoding.encode() was called directly on the dict instead of its text parts. Also removed a dead, equally-buggy duplicate of the same loop in _transform_response. Separately, the embeddings guardrail handler silently skipped multimodal dict inputs instead of scanning them, letting prohibited text/image content bypass the configured guardrail. It now extracts and scans the text parts while leaving image parts untouched.
424 lines
15 KiB
Python
424 lines
15 KiB
Python
import ast
|
|
import os
|
|
import re
|
|
from typing import List, Dict, Any
|
|
|
|
|
|
class SensitiveLogDetector(ast.NodeVisitor):
|
|
"""
|
|
Detects logger.info() statements that might log sensitive request/response data.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.violations = []
|
|
self.current_file = None
|
|
|
|
def set_file(self, file_path: str):
|
|
"""Set the current file being analyzed"""
|
|
self.current_file = file_path
|
|
|
|
def visit_Call(self, node):
|
|
"""Visit function calls to detect logger.info() with sensitive data"""
|
|
if self._is_logger_info_call(node):
|
|
# Check all arguments to the logger.info() call
|
|
for arg in node.args:
|
|
if self._contains_sensitive_data(arg):
|
|
violation = {
|
|
"file": self.current_file,
|
|
"line": node.lineno,
|
|
"call": self._get_call_string(node),
|
|
"reason": self._get_violation_reason(arg),
|
|
"arg": self._get_arg_string(arg),
|
|
}
|
|
self.violations.append(violation)
|
|
|
|
self.generic_visit(node)
|
|
|
|
def _is_logger_info_call(self, node) -> bool:
|
|
"""Check if this is a logger.info() call"""
|
|
if not isinstance(node.func, ast.Attribute):
|
|
return False
|
|
|
|
# Check for various logger patterns:
|
|
# logger.info(), verbose_logger.info(), verbose_proxy_logger.info(), etc.
|
|
if node.func.attr == "info":
|
|
if isinstance(node.func.value, ast.Name):
|
|
logger_name = node.func.value.id
|
|
return any(
|
|
pattern in logger_name.lower() for pattern in ["logger", "log"]
|
|
)
|
|
|
|
return False
|
|
|
|
def _contains_sensitive_data(self, arg) -> bool:
|
|
"""Check if the argument might contain sensitive data"""
|
|
# Convert argument to string for analysis
|
|
arg_str = self._get_arg_string(arg).lower()
|
|
|
|
# Skip obvious non-sensitive patterns
|
|
non_sensitive_patterns = [
|
|
r'^["\'][\w\s\-_:.,!?]*["\']$', # Simple static strings
|
|
r'^["\'][^{%]*["\']$', # Strings without format placeholders
|
|
]
|
|
|
|
# Skip common safe phrases that contain sensitive keywords
|
|
safe_phrases = [
|
|
r"request\s+(completed|finished|started|processing)",
|
|
r"response\s+(sent|received|processed)",
|
|
r"data\s+(inserted|updated|deleted|saved)\s+into",
|
|
r"(successfully|failed)\s+(request|response)",
|
|
r"(starting|ending|completed)\s+(request|response)",
|
|
r"no\s+(usage\s+)?data\s+found",
|
|
r"found\s+\d+.*records",
|
|
r"exported\s+\d+.*records",
|
|
]
|
|
|
|
for pattern in non_sensitive_patterns:
|
|
if re.search(pattern, arg_str):
|
|
# Check if it's a safe phrase first
|
|
for safe_pattern in safe_phrases:
|
|
if re.search(safe_pattern, arg_str, re.IGNORECASE):
|
|
return False
|
|
|
|
# Then check if the static string mentions sensitive keywords
|
|
if not any(
|
|
keyword in arg_str
|
|
for keyword in [
|
|
"request",
|
|
"response",
|
|
"data",
|
|
"body",
|
|
"payload",
|
|
"token",
|
|
"auth",
|
|
"credential",
|
|
]
|
|
):
|
|
return False
|
|
|
|
# Direct variable/attribute patterns that are likely sensitive
|
|
sensitive_patterns = [
|
|
r"\brequest\b(?!\s*(id|status|method))", # request but not request_id, request_status, request_method
|
|
r"\bresponse\b(?!\s*(status|code|time))", # response but not response_status, response_code
|
|
r"\bdata\b(?=[\.\[\s]|$)", # data followed by . [ space or end
|
|
r"\bbody\b(?=[\.\[\s]|$)",
|
|
r"\bpayload\b(?=[\.\[\s]|$)",
|
|
r"\bmessages?\b(?=[\.\[\s]|$)",
|
|
r"\bcontent\b(?=[\.\[\s]|$)",
|
|
r"\binput\b(?=[\.\[\s]|$)",
|
|
r"\boutput\b(?=[\.\[\s]|$)",
|
|
r"\bargs\b(?=[\.\[\s]|$)",
|
|
r"\bkwargs\b(?=[\.\[\s]|$)",
|
|
r"\bparams\b(?=[\.\[\s]|$)",
|
|
r"\bheaders\b(?=[\.\[\s]|$)",
|
|
r"\bapi_key\b",
|
|
r"\btoken\b(?!\s*(name|id))", # token but not token_name, token_id
|
|
r"\bauth\b(?=[\.\[\s]|$)",
|
|
r"\bcredentials?\b",
|
|
]
|
|
|
|
# Check for direct variable references with context
|
|
for pattern in sensitive_patterns:
|
|
if re.search(pattern, arg_str):
|
|
return True
|
|
|
|
# Check for format strings that might interpolate sensitive data
|
|
if self._is_format_string_with_sensitive_data(arg):
|
|
return True
|
|
|
|
# Check for JSON dumps or string formatting of objects
|
|
if self._is_object_serialization(arg):
|
|
return True
|
|
|
|
return False
|
|
|
|
def _is_format_string_with_sensitive_data(self, arg) -> bool:
|
|
"""Check if this is a format string that might contain sensitive data"""
|
|
# Check for f-strings
|
|
if isinstance(arg, ast.JoinedStr):
|
|
for value in arg.values:
|
|
if isinstance(value, ast.FormattedValue):
|
|
value_str = self._get_arg_string(value.value).lower()
|
|
# Check for any sensitive data patterns in f-string interpolations
|
|
sensitive_f_string_patterns = [
|
|
"request",
|
|
"response",
|
|
"data",
|
|
"body",
|
|
"content",
|
|
"messages",
|
|
"token",
|
|
"jwt",
|
|
"auth",
|
|
"api_key",
|
|
"apikey",
|
|
"credential",
|
|
"secret",
|
|
"password",
|
|
"passwd",
|
|
]
|
|
if any(
|
|
pattern in value_str for pattern in sensitive_f_string_patterns
|
|
):
|
|
return True
|
|
|
|
# Check for .format() calls
|
|
if isinstance(arg, ast.Call) and isinstance(arg.func, ast.Attribute):
|
|
if arg.func.attr == "format":
|
|
# Check the base string for suspicious patterns
|
|
base_str = self._get_arg_string(arg.func.value).lower()
|
|
if "{}" in base_str or "{" in base_str:
|
|
# Check format arguments for sensitive data
|
|
sensitive_format_patterns = [
|
|
"request",
|
|
"response",
|
|
"data",
|
|
"body",
|
|
"content",
|
|
"token",
|
|
"jwt",
|
|
"auth",
|
|
"api_key",
|
|
"apikey",
|
|
"credential",
|
|
"secret",
|
|
"password",
|
|
"passwd",
|
|
]
|
|
for format_arg in arg.args:
|
|
format_str = self._get_arg_string(format_arg).lower()
|
|
if any(
|
|
pattern in format_str
|
|
for pattern in sensitive_format_patterns
|
|
):
|
|
return True
|
|
|
|
return False
|
|
|
|
def _is_object_serialization(self, arg) -> bool:
|
|
"""Check if this is serializing an object that might contain sensitive data"""
|
|
arg_str = self._get_arg_string(arg)
|
|
|
|
# Check for json.dumps() calls
|
|
if isinstance(arg, ast.Call):
|
|
if (
|
|
isinstance(arg.func, ast.Attribute)
|
|
and arg.func.attr == "dumps"
|
|
and isinstance(arg.func.value, ast.Name)
|
|
and arg.func.value.id == "json"
|
|
):
|
|
return True
|
|
|
|
# Check for str() calls on potentially sensitive objects
|
|
if (
|
|
isinstance(arg.func, ast.Name)
|
|
and arg.func.id == "str"
|
|
and len(arg.args) > 0
|
|
):
|
|
obj_str = self._get_arg_string(arg.args[0]).lower()
|
|
if any(
|
|
pattern in obj_str
|
|
for pattern in ["request", "response", "data", "body"]
|
|
):
|
|
return True
|
|
|
|
return False
|
|
|
|
def _get_violation_reason(self, arg) -> str:
|
|
"""Get a human-readable reason for the violation"""
|
|
arg_str = self._get_arg_string(arg).lower()
|
|
|
|
if any(
|
|
pattern in arg_str
|
|
for pattern in [
|
|
"jwt",
|
|
"token",
|
|
"api_key",
|
|
"apikey",
|
|
"auth",
|
|
"credential",
|
|
"secret",
|
|
"password",
|
|
"passwd",
|
|
]
|
|
):
|
|
return "Potentially logging authentication/secret data (JWT, token, API key, etc.)"
|
|
elif "request" in arg_str:
|
|
return "Potentially logging request data"
|
|
elif "response" in arg_str:
|
|
return "Potentially logging response data"
|
|
elif any(
|
|
pattern in arg_str for pattern in ["data", "body", "payload", "content"]
|
|
):
|
|
return "Potentially logging sensitive data/body/content"
|
|
elif any(pattern in arg_str for pattern in ["messages", "input", "output"]):
|
|
return "Potentially logging message/input/output data"
|
|
else:
|
|
return "Potentially logging sensitive data"
|
|
|
|
def _get_call_string(self, node) -> str:
|
|
"""Get string representation of the function call"""
|
|
try:
|
|
if hasattr(ast, "unparse"):
|
|
return ast.unparse(node)
|
|
else:
|
|
# Fallback for older Python versions
|
|
return f"{self._get_arg_string(node.func)}(...)"
|
|
except:
|
|
return "logger.info(...)"
|
|
|
|
def _get_arg_string(self, arg) -> str:
|
|
"""Get string representation of an argument"""
|
|
try:
|
|
if hasattr(ast, "unparse"):
|
|
return ast.unparse(arg)
|
|
else:
|
|
# Fallback for older Python versions
|
|
if isinstance(arg, ast.Name):
|
|
return arg.id
|
|
elif isinstance(arg, ast.Attribute):
|
|
return f"{self._get_arg_string(arg.value)}.{arg.attr}"
|
|
elif isinstance(arg, ast.Str):
|
|
return repr(arg.s)
|
|
elif isinstance(arg, ast.Constant):
|
|
return repr(arg.value)
|
|
else:
|
|
return str(type(arg).__name__)
|
|
except:
|
|
return "unknown"
|
|
|
|
|
|
def check_sensitive_logging(base_dir: str) -> List[Dict[str, Any]]:
|
|
"""
|
|
Check for logger.info() statements that might log sensitive data.
|
|
|
|
Args:
|
|
base_dir: Base directory to scan (typically the litellm root)
|
|
|
|
Returns:
|
|
List of violations found
|
|
"""
|
|
detector = SensitiveLogDetector()
|
|
all_violations = []
|
|
|
|
# Directories to scan - only main litellm codebase
|
|
scan_dirs = ["litellm", "enterprise"] # Include enterprise directory if it exists
|
|
|
|
# Directories to exclude (third-party code, venvs, etc.)
|
|
exclude_dirs = {
|
|
"venv",
|
|
"venv313",
|
|
".venv",
|
|
"env",
|
|
".env",
|
|
"node_modules",
|
|
"__pycache__",
|
|
".git",
|
|
"build",
|
|
"dist",
|
|
".tox",
|
|
"clean_env",
|
|
"litellm_env",
|
|
"myenv",
|
|
"py313_env",
|
|
"venv_sip_bypass",
|
|
"mypyc_env",
|
|
}
|
|
|
|
for scan_dir in scan_dirs:
|
|
dir_path = os.path.join(base_dir, scan_dir)
|
|
if not os.path.exists(dir_path):
|
|
print(f"Warning: Directory {dir_path} does not exist, skipping.")
|
|
continue
|
|
|
|
print(f"Scanning directory: {dir_path}")
|
|
|
|
for root, dirs, files in os.walk(dir_path):
|
|
# Skip excluded directories
|
|
dirs[:] = [d for d in dirs if d not in exclude_dirs]
|
|
|
|
# Skip if we're in a virtual environment or third-party directory
|
|
relative_root = os.path.relpath(root, base_dir)
|
|
if any(
|
|
excluded in relative_root.split(os.sep) for excluded in exclude_dirs
|
|
):
|
|
continue
|
|
|
|
for file in files:
|
|
if file.endswith(".py"):
|
|
file_path = os.path.join(root, file)
|
|
relative_path = os.path.relpath(file_path, base_dir)
|
|
|
|
# Skip files that are clearly third-party or generated
|
|
if any(excluded in relative_path for excluded in exclude_dirs):
|
|
continue
|
|
|
|
try:
|
|
with open(file_path, "r", encoding="utf-8") as f:
|
|
content = f.read()
|
|
tree = ast.parse(content)
|
|
|
|
detector.set_file(relative_path)
|
|
detector.visit(tree)
|
|
|
|
except SyntaxError as e:
|
|
print(f"Warning: Syntax error in file {relative_path}: {e}")
|
|
continue
|
|
except UnicodeDecodeError as e:
|
|
print(
|
|
f"Warning: Unicode decode error in file {relative_path}: {e}"
|
|
)
|
|
continue
|
|
except Exception as e:
|
|
print(f"Warning: Error processing file {relative_path}: {e}")
|
|
continue
|
|
|
|
return detector.violations
|
|
|
|
|
|
def main():
|
|
"""Main function to run the sensitive logging check"""
|
|
# Get the base directory (assume we're running from tests/code_coverage_tests/)
|
|
###################
|
|
# Running locally
|
|
###################
|
|
# current_dir = os.path.dirname(os.path.abspath(__file__))
|
|
# base_dir = os.path.join(current_dir, "..", "..")
|
|
# base_dir = os.path.abspath(base_dir)
|
|
|
|
###################
|
|
# Running in CI/CD
|
|
###################
|
|
base_dir = "./litellm" # Adjust this path as needed
|
|
|
|
print(f"Checking for sensitive logging in: {base_dir}")
|
|
|
|
violations = check_sensitive_logging(base_dir)
|
|
|
|
if violations:
|
|
print(f"\n❌ Found {len(violations)} potential violations:")
|
|
print("=" * 80)
|
|
|
|
for i, violation in enumerate(violations, 1):
|
|
print(f"\n{i}. {violation['file']}:{violation['line']}")
|
|
print(f" Reason: {violation['reason']}")
|
|
print(f" Call: {violation['call']}")
|
|
print(f" Argument: {violation['arg']}")
|
|
|
|
print("\n" + "=" * 80)
|
|
print("⚠️ SECURITY WARNING:")
|
|
print("These logger.info() statements may log sensitive request/response data.")
|
|
print("Consider changing them to logger.debug() or removing sensitive data.")
|
|
print("This is critical for PII compliance and security.")
|
|
print(
|
|
"Please contact @ishaan-jaff for more details about this check. DO NOT VIOLATE THIS CHECK."
|
|
)
|
|
|
|
return 1 # Exit with error code
|
|
else:
|
|
print("\n✅ No sensitive logging violations found!")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
exit(main())
|