mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-05 02:41:56 +00:00
fix(passthrough): mask dangling private keys by base64 shape instead of alphabet
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
parent
0005428a76
commit
b84ae93341
2 changed files with 24 additions and 3 deletions
|
|
@ -948,7 +948,7 @@ def _headers_without_body_framing(headers: httpx.Headers) -> httpx.Headers:
|
|||
|
||||
|
||||
_DANGLING_PRIVATE_KEY: Final = re.compile(
|
||||
r"-----BEGIN[A-Z \-]*PRIVATE KEY-----(?:[A-Za-z0-9+/= \\nr]{32,}|[A-Za-z0-9+/= \\nr]*$)"
|
||||
r"-----BEGIN[A-Z \-]*PRIVATE KEY-----[ \\nr]*(?:[A-Za-z0-9+/=]{40,}|[A-Za-z0-9+/=]{16,}$|$)"
|
||||
)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -8498,12 +8498,33 @@ async def test_upstream_error_body_for_log_masks_a_pem_after_a_json_escaped_newl
|
|||
assert output.rstrip().endswith("REDACTED"), output
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"body",
|
||||
[
|
||||
'{"error":{"message":"-----BEGIN PRIVATE KEY----- header is missing from the uploaded credentials file please re-upload it"}}',
|
||||
"Invalid JWT Signature. The private_key field must start with -----BEGIN PRIVATE KEY----- and contain the PEM encoded key from your service account JSON file",
|
||||
"-----BEGIN RSA PRIVATE KEY----- block could not be parsed",
|
||||
"No key could be detected. Expected -----BEGIN PRIVATE KEY----- header in the service account JSON field private_key",
|
||||
"private_key must start with -----BEGIN PRIVATE KEY----- and end with -----END PRIVATE KEY-----",
|
||||
],
|
||||
ids=["missing-header", "jwt-signature", "rsa-parse", "expected-header", "must-start-with"],
|
||||
)
|
||||
async def test_upstream_error_body_for_log_keeps_prose_mentioning_a_pem_header(body: str):
|
||||
"""Error prose that mentions a PEM header without key material must reach the
|
||||
log exactly as the base pipeline's redaction produced it: the dangling mask
|
||||
adds nothing."""
|
||||
output: Final = _upstream_error_body_for_log(body.encode(), "utf-8", redact_secrets)
|
||||
expected: Final = redact_secrets(_sanitize_upstream_error_body(body))
|
||||
assert output == expected, output
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_upstream_error_body_for_log_masks_a_pem_header_at_the_preview_cut():
|
||||
"""A header so close to the head cut that fewer than 32 key chars fit must
|
||||
still be masked instead of leaking the partial key."""
|
||||
prefix: Final = "x" * (PASSTHROUGH_UPSTREAM_ERROR_BODY_MAX_LOG_CHARS - len("-----BEGIN PRIVATE KEY-----") - 11)
|
||||
sanitized: Final = prefix + " -----BEGIN PRIVATE KEY----- MIIEvQIBAD" + "y" * 300
|
||||
prefix: Final = "x" * (PASSTHROUGH_UPSTREAM_ERROR_BODY_MAX_LOG_CHARS - len("-----BEGIN PRIVATE KEY-----") - 22)
|
||||
sanitized: Final = prefix + " -----BEGIN PRIVATE KEY----- MIIEvQIBADANBgkqhkiG" + "y" * 300
|
||||
key_start: Final = sanitized.index("MIIEvQIBAD")
|
||||
assert key_start < PASSTHROUGH_UPSTREAM_ERROR_BODY_MAX_LOG_CHARS
|
||||
assert PASSTHROUGH_UPSTREAM_ERROR_BODY_MAX_LOG_CHARS - key_start < 32
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue