scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-07 02:59:05 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 103

fix(proxy): stop logging license values during verification (#44956)

Browse source 
* fix(proxy): stop logging license values during verification

* fix(tests): address license logging review feedback

---------

Co-authored-by: moyai-devin-berriai[bot] <336287033+moyai-devin-berriai[bot]@users.noreply.github.com>
This commit is contained in:
moyai-devin-berriai[bot] 2026-10-06 17:05:07 -07:00 • committed by GitHub
parent 0b0fdedd1e
commit b9ab1eee1f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 141 additions and 24 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

32
litellm/proxy/auth/litellm_license.py
View file

@ -33,7 +33,7 @@ class LicenseCheck:
def __init__(self) -> None:
self.license_str = os.getenv("LITELLM_LICENSE", None)
verbose_proxy_logger.debug("License Str value - %s", self.license_str)
verbose_proxy_logger.debug("License configured: %s", self.license_str is not None)
self.http_handler = HTTPHandler(timeout=NON_LLM_CONNECTION_TIMEOUT)
self._premium_check_logged = False
self.public_key = None
@ -59,9 +59,8 @@ class LicenseCheck:
def _verify(self, license_str: str) -> bool:
verbose_proxy_logger.debug(
"litellm.proxy.auth.litellm_license.py::_verify - Checking license against %s/verify_license - %s",
"litellm.proxy.auth.litellm_license.py::_verify - Checking license against %s/verify_license",
self.base_url,
license_str,
)
url: Final = f"{self.base_url}/verify_license/{license_str}"
@ -86,14 +85,15 @@ class LicenseCheck:
assert isinstance(premium, bool)
verbose_proxy_logger.debug(
"litellm.proxy.auth.litellm_license.py::_verify - License=%s is premium=%s", license_str, premium
"litellm.proxy.auth.litellm_license.py::_verify - License is premium=%s", premium
)
return premium
except Exception as e:
verbose_proxy_logger.exception(
"litellm.proxy.auth.litellm_license.py::_verify - Unable to verify License=%s via api. - %s",
license_str,
e,
verbose_proxy_logger.error(
"litellm.proxy.auth.litellm_license.py::_verify - Unable to verify license via api. "
"error_type=%s status_code=%s",
type(e).__name__,
e.response.status_code if isinstance(e, httpx.HTTPStatusError) else None,
)
return False
@ -105,8 +105,8 @@ class LicenseCheck:
try:
if not self._premium_check_logged:
verbose_proxy_logger.debug(
"litellm.proxy.auth.litellm_license.py::is_premium() - ENTERING 'IS_PREMIUM' - LiteLLM License=%s",
self.license_str,
"litellm.proxy.auth.litellm_license.py::is_premium() - ENTERING 'IS_PREMIUM' - License configured: %s",
self.license_str is not None,
)
if self.license_str is None:
@ -114,8 +114,8 @@ class LicenseCheck:
if not self._premium_check_logged:
verbose_proxy_logger.debug(
"litellm.proxy.auth.litellm_license.py::is_premium() - Updated 'self.license_str' - %s",
self.license_str,
"litellm.proxy.auth.litellm_license.py::is_premium() - License configured after refresh: %s",
self.license_str is not None,
)
self._premium_check_logged = True
@ -202,9 +202,6 @@ class LicenseCheck:
# Decode and parse the data
license_data: Final = json.loads(message.decode())
# debug information provided in license data
verbose_proxy_logger.debug("License data: %s", license_data)
# Check expiration date
expiration_date: Final = datetime.strptime(license_data["expiration_date"], "%Y-%m-%d")
if expiration_date < datetime.now():
@ -218,7 +215,8 @@ class LicenseCheck:
except Exception as e:
self.airgapped_license_data = None
verbose_proxy_logger.debug(
"litellm.proxy.auth.litellm_license.py::verify_license_without_api_request - Unable to verify License locally. - %s",
e,
"litellm.proxy.auth.litellm_license.py::verify_license_without_api_request - "
"Unable to verify license locally. error_type=%s",
type(e).__name__,
)
return False

133
tests/unit/proxy/auth/test_litellm_license.py
View file

@ -1,13 +1,12 @@
import asyncio
import json
import logging
from typing import Final
from unittest.mock import AsyncMock, MagicMock, patch
from unittest.mock import MagicMock
import httpx
import pytest
from cryptography.hazmat.primitives.asymmetric.rsa import RSAPublicKey
from litellm.llms.custom_httpx.http_handler import HTTPHandler
from litellm.proxy.auth.litellm_license import LicenseCheck
@ -15,9 +14,9 @@ from litellm.proxy.auth.litellm_license import LicenseCheck
def test_read_public_key_loads_successfully():
"""Ensure public_key.pem is valid PEM with no leading whitespace."""
license_check = LicenseCheck()
assert (
license_check.public_key is not None
), "public_key.pem could not be loaded — check for leading whitespace or malformed PEM header"
assert license_check.public_key is not None, (
"public_key.pem could not be loaded — check for leading whitespace or malformed PEM header"
)
def test_is_over_limit():
@ -104,7 +103,9 @@ def test_expired_or_unreadable_license_grants_no_features() -> None:
assert license_check.auto_router_capability_limit() == 1
assert license_check.verify_license_without_api_request(public_key=public_key, license_key=valid_key) is True
assert license_check.verify_license_without_api_request(public_key=public_key, license_key="not-a-license") is not True
assert (
license_check.verify_license_without_api_request(public_key=public_key, license_key="not-a-license") is not True
)
assert license_check.airgapped_license_data is None
@ -157,3 +158,121 @@ def test_is_premium_follows_the_license_server_reply_for_an_unsigned_license(
assert license_check.is_premium() is premium
assert set(requested) == {"https://license.litellm.ai/verify_license/license-the-public-key-did-not-sign"}
@pytest.fixture
def license_logs(monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture) -> pytest.LogCaptureFixture:
logger: Final = logging.getLogger("test.license_verification")
monkeypatch.setattr("litellm.proxy.auth.litellm_license.verbose_proxy_logger", logger)
caplog.set_level(logging.DEBUG, logger=logger.name)
return caplog
def _assert_no_license_in_logs(caplog: pytest.LogCaptureFixture, license_value: str) -> None:
records: Final = tuple(record for record in caplog.records if record.name == "test.license_verification")
assert records
for record in records:
assert license_value not in logging.Formatter().format(record)
assert license_value not in repr(record.args)
assert record.exc_info is None
@pytest.mark.parametrize("configured_at_init", [True, False])
@pytest.mark.parametrize("premium", [True, False])
def test_license_success_logs_do_not_include_license(
monkeypatch: pytest.MonkeyPatch,
license_logs: pytest.LogCaptureFixture,
configured_at_init: bool,
premium: bool,
) -> None:
license_value: Final = "test-only-private-license-marker"
monkeypatch.delenv("LITELLM_LICENSE", raising=False)
if configured_at_init:
monkeypatch.setenv("LITELLM_LICENSE", license_value)
license_check: Final = LicenseCheck()
monkeypatch.setenv("LITELLM_LICENSE", license_value)
with httpx.Client(
transport=httpx.MockTransport(lambda request: httpx.Response(200, json={"verify": premium}))
) as client:
license_check.http_handler = HTTPHandler(client=client)
assert license_check.is_premium() is premium
assert "License configured after refresh: True" in license_logs.text
assert f"License is premium={premium}" in license_logs.text
_assert_no_license_in_logs(license_logs, license_value)
@pytest.mark.parametrize("status_code", [404, 401, 403, 500, 503])
def test_license_http_error_logs_preserve_status_without_license(
license_logs: pytest.LogCaptureFixture,
status_code: int,
) -> None:
license_value: Final = "test-only-private-license-marker"
license_check: Final = LicenseCheck()
with httpx.Client(
transport=httpx.MockTransport(lambda request: httpx.Response(status_code, text=license_value))
) as client:
license_check.http_handler = HTTPHandler(client=client)
assert license_check._verify(license_value) is False
assert f"error_type=HTTPStatusError status_code={status_code}" in license_logs.text
_assert_no_license_in_logs(license_logs, license_value)
@pytest.mark.parametrize("error_class", [httpx.ReadTimeout, httpx.ConnectError, ValueError])
def test_license_exception_logs_omit_secret_message_and_traceback(
license_logs: pytest.LogCaptureFixture,
error_class: type[Exception],
) -> None:
license_value: Final = "test-only-private-license-marker"
license_check: Final = LicenseCheck()
def fail(request: httpx.Request) -> httpx.Response:
try:
raise RuntimeError(license_value)
except RuntimeError as cause:
raise error_class(f"Failed verification: {request.url}") from cause
with httpx.Client(transport=httpx.MockTransport(fail)) as client:
license_check.http_handler = HTTPHandler(client=client)
assert license_check._verify(license_value) is False
assert f"error_type={error_class.__name__}" in license_logs.text
_assert_no_license_in_logs(license_logs, license_value)
def test_license_malformed_response_does_not_log_echoed_license(
license_logs: pytest.LogCaptureFixture,
) -> None:
license_value: Final = "test-only-private-license-marker"
license_check: Final = LicenseCheck()
with httpx.Client(
transport=httpx.MockTransport(lambda request: httpx.Response(200, json={"verify": license_value}))
) as client:
license_check.http_handler = HTTPHandler(client=client)
assert license_check._verify(license_value) is False
_assert_no_license_in_logs(license_logs, license_value)
@pytest.mark.parametrize("expiration_date", ["2999-01-01", "2000-01-01", "private-invalid-expiry-marker"])
def test_local_license_logs_omit_signed_payload_and_validation_errors(
license_logs: pytest.LogCaptureFixture,
expiration_date: str,
) -> None:
payload_marker: Final = "test-only-private-payload-marker"
public_key, license_value = _signed_license(expiration_date, allowed_features=(payload_marker,))
license_check: Final = LicenseCheck()
result: Final = license_check.verify_license_without_api_request(public_key, license_value)
assert (result is True) is (expiration_date == "2999-01-01")
_assert_no_license_in_logs(license_logs, license_value)
assert payload_marker not in license_logs.text
assert expiration_date not in license_logs.text
def test_local_license_exception_does_not_log_license(
license_logs: pytest.LogCaptureFixture,
) -> None:
_public_key, license_value = _signed_license("2999-01-01")
failing_public_key: Final = MagicMock()
failing_public_key.verify.side_effect = ValueError(license_value)
license_check: Final = LicenseCheck()
assert license_check.verify_license_without_api_request(failing_public_key, license_value) is False
assert "error_type=ValueError" in license_logs.text
_assert_no_license_in_logs(license_logs, license_value)