scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-13 23:11:40 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 101

fix(proxy): auto-send user invitation emails when email is configured

Browse source 
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
HammerZhaoTuro 2026-04-22 00:24:21 +00:00
parent 1d56e732e8
commit e56f57fe70
2 changed files with 270 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

63
litellm/proxy/hooks/user_management_event_hooks.py
View file

@ -27,6 +27,59 @@ from litellm.proxy.management_helpers.audit_logs import create_audit_log_for_upd
class UserManagementEventHooks:
@staticmethod
def _is_email_sending_enabled() -> bool:
"""
Check if email sending is enabled via v2 enterprise loggers or v0 alerting config.
Returns True only if email is actually configured, preventing any email
processing when the user has not opted in.
"""
try:
from litellm_enterprise.enterprise_callbacks.send_emails.base_email import (
BaseEmailLogger,
)
initialized_email_loggers = (
litellm.logging_callback_manager.get_custom_loggers_for_type(
callback_type=BaseEmailLogger
)
)
if len(initialized_email_loggers) > 0:
return True
except ImportError:
pass
from litellm.proxy.proxy_server import general_settings
if "email" in general_settings.get("alerting", []):
return True
return False
@staticmethod
def _should_send_user_invitation_email(
data: NewUserRequest,
response: NewUserResponse,
) -> bool:
"""
Determine whether a user invitation email should be sent.
- send_invite_email=True -> send if user has an email address
- send_invite_email=False -> never send
- send_invite_email=None -> auto-detect: send if email infra is configured
AND the user has an email address
"""
if data.send_invite_email is True:
return bool(response.user_email)
if data.send_invite_email is False:
return False
if not UserManagementEventHooks._is_email_sending_enabled():
return False
if not response.user_email:
return False
return True
@staticmethod
async def async_user_created_hook(
data: NewUserRequest,
@ -105,6 +158,10 @@ class UserManagementEventHooks:
key_alias=response.key_alias,
)
should_send_email = UserManagementEventHooks._should_send_user_invitation_email(
data=data, response=response
)
#########################################################
########## V2 USER INVITATION EMAIL ################
#########################################################
@ -121,7 +178,8 @@ class UserManagementEventHooks:
)
use_enterprise_email_hooks = False
if use_enterprise_email_hooks and (data.send_invite_email is True):
enterprise_email_sent = False
if use_enterprise_email_hooks and should_send_email:
initialized_email_loggers = litellm.logging_callback_manager.get_custom_loggers_for_type(
callback_type=BaseEmailLogger # type: ignore
)
@ -131,11 +189,12 @@ class UserManagementEventHooks:
await email_logger.send_user_invitation_email( # type: ignore
event=event,
)
enterprise_email_sent = True
#########################################################
########## LEGACY V1 USER INVITATION EMAIL ################
#########################################################
if data.send_invite_email is True:
if should_send_email and not enterprise_email_sent:
await UserManagementEventHooks.send_legacy_v1_user_invitation_email(
data=data,
response=response,

209
tests/test_litellm/proxy/hooks/test_send_invite_email.py
View file

@ -92,6 +92,44 @@ async def test_v1_user_creation_sends_email_when_send_invite_email_true():
mock_slack_alerting.send_key_created_or_user_invited_email.assert_called_once()
@pytest.mark.asyncio
async def test_v1_user_creation_no_email_when_send_invite_email_true_but_no_user_email():
"""
When send_invite_email=True but the response has no user_email,
no invitation email should be sent (nothing to send to).
"""
mock_slack_alerting = MagicMock()
mock_slack_alerting.send_key_created_or_user_invited_email = AsyncMock()
mock_proxy_logging_obj = MagicMock()
mock_proxy_logging_obj.slack_alerting_instance = mock_slack_alerting
with patch(
"litellm.logging_callback_manager.get_custom_loggers_for_type", return_value=[]
):
mock_proxy_server = SimpleNamespace(
general_settings={"alerting": ["email"]},
proxy_logging_obj=mock_proxy_logging_obj,
litellm_proxy_admin_name="admin-user",
)
with patch.dict(sys.modules, {"litellm.proxy.proxy_server": mock_proxy_server}):
data = NewUserRequest(
send_invite_email=True,
)
response = NewUserResponse(
user_id="test-user",
key="sk-test-key",
)
user_api_key_dict = UserAPIKeyAuth(
user_id="admin-user", api_key="admin-key"
)
await UserManagementEventHooks.async_send_user_invitation_email(
data=data,
response=response,
user_api_key_dict=user_api_key_dict,
)
mock_slack_alerting.send_key_created_or_user_invited_email.assert_not_called()
@pytest.mark.asyncio
async def test_v1_key_generation_sends_email_when_send_invite_email_true():
"""
@ -180,3 +218,174 @@ async def test_v1_key_generation_no_email_when_send_invite_email_false():
user_api_key_dict=user_api_key_dict,
)
mock_send_key_created_email.assert_not_called()
@pytest.mark.asyncio
async def test_v1_user_creation_sends_email_when_send_invite_email_none_and_email_configured():
"""
When send_invite_email is None (default) and email alerting is configured
and the user has an email address, an invitation email should be sent.
"""
mock_slack_alerting = MagicMock()
mock_slack_alerting.send_key_created_or_user_invited_email = AsyncMock()
mock_proxy_logging_obj = MagicMock()
mock_proxy_logging_obj.slack_alerting_instance = mock_slack_alerting
with patch(
"litellm.logging_callback_manager.get_custom_loggers_for_type", return_value=[]
):
mock_proxy_server = SimpleNamespace(
general_settings={"alerting": ["email"]},
proxy_logging_obj=mock_proxy_logging_obj,
litellm_proxy_admin_name="admin-user",
)
with patch.dict(sys.modules, {"litellm.proxy.proxy_server": mock_proxy_server}):
data = NewUserRequest(
user_email="test@example.com",
)
response = NewUserResponse(
user_id="test-user",
user_email="test@example.com",
key="sk-test-key",
)
user_api_key_dict = UserAPIKeyAuth(
user_id="admin-user", api_key="admin-key"
)
await UserManagementEventHooks.async_send_user_invitation_email(
data=data,
response=response,
user_api_key_dict=user_api_key_dict,
)
mock_slack_alerting.send_key_created_or_user_invited_email.assert_called_once()
@pytest.mark.asyncio
async def test_v1_user_creation_no_email_when_send_invite_email_none_and_email_not_configured():
"""
When send_invite_email is None (default) and email alerting is NOT configured,
no invitation email should be sent.
"""
mock_slack_alerting = MagicMock()
mock_slack_alerting.send_key_created_or_user_invited_email = AsyncMock()
mock_proxy_logging_obj = MagicMock()
mock_proxy_logging_obj.slack_alerting_instance = mock_slack_alerting
with patch(
"litellm.logging_callback_manager.get_custom_loggers_for_type", return_value=[]
):
mock_proxy_server = SimpleNamespace(
general_settings={},
proxy_logging_obj=mock_proxy_logging_obj,
litellm_proxy_admin_name="admin-user",
)
with patch.dict(sys.modules, {"litellm.proxy.proxy_server": mock_proxy_server}):
data = NewUserRequest(
user_email="test@example.com",
)
response = NewUserResponse(
user_id="test-user",
user_email="test@example.com",
key="sk-test-key",
)
user_api_key_dict = UserAPIKeyAuth(
user_id="admin-user", api_key="admin-key"
)
await UserManagementEventHooks.async_send_user_invitation_email(
data=data,
response=response,
user_api_key_dict=user_api_key_dict,
)
mock_slack_alerting.send_key_created_or_user_invited_email.assert_not_called()
@pytest.mark.asyncio
async def test_v1_user_creation_no_email_when_send_invite_email_none_and_no_user_email():
"""
When send_invite_email is None (default) and email alerting is configured
but the user has no email address, no invitation email should be sent.
"""
mock_slack_alerting = MagicMock()
mock_slack_alerting.send_key_created_or_user_invited_email = AsyncMock()
mock_proxy_logging_obj = MagicMock()
mock_proxy_logging_obj.slack_alerting_instance = mock_slack_alerting
with patch(
"litellm.logging_callback_manager.get_custom_loggers_for_type", return_value=[]
):
mock_proxy_server = SimpleNamespace(
general_settings={"alerting": ["email"]},
proxy_logging_obj=mock_proxy_logging_obj,
litellm_proxy_admin_name="admin-user",
)
with patch.dict(sys.modules, {"litellm.proxy.proxy_server": mock_proxy_server}):
data = NewUserRequest()
response = NewUserResponse(
user_id="test-user",
key="sk-test-key",
)
user_api_key_dict = UserAPIKeyAuth(
user_id="admin-user", api_key="admin-key"
)
await UserManagementEventHooks.async_send_user_invitation_email(
data=data,
response=response,
user_api_key_dict=user_api_key_dict,
)
mock_slack_alerting.send_key_created_or_user_invited_email.assert_not_called()
@pytest.mark.asyncio
async def test_v1_user_creation_no_legacy_email_when_enterprise_handles_send():
"""
When enterprise email loggers are registered and send the email,
the legacy V1 path should NOT also fire (which would raise ValueError
if "email" is not in general_settings["alerting"]).
"""
mock_slack_alerting = MagicMock()
mock_slack_alerting.send_key_created_or_user_invited_email = AsyncMock()
mock_proxy_logging_obj = MagicMock()
mock_proxy_logging_obj.slack_alerting_instance = mock_slack_alerting
mock_email_logger = MagicMock()
mock_email_logger.send_user_invitation_email = AsyncMock()
mock_proxy_server = SimpleNamespace(
general_settings={},
proxy_logging_obj=mock_proxy_logging_obj,
litellm_proxy_admin_name="admin-user",
)
with patch.dict(sys.modules, {"litellm.proxy.proxy_server": mock_proxy_server}):
with patch(
"litellm.logging_callback_manager.get_custom_loggers_for_type",
return_value=[mock_email_logger],
):
with patch.dict(
sys.modules,
{
"litellm_enterprise": MagicMock(),
"litellm_enterprise.enterprise_callbacks": MagicMock(),
"litellm_enterprise.enterprise_callbacks.send_emails": MagicMock(),
"litellm_enterprise.enterprise_callbacks.send_emails.base_email": MagicMock(
BaseEmailLogger=type(mock_email_logger),
),
},
):
data = NewUserRequest(
user_email="test@example.com",
)
response = NewUserResponse(
user_id="test-user",
user_email="test@example.com",
key="sk-test-key",
)
user_api_key_dict = UserAPIKeyAuth(
user_id="admin-user", api_key="admin-key"
)
await UserManagementEventHooks.async_send_user_invitation_email(
data=data,
response=response,
user_api_key_dict=user_api_key_dict,
)
mock_email_logger.send_user_invitation_email.assert_called_once()
mock_slack_alerting.send_key_created_or_user_invited_email.assert_not_called()