refactor(proxy): ban raw datetime.fromisoformat via parse_utc_datetime helper

Extract parse_utc_datetime into litellm_core_utils/datetime_utils.py as the
single ISO-8601 parse entrypoint for the proxy: it accepts str or datetime,
handles the Z suffix on Python 3.10, and assumes naive values are UTC so
results always compare safely against datetime.now(timezone.utc).

Convert all 27 raw fromisoformat sites under litellm/proxy/ to the helper,
collapsing the hand-rolled tzinfo guards they each carried. This also fixes a
real bug in get_mcp_oauth_user_credential_status: a tz-naive expires_at raised
TypeError against the aware now(), the bare except swallowed it, and the
credential never read as expired.

Two semgrep rules enforce the pattern in CI: raw fromisoformat is banned under
litellm/proxy/, and a repo-wide taint rule flags comparing or subtracting an
unnormalized fromisoformat result.

.semgrep/rules/python/reliability/naive-datetime.yml

@@ -0,0 +1,62 @@
+# Aware-vs-naive datetime comparisons raise TypeError at runtime.
+# ISO-8601 input may carry "Z"/offset (tz-aware) or not (naive), so raw
+# datetime.fromisoformat results must be normalized before comparing.
+# parse_utc_datetime in litellm/litellm_core_utils/datetime_utils.py is the
+# single allowed parse entrypoint under litellm/proxy/.
+
+rules:
+  - id: ban-raw-fromisoformat-in-proxy
+    message: >-
+      Raw datetime.fromisoformat() is banned under litellm/proxy/. Use
+      parse_utc_datetime() from litellm.litellm_core_utils.datetime_utils, which
+      normalizes naive values to UTC so comparisons against
+      datetime.now(timezone.utc) cannot raise TypeError.
+    severity: ERROR
+    languages: [python]
+    paths:
+      include:
+        - litellm/proxy
+    pattern-either:
+      - pattern: datetime.fromisoformat(...)
+      - pattern: datetime.datetime.fromisoformat(...)
+    metadata:
+      category: reliability
+      cwe: "CWE-704: Incorrect Type Conversion or Cast"
+      tags: [python, reliability, datetime]
+      confidence: HIGH
+
+  - id: tz-unnormalized-fromisoformat-compare
+    message: >-
+      Comparing a datetime.fromisoformat(...) result without timezone
+      normalization raises TypeError when one side is tz-aware and the other is
+      naive. Use parse_utc_datetime() from
+      litellm.litellm_core_utils.datetime_utils and compare against
+      datetime.now(timezone.utc).
+    severity: ERROR
+    languages: [python]
+    mode: taint
+    pattern-sources:
+      - pattern: datetime.fromisoformat(...)
+      - pattern: datetime.datetime.fromisoformat(...)
+    pattern-sanitizers:
+      - pattern: $DT.replace(tzinfo=$TZ)
+      - by-side-effect: true
+        patterns:
+          - pattern: $DT
+          - pattern-inside: |
+              if <... $DT.tzinfo is None ...>:
+                $DT = $DT.replace(tzinfo=$TZ)
+              ...
+    pattern-sinks:
+      - patterns:
+          - pattern-either:
+              - pattern: $X > $Y
+              - pattern: $X < $Y
+              - pattern: $X >= $Y
+              - pattern: $X <= $Y
+              - pattern: $X - $Y
+    metadata:
+      category: reliability
+      cwe: "CWE-704: Incorrect Type Conversion or Cast"
+      tags: [python, reliability, datetime]
+      confidence: MEDIUM

litellm/litellm_core_utils/datetime_utils.py

@@ -0,0 +1,24 @@
+"""
+Timezone-safe datetime parsing.
+
+ISO-8601 strings arriving from API input, DB metadata, or serialized state may carry a
+timezone offset ("...Z" / "+00:00") or not. Comparing an aware datetime with a naive one
+raises TypeError, so every parse site must normalize. This module is the single allowed
+entrypoint; a semgrep rule bans raw datetime.fromisoformat under litellm/proxy/.
+"""
+
+from datetime import datetime, timezone
+
+
+def parse_utc_datetime(value: str | datetime) -> datetime:
+    """Parse an ISO-8601 string (or pass through a datetime) into a tz-aware datetime.
+
+    Naive values are assumed to be UTC, matching the convention used across the proxy
+    (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.
+    """
+    parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) if isinstance(value, str) else value
+    if parsed.tzinfo is None:
+        return parsed.replace(tzinfo=timezone.utc)
+    return parsed

litellm/proxy/_experimental/mcp_server/auth/user_api_key_auth_mcp.py

@@ -10,6 +10,7 @@ from typing_extensions import assert_never
 
 import litellm
 from litellm._logging import verbose_logger
