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

"""RDG 302→303 session flap: detect, flag event, build Problem match."""

from __future__ import annotations

from datetime import datetime, timedelta, timezone

from sqlalchemy import select
from sqlalchemy.orm import Session
from sqlalchemy.orm.attributes import flag_modified

from app.config import get_settings
from app.models import Event
from app.services.problem_rules import RULE_RDG_SESSION_FLAP, RuleMatch

RDG_SUCCESS_TYPE = "rdg.connection.success"
RDG_END_TYPES = frozenset({"rdg.connection.disconnected", "rdg.connection.failed"})


def _event_user(event: Event) -> str:
    details = event.details if isinstance(event.details, dict) else {}
    user = details.get("user")
    return str(user).strip() if user else ""


def _event_internal_ip(event: Event) -> str:
    details = event.details if isinstance(event.details, dict) else {}
    for key in ("internal_ip", "client_ip", "ip_address"):
        val = details.get(key)
        if val:
            return str(val).strip()
    return ""


def event_internal_ip(event: Event) -> str:
    return _event_internal_ip(event)


def _users_match(end_event: Event, success_event: Event) -> bool:
    return _event_user(end_event) != "" and _event_user(end_event) == _event_user(success_event)


def _internal_ips_compatible(end_event: Event, success_event: Event) -> bool:
    end_ip = _event_internal_ip(end_event)
    success_ip = _event_internal_ip(success_event)
    if not end_ip or not success_ip:
        return True
    return end_ip == success_ip


def _internal_ips_match_strict(end_event: Event, success_event: Event) -> bool:
    """Для «сессия завершена» — только при совпадении целевого ПК (оба IP заданы)."""
    end_ip = _event_internal_ip(end_event)
    success_ip = _event_internal_ip(success_event)
    if not end_ip or not success_ip:
        return False
    return end_ip == success_ip


def _as_utc(dt: datetime) -> datetime:
    if dt.tzinfo is None:
        return dt.replace(tzinfo=timezone.utc)
    return dt.astimezone(timezone.utc)


def find_rdg_success_before_end(db: Session, end_event: Event) -> Event | None:
    if end_event.type not in RDG_END_TYPES:
        return None
    settings = get_settings()
    min_sec = settings.sac_rdg_flap_window_min_sec
    max_sec = settings.sac_rdg_flap_window_max_sec
    end_at = _as_utc(end_event.occurred_at)

    window_start = end_at - timedelta(seconds=max_sec)
    window_end = end_at - timedelta(seconds=min_sec)

    candidates = db.scalars(
        select(Event)
        .where(
            Event.host_id == end_event.host_id,
            Event.type == RDG_SUCCESS_TYPE,
            Event.occurred_at >= window_start,
            Event.occurred_at <= window_end,
            Event.id != end_event.id,
        )
        .order_by(Event.occurred_at.desc())
    ).all()

    for prior in candidates:
        if not _users_match(end_event, prior):
            continue
        if not _internal_ips_compatible(end_event, prior):
            continue
        prior_at = _as_utc(prior.occurred_at)
        delta = (end_at - prior_at).total_seconds()
        if min_sec <= delta <= max_sec:
            return prior
    return None


def mark_rdg_flap(event: Event, *, pair_event: Event) -> None:
    details = dict(event.details) if isinstance(event.details, dict) else {}
    details["rdg_flap"] = True
    details["rdg_flap_pair_event_id"] = pair_event.id
    event.details = details
    flag_modified(event, "details")


def evaluate_rdg_session_flap(db: Session, event: Event) -> RuleMatch | None:
    prior = find_rdg_success_before_end(db, event)
    if prior is None:
        return None

    mark_rdg_flap(event, pair_event=prior)
    user = _event_user(event)
    internal_ip = _event_internal_ip(event)
    ip_note = f", client {internal_ip}" if internal_ip else ""
    delta_sec = int((_as_utc(event.occurred_at) - _as_utc(prior.occurred_at)).total_seconds())
    return RuleMatch(
        rule_id=RULE_RDG_SESSION_FLAP,
        correlation_type="rdg.session.flap",
        fingerprint_suffix=f"u{user}:ip{internal_ip or 'any'}",
        title=f"RDG session flap: {user}",
        summary=(
            f"302→303 за {delta_sec} с "
            f"({user}{ip_note}). Возможна зависшая сессия на ПК пользователя — qwinsta/logoff."
        ),
        severity="warning",
    )


