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

"""Generic JSON webhook notifications."""

from __future__ import annotations

import logging
from typing import Any

import httpx
from sqlalchemy.orm import Session

from app.models import Event, Problem
from app.services.notification_severity import severity_meets_minimum
from app.services.notification_settings import WebhookConfig, get_effective_webhook_config

logger = logging.getLogger(__name__)


class WebhookNotConfiguredError(Exception):
    """Webhook disabled or missing URL."""


class WebhookSendError(Exception):
    def __init__(self, message: str, *, status_code: int | None = None) -> None:
        super().__init__(message)
        self.status_code = status_code


def _host_payload(entity: Event | Problem) -> dict[str, Any]:
    host = entity.host
    if host is None:
        return {"hostname": "unknown", "display_name": None}
    return {"hostname": host.hostname, "display_name": host.display_name}


def build_event_payload(event: Event) -> dict[str, Any]:
    return {
        "kind": "event",
        "event_id": event.event_id,
        "type": event.type,
        "severity": event.severity,
        "category": event.category,
        "title": event.title,
        "summary": event.summary,
        "host": _host_payload(event),
        "occurred_at": event.occurred_at.isoformat() if event.occurred_at else None,
    }


def build_problem_payload(problem: Problem, event: Event | None = None) -> dict[str, Any]:
    payload: dict[str, Any] = {
        "kind": "problem",
        "problem_id": problem.id,
        "rule_id": problem.rule_id,
        "severity": problem.severity,
        "status": problem.status,
        "title": problem.title,
        "summary": problem.summary,
        "host": _host_payload(problem),
        "opened_at": problem.opened_at.isoformat() if problem.opened_at else None,
    }
    if event is not None:
        payload["trigger_event"] = {
            "event_id": event.event_id,
            "type": event.type,
            "severity": event.severity,
        }
    return payload


def send_webhook_payload(payload: dict[str, Any], *, config: WebhookConfig | None = None, force: bool = False) -> None:
    cfg = config or get_effective_webhook_config()
    if not force and not cfg.active:
        return
    if not cfg.url:
        if force:
            raise WebhookNotConfiguredError("Webhook: не задан URL")
        return

    headers = {"Content-Type": "application/json", "User-Agent": "SecurityAlertCenter/1.0"}
    if cfg.secret_header and cfg.secret:
        headers[cfg.secret_header] = cfg.secret

    try:
        with httpx.Client(timeout=10.0) as client:
            response = client.post(cfg.url, json=payload, headers=headers)
            response.raise_for_status()
    except httpx.HTTPStatusError as exc:
        detail = exc.response.text[:200] if exc.response is not None else str(exc)
        raise WebhookSendError(
            f"Webhook HTTP {exc.response.status_code}: {detail}",
            status_code=exc.response.status_code,
        ) from exc
    except Exception as exc:
        logger.exception("webhook send failed")
        raise WebhookSendError(str(exc)) from exc


def send_webhook_test_message(*, config: WebhookConfig | None = None) -> None:
    cfg = config or get_effective_webhook_config()
    if not cfg.enabled:
        raise WebhookNotConfiguredError("Webhook отключён (enabled=false)")
    if not cfg.configured:
        raise WebhookNotConfiguredError("Не задан webhook URL")
    send_webhook_payload(
        {
            "kind": "test",
            "message": "SAC webhook test",
            "source": "security-alert-center",
        },
        config=cfg,
        force=True,
    )


def notify_event(event: Event, *, db: Session | None = None, apply_policy_gate: bool = True) -> None:
    cfg = get_effective_webhook_config(db)
    if not cfg.active:
        return
    if apply_policy_gate and not severity_meets_minimum(event.severity, cfg.min_severity):
        return
    try:
        send_webhook_payload(build_event_payload(event), config=cfg)
    except WebhookSendError:
        logger.exception("notify_event webhook failed event_id=%s", event.event_id)


def notify_problem(problem: Problem, event: Event | None = None, *, db: Session | None = None, apply_policy_gate: bool = True) -> None:
    cfg = get_effective_webhook_config(db)
    if not cfg.active:
        return
    if apply_policy_gate and not severity_meets_minimum(problem.severity, cfg.min_severity):
        return
    try:
        send_webhook_payload(build_problem_payload(problem, event), config=cfg)
    except WebhookSendError:
        logger.exception("notify_problem webhook failed problem_id=%s", problem.id)