mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-11 22:51:28 +00:00
hibp support in password policy
This commit is contained in:
parent
e5da59336d
commit
6f8104606b
10 changed files with 409 additions and 9 deletions
|
|
@ -1791,6 +1791,17 @@ class NewUserRequest(GenerateRequestBase):
|
|||
send_invite_email: bool | None = None
|
||||
sso_user_id: str | None = None
|
||||
organizations: list[str] | None = None
|
||||
password: str | None = None
|
||||
|
||||
@field_validator("password")
|
||||
@classmethod
|
||||
def password_not_supported(cls, value: str | None) -> str | None:
|
||||
if value is not None:
|
||||
raise ValueError(
|
||||
"password cannot be set via /user/new. Users set their own password through an "
|
||||
"invitation link (POST /invitation/new)."
|
||||
)
|
||||
return value
|
||||
|
||||
|
||||
class NewUserResponse(GenerateKeyResponse):
|
||||
|
|
|
|||
|
|
@ -4,13 +4,26 @@ Applied at every path that persists a new or changed password for a DB-backed
|
|||
user (``/user/update``, ``/user/bulk_update``, and the invitation onboarding
|
||||
claim flow), so the strength bar is configured in one place instead of
|
||||
per-endpoint.
|
||||
|
||||
Also screens new passwords against known data breaches via the
|
||||
haveibeenpwned.com (HIBP) k-anonymity range API: only the first 5 characters
|
||||
of the password's SHA-1 hash ever leave the proxy, and the check fails open
|
||||
(allows the password) when HIBP is unreachable.
|
||||
"""
|
||||
|
||||
import hashlib
|
||||
from collections.abc import Mapping
|
||||
from dataclasses import dataclass
|
||||
from typing import Final
|
||||
|
||||
from litellm._logging import verbose_proxy_logger
|
||||
from litellm._version import version
|
||||
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, get_async_httpx_client
|
||||
from litellm.proxy._types import ProxyErrorTypes, ProxyException
|
||||
from litellm.types.llms.custom_http import httpxSpecialProvider
|
||||
|
||||
HIBP_RANGE_API_BASE: Final = "https://api.pwnedpasswords.com/range"
|
||||
HIBP_TIMEOUT_SECONDS: Final = 5.0
|
||||
|
||||
DEFAULT_MIN_LENGTH: Final = 12
|
||||
MIN_ALLOWED_LENGTH: Final = 8
|
||||
|
|
@ -90,3 +103,58 @@ def validate_password_policy(password: str, general_settings: Mapping[str, objec
|
|||
param="password",
|
||||
code=400,
|
||||
)
|
||||
|
||||
|
||||
def _hibp_client() -> AsyncHTTPHandler:
|
||||
return get_async_httpx_client(
|
||||
llm_provider=httpxSpecialProvider.PasswordBreachCheck,
|
||||
params={"timeout": HIBP_TIMEOUT_SECONDS},
|
||||
)
|
||||
|
||||
|
||||
def _is_suffix_in_range_response(response_body: str, hash_suffix: str) -> bool:
|
||||
for line in response_body.upper().splitlines():
|
||||
entry_suffix, _, count = line.strip().partition(":")
|
||||
if entry_suffix == hash_suffix:
|
||||
return int(count.strip() or "0") > 0
|
||||
return False
|
||||
|
||||
|
||||
async def _is_password_breached(password: str, client: AsyncHTTPHandler) -> bool:
|
||||
# usedforsecurity=False: SHA-1 is only a lookup key into the HIBP dataset, so no security property rests on it
|
||||
sha1_hex: Final = hashlib.sha1(password.encode("utf-8"), usedforsecurity=False).hexdigest().upper()
|
||||
try:
|
||||
response: Final = await client.get(
|
||||
f"{HIBP_RANGE_API_BASE}/{sha1_hex[:5]}",
|
||||
headers={"Add-Padding": "true", "User-Agent": f"litellm-proxy/{version}"},
|
||||
)
|
||||
response.raise_for_status()
|
||||
breached: Final = _is_suffix_in_range_response(response.text, sha1_hex[5:])
|
||||
except Exception as e:
|
||||
verbose_proxy_logger.warning("Breached-password check skipped, HIBP lookup failed: %s", e)
|
||||
return False
|
||||
return breached
|
||||
|
||||
|
||||
async def validate_password_not_breached(
|
||||
password: str,
|
||||
general_settings: Mapping[str, object],
|
||||
client: AsyncHTTPHandler | None = None,
|
||||
) -> None:
|
||||
"""Raise ``ProxyException`` (400) if ``password`` appears in a known data breach.
|
||||
|
||||
Fails open: an unreachable or misbehaving HIBP allows the password."""
|
||||
check_enabled: Final = general_settings.get("password_policy_check_breached_passwords", True) is not False
|
||||
if not check_enabled:
|
||||
return
|
||||
if not await _is_password_breached(password, client if client is not None else _hibp_client()):
|
||||
return
|
||||
raise ProxyException(
|
||||
message=(
|
||||
"This password appears in known data breaches and cannot be used. "
|
||||
"Please choose a different password."
|
||||
),
|
||||
type=ProxyErrorTypes.validation_error,
|
||||
param="password",
|
||||
code=400,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -28,7 +28,7 @@ from litellm._logging import verbose_proxy_logger
|
|||
from litellm._uuid import uuid
|
||||
from litellm.proxy._types import *
|
||||
from litellm.proxy.auth.auth_checks import get_team_object, get_user_object
|
||||
from litellm.proxy.auth.password_policy import validate_password_policy
|
||||
from litellm.proxy.auth.password_policy import validate_password_not_breached, validate_password_policy
|
||||
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
|
||||
from litellm.proxy.common_utils.user_api_key_cache import (
|
||||
object_permission_cache_key,
|
||||
|
|
@ -157,10 +157,11 @@ def _team_membership_table(
|
|||
return team_membership_table
|
||||
|
||||
|
||||
def _hash_password_in_dict(data: dict, general_settings: Mapping[str, object]) -> None:
|
||||
async def _hash_password_in_dict(data: dict, general_settings: Mapping[str, object]) -> None:
|
||||
"""Validate and hash password field in-place if present."""
|
||||
if "password" in data and data["password"] is not None:
|
||||
validate_password_policy(data["password"], general_settings)
|
||||
await validate_password_not_breached(data["password"], general_settings)
|
||||
data["password"] = hash_password(data["password"])
|
||||
|
||||
|
||||
|
|
@ -557,7 +558,7 @@ async def new_user(
|
|||
# generate_key_helper_fn only forwards object_permission_id, so without this the entitlement
|
||||
# the caller sent would be dropped on the floor.
|
||||
data_json = await _set_object_permission(data_json=data_json, prisma_client=prisma_client)
|
||||
_hash_password_in_dict(data_json, general_settings)
|
||||
data_json.pop("password", None) # always None: NewUserRequest.password_not_supported rejects any other value
|
||||
teams = data.teams
|
||||
if teams is None:
|
||||
teams = check_if_default_team_set()
|
||||
|
|
@ -1436,7 +1437,7 @@ async def _update_single_user_helper(
|
|||
|
||||
data_json: Final[dict] = user_request.model_dump(exclude_unset=True)
|
||||
non_default_values = _update_internal_user_params(data_json=data_json, data=user_request)
|
||||
_hash_password_in_dict(non_default_values, general_settings)
|
||||
await _hash_password_in_dict(non_default_values, general_settings)
|
||||
|
||||
existing_user_row: BaseModel | None = None
|
||||
if user_request.user_id:
|
||||
|
|
|
|||
|
|
@ -320,7 +320,7 @@ from litellm.proxy.auth.model_checks import (
|
|||
get_mcp_server_ids,
|
||||
get_team_models,
|
||||
)
|
||||
from litellm.proxy.auth.password_policy import validate_password_policy
|
||||
from litellm.proxy.auth.password_policy import validate_password_not_breached, validate_password_policy
|
||||
from litellm.proxy.auth.user_api_key_auth import (
|
||||
_fetch_global_spend_with_event_coordination,
|
||||
user_api_key_auth,
|
||||
|
|
@ -16190,6 +16190,7 @@ 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)
|
||||
current_time = litellm.utils.get_utc_datetime()
|
||||
async with prisma_client.db.tx() as tx:
|
||||
|
|
|
|||
|
|
@ -31,6 +31,7 @@ class httpxSpecialProvider(str, Enum):
|
|||
UI = "ui"
|
||||
Sandbox = "sandbox"
|
||||
ModelCostMap = "model_cost_map"
|
||||
PasswordBreachCheck = "password_breach_check"
|
||||
|
||||
|
||||
VerifyTypes = str | bool | ssl.SSLContext
|
||||
|
|
|
|||
|
|
@ -8,15 +8,20 @@ Covers the security behavior of:
|
|||
session key only after the password is written
|
||||
"""
|
||||
|
||||
import hashlib
|
||||
from datetime import timedelta
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import httpx
|
||||
import jwt
|
||||
import pytest
|
||||
import respx
|
||||
from fastapi import HTTPException
|
||||
|
||||
import litellm
|
||||
from litellm.proxy._types import InvitationClaim
|
||||
from litellm.proxy._types import InvitationClaim, ProxyException
|
||||
|
||||
_POLICY_NO_BREACH_CHECK = {"password_policy_check_breached_passwords": False}
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
|
|
@ -386,7 +391,9 @@ async def test_claim_token_rejects_concurrent_reuse_before_password_write():
|
|||
with (
|
||||
patch("litellm.proxy.proxy_server.prisma_client", prisma),
|
||||
patch("litellm.proxy.proxy_server.master_key", "sk-test"),
|
||||
patch("litellm.proxy.proxy_server.general_settings", {}),
|
||||
patch( # test-quality-ok: claim_onboarding_link reads proxy_server module globals; no injection seam
|
||||
"litellm.proxy.proxy_server.general_settings", _POLICY_NO_BREACH_CHECK
|
||||
),
|
||||
patch(
|
||||
"litellm.proxy.proxy_server.generate_key_helper_fn",
|
||||
new_callable=AsyncMock,
|
||||
|
|
@ -426,7 +433,9 @@ async def test_claim_token_sets_accepted_at_after_password_written():
|
|||
with (
|
||||
patch("litellm.proxy.proxy_server.prisma_client", prisma),
|
||||
patch("litellm.proxy.proxy_server.master_key", "sk-test"),
|
||||
patch("litellm.proxy.proxy_server.general_settings", {}),
|
||||
patch( # test-quality-ok: claim_onboarding_link reads proxy_server module globals; no injection seam
|
||||
"litellm.proxy.proxy_server.general_settings", _POLICY_NO_BREACH_CHECK
|
||||
),
|
||||
patch("litellm.proxy.proxy_server.premium_user", False),
|
||||
patch(
|
||||
"litellm.proxy.proxy_server.generate_key_helper_fn",
|
||||
|
|
@ -483,7 +492,9 @@ async def test_claim_token_rolls_back_invite_when_session_key_mint_fails():
|
|||
with (
|
||||
patch("litellm.proxy.proxy_server.prisma_client", prisma),
|
||||
patch("litellm.proxy.proxy_server.master_key", "sk-test"),
|
||||
patch("litellm.proxy.proxy_server.general_settings", {}),
|
||||
patch( # test-quality-ok: claim_onboarding_link reads proxy_server module globals; no injection seam
|
||||
"litellm.proxy.proxy_server.general_settings", _POLICY_NO_BREACH_CHECK
|
||||
),
|
||||
patch(
|
||||
"litellm.proxy.proxy_server.generate_key_helper_fn",
|
||||
new_callable=AsyncMock,
|
||||
|
|
@ -505,3 +516,124 @@ async def test_claim_token_rolls_back_invite_when_session_key_mint_fails():
|
|||
}
|
||||
assert rollback_kwargs["data"]["accepted_at"] is None
|
||||
assert rollback_kwargs["data"]["is_accepted"] is False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# POST /onboarding/claim_token - password policy
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _hibp_url_for(password: str) -> str:
|
||||
sha1 = hashlib.sha1(password.encode("utf-8"), usedforsecurity=False).hexdigest().upper()
|
||||
return f"https://api.pwnedpasswords.com/range/{sha1[:5]}"
|
||||
|
||||
|
||||
def _hibp_suffix_for(password: str) -> str:
|
||||
return hashlib.sha1(password.encode("utf-8"), usedforsecurity=False).hexdigest().upper()[5:]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_claim_token_rejects_short_password_before_consuming_invite():
|
||||
"""Default policy requires 12 characters; the invite must stay claimable."""
|
||||
from litellm.proxy.proxy_server import claim_onboarding_link
|
||||
|
||||
invite = _make_invite(is_accepted=False)
|
||||
prisma = _make_prisma(invite, _make_user())
|
||||
request = _make_claim_request(_make_onboarding_token())
|
||||
data = InvitationClaim(
|
||||
invitation_link="invite-abc",
|
||||
user_id="user-123",
|
||||
password="Sh0rt!pw",
|
||||
)
|
||||
|
||||
with (
|
||||
patch("litellm.proxy.proxy_server.prisma_client", prisma), # test-quality-ok: claim_onboarding_link reads proxy_server module globals; no injection seam
|
||||
patch("litellm.proxy.proxy_server.master_key", "sk-test"), # test-quality-ok: same as above
|
||||
patch("litellm.proxy.proxy_server.general_settings", {}), # test-quality-ok: same as above
|
||||
):
|
||||
with pytest.raises(ProxyException) as exc_info:
|
||||
await claim_onboarding_link(data=data, request=request)
|
||||
|
||||
assert exc_info.value.code == "400"
|
||||
assert "at least 12 characters" in exc_info.value.message
|
||||
prisma.db.litellm_invitationlink.update_many.assert_not_called()
|
||||
prisma.db.litellm_usertable.update.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@respx.mock
|
||||
async def test_claim_token_rejects_breached_password_before_consuming_invite():
|
||||
"""A password found in the HIBP corpus must be rejected and never stored."""
|
||||
from litellm.proxy.proxy_server import claim_onboarding_link
|
||||
|
||||
password = "P@ssword123456"
|
||||
respx.get(_hibp_url_for(password)).mock(
|
||||
return_value=httpx.Response(200, text=f"{_hibp_suffix_for(password)}:1387")
|
||||
)
|
||||
|
||||
invite = _make_invite(is_accepted=False)
|
||||
prisma = _make_prisma(invite, _make_user())
|
||||
request = _make_claim_request(_make_onboarding_token())
|
||||
data = InvitationClaim(
|
||||
invitation_link="invite-abc",
|
||||
user_id="user-123",
|
||||
password=password,
|
||||
)
|
||||
|
||||
with (
|
||||
patch("litellm.proxy.proxy_server.prisma_client", prisma), # test-quality-ok: claim_onboarding_link reads proxy_server module globals; no injection seam
|
||||
patch("litellm.proxy.proxy_server.master_key", "sk-test"), # test-quality-ok: same as above
|
||||
patch("litellm.proxy.proxy_server.general_settings", {}), # test-quality-ok: same as above
|
||||
):
|
||||
with pytest.raises(ProxyException) as exc_info:
|
||||
await claim_onboarding_link(data=data, request=request)
|
||||
|
||||
assert exc_info.value.code == "400"
|
||||
assert "data breaches" in exc_info.value.message
|
||||
prisma.db.litellm_invitationlink.update_many.assert_not_called()
|
||||
prisma.db.litellm_usertable.update.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@respx.mock
|
||||
async def test_claim_token_fails_open_when_hibp_unreachable():
|
||||
"""An HIBP outage must never block onboarding: the claim proceeds."""
|
||||
from litellm.proxy.proxy_server import claim_onboarding_link
|
||||
|
||||
password = "NewP@ssw0rd-2026"
|
||||
respx.get(_hibp_url_for(password)).mock(side_effect=httpx.ConnectError("no route to host"))
|
||||
|
||||
invite = _make_invite(is_accepted=False)
|
||||
user = _make_user()
|
||||
prisma = _make_prisma(invite, user)
|
||||
request = _make_claim_request(_make_onboarding_token())
|
||||
data = InvitationClaim(
|
||||
invitation_link="invite-abc",
|
||||
user_id="user-123",
|
||||
password=password,
|
||||
)
|
||||
|
||||
with (
|
||||
patch("litellm.proxy.proxy_server.prisma_client", prisma), # test-quality-ok: claim_onboarding_link reads proxy_server module globals; no injection seam
|
||||
patch("litellm.proxy.proxy_server.master_key", "sk-test"), # test-quality-ok: same as above
|
||||
patch("litellm.proxy.proxy_server.general_settings", {}), # test-quality-ok: same as above
|
||||
patch("litellm.proxy.proxy_server.premium_user", False), # test-quality-ok: same as above
|
||||
patch( # test-quality-ok: same as above
|
||||
"litellm.proxy.proxy_server.generate_key_helper_fn",
|
||||
new_callable=AsyncMock,
|
||||
return_value={"token": "sk-generated-key", "user_id": "user-123"},
|
||||
),
|
||||
patch( # test-quality-ok: same as above
|
||||
"litellm.proxy.proxy_server.get_custom_url",
|
||||
return_value="http://localhost:4000/",
|
||||
),
|
||||
patch( # test-quality-ok: same as above
|
||||
"litellm.proxy.proxy_server.get_disabled_non_admin_personal_key_creation",
|
||||
return_value=False,
|
||||
),
|
||||
patch("litellm.proxy.proxy_server.get_server_root_path", return_value=""), # test-quality-ok: same as above
|
||||
):
|
||||
result = await claim_onboarding_link(data=data, request=request)
|
||||
|
||||
assert "token" in result
|
||||
prisma.db.litellm_usertable.update.assert_called_once()
|
||||
|
|
|
|||
|
|
@ -2,22 +2,54 @@
|
|||
Tests for the configurable password-strength policy in
|
||||
`litellm.proxy.auth.password_policy`, enforced on every path that persists a
|
||||
new or changed password for a locally-managed user.
|
||||
|
||||
The breach-check (HIBP) tests inject a real AsyncHTTPHandler wrapping an
|
||||
httpx.MockTransport, so no network is touched and nothing is monkeypatched.
|
||||
"""
|
||||
|
||||
import hashlib
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
|
||||
from litellm.proxy._types import ProxyErrorTypes, ProxyException
|
||||
from litellm.proxy.auth.password_policy import (
|
||||
DEFAULT_MIN_LENGTH,
|
||||
MIN_ALLOWED_LENGTH,
|
||||
PasswordPolicy,
|
||||
get_password_policy,
|
||||
validate_password_not_breached,
|
||||
validate_password_policy,
|
||||
)
|
||||
|
||||
STRONG_PASSWORD = "Str0ng!Passw0rd"
|
||||
|
||||
|
||||
def _sha1_upper(password: str) -> str:
|
||||
return hashlib.sha1(password.encode("utf-8"), usedforsecurity=False).hexdigest().upper()
|
||||
|
||||
|
||||
def _client_with_transport(handler) -> AsyncHTTPHandler:
|
||||
http_handler = AsyncHTTPHandler()
|
||||
http_handler.client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
|
||||
return http_handler
|
||||
|
||||
|
||||
def _client_never_called() -> AsyncHTTPHandler:
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
raise AssertionError(f"unexpected HTTP call to {request.url}")
|
||||
|
||||
return _client_with_transport(handler)
|
||||
|
||||
|
||||
def _client_returning(body: str, status_code: int = 200) -> AsyncHTTPHandler:
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
return httpx.Response(status_code, text=body)
|
||||
|
||||
return _client_with_transport(handler)
|
||||
|
||||
|
||||
def test_get_password_policy_defaults_to_pif_baseline():
|
||||
policy = get_password_policy({})
|
||||
assert policy == PasswordPolicy(
|
||||
|
|
@ -134,3 +166,105 @@ def test_validate_password_policy_rejects_unicode_letter_as_special_character():
|
|||
def test_validate_password_policy_accepts_real_special_character_with_unicode_letters():
|
||||
"""Same base password as the rejection test above, plus an actual symbol."""
|
||||
assert validate_password_policy("Passwörd1234!", {}) is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_breach_check_skipped_when_disabled():
|
||||
result = await validate_password_not_breached(
|
||||
password="password12345", # breached in reality, but the check is off
|
||||
general_settings={"password_policy_check_breached_passwords": False},
|
||||
client=_client_never_called(),
|
||||
)
|
||||
assert result is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_rejects_breached_password():
|
||||
password = "correct horse battery staple"
|
||||
sha1 = _sha1_upper(password)
|
||||
body = f"AAAA000000000000000000000000000000A:0\r\n{sha1[5:]}:42\r\nBBBB000000000000000000000000000000B:7"
|
||||
|
||||
with pytest.raises(ProxyException) as exc_info:
|
||||
await validate_password_not_breached(password=password, general_settings={}, client=_client_returning(body))
|
||||
assert exc_info.value.code == "400"
|
||||
assert exc_info.value.type == ProxyErrorTypes.validation_error
|
||||
assert exc_info.value.param == "password"
|
||||
assert "data breaches" in exc_info.value.message
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_only_sha1_prefix_leaves_the_proxy():
|
||||
password = "a very secret password"
|
||||
sha1 = _sha1_upper(password)
|
||||
captured_requests: list[httpx.Request] = []
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
captured_requests.append(request)
|
||||
return httpx.Response(200, text="0000000000000000000000000000000000A:1")
|
||||
|
||||
result = await validate_password_not_breached(
|
||||
password=password, general_settings={}, client=_client_with_transport(handler)
|
||||
)
|
||||
assert result is None
|
||||
|
||||
(request,) = captured_requests
|
||||
assert request.url.path == f"/range/{sha1[:5]}"
|
||||
assert sha1[5:] not in str(request.url)
|
||||
assert request.headers["Add-Padding"] == "true"
|
||||
assert "litellm" in request.headers["User-Agent"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_ignores_padding_entries_with_zero_count():
|
||||
"""HIBP padding entries (requested via Add-Padding) carry count 0 and must
|
||||
not be treated as breaches when they collide with the password's suffix."""
|
||||
password = "a padded-away password"
|
||||
sha1 = _sha1_upper(password)
|
||||
|
||||
result = await validate_password_not_breached(
|
||||
password=password, general_settings={}, client=_client_returning(f"{sha1[5:]}:0")
|
||||
)
|
||||
assert result is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_accepts_password_absent_from_breach_corpus():
|
||||
result = await validate_password_not_breached(
|
||||
password="a genuinely novel password",
|
||||
general_settings={},
|
||||
client=_client_returning("0018A45C4D1DEF81644B54AB7F969B88D65:1\r\n00D4F6E8FA6EECAD2A3AA415EEC418D38EC:2"),
|
||||
)
|
||||
assert result is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_breach_check_fails_open_on_network_error():
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
raise httpx.ConnectError("no route to host")
|
||||
|
||||
result = await validate_password_not_breached(
|
||||
password="password12345", # breached, but HIBP is unreachable
|
||||
general_settings={},
|
||||
client=_client_with_transport(handler),
|
||||
)
|
||||
assert result is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_breach_check_fails_open_on_http_error_status():
|
||||
result = await validate_password_not_breached(
|
||||
password="password12345",
|
||||
general_settings={},
|
||||
client=_client_returning("service unavailable", status_code=503),
|
||||
)
|
||||
assert result is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_breach_check_fails_open_on_malformed_response_body():
|
||||
result = await validate_password_not_breached(
|
||||
password="password12345",
|
||||
general_settings={},
|
||||
client=_client_returning(f"{_sha1_upper('password12345')[5:]}:not-a-number"),
|
||||
)
|
||||
assert result is None
|
||||
|
|
|
|||
|
|
@ -1,8 +1,11 @@
|
|||
import hashlib
|
||||
import json
|
||||
from datetime import datetime, timezone
|
||||
from types import SimpleNamespace
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
import respx
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
|
||||
|
|
@ -4358,6 +4361,11 @@ async def test_user_update_hashes_and_persists_strong_password(_admin_prisma, mo
|
|||
_update_single_user_helper,
|
||||
)
|
||||
|
||||
mocker.patch( # test-quality-ok: same module-global mocking every test in this file already uses
|
||||
"litellm.proxy.proxy_server.general_settings",
|
||||
{"password_policy_check_breached_passwords": False},
|
||||
)
|
||||
|
||||
mock_prisma_client = _admin_prisma
|
||||
existing_user = mocker.MagicMock()
|
||||
existing_user.model_dump.return_value = {"user_id": "target-user"}
|
||||
|
|
@ -4375,3 +4383,29 @@ async def test_user_update_hashes_and_persists_strong_password(_admin_prisma, mo
|
|||
written_data = mock_prisma_client.update_data.call_args.kwargs["data"]
|
||||
assert written_data.get("password") is not None
|
||||
assert written_data["password"] != strong_password
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@respx.mock
|
||||
async def test_user_update_rejects_breached_password(_admin_prisma):
|
||||
"""A strength-passing password found in the HIBP corpus must be rejected
|
||||
before it ever reaches the DB write."""
|
||||
from litellm.proxy.management_endpoints.internal_user_endpoints import (
|
||||
_update_single_user_helper,
|
||||
)
|
||||
|
||||
password = "Str0ng!Passw0rd"
|
||||
sha1 = hashlib.sha1(password.encode("utf-8"), usedforsecurity=False).hexdigest().upper()
|
||||
respx.get(f"https://api.pwnedpasswords.com/range/{sha1[:5]}").mock(
|
||||
return_value=httpx.Response(200, text=f"{sha1[5:]}:1387")
|
||||
)
|
||||
|
||||
user_request = UpdateUserRequest(user_id="target-user", password=password)
|
||||
admin_caller = UserAPIKeyAuth(user_id="admin-1", user_role=LitellmUserRoles.PROXY_ADMIN)
|
||||
|
||||
with pytest.raises(ProxyException) as exc_info:
|
||||
await _update_single_user_helper(user_request=user_request, user_api_key_dict=admin_caller)
|
||||
|
||||
assert exc_info.value.code == "400"
|
||||
assert "data breaches" in exc_info.value.message
|
||||
_admin_prisma.db.litellm_usertable.find_first.assert_not_called()
|
||||
|
|
|
|||
|
|
@ -10,6 +10,7 @@ from litellm.proxy._types import (
|
|||
LiteLLM_AuditLogs,
|
||||
LiteLLM_TeamMembership,
|
||||
LitellmUserRoles,
|
||||
NewUserRequest,
|
||||
OrganizationMemberUpdateRequest,
|
||||
ResetSpendRequest,
|
||||
UpdateKeyRequest,
|
||||
|
|
@ -277,3 +278,18 @@ def test_team_membership_budget_table_present_still_works():
|
|||
}
|
||||
result = LiteLLM_TeamMembership.model_validate(data)
|
||||
assert result.litellm_budget_table is None
|
||||
|
||||
|
||||
def test_new_user_request_loudly_rejects_a_password():
|
||||
"""
|
||||
/user/new has never persisted a password (the field used to be silently
|
||||
dropped). Sending one must now fail visibly so the dead path cannot be
|
||||
revived without going through the password policy.
|
||||
"""
|
||||
with pytest.raises(ValidationError, match="invitation link"):
|
||||
NewUserRequest(user_email="alice@example.com", password="hunter2hunter2")
|
||||
|
||||
|
||||
def test_new_user_request_without_password_still_works():
|
||||
request = NewUserRequest(user_email="alice@example.com")
|
||||
assert request.password is None
|
||||
|
|
|
|||
2
ui/litellm-dashboard/src/lib/http/schema.d.ts
generated
vendored
2
ui/litellm-dashboard/src/lib/http/schema.d.ts
generated
vendored
|
|
@ -32764,6 +32764,8 @@ export interface components {
|
|||
object_permission?: components["schemas"]["LiteLLM_ObjectPermissionBase"] | null;
|
||||
/** Organizations */
|
||||
organizations?: string[] | null;
|
||||
/** Password */
|
||||
password?: string | null;
|
||||
/**
|
||||
* Permissions
|
||||
* @default {}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue