feat: audit outbound user notifications

2026-05-31 23:05:40 +03:00
parent 1578a9da36
commit ecf779763c
7 changed files with 241 additions and 1 deletions
@@ -17,8 +17,9 @@ from sqlalchemy.ext.asyncio import AsyncSession

 from bot.middlewares.i18n import JsonI18n
 from bot.services.email_templates import EmailContent, render_login_code
+from bot.services.message_audit import log_user_message_delivery
 from config.settings import Settings
-from db.dal import security_dal
+from db.dal import security_dal, user_dal
 from db.models import EmailVerificationCode

 logger = logging.getLogger(__name__)
@@ -222,6 +223,28 @@ class EmailAuthService:
            magic_link=magic_link,
            purpose=purpose,
        )
+        resolved_target_user_id = target_user_id
+        if resolved_target_user_id is None:
+            try:
+                existing_user = await user_dal.get_user_by_email(session, normalized_email)
+                resolved_target_user_id = (
+                    int(existing_user.user_id) if existing_user is not None else None
+                )
+            except Exception:
+                logger.exception(
+                    "Failed to resolve email auth target user for audit log: %s",
+                    normalized_email,
+                )
+        await log_user_message_delivery(
+            session,
+            target_user_id=resolved_target_user_id,
+            event_type="email_login_code_sent"
+            if purpose == "login"
+            else "email_verification_code_sent",
+            channel="email",
+            recipient=normalized_email,
+            content=f"purpose={purpose} magic_link={bool(magic_link)}",
+        )
        return EmailCodeRequestResult(ok=True)

    async def verify_code(
@@ -0,0 +1,57 @@
+import logging
+from datetime import datetime, timezone
+from typing import Optional
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from db.dal import message_log_dal
+
+logger = logging.getLogger(__name__)
+
+
+def _clean_piece(value: Optional[object]) -> str:
+    return str(value or "").strip()
+
+
+async def log_user_message_delivery(
+    session: AsyncSession,
+    *,
+    target_user_id: Optional[int],
+    event_type: str,
+    channel: str,
+    content: str,
+    recipient: Optional[str] = None,
+    timestamp: Optional[datetime] = None,
+) -> None:
+    """Add a best-effort user log entry for important outbound messages."""
+    clean_event = _clean_piece(event_type)
+    clean_channel = _clean_piece(channel)
+    if not clean_event or not clean_channel:
+        return
+
+    parts = [f"channel={clean_channel}"]
+    clean_recipient = _clean_piece(recipient)
+    if clean_recipient:
+        parts.append(f"recipient={clean_recipient}")
+    clean_content = _clean_piece(content)
+    if clean_content:
+        parts.append(clean_content)
+
+    try:
+        await message_log_dal.create_message_log_no_commit(
+            session,
+            {
+                "user_id": None,
+                "event_type": clean_event,
+                "content": " | ".join(parts)[:4000],
+                "is_admin_event": False,
+                "target_user_id": int(target_user_id) if target_user_id is not None else None,
+                "timestamp": timestamp or datetime.now(timezone.utc),
+            },
+        )
+    except Exception:
+        logger.exception(
+            "Failed to add outbound message audit log for user %s event %s",
+            target_user_id,
+            clean_event,
+        )
@@ -13,6 +13,7 @@ from bot.keyboards.inline.user_keyboards import get_subscribe_only_markup
 from bot.middlewares.i18n import JsonI18n
 from bot.services.email_auth_service import EmailAuthService
 from bot.services.email_templates import render_subscription_lifecycle_notification
+from bot.services.message_audit import log_user_message_delivery
 from bot.services.telegram_notifications import (
    TELEGRAM_NOTIFICATIONS_BLOCKED,
    TELEGRAM_NOTIFICATIONS_ENABLED,
@@ -188,6 +189,18 @@ class SubscriptionLifecycleNotificationService:
            self._channel_key(stage.key, "telegram"),
            sent_at=sent_at,
        )
+        await log_user_message_delivery(
+            session,
+            target_user_id=getattr(sub, "user_id", None),
+            event_type="telegram_subscription_notification_sent",
+            channel="telegram",
+            recipient=str(chat_id),
+            content=(
+                f"stage={stage.key} message_key={stage.message_key} "
+                f"subscription_id={getattr(sub, 'subscription_id', '')}"
+            ),
+            timestamp=sent_at,
+        )
        if user:
            status = normalize_telegram_notification_status(
                getattr(user, "telegram_notifications_status", None)
@@ -253,6 +266,18 @@ class SubscriptionLifecycleNotificationService:
            self._channel_key(stage.key, "email"),
            sent_at=sent_at,
        )
+        await log_user_message_delivery(
+            session,
+            target_user_id=getattr(sub, "user_id", None),
+            event_type="email_subscription_notification_sent",
+            channel="email",
+            recipient=recipient,
+            content=(
+                f"stage={stage.key} message_key={stage.message_key} "
+                f"subscription_id={getattr(sub, 'subscription_id', '')}"
+            ),
+            timestamp=sent_at,
+        )
        return True

    async def _already_sent(
@@ -13,6 +13,7 @@ from sqlalchemy.orm import selectinload, sessionmaker
 from bot.infra.redis import redis_lock
 from bot.keyboards.inline.user_keyboards import get_subscribe_only_markup
 from bot.middlewares.i18n import JsonI18n
+from bot.services.message_audit import log_user_message_delivery
 from bot.services.panel_api_service import PanelApiService
 from bot.services.subscription_lifecycle_notifications import (
    SubscriptionLifecycleNotificationService,
@@ -331,6 +332,17 @@ class SubscriptionNotificationWorker:
                    parse_mode="HTML",
                )
                telegram_sent = True
+                await log_user_message_delivery(
+                    session,
+                    target_user_id=user_id,
+                    event_type="telegram_traffic_warning_sent",
+                    channel="telegram",
+                    recipient=str(telegram_chat_id),
+                    content=(
+                        "kind=trial warning_key=trial_traffic_depleted "
+                        f"used_bytes={used} limit_bytes={limit}"
+                    ),
+                )
            except Exception as exc:
                status = telegram_notification_status_from_error(exc)
                if status and user and user_id:
@@ -355,6 +367,12 @@ class SubscriptionNotificationWorker:
                subject_key="email_trial_traffic_depleted_subject",
                message_text=message_text,
                dashboard_url=(getattr(self.settings, "SUBSCRIPTION_MINI_APP_URL", "") or None),
+                session=session,
+                audit_event_type="email_traffic_warning_sent",
+                audit_content=(
+                    "kind=trial warning_key=trial_traffic_depleted "
+                    f"used_bytes={used} limit_bytes={limit}"
+                ),
            )
        return {"telegram": telegram_sent, "email": email_sent}

@@ -13,6 +13,7 @@ from sqlalchemy.orm import sessionmaker

 from bot.infra.redis import redis_lock
 from bot.middlewares.i18n import JsonI18n
+from bot.services.message_audit import log_user_message_delivery
 from bot.services.panel_api_service import PanelApiService
 from bot.services.subscription_service import SubscriptionService
 from bot.services.user_email_notifications import send_user_notification_email
@@ -111,6 +112,8 @@ class TariffTrafficWorker:
        subject_key: str,
        message_text: str,
        kind: str,
+        warning_key: str,
+        audit_content: str,
    ) -> None:
        try:
            user = await user_dal.get_user_by_id(session, user_id)
