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

1""" 

2UI session revocation. 

3 

4POST /session/logout — revoke the UI session key this request authenticated with. 

5revoke_ui_session_keys — revoke every UI session key a user holds (password writes). 

6 

7Logging out of the dashboard was purely client-side (cookies cleared, redirect); 

8the DB-backed virtual key minted at login stayed valid until 

9LITELLM_UI_SESSION_DURATION elapsed, so a captured token kept working access 

10after logout, and changing a password did not invalidate existing sessions. 

11 

12Deliberately NOT reusing /key/delete: its `can_modify_verification_token` 

13ownership checks can reject low-privilege roles, and a self-revoke endpoint 

14that takes no body cannot be aimed at other keys. 

15""" 

16 

17from typing import TYPE_CHECKING, Annotated, Final, cast 

18 

19from fastapi import APIRouter, Depends, HTTPException, Response 

20from pydantic import TypeAdapter 

21 

22from litellm._logging import verbose_proxy_logger 

23from litellm.constants import UI_SESSION_TOKEN_TEAM_ID 

24from litellm.proxy._types import ( 

25 CommonProxyErrors, 

26 HTTPExceptionErrorDetail, 

27 LiteLLM_VerificationToken, 

28 SessionLogoutResponse, 

29 UserAPIKeyAuth, 

30) 

31from litellm.proxy.auth.auth_checks import delete_cache_key_objects 

32from litellm.proxy.auth.user_api_key_auth import user_api_key_auth 

33from litellm.proxy.management_endpoints.key_management_endpoints import ( 

34 _persist_deleted_verification_tokens, 

35) 

36from litellm.repositories.verification_token_repository import ( 

37 VerificationTokenRepository, 

38) 

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 types as prisma_types 

42 

43router: Final = APIRouter() 

44 

45_TOKEN_LIST: Final = TypeAdapter(list[str]) 

46 

47 

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

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

50 return detail 

51 

52 

53async def revoke_ui_session_keys( 

54 user_id: str, 

55 user_api_key_dict: UserAPIKeyAuth, 

56 *, 

57 keep_hashed_token: str | None = None, 

58 litellm_changed_by: str | None = None, 

59) -> int: 

60 """Revoke every UI session key belonging to ``user_id``, except 

61 ``keep_hashed_token`` (the caller's own session on a self-service password 

62 change; the other password-write paths revoke all). 

63 

64 Best-effort: the password write this runs after has already committed, so a 

65 revocation failure is logged loudly rather than failing the request — the 

66 unrevoked keys still expire at LITELLM_UI_SESSION_DURATION. 

67 

68 Returns the number of sessions revoked. 

69 """ 

70 from litellm.proxy.proxy_server import prisma_client, proxy_logging_obj, user_api_key_cache 

71 

72 if prisma_client is None: 

73 return 0 

74 

75 try: 

76 where_user_sessions: Final[prisma_types.LiteLLM_VerificationTokenWhereInput] = { 

77 "user_id": user_id, 

78 "team_id": UI_SESSION_TOKEN_TEAM_ID, 

79 } 

80 rows: Final = cast( # cast-ok: find_many returns prisma rows shaped like the pydantic model 

81 "tuple[LiteLLM_VerificationToken, ...]", 

82 tuple(await VerificationTokenRepository(prisma_client).table.find_many(where=where_user_sessions)), 

83 ) 

84 revoked_rows: Final = tuple(row for row in rows if row.token is not None and row.token != keep_hashed_token) 

85 if not revoked_rows: 

86 return 0 

87 revoked_tokens: Final = _TOKEN_LIST.validate_python(tuple(row.token for row in revoked_rows)) 

88 

89 await _persist_deleted_verification_tokens( 

90 keys=revoked_rows, 

91 prisma_client=prisma_client, 

92 user_api_key_dict=user_api_key_dict, 

93 litellm_changed_by=litellm_changed_by, 

94 ) 

95 where_revoked: Final[prisma_types.LiteLLM_VerificationTokenWhereInput] = {"token": {"in": revoked_tokens}} 

96 await VerificationTokenRepository(prisma_client).table.delete_many(where=where_revoked) 

97 await delete_cache_key_objects( 

98 hashed_tokens=revoked_tokens, 

99 user_api_key_cache=user_api_key_cache, 

100 proxy_logging_obj=proxy_logging_obj, 

101 ) 

102 verbose_proxy_logger.info( 

103 "Revoked %s UI session key(s) for user_id=%s after password change", 

104 len(revoked_tokens), 

105 user_id, 

106 ) 

107 return len(revoked_tokens) 

108 except Exception: # noqa: BLE001 # the password write committed; revocation must not undo that 

109 verbose_proxy_logger.exception( 

110 "Failed to revoke UI session keys for user_id=%s; existing sessions remain valid until they expire", 

111 user_id, 

112 ) 

113 return 0 

114 

115 

116@router.post( 

117 "/session/logout", 

118 tags=("UI Session",), 

119) 

120async def session_logout( 

121 response: Response, 

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

123) -> SessionLogoutResponse: 

124 """ 

125 Revoke the UI session key this request authenticated with. 

126 

127 Only accepts UI session keys (minted by dashboard login); any other 

128 credential is refused, so this can never be used to delete arbitrary keys. 

129 Revokes only the presented session, not the user's other sessions. 

130 Idempotent: logging out an already-revoked session succeeds. 

131 """ 

132 from litellm.proxy.proxy_server import prisma_client, proxy_logging_obj, user_api_key_cache 

133 

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

135 raise HTTPException( 

136 status_code=500, 

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

138 ) 

139 

140 if user_api_key_dict.team_id != UI_SESSION_TOKEN_TEAM_ID: 140 ↛ 146line 140 didn't jump to line 146 because the condition on line 140 was always true

141 raise HTTPException( 

142 status_code=403, 

143 detail=_error_detail("Only UI session tokens can be revoked through this endpoint."), 

144 ) 

145 

146 hashed_token: Final = user_api_key_dict.token 

147 revoked = False 

148 if hashed_token is not None: 

149 where_token: Final[prisma_types.LiteLLM_VerificationTokenWhereUniqueInput] = {"token": hashed_token} 

150 row: Final = await VerificationTokenRepository(prisma_client).table.find_unique(where=where_token) 

151 # A missing row means the session is already revoked (or an 

152 # EXPERIMENTAL_UI_LOGIN blob token); logout is idempotent either way. 

153 if row is not None: 

154 caller_row: Final = cast( # cast-ok: find_unique returns a prisma row shaped like the pydantic model 

155 "LiteLLM_VerificationToken", row 

156 ) 

157 await _persist_deleted_verification_tokens( 

158 keys=(caller_row,), 

159 prisma_client=prisma_client, 

160 user_api_key_dict=user_api_key_dict, 

161 ) 

162 await VerificationTokenRepository(prisma_client).table.delete_many(where=where_token) 

163 revoked = True 

164 await delete_cache_key_objects( 

165 hashed_tokens=(hashed_token,), 

166 user_api_key_cache=user_api_key_cache, 

167 proxy_logging_obj=proxy_logging_obj, 

168 ) 

169 

170 # The server set this cookie at login (set_session_token_cookie); clear it 

171 # here too so logout works even if the client-side clear is skipped. 

172 response.delete_cookie("token") 

173 return SessionLogoutResponse( 

174 message="Session revoked." if revoked else "Session already revoked.", 

175 )