454 lines
16 KiB
Python
454 lines
16 KiB
Python
"""Normalized capture data consumed by every detector.
|
|
|
|
The module is independent of Django and tshark so detector logic can be tested
|
|
with synthetic data. ``extract_tshark.py`` performs the actual PCAP extraction.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ipaddress
|
|
from dataclasses import dataclass, field
|
|
from typing import Optional
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Normalized records
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
@dataclass
|
|
class Host:
|
|
ip: str
|
|
macs: set[str] = field(default_factory=set)
|
|
hostnames: set[str] = field(default_factory=set) # DHCP, mDNS, reverse DNS names
|
|
bytes_sent: int = 0
|
|
bytes_received: int = 0
|
|
packets_sent: int = 0
|
|
packets_received: int = 0
|
|
first_seen: Optional[float] = None
|
|
last_seen: Optional[float] = None
|
|
ports_offered: set[int] = field(default_factory=set) # Ports the host answered on (server role)
|
|
ports_contacted: set[int] = field(default_factory=set) # Destination ports the host contacted
|
|
dhcp_vendor_class: Optional[str] = None
|
|
user_agents: set[str] = field(default_factory=set)
|
|
# Application protocols seen with this host as an endpoint (from frame.protocols).
|
|
app_protocols: set[str] = field(default_factory=set)
|
|
|
|
@property
|
|
def is_private(self) -> bool:
|
|
return _is_private(self.ip)
|
|
|
|
@property
|
|
def packets(self) -> int:
|
|
return self.packets_sent + self.packets_received
|
|
|
|
|
|
@dataclass
|
|
class Conversation:
|
|
"""A five-tuple flow aggregated between two endpoints."""
|
|
|
|
src_ip: str
|
|
dst_ip: str
|
|
src_port: Optional[int]
|
|
dst_port: Optional[int]
|
|
protocol: str # For example, "tcp" or "udp"
|
|
packets: int = 0
|
|
bytes: int = 0
|
|
bytes_src_to_dst: int = 0
|
|
bytes_dst_to_src: int = 0
|
|
packets_src_to_dst: int = 0
|
|
packets_dst_to_src: int = 0
|
|
payload_bytes: int = 0 # Application payload only (no headers)
|
|
first_seen: Optional[float] = None
|
|
last_seen: Optional[float] = None
|
|
stream_id: Optional[int] = None
|
|
timestamps: list[float] = field(default_factory=list) # Used to detect beaconing
|
|
packet_numbers: list[int] = field(default_factory=list) # Bounded evidence packet list
|
|
# TCP state summary.
|
|
syn_count: int = 0 # SYNs sent by src (connection attempts)
|
|
synack_count: int = 0 # SYN/ACKs sent back by dst (service answered)
|
|
rst_count: int = 0
|
|
fin_count: int = 0
|
|
retransmissions: int = 0
|
|
# Application protocols dissected on this flow, e.g. {"tls", "http"}.
|
|
app_protocols: set[str] = field(default_factory=set)
|
|
|
|
@property
|
|
def answered(self) -> bool:
|
|
"""Whether the destination visibly responded (TCP handshake or return traffic)."""
|
|
if self.protocol == "tcp":
|
|
return self.synack_count > 0 or self.packets_dst_to_src > 0
|
|
return self.packets_dst_to_src > 0
|
|
|
|
@property
|
|
def is_encrypted(self) -> bool:
|
|
return bool(self.app_protocols & ENCRYPTED_PROTOCOLS)
|
|
|
|
|
|
# Protocols whose payload is ciphertext and must never be scanned as text.
|
|
ENCRYPTED_PROTOCOLS = {
|
|
"tls", "ssl", "dtls", "quic", "ssh", "esp", "ipsec", "wireguard", "openvpn",
|
|
"ikev2", "isakmp", "kerberos", "ntlmssp",
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class DnsQuery:
|
|
query_name: str
|
|
qtype: Optional[str] = None
|
|
is_response: bool = False
|
|
response_ips: list[str] = field(default_factory=list)
|
|
cnames: list[str] = field(default_factory=list)
|
|
txt: list[str] = field(default_factory=list)
|
|
rcode: Optional[str] = None # For example, "NOERROR" or "NXDOMAIN"
|
|
timestamp: Optional[float] = None
|
|
src_ip: Optional[str] = None
|
|
dst_ip: Optional[str] = None
|
|
resolver: Optional[str] = None
|
|
transport: Optional[str] = None # "dns", "mdns", "llmnr", "nbns"
|
|
packet_number: Optional[int] = None
|
|
|
|
|
|
@dataclass
|
|
class HttpRequest:
|
|
method: Optional[str] = None
|
|
host: Optional[str] = None
|
|
uri: Optional[str] = None
|
|
user_agent: Optional[str] = None
|
|
authorization: Optional[str] = None # Authorization header value
|
|
cookies: Optional[str] = None
|
|
headers: dict[str, str] = field(default_factory=dict)
|
|
body: Optional[str] = None
|
|
content_type: Optional[str] = None
|
|
request_bytes: int = 0
|
|
is_tls: bool = False # True when only TLS metadata is visible
|
|
src_ip: Optional[str] = None
|
|
dst_ip: Optional[str] = None
|
|
dst_port: Optional[int] = None
|
|
timestamp: Optional[float] = None
|
|
stream_id: Optional[int] = None
|
|
packet_number: Optional[int] = None
|
|
|
|
|
|
@dataclass
|
|
class HttpResponse:
|
|
status_code: Optional[int] = None
|
|
content_type: Optional[str] = None
|
|
server: Optional[str] = None
|
|
content_disposition: Optional[str] = None
|
|
content_length: Optional[int] = None
|
|
set_cookie: list[str] = field(default_factory=list)
|
|
www_authenticate: Optional[str] = None
|
|
request_uri: Optional[str] = None
|
|
src_ip: Optional[str] = None
|
|
dst_ip: Optional[str] = None
|
|
src_port: Optional[int] = None
|
|
timestamp: Optional[float] = None
|
|
stream_id: Optional[int] = None
|
|
packet_number: Optional[int] = None
|
|
|
|
|
|
@dataclass
|
|
class TlsSession:
|
|
"""One TLS handshake, aggregated per TCP stream (or QUIC connection)."""
|
|
|
|
src_ip: Optional[str] = None
|
|
dst_ip: Optional[str] = None
|
|
dst_port: Optional[int] = 443
|
|
src_port: Optional[int] = None
|
|
transport: str = "tcp" # "tcp" or "quic"
|
|
version: Optional[str] = None # Negotiated version from the ServerHello
|
|
cipher: Optional[str] = None # Negotiated cipher-suite name from the ServerHello
|
|
sni: Optional[str] = None
|
|
alpn: list[str] = field(default_factory=list)
|
|
client_version: Optional[str] = None # Highest version the client offered
|
|
client_offered_versions: list[str] = field(default_factory=list)
|
|
client_offered_ciphers: list[str] = field(default_factory=list)
|
|
client_hello_seen: bool = False
|
|
server_hello_seen: bool = False
|
|
cert_subject_cn: Optional[str] = None
|
|
cert_issuer_cn: Optional[str] = None
|
|
cert_san: list[str] = field(default_factory=list)
|
|
cert_not_before: Optional[str] = None
|
|
cert_not_after: Optional[str] = None
|
|
cert_self_signed: Optional[bool] = None
|
|
cert_expired: Optional[bool] = None
|
|
cert_sig_algorithm: Optional[str] = None
|
|
cert_key_type: Optional[str] = None
|
|
cert_key_bits: Optional[int] = None
|
|
cert_seen: bool = False
|
|
client_cert_seen: bool = False # Client presented a certificate (mutual TLS)
|
|
alerts: list[str] = field(default_factory=list)
|
|
timestamp: Optional[float] = None
|
|
stream_id: Optional[int] = None
|
|
packet_number: Optional[int] = None
|
|
packet_numbers: list[int] = field(default_factory=list)
|
|
|
|
|
|
@dataclass
|
|
class AuthMaterial:
|
|
"""Hashes, challenge-response material, and authentication tickets."""
|
|
|
|
kind: str # "ntlmssp", "netntlmv2", "kerberos_asrep", ...
|
|
value: Optional[str] = None
|
|
account: Optional[str] = None
|
|
domain: Optional[str] = None
|
|
src_ip: Optional[str] = None
|
|
dst_ip: Optional[str] = None
|
|
protocol: Optional[str] = None
|
|
timestamp: Optional[float] = None
|
|
packet_number: Optional[int] = None
|
|
|
|
# Crackable material, when reconstruction is possible.
|
|
hashcat: Optional[str] = None # Complete hashcat input line
|
|
hashcat_mode: Optional[int] = None # For example, 5600 (NetNTLMv2)
|
|
hash_format: Optional[str] = None # Human-readable format name
|
|
server_challenge: Optional[str] = None # NTLMSSP server challenge (hex)
|
|
nt_response: Optional[str] = None # NTLMv2 response (hex)
|
|
etype: Optional[str] = None # Kerberos encryption type
|
|
|
|
|
|
@dataclass
|
|
class ArpEvent:
|
|
sender_ip: Optional[str] = None
|
|
sender_mac: Optional[str] = None
|
|
target_ip: Optional[str] = None
|
|
opcode: Optional[int] = None # 1=request, 2=reply
|
|
is_gratuitous: bool = False
|
|
timestamp: Optional[float] = None
|
|
packet_number: Optional[int] = None
|
|
|
|
|
|
@dataclass
|
|
class DhcpEvent:
|
|
message_type: Optional[str] = None # "discover", "offer", "request", "ack", ...
|
|
client_mac: Optional[str] = None
|
|
client_ip: Optional[str] = None # Assigned/requested address when known
|
|
hostname: Optional[str] = None
|
|
vendor_class: Optional[str] = None
|
|
server_id: Optional[str] = None
|
|
src_ip: Optional[str] = None
|
|
timestamp: Optional[float] = None
|
|
packet_number: Optional[int] = None
|
|
|
|
|
|
@dataclass
|
|
class MqttEvent:
|
|
"""MQTT control packet visible in clear text (port 1883 or any unencrypted flow)."""
|
|
|
|
msg_type: Optional[str] = None # "connect", "connack", "publish", "subscribe", ...
|
|
client_id: Optional[str] = None
|
|
username: Optional[str] = None
|
|
password: Optional[str] = None
|
|
topic: Optional[str] = None
|
|
return_code: Optional[int] = None # CONNACK return code
|
|
src_ip: Optional[str] = None
|
|
dst_ip: Optional[str] = None
|
|
dst_port: Optional[int] = None
|
|
timestamp: Optional[float] = None
|
|
stream_id: Optional[int] = None
|
|
packet_number: Optional[int] = None
|
|
|
|
|
|
@dataclass
|
|
class ServiceEvent:
|
|
"""Generic clear-text protocol observation (SNMP community, SSH banner, ...)."""
|
|
|
|
protocol: str
|
|
kind: str # "snmp_community", "ssh_banner", "coap_request", ...
|
|
value: Optional[str] = None
|
|
detail: Optional[str] = None
|
|
src_ip: Optional[str] = None
|
|
dst_ip: Optional[str] = None
|
|
dst_port: Optional[int] = None
|
|
timestamp: Optional[float] = None
|
|
packet_number: Optional[int] = None
|
|
|
|
|
|
@dataclass
|
|
class FileTransfer:
|
|
filename: Optional[str] = None
|
|
extension: Optional[str] = None
|
|
content_type: Optional[str] = None
|
|
size: Optional[int] = None
|
|
protocol: Optional[str] = None # "http", "ftp-data", "smb", "tftp"
|
|
src_ip: Optional[str] = None
|
|
dst_ip: Optional[str] = None
|
|
direction: Optional[str] = None # "download" / "upload"
|
|
uri: Optional[str] = None
|
|
host: Optional[str] = None
|
|
timestamp: Optional[float] = None
|
|
stream_id: Optional[int] = None
|
|
packet_number: Optional[int] = None
|
|
|
|
|
|
@dataclass
|
|
class PlaintextAuth:
|
|
"""Clear-text authentication observed in an application protocol."""
|
|
|
|
protocol: str
|
|
username: Optional[str] = None
|
|
password: Optional[str] = None
|
|
src_ip: Optional[str] = None
|
|
dst_ip: Optional[str] = None
|
|
dst_port: Optional[int] = None
|
|
# "success", "failure" or None when the server response was not observed.
|
|
outcome: Optional[str] = None
|
|
timestamp: Optional[float] = None
|
|
stream_id: Optional[int] = None
|
|
packet_number: Optional[int] = None
|
|
|
|
|
|
@dataclass
|
|
class AuthAttempt:
|
|
"""A single authentication attempt result used for brute-force detection."""
|
|
|
|
protocol: str
|
|
success: Optional[bool]
|
|
src_ip: Optional[str] = None # Client
|
|
dst_ip: Optional[str] = None # Server
|
|
dst_port: Optional[int] = None
|
|
username: Optional[str] = None
|
|
timestamp: Optional[float] = None
|
|
stream_id: Optional[int] = None
|
|
packet_number: Optional[int] = None
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Central capture context
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
@dataclass
|
|
class CaptureContext:
|
|
"""All detector input extracted during a single capture pass."""
|
|
|
|
path: Optional[str] = None
|
|
packet_count: int = 0
|
|
# Set when limits stopped the extraction pass before the capture was fully read.
|
|
truncated: bool = False
|
|
truncation_reason: Optional[str] = None
|
|
truncation_reasons: list[str] = field(default_factory=list)
|
|
# Non-fatal coverage notes (protocols not dissected, undecoded ports, ...).
|
|
coverage_notes: list[str] = field(default_factory=list)
|
|
start_time: Optional[float] = None
|
|
end_time: Optional[float] = None
|
|
extractor: Optional[str] = None # e.g. "tshark 4.2.2"
|
|
|
|
# The device under test. When set, findings are attributed to it.
|
|
device_ip: Optional[str] = None
|
|
device_macs: set[str] = field(default_factory=set)
|
|
|
|
hosts: dict[str, Host] = field(default_factory=dict)
|
|
conversations: list[Conversation] = field(default_factory=list)
|
|
dns_queries: list[DnsQuery] = field(default_factory=list)
|
|
http_requests: list[HttpRequest] = field(default_factory=list)
|
|
http_responses: list[HttpResponse] = field(default_factory=list)
|
|
tls_sessions: list[TlsSession] = field(default_factory=list)
|
|
auth_material: list[AuthMaterial] = field(default_factory=list)
|
|
arp_events: list[ArpEvent] = field(default_factory=list)
|
|
dhcp_events: list[DhcpEvent] = field(default_factory=list)
|
|
mqtt_events: list[MqttEvent] = field(default_factory=list)
|
|
service_events: list[ServiceEvent] = field(default_factory=list)
|
|
file_transfers: list[FileTransfer] = field(default_factory=list)
|
|
plaintext_auth: list[PlaintextAuth] = field(default_factory=list)
|
|
auth_attempts: list[AuthAttempt] = field(default_factory=list)
|
|
|
|
# protocol_stats: highest-layer protocol name -> packet count (legacy, kept for detectors)
|
|
protocol_stats: dict[str, int] = field(default_factory=dict)
|
|
# app_protocol_stats: every protocol in frame.protocols -> packet count
|
|
app_protocol_stats: dict[str, int] = field(default_factory=dict)
|
|
icmp_stats: dict[str, int] = field(default_factory=dict)
|
|
vlan_ids: set[int] = field(default_factory=set)
|
|
# Generic text payloads with packet metadata for secret detection.
|
|
text_payloads: list[dict] = field(default_factory=list)
|
|
|
|
@property
|
|
def duration(self) -> float:
|
|
if self.start_time is None or self.end_time is None:
|
|
return 0.0
|
|
return max(0.0, self.end_time - self.start_time)
|
|
|
|
def host(self, ip: str) -> Host:
|
|
h = self.hosts.get(ip)
|
|
if h is None:
|
|
h = Host(ip=ip)
|
|
self.hosts[ip] = h
|
|
return h
|
|
|
|
def is_external(self, ip: Optional[str]) -> bool:
|
|
"""Return whether an address is globally routable."""
|
|
if not ip:
|
|
return False
|
|
return not _is_private(ip)
|
|
|
|
def is_device(self, ip: Optional[str]) -> bool:
|
|
"""Whether the address belongs to the device under test."""
|
|
return bool(ip) and self.device_ip is not None and ip == self.device_ip
|
|
|
|
def involves_device(self, *ips: Optional[str]) -> Optional[bool]:
|
|
"""True/False when a device is configured, None when no device is known."""
|
|
if self.device_ip is None:
|
|
return None
|
|
return any(self.is_device(ip) for ip in ips)
|
|
|
|
def protocol_count(self, *names: str) -> int:
|
|
"""Return a case-insensitive packet total for the requested protocols.
|
|
|
|
Both the highest-layer statistics and the complete protocol chain are
|
|
consulted so that e.g. "smb" is found even when a higher layer sits on top.
|
|
"""
|
|
wanted = {n.lower() for n in names}
|
|
total = sum(
|
|
count for proto, count in self.app_protocol_stats.items()
|
|
if str(proto).lower() in wanted
|
|
)
|
|
if total:
|
|
return total
|
|
return sum(
|
|
count for proto, count in self.protocol_stats.items()
|
|
if str(proto).lower() in wanted
|
|
)
|
|
|
|
def names_for_ip(self, ip: Optional[str]) -> set[str]:
|
|
"""DNS names that resolved to an address, plus SNI/Host names used with it."""
|
|
names: set[str] = set()
|
|
if not ip:
|
|
return names
|
|
for q in self.dns_queries:
|
|
if ip in q.response_ips and q.query_name:
|
|
names.add(q.query_name.rstrip(".").lower())
|
|
for t in self.tls_sessions:
|
|
if t.dst_ip == ip and t.sni:
|
|
names.add(t.sni.lower())
|
|
for r in self.http_requests:
|
|
if r.dst_ip == ip and r.host:
|
|
names.add(str(r.host).split(":")[0].lower())
|
|
return names
|
|
|
|
|
|
def _is_private(ip: str) -> bool:
|
|
try:
|
|
addr = ipaddress.ip_address(ip)
|
|
except ValueError:
|
|
return False
|
|
return (
|
|
addr.is_private
|
|
or addr.is_loopback
|
|
or addr.is_link_local
|
|
or addr.is_multicast
|
|
or addr.is_reserved
|
|
or addr.is_unspecified
|
|
)
|
|
|
|
|
|
def is_special(ip: Optional[str]) -> bool:
|
|
"""Broadcast, multicast, unspecified and link-scope addresses that never identify one host."""
|
|
if not ip:
|
|
return True
|
|
if ip in ("0.0.0.0", "255.255.255.255", "::"):
|
|
return True
|
|
try:
|
|
addr = ipaddress.ip_address(ip)
|
|
except ValueError:
|
|
return True
|
|
return addr.is_multicast or addr.is_unspecified or addr.is_loopback
|