security-alert-center/backend/app/services/daily_report.py

"""SAC-generated daily reports from ingested events (F-NOT-05 / notif-32)."""

from __future__ import annotations

import html
import logging
import uuid
from collections import Counter
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Any
from zoneinfo import ZoneInfo

from sqlalchemy import select
from sqlalchemy.orm import Session

from app.config import Settings, get_settings
from app.models import Event, Host
from app.services.ingest import ingest_event
from app.services.notify_dispatch import notify_daily_report

logger = logging.getLogger(__name__)

REPORT_TYPES = {
    "ssh-monitor": "report.daily.ssh",
    "rdp-login-monitor": "report.daily.rdp",
}

SSH_SUCCESS = "ssh.login.success"
SSH_FAILED = "ssh.login.failed"
SSH_SUDO = "privilege.sudo.command"
RDP_SUCCESS = "rdp.login.success"
RDP_FAILED = "rdp.login.failed"


@dataclass(frozen=True)
class DailyReportResult:
    host_id: int
    hostname: str
    report_type: str
    event_id: str
    created: bool
    skipped_reason: str | None = None


def _now_in_tz(settings: Settings) -> datetime:
    tz = ZoneInfo(settings.sac_daily_report_timezone.strip() or "Europe/Moscow")
    return datetime.now(timezone.utc).astimezone(tz)


def _report_type_for_host(host: Host) -> str | None:
    return REPORT_TYPES.get((host.product or "").strip())


def _day_start_in_tz(settings: Settings, ref: datetime | None = None) -> datetime:
    local = ref or _now_in_tz(settings)
    start = local.replace(hour=0, minute=0, second=0, microsecond=0)
    return start.astimezone(timezone.utc)


def host_has_report_today(db: Session, host_id: int, report_type: str, settings: Settings) -> bool:
    since = _day_start_in_tz(settings)
    row = db.scalar(
        select(Event.id)
        .where(
            Event.host_id == host_id,
            Event.type == report_type,
            Event.received_at >= since,
        )
        .limit(1)
    )
    return row is not None


def _ip_from_details(details: dict[str, Any] | None) -> str:
    if not details:
        return ""
    for key in ("source_ip", "ip_address", "ip"):
        val = details.get(key)
        if val is not None and str(val).strip() not in ("", "-"):
            return str(val).strip()
    return ""


def _user_from_details(details: dict[str, Any] | None) -> str:
    if not details:
        return ""
    for key in ("user", "username"):
        val = details.get(key)
        if val is not None and str(val).strip():
            return str(val).strip()
    return ""


def _aggregate_ssh(events: list[Event]) -> dict[str, Any]:
    ok = failed = sudo = 0
    failed_ips: Counter[str] = Counter()
    for ev in events:
        if ev.type == SSH_SUCCESS:
            ok += 1
        elif ev.type == SSH_FAILED:
            failed += 1
            ip = _ip_from_details(ev.details if isinstance(ev.details, dict) else None)
            if ip:
                failed_ips[ip] += 1
        elif ev.type == SSH_SUDO:
            sudo += 1
    top_ips = [f"{ip} — {cnt}" for ip, cnt in failed_ips.most_common(5)]
    return {
        "successful_ssh": ok,
        "failed_ssh": failed,
        "sudo_commands": sudo,
        "active_bans": sum(1 for e in events if e.type == "ssh.ip.banned"),
        "top_failed_ips": top_ips,
    }


def _aggregate_rdp(events: list[Event]) -> dict[str, Any]:
    ok = failed = 0
    failed_ips: Counter[str] = Counter()
    users: set[str] = set()
    for ev in events:
        details = ev.details if isinstance(ev.details, dict) else None
        if ev.type == RDP_SUCCESS:
            ok += 1
            u = _user_from_details(details)
            if u:
                users.add(u)
        elif ev.type == RDP_FAILED:
            failed += 1
            ip = _ip_from_details(details)
            if ip:
                failed_ips[ip] += 1
            u = _user_from_details(details)
            if u:
                users.add(u)
    return {
        "rdp_success": ok,
        "rdp_failed": failed,
        "unique_users": sorted(users),
        "top_failed_ips": [f"{ip} — {cnt}" for ip, cnt in failed_ips.most_common(5)],
    }


def _build_report_body_ssh(host: Host, stats: dict[str, Any], when_local: datetime) -> str:
    label = host.display_name or host.hostname
    lines = [
        "📊 ЕЖЕДНЕВНЫЙ ОТЧЁТ SSH (SAC)",
        f"🖥️ Хост: {label}",
        f"🕐 Период: последние 24 ч (сводка на {when_local.strftime('%d.%m.%Y %H:%M')})",
        "",
        "📈 СТАТИСТИКА ИЗ INGEST:",
        f"✅ Успешных SSH: {stats['successful_ssh']}",
        f"❌ Неудачных SSH: {stats['failed_ssh']}",
        f"⚠️ Sudo: {stats['sudo_commands']}",
        f"🚫 Событий ban: {stats['active_bans']}",
        "",
        "🧾 ТОП IP (неудачные входы):",
    ]
    top = stats.get("top_failed_ips") or []
    if top:
        lines.extend(f" • {x}" for x in top)
    else:
        lines.append(" (нет данных)")
    lines.append("")
    lines.append("Источник: Security Alert Center (агрегация событий SAC).")
    return "\n".join(lines)