@@ -131,6 +134,9 @@ class TariffTrafficWorker:
                if kind == "premium"
                else "email_traffic_warning_regular_cta"
            ),
+            session=session,
+            audit_event_type="email_traffic_warning_sent",
+            audit_content=f"{audit_content} subject_key={subject_key} warning_key={warning_key}",
        )

    async def run(self) -> None:
@@ -602,6 +608,15 @@ class TariffTrafficWorker:
                    **usage,
                )
                subject_key = "email_traffic_warning_regular_depleted_subject"
+            warning_key = (
+                "traffic_warning_regular_almost"
+                if level < 100
+                else "traffic_warning_regular_depleted"
+            )
+            audit_content = (
+                f"kind=regular warning_key={warning_key} level={level} "
+                f"used_bytes={used_val} limit_bytes={limit_val}"
+            )
            if self.bot:
                try:
                    markup = self._traffic_topup_markup(user_lang, "regular")
@@ -611,6 +626,14 @@ class TariffTrafficWorker:
                        reply_markup=markup,
                        parse_mode="HTML",
                    )
+                    await log_user_message_delivery(
+                        session,
+                        target_user_id=sub.user_id,
+                        event_type="telegram_traffic_warning_sent",
+                        channel="telegram",
+                        recipient=str(sub.user_id),
+                        content=audit_content,
+                    )
                except Exception:
                    logging.exception("Failed to send traffic warning to user %s", sub.user_id)
            await self._send_traffic_warning_email(
@@ -619,6 +642,8 @@ class TariffTrafficWorker:
                subject_key=subject_key,
                message_text=text,
                kind="regular",
+                warning_key=warning_key,
+                audit_content=audit_content,
            )
        if ratio >= 1.0 and not sub.is_throttled:
            logging.info(
@@ -1066,6 +1091,11 @@ class TariffTrafficWorker:
                servers=servers,
                **usage,
            )
+            warning_key = "traffic_warning_premium_depleted"
+            audit_content = (
+                f"kind=premium warning_key={warning_key} "
+                f"used_bytes={used_val} limit_bytes={limit_val}"
+            )
            if self.bot:
                try:
                    markup = self._traffic_topup_markup(user_lang, "premium")
