Prevent password leakage in login reports

app/notification/__init__.py

@@ -219,12 +219,12 @@ async def admin_usage_limit_reached(admin: AdminDetails, usage_percentage: int,
         )
 
 
-async def admin_login(username: str, password: str, client_ip: str, success: bool):
+async def admin_login(username: str, client_ip: str, success: bool):
     if (await notification_enable()).admin.login:
         await _gather_notifications(
             "admin_login",
-            ds.admin_login(username, password, client_ip, success),
-            tg.admin_login(username, password, client_ip, success),
+            ds.admin_login(username, client_ip, success),
+            tg.admin_login(username, client_ip, success),
         )
 
 

app/notification/discord/admin.py

@@ -107,12 +107,11 @@ async def admin_usage_limit_reached(admin: AdminDetails, usage_percentage: int,
         await send_discord_webhook(data, admin.discord_webhook)
 
 
-async def admin_login(username: str, password: str, client_ip: str, success: bool):
-    username, password = escape_ds_markdown_list((username, password))
+async def admin_login(username: str, client_ip: str, success: bool):
+    username = escape_ds_markdown_list((username,))[0]
     message = {**messages.ADMIN_LOGIN, "footer": dict(messages.ADMIN_LOGIN["footer"])}
     message["description"] = message["description"].format(
         username=username,
-        password="🔒" if success else password,
         client_ip=client_ip,
     )
     message["footer"]["text"] = message["footer"]["text"].format(status="Successful" if success else "Failed")

app/notification/discord/messages.py

@@ -133,7 +133,7 @@
 
 ADMIN_LOGIN = {
     "title": "Login Attempt",
-    "description": "**Username:** {username}\n**Password:** {password}\n**IP:** {client_ip}",
+    "description": "**Username:** {username}\n**IP:** {client_ip}",
     "footer": {"text": "{status}"},
 }
 

app/notification/telegram/admin.py

@@ -75,12 +75,11 @@ async def admin_usage_limit_reached(admin: AdminDetails, usage_percentage: int,
         await send_telegram_message(data, chat_id=admin.telegram_id)
 
 
-async def admin_login(username: str, password: str, client_ip: str, success: bool):
-    username, password = escape_tg_html((username, password))
+async def admin_login(username: str, client_ip: str, success: bool):
+    username = escape_tg_html((username,))[0]
     data = messages.ADMIN_LOGIN.format(
         status="Successful" if success else "Failed",
         username=username,
-        password="🔒" if success else password,
         client_ip=client_ip,
     )
     settings: NotificationSettings = await notification_settings()

app/notification/telegram/messages.py

@@ -128,7 +128,6 @@
 <i>Status</i>: {status}
 ➖➖➖➖➖➖➖➖➖
 <b>Username:</b> <code>{username}</code>
-<b>Password:</b> <code>{password}</code>
 <b>IP:</b> <code>{client_ip}</code>
 """
 

app/routers/admin.py

@@ -49,16 +49,16 @@ async def admin_token(
     client_ip = get_client_ip(request)
     db_admin = await validate_admin(db, form_data.username, form_data.password)
     if not db_admin:
-        asyncio.create_task(notification.admin_login(form_data.username, form_data.password, client_ip, False))
+        asyncio.create_task(notification.admin_login(form_data.username, client_ip, False))
         raise HTTPException(
             status_code=401, detail="Incorrect username or password", headers={"WWW-Authenticate": "Bearer"}
         )
     if db_admin.status == AdminStatus.disabled:
-        asyncio.create_task(notification.admin_login(form_data.username, form_data.password, client_ip, False))
+        asyncio.create_task(notification.admin_login(form_data.username, client_ip, False))
         raise HTTPException(
             status_code=403, detail="your account has been disabled", headers={"WWW-Authenticate": "Bearer"}
         )
-    asyncio.create_task(notification.admin_login(db_admin.username, "", client_ip, True))
+    asyncio.create_task(notification.admin_login(db_admin.username, client_ip, True))
     return Token(access_token=await create_admin_token(db_admin.id, form_data.username))
 
 
@@ -75,7 +75,7 @@ async def admin_mini_app_token(
         raise HTTPException(
             status_code=403, detail="your account has been disabled", headers={"WWW-Authenticate": "Bearer"}
         )
-    asyncio.create_task(notification.admin_login(db_admin.username, "", client_ip, True))
+    asyncio.create_task(notification.admin_login(db_admin.username, client_ip, True))
     return Token(access_token=await create_admin_token(db_admin.id, db_admin.username))
 
 

tests/test_admin_login_notifications.py

@@ -0,0 +1,88 @@
+import asyncio
+import json
+from types import SimpleNamespace
+from unittest.mock import ANY, AsyncMock
+
+import pytest
+from fastapi import HTTPException
+
+from app.models.admin import AdminStatus
+from app.notification.discord import admin as discord_admin
+from app.notification.telegram import admin as telegram_admin
+from app.routers import admin as admin_router
+
+SUBMITTED_PASSWORD = "correct-horse-battery-staple"
+
+
+@pytest.mark.parametrize(
+    ("db_admin", "expected_status"),
+    [
+        (None, 401),
+        (SimpleNamespace(id=7, username="disabled-admin", status=AdminStatus.disabled), 403),
+    ],
+)
+@pytest.mark.asyncio
+async def test_failed_login_report_never_receives_submitted_password(
+    monkeypatch: pytest.MonkeyPatch, db_admin: SimpleNamespace | None, expected_status: int
+):
+    validate_admin = AsyncMock(return_value=db_admin)
+    report_login = AsyncMock()
+    monkeypatch.setattr(admin_router, "validate_admin", validate_admin)
+    monkeypatch.setattr(admin_router.notification, "admin_login", report_login)
+    monkeypatch.setattr(admin_router, "get_client_ip", lambda request: "203.0.113.10")
+
+    form_data = SimpleNamespace(username="disabled-admin", password=SUBMITTED_PASSWORD)
+    with pytest.raises(HTTPException) as exc_info:
+        await admin_router.admin_token(SimpleNamespace(), form_data, SimpleNamespace())
+    await asyncio.sleep(0)
+
+    assert exc_info.value.status_code == expected_status
+    validate_admin.assert_awaited_once_with(ANY, "disabled-admin", SUBMITTED_PASSWORD)
+    report_login.assert_awaited_once_with("disabled-admin", "203.0.113.10", False)
+    assert SUBMITTED_PASSWORD not in repr(report_login.await_args)
+
+
+@pytest.mark.asyncio
+async def test_db_admin_login_uses_submitted_password_but_reports_only_safe_metadata(monkeypatch: pytest.MonkeyPatch):
+    db_admin = SimpleNamespace(id=7, username="db-admin", status=AdminStatus.active)
+    validate_admin = AsyncMock(return_value=db_admin)
+    report_login = AsyncMock()
+    create_admin_token = AsyncMock(return_value="access-token")
+    monkeypatch.setattr(admin_router, "validate_admin", validate_admin)
+    monkeypatch.setattr(admin_router.notification, "admin_login", report_login)
+    monkeypatch.setattr(admin_router, "create_admin_token", create_admin_token)
+    monkeypatch.setattr(admin_router, "get_client_ip", lambda request: "203.0.113.10")
+
+    form_data = SimpleNamespace(username="db-admin", password=SUBMITTED_PASSWORD)
+    token = await admin_router.admin_token(SimpleNamespace(), form_data, SimpleNamespace())
+    await asyncio.sleep(0)
+
+    assert token.access_token == "access-token"
+    validate_admin.assert_awaited_once_with(ANY, "db-admin", SUBMITTED_PASSWORD)
+    report_login.assert_awaited_once_with("db-admin", "203.0.113.10", True)
+    assert SUBMITTED_PASSWORD not in repr(report_login.await_args)
+
+
+@pytest.mark.asyncio
+async def test_login_renderers_emit_safe_metadata_only(monkeypatch: pytest.MonkeyPatch):
+    telegram_send = AsyncMock()
+    discord_send = AsyncMock()
+    settings = SimpleNamespace(notify_telegram=True, notify_discord=True)
+
+    monkeypatch.setattr(telegram_admin, "notification_settings", AsyncMock(return_value=settings))
+    monkeypatch.setattr(telegram_admin, "get_telegram_channel", lambda settings, entity: (123, None))
+    monkeypatch.setattr(telegram_admin, "send_telegram_message", telegram_send)
+    monkeypatch.setattr(discord_admin, "notification_settings", AsyncMock(return_value=settings))
+    monkeypatch.setattr(discord_admin, "get_discord_webhook", lambda settings, entity: "https://example.test/hook")
+    monkeypatch.setattr(discord_admin, "send_discord_webhook", discord_send)
+
+    await telegram_admin.admin_login("db-admin", "203.0.113.10", False)
+    await discord_admin.admin_login("db-admin", "203.0.113.10", False)
+
+    telegram_payload = telegram_send.await_args.args[0]
+    discord_payload = json.dumps(discord_send.await_args.args[0])
+    for payload in (telegram_payload, discord_payload):
+        assert "db-admin" in payload
+        assert "203.0.113.10" in payload
+        assert "password" not in payload.lower()
+        assert SUBMITTED_PASSWORD not in payload