def _build_report_body_rdp(host: Host, stats: dict[str, Any], when_local: datetime) -> str:
    label = host.display_name or host.hostname
    users = stats.get("unique_users") or []
    lines = [
        "📊 ЕЖЕДНЕВНЫЙ ОТЧЁТ RDP (SAC)",
        f"🖥️ Сервер: {label}",
        f"🕐 Период: последние 24 ч (сводка на {when_local.strftime('%d.%m.%Y %H:%M')})",
        "",
        f"✅ Успешных RDP (ingest): {stats['rdp_success']}",
        f"❌ Неудачных RDP: {stats['rdp_failed']}",
        f"👥 Уникальных пользователей в событиях: {len(users)}",
        "",
        "🧾 ТОП IP (неудачные входы):",
    ]
    top = stats.get("top_failed_ips") or []
    if top:
        lines.extend(f" • {x}" for x in top)
    else:
        lines.append(" (нет данных)")
    if users:
        lines.append("")
        lines.append("Уникальные логины:")
        for u in users[:20]:
            lines.append(f" • {u}")
    lines.append("")
    lines.append("Источник: SAC ingest (активные сессии quser — только на агенте).")
    return "\n".join(lines)


def _details_from_body(body: str, stats: dict[str, Any]) -> dict[str, Any]:
    escaped = html.escape(body)
    # UI (Vue) допускает div; для единообразия — обёртка без br (Telegram читает через sanitize)
    report_html = f'<div class="agent-report">{escaped.replace(chr(10), "<br>")}</div>'
    return {
        "stats": stats,
        "report_body": body,
        "report_format": "plain",
        "report_html": report_html,
        "generated_by": "sac",
    }


def _events_last_24h(db: Session, host_id: int) -> list[Event]:
    since = datetime.now(timezone.utc) - timedelta(hours=24)
    return list(
        db.scalars(
            select(Event)
            .where(Event.host_id == host_id, Event.occurred_at >= since)
            .order_by(Event.occurred_at.desc())
        ).all()
    )


def generate_daily_report_for_host(db: Session, host: Host, settings: Settings | None = None) -> DailyReportResult | None:
    cfg = settings or get_settings()
    report_type = _report_type_for_host(host)
    if report_type is None:
        return None

    if cfg.sac_daily_report_skip_if_agent_sent and host_has_report_today(db, host.id, report_type, cfg):
        return DailyReportResult(
            host_id=host.id,
            hostname=host.hostname,
            report_type=report_type,
            event_id="",
            created=False,
            skipped_reason="agent_report_exists_today",
        )

    events = _events_last_24h(db, host.id)
    if not events and cfg.sac_daily_report_require_activity:
        return DailyReportResult(
            host_id=host.id,
            hostname=host.hostname,
            report_type=report_type,
            event_id="",
            created=False,
            skipped_reason="no_events_24h",
        )

    when_local = _now_in_tz(cfg)
    if report_type == "report.daily.ssh":
        stats = _aggregate_ssh(events)
        body = _build_report_body_ssh(host, stats, when_local)
        title = "Ежедневный отчёт SSH (SAC)"
        summary = (
            f"SSH 24ч: успех {stats['successful_ssh']}, неудач {stats['failed_ssh']}, "
            f"sudo {stats['sudo_commands']}"
        )
    else:
        stats = _aggregate_rdp(events)
        body = _build_report_body_rdp(host, stats, when_local)
        title = "Ежедневный отчёт RDP (SAC)"
        summary = f"RDP 24ч: успех {stats['rdp_success']}, неудач {stats['rdp_failed']}"

    event_id = str(uuid.uuid4())
    day_key = when_local.strftime("%Y-%m-%d")
    payload = {
        "schema_version": "1.0",
        "event_id": event_id,
        "occurred_at": datetime.now(timezone.utc).isoformat(),
        "source": {
            "product": host.product or "unknown",
            "product_version": host.product_version or "sac",
        },
        "host": {
            "hostname": host.hostname,
            "os_family": host.os_family or "linux",
            "display_name": host.display_name,
        },
        "category": "report",
        "type": report_type,
        "severity": "info",
        "title": title,
        "summary": summary,
        "details": _details_from_body(body, stats),
        "dedup_key": f"sac|{host.id}|{report_type}|{day_key}",
    }

    event, created = ingest_event(db, payload)
    if created:
        notify_daily_report(event, db=db)
    return DailyReportResult(
        host_id=host.id,
        hostname=host.hostname,
        report_type=report_type,
        event_id=event.event_id,
        created=created,
        skipped_reason=None if created else "duplicate_event_id",
    )


def run_daily_reports(db: Session, settings: Settings | None = None, *, force: bool = False) -> list[DailyReportResult]:
    cfg = settings or get_settings()
    if not cfg.sac_daily_report_enabled:
        logger.info("daily report disabled (SAC_DAILY_REPORT_ENABLED=false)")
        return []

    local_now = _now_in_tz(cfg)
    if not force and local_now.hour < cfg.sac_daily_report_hour:
        logger.info(
            "daily report skipped: local hour %s < SAC_DAILY_REPORT_HOUR=%s",
            local_now.hour,
            cfg.sac_daily_report_hour,
        )
        return []

    hosts = db.scalars(select(Host).order_by(Host.hostname)).all()
    results: list[DailyReportResult] = []
    for host in hosts:
        try:
            res = generate_daily_report_for_host(db, host, cfg)
            if res is not None:
                results.append(res)
        except Exception:
            logger.exception("daily report failed host_id=%s hostname=%s", host.id, host.hostname)
    db.commit()
    created_n = sum(1 for r in results if r.created)
    logger.info("daily report finished hosts=%s created=%s", len(results), created_n)
    return results