@@ -1075,6 +1105,14 @@ class TariffTrafficWorker:
                        reply_markup=markup,
                        parse_mode="HTML",
                    )
+                    await log_user_message_delivery(
+                        session,
+                        target_user_id=sub.user_id,
+                        event_type="telegram_traffic_warning_sent",
+                        channel="telegram",
+                        recipient=str(sub.user_id),
+                        content=audit_content,
+                    )
                except Exception:
                    logging.exception(
                        "Failed to send premium traffic depleted warning to user %s", sub.user_id
@@ -1085,6 +1123,8 @@ class TariffTrafficWorker:
                subject_key="email_traffic_warning_premium_depleted_subject",
                message_text=text,
                kind="premium",
+                warning_key=warning_key,
+                audit_content=audit_content,
            )
            return

@@ -1134,6 +1174,11 @@ class TariffTrafficWorker:
                servers=servers,
                **usage,
            )
+            warning_key = "traffic_warning_premium_almost"
+            audit_content = (
+                f"kind=premium warning_key={warning_key} level={int(level)} "
+                f"used_bytes={used_val} limit_bytes={limit_val}"
+            )
            if self.bot:
                try:
                    markup = self._traffic_topup_markup(user_lang, "premium")
@@ -1143,6 +1188,14 @@ class TariffTrafficWorker:
                        reply_markup=markup,
                        parse_mode="HTML",
                    )
+                    await log_user_message_delivery(
+                        session,
+                        target_user_id=sub.user_id,
+                        event_type="telegram_traffic_warning_sent",
+                        channel="telegram",
+                        recipient=str(sub.user_id),
+                        content=audit_content,
+                    )
                except Exception:
                    logging.exception(
                        "Failed to send premium traffic warning to user %s", sub.user_id
@@ -1153,6 +1206,8 @@ class TariffTrafficWorker:
                subject_key="email_traffic_warning_premium_almost_subject",
                message_text=text,
                kind="premium",
+                warning_key=warning_key,
+                audit_content=audit_content,
            )

    async def _premium_node_uuids_for_tariff(self, tariff) -> list[str]:
@@ -1,9 +1,12 @@
 import logging
 from typing import Any, Optional

+from sqlalchemy.ext.asyncio import AsyncSession
+
 from bot.middlewares.i18n import JsonI18n
 from bot.services.email_auth_service import EmailAuthService
 from bot.services.email_templates import render_user_notification
+from bot.services.message_audit import log_user_message_delivery
 from config.settings import Settings


@@ -34,6 +37,9 @@ async def send_user_notification_email(
    subject_kwargs: Optional[dict[str, Any]] = None,
    heading_key: Optional[str] = None,
    intro_key: Optional[str] = None,
+    session: Optional[AsyncSession] = None,
+    audit_event_type: Optional[str] = None,
+    audit_content: Optional[str] = None,
 ) -> bool:
    if not getattr(settings, "email_auth_configured", False):
        return False
@@ -78,6 +84,15 @@ async def send_user_notification_email(
            email=recipient,
            content=content,
        )
+        if session is not None and audit_event_type:
+            await log_user_message_delivery(
+                session,
+                target_user_id=getattr(user, "user_id", None),
+                event_type=audit_event_type,
+                channel="email",
+                recipient=recipient,
+                content=audit_content or f"subject_key={subject_key}",
+            )
        return True
    except Exception:
        logging.exception("Failed to send user notification email to %s.", recipient)
@@ -0,0 +1,47 @@
+import unittest
+from datetime import datetime, timezone
+
+from bot.services import message_audit
+
+
+class MessageAuditTests(unittest.IsolatedAsyncioTestCase):
+    async def test_log_user_message_delivery_adds_targeted_log(self):
+        calls = []
+        sent_at = datetime(2026, 5, 31, tzinfo=timezone.utc)
+
+        async def fake_create(_session, payload):
+            calls.append(payload)
+
+        original = message_audit.message_log_dal.create_message_log_no_commit
+        message_audit.message_log_dal.create_message_log_no_commit = fake_create
+        self.addCleanup(
+            self._restore_create_message_log,
+            original,
+        )
+
+        await message_audit.log_user_message_delivery(
+            object(),
+            target_user_id=42,
+            event_type="telegram_traffic_warning_sent",
+            channel="telegram",
+            recipient="100500",
+            content="kind=regular level=90",
+            timestamp=sent_at,
+        )
+
+        self.assertEqual(
+            calls,
+            [
+                {
+                    "user_id": None,
+                    "event_type": "telegram_traffic_warning_sent",
+                    "content": "channel=telegram | recipient=100500 | kind=regular level=90",
+                    "is_admin_event": False,
+                    "target_user_id": 42,
+                    "timestamp": sent_at,
+                }
+            ],
+        )
+
+    def _restore_create_message_log(self, original):
+        message_audit.message_log_dal.create_message_log_no_commit = original