167 lines
4.9 KiB
Python
167 lines
4.9 KiB
Python
"""Shared finding taxonomy for IoT PreTester 4.6 Demo.
|
|
|
|
The module deliberately has no Django or third-party dependencies so the same
|
|
model works in the web backend, CLI, and unit tests.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import enum
|
|
import hashlib
|
|
from dataclasses import dataclass, field, asdict
|
|
from typing import Any, Optional
|
|
|
|
|
|
class Severity(enum.Enum):
|
|
"""Potential impact if the finding is valid."""
|
|
|
|
CRITICAL = "critical"
|
|
HIGH = "high"
|
|
MEDIUM = "medium"
|
|
LOW = "low"
|
|
INFO = "info"
|
|
|
|
@property
|
|
def rank(self) -> int:
|
|
return _SEVERITY_RANK[self]
|
|
|
|
|
|
class Confidence(enum.Enum):
|
|
"""Confidence in the evidence extracted from the capture."""
|
|
|
|
HIGH = "high"
|
|
MEDIUM = "medium"
|
|
LOW = "low"
|
|
|
|
@property
|
|
def rank(self) -> int:
|
|
return _CONFIDENCE_RANK[self]
|
|
|
|
|
|
class Category(enum.Enum):
|
|
"""Security finding categories covered by the detector set."""
|
|
|
|
PLAINTEXT_CREDENTIALS = "plaintext_credentials"
|
|
PASSWORD_HASHES = "password_hashes"
|
|
TOKENS_SECRETS = "tokens_secrets"
|
|
MALWARE_BEACONING = "malware_beaconing"
|
|
EXFILTRATION = "exfiltration"
|
|
MITM = "mitm"
|
|
INSECURE_PROTOCOL = "insecure_protocol"
|
|
TLS_WEAKNESS = "tls_weakness"
|
|
DNS_ANOMALY = "dns_anomaly"
|
|
BRUTEFORCE_SCAN_FILETRANSFER = "bruteforce_scan_filetransfer"
|
|
|
|
|
|
_SEVERITY_RANK = {
|
|
Severity.CRITICAL: 4,
|
|
Severity.HIGH: 3,
|
|
Severity.MEDIUM: 2,
|
|
Severity.LOW: 1,
|
|
Severity.INFO: 0,
|
|
}
|
|
|
|
_CONFIDENCE_RANK = {
|
|
Confidence.HIGH: 2,
|
|
Confidence.MEDIUM: 1,
|
|
Confidence.LOW: 0,
|
|
}
|
|
|
|
|
|
def _coerce(value, enum_cls):
|
|
"""Accept enum members and their serialized string values."""
|
|
if isinstance(value, enum_cls):
|
|
return value
|
|
if isinstance(value, str):
|
|
return enum_cls(value)
|
|
raise TypeError(f"{value!r} is not a valid {enum_cls.__name__}")
|
|
|
|
|
|
@dataclass
|
|
class Finding:
|
|
"""Shared data model for detector output and packet evidence."""
|
|
|
|
title: str
|
|
category: Category
|
|
severity: Severity
|
|
confidence: Confidence
|
|
risk_description: str
|
|
recommendation: str
|
|
|
|
affected_device: Optional[str] = None
|
|
source_ip: Optional[str] = None
|
|
destination_ip: Optional[str] = None
|
|
protocol: Optional[str] = None
|
|
port: Optional[int] = None
|
|
timestamp_first_seen: Optional[float] = None
|
|
timestamp_last_seen: Optional[float] = None
|
|
|
|
# Structured evidence captured by the detector.
|
|
evidence: dict[str, Any] = field(default_factory=dict)
|
|
related_packets: list[int] = field(default_factory=list)
|
|
related_streams: list[int] = field(default_factory=list)
|
|
|
|
# Filled by the engine with the detector class name.
|
|
detector: Optional[str] = None
|
|
finding_id: Optional[str] = None
|
|
|
|
def __post_init__(self) -> None:
|
|
self.category = _coerce(self.category, Category)
|
|
self.severity = _coerce(self.severity, Severity)
|
|
self.confidence = _coerce(self.confidence, Confidence)
|
|
if self.finding_id is None:
|
|
self.finding_id = self.compute_id()
|
|
|
|
def compute_id(self) -> str:
|
|
"""Build a stable identifier from the finding's identity fields."""
|
|
basis = "|".join(
|
|
str(x)
|
|
for x in (
|
|
self.category.value,
|
|
self.title,
|
|
self.source_ip,
|
|
self.destination_ip,
|
|
self.protocol,
|
|
self.port,
|
|
_evidence_fingerprint(self.evidence),
|
|
)
|
|
)
|
|
return hashlib.sha1(basis.encode("utf-8")).hexdigest()[:16]
|
|
|
|
@property
|
|
def priority(self) -> tuple[int, int]:
|
|
"""Sort by severity first, then confidence."""
|
|
return (self.severity.rank, self.confidence.rank)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
data = asdict(self)
|
|
data["category"] = self.category.value
|
|
data["severity"] = self.severity.value
|
|
data["confidence"] = self.confidence.value
|
|
return data
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> "Finding":
|
|
"""Reconstruct a finding from serialized data."""
|
|
fields = {
|
|
"title", "category", "severity", "confidence", "risk_description",
|
|
"recommendation", "affected_device", "source_ip", "destination_ip",
|
|
"protocol", "port", "timestamp_first_seen", "timestamp_last_seen",
|
|
"evidence", "related_packets", "related_streams", "detector", "finding_id",
|
|
}
|
|
kwargs = {k: v for k, v in data.items() if k in fields}
|
|
return cls(**kwargs)
|
|
|
|
|
|
def _evidence_fingerprint(evidence: dict[str, Any]) -> str:
|
|
"""Prefer an explicit deduplication key, then stable sorted evidence."""
|
|
if not evidence:
|
|
return ""
|
|
if "dedup_key" in evidence:
|
|
return str(evidence["dedup_key"])
|
|
try:
|
|
items = sorted((str(k), str(v)) for k, v in evidence.items())
|
|
except Exception:
|
|
return str(evidence)
|
|
return ";".join(f"{k}={v}" for k, v in items)
|