Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/management_endpoints/password_endpoints.py: 47%

1""" 

2Self-service password management. 

3 

4/user/password/change 

5 

6Deliberately NOT wrapped in `management_endpoint_wrapper`: the wrapper emits 

7request kwargs to OTEL spans, which would log plaintext passwords. The audit 

8signal is emitted by hand below, with field names only, never values. 

9""" 

10 

11from typing import TYPE_CHECKING, Annotated, Final 

12 

13from fastapi import APIRouter, Depends, HTTPException 

14from pydantic import TypeAdapter 

15 

16from litellm._logging import verbose_proxy_logger 

17from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler 

18from litellm.proxy._types import ( 

19 UI_TEAM_ID, 

20 ChangePasswordRequest, 

21 ChangePasswordResponse, 

22 CommonProxyErrors, 

23 HTTPExceptionErrorDetail, 

24 LitellmTableNames, 

25 UserAPIKeyAuth, 

26) 

27from litellm.proxy.auth.login_utils import PASSWORD_SESSION_METADATA 

28from litellm.proxy.auth.password_policy import ( 

29 get_hibp_client, 

30 validate_password_not_breached, 

31 validate_password_policy, 

32) 

33from litellm.proxy.auth.user_api_key_auth import user_api_key_auth 

34from litellm.proxy.management_endpoints.session_endpoints import revoke_ui_session_keys 

35from litellm.proxy.management_helpers.audit_logs import create_object_audit_log 

36from litellm.proxy.utils import hash_password, verify_password 

37from litellm.repositories.prisma_protocols import TableActions 

38from litellm.repositories.user_repository import UserRepository 

39 

40if TYPE_CHECKING: 40 ↛ 41line 40 didn't jump to line 41 because the condition on line 40 was never true

41 from prisma import models as prisma_models 

42 from prisma import types as prisma_types 

43 

44 from litellm.proxy.utils import PrismaClient 

45 

46router: Final = APIRouter() 

47 

48_PASSWORD_CHANGED_AUDIT_VALUES: Final = '{"fields_changed": ["password"]}' 

49_KEY_METADATA: Final = TypeAdapter(dict[str, object]) 

50 

51 

52def _error_detail(message: str) -> HTTPExceptionErrorDetail: 

53 detail: Final[HTTPExceptionErrorDetail] = {"error": message} 

54 return detail 

55 

56 

57def _is_password_login_session(user_api_key_dict: UserAPIKeyAuth) -> bool: 

58 if user_api_key_dict.team_id != UI_TEAM_ID: 58 ↛ 60line 58 didn't jump to line 60 because the condition on line 58 was always true

59 return False 

60 key_metadata: Final = _KEY_METADATA.validate_python(user_api_key_dict.metadata) 

61 return all(key_metadata.get(k) == v for k, v in PASSWORD_SESSION_METADATA.items()) 

62 

63 

64def _user_table( 

65 prisma_client: "PrismaClient | None", 

66) -> "TableActions[prisma_models.LiteLLM_UserTable]": 

67 user_table: Final[TableActions[prisma_models.LiteLLM_UserTable]] = UserRepository(prisma_client).table 

68 return user_table 

69 

70 

71@router.post( 

72 "/user/password/change", 

73 tags=("Internal User management",), 

74 dependencies=(Depends(user_api_key_auth),), 

75) 

76async def change_password( 

77 data: ChangePasswordRequest, 

78 user_api_key_dict: Annotated[UserAPIKeyAuth, Depends(user_api_key_auth)], 

79 hibp_client: Annotated[AsyncHTTPHandler, Depends(get_hibp_client)], 

80) -> ChangePasswordResponse: 

81 """ 

82 Change the calling user's own password. 

83 

84 Only callable with the dashboard session issued by a username/password 

85 login; SSO sessions and virtual keys are rejected with 403. Requires the 

86 current password. The new password must differ from the 

87 current one and satisfy the configured password policy 

88 (`general_settings.password_policy_*`: minimum length, character classes, 

89 and, when enabled, breached-password screening via haveibeenpwned.com). 

90 A successful change lifts any pending forced password reset 

91 (`password_reset_required`) on the account. 

92 

93 Parameters: 

94 - current_password: str - The user's current password. 

95 - new_password: str - The password to change to. 

96 """ 

97 from litellm.proxy.proxy_server import general_settings, litellm_proxy_admin_name, prisma_client 

98 

99 if prisma_client is None: 99 ↛ 100line 99 didn't jump to line 100 because the condition on line 99 was never true

100 raise HTTPException( 

101 status_code=500, 

102 detail=_error_detail(CommonProxyErrors.db_not_connected_error.value), 

103 ) 

104 

105 if not _is_password_login_session(user_api_key_dict): 105 ↛ 113line 105 didn't jump to line 113 because the condition on line 105 was always true

106 raise HTTPException( 

107 status_code=403, 

108 detail=_error_detail( 

109 "Passwords can only be changed from a dashboard session created by logging in with a password." 

110 ), 

111 ) 

112 

113 user_id: Final = user_api_key_dict.user_id 

114 if user_id is None: 

115 raise HTTPException( 

116 status_code=400, 

117 detail=_error_detail("No user is associated with this session, so there is no password to change."), 

118 ) 

119 

120 find_user: Final[prisma_types.LiteLLM_UserTableWhereInput] = {"user_id": user_id} 

121 user_row: Final = await _user_table(prisma_client).find_first(where=find_user) 

122 stored_password: Final = user_row.password if user_row is not None else None 

123 if stored_password is None: 

124 raise HTTPException( 

125 status_code=400, 

126 detail=_error_detail( 

127 "This account has no password set, so there is no password to change. " 

128 "Passwords are set through an invitation link (POST /invitation/new)." 

129 ), 

130 ) 

131 

132 if not verify_password(data.current_password, stored_password): 

133 raise HTTPException(status_code=400, detail=_error_detail("Current password is incorrect.")) 

134 

135 if data.new_password == data.current_password: 

136 raise HTTPException( 

137 status_code=400, 

138 detail=_error_detail("New password must be different from the current password."), 

139 ) 

140 

141 validate_password_policy(data.new_password, general_settings) 

142 await validate_password_not_breached(data.new_password, general_settings, hibp_client) 

143 

144 password_update: Final[prisma_types.LiteLLM_UserTableUpdateInput] = { 

145 "password": hash_password(data.new_password), 

146 "password_reset_required": False, 

147 "last_breach_check_at": None, 

148 } 

149 await _user_table(prisma_client).update(where=find_user, data=password_update) 

150 

151 # The old password may have been compromised; revoke every other UI session 

152 # so a holder of a stolen session token is cut off. The caller's own session 

153 # is kept — they just proved they hold the current password. 

154 await revoke_ui_session_keys( 

155 user_id=user_id, 

156 user_api_key_dict=user_api_key_dict, 

157 keep_hashed_token=user_api_key_dict.token, 

158 ) 

159 

160 verbose_proxy_logger.info("Password changed via /user/password/change for user_id=%s", user_id) 

161 await create_object_audit_log( 

162 object_id=user_id, 

163 action="updated", 

164 litellm_changed_by=None, 

165 user_api_key_dict=user_api_key_dict, 

166 litellm_proxy_admin_name=litellm_proxy_admin_name, 

167 table_name=LitellmTableNames.USER_TABLE_NAME, 

168 after_value=_PASSWORD_CHANGED_AUDIT_VALUES, 

169 ) 

170 return ChangePasswordResponse(user_id=user_id, message="Password updated successfully.")