84 lines
2.8 KiB
Python
84 lines
2.8 KiB
Python
from datetime import UTC, datetime
|
|
from typing import Literal
|
|
|
|
from pydantic import BaseModel, Field, field_validator, model_validator
|
|
|
|
Category = Literal["system", "service", "network", "authentication", "application", "crash", "other"]
|
|
ParseStatus = Literal["complete", "partial", "failed"]
|
|
TimestampSource = Literal["event", "record", "missing", "invalid"]
|
|
|
|
|
|
class NormalizedEvent(BaseModel):
|
|
id: str = Field(max_length=128)
|
|
timestamp: datetime | None = None
|
|
timestamp_source: TimestampSource = "event"
|
|
timestamp_valid: bool = True
|
|
provider: str = Field(default="", max_length=256)
|
|
event_id: int | None = Field(default=None, ge=0, le=4_294_967_295)
|
|
level: str = Field(default="information", max_length=32)
|
|
channel: str = Field(default="", max_length=256)
|
|
computer: str = Field(default="", max_length=256)
|
|
category: Category = "other"
|
|
summary: str = Field(default="", max_length=512)
|
|
fields: dict[str, str] = Field(default_factory=dict, max_length=200)
|
|
|
|
@field_validator("timestamp")
|
|
@classmethod
|
|
def normalize_naive_timestamp(cls, value: datetime | None) -> datetime | None:
|
|
if value is not None and value.tzinfo is None:
|
|
return value.replace(tzinfo=UTC)
|
|
return value
|
|
|
|
@model_validator(mode="after")
|
|
def validate_timestamp_metadata(self) -> "NormalizedEvent":
|
|
if self.timestamp is None:
|
|
self.timestamp_valid = False
|
|
if self.timestamp_source in {"event", "record"}:
|
|
self.timestamp_source = "missing"
|
|
elif self.timestamp_source in {"missing", "invalid"}:
|
|
self.timestamp_valid = False
|
|
else:
|
|
self.timestamp_valid = True
|
|
return self
|
|
|
|
|
|
class IncidentCluster(BaseModel):
|
|
id: str = Field(max_length=128)
|
|
rule_id: str = Field(max_length=128)
|
|
start: datetime | None = None
|
|
end: datetime | None = None
|
|
title: str = Field(max_length=256)
|
|
category: Category
|
|
severity: str = Field(max_length=32)
|
|
reasons: list[str] = Field(max_length=8)
|
|
evidence_event_ids: list[str]
|
|
original_count: int = Field(ge=1)
|
|
compressed_count: int = Field(ge=1)
|
|
|
|
|
|
class ParseIssue(BaseModel):
|
|
code: str = Field(max_length=64)
|
|
record_id: int | None = None
|
|
message: str = Field(max_length=256)
|
|
skipped_record: bool = False
|
|
|
|
|
|
class ParseStats(BaseModel):
|
|
parser: str = "evtx-rs"
|
|
status: ParseStatus
|
|
parsed_events: int = Field(ge=0)
|
|
records_seen: int = Field(ge=0)
|
|
skipped_records: int = Field(ge=0)
|
|
issue_count: int = Field(ge=0)
|
|
truncated: bool
|
|
truncation_reason: str | None = Field(default=None, max_length=64)
|
|
duration_ms: int = Field(ge=0)
|
|
issues: list[ParseIssue] = Field(default_factory=list)
|
|
|
|
|
|
class AnalysisResponse(BaseModel):
|
|
source: str
|
|
events: list[NormalizedEvent]
|
|
incidents: list[IncidentCluster]
|
|
parse: ParseStats | None = None
|