feat(proxy): automatic rotation of user credentials

litellm-proxy-extras/litellm_proxy_extras/schema.prisma

@@ -239,6 +239,7 @@ model LiteLLM_UserTable {
     organization_id String?
     object_permission_id String?
     password  String?
+    password_expiry DateTime?
     teams    String[] @default([])
     user_role  String?
 		max_budget Float?

litellm/constants.py

@@ -1382,6 +1382,24 @@
 # ``ProxyLogging._handle_logging_proxy_only_error``.
 LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL = "litellm_no_upstream_llm_call"
 
+# User credentials Rotation Constants
+LITELLM_USER_CREDENTIALS_ROTATION_ENABLED = os.getenv(
+    "LITELLM_USER_CREDENTIALS_ROTATION_ENABLED", "false"
+)
+LITELLM_USER_CREDENTIALS_ROTATION_CHECK_INTERVAL_SECONDS = int(
+    os.getenv("LITELLM_USER_CREDENTIALS_ROTATION_CHECK_INTERVAL_SECONDS", 86400)
+)  # 24 hours default
+LITELLM_USER_CREDENTIALS_ROTATION_INTERVAL = os.getenv(
+    "LITELLM_USER_CREDENTIALS_ROTATION_INTERVAL", ""
+)
+LITELLM_USER_CREDENTIALS_ROTATION_REMINDER_DAYS = int(
+    os.getenv("LITELLM_USER_CREDENTIALS_ROTATION_REMINDER_DAYS", 10)
+)
+LITELLM_USER_CREDENTIALS_ROTATION_LOCK_TTL_SECONDS = int(
+    os.getenv("LITELLM_USER_CREDENTIALS_ROTATION_LOCK_TTL_SECONDS", 600)
+)  # 10 minutes default — caps the deadlock window if a pod crashes mid-rotation
+
+
 # Key Rotation Constants
 LITELLM_KEY_ROTATION_ENABLED = os.getenv("LITELLM_KEY_ROTATION_ENABLED", "false")
 LITELLM_KEY_ROTATION_CHECK_INTERVAL_SECONDS = int(
@@ -1431,6 +1449,7 @@
 MAVVRIK_FOCUS_EXPORT_JOB_NAME = "mavvrik_focus_export_usage_data"
 CLOUDZERO_MAX_FETCHED_DATA_RECORDS = int(os.getenv("CLOUDZERO_MAX_FETCHED_DATA_RECORDS", 50000))
 SPEND_LOG_CLEANUP_JOB_NAME = "spend_log_cleanup"
+USER_CREDENTIALS_ROTATION_JOB_NAME = "litellm_user_credentials_rotation_job"
 KEY_ROTATION_JOB_NAME = "litellm_key_rotation_job"
 EXPIRED_UI_SESSION_KEY_CLEANUP_JOB_NAME = "litellm_expired_ui_session_key_cleanup_job"
 SPEND_LOG_RUN_LOOPS = int(os.getenv("SPEND_LOG_RUN_LOOPS", 500))

litellm/models/user.py

@@ -6,7 +6,7 @@
 """
 
 from datetime import datetime
-from typing import Dict, List, Optional
+from typing import Any, Dict, List, Optional
 
 from pydantic import ConfigDict, Field, model_validator
 
@@ -25,6 +25,7 @@ class LiteLLM_UserTable(LiteLLMPydanticObjectBase):
     organization_id: Optional[str] = None
     object_permission_id: Optional[str] = None
     password: Optional[str] = Field(default=None, exclude=True)
+    password_expiry: Optional[datetime] = Field(default=None)
     teams: List[str] = []
     user_role: Optional[str] = None
     max_budget: Optional[float] = None
@@ -50,7 +51,7 @@ class LiteLLM_UserTable(LiteLLMPydanticObjectBase):
 
     @model_validator(mode="before")
     @classmethod
-    def set_model_info(cls, values):
+    def set_model_info(cls, values: Dict[str, Any]) -> Dict[str, Any]:
         if values.get("spend") is None:
             values.update({"spend": 0.0})
         if values.get("models") is None:

litellm/proxy/_types.py

@@ -1516,6 +1516,7 @@ class NewUserRequestTeam(LiteLLMPydanticObjectBase):
 
 class NewUserRequest(GenerateRequestBase):
     max_budget: Optional[float] = None
+    password_expiry: Optional[datetime] = None
     user_email: Optional[str] = None
     user_alias: Optional[str] = None
     user_role: Optional[
@@ -1553,6 +1554,7 @@ class NewUserResponse(GenerateKeyResponse):
 
 class UpdateUserRequestNoUserIDorEmail(GenerateRequestBase):  # shared with BulkUpdateUserRequest
     password: Optional[str] = None
+    password_expiry: Optional[datetime] = None
     spend: Optional[float] = None
     metadata: Optional[dict] = None
     user_alias: Optional[str] = None
@@ -2233,6 +2235,14 @@ class ConfigGeneralSettings(LiteLLMPydanticObjectBase):
         description="sends alerts if requests hang for 5min+",
     )
     ui_access_mode: Optional[Literal["admin_only", "all"]] = Field("all", description="Control access to the Proxy UI")
+    user_credentials_rotation_interval: Optional[str] = Field(
+        None,
+        description="How often non-SSO username/password users must rotate credentials, e.g. '30d'",
+    )
+    user_credentials_rotation_reminder_days: Optional[int] = Field(
+        None,
+        description="How many days before password_expiry to send daily reminder emails for non-SSO users",
+    )
     allowed_routes: Optional[List] = Field(None, description="Proxy API Endpoints you want users to be able to access")
     reject_clientside_metadata_tags: Optional[bool] = Field(
         None,
@@ -2749,7 +2759,6 @@ class LiteLLM_UserTableFiltered(BaseModel):  # done to avoid exposing sensitive
     user_id: str
     user_email: Optional[str] = None
 
-
 class LiteLLM_UserTableWithKeyCount(LiteLLM_UserTable):
     key_count: int = 0
 

litellm/proxy/auth/login_utils.py

@@ -7,6 +7,7 @@
 
 import os
 import secrets
+from datetime import datetime, timezone
 from typing import Literal, Optional, cast
 
 from fastapi import HTTPException
@@ -103,6 +104,26 @@ def __init__(
         self.login_method = login_method
 
 
+def _get_password_expiry(user_row: LiteLLM_UserTable) -> Optional[datetime]:
+    password_expiry = getattr(user_row, "password_expiry", None)
+    if isinstance(password_expiry, datetime):
+        return password_expiry if password_expiry.tzinfo is not None else password_expiry.replace(tzinfo=timezone.utc)
+    return None
+
+
+def _raise_if_password_expired(user_row: LiteLLM_UserTable) -> None:
+    password_expiry = _get_password_expiry(user_row)
+    if password_expiry is None:
+        return
+    if datetime.now(timezone.utc) >= password_expiry:
+        raise ProxyException(
+            message="User credentials have expired. Please rotate your password before logging in again.",
+            type=ProxyErrorTypes.auth_error,
+            param="password_expiry",
+            code=401,
+        )
+
+
 async def authenticate_user(
     username: str,
     password: str,
@@ -264,6 +285,7 @@ async def authenticate_user(
             )
 
         if verify_password(password, _password):
+            _raise_if_password_expired(_user_row)
             await _rehash_password_if_needed(_user_row.user_id, password, _password)
             if os.getenv("DATABASE_URL") is not None:
                 response = await generate_key_helper_fn(

litellm/proxy/common_utils/user_credentials_rotation_manager.py

@@ -0,0 +1,114 @@
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+from litellm._logging import verbose_proxy_logger
+from litellm.constants import LITELLM_USER_CREDENTIALS_ROTATION_LOCK_TTL_SECONDS
+from litellm.models.user import LiteLLM_UserTable
+from litellm.proxy.utils import PrismaClient, send_email
+from litellm.repositories.user_repository import UserRepository
+
+
+UserCredentialsReminderRow = LiteLLM_UserTable
+
+
+class UserCredentialsRotationManager:
+    def __init__(
+        self,
+        prisma_client: PrismaClient,
+        pod_lock_manager: Any = None,
+        reminder_days: int = 10,
+    ):
+        self.prisma_client = prisma_client
+        self.pod_lock_manager = pod_lock_manager
+        self.reminder_days = max(reminder_days, 0)
+
+    async def process_rotations(self) -> None:
+        from litellm.constants import USER_CREDENTIALS_ROTATION_JOB_NAME
+
+        lock_acquired = False
+        try:
+            if self.pod_lock_manager and self.pod_lock_manager.redis_cache:
+                lock_ttl = max(LITELLM_USER_CREDENTIALS_ROTATION_LOCK_TTL_SECONDS, 300)
+                lock_acquired = (
+                    await self.pod_lock_manager.acquire_lock(
+                        cronjob_id=USER_CREDENTIALS_ROTATION_JOB_NAME,
+                        ttl=lock_ttl,
+                    )
+                    or False
+                )
+                if not lock_acquired:
+                    verbose_proxy_logger.warning(
+                        "User credentials rotation: another pod is already running reminders or Redis lock acquisition failed; skipping this cycle"
+                    )
+                    return
+
+            users_to_notify = await self._find_users_needing_reminders()
+            if not users_to_notify:
+                verbose_proxy_logger.debug(
+                    "No user credential rotation reminders are due at this time"
+                )
+                return
+
+            for user in users_to_notify:
+                try:
+                    await self._notify_user(user=user)
+                    verbose_proxy_logger.info(
+                        "Sent user credential rotation reminder to user_id=%s",
+                        user.user_id,
+                    )
+                except Exception as e:
+                    verbose_proxy_logger.error(
+                        "Failed to send user credential rotation reminder to user_id=%s: %s",
+                        user.user_id,
+                        e,
+                    )
+        except Exception as e:
+            verbose_proxy_logger.error(
+                "User credential rotation reminder process failed: %s",
+                e,
+            )
+        finally:
+            if (
+                lock_acquired
+                and self.pod_lock_manager
+                and self.pod_lock_manager.redis_cache
+            ):
+                await self.pod_lock_manager.release_lock(
+                    cronjob_id=USER_CREDENTIALS_ROTATION_JOB_NAME,
+                )
+
+    async def _find_users_needing_reminders(self) -> list[UserCredentialsReminderRow]:
+        now = datetime.now(timezone.utc)
+        reminder_threshold = now + timedelta(days=self.reminder_days)
+        return await UserRepository(self.prisma_client).table.find_many(
+            where={
+                "password": {"not": None},
+                "password_expiry": {
+                    "gte": now,
+                    "lte": reminder_threshold,
+                },
+                "user_email": {"not": None},
+                "sso_user_id": None,
+            }
+        )
+
+    async def _notify_user(self, user: UserCredentialsReminderRow) -> None:
+        user_email = getattr(user, "user_email", None)
+        password_expiry = getattr(user, "password_expiry", None)
+        if not isinstance(user_email, str) or not user_email.strip() or password_expiry is None:
+            return
+
+        subject = "LiteLLM password rotation reminder"
+        html = (
+            "<p>Your LiteLLM password will expire soon.</p>"
+            f"<p>Password expiry: <b>{password_expiry}</b></p>"
+            "<p>Please rotate your password before the expiry date to keep access.</p>"
+        )
+        await send_email(
+            receiver_email=user_email,
+            subject=subject,
+            html=html,
+        )
+
+
+__all__ = ["UserCredentialsRotationManager"]

litellm/proxy/example_config_yaml/bad_schema.prisma

@@ -104,6 +104,7 @@ model LiteLLM_UserTable {
     team_id    String?
     organization_id String?
     password  String?
+    password_expiry DateTime?
     teams    String[] @default([])
     user_role  String?
 		max_budget Float?

litellm/proxy/management_endpoints/internal_user_endpoints.py

@@ -15,9 +15,11 @@
 import asyncio
 import json
 import traceback
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from typing import Any, Dict, List, Optional, Union, cast
 
+from litellm.litellm_core_utils.duration_parser import duration_in_seconds
+
 import fastapi
 from fastapi import APIRouter, Depends, Header, HTTPException, Request, status
 
@@ -76,10 +78,35 @@
 router = APIRouter()
 
 
+def _get_user_credentials_rotation_interval() -> Optional[str]:
+    from litellm.constants import LITELLM_USER_CREDENTIALS_ROTATION_INTERVAL
+    from litellm.proxy.proxy_server import general_settings
+
+    configured_interval = general_settings.get(
+        "user_credentials_rotation_interval",
+        LITELLM_USER_CREDENTIALS_ROTATION_INTERVAL,
+    )
+    if not isinstance(configured_interval, str):
+        return None
+    stripped_interval = configured_interval.strip()
+    return stripped_interval or None
+
+
+def _calculate_password_expiry(rotation_interval: str) -> datetime:
+    return datetime.now(timezone.utc) + timedelta(
+        seconds=duration_in_seconds(rotation_interval)
+    )
+
+
 def _hash_password_in_dict(data: dict) -> None:
     """Hash password field in-place if present."""
-    if "password" in data and data["password"] is not None:
-        data["password"] = hash_password(data["password"])
+    password = data.get("password")
+    if password is None:
+        return
+    data["password"] = hash_password(password)
+    rotation_interval = _get_user_credentials_rotation_interval()
+    if rotation_interval is not None:
+        data["password_expiry"] = _calculate_password_expiry(rotation_interval)
 
 
 def _strip_password_from_response(response) -> None: