scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-28 06:23:07 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 94

Merge 00d5c1bb43 into b6f91a221b

Browse source
This commit is contained in:
devin-ai-integration[bot] 2026-10-21 14:29:30 +00:00 • committed by GitHub
commit ee77549f17
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
13 changed files with 778 additions and 86 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

21
.github/workflows/codeql.yml vendored
View file

@ -71,13 +71,32 @@ jobs:
# a lookup key into the haveibeenpwned range API (the protocol mandates
# SHA-1) and the digest itself never leaves the proxy beyond its first 5
# characters.
- name: Filter SARIF (OCI sha256, HIBP sha1)
# The same query fires on the legacy-row verifier in
# litellm/proxy/utils.py, where a stored 64-hex SHA256 row is matched by
# hashing the submitted password to compare. No new SHA256 rows are
# written (new passwords are PBKDF2-HMAC-SHA256); the call exists only so
# pre-migration rows can still sign in and be rehashed.
- name: Filter SARIF (OCI sha256, HIBP sha1, legacy password row verify)
if: matrix.language == 'python'
uses: advanced-security/filter-sarif@2da736ff05ef065cb2894ac6892e47b5eac2c3c0 # v1.1
with:
# Each suppression below is an adjudicated intentional use, not a
# lingering weakness:
# - litellm/llms/oci/common_utils.py: the OCI request-signing spec
# mandates a SHA-256 digest (FIPS-approved, content integrity —
# not password hashing).
# - litellm/proxy/auth/password_policy.py: the HIBP k-anonymity
# protocol requires SHA-1 prefixes for its range queries.
# - litellm/proxy/utils.py: read-only legacy verifier that must hash
# submitted passwords with the OLD algorithm to authenticate
# pre-migration rows (no new SHA-256 rows are written).
# These patterns suppress the rule for the whole file, so any FUTURE
# weak-hashing added to these three files is hidden too — new code
# here needs explicit adjudication before it may rely on this filter.
patterns: |
-litellm/llms/oci/common_utils.py:py/weak-sensitive-data-hashing
-litellm/proxy/auth/password_policy.py:py/weak-sensitive-data-hashing
-litellm/proxy/utils.py:py/weak-sensitive-data-hashing
input: sarif-results/python.sarif
output: sarif-results/python.sarif

24
litellm/proxy/auth/login_utils.py
View file

