fix(datetime_utils): raise TypeError for non-str/datetime input

Greptile flagged that narrowing get_mcp_oauth_user_credential_status from a
bare except to except ValueError left a truthy non-string expires_at (possible
in raw DB JSON) raising AttributeError through the endpoint as a 500. The same
gap applied to every refactored site that catches (ValueError, TypeError):
datetime.fromisoformat raised TypeError for non-strings, which those handlers
were written against, but the helper's passthrough branch leaked
AttributeError instead.

parse_utc_datetime now validates its input and raises TypeError for anything
that is not str or datetime, restoring the stdlib error contract at every call
site. The MCP status endpoint catches it and also omits non-string
expires_at/connected_at values from the response instead of failing response
validation.
This commit is contained in:
ryan-crabbe-berri 2026-07-21 14:42:43 -07:00
parent c3eb118f8c
commit e315984f54
4 changed files with 72 additions and 9 deletions

View file

@ -17,8 +17,17 @@ def parse_utc_datetime(value: str | datetime) -> datetime:
(key expiry checks, budget windows, spend reports). The "Z" suffix is handled
explicitly because datetime.fromisoformat only accepts it from Python 3.11 and the
project floor is 3.10.
Raises ValueError for unparseable strings and TypeError for any other type,
mirroring datetime.fromisoformat's own contract so existing
``except (ValueError, TypeError)`` handlers stay fail-closed.
"""
parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) if isinstance(value, str) else value
if isinstance(value, str):
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
elif isinstance(value, datetime):
parsed = value
else:
raise TypeError(f"parse_utc_datetime expects str or datetime, got {type(value).__name__}")
if parsed.tzinfo is None:
return parsed.replace(tzinfo=timezone.utc)
return parsed

View file

@ -2005,14 +2005,15 @@ if MCP_AVAILABLE:
try:
exp = parse_utc_datetime(expires_at)
is_expired = exp < datetime.now(timezone.utc)
except ValueError:
except (ValueError, TypeError):
pass
connected_at = cred.get("connected_at")
return MCPOAuthUserCredentialStatus(
server_id=server_id,
has_credential=True,
expires_at=expires_at,
expires_at=expires_at if isinstance(expires_at, str) else None,
is_expired=is_expired,
connected_at=cred.get("connected_at"),
connected_at=connected_at if isinstance(connected_at, str) else None,
)
@router.get(

View file

@ -39,11 +39,17 @@ def test_aware_datetime_passthrough_unchanged():
def test_result_always_comparable_to_utc_now():
for value in ("2026-01-20T00:00:00", "2026-01-20T00:00:00Z", datetime(2026, 1, 20)):
assert parse_utc_datetime(value) < datetime.now(timezone.utc) or parse_utc_datetime(
value
) >= datetime.now(timezone.utc)
assert parse_utc_datetime(value) < datetime.now(timezone.utc) or parse_utc_datetime(value) >= datetime.now(
timezone.utc
)
def test_invalid_string_raises():
with pytest.raises(ValueError):
parse_utc_datetime("not-a-date")
def test_non_str_non_datetime_raises_typeerror():
for bad in (1755000000, 1755000000.0, None, {"expires_at": "2026-01-01"}):
with pytest.raises(TypeError):
parse_utc_datetime(bad) # pyright: ignore[reportArgumentType] # exercising the runtime guard

View file

@ -3565,6 +3565,45 @@ async def test_get_mcp_oauth_user_credential_status_naive_past_expiry_is_expired
assert result.expires_at == "2020-01-01T00:00:00"
@pytest.mark.asyncio
async def test_get_mcp_oauth_user_credential_status_non_string_expiry_degrades_gracefully():
"""A malformed non-string expires_at must report status instead of raising a 500."""
if not mgmt_endpoints.MCP_AVAILABLE:
pytest.skip("MCP module not installed")
from litellm.proxy.management_endpoints.mcp_management_endpoints import (
get_mcp_oauth_user_credential_status,
)
server_id = "srv-1"
stored_payload = {
"type": "oauth2",
"access_token": "tok",
"expires_at": 1577836800,
"connected_at": "2019-01-01T00:00:00+00:00",
"server_id": server_id,
}
with (
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints.get_prisma_client_or_throw",
return_value=_make_prisma_client(),
),
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints.get_user_oauth_credential",
new=AsyncMock(return_value=stored_payload),
),
):
result = await get_mcp_oauth_user_credential_status(
server_id=server_id,
user_api_key_dict=_make_user_auth("user-123"),
)
assert result.has_credential is True
assert result.is_expired is False
assert result.expires_at is None
@pytest.mark.asyncio
async def test_delete_mcp_oauth_user_credential_only_deletes_oauth():
"""delete_mcp_oauth_user_credential only deletes OAuth2 credentials, not BYOK."""
@ -5021,7 +5060,9 @@ def _edit_endpoint_patches(old_record, update_mock):
),
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints.get_mcp_server",
AsyncMock(side_effect=old_record) if isinstance(old_record, Exception) else AsyncMock(return_value=old_record),
AsyncMock(side_effect=old_record)
if isinstance(old_record, Exception)
else AsyncMock(return_value=old_record),
),
patch(
"litellm.proxy.management_endpoints.mcp_management_endpoints.update_mcp_server",
@ -5430,7 +5471,13 @@ def test_bundled_openapi_registry_parses_and_entries_are_well_formed():
registry_path = os.path.join(
os.path.dirname(os.path.abspath(__file__)),
"..", "..", "..", "..", "litellm", "proxy", "openapi_registry.json",
"..",
"..",
"..",
"..",
"litellm",
"proxy",
"openapi_registry.json",
)
with open(registry_path) as f:
registry = json.load(f)