"""Dispatch ingest notifications per global policy (severity → channels).""" import logging from sqlalchemy.orm import Session from app.models import Event, Host, Problem from app.services import email_notify, mobile_notify, telegram_notify, webhook_notify from app.services.notification_cooldown import should_notify_event, should_notify_problem from app.services.notification_policy import get_effective_notification_policy from app.services.event_type_visibility import event_type_notifications_enabled from app.services.host_health import HEARTBEAT_TYPE from app.services.notification_severity import severity_meets_minimum logger = logging.getLogger(__name__) LIFECYCLE_EVENT_TYPE = "agent.lifecycle" PRIVILEGE_SUDO_TYPE = "privilege.sudo.command" SUDO_MAINTENANCE_MARKERS = ( "update_ssh_monitor.sh", "update_via_sac", "update_script.log", "/opt/scripts/update", "agent-update-in-progress", ) DAILY_REPORT_EVENT_TYPES = frozenset({"report.daily.ssh", "report.daily.rdp"}) AUTH_LOGIN_SUCCESS_TYPES = frozenset({"rdp.login.success", "ssh.login.success"}) RDG_CONNECTION_TYPES = frozenset({ "rdg.connection.success", "rdg.connection.disconnected", "rdg.connection.failed", }) def _event_telegram_via_agent(event: Event) -> bool: details = event.details if isinstance(event.details, dict) else {} via = str(details.get("telegram_via") or "").strip().lower() return via == "agent" def _skip_notifications_for_hidden_event(event: Event, db: Session | None) -> bool: if event_type_notifications_enabled(event.type, db): return False logger.info( "notify skipped hidden event type=%s event_id=%s", event.type, event.event_id, ) return True def _dispatch_event_channels(event: Event, *, db: Session | None, policy) -> None: if policy.use_telegram: telegram_notify.notify_event(event, db=db, apply_policy_gate=False) if policy.use_webhook: webhook_notify.notify_event(event, db=db, apply_policy_gate=False) if policy.use_email: email_notify.notify_event(event, db=db, apply_policy_gate=False) if policy.use_mobile: mobile_notify.notify_event(event, db=db, apply_policy_gate=False) def _dispatch_problem_channels(problem: Problem, event: Event | None, *, db: Session | None, policy) -> None: if policy.use_telegram: telegram_notify.notify_problem(problem, event, db=db, apply_policy_gate=False) if policy.use_webhook: webhook_notify.notify_problem(problem, event, db=db, apply_policy_gate=False) if policy.use_email: email_notify.notify_problem(problem, event, db=db, apply_policy_gate=False) if policy.use_mobile: mobile_notify.notify_problem(problem, event, db=db, apply_policy_gate=False) def notify_event(event: Event, *, db: Session | None = None) -> None: # Heartbeat — только для UI/статуса хоста, не для Telegram/email/push. if event.type == HEARTBEAT_TYPE: return if _should_suppress_sudo_notify(event, db=db): return if _skip_notifications_for_hidden_event(event, db): return policy = get_effective_notification_policy(db) if not severity_meets_minimum(event.severity, policy.min_severity): return if not should_notify_event(event, db): return _dispatch_event_channels(event, db=db, policy=policy) def notify_problem(problem: Problem, event: Event | None = None, *, db: Session | None = None) -> None: policy = get_effective_notification_policy(db) if not severity_meets_minimum(problem.severity, policy.min_severity): return if not should_notify_problem(problem, db): return _dispatch_problem_channels(problem, event, db=db, policy=policy) def _dispatch_lifecycle_channels(event: Event, *, db: Session | None, policy) -> None: skip_telegram = _event_telegram_via_agent(event) if policy.use_telegram and not skip_telegram: telegram_notify.notify_event(event, db=db, apply_policy_gate=False) if policy.use_webhook: webhook_notify.notify_event(event, db=db, apply_policy_gate=False) if policy.use_email: email_notify.notify_event(event, db=db, apply_policy_gate=False) if policy.use_mobile: mobile_notify.notify_event(event, db=db, apply_policy_gate=False) def notify_daily_report(event: Event, *, db: Session | None = None) -> None: """Оповещение по суточному отчёту (severity=info, вне порога policy). При UseSAC=dual агент шлёт TG сам (telegram_via=agent) — SAC не дублирует. """ if _skip_notifications_for_hidden_event(event, db): return policy = get_effective_notification_policy(db) if not should_notify_event(event, db): return _dispatch_lifecycle_channels(event, db=db, policy=policy) def _host_sac_update_running(event: Event, db: Session | None) -> bool: """Пока SAC выполняет agent-update на хосте — не слать шумные TG.""" if db is None or not event.host_id: return False host = db.get(Host, event.host_id) if host is None: return False if (host.agent_update_state or "").strip().lower() == "running": logger.info( "notify skipped (host update running) type=%s host_id=%s event_id=%s", event.type, host.id, event.event_id, ) return True return False def _sudo_event_is_agent_maintenance(event: Event) -> bool: details = event.details if isinstance(event.details, dict) else {} cmd = str(details.get("command") or "").strip() if not cmd: cmd = str(event.summary or "").strip() text = cmd.casefold() return any(marker in text for marker in SUDO_MAINTENANCE_MARKERS) def _should_suppress_sudo_notify(event: Event, *, db: Session | None) -> bool: if event.type != PRIVILEGE_SUDO_TYPE: return False if _host_sac_update_running(event, db): return True if _sudo_event_is_agent_maintenance(event): logger.info( "notify sudo skipped maintenance command event_id=%s host_id=%s", event.event_id, event.host_id, ) return True return False def _lifecycle_suppress_notifications(event: Event, *, db: Session | None = None) -> bool: """Не слать TG при штатном SAC/cron update (lifecycle с trigger deploy_recycle).""" if _host_sac_update_running(event, db): return True details = event.details if isinstance(event.details, dict) else {} trigger = str(details.get("trigger") or "").strip().lower() if trigger in ("deploy_recycle", "sac_update", "agent_update"): logger.info( "notify lifecycle skipped trigger=%s event_id=%s", trigger, event.event_id, ) return True return False def notify_lifecycle(event: Event, *, db: Session | None = None) -> None: """Старт/стоп/reload агента — всегда в каналы SAC (кроме TG, если telegram_via=agent).""" if _lifecycle_suppress_notifications(event, db=db): return if _skip_notifications_for_hidden_event(event, db): return policy = get_effective_notification_policy(db) if not should_notify_event(event, db): return _dispatch_lifecycle_channels(event, db=db, policy=policy) def notify_auth_login(event: Event, *, db: Session | None = None) -> None: """Успешный удалённый вход RDP/SSH — всегда в Telegram SAC (info вне min_severity). При UseSAC=dual агент шлёт TG сам (telegram_via=agent) — SAC не дублирует. """ if _skip_notifications_for_hidden_event(event, db): return policy = get_effective_notification_policy(db) if not should_notify_event(event, db): return _dispatch_lifecycle_channels(event, db=db, policy=policy) def notify_rdg_connection(event: Event, *, db: Session | None = None) -> None: """RD Gateway 302/303 — ingest всегда; Telegram SAC вне min_severity (как auth login).""" if _skip_notifications_for_hidden_event(event, db): return policy = get_effective_notification_policy(db) if not should_notify_event(event, db): return _dispatch_lifecycle_channels(event, db=db, policy=policy) def schedule_notify_daily_report(event_db_id: int) -> None: """Отложенное оповещение по суточному отчёту (после commit ingest, вне горячего POST).""" _schedule_deferred_event_notify(event_db_id, notify_daily_report, label="daily report") def schedule_notify_lifecycle(event_db_id: int) -> None: """Отложенное lifecycle-оповещение (после commit ingest).""" _schedule_deferred_event_notify(event_db_id, notify_lifecycle, label="lifecycle") def schedule_notify_auth_login(event_db_id: int) -> None: """Отложенное оповещение об успешном RDP/SSH входе (после commit ingest).""" _schedule_deferred_event_notify(event_db_id, notify_auth_login, label="auth login") def _schedule_deferred_event_notify(event_db_id: int, handler, *, label: str) -> None: from app.database import SessionLocal db = SessionLocal() try: event = db.get(Event, event_db_id) if event is None: logger.warning("deferred %s notify: event id=%s not found", label, event_db_id) return handler(event, db=db) except Exception: logger.exception("deferred %s notify failed event_db_id=%s", label, event_db_id) finally: db.close()