scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-09 22:31:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 133

fix(s3_v2): encode the object key and sign the URL we send

Browse source 
S3SigV4Auth now signs these requests, which fixes an object key containing a space.
The path itself is still produced by requests.Request(...).prepare(), so it is not
percent-encoded the way S3 canonicalizes it and three shapes still misbehave.

"+" and "&" are left literal where S3 encodes them as %2B and %26, so the canonical
request differs from ours and the upload fails with 403 SignatureDoesNotMatch. "#"
is worse: requests treats it as a fragment delimiter, truncates the key there, and
the log is written to the wrong S3 object with a 200 and no error anywhere.

Build the object URL once, percent-encoding the key a single time and normalizing
the result with httpx so the string that is signed is byte-identical to the one
httpx transmits, and hand that URL straight to AWSRequest instead of routing it
through requests.

The key is encoded without being decoded first, so an alias holding a literal
percent escape stays a single path segment; decoding it would turn %2F into real
separators and move the object out of its prefix, and out of its bucket on the
path-style URL.
This commit is contained in:
Yucheng He 2026-08-04 12:06:55 -07:00 • committed by milan
parent ff02d5cfc0
commit d7bf49321c
2 changed files with 282 additions and 76 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

101
litellm/integrations/s3_v2.py
View file

@ -11,6 +11,9 @@ import time
from collections.abc import Mapping
from datetime import datetime
from typing import Final, cast
from urllib.parse import quote
import httpx
import litellm
from litellm._logging import print_verbose, verbose_logger
@ -287,12 +290,27 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM):
verbose_logger.exception("s3 Layer Error - %s", e)
self.handle_callback_failure(callback_name="S3Logger")
def _build_object_url(self, s3_object_key: str) -> str:
"""
Build the object URL, percent-encoding the key once and normalizing it the way httpx will.
The returned string is what gets both signed and sent; the two must be byte-identical or S3
answers 403 SignatureDoesNotMatch.
"""
encoded_key: Final = quote(s3_object_key, safe="/")
if not self.s3_endpoint_url or not self.s3_bucket_name:
return str(httpx.URL(f"https://{self.s3_bucket_name}.s3.{self.s3_region_name}.amazonaws.com/{encoded_key}"))
if self.s3_use_virtual_hosted_style:
endpoint_host: Final = self.s3_endpoint_url.replace("https://", "").replace("http://", "")
protocol: Final = "https://" if self.s3_endpoint_url.startswith("https://") else "http://"
return str(httpx.URL(f"{protocol}{self.s3_bucket_name}.{endpoint_host}/{encoded_key}"))
return str(httpx.URL(f"{self.s3_endpoint_url}/{self.s3_bucket_name}/{encoded_key}"))
async def async_upload_data_to_s3(self, batch_logging_element: s3BatchLoggingElement):
try:
import base64
import hashlib
import requests
from botocore.auth import S3SigV4Auth
from botocore.awsrequest import AWSRequest
except ImportError:
@ -316,18 +334,7 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM):
verbose_logger.debug("s3_v2 logger - uploading data to s3 - %s", batch_logging_element.s3_object_key)
verbose_logger.debug("s3_v2 logger - s3_verify setting: %s", self.s3_verify)
# Prepare the URL
url = f"https://{self.s3_bucket_name}.s3.{self.s3_region_name}.amazonaws.com/{batch_logging_element.s3_object_key}"
if self.s3_endpoint_url and self.s3_bucket_name:
if self.s3_use_virtual_hosted_style:
# Virtual-hosted-style: bucket.endpoint/key
endpoint_host: Final = self.s3_endpoint_url.replace("https://", "").replace("http://", "")
protocol: Final = "https://" if self.s3_endpoint_url.startswith("https://") else "http://"
url = f"{protocol}{self.s3_bucket_name}.{endpoint_host}/{batch_logging_element.s3_object_key}"
else:
# Path-style: endpoint/bucket/key
url = self.s3_endpoint_url + "/" + self.s3_bucket_name + "/" + batch_logging_element.s3_object_key
request_url: Final = self._build_object_url(batch_logging_element.s3_object_key)
# Convert JSON to string
json_string: Final = safe_dumps(batch_logging_element.payload)
@ -348,25 +355,13 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM):
"Cache-Control": "private, immutable, max-age=31536000, s-maxage=0",
**self._sse_headers(),
}
req: Final = requests.Request("PUT", url, data=json_string, headers=headers)
prepped: Final = req.prepare()
# Sign the request
aws_request: Final = AWSRequest(
method=prepped.method,
url=prepped.url,
data=prepped.body,
headers=prepped.headers,
)
aws_request: Final = AWSRequest(method="PUT", url=request_url, data=json_string, headers=headers)
aws_region_name: Final = self.get_aws_region_name_for_non_llm_api_calls(aws_region_name=self.s3_region_name)
S3SigV4Auth(credentials, "s3", aws_region_name).add_auth(aws_request)
# Prepare the signed headers
signed_headers: Final = dict(aws_request.headers.items())
# Use prepared URL so path segments match SigV4 canonical request (e.g. %20 for spaces).
request_url: Final = prepped.url or url
# Make the request with retry for transient S3 errors (500/503)
max_retries: Final = 3
for attempt in range(max_retries):
@ -478,7 +473,6 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM):
import base64
import hashlib
import requests
from botocore.auth import S3SigV4Auth
from botocore.awsrequest import AWSRequest
from botocore.credentials import Credentials
@ -493,18 +487,7 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM):
aws_region_name=self.s3_region_name,
)
# Prepare the URL
url = f"https://{self.s3_bucket_name}.s3.{self.s3_region_name}.amazonaws.com/{batch_logging_element.s3_object_key}"
if self.s3_endpoint_url and self.s3_bucket_name:
if self.s3_use_virtual_hosted_style:
# Virtual-hosted-style: bucket.endpoint/key
endpoint_host: Final = self.s3_endpoint_url.replace("https://", "").replace("http://", "")
protocol: Final = "https://" if self.s3_endpoint_url.startswith("https://") else "http://"
url = f"{protocol}{self.s3_bucket_name}.{endpoint_host}/{batch_logging_element.s3_object_key}"
else:
# Path-style: endpoint/bucket/key
url = self.s3_endpoint_url + "/" + self.s3_bucket_name + "/" + batch_logging_element.s3_object_key
request_url: Final = self._build_object_url(batch_logging_element.s3_object_key)
# Convert JSON to string
json_string: Final = safe_dumps(batch_logging_element.payload)
@ -525,25 +508,13 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM):
"Cache-Control": "private, immutable, max-age=31536000, s-maxage=0",
**self._sse_headers(),
}
req: Final = requests.Request("PUT", url, data=json_string, headers=headers)
prepped: Final = req.prepare()
# Sign the request
aws_request: Final = AWSRequest(
method=prepped.method,
url=prepped.url,
data=prepped.body,
headers=prepped.headers,
)
aws_request: Final = AWSRequest(method="PUT", url=request_url, data=json_string, headers=headers)
aws_region_name: Final = self.get_aws_region_name_for_non_llm_api_calls(aws_region_name=self.s3_region_name)
S3SigV4Auth(credentials, "s3", aws_region_name).add_auth(aws_request)
# Prepare the signed headers
signed_headers: Final = dict(aws_request.headers.items())
# Use prepared URL so path segments match SigV4 canonical request (e.g. %20 for spaces).
request_url: Final = prepped.url or url
httpx_client: Final = _get_httpx_client(
params=({"ssl_verify": self.s3_verify} if self.s3_verify is not None else None)
)
@ -582,7 +553,6 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM):
try:
import hashlib
import requests
from botocore.auth import S3SigV4Auth
from botocore.awsrequest import AWSRequest
except ImportError:
@ -607,18 +577,7 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM):
verbose_logger.debug("s3_v2 logger - downloading data from s3 - %s", s3_object_key)
# Prepare the URL
url = f"https://{self.s3_bucket_name}.s3.{self.s3_region_name}.amazonaws.com/{s3_object_key}"
if self.s3_endpoint_url and self.s3_bucket_name:
if self.s3_use_virtual_hosted_style:
# Virtual-hosted-style: bucket.endpoint/key
endpoint_host: Final = self.s3_endpoint_url.replace("https://", "").replace("http://", "")
protocol: Final = "https://" if self.s3_endpoint_url.startswith("https://") else "http://"
url = f"{protocol}{self.s3_bucket_name}.{endpoint_host}/{s3_object_key}"
else:
# Path-style: endpoint/bucket/key
url = self.s3_endpoint_url + "/" + self.s3_bucket_name + "/" + s3_object_key
request_url: Final = self._build_object_url(s3_object_key)
# Prepare the request for GET operation
# For GET requests, we need x-amz-content-sha256 with hash of empty string
@ -626,21 +585,11 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM):
headers: Final = {
"x-amz-content-sha256": empty_string_hash,
}
req: Final = requests.Request("GET", url, headers=headers)
prepped: Final = req.prepare()
# Sign the request
aws_request: Final = AWSRequest(
method=prepped.method,
url=prepped.url,
headers=prepped.headers,
)
aws_request: Final = AWSRequest(method="GET", url=request_url, headers=headers)
S3SigV4Auth(credentials, "s3", self.s3_region_name).add_auth(aws_request)
# Prepare the signed headers
signed_headers: Final = dict(aws_request.headers.items())
request_url: Final = prepped.url or url
response: Final = await self.async_httpx_client.get(request_url, headers=signed_headers)
if response.status_code != 200:

