111 lines
4.0 KiB
Python
111 lines
4.0 KiB
Python
"""Build packet-level provenance without coupling findings to PCAP I/O."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Optional
|
|
|
|
# Keep reports compact while preserving the complete packet count separately.
|
|
MAX_SOURCE_PACKETS = 8
|
|
|
|
|
|
def source_ref(*, packet: Optional[int] = None, stream: Optional[int] = None,
|
|
timestamp: Optional[float] = None, src_ip: Optional[str] = None,
|
|
dst_ip: Optional[str] = None, protocol: Optional[str] = None,
|
|
port: Optional[int] = None, capture: Optional[str] = None,
|
|
note: Optional[str] = None) -> dict:
|
|
"""Build a normalized source reference while omitting empty fields."""
|
|
ref: dict[str, Any] = {}
|
|
if packet is not None:
|
|
ref["packet"] = packet
|
|
if stream is not None:
|
|
ref["stream"] = stream
|
|
if timestamp is not None:
|
|
ref["timestamp"] = timestamp
|
|
if src_ip:
|
|
ref["src_ip"] = src_ip
|
|
if dst_ip:
|
|
ref["dst_ip"] = dst_ip
|
|
if protocol:
|
|
ref["protocol"] = protocol
|
|
if port:
|
|
ref["port"] = port
|
|
if capture:
|
|
ref["capture"] = capture
|
|
if note:
|
|
ref["note"] = note
|
|
return ref
|
|
|
|
|
|
# Priority-ordered evidence keys used to select the primary artifact.
|
|
_ARTIFACT_KEYS = [
|
|
("hashcat", "Hash (hashcat)"),
|
|
("secret_preview", "Secret/Token"),
|
|
("query_name", "DNS query"),
|
|
("cipher", "TLS-Cipher"),
|
|
("version", "TLS-Version"),
|
|
("user_agent", "User-Agent"),
|
|
("filename", "File"),
|
|
("uri", "URI"),
|
|
("subject_cn", "Certificate CN"),
|
|
("registered_domain", "Domain"),
|
|
("macs", "MAC addresses"),
|
|
]
|
|
|
|
|
|
def derive_artifact(evidence: dict) -> Optional[dict]:
|
|
"""Derive the most representative artifact from structured evidence."""
|
|
if not evidence:
|
|
return None
|
|
|
|
# Treat a username and masked password as one artifact.
|
|
user = evidence.get("username")
|
|
pw = evidence.get("password")
|
|
if user or pw:
|
|
return {"label": "Credentials", "value": f"{user or '?'} : {pw or '?'}"}
|
|
|
|
for key, label in _ARTIFACT_KEYS:
|
|
val = evidence.get(key)
|
|
if val:
|
|
if isinstance(val, (list, tuple)):
|
|
val = ", ".join(str(x) for x in val)
|
|
return {"label": label, "value": str(val)}
|
|
return None
|
|
|
|
|
|
def build_sources(*, related_packets: list[int], related_streams: list[int],
|
|
src_ip: Optional[str], dst_ip: Optional[str],
|
|
protocol: Optional[str], port: Optional[int],
|
|
timestamp_first_seen: Optional[float],
|
|
capture: Optional[str],
|
|
max_packets: int = MAX_SOURCE_PACKETS) -> list[dict]:
|
|
"""Build a bounded source list from packet, stream, and endpoint fields."""
|
|
packets = list(dict.fromkeys(p for p in (related_packets or []) if p is not None))
|
|
streams = list(dict.fromkeys(s for s in (related_streams or []) if s is not None))
|
|
single_stream = streams[0] if len(streams) == 1 else None
|
|
|
|
sources: list[dict] = []
|
|
if packets:
|
|
single_ts = timestamp_first_seen if len(packets) == 1 else None
|
|
for p in packets[:max_packets]:
|
|
sources.append(source_ref(
|
|
packet=p, stream=single_stream, timestamp=single_ts,
|
|
src_ip=src_ip, dst_ip=dst_ip, protocol=protocol, port=port,
|
|
capture=capture,
|
|
))
|
|
elif streams:
|
|
for s in streams[:max_packets]:
|
|
sources.append(source_ref(
|
|
stream=s, src_ip=src_ip, dst_ip=dst_ip, protocol=protocol,
|
|
port=port, capture=capture,
|
|
))
|
|
elif src_ip or dst_ip or timestamp_first_seen is not None:
|
|
sources.append(source_ref(
|
|
src_ip=src_ip, dst_ip=dst_ip, protocol=protocol, port=port,
|
|
timestamp=timestamp_first_seen, capture=capture,
|
|
note="aggregated evidence; no single packet is available",
|
|
))
|
|
elif capture:
|
|
sources.append(source_ref(capture=capture,
|
|
note="aggregated evidence; no single packet is available"))
|
|
return sources
|