+from litellm.litellm_core_utils.datetime_utils import parse_utc_datetime
 from litellm.proxy._experimental.mcp_server.oauth_utils import (
     get_request_base_url,
     well_known_root_suffix,
@@ -867,9 +868,7 @@ class MCPRequestHandler:
         expires = key_object.expires
         if expires is None:
             return True
-        expiry = expires if isinstance(expires, datetime) else datetime.fromisoformat(expires)
-        if expiry.tzinfo is None or expiry.tzinfo.utcoffset(expiry) is None:
-            expiry = expiry.replace(tzinfo=timezone.utc)
+        expiry = parse_utc_datetime(expires)
         return expiry >= datetime.now(timezone.utc)
 
     @staticmethod

litellm/proxy/_experimental/mcp_server/bridge_token_flow.py

@@ -11,6 +11,7 @@ from pydantic import SecretStr
 from typing_extensions import assert_never
 
 from litellm._logging import verbose_logger
+from litellm.litellm_core_utils.datetime_utils import parse_utc_datetime
 from litellm.proxy._experimental.mcp_server.oauth_utils import TOKEN_NO_CACHE_HEADERS
 from litellm.types.mcp_server.mcp_server_manager import MCPServer
 
@@ -62,7 +63,7 @@ def _key_is_active(key_obj: "UserAPIKeyAuth") -> bool:
     store) derive it separately via :func:`_active_key_user_id`.
 
     Total by design: ``expires`` is typed ``str | datetime``, and an unparseable string would make
