83 lines
1.8 KiB
Python
83 lines
1.8 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
|
|
from .normalization import canonical_value
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class LogEvent:
|
|
raw: str
|
|
fields: dict[str, str]
|
|
|
|
@property
|
|
def src_ip(self) -> str | None:
|
|
return canonical_value(self.fields, "srcip") or None
|
|
|
|
@property
|
|
def dst_ip(self) -> str | None:
|
|
return canonical_value(self.fields, "dstip") or None
|
|
|
|
@property
|
|
def subtype(self) -> str:
|
|
return canonical_value(self.fields, "subtype").lower()
|
|
|
|
@property
|
|
def action(self) -> str:
|
|
return canonical_value(self.fields, "action").lower()
|
|
|
|
@property
|
|
def severity(self) -> str:
|
|
return canonical_value(self.fields, "severity").lower()
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PolicyRule:
|
|
policy_id: str
|
|
settings: dict[str, list[str]] = field(default_factory=dict)
|
|
|
|
def values(self, key: str) -> list[str]:
|
|
return self.settings.get(key, [])
|
|
|
|
def first(self, key: str, default: str = "") -> str:
|
|
values = self.values(key)
|
|
return values[0] if values else default
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Finding:
|
|
severity: str
|
|
title: str
|
|
detail: str
|
|
reference: str | None = None
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class BlockCandidate:
|
|
src_ip: str
|
|
score: int
|
|
reasons: list[str]
|
|
sample_events: list[LogEvent]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AnomalyFinding:
|
|
subject: str
|
|
score: int
|
|
severity: str
|
|
confidence: str
|
|
reasons: list[str]
|
|
evidence: dict[str, int | str | float]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Recommendation:
|
|
subject: str
|
|
score: int
|
|
severity: str
|
|
title: str
|
|
recommendation: str
|
|
reasons: list[str]
|
|
related_policy_ids: list[str] = field(default_factory=list)
|
|
related_services: list[str] = field(default_factory=list)
|