src/common/models.py - master - azathd/mutiagent

src/common/models.py
106 строк2 KB
Your Name
init
15 май 2026, 17:47
15 май 2026, 17:47359c81e
"""Pydantic event envelope and domain enums."""

from __future__ import annotations

from datetime import datetime
from enum import Enum
from typing import Any
from uuid import UUID, uuid4

from pydantic import BaseModel, Field


class Envelope(BaseModel):
    """Envelope for all bus messages."""

    event_id: UUID = Field(default_factory=uuid4)
    occurred_at: datetime
    schema_version: int = 1
    producer: str
    trace_id: str | None = None
    payload: dict[str, Any]


class Severity(str, Enum):
    """Alert / incident severity tiers."""

    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"


class ActionType(str, Enum):
    """High-level action kinds for Executor / Orchestrator."""

    USER_SUSPEND = "user.suspend"
    USER_UNSUSPEND = "user.unsuspend"
    USER_SET_LIMIT = "user.set_limit"
    USER_DELETE = "user.delete"
    FIREWALL_BLOCK_IP = "firewall.block_ip"
    NOTIFY = "notify"


class ReportPeriod(str, Enum):
    """Reporting cadence."""

    DAY = "day"
    WEEK = "week"
    MONTH = "month"
    CUSTOM = "custom"


class IncidentStatus(str, Enum):
    NEW = "new"
    TRIAGING = "triaging"
    ACTION_PROPOSED = "action_proposed"
    AWAITING_APPROVAL = "awaiting_approval"
    EXECUTING = "executing"
    RESOLVED = "resolved"
    ROLLED_BACK = "rolled_back"


class NormalizedEvent(BaseModel):
    """Parsed Xray access/error line."""

    occurred_at: datetime
    user_email: str | None = None
    source_ip: str | None = None
    dest_host: str | None = None
    dest_port: int | None = None
    protocol: str | None = None
    bytes_up: int = 0
    bytes_down: int = 0
    country: str | None = None
    asn: str | None = None
    inbound_tag: str | None = None
    outbound_tag: str | None = None
    log_type: str = "access"
    raw: str | None = None


class SecurityAlert(BaseModel):
    rule_id: str
    severity: Severity
    user_email: str | None = None
    source_ip: str | None = None
    message: str
    context_events: list[dict[str, Any]] = Field(default_factory=list)
    action_hint: str | None = None


class ProposedAction(BaseModel):
    action_id: str
    action_type: ActionType
    target: str
    severity: Severity = Severity.LOW
    rationale: str = ""
    payload: dict[str, Any] = Field(default_factory=dict)


class TrafficRecommendation(BaseModel):
    outbound_tag: str
    latency_p95_ms: float
    loss_pct: float
    recommendation: str
    confidence: float = 0.0