257
tests/test_litellm/integrations/test_s3_v2.py
View file

@ -2,6 +2,7 @@ import asyncio
from datetime import datetime
from unittest.mock import MagicMock, patch
import httpx
import pytest
from litellm.integrations.s3_v2 import S3Logger
@ -1760,3 +1761,259 @@ async def test_download_signs_object_key_with_space_the_way_s3_does():
body=None,
headers=call.kwargs["headers"],
)
AWS_TEST_ACCESS_KEY = "AKIAIOSFODNN7EXAMPLE"
AWS_TEST_SECRET_KEY = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
S3_OBJECT_KEY_CASES = (
("space", "LOGS/LLM AI Projects/2026-08-04/log.json", "/LOGS/LLM%20AI%20Projects/2026-08-04/log.json"),
("plus", "LOGS/Team+Plus/2026-08-04/log.json", "/LOGS/Team%2BPlus/2026-08-04/log.json"),
("ampersand", "LOGS/A&B Team/2026-08-04/log.json", "/LOGS/A%26B%20Team/2026-08-04/log.json"),
("hash", "LOGS/Team#Hash/2026-08-04/log.json", "/LOGS/Team%23Hash/2026-08-04/log.json"),
("unicode", "LOGS/Equipe Café/2026-08-04/log.json", "/LOGS/Equipe%20Caf%C3%A9/2026-08-04/log.json"),
("percent", "LOGS/100%Team/2026-08-04/log.json", "/LOGS/100%25Team/2026-08-04/log.json"),
)
def _signing_logger() -> S3Logger:
return S3Logger(
s3_bucket_name="test-bucket",
s3_aws_access_key_id=AWS_TEST_ACCESS_KEY,
s3_aws_secret_access_key=AWS_TEST_SECRET_KEY,
s3_region_name="us-east-1",
)
def _resign(signer_cls, method: str, url: str, body, sent_headers):
"""Recompute the signature over the URL actually put on the wire, the way S3 does."""
from botocore.awsrequest import AWSRequest
from botocore.credentials import Credentials
lowered = {name.lower(): value for name, value in sent_headers.items()}
signed_names = lowered["authorization"].split("SignedHeaders=")[1].split(",")[0].split(";")
request = AWSRequest(
method=method,
url=url,
data=body,
headers={name: lowered[name] for name in signed_names if name in lowered},
)
request.context["timestamp"] = lowered["x-amz-date"]
signer = signer_cls(Credentials(AWS_TEST_ACCESS_KEY, AWS_TEST_SECRET_KEY), "s3", "us-east-1")
return signer.signature(signer.string_to_sign(request, signer.canonical_request(request)), request)
def _assert_signature_matches_wire(method: str, url: str, body, sent_headers, expected_path: str):
"""Assert the sent signature is the one S3 computes, and not the generic SigV4Auth one."""
from urllib.parse import urlsplit
from botocore.auth import S3SigV4Auth, SigV4Auth
assert urlsplit(url).path == expected_path
assert httpx.Request(method, url).url.raw_path.decode().split("?")[0] == expected_path
sent_signature = sent_headers["Authorization"].split("Signature=")[1].strip()
assert sent_signature == _resign(S3SigV4Auth, method, url, body, sent_headers)
assert sent_signature != _resign(SigV4Auth, method, url, body, sent_headers)
def _put_element(s3_object_key: str):
from litellm.types.integrations.s3_v2 import s3BatchLoggingElement
return s3BatchLoggingElement(
s3_object_key=s3_object_key,
payload={"test": "lit5146"},
s3_object_download_filename="log.json",
)
@pytest.mark.parametrize(
"case_id,s3_object_key,expected_path",
S3_OBJECT_KEY_CASES,
ids=[case[0] for case in S3_OBJECT_KEY_CASES],
)
@pytest.mark.asyncio
async def test_async_upload_signs_the_url_it_sends(case_id, s3_object_key, expected_path):
from unittest.mock import AsyncMock
logger = _signing_logger()
response = MagicMock()
response.status_code = 200
response.raise_for_status = MagicMock()
logger.async_httpx_client = AsyncMock()
logger.async_httpx_client.put.return_value = response
await logger.async_upload_data_to_s3(_put_element(s3_object_key))
call = logger.async_httpx_client.put.call_args
_assert_signature_matches_wire(
method="PUT",
url=call[0][0],
body=call.kwargs["data"].encode("utf-8"),
sent_headers=call.kwargs["headers"],
expected_path=expected_path,
)
@pytest.mark.parametrize(
"case_id,s3_object_key,expected_path",
S3_OBJECT_KEY_CASES,
ids=[case[0] for case in S3_OBJECT_KEY_CASES],
)
def test_sync_upload_signs_the_url_it_sends(case_id, s3_object_key, expected_path):
logger = _signing_logger()
response = MagicMock()
response.status_code = 200
response.raise_for_status = MagicMock()
sync_client = MagicMock()
sync_client.put.return_value = response
with patch("litellm.integrations.s3_v2._get_httpx_client", return_value=sync_client):
logger.upload_data_to_s3(_put_element(s3_object_key))
call = sync_client.put.call_args
_assert_signature_matches_wire(
method="PUT",
url=call[0][0],
body=call.kwargs["data"].encode("utf-8"),
sent_headers=call.kwargs["headers"],
expected_path=expected_path,
)
@pytest.mark.parametrize(
"case_id,s3_object_key,expected_path",
S3_OBJECT_KEY_CASES,
ids=[case[0] for case in S3_OBJECT_KEY_CASES],
)
@pytest.mark.asyncio
async def test_download_signs_the_url_it_sends(case_id, s3_object_key, expected_path):
from unittest.mock import AsyncMock
logger = _signing_logger()
response = MagicMock()
response.status_code = 200
response.json = MagicMock(return_value={"downloaded": "ok"})
logger.async_httpx_client = AsyncMock()
logger.async_httpx_client.get.return_value = response
assert await logger._download_object_from_s3(s3_object_key) == {"downloaded": "ok"}
call = logger.async_httpx_client.get.call_args
_assert_signature_matches_wire(
method="GET",
url=call[0][0],
body=None,
sent_headers=call.kwargs["headers"],
expected_path=expected_path,
)
credential = call.kwargs["headers"]["Authorization"].split("Credential=")[1].split(",")[0]
assert credential.split("/")[2] == "us-east-1"
@pytest.mark.asyncio
async def test_async_upload_preserves_payload_hash_header():
"""S3SigV4Auth recomputes x-amz-content-sha256; it must still describe the body we send."""
import hashlib
from unittest.mock import AsyncMock
logger = _signing_logger()
response = MagicMock()
response.status_code = 200
response.raise_for_status = MagicMock()
logger.async_httpx_client = AsyncMock()
logger.async_httpx_client.put.return_value = response
await logger.async_upload_data_to_s3(_put_element("LOGS/LLM AI Projects/2026-08-04/log.json"))
call = logger.async_httpx_client.put.call_args
sent = {name.lower(): value for name, value in call.kwargs["headers"].items()}
assert sent["x-amz-content-sha256"] == hashlib.sha256(call.kwargs["data"].encode("utf-8")).hexdigest()
@pytest.mark.parametrize(
"s3_object_key,expected_path",
[
("LOGS/a/./b/log.json", "/LOGS/a/b/log.json"),
("LOGS/a/../b/log.json", "/LOGS/b/log.json"),
],
)
@pytest.mark.asyncio
async def test_dot_segment_keys_keep_collapsing_and_stay_signable(s3_object_key, expected_path):
"""Dot segments collapse as before, and the signature covers the collapsed path."""
from unittest.mock import AsyncMock
from botocore.auth import S3SigV4Auth
logger = _signing_logger()
response = MagicMock()
response.status_code = 200
response.raise_for_status = MagicMock()
logger.async_httpx_client = AsyncMock()
logger.async_httpx_client.put.return_value = response
await logger.async_upload_data_to_s3(_put_element(s3_object_key))
call = logger.async_httpx_client.put.call_args
url = call[0][0]
assert str(httpx.URL(url)) == url
from urllib.parse import urlsplit
assert urlsplit(url).path == expected_path
sent_signature = call.kwargs["headers"]["Authorization"].split("Signature=")[1].strip()
assert sent_signature == _resign(
S3SigV4Auth, "PUT", url, call.kwargs["data"].encode("utf-8"), call.kwargs["headers"]
)
@pytest.mark.parametrize(
"endpoint_url,virtual_hosted,expected_url",
[
(None, False, "https://test-bucket.s3.us-east-1.amazonaws.com/LOGS/My%20Team/log.json"),
("https://s3.example.com", False, "https://s3.example.com/test-bucket/LOGS/My%20Team/log.json"),
("https://s3.example.com", True, "https://test-bucket.s3.example.com/LOGS/My%20Team/log.json"),
("http://localhost:9000", False, "http://localhost:9000/test-bucket/LOGS/My%20Team/log.json"),
],
)
def test_build_object_url_shapes(endpoint_url, virtual_hosted, expected_url):
"""Each endpoint shape must encode the object key and leave the rest of the URL alone."""
logger = S3Logger(
s3_bucket_name="test-bucket",
s3_endpoint_url=endpoint_url,
s3_use_virtual_hosted_style=virtual_hosted,
s3_aws_access_key_id=AWS_TEST_ACCESS_KEY,
s3_aws_secret_access_key=AWS_TEST_SECRET_KEY,
s3_region_name="us-east-1",
)
built = logger._build_object_url("LOGS/My Team/log.json")
assert built == expected_url
@pytest.mark.parametrize(
"s3_object_key",
[
"LOGS/LLM AI Projects/log.json",
"LOGS/100%Team/log.json",
"LOGS/A%20B/log.json",
"LOGS/Sale%2FDiscount/log.json",
"LOGS/..%2F..%2Fescape/2026-08-04/log.json",
],
)
def test_object_key_round_trips_through_the_url(s3_object_key):
"""The path we send must decode back to the exact key we were asked to write."""
from urllib.parse import unquote, urlsplit
url = _signing_logger()._build_object_url(s3_object_key)
assert unquote(urlsplit(url).path.lstrip("/")) == s3_object_key
def test_object_key_cannot_escape_the_bucket_path_style():
"""A key holding encoded traversal must stay one path segment, bucket included."""
logger = S3Logger(
s3_bucket_name="test-bucket",
s3_endpoint_url="https://minio.internal",
s3_aws_access_key_id=AWS_TEST_ACCESS_KEY,
s3_aws_secret_access_key=AWS_TEST_SECRET_KEY,
s3_region_name="us-east-1",
)
url = logger._build_object_url("LOGS/..%2F..%2Fescape/log.json")
assert url.startswith("https://minio.internal/test-bucket/LOGS/")