"""Report assembly and module definitions for eleVADR output."""
# Standard Python Libraries
import ipaddress
from abc import ABC, abstractmethod
from collections import Counter
from collections.abc import Callable
from typing import Any, cast
# Third-Party Libraries
import numpy as np
import pandas as pd
from .analysis import Analyzer
from .utils import PortType, count_values_in_list_column
type JSONValue = dict[str, "JSONValue"] | list["JSONValue"] | str | int | float | bool | None
[docs]
class Report:
"""Main report orchestrator class.
Instantiates module classes and aggregates
their data into a unified report structure.
"""
REPORT_VERSION = "2.0"
def __init__(self, analyzer: Analyzer, report_id: str | None = None):
"""Initialize report modules and assemble the top-level payload."""
self.analyzer = analyzer
self.report_id = report_id
# Initialize all report modules
modules_instances = [
# Aggregate Analysis Modules
ServicePanelModule(analyzer),
DevicePanelModule(analyzer),
ServiceRiskBreakdownModule(analyzer),
ServiceCountModule(analyzer),
RiskBasisBreakdownModule(analyzer),
ExposureBreakdownModule(analyzer),
ProtocolPostureModule(analyzer),
ConnectionSuccessModule(analyzer),
# Table Analysis Modules
SuspiciousOutboundConnectionsModule(analyzer),
OTcrossSegmentLinesModule(analyzer),
DevicesModule(analyzer, name="ot_devices", device_filter=lambda df: df["device.is_ot"]),
DevicesModule(
analyzer,
name="it_devices",
device_filter=lambda df: (~df["device.is_ot"]) & (~df["device.is_edge"]),
),
DevicesModule(
analyzer,
name="edge_devices",
device_filter=lambda df: df["device.is_edge"],
),
OTServicesModule(analyzer),
]
self.modules = {module.name: module for module in modules_instances}
# Initialize detection modules
self.detections = [
SuspiciousOutboundConnectionsDetection([self.modules["suspicious_outbound_connections_panel"]]),
OTcrossSegmentDetection(
[
self.modules["ot_cross_segment_lines_panel"],
]
),
]
# Build executive summary from detections
executive_summary = {}
for detection in self.detections:
if detection.run_detection():
executive_summary[detection.name] = detection.executive_summary
# Build the report data structure
self.data: JSONValue = {
"report_version": self.REPORT_VERSION,
"report_id": self.report_id,
"executive_summary": cast(JSONValue, executive_summary),
"modules": {module_name: cast(JSONValue, module.data) for module_name, module in self.modules.items()},
"arch_insights": {},
}
class ReportModule(ABC):
"""Abstract base class for report modules."""
def __init__(self, analyzer: Analyzer):
"""Store common dataframe references for concrete modules."""
self.analyzer = analyzer
self.traffic_df = analyzer.traffic_df
self.endpoints_df = analyzer.endpoints_df
self.services_df = analyzer.services_df
self._data: object = None
@property
@abstractmethod
def name(self) -> str:
"""Return the identifier."""
pass
@abstractmethod
def generate_data(self) -> object:
"""Generate module-specific data."""
pass
@property
def data(self) -> object:
"""Lazy-load and cache data."""
if self._data is None:
self._data = self.generate_data()
return self._data
#####
#
# Aggregate Analysis Modules
#
#####
[docs]
class DevicePanelModule(ReportModule):
"""Device statistics module."""
@property
def name(self) -> str:
"""Return `device_panel`."""
return "device_panel"
def generate_data(self) -> dict[str, object]:
"""Summarize endpoint counts and OT cross\u2011segment communication.
Returns
-------
dict
{
"hosts": total endpoint count,
"ot_hosts": OT\u2011marked endpoint count,
"it_hosts": IT\u2011only endpoint count,
"edge_hosts": edge\u2011marked endpoint count,
"ot_cross_segment": count of OT cross\u2011segment communications
}
"""
ot_hosts = self.endpoints_df[self.endpoints_df["device.is_ot"]]
edge_hosts = self.endpoints_df[self.endpoints_df["device.is_edge"]]
it_hosts = self.endpoints_df[(~self.endpoints_df["device.is_ot"]) & (~self.endpoints_df["device.is_edge"])]
return {
"hosts": len(self.endpoints_df),
"ot_hosts": len(ot_hosts),
"it_hosts": max(len(it_hosts) - len(edge_hosts), 0),
"edge_hosts": len(edge_hosts),
"ot_cross_segment": self.analyzer.ot_cross_segment_communication_count(),
}
[docs]
class ServicePanelModule(ReportModule):
"""Service overview statistics module."""
@property
def name(self) -> str:
"""Return `service_panel`."""
return "service_panel"
def generate_data(self) -> dict[str, object]:
"""Return a summary of service statistics.
- known, unknown and OT services counts
- risky services count
"""
unknown_services = self.traffic_df[
self.traffic_df["service.port_type"].isin(
[
PortType.EPHEMERAL.name,
PortType.UNKNOWN.name,
PortType.UNKNOWN_PRIV.name,
]
)
]
# Known services
known_services = self.traffic_df[self.traffic_df["service.port_type"].isin([PortType.KNOWN.name])]
return {
"num_known_services": len(known_services["service.name"].dropna().unique()),
"num_ot_services": len(
self.services_df[
self.services_df["service.information_categories"].apply(
lambda x: "Industrial Protocol" in x if isinstance(x, str) else False
)
]
),
"num_risky_services": len(
self.services_df[self.services_df["service.risk_categories"].notna()]["service.name"].unique()
),
"num_unknown_services": len(unknown_services["service.name"].dropna().unique()),
}
[docs]
class ServiceRiskBreakdownModule(ReportModule):
"""Service risk categorization module."""
@property
def name(self) -> str:
"""Return `service_risk_breakdown_panel`."""
return "service_risk_breakdown_panel"
def generate_data(self) -> dict[str, object]:
"""Summarize service risk information.
Returns
-------
dict
{
"risk_category_counts": dict mapping risk category \u2192 occurrence count,
"risk_category_services": dict mapping risk category \u2192 list of service names
}
"""
risk_category_counts = count_values_in_list_column(self.services_df, "service.risk_categories")
risk_category_services = self.analyzer.service_category_map("service.risk_categories")
return {
"risk_category_counts": risk_category_counts,
"risk_category_services": risk_category_services,
}
[docs]
class ServiceCountModule(ReportModule):
"""Service connection counts module."""
@property
def name(self) -> str:
"""Return `service_count_panel`."""
return "service_count_panel"
def generate_data(self) -> dict[str, object]:
"""Return the total number of distinct services observed in the traffic data.
The method:
* Splits the traffic frame into *known* and *unknown* services based on
``service.port_type`` (known, ephem\u2011eral, unknown, unknown\u2011priv).
* Counts the unique, non\u2011null ``service.name`` values in each group.
* Sums the two counts and returns the result.
Returns:
-------
dict
{
"service_count": int # total unique services (known + unknown)
}
Note:
----
This routine depends on the ServicePanelModule data that the analyzer
populates; the ``self.traffic_df`` must already contain the ``service.*``
columns.
"""
# Note: This module depends on ServicePanelModule data
# We'll need to access the analyzer's services for counting
unknown_services = self.traffic_df[
self.traffic_df["service.port_type"].isin(
[
PortType.EPHEMERAL.name,
PortType.UNKNOWN.name,
PortType.UNKNOWN_PRIV.name,
]
)
]
# Known services
known_services = self.traffic_df[self.traffic_df["service.port_type"].isin([PortType.KNOWN.name])]
service_total_count = len(known_services["service.name"].dropna().unique()) + len(
unknown_services["service.name"].dropna().unique()
)
return {
"service_count": service_total_count,
"service_connections_count": self.analyzer.service_counts_in_traffic(),
}
[docs]
class RiskBasisBreakdownModule(ReportModule):
"""Breaks down observed services by risk basis (Observed vs Credible).
Observed - directly used in compromises with evidence.
Credible - known threat actor TTPs, not necessarily active now.
"""
@property
def name(self) -> str:
"""Return `risk_basis_breakdown_panel`."""
return "risk_basis_breakdown_panel"
def generate_data(self) -> dict[str, object]:
"""Return risk-basis counts and associated service names."""
known_services = self.services_df[self.services_df["service.port_type"] == "KNOWN"]
# Count services per basis value, excluding nulls
basis_counts = known_services["service.risk_basis"].dropna().value_counts().to_dict()
# Map each basis value to the services that carry it
basis_service_map: dict[str, list[str]] = {}
for _, row in known_services.iterrows():
basis = row["service.risk_basis"]
if isinstance(basis, str) and basis:
basis_service_map.setdefault(basis, []).append(row["service.name"])
return {
"risk_basis_counts": basis_counts,
"risk_basis_services": basis_service_map,
}
[docs]
class ExposureBreakdownModule(ReportModule):
"""Breaks down observed services by environment/exposure classification.
External - internet-facing or reachable from outside the org.
Cross-Zone - traverses trust boundaries without broker/inspection.
Internal - confined within a protected cell/zone.
"""
@property
def name(self) -> str:
"""Return `exposure_breakdown_panel`."""
return "exposure_breakdown_panel"
def generate_data(self) -> dict[str, object]:
"""Return exposure counts and associated service names."""
known_services = self.services_df[self.services_df["service.port_type"] == "KNOWN"]
exposure_counts = known_services["service.environment_exposure"].dropna().value_counts().to_dict()
exposure_service_map: dict[str, list[str]] = {}
for _, row in known_services.iterrows():
exposure = row["service.environment_exposure"]
if isinstance(exposure, str) and exposure:
exposure_service_map.setdefault(exposure, []).append(row["service.name"])
return {
"exposure_counts": exposure_counts,
"exposure_services": exposure_service_map,
}
[docs]
class ConnectionSuccessModule(ReportModule):
"""Connection success vs failure breakdown based on Zeek conn.log."""
@property
def name(self) -> str:
"""Return `connection_success_panel`."""
return "connection_success_panel"
def generate_data(self) -> dict[str, object]:
"""Return connection success summary data and representative lines."""
return {
"summary": self.analyzer.connection_success_summary(),
"connections": self.analyzer.connection_success_lines(limit=200),
}
[docs]
class ProtocolPostureModule(ReportModule):
"""Breaks down observed services by protocol posture.
Inherently Risky - clear-text or long-standing exploitation history.
Conditionally Risky - secure if hardened, risky when exposed or tunnelled.
"""
@property
def name(self) -> str:
"""Return `protocol_posture_panel`."""
return "protocol_posture_panel"
def generate_data(self) -> dict[str, object]:
"""Return protocol-posture counts and associated service names."""
known_services = self.services_df[self.services_df["service.port_type"] == "KNOWN"]
posture_counts = known_services["service.protocol_posture"].dropna().value_counts().to_dict()
posture_service_map: dict[str, list[str]] = {}
for _, row in known_services.iterrows():
posture = row["service.protocol_posture"]
if isinstance(posture, str) and posture:
posture_service_map.setdefault(posture, []).append(row["service.name"])
return {
"posture_counts": posture_counts,
"posture_services": posture_service_map,
}
#####
#
# Table Analysis Modules
#
#####
[docs]
class SuspiciousOutboundConnectionsModule(ReportModule):
"""Suspicious outbound connections from OT devices module."""
@property
def name(self) -> str:
"""Return `suspicious_outbound_connections_panel`."""
return "suspicious_outbound_connections_panel"
def generate_data(self) -> list[dict[str, object]]:
"""Return grouped outbound OT connection records for reporting."""
outbound_traffic = self.traffic_df[self.traffic_df["connection_info.direction_name"] == "outbound"]
ot_join_col = "src_endpoint.mac" if "src_endpoint.mac" in outbound_traffic.columns else "dst_endpoint.mac"
outbound_traffic_w_ot = outbound_traffic.merge(
self.endpoints_df["device.is_ot"],
left_on=ot_join_col,
right_index=True,
how="left",
)
display_cols = [
"src_endpoint.ip",
"dst_endpoint.ip",
"dst_endpoint.port",
"service.name",
]
outbound_traffic_ot = outbound_traffic_w_ot[outbound_traffic_w_ot["device.is_ot"].fillna(False)][display_cols]
return list(
outbound_traffic_ot.groupby(display_cols, sort=False)
.size()
.reset_index(name="count")
.to_dict(orient="records")
)
[docs]
class DevicesModule(ReportModule):
"""Generic devices module to list devices based on a filter."""
def __init__(
self,
analyzer: Analyzer,
name: str,
device_filter: Callable[[pd.DataFrame], pd.Series],
):
"""Initialize the module with a report name and dataframe filter."""
super().__init__(analyzer)
self._name = name
self.device_filter = device_filter
@property
def name(self) -> str:
"""Return the configured module name."""
return self._name
def generate_data(self) -> list[dict[str, Any]]:
"""Return filtered device records prepared for JSON serialization."""
devices_df = self.endpoints_df[self.device_filter(self.endpoints_df)]
device_columns = {
"device.manufacturer": "manufacturer",
"device.ipv4_ips": "ipv4_ips",
"device.ipv6_ips": "ipv6_ips",
"device.ipv4_subnets": "ipv4_subnets",
"device.ipv6_subnets": "ipv6_subnets",
"device.incoming_services": "incoming_services",
"device.sent_services": "sent_services",
}
return self._prep_df_for_json(devices_df, device_columns)
def _prep_df_for_json(self, df: pd.DataFrame, columns: dict[str, str]) -> list[dict[str, Any]]:
"""Prepare a dataframe for JSON serialization.
Selects and renames columns before converting rows to records.
"""
if df.empty:
return []
existing_cols = {k: v for k, v in columns.items() if k in df.columns}
report_df = df[list(existing_cols.keys())].rename(columns=existing_cols)
report_df = report_df.replace({np.nan: None})
for column in report_df.columns:
report_df[column] = report_df[column].apply(
lambda value: sorted(value) if isinstance(value, set) else value
)
return list(report_df.astype(object).to_dict("records"))
# Removed OTManufacturersModule as it's being replaced
# class OTManufacturersModule(ReportModule):
# """OT device manufacturers module"""
#
# @property
# def name(self) -> str:
# return "ot_manufacturers"
#
# def generate_data(self) -> dict:
# return (
# self.endpoints_df[self.endpoints_df["device.is_ot"]]["device.manufacturer"]
# .value_counts()
# .to_dict()
# )
[docs]
class OTServicesModule(ReportModule):
"""OT services/protocols module."""
@property
def name(self) -> str:
"""Return `ot_services`."""
return "ot_services"
def generate_data(self) -> list[dict[str, object]]:
"""Return service records tagged as industrial protocols."""
return list(
self.services_df[
self.services_df["service.information_categories"].apply(
lambda x: "Industrial Protocol" in x if isinstance(x, str) else False
)
].to_dict(orient="records")
)
#####
#
# Detection Modules
#
#####
[docs]
class DetectionModule(ABC):
"""Abstract base class for detections based on report modules."""
def __init__(self, report_modules: list[ReportModule]):
"""Store module dependencies used by a detection."""
self.report_modules = report_modules
self._data: object = None
@property
@abstractmethod
def name(self) -> str:
"""Return the identifier."""
pass
@property
@abstractmethod
def executive_summary(self) -> str:
"""Return the human-readable finding summary."""
pass
@abstractmethod
def run_detection(self) -> bool:
"""Evaluate the detection against the available module data."""
pass
@property
def data(self) -> object:
"""Lazy-load and cache data."""
if self._data is None:
self._data = self.generate_data() # type: ignore[attr-defined]
return self._data
class OTcrossSegmentLinesModule(ReportModule):
"""OT cross-segment communication lines module."""
@property
def name(self) -> str:
"""Return `ot_cross_segment_lines_panel`."""
return "ot_cross_segment_lines_panel"
@staticmethod
def _is_excluded_cross_segment_ip(ip: object) -> bool:
"""Return whether an IP should be excluded from cross-segment reporting."""
if not isinstance(ip, str) or not ip.strip():
return True
try:
parsed_ip = ipaddress.ip_address(ip)
except ValueError:
return True
return bool(
getattr(parsed_ip, "is_multicast", False)
or getattr(parsed_ip, "is_link_local", False)
or ip == "255.255.255.255"
)
@classmethod
def _preferred_ip_for_mac(cls, endpoints_df: pd.DataFrame, mac: str) -> str | None:
"""Return a stable preferred IP for a MAC, preferring IPv4 over IPv6."""
if mac not in endpoints_df.index:
return None
endpoint = endpoints_df.loc[mac]
if isinstance(endpoint, pd.DataFrame):
endpoint = endpoint.iloc[0]
for column in ("device.ipv4_ips", "device.ipv6_ips"):
if column not in endpoint.index:
continue
ips = endpoint[column]
if isinstance(ips, (list, tuple, np.ndarray)):
for ip in ips:
if isinstance(ip, str) and ip.strip():
return str(ip)
return None
def generate_data(self) -> dict[str, list[dict[str, object]]]:
"""Return grouped OT cross-segment reporting data."""
cross_seg_df = self.traffic_df.copy()
empty_result: dict[str, list[dict[str, object]]] = {
"lines": [],
"subnet_pair_counts": [],
"dst_subnet_counts": [],
"ot_device_counts": [],
}
if cross_seg_df.empty:
return empty_result
ot_macs = set(self.endpoints_df[self.endpoints_df["device.is_ot"].fillna(False)].index)
ot_mask = cross_seg_df["src_endpoint.mac"].isin(ot_macs) | cross_seg_df["dst_endpoint.mac"].isin(ot_macs)
ot_cross_seg_df = cross_seg_df[ot_mask].copy()
if ot_cross_seg_df.empty:
return empty_result
ot_cross_seg_df = ot_cross_seg_df[
ot_cross_seg_df["src_endpoint.subnet"] != ot_cross_seg_df["dst_endpoint.subnet"]
].copy()
if ot_cross_seg_df.empty:
return empty_result
ot_cross_seg_df = ot_cross_seg_df[
~ot_cross_seg_df["src_endpoint.ip"].apply(self._is_excluded_cross_segment_ip)
& ~ot_cross_seg_df["dst_endpoint.ip"].apply(self._is_excluded_cross_segment_ip)
].copy()
if ot_cross_seg_df.empty:
return empty_result
line_cols = [
"src_endpoint.ip",
"src_endpoint.subnet",
"dst_endpoint.ip",
"dst_endpoint.subnet",
"dst_endpoint.port",
"service.name",
]
lines = list(
ot_cross_seg_df[line_cols]
.groupby(line_cols, sort=False)
.size()
.reset_index(name="count")
.to_dict(orient="records")
)
subnet_pair_counts = list(
ot_cross_seg_df.groupby(["src_endpoint.subnet", "dst_endpoint.subnet"], sort=False)
.size()
.reset_index(name="count")
.to_dict(orient="records")
)
dst_subnet_counts = list(
ot_cross_seg_df.groupby(["dst_endpoint.subnet"], sort=False)
.size()
.reset_index(name="count")
.to_dict(orient="records")
)
ot_device_counter: Counter[str] = Counter()
for _, row in ot_cross_seg_df.iterrows():
for mac_column in ("src_endpoint.mac", "dst_endpoint.mac"):
mac = row[mac_column]
if mac in ot_macs:
ot_device_counter[mac] += 1
ot_device_counts = [
{
"mac": mac,
"ip": self._preferred_ip_for_mac(self.endpoints_df, mac),
"count": count,
}
for mac, count in ot_device_counter.items()
]
return {
"lines": lines,
"subnet_pair_counts": subnet_pair_counts,
"dst_subnet_counts": dst_subnet_counts,
"ot_device_counts": ot_device_counts,
}
[docs]
class SuspiciousOutboundConnectionsDetection(DetectionModule):
"""Detection: OT devices making suspicious outbound connections."""
@property
def name(self) -> str:
"""Return `suspicious_outbound_connections`."""
return "suspicious_outbound_connections"
@property
def executive_summary(self) -> str:
"""Return a human-readable finding summary."""
if not self.run_detection():
return ""
module = self.report_modules[0]
data = module.data
assert type(data) is list
count = len(data)
examples = ""
example_lines = [
(
f"- {entry['src_endpoint.ip']} → "
f"{entry['dst_endpoint.ip']}:{entry['dst_endpoint.port']} "
f"({entry['service.name']}) - {entry['count']}x"
)
for entry in data
]
examples = "\n".join(example_lines)
return (
"**FINDING: Suspicious Outbound Connections Detected**\n\n"
f"{count} suspicious outbound connection(s) from OT devices were identified.\n\n"
f"{examples}\n\n"
"OT devices were observed making outbound connections to external hosts. "
"This may indicate unauthorized communication or a compromised device.\n\n"
"**Recommended Actions:**\n"
"- Validate whether the destinations are approved external dependencies.\n"
"- Review device ownership, change history, and recent operational intent.\n"
"- Investigate for compromise, tunneling, or policy bypass if the traffic is unexpected."
)
def run_detection(self) -> bool:
"""Trip if any suspicious outbound OT connections are present."""
module = self.report_modules[0]
data = module.data
return isinstance(data, list) and len(data) > 0
[docs]
class OTcrossSegmentDetection(DetectionModule):
"""Detection: OT devices communicating across network segments."""
@property
def name(self) -> str:
"""Return `ot_cross_segment`."""
return "ot_cross_segment"
@property
def executive_summary(self) -> str:
"""Return a human-readable finding summary."""
if not self.run_detection():
return ""
module = self.report_modules[0]
data = module.data
lines = data.get("lines", []) if isinstance(data, dict) else []
count = len(lines)
examples = ""
example_lines = [
(
f"- {entry['src_endpoint.ip']} → "
f"{entry['dst_endpoint.ip']}:{entry['dst_endpoint.port']} "
f"({entry['service.name']}) - {entry['count']}x"
)
for entry in lines
]
if example_lines:
examples = "\n".join(example_lines)
return (
"**FINDING: OT Cross-Segment Communications Detected**\n\n"
f"{count} cross-segment communication example(s) involving OT assets were identified.\n\n"
f"{examples}\n\n"
"OT devices were observed communicating across network segments. "
"This may indicate a misconfiguration or lateral movement attempt.\n\n"
"**Recommended Actions:**\n"
"- Validate whether the observed communications are expected for operations.\n"
"- Review segmentation policy, firewall rules, and routing between the affected subnets.\n"
"- Investigate potential unauthorized lateral movement if the traffic is unexpected."
)
def run_detection(self) -> bool:
"""Trip if any cross-segment OT lines are present."""
module = self.report_modules[0]
data = module.data
return isinstance(data, dict) and isinstance(data.get("lines"), list) and len(data["lines"]) > 0