172 lines
6.7 KiB
Python
172 lines
6.7 KiB
Python
"""Run all registered detectors and return a prioritized analysis result."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import time
|
|
from dataclasses import dataclass, field
|
|
from typing import Optional
|
|
|
|
from .context import CaptureContext
|
|
from .findings import Finding
|
|
from . import scoring
|
|
from . import provenance
|
|
from . import detectors as detectors_pkg
|
|
from . import inventory as inventory_mod
|
|
|
|
logger = logging.getLogger("analyzer.engine")
|
|
|
|
|
|
@dataclass
|
|
class AnalysisResult:
|
|
findings: list[Finding] = field(default_factory=list)
|
|
summary: dict = field(default_factory=dict)
|
|
meta: dict = field(default_factory=dict)
|
|
# Complete observation inventory (hosts, endpoints, protocols, TLS posture, ...).
|
|
inventory: dict = field(default_factory=dict)
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"meta": self.meta,
|
|
"summary": _summary_to_dict(self.summary),
|
|
"findings": [f.to_dict() for f in self.findings],
|
|
"inventory": self.inventory,
|
|
}
|
|
|
|
|
|
class AnalysisEngine:
|
|
def __init__(self, detector_list=None):
|
|
# Use every registered detector unless a test supplies a subset.
|
|
self.detectors = detector_list if detector_list is not None else detectors_pkg.build_all()
|
|
|
|
def analyze_context(self, ctx: CaptureContext) -> AnalysisResult:
|
|
started = time.time()
|
|
raw: list[Finding] = []
|
|
errors: dict[str, str] = {}
|
|
timings: dict[str, float] = {}
|
|
logger.info("Analysis start: %d detectors, %d packets, device_ip=%s",
|
|
len(self.detectors), ctx.packet_count, ctx.device_ip)
|
|
|
|
for det in self.detectors:
|
|
t0 = time.time()
|
|
try:
|
|
found = list(det.detect(ctx))
|
|
raw.extend(found)
|
|
logger.debug("%s: %d findings in %.3fs", det.name, len(found), time.time() - t0)
|
|
except Exception as exc: # One detector must not abort the full analysis.
|
|
errors[det.name] = repr(exc)
|
|
logger.exception("%s failed: %r", det.name, exc)
|
|
timings[det.name] = round(time.time() - t0, 3)
|
|
|
|
deduped = scoring.dedupe(raw)
|
|
# Attribute findings to the device under test before prioritizing.
|
|
for f in deduped:
|
|
_attribute_device(f, ctx)
|
|
ordered = scoring.sort_by_priority(deduped)
|
|
# Attach packet-level provenance before the summary is generated.
|
|
capture_file = _basename(ctx.path)
|
|
for f in ordered:
|
|
_enrich_provenance(f, capture_file)
|
|
summary = scoring.summarize(ordered)
|
|
try:
|
|
inventory = inventory_mod.build_inventory(ctx)
|
|
except Exception as exc: # The inventory must never break the analysis.
|
|
logger.exception("Inventory build failed: %r", exc)
|
|
inventory = {"error": repr(exc)}
|
|
logger.info("Analysis done: %d raw -> %d findings (%s) in %.2fs; detector errors: %s",
|
|
len(raw), len(ordered), summary.get("by_severity"), time.time() - started,
|
|
errors or "none")
|
|
|
|
return AnalysisResult(
|
|
findings=ordered,
|
|
summary=summary,
|
|
inventory=inventory,
|
|
meta={
|
|
"capture_path": ctx.path,
|
|
"packet_count": ctx.packet_count,
|
|
"duration_seconds": round(ctx.duration, 3),
|
|
"capture_start": ctx.start_time,
|
|
"capture_end": ctx.end_time,
|
|
"device_ip": ctx.device_ip,
|
|
"device_macs": sorted(ctx.device_macs),
|
|
"extractor": ctx.extractor,
|
|
"coverage_notes": list(ctx.coverage_notes),
|
|
"detectors_run": [d.name for d in self.detectors],
|
|
"detector_errors": errors,
|
|
"detector_timings": timings,
|
|
"analysis_time_seconds": round(time.time() - started, 3),
|
|
"raw_finding_count": len(raw),
|
|
"finding_count": len(ordered),
|
|
# A partial pass must never be mistaken for a complete one.
|
|
"truncated": bool(getattr(ctx, "truncated", False)),
|
|
"truncation_reason": getattr(ctx, "truncation_reason", None),
|
|
},
|
|
)
|
|
|
|
def analyze_pcap(self, path: str, packet_limit: Optional[int] = None,
|
|
device_ip: Optional[str] = None) -> AnalysisResult:
|
|
# Keep context-only analysis usable without tshark.
|
|
from .extract_tshark import build_context_from_pcap
|
|
ctx = build_context_from_pcap(path, packet_limit=packet_limit, device_ip=device_ip)
|
|
return self.analyze_context(ctx)
|
|
|
|
|
|
def result_from_dict(data: dict) -> AnalysisResult:
|
|
"""Rebuild a result from stored JSON; the summary is recalculated."""
|
|
findings = [Finding.from_dict(f) for f in data.get("findings", [])]
|
|
summary = scoring.summarize(findings)
|
|
meta = data.get("meta", {})
|
|
return AnalysisResult(findings=findings, summary=summary, meta=meta,
|
|
inventory=data.get("inventory") or {})
|
|
|
|
|
|
def _attribute_device(f: Finding, ctx: CaptureContext) -> None:
|
|
"""Set affected_device when the finding touches the device under test."""
|
|
if ctx.device_ip is None:
|
|
return
|
|
involved = ctx.involves_device(f.source_ip, f.destination_ip)
|
|
if involved is None:
|
|
return
|
|
f.evidence["device_related"] = bool(involved)
|
|
if involved and not f.affected_device:
|
|
f.affected_device = ctx.device_ip
|
|
|
|
|
|
def _summary_to_dict(summary: dict) -> dict:
|
|
out = dict(summary)
|
|
if "highlights" in out:
|
|
out["highlights"] = [f.to_dict() for f in out["highlights"]]
|
|
if "top" in out:
|
|
out["top"] = [f.to_dict() for f in out["top"]]
|
|
return out
|
|
|
|
|
|
def _basename(path) -> Optional[str]:
|
|
if not path:
|
|
return None
|
|
import os
|
|
return os.path.basename(str(path)) or str(path)
|
|
|
|
|
|
def _enrich_provenance(f: Finding, capture_file) -> None:
|
|
"""Add capture, artifact, and packet sources without replacing detector data."""
|
|
ev = f.evidence
|
|
if capture_file and not ev.get("capture_file"):
|
|
ev["capture_file"] = capture_file
|
|
if "artifact" not in ev:
|
|
art = provenance.derive_artifact(ev)
|
|
if art is not None:
|
|
ev["artifact"] = art
|
|
if "sources" not in ev:
|
|
ev["sources"] = provenance.build_sources(
|
|
related_packets=f.related_packets,
|
|
related_streams=f.related_streams,
|
|
src_ip=f.source_ip, dst_ip=f.destination_ip,
|
|
protocol=f.protocol, port=f.port,
|
|
timestamp_first_seen=f.timestamp_first_seen,
|
|
capture=capture_file,
|
|
)
|
|
# Preserve the complete evidence count when the displayed source list is bounded.
|
|
if f.related_packets and "source_packet_count" not in ev:
|
|
ev["source_packet_count"] = len(f.related_packets)
|