def event_has_rdg_flap(event: Event) -> bool:
    details = event.details if isinstance(event.details, dict) else {}
    return details.get("rdg_flap") is True


def _stored_flap_pair_id(event: Event) -> int | None:
    details = event.details if isinstance(event.details, dict) else {}
    raw = details.get("rdg_flap_pair_event_id")
    if raw is None:
        return None
    try:
        return int(raw)
    except (TypeError, ValueError):
        return None


def find_rdg_end_after_success(db: Session, success_event: Event) -> Event | None:
    """303 с rdg_flap, у которого пара — этот 302 (или вычисляется по окну)."""
    if success_event.type != RDG_SUCCESS_TYPE:
        return None
    settings = get_settings()
    min_sec = settings.sac_rdg_flap_window_min_sec
    max_sec = settings.sac_rdg_flap_window_max_sec
    start_at = _as_utc(success_event.occurred_at)
    window_start = start_at + timedelta(seconds=min_sec)
    window_end = start_at + timedelta(seconds=max_sec)

    candidates = db.scalars(
        select(Event)
        .where(
            Event.host_id == success_event.host_id,
            Event.type.in_(RDG_END_TYPES),
            Event.occurred_at >= window_start,
            Event.occurred_at <= window_end,
            Event.id != success_event.id,
        )
        .order_by(Event.occurred_at.asc())
    ).all()

    for end in candidates:
        if not _users_match(end, success_event):
            continue
        if not _internal_ips_compatible(end, success_event):
            continue
        pair_id = _stored_flap_pair_id(end)
        if pair_id == success_event.id:
            return end
        prior = find_rdg_success_before_end(db, end)
        if prior is not None and prior.id == success_event.id:
            return end
    return None


def find_normal_rdg_end_after_success(db: Session, success_event: Event) -> Event | None:
    """303 после 302 с паузой больше flap-окна — штатное завершение сессии."""
    if success_event.type != RDG_SUCCESS_TYPE:
        return None
    settings = get_settings()
    max_sec = settings.sac_rdg_flap_window_max_sec
    start_at = _as_utc(success_event.occurred_at)
    after_flap = start_at + timedelta(seconds=max_sec)

    candidates = db.scalars(
        select(Event)
        .where(
            Event.host_id == success_event.host_id,
            Event.type.in_(RDG_END_TYPES),
            Event.occurred_at > after_flap,
            Event.id != success_event.id,
        )
        .order_by(Event.occurred_at.asc())
    ).all()

    for end in candidates:
        if not _users_match(end, success_event):
            continue
        if not _internal_ips_match_strict(end, success_event):
            continue
        return end
    return None


def resolve_rdg_qwinsta_enabled(db: Session | None, event: Event) -> bool:
    """Кнопка qwinsta/logoff только на 302, пока сессия может быть активна (или RDG flap)."""
    if event.type in RDG_END_TYPES:
        return False
    if event.type != RDG_SUCCESS_TYPE:
        return False
    if not _event_internal_ip(event):
        return False
    if db is None:
        return True
    if find_rdg_end_after_success(db, event) is not None:
        return True
    if find_normal_rdg_end_after_success(db, event) is not None:
        return False
    return True


def resolve_rdg_flap_summary(
    db: Session, event: Event
) -> tuple[bool, int | None, int | None]:
    """
    (rdg_flap, pair_event_id, qwinsta_event_id).
    qwinsta_event_id — id события 302 для qwinsta (на 303 кнопку не показываем).
    """
    if event_has_rdg_flap(event):
        pair_id = _stored_flap_pair_id(event)
        if event.type == RDG_SUCCESS_TYPE:
            return True, pair_id, event.id
        return True, pair_id, pair_id

    if event.type in RDG_END_TYPES:
        prior = find_rdg_success_before_end(db, event)
        if prior is not None:
            return True, prior.id, prior.id

    if event.type == RDG_SUCCESS_TYPE:
        end = find_rdg_end_after_success(db, event)
        if end is not None:
            return True, end.id, event.id

    return False, None, None