@ -5,6 +5,7 @@ This module contains the core login logic that can be reused across different
login endpoints (e.g., /login and /v2/login).
"""
import asyncio
import os
import secrets
from collections.abc import Mapping
@ -42,6 +43,7 @@ from litellm.proxy.utils import (
PrismaClient,
get_server_root_path,
hash_password,
needs_password_rehash,
verify_password,
)
from litellm.repositories.user_repository import UserRepository
@ -106,16 +108,22 @@ async def screen_login_password_for_breach(
async def _rehash_password_if_needed(user_id: str, password: str, stored: str) -> None:
"""Rehash legacy password (SHA256) to scrypt on successful login."""
if stored.startswith("scrypt:"):
"""Rehash legacy scrypt or SHA256 password rows to pbkdf2 on successful login."""
if not needs_password_rehash(stored):
return
from litellm.proxy.proxy_server import prisma_client
if prisma_client is not None:
await UserRepository(prisma_client).table.update(
where={"user_id": user_id},
data={"password": hash_password(password)},
)
try:
# PBKDF2 at 600k iterations is CPU-bound (~240ms); hash it off the
# event loop so concurrent requests do not stall behind a rehash.
rehashed: Final = await asyncio.to_thread(hash_password, password)
await UserRepository(prisma_client).table.update_many(
where={"user_id": user_id, "password": stored},
data={"password": rehashed},
)
except Exception as e: # noqa: BLE001 # a failed rehash must never surface into the login
verbose_proxy_logger.warning("Login-time password rehash could not update user %s: %s", user_id, e)
def get_ui_credentials(master_key: str | None) -> tuple[str, str]:
@ -411,7 +419,9 @@ async def _sign_in(
code=401,
)
if verify_password(password, _password):
# PBKDF2 at 600k iterations is CPU-bound (~240ms); run the KDF on a
# worker thread so the event loop keeps serving requests during login.
if await asyncio.to_thread(verify_password, password, _password):
await _rehash_password_if_needed(_user_row.user_id, password, _password)
breached_now: Final = prisma_client is not None and await screen_login_password_for_breach(
user_id=_user_row.user_id,

4
litellm/proxy/management_endpoints/internal_user_endpoints.py
View file

@ -185,7 +185,9 @@ async def _hash_password_in_dict(
if not password_prevalidated:
validate_password_policy(data["password"], general_settings)
await validate_password_not_breached(data["password"], general_settings, hibp_client)
data["password"] = hash_password(data["password"])
# PBKDF2 at 600k iterations is CPU-bound (~240ms); hash off the event
# loop so concurrent requests do not stall behind this write path.
data["password"] = await asyncio.to_thread(hash_password, data["password"])
data["password_reset_required"] = True
data["last_breach_check_at"] = None

8
litellm/proxy/management_endpoints/password_endpoints.py
View file

@ -8,6 +8,7 @@ request kwargs to OTEL spans, which would log plaintext passwords. The audit
signal is emitted by hand below, with field names only, never values.
"""
import asyncio
from typing import TYPE_CHECKING, Annotated, Final
from fastapi import APIRouter, Depends, HTTPException
@ -129,7 +130,9 @@ async def change_password(
),
)
if not verify_password(data.current_password, stored_password):
# PBKDF2 at 600k iterations is CPU-bound (~240ms); run the KDF on a worker
# thread so the event loop keeps serving requests during the checks.
if not await asyncio.to_thread(verify_password, data.current_password, stored_password):
raise HTTPException(status_code=400, detail=_error_detail("Current password is incorrect."))
if data.new_password == data.current_password:
@ -141,8 +144,9 @@ async def change_password(
validate_password_policy(data.new_password, general_settings)
await validate_password_not_breached(data.new_password, general_settings, hibp_client)
new_hash: Final = await asyncio.to_thread(hash_password, data.new_password)
password_update: Final[prisma_types.LiteLLM_UserTableUpdateInput] = {
"password": hash_password(data.new_password),
"password": new_hash,
"password_reset_required": False,
"last_breach_check_at": None,
}

7
litellm/proxy/proxy_server.py
View file

@ -947,7 +947,7 @@ from litellm.proxy.utils import ( # noqa: F401, RUF100 # legacy module exports
is_projected_spend_over_limit, # pyright: ignore[reportUnusedImport] # backwards-compatible package export
is_valid_team_configs,
litellm_config_cache,
migrate_passwords_to_scrypt_async,
migrate_plaintext_passwords_async,
model_dump_with_preserved_fields,
prefetch_config_params,
update_spend,
@ -1557,7 +1557,7 @@ async def proxy_startup_event(app: FastAPI) -> AsyncGenerator[ProxyLifespanState
async def _run_pw_migration():
try:
result: Final = await migrate_passwords_to_scrypt_async(prisma_client)
result: Final = await migrate_plaintext_passwords_async(prisma_client)
verbose_proxy_logger.info("Password migration: %s", result)
except Exception as e:
verbose_proxy_logger.warning("Password migration skipped: %s", e)
@ -18922,7 +18922,8 @@ async def claim_onboarding_link(data: InvitationClaim, request: Request):
validate_password_policy(data.password, general_settings)
await validate_password_not_breached(data.password, general_settings)
hashed_pw: Final = hash_password(data.password)
# PBKDF2 at 600k iterations is CPU-bound (~240ms); hash off the event loop.
hashed_pw: Final = await asyncio.to_thread(hash_password, data.password)
current_time = litellm.utils.get_utc_datetime()
tx_manager: Final[_InvitationAcceptTxManager] = prisma_client.db.tx()
async with tx_manager as tx:

98
litellm/proxy/utils.py
View file

@ -1,4 +1,6 @@
import asyncio
import base64
import binascii
import contextlib
import copy
import hashlib
@ -6,6 +8,7 @@ import inspect
import json
import math
import os
import secrets
import smtplib
import ssl
import sys
@ -82,6 +85,7 @@ from litellm.proxy._types import (
SpendLogsPayload,
)
from litellm.proxy.bug_report_config import build_proxy_bug_report
from litellm.proxy.common_utils.fips import is_fips_mode
from litellm.proxy.common_utils.openai_error_payload import (
attribute_of,
litellm_call_id_headers,
@ -7842,64 +7846,104 @@ def hash_token(token: str):
return hashed_token
def hash_password(password: str) -> str:
"""Hash a password using scrypt with a random salt."""
import base64
import hashlib
import os
# OWASP Password Storage Cheat Sheet recommendation
# for PBKDF2-HMAC-SHA256.
PBKDF2_ITERATIONS: Final = 600_000
# Defensive upper bound against excessive CPU usage from
# malformed or untrusted stored iteration counts.
PBKDF2_MAX_ITERATIONS: Final = 10_000_000
PBKDF2_PREFIX: Final = "pbkdf2:sha256:"
SCRYPT_PREFIX: Final = "scrypt:"
def hash_password(password: str) -> str:
"""Hash a password as ``pbkdf2:sha256:<iterations>:<salt b64>:<key b64>``.
Iteration count is the OWASP Password Storage Cheat Sheet floor for
PBKDF2-HMAC-SHA256; PBKDF2 is a FIPS-approved primitive so the same row
format is written on every image, FIPS mode or not.
"""
salt: Final = os.urandom(16)
dk: Final = hashlib.scrypt(password.encode(), salt=salt, n=16384, r=8, p=1, dklen=32)
return "scrypt:" + base64.b64encode(salt + dk).decode()
derived: Final = hashlib.pbkdf2_hmac("sha256", password.encode(), salt, PBKDF2_ITERATIONS, dklen=32)
return f"{PBKDF2_PREFIX}{PBKDF2_ITERATIONS}:{base64.b64encode(salt).decode()}:{base64.b64encode(derived).decode()}"
def _is_sha256_hex(value: str) -> bool:
return len(value) == 64 and all(c in "0123456789abcdef" for c in value)
def _is_hashed_password(value: str) -> bool:
return value.startswith((PBKDF2_PREFIX, SCRYPT_PREFIX)) or _is_sha256_hex(value)
def needs_password_rehash(stored: str) -> bool:
return not stored.startswith(PBKDF2_PREFIX)
def _verify_pbkdf2(password: str, stored: str) -> bool:
try:
scheme, digest, iterations, salt, derived = stored.split(":")
if (scheme, digest) != ("pbkdf2", "sha256"):
return False
count: Final = int(iterations)
if not 1 <= count <= PBKDF2_MAX_ITERATIONS:
return False
expected: Final = hashlib.pbkdf2_hmac("sha256", password.encode(), base64.b64decode(salt, validate=True), count)
return secrets.compare_digest(base64.b64decode(derived, validate=True), expected)
except (ValueError, binascii.Error, TypeError, OverflowError):
return False
def verify_password(password: str, stored: str) -> bool:
"""Verify a password against a stored hash. Supports scrypt and SHA256."""
import base64
import hashlib
import secrets
if stored.startswith("scrypt:"):
"""Verify a password against a stored hash. Supports pbkdf2, scrypt and SHA256 rows."""
if stored.startswith(PBKDF2_PREFIX):
return _verify_pbkdf2(password, stored)
if stored.startswith(SCRYPT_PREFIX):
if is_fips_mode():
verbose_proxy_logger.error(
"LITELLM_FIPS_MODE is on and this account still has a scrypt password hash, "
"which is not a FIPS approved primitive. An admin must set a new password with "
"POST /user/update so it is stored as pbkdf2 and the account can sign in again"
)
return False
try:
raw: Final = base64.b64decode(stored[7:])
salt, dk = raw[:16], raw[16:]
dk2: Final = hashlib.scrypt(password.encode(), salt=salt, n=16384, r=8, p=1, dklen=32)
return secrets.compare_digest(dk, dk2)
except Exception:
except (ValueError, binascii.Error, TypeError):
return False
# SHA256 fallback (not vulnerable to pass-the-hash: checks sha256(input) == stored)
if len(stored) == 64 and all(c in "0123456789abcdef" for c in stored):
if _is_sha256_hex(stored):
return secrets.compare_digest(hashlib.sha256(password.encode()).hexdigest().encode(), stored.encode())
return False
async def migrate_passwords_to_scrypt_async(prisma_client) -> str:
async def migrate_plaintext_passwords_async(prisma_client) -> str:
"""
Migrate plaintext passwords in the DB to scrypt. SHA256 passwords
are left alone (they migrate on next login via the SHA256 fallback).
Skips quickly if no plaintext passwords exist.
Migrate plaintext passwords in the DB to pbkdf2. Already-hashed rows
(pbkdf2, scrypt, sha256) are left alone; scrypt and sha256 rows rehash
on next successful login. Skips quickly if no plaintext passwords exist.
"""
all_with_pw: Final = await UserRepository(prisma_client).table.find_many(
where={"password": {"not": None}},
)
def _is_sha256_hex(s: str) -> bool:
return len(s) == 64 and all(c in "0123456789abcdef" for c in s)
plaintext_users: Final = [
(u.user_id, u.password)
for u in all_with_pw
if u.password and not u.password.startswith("scrypt:") and not _is_sha256_hex(u.password)
(u.user_id, u.password) for u in all_with_pw if u.password and not _is_hashed_password(u.password)
]
if not plaintext_users:
return "No plaintext passwords found"
for user_id, plaintext_password in plaintext_users:
# PBKDF2 at 600k iterations is CPU-bound (~240ms per row); hash off
# the event loop so startup does not stall other async work.
await UserRepository(prisma_client).table.update(
where={"user_id": user_id},
data={"password": hash_password(plaintext_password)},
data={"password": await asyncio.to_thread(hash_password, plaintext_password)},
)
return f"Migrated {len(plaintext_users)} plaintext passwords to scrypt"
return f"Migrated {len(plaintext_users)} plaintext passwords to pbkdf2"
def hash_token_if_needed(token: str) -> str:

284
tests/integration/authorization/test_password_hash_login.py Normal file
View file

@ -0,0 +1,284 @@
"""Stored passwords are PBKDF2-HMAC-SHA256 rows; scrypt and legacy SHA256 rows still sign in and get rehashed.
Every cell drives the real /login form and the management endpoints of a proxy this module owns (two workers,
login throttle off, breach screening off so no request leaves the box) and reads the stored row back from
Postgres. The expected hash is recomputed here with hashlib, never with litellm code.
"""
import base64
import hashlib
import os
import uuid
from collections.abc import Callable, Iterator
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from typing import Final
import httpx
import jwt
import psycopg
import pytest
import yaml
from tests.integration._support.client import Gateway, eventually, gateway_from_environment
from tests.integration._support.database import read_rows
from tests.integration._support.process import OwnedProxy, owned_proxy_process
PBKDF2_ITERATIONS: Final = 600_000
PASSWORD: Final = "Correct-Horse-9-Battery"
WRONG_PASSWORD: Final = "Wrong-Horse-9-Battery"
BURST_SIZE: Final = 30
@pytest.fixture(scope="module")
def password_proxy(tmp_path_factory: pytest.TempPathFactory) -> Iterator[OwnedProxy]:
directory: Final = tmp_path_factory.mktemp("password-proxy")
config: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text())
settings: Final = {**config["general_settings"], "password_policy_check_breached_passwords": False}
path: Final = directory / "config.yaml"
path.write_text(yaml.safe_dump({**config, "general_settings": settings}))
with gateway_from_environment() as suite_gateway:
with owned_proxy_process(
suite_gateway,
directory,
{"LITELLM_DISABLE_LOGIN_RATE_LIMIT": "true"},
config=path,
workers=2,
) as owned:
yield owned
@pytest.fixture
def proxy(password_proxy: OwnedProxy) -> Gateway:
return password_proxy.gateway
def _stored_password(user_id: str) -> str:
rows: Final = read_rows('SELECT password FROM "LiteLLM_UserTable" WHERE user_id = %s', (user_id,))
assert len(rows) == 1, rows
stored: Final = rows[0]["password"]
assert isinstance(stored, str), rows
return stored
def _write_stored_password(user_id: str, stored: str) -> None:
with psycopg.connect(os.environ["DATABASE_URL"]) as connection:
connection.execute('UPDATE "LiteLLM_UserTable" SET password = %s WHERE user_id = %s', (stored, user_id))
connection.commit()
def _scrypt_row(password: str) -> str:
salt: Final = os.urandom(16)
derived: Final = hashlib.scrypt(password.encode(), salt=salt, n=16384, r=8, p=1, dklen=32)
return "scrypt:" + base64.b64encode(salt + derived).decode()
def _sha256_row(password: str) -> str:
return hashlib.sha256(password.encode()).hexdigest()
def _pbkdf2_matches(stored: str, password: str) -> bool:
"""Independent check of a ``pbkdf2:sha256:<iterations>:<salt b64>:<key b64>`` row with the stdlib."""
scheme, digest, iterations, salt, derived = stored.split(":")
assert (scheme, digest, int(iterations)) == ("pbkdf2", "sha256", PBKDF2_ITERATIONS), stored
expected: Final = hashlib.pbkdf2_hmac("sha256", password.encode(), base64.b64decode(salt), int(iterations))
return base64.b64decode(derived) == expected
def _login(proxy: Gateway, email: str, password: str) -> int:
response: Final = proxy.client.post(
"/login", data={"username": email, "password": password}, follow_redirects=False
)
if response.status_code == 303:
assert response.headers["location"].endswith("/ui?login=success"), response.headers
assert "token=" in response.headers.get("set-cookie", ""), response.headers
return response.status_code
def _login_retrying_dropped_connections(proxy: Gateway, email: str, password: str) -> int:
try:
return _login(proxy, email, password)
except httpx.TransportError:
return _login(proxy, email, password)
def _user_with_password(proxy: Gateway, password: str | None) -> tuple[str, str]:
email: Final = f"integration-{uuid.uuid4().hex}@example.com"
created: Final = proxy.post(
"/user/new",
{"user_id": f"integration-{uuid.uuid4().hex}", "user_email": email, "auto_create_key": False},
)
user_id: Final = created["user_id"]
assert isinstance(user_id, str), created
if password is not None:
proxy.post("/user/update", {"user_id": user_id, "password": password})
return user_id, email
def _delete_user(proxy: Gateway, user_id: str) -> None:
response: Final = proxy.request("POST", "/user/delete", {"user_ids": [user_id]})
assert response.status_code == 200, response.text
def test_new_user_password_is_stored_as_pbkdf2_and_signs_in(proxy: Gateway) -> None:
user_id, email = _user_with_password(proxy, PASSWORD)
try:
stored: Final = _stored_password(user_id)
assert stored.startswith("pbkdf2:sha256:"), stored
assert _pbkdf2_matches(stored, PASSWORD), stored
assert _login(proxy, email, PASSWORD) == 303
assert _stored_password(user_id) == stored, "a pbkdf2 row must not be rewritten on login"
finally:
_delete_user(proxy, user_id)
def test_wrong_password_is_rejected_and_row_is_untouched(proxy: Gateway) -> None:
user_id, email = _user_with_password(proxy, PASSWORD)
try:
before: Final = _stored_password(user_id)
assert _login(proxy, email, WRONG_PASSWORD) == 401
assert _login(proxy, email, "") == 401
assert _stored_password(user_id) == before
finally:
_delete_user(proxy, user_id)
@pytest.mark.parametrize("legacy_row", (_scrypt_row, _sha256_row), ids=("scrypt", "sha256"))
def test_legacy_hash_signs_in_and_is_rehashed_to_pbkdf2(proxy: Gateway, legacy_row: Callable[[str], str]) -> None:
user_id, email = _user_with_password(proxy, None)
try:
legacy: Final = legacy_row(PASSWORD)
_write_stored_password(user_id, legacy)
assert _login(proxy, email, WRONG_PASSWORD) == 401
assert _stored_password(user_id) == legacy, "a rejected login must not rewrite the row"
assert _login(proxy, email, PASSWORD) == 303
rehashed: Final = eventually(lambda: _stored_password(user_id), lambda row: row.startswith("pbkdf2:"))
assert _pbkdf2_matches(rehashed, PASSWORD), rehashed
assert _login(proxy, email, PASSWORD) == 303
assert _login(proxy, email, WRONG_PASSWORD) == 401
finally:
_delete_user(proxy, user_id)
@pytest.mark.parametrize("garbage", ("", "plaintext-password", "pbkdf2:sha256:1:not-base64:!!", "scrypt:%%%"))
def test_unreadable_stored_row_rejects_every_password(proxy: Gateway, garbage: str) -> None:
user_id, email = _user_with_password(proxy, None)
try:
_write_stored_password(user_id, garbage)
assert _login(proxy, email, garbage) == 401
assert _login(proxy, email, PASSWORD) == 401
assert _stored_password(user_id) == garbage
finally:
_delete_user(proxy, user_id)
def test_changed_password_is_stored_as_pbkdf2(proxy: Gateway) -> None:
user_id, email = _user_with_password(proxy, PASSWORD)
try:
signed_in: Final = proxy.client.post(
"/login", data={"username": email, "password": PASSWORD}, follow_redirects=False
)
assert signed_in.status_code == 303, signed_in.text
session: Final = jwt.decode(signed_in.cookies["token"], options={"verify_signature": False})["key"]
assert isinstance(session, str) and session.startswith("sk-"), session
new_password: Final = "Fresh-Horse-7-Battery"
changed: Final = proxy.request(
"POST",
"/user/password/change",
{"current_password": PASSWORD, "new_password": new_password},
key=session,
)
assert changed.status_code == 200, changed.text
stored: Final = _stored_password(user_id)
assert stored.startswith("pbkdf2:sha256:"), stored
assert _pbkdf2_matches(stored, new_password), stored
assert _login(proxy, email, PASSWORD) == 401
assert _login(proxy, email, new_password) == 303
finally:
_delete_user(proxy, user_id)
def test_pbkdf2_row_with_a_higher_iteration_count_signs_in_and_is_kept(proxy: Gateway) -> None:
user_id, email = _user_with_password(proxy, None)
try:
salt: Final = os.urandom(16)
derived: Final = hashlib.pbkdf2_hmac("sha256", PASSWORD.encode(), salt, 700_000)
row: Final = f"pbkdf2:sha256:700000:{base64.b64encode(salt).decode()}:{base64.b64encode(derived).decode()}"
_write_stored_password(user_id, row)
assert _login(proxy, email, WRONG_PASSWORD) == 401
assert _login(proxy, email, PASSWORD) == 303
assert _stored_password(user_id) == row, "a 700k-iteration row must not be downgraded to 600k on login"
finally:
_delete_user(proxy, user_id)
def test_user_new_refuses_a_password(proxy: Gateway) -> None:
refused: Final = proxy.request(
"POST",
"/user/new",
{
"user_id": f"integration-{uuid.uuid4().hex}",
"user_email": f"integration-{uuid.uuid4().hex}@example.com",
"auto_create_key": False,
"password": PASSWORD,
},
)
assert refused.status_code == 422, refused.text
assert "password cannot be set via /user/new" in refused.text, refused.text
def test_unrelated_routes_keep_serving_during_a_login_burst(proxy: Gateway) -> None:
user_id, email = _user_with_password(proxy, None)
try:
_write_stored_password(user_id, _scrypt_row(PASSWORD))
with proxy.scenario() as scenario:
model: Final = scenario.model()
key: Final = scenario.key(models=[model])
attempts: Final = tuple(PASSWORD if index % 3 else WRONG_PASSWORD for index in range(BURST_SIZE))
with ThreadPoolExecutor(max_workers=BURST_SIZE + 2) as pool:
login_futures: Final = tuple(pool.submit(_login, proxy, email, password) for password in attempts)
key_future: Final = pool.submit(proxy.request, "POST", "/key/generate", {})
chat_future: Final = pool.submit(
proxy.request,
"POST",
"/v1/chat/completions",
{"model": model, "messages": [{"role": "user", "content": "login burst liveness"}]},
key=key,
)
statuses: Final = tuple(future.result() for future in login_futures)
generated: Final = key_future.result()
chat: Final = chat_future.result()
assert sorted(statuses) == sorted(303 if password == PASSWORD else 401 for password in attempts), statuses
assert generated.status_code == 200, generated.text
assert chat.status_code == 200, chat.text
rehashed: Final = eventually(lambda: _stored_password(user_id), lambda row: row.startswith("pbkdf2:"))
assert _pbkdf2_matches(rehashed, PASSWORD), rehashed
finally:
_delete_user(proxy, user_id)
def test_concurrent_logins_on_a_scrypt_row_rehash_once_while_one_worker_is_killed(
password_proxy: OwnedProxy,
) -> None:
proxy: Final = password_proxy.gateway
user_id, email = _user_with_password(proxy, None)
try:
_write_stored_password(user_id, _scrypt_row(PASSWORD))
workers: Final = password_proxy.worker_pids()
assert len(workers) >= 2, workers
victim: Final = next(iter(workers))
attempts: Final = tuple(PASSWORD if index % 3 else WRONG_PASSWORD for index in range(BURST_SIZE))
with ThreadPoolExecutor(max_workers=BURST_SIZE) as pool:
futures: Final = tuple(
pool.submit(_login_retrying_dropped_connections, proxy, email, password) for password in attempts
)
eventually(lambda: any(future.done() for future in futures), lambda done: done)
password_proxy.kill_worker(victim)
statuses: Final = tuple(future.result() for future in futures)
assert sorted(statuses) == sorted(303 if password == PASSWORD else 401 for password in attempts), statuses
rehashed: Final = eventually(lambda: _stored_password(user_id), lambda row: row.startswith("pbkdf2:"))
assert _pbkdf2_matches(rehashed, PASSWORD), rehashed
assert _login(proxy, email, PASSWORD) == 303
assert _stored_password(user_id) == rehashed, "a settled pbkdf2 row must stay stable across logins"
finally:
_delete_user(proxy, user_id)

162
tests/unit/proxy/auth/test_login_utils.py
View file

@ -597,6 +597,102 @@ async def test_authenticate_user_database_login_with_non_ascii_password():
assert result.user_email == user_email
@pytest.mark.asyncio
async def test_login_password_kdf_runs_off_event_loop_thread(monkeypatch):
"""PBKDF2 at 600k iterations is CPU-bound (~240ms per verify, plus a full
rehash on legacy rows). If it runs on the event loop thread, every concurrent
request on that worker stalls for the duration of a login. The async login
path must execute the KDF on a worker thread instead."""
import base64
import threading
from litellm.proxy.auth import login_utils as login_module
def _scrypt_row(password: str) -> str:
salt = b"\x01" * 16
derived = hashlib.scrypt(password.encode(), salt=salt, n=16384, r=8, p=1, dklen=32)
return "scrypt:" + base64.b64encode(salt + derived).decode()
user_email = "kdf-thread@example.com"
password = "correct-password"
# Legacy scrypt row: exercises both the verify and the transparent rehash path.
stored_password = _scrypt_row(password)
mock_user = MagicMock()
mock_user.user_id = "test-user-kdf"
mock_user.user_email = user_email
mock_user.password = stored_password
mock_user.user_role = LitellmUserRoles.INTERNAL_USER
def mock_find_first(**kwargs):
where = kwargs.get("where", {})
user_email_filter = where.get("user_email", {})
if str(user_email_filter.get("equals", "")).lower() == user_email.lower():
return mock_user
return None
mock_prisma_client = MagicMock()
mock_prisma_client.db.litellm_usertable.find_first = AsyncMock(side_effect=mock_find_first)
seen_threads: dict[str, int] = {}
real_verify = login_module.verify_password
real_hash = login_module.hash_password
def recording_verify(password: str, stored: str) -> bool:
seen_threads.setdefault("verify", threading.get_ident())
return real_verify(password, stored)
def recording_hash(password: str) -> str:
seen_threads.setdefault("hash", threading.get_ident())
return real_hash(password)
monkeypatch.setattr(login_module, "verify_password", recording_verify)
monkeypatch.setattr(login_module, "hash_password", recording_hash)
mock_user_repository = MagicMock()
mock_user_repository.table.find_first = AsyncMock(side_effect=mock_find_first)
mock_user_repository.table.update_many = AsyncMock(return_value=None)
monkeypatch.setattr(login_module, "UserRepository", lambda client: mock_user_repository)
with patch.dict(
os.environ,
{"DATABASE_URL": "postgresql://test:test@localhost/test"},
):
with (
patch( # test-quality-ok: _rehash_password_if_needed reads the proxy_server global
"litellm.proxy.proxy_server.prisma_client", mock_prisma_client
),
patch(
"litellm.proxy.auth.login_utils.generate_key_helper_fn",
new_callable=AsyncMock,
) as mock_generate_key,
):
mock_generate_key.return_value = {"token": "token-123"}
result = await authenticate_user(
username=user_email,
password=password,
master_key="sk-master-for-tests",
prisma_client=mock_prisma_client,
throttle=_unlimited_throttle(),
general_settings=_POLICY_NO_BREACH_CHECK,
)
assert isinstance(result, LoginResult)
assert result.user_id == "test-user-kdf"
# The KDF calls made during the async login must not run on the thread
# hosting the event loop.
loop_thread: Final = threading.get_ident()
assert seen_threads.get("verify") is not None, "login never called verify_password"
assert seen_threads["verify"] != loop_thread
assert seen_threads.get("hash") is not None, "login never rehashed the legacy row"
assert seen_threads["hash"] != loop_thread
# The rehash still wrote a pbkdf2 row (non-fatal path stays functional).
rehash_kwargs = mock_user_repository.table.update_many.call_args.kwargs
assert rehash_kwargs["where"] == {"user_id": "test-user-kdf", "password": stored_password}
assert rehash_kwargs["data"]["password"].startswith("pbkdf2:sha256:")
class TestEncodeUiSessionJwt:
"""The UI session cookie must carry a bounded exp so it does not stay
signature-valid until the master key rotates, and so the session-cookie readers
@ -2350,3 +2446,69 @@ class TestScreenLoginPasswordForBreach:
)
is True
)
class TestRehashPasswordIfNeeded:
"""A successful login rewrites legacy hash rows to pbkdf2 in place."""
@pytest.mark.asyncio
@pytest.mark.parametrize("legacy_kind", ("scrypt", "sha256"))
async def test_legacy_row_is_rewritten_to_a_verifying_pbkdf2_row(self, legacy_kind):
import base64
from litellm.proxy.auth.login_utils import _rehash_password_if_needed
from litellm.proxy.utils import verify_password
password = "rehash-me-1"
if legacy_kind == "scrypt":
salt = os.urandom(16)
derived = hashlib.scrypt(password.encode(), salt=salt, n=16384, r=8, p=1, dklen=32)
stored = "scrypt:" + base64.b64encode(salt + derived).decode()
else:
stored = hashlib.sha256(password.encode()).hexdigest()
mock_prisma_client = MagicMock()
mock_prisma_client.db.litellm_usertable.update_many = AsyncMock()
with patch( # test-quality-ok: the rehash writes to the database; faked so no DB is needed
"litellm.proxy.proxy_server.prisma_client", mock_prisma_client
):
await _rehash_password_if_needed("u-1", password, stored)
update_kwargs = mock_prisma_client.db.litellm_usertable.update_many.await_args.kwargs
assert update_kwargs["where"] == {"user_id": "u-1", "password": stored}
written = update_kwargs["data"]["password"]
assert written.startswith("pbkdf2:sha256:")
assert verify_password(password, written)
@pytest.mark.asyncio
async def test_pbkdf2_row_triggers_no_update(self):
from litellm.proxy.auth.login_utils import _rehash_password_if_needed
from litellm.proxy.utils import hash_password, verify_password
stored = hash_password("rehash-me-1")
mock_prisma_client = MagicMock()
mock_prisma_client.db.litellm_usertable.update_many = AsyncMock()
with patch( # test-quality-ok: the rehash writes to the database; faked so no DB is needed
"litellm.proxy.proxy_server.prisma_client", mock_prisma_client
):
await _rehash_password_if_needed("u-1", "rehash-me-1", stored)
assert verify_password("rehash-me-1", stored)
mock_prisma_client.db.litellm_usertable.update_many.assert_not_called()
@pytest.mark.asyncio
async def test_a_failed_rehash_logs_a_warning_and_never_raises(self, caplog):
import logging
from litellm._logging import verbose_proxy_logger
from litellm.proxy.auth.login_utils import _rehash_password_if_needed
mock_prisma_client = MagicMock()
mock_prisma_client.db.litellm_usertable.update_many = AsyncMock(side_effect=RuntimeError("db down"))
with patch( # test-quality-ok: the rehash writes to the database; faked so no DB is needed
"litellm.proxy.proxy_server.prisma_client", mock_prisma_client
):
with caplog.at_level(logging.WARNING, logger=verbose_proxy_logger.name):
await _rehash_password_if_needed("u-1", "rehash-me-1", "scrypt:stored")
assert "Login-time password rehash could not update user u-1" in caplog.text

121
tests/unit/proxy/auth/test_password_hashing.py
View file

@ -1,31 +1,41 @@
"""Tests for password hashing and verification utilities."""
import base64
import hashlib
import logging
import os
import pytest
from litellm.proxy.utils import hash_password, verify_password
from litellm._logging import verbose_proxy_logger
from litellm.proxy.utils import hash_password, needs_password_rehash, verify_password
def _scrypt_row(password: str) -> str:
salt = os.urandom(16)
derived = hashlib.scrypt(password.encode(), salt=salt, n=16384, r=8, p=1, dklen=32)
return "scrypt:" + base64.b64encode(salt + derived).decode()
def _pbkdf2_row(password: str, iterations: int) -> str:
salt = os.urandom(16)
derived = hashlib.pbkdf2_hmac("sha256", password.encode(), salt, iterations)
return f"pbkdf2:sha256:{iterations}:{base64.b64encode(salt).decode()}:{base64.b64encode(derived).decode()}"
class TestHashPassword:
def test_produces_scrypt_prefix(self):
assert hash_password("test").startswith("scrypt:")
def test_produces_pbkdf2_prefix_at_owasp_floor(self):
h = hash_password("test")
assert h.startswith("pbkdf2:sha256:600000:")
iterations = int(h.split(":")[2])
assert iterations >= 600_000
def test_unique_salt_per_call(self):
assert hash_password("same") != hash_password("same")
def test_output_length(self):
# "scrypt:" (7) + base64(48 bytes) (64) = 71
assert len(hash_password("test")) == 71
class TestVerifyPassword:
def test_correct_password(self):
def test_round_trip_and_wrong_password(self):
h = hash_password("correct")
assert verify_password("correct", h) is True
def test_wrong_password(self):
h = hash_password("correct")
assert verify_password("wrong", h) is False
def test_empty_password(self):
@ -44,9 +54,74 @@ class TestVerifyPassword:
assert verify_password(pw, h) is True
class TestVerifyPasswordFormats:
def test_row_with_higher_iteration_count_verifies(self):
stored = _pbkdf2_row("iterated", 700_000)
assert verify_password("iterated", stored) is True
assert verify_password("other", stored) is False
def test_scrypt_row_verifies_when_fips_off(self, monkeypatch):
monkeypatch.delenv("LITELLM_FIPS_MODE", raising=False)
stored = _scrypt_row("legacy-scrypt-pass")
assert verify_password("legacy-scrypt-pass", stored) is True
assert verify_password("wrong", stored) is False
def test_scrypt_row_rejected_and_logged_when_fips_on(self, monkeypatch, caplog):
monkeypatch.setenv("LITELLM_FIPS_MODE", "true")
stored = _scrypt_row("legacy-scrypt-pass")
with caplog.at_level(logging.ERROR, logger=verbose_proxy_logger.name):
assert verify_password("legacy-scrypt-pass", stored) is False
assert "scrypt" in caplog.text
assert "/user/update" in caplog.text
def test_sha256_fallback_verifies_in_both_modes(self, monkeypatch):
stored = hashlib.sha256(b"oldpass").hexdigest()
monkeypatch.delenv("LITELLM_FIPS_MODE", raising=False)
assert verify_password("oldpass", stored) is True
assert verify_password("wrong", stored) is False
monkeypatch.setenv("LITELLM_FIPS_MODE", "true")
assert verify_password("oldpass", stored) is True
@pytest.mark.parametrize(
"stored",
(
"pbkdf2:sha256:600000",
"pbkdf2:sha256:600000:c2FsdA==",
"pbkdf2:sha256:600000:c2FsdA==:a2V5:extra",
"pbkdf2:sha256:not-an-int:c2FsdA==:a2V5",
"pbkdf2:sha256:600000:not-base64-!!:a2V5",
"pbkdf2:sha256:600000:c2FsdA==:not-base64-!!",
"pbkdf2:md5:600000:c2FsdA==:a2V5",
),
ids=(
"missing_fields",
"four_fields",
"six_fields",
"non_int_iterations",
"bad_salt_base64",
"bad_key_base64",
"wrong_digest",
),
)
def test_malformed_pbkdf2_rows_return_false(self, stored):
assert verify_password("test", stored) is False
@pytest.mark.parametrize(
"iterations",
(10_000_001, 10**30, 0),
ids=("above_max", "overflows_c_long", "zero"),
)
def test_out_of_range_iteration_counts_return_false(self, iterations):
stored = _pbkdf2_row("test", 600_000).replace(":600000:", f":{iterations}:")
assert verify_password("test", stored) is False
def test_scrypt_invalid_base64_rejected(self):
assert verify_password("test", "scrypt:not-valid-base64!!!") is False
class TestVerifyPasswordFallbacks:
def test_sha256_fallback(self):
stored = hashlib.sha256("oldpass".encode()).hexdigest()
stored = hashlib.sha256(b"oldpass").hexdigest()
assert verify_password("oldpass", stored) is True
assert verify_password("wrong", stored) is False
@ -54,16 +129,18 @@ class TestVerifyPasswordFallbacks:
# Plaintext fallback removed to prevent pass-the-hash attacks
assert verify_password("plaintext", "plaintext") is False
def test_scrypt_preferred_over_fallbacks(self):
h = hash_password("test")
# Scrypt hash should not accidentally match as plaintext or SHA256
assert verify_password("test", h) is True
assert h.startswith("scrypt:")
def test_sha256_not_confused_with_plaintext(self):
# A 64-char hex string that isn't a valid SHA256 of the password
fake_hex = "a" * 64
assert verify_password("test", fake_hex) is False
def test_scrypt_invalid_base64_rejected(self):
assert verify_password("test", "scrypt:not-valid-base64!!!") is False
class TestNeedsPasswordRehash:
def test_legacy_rows_need_rehash(self):
assert needs_password_rehash(_scrypt_row("x")) is True
assert needs_password_rehash(hashlib.sha256(b"x").hexdigest()) is True
assert needs_password_rehash("plaintext") is True
def test_pbkdf2_rows_do_not_need_rehash(self):
assert needs_password_rehash(hash_password("x")) is False
assert needs_password_rehash(_pbkdf2_row("x", 700_000)) is False

30
tests/unit/proxy/management_endpoints/test_internal_user_endpoints.py
View file

@ -5211,3 +5211,33 @@ async def test_user_update_without_password_revokes_nothing(_admin_prisma, mocke
await _update_single_user_helper(user_request=user_request, user_api_key_dict=admin_caller)
revoke_mock.assert_not_awaited()
@pytest.mark.asyncio
async def test_hash_password_in_dict_runs_kdf_off_event_loop_thread(monkeypatch):
"""_hash_password_in_dict runs inside async request handlers (/user/update,
invitation onboarding); its PBKDF2 hash is CPU-bound and must not run on the
event loop thread."""
import threading
from litellm.proxy.management_endpoints import internal_user_endpoints as endpoints_module
from litellm.proxy.management_endpoints.internal_user_endpoints import _hash_password_in_dict
seen_thread: dict[str, int] = {}
real_hash = endpoints_module.hash_password
def recording_hash(password: str) -> str:
seen_thread.setdefault("hash", threading.get_ident())
return real_hash(password)
monkeypatch.setattr(endpoints_module, "hash_password", recording_hash)
data: dict = {"user_id": "u1", "password": "some-new-password"}
await _hash_password_in_dict(data, general_settings={}, password_prevalidated=True)
assert data["password"].startswith("pbkdf2:sha256:")
assert data["password_reset_required"] is True
assert data["last_breach_check_at"] is None
loop_thread: Final = threading.get_ident()
assert seen_thread.get("hash") is not None, "_hash_password_in_dict never hashed the password"
assert seen_thread["hash"] != loop_thread

61
tests/unit/proxy/management_endpoints/test_password_endpoints.py
View file

@ -5,6 +5,7 @@ HIBP is served by an AsyncHTTPHandler wrapping an httpx.MockTransport that is
injected straight into change_password; no test here touches the network.
"""
import base64
import hashlib
from typing import Final
from unittest.mock import AsyncMock, MagicMock, patch
@ -32,6 +33,12 @@ def _make_user_row(password: str | None) -> MagicMock:
return user
def _legacy_scrypt_row(password: str) -> str:
salt = b"\x01" * 16
derived = hashlib.scrypt(password.encode(), salt=salt, n=16384, r=8, p=1, dklen=32)
return "scrypt:" + base64.b64encode(salt + derived).decode()
def _make_prisma(user: MagicMock | None) -> MagicMock:
prisma = MagicMock()
prisma.db.litellm_usertable.find_first = AsyncMock(return_value=user)
@ -75,7 +82,7 @@ def _hibp_client_recording(calls: list[httpx.Request]) -> AsyncHTTPHandler:
@pytest.mark.asyncio
async def test_change_password_success_writes_new_scrypt_hash():
async def test_change_password_success_writes_new_pbkdf2_hash():
from litellm.proxy._types import ChangePasswordRequest
prisma = _make_prisma(_make_user_row(hash_password(CURRENT_PASSWORD)))
@ -99,6 +106,7 @@ async def test_change_password_success_writes_new_scrypt_hash():
assert update_kwargs["where"] == {"user_id": "user-123"}
stored = update_kwargs["data"]["password"]
assert stored != NEW_PASSWORD
assert stored.startswith("pbkdf2:sha256:")
assert verify_password(NEW_PASSWORD, stored)
# A successful change lifts any pending forced reset and re-arms the
# login-time breach screen for the new password.
@ -132,6 +140,57 @@ async def test_change_password_rejects_wrong_current_password():
prisma.db.litellm_usertable.update.assert_not_called()
@pytest.mark.asyncio
async def test_change_password_kdf_runs_off_event_loop_thread(monkeypatch):
"""The change-password endpoint's verify/hash are CPU-bound PBKDF2 calls
(~240ms each); they must run on a worker thread, not the event loop thread
that serves the rest of the proxy."""
import threading
from litellm.proxy._types import ChangePasswordRequest
from litellm.proxy.management_endpoints import password_endpoints as endpoints_module
seen_threads: dict[str, int] = {}
real_verify = endpoints_module.verify_password
real_hash = endpoints_module.hash_password
def recording_verify(password: str, stored: str) -> bool:
seen_threads.setdefault("verify", threading.get_ident())
return real_verify(password, stored)
def recording_hash(password: str) -> str:
seen_threads.setdefault("hash", threading.get_ident())
return real_hash(password)
monkeypatch.setattr(endpoints_module, "verify_password", recording_verify)
monkeypatch.setattr(endpoints_module, "hash_password", recording_hash)
prisma = _make_prisma(_make_user_row(_legacy_scrypt_row(CURRENT_PASSWORD)))
with (
patch( # test-quality-ok: change_password reads proxy_server module globals; no injection seam
"litellm.proxy.proxy_server.prisma_client", prisma
),
patch( # test-quality-ok: change_password reads proxy_server module globals; no injection seam
"litellm.proxy.proxy_server.general_settings", _POLICY_NO_BREACH_CHECK
),
):
response = await change_password(
data=ChangePasswordRequest(current_password=CURRENT_PASSWORD, new_password=NEW_PASSWORD),
user_api_key_dict=_caller(),
hibp_client=_hibp_client_never_called(),
)
assert response.user_id == "user-123"
loop_thread: Final = threading.get_ident()
assert seen_threads.get("verify") is not None, "change_password never called verify_password"
assert seen_threads["verify"] != loop_thread
assert seen_threads.get("hash") is not None, "change_password never hashed the new password"
assert seen_threads["hash"] != loop_thread
stored = prisma.db.litellm_usertable.update.call_args.kwargs["data"]["password"]
assert verify_password(NEW_PASSWORD, stored)
@pytest.mark.asyncio
async def test_change_password_rejects_unchanged_password():
from litellm.proxy._types import ChangePasswordRequest

39
tests/unit/proxy/utils/prisma_and_spend/test_password_helpers.py
View file

@ -4,18 +4,17 @@ Symbols pinned here:
- ``hash_token``
- ``hash_password``
- ``verify_password``
- ``migrate_passwords_to_scrypt_async``
- ``migrate_plaintext_passwords_async``
- ``_hash_token_if_needed``
- ``PrismaClient._is_sha256_hex`` (a nested helper inside
``migrate_passwords_to_scrypt_async``; the pin list labels it under the
PrismaClient health cluster as a documentation artifact)
- ``_is_sha256_hex`` (used by ``migrate_plaintext_passwords_async``; the
pin list labels it under the PrismaClient health cluster as a
documentation artifact)
"""
from __future__ import annotations
import hashlib
from types import SimpleNamespace
from typing import List
from unittest.mock import AsyncMock, MagicMock
import pytest
@ -24,7 +23,7 @@ from litellm.proxy.utils import (
hash_token_if_needed,
hash_password,
hash_token,
migrate_passwords_to_scrypt_async,
migrate_plaintext_passwords_async,
verify_password,
)
@ -57,16 +56,16 @@ def test_hash_token_raises_for_non_string() -> None:
hash_token(None) # type: ignore[arg-type]
def test_hash_password_uses_scrypt_prefix() -> None:
def test_hash_password_uses_pbkdf2_prefix() -> None:
h = hash_password("hunter2")
fields = {
"prefix": h[:7],
"prefix": h[:14],
"min_length": len(h) > 60,
"verifies_self": verify_password("hunter2", h),
"rejects_other": verify_password("hunter3", h),
}
assert fields == {
"prefix": "scrypt:",
"prefix": "pbkdf2:sha256:",
"min_length": True,
"verifies_self": True,
"rejects_other": False,
@ -133,9 +132,9 @@ def test_hash_token_if_needed_error_on_non_string() -> None:
# ---------------------------------------------------------------------------
# migrate_passwords_to_scrypt_async — pins behavior of the nested
# ``_is_sha256_hex`` helper too: scrypt-prefixed and sha256-hex rows are
# left alone, plaintext rows are upgraded in place.
# migrate_plaintext_passwords_async — pins behavior of the
# ``_is_hashed_password`` helper too: pbkdf2-prefixed, scrypt-prefixed and
# sha256-hex rows are left alone, plaintext rows are upgraded in place.
# ---------------------------------------------------------------------------
@ -152,11 +151,12 @@ async def test_migrate_passwords_skips_when_no_plaintext() -> None:
return_value=[
_make_user("a", "scrypt:abc"),
_make_user("b", sha),
_make_user("c", "pbkdf2:sha256:600000:c2FsdA==:a2V5"),
]
)
pc.db.litellm_usertable.update = AsyncMock()
result = await migrate_passwords_to_scrypt_async(pc)
result = await migrate_plaintext_passwords_async(pc)
outcome = {
"message": result,
"updates": pc.db.litellm_usertable.update.await_count,
@ -175,10 +175,11 @@ async def test_migrate_passwords_skips_when_no_plaintext() -> None:
async def test_migrate_passwords_upgrades_only_plaintext_rows() -> None:
pc = MagicMock()
pc.db = MagicMock()
users: List[SimpleNamespace] = [
users: list[SimpleNamespace] = [
_make_user("plaintext-user-1", "plain-1"),
_make_user("plaintext-user-2", "plain-2"),
_make_user("scrypt-user", "scrypt:already"),
_make_user("pbkdf2-user", "pbkdf2:sha256:600000:c2FsdA==:a2V5"),
_make_user(
"sha-user",
hashlib.sha256(b"alreadyhashed").hexdigest(),
@ -188,7 +189,7 @@ async def test_migrate_passwords_upgrades_only_plaintext_rows() -> None:
pc.db.litellm_usertable.find_many = AsyncMock(return_value=users)
pc.db.litellm_usertable.update = AsyncMock()
result = await migrate_passwords_to_scrypt_async(pc)
result = await migrate_plaintext_passwords_async(pc)
updated_user_ids = sorted(
call.kwargs["where"]["user_id"] for call in pc.db.litellm_usertable.update.await_args_list
@ -200,13 +201,13 @@ async def test_migrate_passwords_upgrades_only_plaintext_rows() -> None:
"message": result,
"update_count": pc.db.litellm_usertable.update.await_count,
"updated_ids": updated_user_ids,
"all_scrypt_prefixed": new_password_prefixes,
"all_pbkdf2_prefixed": new_password_prefixes,
}
assert outcome == {
"message": "Migrated 2 plaintext passwords to scrypt",
"message": "Migrated 2 plaintext passwords to pbkdf2",
"update_count": 2,
"updated_ids": ["plaintext-user-1", "plaintext-user-2"],
"all_scrypt_prefixed": ["scrypt:", "scrypt:"],
"all_pbkdf2_prefixed": ["pbkdf2:", "pbkdf2:"],
}
@ -216,4 +217,4 @@ async def test_migrate_passwords_raises_on_db_failure() -> None:
pc.db = MagicMock()
pc.db.litellm_usertable.find_many = AsyncMock(side_effect=RuntimeError("db unavailable"))
with pytest.raises(RuntimeError, match="db unavailable"):
await migrate_passwords_to_scrypt_async(pc)
await migrate_plaintext_passwords_async(pc)

5
tests/unit/proxy/utils/prisma_and_spend/test_prisma_client_health.py
View file

@ -10,9 +10,8 @@ Symbols pinned here:
- ``PrismaClient.get_health_check_history``
- ``PrismaClient.get_all_latest_health_checks``
- ``PrismaClient.get_latest_health_checks_for_models``
- ``PrismaClient._is_sha256_hex`` (a nested helper inside
``migrate_passwords_to_scrypt_async``; the pin list assigns it to this
cluster as a documentation artifact)
- ``_is_sha256_hex`` (used by ``migrate_plaintext_passwords_async``; the
pin list assigns it to this cluster as a documentation artifact)
"""
from __future__ import annotations