-    ``datetime.fromisoformat`` raise. Since the callers run this outside their key-resolution
+    ``parse_utc_datetime`` raise. Since the callers run this outside their key-resolution
     ``try``, an uncaught parse error would surface as a 500 instead of the endpoint's fail-closed
     behavior, so a malformed expiry is treated as inactive (return ``False``) rather than raising.
     """
@@ -70,15 +71,10 @@ def _key_is_active(key_obj: "UserAPIKeyAuth") -> bool:
         return False
     expires = key_obj.expires
     if expires is not None:
-        if isinstance(expires, datetime):
-            expiry = expires
-        else:
-            try:
-                expiry = datetime.fromisoformat(expires)
-            except (ValueError, TypeError):
-                return False
-        if expiry.tzinfo is None or expiry.tzinfo.utcoffset(expiry) is None:
-            expiry = expiry.replace(tzinfo=timezone.utc)
+        try:
+            expiry = parse_utc_datetime(expires)
+        except (ValueError, TypeError):
+            return False
         if expiry < datetime.now(timezone.utc):
             return False
     return True

litellm/proxy/_experimental/mcp_server/db.py

@@ -8,6 +8,7 @@ from typing import TYPE_CHECKING, Any, Awaitable, Callable, Dict, Iterable, List
 from litellm._logging import verbose_proxy_logger
 from litellm._uuid import uuid
 from litellm.constants import MCP_PER_USER_TOKEN_EXPIRY_BUFFER_SECONDS
+from litellm.litellm_core_utils.datetime_utils import parse_utc_datetime
 from litellm.llms.custom_httpx.http_handler import get_async_httpx_client
 from litellm.proxy._experimental.mcp_server.auth.token_endpoint_auth import (
     build_token_endpoint_client_auth,
@@ -1192,9 +1193,7 @@ def is_oauth_credential_expired(cred: Dict[str, Any], buffer_seconds: int = 0) -
     if not expires_at:
         return False
     try:
-        exp_dt = datetime.fromisoformat(expires_at)
-        if exp_dt.tzinfo is None:
-            exp_dt = exp_dt.replace(tzinfo=timezone.utc)
+        exp_dt = parse_utc_datetime(expires_at)
         return datetime.now(timezone.utc) + timedelta(seconds=buffer_seconds) > exp_dt
     except (ValueError, TypeError):
         return False
@@ -1551,11 +1550,9 @@ def _remaining_token_seconds(expires_at: str | None) -> int | None:
     if not expires_at:
         return None
     try:
-        exp_dt = datetime.fromisoformat(expires_at)
+        exp_dt = parse_utc_datetime(expires_at)
     except (ValueError, TypeError):
         return None
-    if exp_dt.tzinfo is None:
-        exp_dt = exp_dt.replace(tzinfo=timezone.utc)
     remaining = int((exp_dt - datetime.now(timezone.utc)).total_seconds())
     return remaining if remaining > 0 else None
 

litellm/proxy/_experimental/mcp_server/outbound_credentials/v2_token_store.py

@@ -11,8 +11,8 @@ injected, so the DB/decoding plumbing stays testable and out of this seam.
 from __future__ import annotations
 
 from collections.abc import Awaitable, Callable
-from datetime import datetime, timezone
 
+from litellm.litellm_core_utils.datetime_utils import parse_utc_datetime
 from litellm.proxy._experimental.mcp_server.outbound_credentials.oauth_token_store import (
     OAuthToken,
 )
@@ -22,15 +22,9 @@ CredentialReader = Callable[[str, str], Awaitable["dict[str, object] | None"]]
 
 def _iso_to_epoch(expires_at: str) -> float | None:
     try:
-        dt = datetime.fromisoformat(expires_at)
+        return parse_utc_datetime(expires_at).timestamp()
     except ValueError:
         return None
-    # A timezone-naive expiry is stored as UTC (db.py writes ``datetime.now(timezone.utc)``),
-    # so anchor it to UTC before ``.timestamp()`` - otherwise a non-UTC host would read it as
-    # local time and skew the expiry, diverging from v1's ``_remaining_token_seconds``.
-    if dt.tzinfo is None:
-        dt = dt.replace(tzinfo=timezone.utc)
-    return dt.timestamp()
 
 
 def _to_scopes(raw: object) -> tuple[str, ...]:

litellm/proxy/auth/user_api_key_auth.py

@@ -26,6 +26,7 @@ from litellm._service_logger import ServiceLogging
 from litellm.constants import LITELLM_PROXY_MASTER_KEY_ALIAS
 from litellm.integrations.otel.model.config import is_otel_v2_enabled
 from litellm.integrations.otel.runtime import phase_span, seed_request_identity
+from litellm.litellm_core_utils.datetime_utils import parse_utc_datetime
 from litellm.litellm_core_utils.dd_tracing import tracer
 from litellm.litellm_core_utils.dot_notation_indexing import get_nested_value
 from litellm.proxy._types import *
@@ -1515,12 +1516,7 @@ async def _user_api_key_auth_builder(
         ):
             if valid_token.expires is not None:
                 current_time = datetime.now(timezone.utc)
-                if isinstance(valid_token.expires, datetime):
-                    expiry_time = valid_token.expires
-                else:
-                    expiry_time = datetime.fromisoformat(valid_token.expires)
-                if expiry_time.tzinfo is None or expiry_time.tzinfo.utcoffset(expiry_time) is None:
-                    expiry_time = expiry_time.replace(tzinfo=timezone.utc)
+                expiry_time = parse_utc_datetime(valid_token.expires)
                 if expiry_time < current_time:
                     await _delete_cache_key_object(
                         hashed_token=hash_token(api_key),
@@ -1804,12 +1800,7 @@ async def _user_api_key_auth_builder(
             # Check 3. If token is expired
             if valid_token.expires is not None:
                 current_time = datetime.now(timezone.utc)
-                if isinstance(valid_token.expires, datetime):
-                    expiry_time = valid_token.expires
-                else:
-                    expiry_time = datetime.fromisoformat(valid_token.expires)
-                if expiry_time.tzinfo is None or expiry_time.tzinfo.utcoffset(expiry_time) is None:
-                    expiry_time = expiry_time.replace(tzinfo=timezone.utc)
+                expiry_time = parse_utc_datetime(valid_token.expires)
                 verbose_proxy_logger.debug(
                     f"Checking if token expired, expiry time {expiry_time} and current time {current_time}"
                 )
@@ -2683,9 +2674,7 @@ def get_api_key_from_custom_header(request: Request, custom_litellm_key_header_n
 def _get_temp_budget_increase(valid_token: UserAPIKeyAuth):
     valid_token_metadata = valid_token.metadata
     if "temp_budget_increase" in valid_token_metadata and "temp_budget_expiry" in valid_token_metadata:
-        expiry = datetime.fromisoformat(valid_token_metadata["temp_budget_expiry"])
-        if expiry.tzinfo is None:
-            expiry = expiry.replace(tzinfo=timezone.utc)
+        expiry = parse_utc_datetime(valid_token_metadata["temp_budget_expiry"])
         if expiry > datetime.now(timezone.utc):
             return valid_token_metadata["temp_budget_increase"]
     return None
@@ -2894,12 +2883,7 @@ async def _run_post_custom_auth_checks(
     # 2. Check token expiry
     if valid_token.expires is not None:
         current_time = datetime.now(timezone.utc)
-        if isinstance(valid_token.expires, datetime):
-            expiry_time = valid_token.expires
-        else:
-            expiry_time = datetime.fromisoformat(valid_token.expires)
-        if expiry_time.tzinfo is None or expiry_time.tzinfo.utcoffset(expiry_time) is None:
-            expiry_time = expiry_time.replace(tzinfo=timezone.utc)
+        expiry_time = parse_utc_datetime(valid_token.expires)
         if expiry_time < current_time:
             raise ProxyException(
                 message=f"Authentication Error - Expired Key. Key Expiry time {expiry_time} and current time {current_time}",

litellm/proxy/client/cli/commands/keys.py

@@ -7,6 +7,8 @@ import rich
 import requests
 from rich.table import Table
 
+from litellm.litellm_core_utils.datetime_utils import parse_utc_datetime
+
 from ...keys import KeysManagementClient
 
 
@@ -224,22 +226,14 @@ def _filter_keys_by_created_since(
     if not created_since_dt:
         return source_keys
 
+    created_since_utc = parse_utc_datetime(created_since_dt)
     filtered_keys = []
     for key in source_keys:
         key_created_at = key.get("created_at")
         if key_created_at:
-            # Parse the key's created_at timestamp
             if isinstance(key_created_at, str):
-                if "T" in key_created_at:
-                    key_dt = datetime.fromisoformat(key_created_at.replace("Z", "+00:00"))
-                else:
-                    key_dt = datetime.fromisoformat(key_created_at)
-
-                # Convert to naive datetime for comparison (assuming UTC)
-                if key_dt.tzinfo:
-                    key_dt = key_dt.replace(tzinfo=None)
-
-                if key_dt >= created_since_dt:
+                key_dt = parse_utc_datetime(key_created_at)
+                if key_dt >= created_since_utc:
                     filtered_keys.append(key)
 
     click.echo(f"Filtered {len(source_keys)} keys to {len(filtered_keys)} keys created since {created_since}")
@@ -262,7 +256,7 @@ def _display_dry_run_table(source_keys: List[Dict[str, Any]]) -> None:
             if isinstance(created_at, str):
                 # Handle common timestamp formats
                 if "T" in created_at:
-                    dt = datetime.fromisoformat(created_at.replace("Z", "+00:00"))
+                    dt = parse_utc_datetime(created_at)
                     created_at = dt.strftime("%Y-%m-%d %H:%M")
 
         table.add_row(str(key.get("key_alias", "")), str(key.get("user_id", "")), str(created_at))

litellm/proxy/client/cli/commands/models.py

@@ -11,6 +11,8 @@ import click
 import rich
 
 # local imports
+from litellm.litellm_core_utils.datetime_utils import parse_utc_datetime
+
 from ... import Client
 
 
@@ -51,8 +53,7 @@ def format_iso_datetime_str(iso_datetime_str: Optional[str]) -> str:
     if not iso_datetime_str:
         return ""
     try:
-        # Parse ISO format datetime string
-        dt = datetime.fromisoformat(iso_datetime_str.replace("Z", "+00:00"))
+        dt = parse_utc_datetime(iso_datetime_str)
         return dt.strftime("%Y-%m-%d %H:%M")
     except (TypeError, ValueError):
         return str(iso_datetime_str)

litellm/proxy/common_utils/reset_budget_job.py

@@ -6,6 +6,7 @@ from typing import Any, Callable, List, Literal, Optional, Union
 
 import litellm
 from litellm._logging import verbose_proxy_logger
+from litellm.litellm_core_utils.datetime_utils import parse_utc_datetime
 from litellm.proxy._types import (
     LiteLLM_BudgetTableFull,
     LiteLLM_EndUserTable,
@@ -662,8 +663,7 @@ class ResetBudgetJob:
         reset_at_str = window.get("reset_at")
         if not reset_at_str:
             return False
-        reset_at = datetime.fromisoformat(reset_at_str.replace("Z", "+00:00")).replace(tzinfo=None)
-        if reset_at > now:
+        if parse_utc_datetime(reset_at_str) > parse_utc_datetime(now):
             return False
         spend_counter_cache.in_memory_cache.set_cache(key=counter_key, value=0.0)
         if spend_counter_cache.redis_cache is not None:

litellm/proxy/db/db_transaction_queue/spend_logs_partition_manager.py

@@ -18,6 +18,7 @@ from datetime import date, datetime, timedelta, timezone
 from typing import List, Optional, Tuple
 
 from litellm._logging import verbose_proxy_logger
+from litellm.litellm_core_utils.datetime_utils import parse_utc_datetime
 from litellm.constants import (
     SPEND_LOG_PARTITION_INTERVAL,
     SPEND_LOG_PARTITION_PRECREATE_AHEAD,
@@ -86,7 +87,7 @@ def parse_partition_upper_bound(bound_expr: str) -> Optional[datetime]:
     if match is None:
         return None
     try:
-        return datetime.fromisoformat(match.group(1))
+        return parse_utc_datetime(match.group(1))
     except ValueError:
         return None
 
@@ -94,10 +95,12 @@ def parse_partition_upper_bound(bound_expr: str) -> Optional[datetime]:
 def select_partitions_to_drop(partitions: List[Tuple[str, Optional[datetime]]], cutoff: datetime) -> List[str]:
     """
     Names of partitions whose entire range is older than `cutoff` (upper bound
-    <= cutoff). `cutoff` and the bounds are UTC-naive. Partitions without a
-    parseable upper bound (e.g. DEFAULT) are kept.
+    <= cutoff). `cutoff` and the bounds are normalized to tz-aware UTC before
+    comparing (naive values are assumed UTC). Partitions without a parseable
+    upper bound (e.g. DEFAULT) are kept.
     """
-    return [name for name, upper in partitions if upper is not None and upper <= cutoff]
+    aware_cutoff = parse_utc_datetime(cutoff)
+    return [name for name, upper in partitions if upper is not None and upper <= aware_cutoff]
 
 
 class SpendLogsPartitionManager:
@@ -179,9 +182,8 @@ class SpendLogsPartitionManager:
 
     async def drop_partitions_older_than(self, prisma_client, cutoff: datetime) -> List[str]:
         """DROP every partition whose whole range is older than `cutoff`."""
-        cutoff_naive = cutoff.astimezone(timezone.utc).replace(tzinfo=None)
         partitions = await self._list_partitions(prisma_client)
-        to_drop = select_partitions_to_drop(partitions, cutoff_naive)
+        to_drop = select_partitions_to_drop(partitions, cutoff)
         dropped: List[str] = []
         for name in to_drop:
             try:

litellm/proxy/db/spend_log_tool_index.py

@@ -4,10 +4,10 @@ are written, so "last N requests for tool X" and "how is this tool called in pro
 queries are fast.
 """
 
-from datetime import datetime, timezone
 from typing import Any, Dict, List, Set
 
 from litellm._logging import verbose_proxy_logger
+from litellm.litellm_core_utils.datetime_utils import parse_utc_datetime
 from litellm.litellm_core_utils.safe_json_loads import safe_json_loads
 from litellm.proxy.utils import PrismaClient
 from litellm.repositories.table_repositories import SpendLogToolIndexRepository
@@ -92,13 +92,10 @@ async def process_spend_logs_tool_usage(
         start_time = payload.get("startTime")
         if not request_id or not start_time:
             continue
-        if isinstance(start_time, str):
-            try:
-                start_time = datetime.fromisoformat(start_time.replace("Z", "+00:00"))
-            except (ValueError, TypeError):
-                continue
-        if start_time.tzinfo is None:
-            start_time = start_time.replace(tzinfo=timezone.utc)
+        try:
+            start_time = parse_utc_datetime(start_time)
+        except (ValueError, TypeError):
+            continue
 
         tool_names = _parse_tool_names_from_payload(payload)
         for tool_name in tool_names:
@@ -114,23 +111,14 @@ async def process_spend_logs_tool_usage(
         return
 
     try:
-        index_data = []
-        for r in index_rows:
-            st = r["start_time"]
-            if isinstance(st, str):
-                try:
-                    st = datetime.fromisoformat(st.replace("Z", "+00:00"))
-                except (ValueError, TypeError):
-                    continue
-            if st.tzinfo is None:
-                st = st.replace(tzinfo=timezone.utc)
-            index_data.append(
-                {
-                    "request_id": r["request_id"],
-                    "tool_name": r["tool_name"],
-                    "start_time": st,
-                }
-            )
+        index_data = [
+            {
+                "request_id": r["request_id"],
+                "tool_name": r["tool_name"],
+                "start_time": parse_utc_datetime(r["start_time"]),
+            }
+            for r in index_rows
+        ]
         if index_data:
             await SpendLogToolIndexRepository(prisma_client).table.create_many(
                 data=index_data,

litellm/proxy/guardrails/usage_endpoints.py

@@ -10,6 +10,7 @@ from typing import Any, Dict, List, Optional
 from fastapi import APIRouter, Depends, Query
 from pydantic import BaseModel
 
+from litellm.litellm_core_utils.datetime_utils import parse_utc_datetime
 from litellm.proxy._types import UserAPIKeyAuth
 from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
 from litellm.repositories.table_repositories import (
@@ -425,15 +426,15 @@ def _build_usage_logs_where(
     if start_date or end_date:
         st_filter: Dict[str, Any] = {}
         if start_date:
-            sd = start_date.replace("Z", "+00:00").strip()
+            sd = start_date.strip()
             if "T" not in sd:
-                sd += "T00:00:00+00:00"
-            st_filter["gte"] = datetime.fromisoformat(sd)
+                sd += "T00:00:00"
+            st_filter["gte"] = parse_utc_datetime(sd)
         if end_date:
-            ed = end_date.replace("Z", "+00:00").strip()
+            ed = end_date.strip()
             if "T" not in ed:
-                ed += "T23:59:59+00:00"
-            st_filter["lte"] = datetime.fromisoformat(ed)
+                ed += "T23:59:59"
+            st_filter["lte"] = parse_utc_datetime(ed)
         where["start_time"] = st_filter
     return where
 

litellm/proxy/guardrails/usage_tracking.py

@@ -9,6 +9,7 @@ from datetime import datetime, timezone
 from typing import Any, Dict, List, Optional
 
 from litellm._logging import verbose_proxy_logger
+from litellm.litellm_core_utils.datetime_utils import parse_utc_datetime
 from litellm.proxy.utils import PrismaClient
 from litellm.repositories.table_repositories import (
     DailyGuardrailMetricsRepository,
@@ -48,9 +49,7 @@ def _parse_guardrail_info_from_payload(payload: Dict[str, Any]) -> List[Dict[str
 
 def _date_str(dt: datetime) -> str:
     """YYYY-MM-DD in UTC."""
-    if dt.tzinfo is None:
-        dt = dt.replace(tzinfo=timezone.utc)
-    return dt.astimezone(timezone.utc).strftime("%Y-%m-%d")
+    return parse_utc_datetime(dt).astimezone(timezone.utc).strftime("%Y-%m-%d")
 
 
 async def process_spend_logs_guardrail_usage(
@@ -81,7 +80,7 @@ async def process_spend_logs_guardrail_usage(
             continue
         if isinstance(start_time, str):
             try:
-                start_time = datetime.fromisoformat(start_time.replace("Z", "+00:00"))
+                start_time = parse_utc_datetime(start_time)
             except (ValueError, TypeError):
                 continue
         date_key = _date_str(start_time)
@@ -120,7 +119,7 @@ async def process_spend_logs_guardrail_usage(
                 st = r["start_time"]
                 if isinstance(st, str):
                     try:
-                        st = datetime.fromisoformat(st.replace("Z", "+00:00"))
+                        st = parse_utc_datetime(st)
                     except (ValueError, TypeError):
                         continue
                 index_data.append(

litellm/proxy/management_endpoints/mcp_management_endpoints.py

@@ -46,6 +46,7 @@ import litellm
 from litellm._logging import verbose_logger, verbose_proxy_logger
 from litellm._uuid import uuid
 from litellm.constants import LITELLM_PROXY_ADMIN_NAME
+from litellm.litellm_core_utils.datetime_utils import parse_utc_datetime
 from litellm.proxy._experimental.mcp_server.utils import (
     build_env_var_setup_url,
     collect_env_var_references,
@@ -2002,9 +2003,9 @@ if MCP_AVAILABLE:
         is_expired = False
         if expires_at:
             try:
-                exp = datetime.fromisoformat(expires_at)
+                exp = parse_utc_datetime(expires_at)
                 is_expired = exp < datetime.now(timezone.utc)
-            except Exception:
+            except ValueError:
                 pass
         return MCPOAuthUserCredentialStatus(
             server_id=server_id,

litellm/proxy/proxy_server.py

@@ -246,6 +246,7 @@ from litellm.litellm_core_utils.core_helpers import (
     get_litellm_metadata_from_kwargs,
 )
 from litellm.litellm_core_utils.credential_accessor import CredentialAccessor
+from litellm.litellm_core_utils.datetime_utils import parse_utc_datetime
 from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
 from litellm.litellm_core_utils.sensitive_data_masker import (
     SensitiveDataMasker,
@@ -6312,7 +6313,7 @@ class ProxyConfig:
             if interval_hours is None and force_reload is False:
                 return  # No interval configured, skip reload
 
-            current_time = datetime.utcnow()
+            current_time = datetime.now(timezone.utc)
 
             # Check if we need to reload based on interval or force reload
             should_reload = False
@@ -6325,7 +6326,7 @@ class ProxyConfig:
                 global last_model_cost_map_reload
                 if last_model_cost_map_reload is not None:
                     try:
-                        last_reload_time = datetime.fromisoformat(last_model_cost_map_reload)
+                        last_reload_time = parse_utc_datetime(last_model_cost_map_reload)
                         time_since_last_reload = current_time - last_reload_time
                         hours_since_last_reload = time_since_last_reload.total_seconds() / 3600
 
@@ -6412,7 +6413,7 @@ class ProxyConfig:
             if interval_hours is None and force_reload is False:
                 return  # No interval configured, skip reload
 
-            current_time = datetime.utcnow()
+            current_time = datetime.now(timezone.utc)
 
             # Check if we need to reload based on interval or force reload
             should_reload = False
@@ -6425,7 +6426,7 @@ class ProxyConfig:
                 global last_anthropic_beta_headers_reload
                 if last_anthropic_beta_headers_reload is not None:
                     try:
-                        last_reload_time = datetime.fromisoformat(last_anthropic_beta_headers_reload)
+                        last_reload_time = parse_utc_datetime(last_anthropic_beta_headers_reload)
                         time_since_last_reload = current_time - last_reload_time
                         hours_since_last_reload = time_since_last_reload.total_seconds() / 3600
 
@@ -11534,15 +11535,7 @@ def _normalize_datetime_for_sorting(dt: Any) -> Optional[datetime]:
 
     if isinstance(dt, str):
         try:
-            # Handle ISO format strings, including 'Z' suffix
-            dt_str = dt.replace("Z", "+00:00") if dt.endswith("Z") else dt
-            parsed_dt = datetime.fromisoformat(dt_str)
-            # Ensure it's UTC-aware
-            if parsed_dt.tzinfo is None:
-                parsed_dt = parsed_dt.replace(tzinfo=timezone.utc)
-            else:
-                parsed_dt = parsed_dt.astimezone(timezone.utc)
-            return parsed_dt
+            return parse_utc_datetime(dt).astimezone(timezone.utc)
         except (ValueError, AttributeError):
             return None
 
@@ -15528,7 +15521,7 @@ async def reload_model_cost_map(
 
         # Update pod's in-memory last reload time
         global last_model_cost_map_reload
-        current_time = datetime.utcnow()
+        current_time = datetime.now(timezone.utc)
         last_model_cost_map_reload = current_time.isoformat()
 
         # Set force reload flag in database for other pods, preserving existing interval_hours
@@ -15728,13 +15721,13 @@ async def get_model_cost_map_reload_status(
                 "next_run": None,
             }
 
-        current_time = datetime.utcnow()
+        current_time = datetime.now(timezone.utc)
         next_run = None
 
         # Use pod's in-memory last reload time
         if last_model_cost_map_reload is not None:
             try:
-                last_reload_time = datetime.fromisoformat(last_model_cost_map_reload)
+                last_reload_time = parse_utc_datetime(last_model_cost_map_reload)
                 time_since_last_reload = current_time - last_reload_time
                 hours_since_last_reload = time_since_last_reload.total_seconds() / 3600
 
@@ -15842,7 +15835,7 @@ async def reload_anthropic_beta_headers(
 
         # Update pod's in-memory last reload time
         global last_anthropic_beta_headers_reload
-        current_time = datetime.utcnow()
+        current_time = datetime.now(timezone.utc)
         last_anthropic_beta_headers_reload = current_time.isoformat()
 
         # Set force reload flag in database for other pods, preserving existing interval_hours
@@ -16049,13 +16042,13 @@ async def get_anthropic_beta_headers_reload_status(
                 "next_run": None,
             }
 
-        current_time = datetime.utcnow()
+        current_time = datetime.now(timezone.utc)
         next_run = None
 
         # Use pod's in-memory last reload time
         if last_anthropic_beta_headers_reload is not None:
             try:
-                last_reload_time = datetime.fromisoformat(last_anthropic_beta_headers_reload)
+                last_reload_time = parse_utc_datetime(last_anthropic_beta_headers_reload)
                 time_since_last_reload = current_time - last_reload_time
                 hours_since_last_reload = time_since_last_reload.total_seconds() / 3600
 

litellm/proxy/spend_tracking/budget_reservation.py

@@ -9,6 +9,7 @@ from typing import Any, Dict, List, Mapping, Optional, Sequence, cast
 import litellm
 from litellm._logging import verbose_proxy_logger
 from litellm.caching import DualCache
+from litellm.litellm_core_utils.datetime_utils import parse_utc_datetime
 from litellm.litellm_core_utils.duration_parser import duration_in_seconds
 from litellm.litellm_core_utils.llm_cost_calc.tiered_pricing import select_tier_for_input, tier_rate
 from litellm.proxy._types import (
@@ -862,8 +863,6 @@ def get_budget_window_start(window: Any) -> Optional[datetime]:
     reset_at = _coerce_datetime(window_dict.get("reset_at"))
     if reset_at is None:
         return datetime.now(timezone.utc) - timedelta(seconds=duration_seconds)
-    if reset_at.tzinfo is None:
-        reset_at = reset_at.replace(tzinfo=timezone.utc)
     return reset_at - timedelta(seconds=duration_seconds)
 
 
@@ -871,10 +870,10 @@ def _coerce_datetime(value: Any) -> Optional[datetime]:
     if value is None:
         return None
     if isinstance(value, datetime):
-        return value
+        return parse_utc_datetime(value)
     if isinstance(value, str):
         try:
-            return datetime.fromisoformat(value.replace("Z", "+00:00"))
+            return parse_utc_datetime(value)
         except ValueError:
             return None
     return None

litellm/proxy/spend_tracking/spend_management_endpoints.py

@@ -19,6 +19,7 @@ from fastapi import APIRouter, Depends, HTTPException, Request, status
 
 import litellm
 from litellm._logging import verbose_proxy_logger
+from litellm.litellm_core_utils.datetime_utils import parse_utc_datetime
 from litellm.proxy._types import *
 from litellm.proxy._types import ProviderBudgetResponse, ProviderBudgetResponseObject
 from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
@@ -370,7 +371,7 @@ async def get_global_activity(
         daily_data = []
         for row in db_response:
             # cast date to datetime
-            _date_obj = datetime.fromisoformat(row["date"])
+            _date_obj = parse_utc_datetime(row["date"])
             row["date"] = _date_obj.strftime("%b %d")
 
             daily_data.append(row)
@@ -537,7 +538,7 @@ async def get_global_activity_model(
                     "sum_api_requests": 0,
                     "sum_total_tokens": 0,
                 }
-            _date_obj = datetime.fromisoformat(row["date"])
+            _date_obj = parse_utc_datetime(row["date"])
             row["date"] = _date_obj.strftime("%b %d")
 
             model_ui_data[_model]["daily_data"].append(row)
@@ -684,7 +685,7 @@ async def get_global_activity_exceptions_per_deployment(
                     "daily_data": [],
                     "sum_num_rate_limit_exceptions": 0,
                 }
-            _date_obj = datetime.fromisoformat(row["date"])
+            _date_obj = parse_utc_datetime(row["date"])
             row["date"] = _date_obj.strftime("%b %d")
 
             model_ui_data[_model]["daily_data"].append(row)
@@ -803,7 +804,7 @@ async def get_global_activity_exceptions(
         daily_data = []
         for row in db_response:
             # cast date to datetime
-            _date_obj = datetime.fromisoformat(row["date"])
+            _date_obj = parse_utc_datetime(row["date"])
             row["date"] = _date_obj.strftime("%b %d")
 
             daily_data.append(row)

tests/test_litellm/litellm_core_utils/test_datetime_utils.py

@@ -0,0 +1,49 @@
+"""
+parse_utc_datetime is the single allowed ISO-8601 parse entrypoint under litellm/proxy/
+(enforced by semgrep). Naive input is defined as UTC; aware input keeps its offset. The
+result must always be comparable against datetime.now(timezone.utc) without TypeError.
+"""
+
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+from litellm.litellm_core_utils.datetime_utils import parse_utc_datetime
+
+
+def test_naive_string_is_assumed_utc():
+    result = parse_utc_datetime("2026-01-20T00:00:00")
+    assert result == datetime(2026, 1, 20, tzinfo=timezone.utc)
+
+
+def test_z_suffix_string_parses_aware():
+    result = parse_utc_datetime("2026-01-20T00:00:00Z")
+    assert result == datetime(2026, 1, 20, tzinfo=timezone.utc)
+
+
+def test_offset_string_keeps_offset():
+    result = parse_utc_datetime("2026-01-20T05:30:00+05:30")
+    assert result == datetime(2026, 1, 20, tzinfo=timezone.utc)
+    assert result.utcoffset() == timedelta(hours=5, minutes=30)
+
+
+def test_naive_datetime_passthrough_becomes_aware():
+    result = parse_utc_datetime(datetime(2026, 1, 20))
+    assert result == datetime(2026, 1, 20, tzinfo=timezone.utc)
+
+
+def test_aware_datetime_passthrough_unchanged():
+    aware = datetime(2026, 1, 20, tzinfo=timezone.utc)
+    assert parse_utc_datetime(aware) is aware
+
+
+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)
+
+
+def test_invalid_string_raises():
+    with pytest.raises(ValueError):
+        parse_utc_datetime("not-a-date")

tests/test_litellm/proxy/db/db_transaction_queue/test_spend_logs_partition_manager.py

@@ -53,7 +53,7 @@ def test_upcoming_partitions_count_and_contiguous_ranges():
 
 def test_parse_partition_upper_bound_extracts_to_value():
     bound = "FOR VALUES FROM ('2026-06-01 00:00:00') TO ('2026-06-02 00:00:00')"
-    assert parse_partition_upper_bound(bound) == datetime(2026, 6, 2, 0, 0, 0)
+    assert parse_partition_upper_bound(bound) == datetime(2026, 6, 2, 0, 0, 0, tzinfo=timezone.utc)
 
 
 def test_parse_partition_upper_bound_default_is_none():
@@ -62,11 +62,11 @@ def test_parse_partition_upper_bound_default_is_none():
 
 
 def test_select_partitions_to_drop_only_fully_expired():
-    cutoff = datetime(2026, 6, 10, 0, 0, 0)
+    cutoff = datetime(2026, 6, 10, 0, 0, 0, tzinfo=timezone.utc)
     partitions = [
-        ("p_old", datetime(2026, 6, 9, 0, 0, 0)),  # upper < cutoff -> drop
-        ("p_boundary", datetime(2026, 6, 10, 0, 0, 0)),  # upper == cutoff -> drop
-        ("p_partial", datetime(2026, 6, 11, 0, 0, 0)),  # straddles cutoff -> keep
+        ("p_old", datetime(2026, 6, 9, 0, 0, 0, tzinfo=timezone.utc)),  # upper < cutoff -> drop
+        ("p_boundary", datetime(2026, 6, 10, 0, 0, 0, tzinfo=timezone.utc)),  # upper == cutoff -> drop
+        ("p_partial", datetime(2026, 6, 11, 0, 0, 0, tzinfo=timezone.utc)),  # straddles cutoff -> keep
         ("p_default", None),  # DEFAULT -> keep
     ]
     assert select_partitions_to_drop(partitions, cutoff) == ["p_old", "p_boundary"]

tests/test_litellm/proxy/management_endpoints/test_mcp_management_endpoints.py

@@ -3525,6 +3525,46 @@ async def test_store_mcp_oauth_user_credential_returns_status():
     assert result.expires_at == "2099-01-01T00:00:00+00:00"
 
 
+@pytest.mark.asyncio
+async def test_get_mcp_oauth_user_credential_status_naive_past_expiry_is_expired():
+    """Regression: a tz-naive past expires_at used to raise TypeError against the
+    tz-aware now(), which the except swallowed, leaving is_expired False."""
+    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": "2020-01-01T00:00:00",
+        "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 True
+    assert result.expires_at == "2020-01-01T00:00:00"
+
+
 @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."""

tests/test_litellm/proxy/test_proxy_server.py

@@ -3453,8 +3453,7 @@ class TestPriceDataReloadAPI:
             ):
                 with patch("litellm.proxy.proxy_server.datetime") as mock_datetime:
                     # Mock current time to be 1 hour after last reload
-                    mock_datetime.utcnow.return_value = datetime(2024, 1, 1, 7, 0, 0)
-                    mock_datetime.fromisoformat = datetime.fromisoformat
+                    mock_datetime.now.return_value = datetime(2024, 1, 1, 7, 0, 0, tzinfo=timezone.utc)
 
                     response = client_with_auth.get(
                         "/schedule/model_cost_map_reload/status"
@@ -3465,7 +3464,7 @@ class TestPriceDataReloadAPI:
                     assert data["scheduled"] == True
                     assert data["interval_hours"] == 6
                     assert data["last_run"] == "2024-01-01T06:00:00"
-                    assert data["next_run"] == "2024-01-01T12:00:00"
+                    assert data["next_run"] == "2024-01-01T12:00:00+00:00"
 
     def test_get_model_cost_map_reload_status_non_admin_access(self, client_with_auth):
         """Test that non-admin users cannot get reload status"""