1501 lines
67 KiB
Python
1501 lines
67 KiB
Python
"""Build a normalized capture context by driving tshark directly.
|
|
|
|
Pass 1 streams packet metadata for every packet, pass 2 streams payload text
|
|
from unencrypted flows only. Captures are read completely by default.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import time
|
|
from collections import defaultdict
|
|
from typing import Optional
|
|
|
|
from .context import (
|
|
CaptureContext,
|
|
Conversation,
|
|
DnsQuery,
|
|
HttpRequest,
|
|
HttpResponse,
|
|
TlsSession,
|
|
ArpEvent,
|
|
DhcpEvent,
|
|
MqttEvent,
|
|
ServiceEvent,
|
|
PlaintextAuth,
|
|
AuthAttempt,
|
|
AuthMaterial,
|
|
FileTransfer,
|
|
ENCRYPTED_PROTOCOLS,
|
|
is_special,
|
|
)
|
|
from . import crackformats
|
|
from . import tls_tables
|
|
|
|
logger = logging.getLogger("analyzer.extract")
|
|
|
|
# Known clear-text protocols used by extraction heuristics.
|
|
CLEARTEXT_PORTS = {
|
|
21: "ftp", 23: "telnet", 25: "smtp", 80: "http", 110: "pop3", 143: "imap",
|
|
389: "ldap", 161: "snmp", 69: "tftp", 1883: "mqtt", 5683: "coap", 554: "rtsp",
|
|
502: "modbus", 8080: "http", 8000: "http", 8888: "http",
|
|
}
|
|
|
|
# Payload retention limits (safety net; only clear-text flows are retained).
|
|
_MAX_TEXT_PAYLOADS = int(os.environ.get("CAPTURE_MAX_TEXT_PAYLOADS", "50000") or 0) or 50000
|
|
_MAX_PAYLOAD_CHARS = int(os.environ.get("CAPTURE_MAX_PAYLOAD_CHARS", "16384") or 0) or 16384
|
|
_MAX_TIMESTAMPS_PER_FLOW = 5000
|
|
|
|
# Fields exported in the metadata pass. Order does not matter.
|
|
_META_FIELDS = [
|
|
"frame.number", "frame.time_epoch", "frame.len", "frame.protocols",
|
|
"eth.src", "eth.dst", "vlan.id",
|
|
"ip.src", "ip.dst", "ipv6.src", "ipv6.dst",
|
|
"tcp.srcport", "tcp.dstport", "tcp.stream", "tcp.flags", "tcp.len",
|
|
"tcp.analysis.retransmission",
|
|
"udp.srcport", "udp.dstport", "udp.stream", "udp.length",
|
|
"icmp.type", "icmp.code", "icmpv6.type",
|
|
"arp.opcode", "arp.src.proto_ipv4", "arp.src.hw_mac", "arp.dst.proto_ipv4",
|
|
"arp.isgratuitous",
|
|
"dns.flags.response", "dns.qry.name", "dns.qry.type", "dns.flags.rcode",
|
|
"dns.a", "dns.aaaa", "dns.cname", "dns.txt",
|
|
"dhcp.option.dhcp", "dhcp.hw.mac_addr", "dhcp.option.hostname",
|
|
"dhcp.option.vendor_class_id", "dhcp.option.dhcp_server_id", "dhcp.ip.your",
|
|
"dhcp.option.requested_ip_address",
|
|
"http.request.method", "http.host", "http.request.uri", "http.user_agent",
|
|
"http.authorization", "http.cookie", "http.content_type", "http.content_length",
|
|
"http.file_data", "http.request.line", "http.response.code", "http.response.line",
|
|
"http.server", "http.set_cookie", "http.www_authenticate", "http.response_for.uri",
|
|
"http.location",
|
|
"tls.handshake.type", "tls.handshake.version",
|
|
"tls.handshake.extensions.supported_version", "tls.handshake.ciphersuite",
|
|
"tls.handshake.extensions_server_name", "tls.handshake.extensions_alpn_str",
|
|
"tls.handshake.certificate", "tls.alert_message.desc", "tls.record.version",
|
|
"ftp.request.command", "ftp.request.arg", "ftp.response.code",
|
|
"telnet.data",
|
|
"pop.request.command", "pop.request.parameter", "pop.response.indicator",
|
|
"imap.request", "imap.request.command", "imap.response.status",
|
|
"smtp.req.command", "smtp.req.parameter", "smtp.response.code",
|
|
"ntlmssp.messagetype", "ntlmssp.ntlmserverchallenge", "ntlmssp.auth.username",
|
|
"ntlmssp.auth.domain", "ntlmssp.ntlmv2_response", "ntlmssp.auth.ntresponse",
|
|
"ntlmssp.auth.lmresponse",
|
|
"kerberos.msg_type", "kerberos.CNameString", "kerberos.realm", "kerberos.crealm",
|
|
"kerberos.etype", "kerberos.cipher", "kerberos.SNameString",
|
|
"mqtt.msgtype", "mqtt.clientid", "mqtt.username", "mqtt.passwd", "mqtt.topic",
|
|
"mqtt.conack.val",
|
|
"coap.code", "coap.opt.uri_path",
|
|
"snmp.community", "snmp.version",
|
|
"ssh.protocol",
|
|
"rtsp.request", "rtsp.status",
|
|
"sip.Method", "sip.auth",
|
|
"tftp.opcode", "tftp.source_file", "tftp.destination_file",
|
|
"smb2.nt_status", "smb.nt_status",
|
|
]
|
|
|
|
_PAYLOAD_FIELDS = [
|
|
"frame.number", "frame.time_epoch", "frame.protocols",
|
|
"ip.src", "ip.dst", "ipv6.src", "ipv6.dst",
|
|
"tcp.stream", "tcp.srcport", "tcp.dstport", "udp.srcport", "udp.dstport",
|
|
"tcp.payload", "udp.payload", "data.data",
|
|
]
|
|
|
|
# tshark protocol names whose frames carry ciphertext or binary material only.
|
|
_NO_TEXT_PROTOCOLS = ENCRYPTED_PROTOCOLS | {"eapol", "ipsec", "l2tp", "gre"}
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Public entry point
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def build_context_from_pcap(path: str, packet_limit: Optional[int] = None,
|
|
device_ip: Optional[str] = None,
|
|
tshark_path: Optional[str] = None) -> CaptureContext:
|
|
"""Read a capture and return its normalized context.
|
|
|
|
``packet_limit`` (or CAPTURE_PACKET_LIMIT) bounds both passes; 0 reads the
|
|
whole capture. ``device_ip`` marks the device under test.
|
|
"""
|
|
if not os.path.exists(path):
|
|
raise FileNotFoundError(f"Capture not found: {path}")
|
|
|
|
if packet_limit is None:
|
|
try:
|
|
packet_limit = int(os.environ.get("CAPTURE_PACKET_LIMIT", "0"))
|
|
except ValueError:
|
|
packet_limit = 0
|
|
packet_limit = packet_limit if packet_limit and packet_limit > 0 else None
|
|
|
|
tshark = tshark_path or os.environ.get("TSHARK_PATH") or shutil.which("tshark")
|
|
if not tshark or not os.path.exists(tshark):
|
|
logger.error("tshark not found (TSHARK_PATH=%r, PATH lookup failed)", os.environ.get("TSHARK_PATH"))
|
|
raise RuntimeError(
|
|
"tshark was not found. Install Wireshark/tshark or use the backend container."
|
|
)
|
|
|
|
ctx = CaptureContext(path=path, device_ip=device_ip)
|
|
ctx.extractor = _tshark_version(tshark)
|
|
try:
|
|
size = os.path.getsize(path)
|
|
except OSError:
|
|
size = -1
|
|
logger.info("Extraction start: file=%s size=%d bytes tshark=%s (%s) packet_limit=%s device_ip=%s "
|
|
"max_text_payloads=%d", path, size, ctx.extractor, tshark, packet_limit, device_ip,
|
|
_MAX_TEXT_PAYLOADS)
|
|
|
|
ex = _Extractor(ctx)
|
|
t0 = time.time()
|
|
ex.run_metadata_pass(tshark, path, packet_limit)
|
|
t1 = time.time()
|
|
logger.info("Metadata pass done: %d packets in %.1fs (%d skipped as malformed)",
|
|
ctx.packet_count, t1 - t0, ex.skipped_packets)
|
|
ex.finish_metadata()
|
|
logger.info("Records: hosts=%d flows=%d dns=%d http_req=%d http_resp=%d tls=%d arp=%d dhcp=%d "
|
|
"mqtt=%d auth=%d auth_material=%d files=%d services=%d encrypted_tcp_streams=%d",
|
|
len(ctx.hosts), len(ctx.conversations), len(ctx.dns_queries), len(ctx.http_requests),
|
|
len(ctx.http_responses), len(ctx.tls_sessions), len(ctx.arp_events), len(ctx.dhcp_events),
|
|
len(ctx.mqtt_events), len(ctx.plaintext_auth), len(ctx.auth_material),
|
|
len(ctx.file_transfers), len(ctx.service_events), len(ex.encrypted_tcp_streams))
|
|
t2 = time.time()
|
|
ex.run_payload_pass(tshark, path, packet_limit)
|
|
logger.info("Payload pass done: %d clear-text payloads retained in %.1fs", len(ctx.text_payloads),
|
|
time.time() - t2)
|
|
ex.finalize()
|
|
if ctx.truncated:
|
|
logger.warning("Extraction truncated: %s", ctx.truncation_reason)
|
|
for note in ctx.coverage_notes:
|
|
logger.info("Coverage: %s", note)
|
|
logger.info("Extraction finished in %.1fs", time.time() - t0)
|
|
return ctx
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Extraction state machine
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
class _Extractor:
|
|
def __init__(self, ctx: CaptureContext):
|
|
self.ctx = ctx
|
|
self.conv_index: dict[tuple, Conversation] = {}
|
|
self.tls_index: dict[tuple, TlsSession] = {}
|
|
self.encrypted_tcp_streams: set[int] = set()
|
|
self.encrypted_udp_streams: set[int] = set()
|
|
self.ntlm_chal: dict = {}
|
|
self.smtp_auth: dict = {}
|
|
self.telnet_buf: dict = defaultdict(lambda: {"c2s": [], "s2c": [], "meta": None})
|
|
self.mqtt_connect: dict = {}
|
|
self.last_auth: dict[tuple, PlaintextAuth] = {}
|
|
self.dhcp_by_mac: dict[str, dict] = {}
|
|
self.packets_seen = 0
|
|
self.skipped_packets = 0
|
|
self.last_frame = 0
|
|
|
|
# ---- tshark invocation --------------------------------------------- #
|
|
|
|
def _run(self, tshark: str, path: str, fields: list[str], packet_limit,
|
|
display_filter: Optional[str] = None):
|
|
cmd = [tshark, "-r", path, "-n", "-l", "-T", "ek", "-o", "tcp.desegment_tcp_streams:TRUE"]
|
|
if packet_limit:
|
|
cmd += ["-c", str(packet_limit)]
|
|
if display_filter:
|
|
cmd += ["-Y", display_filter]
|
|
for f in supported_fields(tshark, fields):
|
|
cmd += ["-e", f]
|
|
logger.debug("tshark command: %s", " ".join(cmd))
|
|
started = time.time()
|
|
records = 0
|
|
try:
|
|
proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
|
bufsize=1024 * 1024)
|
|
except OSError as exc:
|
|
logger.error("tshark could not be started: %s", exc)
|
|
raise RuntimeError(f"tshark could not be started: {exc}") from exc
|
|
try:
|
|
assert proc.stdout is not None
|
|
for raw in proc.stdout:
|
|
records += 1
|
|
if not raw.startswith(b'{"timestamp"') and b'"layers"' not in raw:
|
|
continue
|
|
try:
|
|
rec = json.loads(raw)
|
|
except ValueError:
|
|
continue
|
|
layers = rec.get("layers")
|
|
if isinstance(layers, dict):
|
|
yield layers
|
|
finally:
|
|
try:
|
|
err = proc.stderr.read().decode("utf-8", "ignore") if proc.stderr else ""
|
|
except Exception:
|
|
err = ""
|
|
rc = proc.wait()
|
|
err_lines = [ln for ln in err.strip().splitlines() if ln.strip()
|
|
and "Running as user" not in ln]
|
|
logger.info("tshark pass finished: rc=%s records=%d %.1fs filter=%s",
|
|
rc, records, time.time() - started, display_filter or "-")
|
|
for ln in err_lines[:20]:
|
|
logger.warning("tshark stderr: %s", ln)
|
|
err_text = " | ".join(err_lines)[:400]
|
|
if rc not in (0, None) and "isn't a capture file" in err:
|
|
raise RuntimeError(f"tshark could not read the capture: {err_text}")
|
|
if rc not in (0, None) and records == 0:
|
|
raise RuntimeError(f"tshark failed (exit {rc}): {err_text or 'no output'}")
|
|
|
|
def run_metadata_pass(self, tshark, path, packet_limit):
|
|
for layers in self._run(tshark, path, _META_FIELDS, packet_limit):
|
|
self.packets_seen += 1
|
|
try:
|
|
self._handle_packet(layers)
|
|
except Exception as exc:
|
|
# One malformed packet must never abort the complete analysis.
|
|
self.skipped_packets += 1
|
|
if self.skipped_packets <= 5:
|
|
logger.debug("Packet %s skipped: %r", layers.get("frame_number"), exc)
|
|
continue
|
|
self.ctx.packet_count = self.packets_seen
|
|
if packet_limit and self.packets_seen >= packet_limit:
|
|
_mark_truncated(
|
|
self.ctx,
|
|
f"Analysis stopped at the configured limit of {packet_limit} packets; "
|
|
"the capture may contain more traffic than was inspected.",
|
|
)
|
|
|
|
def run_payload_pass(self, tshark, path, packet_limit):
|
|
ctx = self.ctx
|
|
parts = ["(tcp.len > 0 || udp.length > 8)",
|
|
"!(tls || ssl || dtls || quic || ssh || esp || kerberos || ntlmssp || eapol)"]
|
|
# Skip whole flows that carried an encrypted protocol at any point.
|
|
# Set members are comma-separated (required since Wireshark 4.4).
|
|
tcp_ex = sorted(self.encrypted_tcp_streams)
|
|
udp_ex = sorted(self.encrypted_udp_streams)
|
|
if tcp_ex and len(tcp_ex) <= 20000:
|
|
parts.append("!(tcp.stream in {" + ", ".join(map(str, tcp_ex)) + "})")
|
|
elif tcp_ex:
|
|
parts.append("!(tcp.port in {443, 465, 563, 636, 853, 993, 995, 5223, 5228, 8443, 8883})")
|
|
ctx.coverage_notes.append(
|
|
"More than 20000 encrypted TCP flows: payload scanning excluded only well-known TLS ports.")
|
|
if udp_ex and len(udp_ex) <= 20000:
|
|
parts.append("!(udp.stream in {" + ", ".join(map(str, udp_ex)) + "})")
|
|
filt = " && ".join(parts)
|
|
|
|
count = 0
|
|
for layers in self._run(tshark, path, _PAYLOAD_FIELDS, packet_limit, filt):
|
|
if len(ctx.text_payloads) >= _MAX_TEXT_PAYLOADS:
|
|
_mark_truncated(
|
|
ctx,
|
|
f"Retained payload text reached the cap of {_MAX_TEXT_PAYLOADS} entries; "
|
|
"later clear-text payloads were not scanned for secrets.",
|
|
)
|
|
break
|
|
try:
|
|
self._handle_payload(layers)
|
|
count += 1
|
|
except Exception as exc:
|
|
logger.debug("Payload record skipped: %r", exc)
|
|
continue
|
|
|
|
# ---- helpers --------------------------------------------------------- #
|
|
|
|
@staticmethod
|
|
def _first(layers, key) -> Optional[str]:
|
|
v = layers.get(key)
|
|
if isinstance(v, list):
|
|
return v[0] if v else None
|
|
return v
|
|
|
|
@staticmethod
|
|
def _all(layers, key) -> list[str]:
|
|
v = layers.get(key)
|
|
if v is None:
|
|
return []
|
|
if isinstance(v, list):
|
|
return [str(x) for x in v]
|
|
return [str(v)]
|
|
|
|
def _int(self, layers, key) -> Optional[int]:
|
|
return tls_tables.parse_int(self._first(layers, key))
|
|
|
|
def _bool(self, layers, key) -> bool:
|
|
v = self._first(layers, key)
|
|
return str(v).lower() in ("1", "true") if v is not None else False
|
|
|
|
# ---- per-packet handling -------------------------------------------- #
|
|
|
|
def _handle_packet(self, L: dict) -> None:
|
|
ctx = self.ctx
|
|
F, A, I = self._first, self._all, self._int
|
|
|
|
pnum = I(L, "frame_number")
|
|
if pnum is not None:
|
|
self.last_frame = pnum
|
|
ts = _float(F(L, "frame_time_epoch"))
|
|
if ts is not None:
|
|
ctx.start_time = ts if ctx.start_time is None else min(ctx.start_time, ts)
|
|
ctx.end_time = ts if ctx.end_time is None else max(ctx.end_time, ts)
|
|
|
|
protos_raw = F(L, "frame_protocols") or ""
|
|
protos = [p for p in protos_raw.split(":") if p]
|
|
proto_set = set(protos)
|
|
top = _highest_layer(protos)
|
|
ctx.protocol_stats[top] = ctx.protocol_stats.get(top, 0) + 1
|
|
for p in proto_set:
|
|
ctx.app_protocol_stats[p] = ctx.app_protocol_stats.get(p, 0) + 1
|
|
|
|
vid = I(L, "vlan_id")
|
|
if vid is not None:
|
|
ctx.vlan_ids.add(vid)
|
|
|
|
length = I(L, "frame_len") or 0
|
|
src_ip = F(L, "ip_src") or F(L, "ipv6_src")
|
|
dst_ip = F(L, "ip_dst") or F(L, "ipv6_dst")
|
|
src_mac, dst_mac = F(L, "eth_src"), F(L, "eth_dst")
|
|
|
|
transport = None
|
|
src_port = dst_port = None
|
|
stream_id = None
|
|
if "tcp" in proto_set:
|
|
transport = "tcp"
|
|
src_port, dst_port = I(L, "tcp_srcport"), I(L, "tcp_dstport")
|
|
stream_id = I(L, "tcp_stream")
|
|
elif "udp" in proto_set:
|
|
transport = "udp"
|
|
src_port, dst_port = I(L, "udp_srcport"), I(L, "udp_dstport")
|
|
stream_id = I(L, "udp_stream")
|
|
|
|
app_protos = proto_set - {"eth", "ethertype", "vlan", "ip", "ipv6", "tcp", "udp",
|
|
"data", "llc", "sll", "frame", "tcp.segments"}
|
|
|
|
# Host bookkeeping.
|
|
if src_ip:
|
|
h = ctx.host(src_ip)
|
|
h.bytes_sent += length
|
|
h.packets_sent += 1
|
|
if ts is not None:
|
|
h.first_seen = ts if h.first_seen is None else min(h.first_seen, ts)
|
|
h.last_seen = ts if h.last_seen is None else max(h.last_seen, ts)
|
|
if src_mac and not is_special(src_ip):
|
|
h.macs.add(src_mac)
|
|
if ctx.is_device(src_ip):
|
|
ctx.device_macs.add(src_mac)
|
|
h.app_protocols.update(app_protos)
|
|
if dst_ip:
|
|
h = ctx.host(dst_ip)
|
|
h.bytes_received += length
|
|
h.packets_received += 1
|
|
if ts is not None:
|
|
h.first_seen = ts if h.first_seen is None else min(h.first_seen, ts)
|
|
h.last_seen = ts if h.last_seen is None else max(h.last_seen, ts)
|
|
h.app_protocols.update(app_protos)
|
|
|
|
# TCP flags.
|
|
flags = I(L, "tcp_flags") or 0
|
|
is_syn = bool(flags & 0x02)
|
|
is_ack = bool(flags & 0x10)
|
|
is_rst = bool(flags & 0x04)
|
|
is_fin = bool(flags & 0x01)
|
|
retrans = self._bool(L, "tcp_analysis_retransmission")
|
|
tcp_len = I(L, "tcp_len") or 0
|
|
udp_len = I(L, "udp_length")
|
|
payload_len = tcp_len if transport == "tcp" else max(0, (udp_len or 8) - 8) if transport == "udp" else 0
|
|
|
|
# Conversation aggregation (bidirectional 5-tuple).
|
|
conv = None
|
|
if src_ip and dst_ip and transport:
|
|
key = (tuple(sorted([(src_ip, src_port or 0), (dst_ip, dst_port or 0)])), transport)
|
|
conv = self.conv_index.get(key)
|
|
if conv is None:
|
|
conv = Conversation(src_ip=src_ip, dst_ip=dst_ip, src_port=src_port,
|
|
dst_port=dst_port, protocol=transport,
|
|
first_seen=ts, last_seen=ts, stream_id=stream_id)
|
|
self.conv_index[key] = conv
|
|
if dst_port:
|
|
ctx.host(src_ip).ports_contacted.add(dst_port)
|
|
conv.packets += 1
|
|
conv.bytes += length
|
|
conv.payload_bytes += payload_len
|
|
if retrans:
|
|
conv.retransmissions += 1
|
|
if pnum is not None and len(conv.packet_numbers) < 20:
|
|
conv.packet_numbers.append(pnum)
|
|
if ts is not None:
|
|
if len(conv.timestamps) < _MAX_TIMESTAMPS_PER_FLOW:
|
|
conv.timestamps.append(ts)
|
|
conv.last_seen = ts if conv.last_seen is None else max(conv.last_seen, ts)
|
|
conv.first_seen = ts if conv.first_seen is None else min(conv.first_seen, ts)
|
|
conv.app_protocols.update(app_protos)
|
|
forward = (src_ip == conv.src_ip and (src_port or 0) == (conv.src_port or 0))
|
|
if forward:
|
|
conv.bytes_src_to_dst += length
|
|
conv.packets_src_to_dst += 1
|
|
else:
|
|
conv.bytes_dst_to_src += length
|
|
conv.packets_dst_to_src += 1
|
|
# A reply from the responder marks its port as an offered service.
|
|
if src_port and src_ip:
|
|
ctx.host(src_ip).ports_offered.add(src_port)
|
|
if transport == "tcp":
|
|
if is_syn and not is_ack:
|
|
conv.syn_count += 1
|
|
elif is_syn and is_ack:
|
|
conv.synack_count += 1
|
|
if is_rst:
|
|
conv.rst_count += 1
|
|
if is_fin:
|
|
conv.fin_count += 1
|
|
if app_protos & ENCRYPTED_PROTOCOLS and stream_id is not None:
|
|
(self.encrypted_tcp_streams if transport == "tcp" else self.encrypted_udp_streams).add(stream_id)
|
|
|
|
meta = dict(src_ip=src_ip, dst_ip=dst_ip, src_port=src_port, dst_port=dst_port,
|
|
ts=ts, pnum=pnum, stream_id=stream_id, transport=transport, protos=proto_set)
|
|
|
|
# Protocol-specific extraction.
|
|
if "arp" in proto_set:
|
|
self._arp(L, meta)
|
|
if "icmp" in proto_set or "icmpv6" in proto_set:
|
|
self._icmp(L, meta)
|
|
if "dns" in proto_set or "mdns" in proto_set or "llmnr" in proto_set:
|
|
self._dns(L, meta)
|
|
if "dhcp" in proto_set or "bootp" in proto_set:
|
|
self._dhcp(L, meta)
|
|
if "http" in proto_set and "ssdp" not in proto_set:
|
|
self._http(L, meta)
|
|
elif "ssdp" in proto_set:
|
|
self._ssdp(L, meta)
|
|
if "tls" in proto_set or "ssl" in proto_set or "quic" in proto_set:
|
|
self._tls(L, meta)
|
|
if "ftp" in proto_set:
|
|
self._ftp(L, meta)
|
|
if "telnet" in proto_set:
|
|
self._telnet(L, meta)
|
|
if "pop" in proto_set:
|
|
self._pop3(L, meta)
|
|
if "imap" in proto_set:
|
|
self._imap(L, meta)
|
|
if "smtp" in proto_set:
|
|
self._smtp(L, meta)
|
|
if "ntlmssp" in proto_set:
|
|
self._ntlm(L, meta)
|
|
if "kerberos" in proto_set:
|
|
self._kerberos(L, meta)
|
|
if "mqtt" in proto_set:
|
|
self._mqtt(L, meta)
|
|
if "snmp" in proto_set:
|
|
self._snmp(L, meta)
|
|
if "ssh" in proto_set:
|
|
self._ssh(L, meta)
|
|
if "coap" in proto_set:
|
|
self._coap(L, meta)
|
|
if "rtsp" in proto_set:
|
|
self._rtsp(L, meta)
|
|
if "sip" in proto_set:
|
|
self._sip(L, meta)
|
|
if "tftp" in proto_set:
|
|
self._tftp(L, meta)
|
|
if "smb" in proto_set or "smb2" in proto_set:
|
|
self._smb(L, meta)
|
|
|
|
# ---- link / network layer ------------------------------------------- #
|
|
|
|
def _arp(self, L, m):
|
|
self.ctx.arp_events.append(ArpEvent(
|
|
sender_ip=self._first(L, "arp_src_proto_ipv4"),
|
|
sender_mac=self._first(L, "arp_src_hw_mac"),
|
|
target_ip=self._first(L, "arp_dst_proto_ipv4"),
|
|
opcode=self._int(L, "arp_opcode"),
|
|
is_gratuitous=self._bool(L, "arp_isgratuitous"),
|
|
timestamp=m["ts"], packet_number=m["pnum"],
|
|
))
|
|
|
|
def _icmp(self, L, m):
|
|
ctx = self.ctx
|
|
t4 = self._int(L, "icmp_type")
|
|
t6 = self._int(L, "icmpv6_type")
|
|
names4 = {0: "echo_reply", 3: "unreachable", 5: "redirect", 8: "echo_request",
|
|
11: "time_exceeded", 13: "timestamp_request"}
|
|
names6 = {128: "echo_request", 129: "echo_reply", 133: "router_solicitation",
|
|
134: "router_advertisement", 135: "neighbor_solicitation",
|
|
136: "neighbor_advertisement", 137: "redirect"}
|
|
name = None
|
|
if t4 is not None:
|
|
name = names4.get(t4, f"icmp_type_{t4}")
|
|
elif t6 is not None:
|
|
name = names6.get(t6, f"icmpv6_type_{t6}")
|
|
if name:
|
|
ctx.icmp_stats[name] = ctx.icmp_stats.get(name, 0) + 1
|
|
if name in ("redirect", "router_advertisement"):
|
|
ctx.service_events.append(ServiceEvent(
|
|
protocol="icmp" if t4 is not None else "icmpv6", kind=name,
|
|
src_ip=m["src_ip"], dst_ip=m["dst_ip"],
|
|
timestamp=m["ts"], packet_number=m["pnum"]))
|
|
|
|
def _dns(self, L, m):
|
|
qname = self._first(L, "dns_qry_name")
|
|
if not qname:
|
|
return
|
|
is_resp = self._bool(L, "dns_flags_response")
|
|
rcode = self._first(L, "dns_flags_rcode")
|
|
rcode_map = {"0": "NOERROR", "1": "FORMERR", "2": "SERVFAIL", "3": "NXDOMAIN",
|
|
"5": "REFUSED"}
|
|
transport = "mdns" if "mdns" in m["protos"] else "llmnr" if "llmnr" in m["protos"] else "dns"
|
|
q = DnsQuery(
|
|
query_name=qname,
|
|
qtype=_dns_qtype_name(self._first(L, "dns_qry_type")),
|
|
is_response=is_resp,
|
|
response_ips=self._all(L, "dns_a") + self._all(L, "dns_aaaa"),
|
|
cnames=self._all(L, "dns_cname"),
|
|
txt=self._all(L, "dns_txt"),
|
|
rcode=rcode_map.get(str(rcode), rcode) if is_resp else None,
|
|
timestamp=m["ts"], src_ip=m["src_ip"], dst_ip=m["dst_ip"],
|
|
resolver=m["src_ip"] if is_resp else m["dst_ip"],
|
|
transport=transport, packet_number=m["pnum"],
|
|
)
|
|
self.ctx.dns_queries.append(q)
|
|
if transport == "mdns" and is_resp and m["src_ip"] and qname.endswith(".local"):
|
|
self.ctx.host(m["src_ip"]).hostnames.add(qname.rstrip("."))
|
|
|
|
def _dhcp(self, L, m):
|
|
types = {"1": "discover", "2": "offer", "3": "request", "4": "decline",
|
|
"5": "ack", "6": "nak", "7": "release", "8": "inform"}
|
|
mtype = types.get(str(self._first(L, "dhcp_option_dhcp")), self._first(L, "dhcp_option_dhcp"))
|
|
mac = self._first(L, "dhcp_hw_mac_addr")
|
|
ev = DhcpEvent(
|
|
message_type=mtype, client_mac=mac,
|
|
client_ip=self._first(L, "dhcp_ip_your") if mtype in ("ack", "offer")
|
|
else self._first(L, "dhcp_option_requested_ip_address"),
|
|
hostname=self._first(L, "dhcp_option_hostname"),
|
|
vendor_class=self._first(L, "dhcp_option_vendor_class_id"),
|
|
server_id=self._first(L, "dhcp_option_dhcp_server_id"),
|
|
src_ip=m["src_ip"], timestamp=m["ts"], packet_number=m["pnum"],
|
|
)
|
|
self.ctx.dhcp_events.append(ev)
|
|
if mac:
|
|
info = self.dhcp_by_mac.setdefault(mac, {})
|
|
if ev.hostname:
|
|
info["hostname"] = ev.hostname
|
|
if ev.vendor_class:
|
|
info["vendor"] = ev.vendor_class
|
|
if ev.client_ip and ev.client_ip not in ("0.0.0.0",):
|
|
info["ip"] = ev.client_ip
|
|
|
|
# ---- HTTP -------------------------------------------------------------- #
|
|
|
|
def _http(self, L, m):
|
|
ctx = self.ctx
|
|
F, A = self._first, self._all
|
|
method = F(L, "http_request_method")
|
|
if method or F(L, "http_request_uri"):
|
|
headers = _parse_header_lines(A(L, "http_request_line"))
|
|
body = _decode_maybe_hex(F(L, "http_file_data"))
|
|
req = HttpRequest(
|
|
method=method, host=F(L, "http_host") or headers.get("host"),
|
|
uri=F(L, "http_request_uri"),
|
|
user_agent=F(L, "http_user_agent") or headers.get("user-agent"),
|
|
authorization=F(L, "http_authorization") or headers.get("authorization"),
|
|
cookies=F(L, "http_cookie") or headers.get("cookie"),
|
|
headers=headers, body=body,
|
|
content_type=F(L, "http_content_type") or headers.get("content-type"),
|
|
request_bytes=tls_tables.parse_int(F(L, "http_content_length")) or 0,
|
|
src_ip=m["src_ip"], dst_ip=m["dst_ip"], dst_port=m["dst_port"],
|
|
timestamp=m["ts"], stream_id=m["stream_id"], packet_number=m["pnum"],
|
|
)
|
|
ctx.http_requests.append(req)
|
|
if req.user_agent and m["src_ip"]:
|
|
ctx.host(m["src_ip"]).user_agents.add(req.user_agent[:120])
|
|
for piece in (req.uri, req.cookies, req.authorization, req.body):
|
|
if piece:
|
|
_add_text_payload(ctx, piece, "http", m["src_ip"], m["dst_ip"], m["pnum"], m["ts"])
|
|
# Uploads of files or firmware via HTTP.
|
|
ct = (req.content_type or "").lower()
|
|
if method and method.upper() in ("POST", "PUT") and (
|
|
_is_download_content_type(ct) or "multipart/form-data" in ct):
|
|
name = _filename_from_disposition(req.body[:2000] if req.body else None)
|
|
ctx.file_transfers.append(FileTransfer(
|
|
filename=name, extension=_ext_of(name or req.uri), content_type=req.content_type,
|
|
size=req.request_bytes or None, protocol="http",
|
|
src_ip=m["src_ip"], dst_ip=m["dst_ip"], direction="upload",
|
|
uri=req.uri, host=req.host, timestamp=m["ts"], stream_id=m["stream_id"],
|
|
packet_number=m["pnum"]))
|
|
code = F(L, "http_response_code")
|
|
if code:
|
|
headers = _parse_header_lines(A(L, "http_response_line"))
|
|
ct = F(L, "http_content_type") or headers.get("content-type")
|
|
disposition = headers.get("content-disposition")
|
|
resp = HttpResponse(
|
|
status_code=tls_tables.parse_int(code), content_type=ct,
|
|
server=F(L, "http_server") or headers.get("server"),
|
|
content_disposition=disposition,
|
|
content_length=tls_tables.parse_int(F(L, "http_content_length")),
|
|
set_cookie=A(L, "http_set_cookie") or [v for k, v in headers.items() if k == "set-cookie"],
|
|
www_authenticate=F(L, "http_www_authenticate") or headers.get("www-authenticate"),
|
|
request_uri=F(L, "http_response_for_uri"),
|
|
src_ip=m["src_ip"], dst_ip=m["dst_ip"], src_port=m["src_port"],
|
|
timestamp=m["ts"], stream_id=m["stream_id"], packet_number=m["pnum"],
|
|
)
|
|
ctx.http_responses.append(resp)
|
|
for sc in resp.set_cookie:
|
|
_add_text_payload(ctx, f"Set-Cookie: {sc}", "http", m["src_ip"], m["dst_ip"], m["pnum"], m["ts"])
|
|
filename = _filename_from_disposition(disposition)
|
|
uri_path = (resp.request_uri or "").split("?")[0]
|
|
ext = _ext_of(filename) or _ext_of(uri_path)
|
|
if filename or _is_download_content_type(ct) or (ext and ext in _FILE_EXTENSIONS):
|
|
ctx.file_transfers.append(FileTransfer(
|
|
filename=filename or (os.path.basename(uri_path) or None),
|
|
extension=ext, content_type=ct, size=resp.content_length,
|
|
protocol="http", src_ip=m["src_ip"], dst_ip=m["dst_ip"],
|
|
direction="download", uri=resp.request_uri,
|
|
timestamp=m["ts"], stream_id=m["stream_id"], packet_number=m["pnum"]))
|
|
|
|
def _ssdp(self, L, m):
|
|
method = self._first(L, "http_request_method")
|
|
headers = _parse_header_lines(self._all(L, "http_request_line") + self._all(L, "http_response_line"))
|
|
self.ctx.service_events.append(ServiceEvent(
|
|
protocol="ssdp", kind=(method or "response").lower(),
|
|
value=headers.get("location") or headers.get("st") or headers.get("nt"),
|
|
detail=headers.get("server") or headers.get("user-agent"),
|
|
src_ip=m["src_ip"], dst_ip=m["dst_ip"], dst_port=m["dst_port"],
|
|
timestamp=m["ts"], packet_number=m["pnum"]))
|
|
|
|
# ---- TLS / QUIC -------------------------------------------------------- #
|
|
|
|
def _tls(self, L, m):
|
|
A, F = self._all, self._first
|
|
types = A(L, "tls_handshake_type")
|
|
alerts = A(L, "tls_alert_message_desc")
|
|
if not types and not alerts:
|
|
return
|
|
is_quic = "quic" in m["protos"]
|
|
transport = "quic" if is_quic else "tcp"
|
|
stream = m["stream_id"]
|
|
key = (transport, stream if stream is not None else (m["src_ip"], m["dst_ip"], m["src_port"], m["dst_port"]))
|
|
sess = self.tls_index.get(key)
|
|
if sess is None:
|
|
# The ClientHello direction defines client and server.
|
|
if "1" in types or not types:
|
|
src, dst, sport, dport = m["src_ip"], m["dst_ip"], m["src_port"], m["dst_port"]
|
|
else:
|
|
src, dst, sport, dport = m["dst_ip"], m["src_ip"], m["dst_port"], m["src_port"]
|
|
sess = TlsSession(src_ip=src, dst_ip=dst, dst_port=dport or 443, src_port=sport,
|
|
transport=transport, timestamp=m["ts"], stream_id=stream,
|
|
packet_number=m["pnum"])
|
|
self.tls_index[key] = sess
|
|
self.ctx.tls_sessions.append(sess)
|
|
if m["pnum"] is not None and len(sess.packet_numbers) < 8:
|
|
sess.packet_numbers.append(m["pnum"])
|
|
|
|
versions = [tls_tables.parse_int(v) for v in A(L, "tls_handshake_extensions_supported_version")]
|
|
versions = [v for v in versions if v is not None and not tls_tables.is_grease(v)]
|
|
ciphers = [tls_tables.parse_int(c) for c in A(L, "tls_handshake_ciphersuite")]
|
|
ciphers = [c for c in ciphers if c is not None and not tls_tables.is_grease(c)]
|
|
hs_version = tls_tables.parse_int(F(L, "tls_handshake_version"))
|
|
|
|
if "1" in types:
|
|
sess.client_hello_seen = True
|
|
sess.sni = sess.sni or F(L, "tls_handshake_extensions_server_name")
|
|
alpn = A(L, "tls_handshake_extensions_alpn_str")
|
|
if alpn and not sess.alpn:
|
|
sess.alpn = alpn
|
|
offered = versions or ([hs_version] if hs_version else [])
|
|
sess.client_offered_versions = [tls_tables.version_name(v) for v in offered]
|
|
if offered:
|
|
sess.client_version = tls_tables.version_name(max(offered))
|
|
sess.client_offered_ciphers = [tls_tables.cipher_name(c) for c in ciphers]
|
|
if "2" in types:
|
|
sess.server_hello_seen = True
|
|
negotiated = versions[0] if versions else hs_version
|
|
if negotiated:
|
|
sess.version = tls_tables.version_name(negotiated)
|
|
if ciphers:
|
|
sess.cipher = tls_tables.cipher_name(ciphers[0])
|
|
if "11" in types:
|
|
chain = A(L, "tls_handshake_certificate")
|
|
from_server = (m["src_ip"] == sess.dst_ip and (m["src_port"] or 0) == (sess.dst_port or 0))
|
|
if chain and not from_server:
|
|
# A certificate sent by the client indicates mutual TLS authentication.
|
|
sess.client_cert_seen = True
|
|
elif chain and not sess.cert_seen:
|
|
sess.cert_seen = True
|
|
info = parse_certificate(chain[0], now_epoch=m["ts"] or self.ctx.end_time)
|
|
if info:
|
|
sess.cert_subject_cn = info.get("subject_cn")
|
|
sess.cert_issuer_cn = info.get("issuer_cn")
|
|
sess.cert_san = info.get("san") or []
|
|
sess.cert_not_before = info.get("not_before")
|
|
sess.cert_not_after = info.get("not_after")
|
|
sess.cert_self_signed = info.get("self_signed")
|
|
sess.cert_expired = info.get("expired")
|
|
sess.cert_sig_algorithm = info.get("sig_algorithm")
|
|
sess.cert_key_type = info.get("key_type")
|
|
sess.cert_key_bits = info.get("key_bits")
|
|
for a in alerts:
|
|
name = _TLS_ALERTS.get(str(a), str(a))
|
|
if name not in sess.alerts:
|
|
sess.alerts.append(name)
|
|
|
|
# ---- clear-text authentication protocols ------------------------------ #
|
|
|
|
def _auth_record(self, protocol, m) -> PlaintextAuth:
|
|
key = (protocol, m["src_ip"], m["dst_ip"], m["stream_id"])
|
|
pa = self.last_auth.get(key)
|
|
if pa is None or (pa.username is not None and pa.password is not None):
|
|
pa = PlaintextAuth(protocol=protocol, src_ip=m["src_ip"], dst_ip=m["dst_ip"],
|
|
dst_port=m["dst_port"], stream_id=m["stream_id"])
|
|
self.ctx.plaintext_auth.append(pa)
|
|
self.last_auth[key] = pa
|
|
return pa
|
|
|
|
def _auth_outcome(self, protocol, m, success: bool):
|
|
"""Attach a server verdict to the client's pending authentication."""
|
|
key = (protocol, m["dst_ip"], m["src_ip"], m["stream_id"])
|
|
pa = self.last_auth.get(key)
|
|
username = None
|
|
if pa is not None:
|
|
pa.outcome = "success" if success else "failure"
|
|
username = pa.username
|
|
self.ctx.auth_attempts.append(AuthAttempt(
|
|
protocol=protocol, success=success, src_ip=m["dst_ip"], dst_ip=m["src_ip"],
|
|
dst_port=m["src_port"], username=username, timestamp=m["ts"],
|
|
stream_id=m["stream_id"], packet_number=m["pnum"]))
|
|
|
|
def _ftp(self, L, m):
|
|
cmd = (self._first(L, "ftp_request_command") or "").upper()
|
|
arg = self._first(L, "ftp_request_arg")
|
|
if cmd in ("USER", "PASS"):
|
|
pa = self._auth_record("ftp", m)
|
|
if cmd == "USER":
|
|
pa.username = arg
|
|
else:
|
|
pa.password = arg
|
|
pa.timestamp, pa.packet_number = m["ts"], m["pnum"]
|
|
code = self._int(L, "ftp_response_code")
|
|
if code == 230:
|
|
self._auth_outcome("ftp", m, True)
|
|
elif code in (530, 430):
|
|
self._auth_outcome("ftp", m, False)
|
|
if cmd in ("RETR", "STOR") and arg:
|
|
self.ctx.file_transfers.append(FileTransfer(
|
|
filename=arg, extension=_ext_of(arg), protocol="ftp",
|
|
src_ip=m["src_ip"], dst_ip=m["dst_ip"],
|
|
direction="download" if cmd == "RETR" else "upload",
|
|
timestamp=m["ts"], stream_id=m["stream_id"], packet_number=m["pnum"]))
|
|
|
|
def _telnet(self, L, m):
|
|
data = self._all(L, "telnet_data")
|
|
if not data:
|
|
return
|
|
stream = m["stream_id"]
|
|
buf = self.telnet_buf[stream]
|
|
if buf["meta"] is None:
|
|
buf["meta"] = dict(m)
|
|
text = "".join(data)
|
|
client_side = (m["dst_port"] == 23) or (buf["meta"]["src_ip"] == m["src_ip"] and buf["meta"]["dst_port"] == 23)
|
|
(buf["c2s"] if client_side else buf["s2c"]).append(text)
|
|
|
|
def _pop3(self, L, m):
|
|
cmd = (self._first(L, "pop_request_command") or "").upper()
|
|
param = self._first(L, "pop_request_parameter")
|
|
if cmd in ("USER", "PASS") and param:
|
|
pa = self._auth_record("pop3", m)
|
|
if cmd == "USER":
|
|
pa.username = param
|
|
else:
|
|
pa.password = param
|
|
pa.timestamp, pa.packet_number = m["ts"], m["pnum"]
|
|
ind = (self._first(L, "pop_response_indicator") or "").upper()
|
|
key = ("pop3", m["dst_ip"], m["src_ip"], m["stream_id"])
|
|
pending = self.last_auth.get(key)
|
|
if ind and pending is not None and pending.password is not None and pending.outcome is None:
|
|
self._auth_outcome("pop3", m, ind.startswith("+OK"))
|
|
|
|
def _imap(self, L, m):
|
|
line = self._first(L, "imap_request") or ""
|
|
parts = str(line).split()
|
|
idx = next((k for k, tok in enumerate(parts) if tok.upper() == "LOGIN"), None)
|
|
if idx is not None and len(parts) >= idx + 3:
|
|
pa = self._auth_record("imap", m)
|
|
pa.username, pa.password = parts[idx + 1].strip('"'), parts[idx + 2].strip('"')
|
|
pa.timestamp, pa.packet_number = m["ts"], m["pnum"]
|
|
status = (self._first(L, "imap_response_status") or "").upper()
|
|
key = ("imap", m["dst_ip"], m["src_ip"], m["stream_id"])
|
|
pending = self.last_auth.get(key)
|
|
if status in ("OK", "NO", "BAD") and pending is not None and pending.password is not None \
|
|
and pending.outcome is None:
|
|
self._auth_outcome("imap", m, status == "OK")
|
|
|
|
def _smtp(self, L, m):
|
|
param = self._first(L, "smtp_req_parameter")
|
|
cmd = (self._first(L, "smtp_req_command") or "").upper()
|
|
code = self._int(L, "smtp_response_code")
|
|
if code == 235:
|
|
self._auth_outcome("smtp", m, True)
|
|
return
|
|
if code in (535, 534, 530):
|
|
self._auth_outcome("smtp", m, False)
|
|
return
|
|
if not param and not cmd:
|
|
return
|
|
pending = self.smtp_auth.setdefault(m["stream_id"], {"mech": None, "tokens": []})
|
|
text = f"{cmd} {param or ''}".strip()
|
|
upper = text.upper()
|
|
if "AUTH LOGIN" in upper:
|
|
pending["mech"] = "login"
|
|
return
|
|
if "AUTH PLAIN" in upper:
|
|
pending["mech"] = "plain"
|
|
tail = text.split()
|
|
if len(tail) >= 3:
|
|
self._emit_smtp(crackformats.decode_smtp_auth_plain(tail[2]), m)
|
|
self.smtp_auth.pop(m["stream_id"], None)
|
|
return
|
|
if pending["mech"] == "login" and param:
|
|
pending["tokens"].append(param)
|
|
if len(pending["tokens"]) >= 2:
|
|
creds = crackformats.decode_smtp_auth_login(pending["tokens"][0], pending["tokens"][1])
|
|
self._emit_smtp(creds, m)
|
|
self.smtp_auth.pop(m["stream_id"], None)
|
|
elif pending["mech"] == "plain" and param:
|
|
self._emit_smtp(crackformats.decode_smtp_auth_plain(param), m)
|
|
self.smtp_auth.pop(m["stream_id"], None)
|
|
|
|
def _emit_smtp(self, creds, m):
|
|
if not creds:
|
|
return
|
|
user, pw = creds
|
|
if not (user or pw):
|
|
return
|
|
pa = self._auth_record("smtp", m)
|
|
pa.username, pa.password = user, pw
|
|
pa.timestamp, pa.packet_number = m["ts"], m["pnum"]
|
|
|
|
def _ntlm(self, L, m):
|
|
F = self._first
|
|
msg_type = tls_tables.parse_int(F(L, "ntlmssp_messagetype"))
|
|
chal = F(L, "ntlmssp_ntlmserverchallenge")
|
|
if chal and msg_type == 2:
|
|
self.ntlm_chal[m["stream_id"]] = str(chal)
|
|
account = F(L, "ntlmssp_auth_username")
|
|
domain = F(L, "ntlmssp_auth_domain")
|
|
nt_resp = F(L, "ntlmssp_ntlmv2_response") or F(L, "ntlmssp_auth_ntresponse")
|
|
if account or nt_resp:
|
|
chal = self.ntlm_chal.get(m["stream_id"])
|
|
am = AuthMaterial(kind="ntlmssp", account=account, domain=domain,
|
|
src_ip=m["src_ip"], dst_ip=m["dst_ip"], protocol="ntlm",
|
|
timestamp=m["ts"], packet_number=m["pnum"],
|
|
server_challenge=chal, nt_response=str(nt_resp) if nt_resp else None)
|
|
built = crackformats.build_netntlmv2(account, domain, chal, nt_resp)
|
|
if built:
|
|
am.kind = "netntlmv2"
|
|
am.hashcat, am.hashcat_mode, am.hash_format = built["hashcat"], built["hashcat_mode"], built["format"]
|
|
self.ctx.auth_material.append(am)
|
|
|
|
def _kerberos(self, L, m):
|
|
F, A = self._first, self._all
|
|
msg_type = str(F(L, "kerberos_msg_type") or "")
|
|
account = F(L, "kerberos_cnamestring")
|
|
realm = F(L, "kerberos_realm") or F(L, "kerberos_crealm")
|
|
etypes = A(L, "kerberos_etype")
|
|
ciphers = A(L, "kerberos_cipher")
|
|
am = AuthMaterial(kind=f"kerberos_{msg_type}" if msg_type else "kerberos",
|
|
account=account, domain=realm, src_ip=m["src_ip"], dst_ip=m["dst_ip"],
|
|
protocol="kerberos", timestamp=m["ts"], packet_number=m["pnum"],
|
|
etype=etypes[0] if etypes else None)
|
|
built = None
|
|
# AS-REP: the second cipher (enc-part) is the crackable one; TGS-REP: the ticket cipher.
|
|
if msg_type == "11" and ciphers:
|
|
cipher = ciphers[1] if len(ciphers) > 1 else ciphers[0]
|
|
etype = etypes[-1] if etypes else None
|
|
am.etype = etype
|
|
built = crackformats.build_krb5_asrep(account, realm, etype, cipher)
|
|
elif msg_type == "13" and ciphers:
|
|
spn = "/".join(A(L, "kerberos_snamestring")) or None
|
|
etype = etypes[0] if etypes else None
|
|
built = crackformats.build_krb5_tgsrep(account, realm, spn, etype, ciphers[0])
|
|
elif msg_type not in ("11", "13"):
|
|
return # Requests carry no offline-usable material.
|
|
if built:
|
|
am.hashcat, am.hashcat_mode, am.hash_format = built["hashcat"], built["hashcat_mode"], built["format"]
|
|
self.ctx.auth_material.append(am)
|
|
|
|
def _mqtt(self, L, m):
|
|
types = {"1": "connect", "2": "connack", "3": "publish", "8": "subscribe", "9": "suback",
|
|
"12": "pingreq", "13": "pingresp", "14": "disconnect"}
|
|
for raw in self._all(L, "mqtt_msgtype") or [None]:
|
|
mtype = types.get(str(raw), str(raw) if raw else None)
|
|
ev = MqttEvent(msg_type=mtype, client_id=self._first(L, "mqtt_clientid"),
|
|
username=self._first(L, "mqtt_username"), password=self._first(L, "mqtt_passwd"),
|
|
topic=self._first(L, "mqtt_topic"), return_code=self._int(L, "mqtt_conack_val"),
|
|
src_ip=m["src_ip"], dst_ip=m["dst_ip"], dst_port=m["dst_port"],
|
|
timestamp=m["ts"], stream_id=m["stream_id"], packet_number=m["pnum"])
|
|
self.ctx.mqtt_events.append(ev)
|
|
if mtype == "connect" and (ev.username or ev.password):
|
|
pa = self._auth_record("mqtt", m)
|
|
pa.username, pa.password = ev.username, ev.password
|
|
pa.timestamp, pa.packet_number = m["ts"], m["pnum"]
|
|
if mtype == "connack" and ev.return_code is not None:
|
|
self._auth_outcome("mqtt", m, ev.return_code == 0)
|
|
break
|
|
|
|
def _snmp(self, L, m):
|
|
community = self._first(L, "snmp_community")
|
|
version = self._first(L, "snmp_version")
|
|
if community is not None:
|
|
self.ctx.service_events.append(ServiceEvent(
|
|
protocol="snmp", kind="snmp_community", value=community,
|
|
detail=f"version {version}" if version is not None else None,
|
|
src_ip=m["src_ip"], dst_ip=m["dst_ip"], dst_port=m["dst_port"],
|
|
timestamp=m["ts"], packet_number=m["pnum"]))
|
|
pa = PlaintextAuth(protocol="snmp", username=None, password=community,
|
|
src_ip=m["src_ip"], dst_ip=m["dst_ip"], dst_port=m["dst_port"],
|
|
timestamp=m["ts"], packet_number=m["pnum"])
|
|
self.ctx.plaintext_auth.append(pa)
|
|
|
|
def _ssh(self, L, m):
|
|
banner = self._first(L, "ssh_protocol")
|
|
if banner:
|
|
self.ctx.service_events.append(ServiceEvent(
|
|
protocol="ssh", kind="ssh_banner", value=str(banner).strip(),
|
|
src_ip=m["src_ip"], dst_ip=m["dst_ip"], dst_port=m["dst_port"],
|
|
timestamp=m["ts"], packet_number=m["pnum"]))
|
|
|
|
def _coap(self, L, m):
|
|
code = self._first(L, "coap_code")
|
|
path = "/".join(self._all(L, "coap_opt_uri_path"))
|
|
self.ctx.service_events.append(ServiceEvent(
|
|
protocol="coap", kind="coap_message", value=path or None, detail=code,
|
|
src_ip=m["src_ip"], dst_ip=m["dst_ip"], dst_port=m["dst_port"],
|
|
timestamp=m["ts"], packet_number=m["pnum"]))
|
|
|
|
def _rtsp(self, L, m):
|
|
req = self._first(L, "rtsp_request")
|
|
status = self._int(L, "rtsp_status")
|
|
if req:
|
|
self.ctx.service_events.append(ServiceEvent(
|
|
protocol="rtsp", kind="rtsp_request", value=str(req).strip(),
|
|
src_ip=m["src_ip"], dst_ip=m["dst_ip"], dst_port=m["dst_port"],
|
|
timestamp=m["ts"], packet_number=m["pnum"]))
|
|
if status == 401:
|
|
self.ctx.auth_attempts.append(AuthAttempt(
|
|
protocol="rtsp", success=False, src_ip=m["dst_ip"], dst_ip=m["src_ip"],
|
|
dst_port=m["src_port"], timestamp=m["ts"], stream_id=m["stream_id"],
|
|
packet_number=m["pnum"]))
|
|
|
|
def _sip(self, L, m):
|
|
method = self._first(L, "sip_method")
|
|
auth = self._first(L, "sip_auth")
|
|
if method:
|
|
self.ctx.service_events.append(ServiceEvent(
|
|
protocol="sip", kind="sip_request", value=str(method),
|
|
src_ip=m["src_ip"], dst_ip=m["dst_ip"], dst_port=m["dst_port"],
|
|
timestamp=m["ts"], packet_number=m["pnum"]))
|
|
if auth:
|
|
_add_text_payload(self.ctx, f"Authorization: {auth}", "sip", m["src_ip"], m["dst_ip"], m["pnum"], m["ts"])
|
|
|
|
def _tftp(self, L, m):
|
|
op = self._int(L, "tftp_opcode")
|
|
name = self._first(L, "tftp_source_file") or self._first(L, "tftp_destination_file")
|
|
if op in (1, 2) and name:
|
|
self.ctx.file_transfers.append(FileTransfer(
|
|
filename=name, extension=_ext_of(name), protocol="tftp",
|
|
src_ip=m["src_ip"], dst_ip=m["dst_ip"],
|
|
direction="download" if op == 1 else "upload",
|
|
timestamp=m["ts"], packet_number=m["pnum"]))
|
|
|
|
def _smb(self, L, m):
|
|
status = self._first(L, "smb2_nt_status") or self._first(L, "smb_nt_status")
|
|
if status is None:
|
|
return
|
|
code = tls_tables.parse_int(status)
|
|
if code == 0xC000006D: # STATUS_LOGON_FAILURE
|
|
self.ctx.auth_attempts.append(AuthAttempt(
|
|
protocol="smb", success=False, src_ip=m["dst_ip"], dst_ip=m["src_ip"],
|
|
dst_port=m["src_port"], timestamp=m["ts"], stream_id=m["stream_id"],
|
|
packet_number=m["pnum"]))
|
|
|
|
# ---- payload pass ----------------------------------------------------- #
|
|
|
|
def _handle_payload(self, L: dict) -> None:
|
|
F = self._first
|
|
raw = F(L, "tcp_payload") or F(L, "udp_payload") or F(L, "data_data")
|
|
if not raw:
|
|
return
|
|
protos = [p for p in (F(L, "frame_protocols") or "").split(":") if p]
|
|
if set(protos) & _NO_TEXT_PROTOCOLS:
|
|
return
|
|
text = _bytes_from_hexish(str(raw))
|
|
if not text or not text.strip() or _printable_ratio(text) < 0.75:
|
|
return
|
|
proto = _highest_layer(protos).lower()
|
|
if proto in ("tcp", "udp", "data", "raw"):
|
|
proto = _guess_cleartext_proto(tls_tables.parse_int(F(L, "tcp_dstport")) or
|
|
tls_tables.parse_int(F(L, "udp_dstport")),
|
|
tls_tables.parse_int(F(L, "tcp_srcport")) or
|
|
tls_tables.parse_int(F(L, "udp_srcport")), text) or proto
|
|
src_ip = F(L, "ip_src") or F(L, "ipv6_src")
|
|
dst_ip = F(L, "ip_dst") or F(L, "ipv6_dst")
|
|
_add_text_payload(self.ctx, text, proto, src_ip, dst_ip,
|
|
tls_tables.parse_int(F(L, "frame_number")), _float(F(L, "frame_time_epoch")))
|
|
|
|
# ---- post-processing -------------------------------------------------- #
|
|
|
|
def finish_metadata(self):
|
|
ctx = self.ctx
|
|
ctx.conversations = list(self.conv_index.values())
|
|
for conv in ctx.conversations:
|
|
_orient_conversation(conv)
|
|
# DHCP hostnames and vendor classes belong to the host that owns the MAC.
|
|
for host in ctx.hosts.values():
|
|
for mac in host.macs:
|
|
info = self.dhcp_by_mac.get(mac)
|
|
if not info:
|
|
continue
|
|
if info.get("hostname"):
|
|
host.hostnames.add(info["hostname"])
|
|
if info.get("vendor"):
|
|
host.dhcp_vendor_class = info["vendor"]
|
|
for mac, info in self.dhcp_by_mac.items():
|
|
ip = info.get("ip")
|
|
if ip and not is_special(ip):
|
|
h = ctx.host(ip)
|
|
h.macs.add(mac)
|
|
if info.get("hostname"):
|
|
h.hostnames.add(info["hostname"])
|
|
if info.get("vendor"):
|
|
h.dhcp_vendor_class = info["vendor"]
|
|
if ctx.device_ip and ctx.device_ip in ctx.hosts:
|
|
ctx.device_macs.update(ctx.hosts[ctx.device_ip].macs)
|
|
# Telnet sessions: reconstruct typed text and a login exchange when visible.
|
|
for stream, buf in self.telnet_buf.items():
|
|
m = buf["meta"] or {}
|
|
typed = "".join(buf["c2s"])
|
|
shown = "".join(buf["s2c"])
|
|
if typed.strip():
|
|
_add_text_payload(ctx, typed, "telnet", m.get("src_ip"), m.get("dst_ip"), m.get("pnum"), m.get("ts"))
|
|
if shown.strip():
|
|
_add_text_payload(ctx, shown, "telnet", m.get("dst_ip"), m.get("src_ip"), m.get("pnum"), m.get("ts"))
|
|
creds = _telnet_login(typed, shown)
|
|
if creds:
|
|
user, pw, outcome = creds
|
|
ctx.plaintext_auth.append(PlaintextAuth(
|
|
protocol="telnet", username=user, password=pw, outcome=outcome,
|
|
src_ip=m.get("src_ip"), dst_ip=m.get("dst_ip"), dst_port=m.get("dst_port"),
|
|
timestamp=m.get("ts"), stream_id=stream, packet_number=m.get("pnum")))
|
|
if outcome is not None:
|
|
ctx.auth_attempts.append(AuthAttempt(
|
|
protocol="telnet", success=(outcome == "success"), src_ip=m.get("src_ip"),
|
|
dst_ip=m.get("dst_ip"), dst_port=m.get("dst_port"), username=user,
|
|
timestamp=m.get("ts"), stream_id=stream, packet_number=m.get("pnum")))
|
|
# Coverage notes: flows with payload on ports tshark did not dissect.
|
|
undecoded: dict[int, int] = defaultdict(int)
|
|
for conv in ctx.conversations:
|
|
if conv.payload_bytes > 0 and not (conv.app_protocols - {"tcp.segments"}) and conv.dst_port:
|
|
undecoded[conv.dst_port] += 1
|
|
if undecoded:
|
|
top = sorted(undecoded.items(), key=lambda kv: -kv[1])[:8]
|
|
ctx.coverage_notes.append(
|
|
"Flows with application data that tshark did not decode (port: flows): "
|
|
+ ", ".join(f"{p}: {n}" for p, n in top))
|
|
|
|
def finalize(self):
|
|
pass
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Certificate parsing
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
def parse_certificate(cert_hex: str, now_epoch: Optional[float] = None) -> Optional[dict]:
|
|
"""Parse a DER certificate (hex from tshark) into the fields the detectors use."""
|
|
try:
|
|
der = bytes.fromhex(str(cert_hex).replace(":", ""))
|
|
except ValueError:
|
|
return None
|
|
try:
|
|
from cryptography import x509 # optional dependency
|
|
from cryptography.hazmat.primitives.asymmetric import rsa, ec, ed25519, ed448
|
|
except ImportError: # pragma: no cover
|
|
return _parse_certificate_fallback(der, now_epoch)
|
|
try:
|
|
cert = x509.load_der_x509_certificate(der)
|
|
except Exception:
|
|
return _parse_certificate_fallback(der, now_epoch)
|
|
|
|
def _cn(name) -> Optional[str]:
|
|
try:
|
|
attrs = name.get_attributes_for_oid(x509.oid.NameOID.COMMON_NAME)
|
|
if attrs:
|
|
return str(attrs[0].value)
|
|
attrs = name.get_attributes_for_oid(x509.oid.NameOID.ORGANIZATION_NAME)
|
|
return str(attrs[0].value) if attrs else name.rfc4514_string()
|
|
except Exception:
|
|
return None
|
|
|
|
san: list[str] = []
|
|
try:
|
|
ext = cert.extensions.get_extension_for_class(x509.SubjectAlternativeName)
|
|
san = [str(v) for v in ext.value.get_values_for_type(x509.DNSName)][:20]
|
|
except Exception:
|
|
pass
|
|
try:
|
|
nb = cert.not_valid_before_utc
|
|
na = cert.not_valid_after_utc
|
|
except AttributeError: # cryptography < 42
|
|
import datetime as _dt
|
|
nb = cert.not_valid_before.replace(tzinfo=_dt.timezone.utc)
|
|
na = cert.not_valid_after.replace(tzinfo=_dt.timezone.utc)
|
|
key = cert.public_key()
|
|
key_type, key_bits = "unknown", None
|
|
if isinstance(key, rsa.RSAPublicKey):
|
|
key_type, key_bits = "RSA", key.key_size
|
|
elif isinstance(key, ec.EllipticCurvePublicKey):
|
|
key_type, key_bits = f"EC {key.curve.name}", key.curve.key_size
|
|
elif isinstance(key, (ed25519.Ed25519PublicKey, ed448.Ed448PublicKey)):
|
|
key_type, key_bits = type(key).__name__.replace("PublicKey", ""), 256
|
|
try:
|
|
sig = cert.signature_hash_algorithm.name if cert.signature_hash_algorithm else None
|
|
except Exception:
|
|
sig = None
|
|
expired = None
|
|
if now_epoch is not None:
|
|
expired = float(now_epoch) > na.timestamp()
|
|
return {
|
|
"subject_cn": _cn(cert.subject), "issuer_cn": _cn(cert.issuer), "san": san,
|
|
"not_before": nb.strftime("%Y-%m-%d %H:%M:%S UTC"),
|
|
"not_after": na.strftime("%Y-%m-%d %H:%M:%S UTC"),
|
|
"self_signed": cert.subject == cert.issuer,
|
|
"expired": expired, "not_yet_valid": (float(now_epoch) < nb.timestamp()) if now_epoch else None,
|
|
"sig_algorithm": sig, "key_type": key_type, "key_bits": key_bits,
|
|
}
|
|
|
|
|
|
def _parse_certificate_fallback(der: bytes, now_epoch) -> Optional[dict]:
|
|
"""Minimal DER walk used when ``cryptography`` is unavailable."""
|
|
try:
|
|
times = re.findall(rb"\x17\x0d(\d{12})Z", der)
|
|
nb = _parse_cert_time(times[0].decode() + "Z") if times else None
|
|
na = _parse_cert_time(times[1].decode() + "Z") if len(times) > 1 else None
|
|
# CN (OID 2.5.4.3) values in order: issuer first, then subject.
|
|
cns = re.findall(rb"\x06\x03\x55\x04\x03[\x0c\x13\x16]([\x01-\x7f])(.*?)\x31?", der, re.DOTALL)
|
|
names = []
|
|
for ln, rest in cns:
|
|
n = rest[:ln[0]]
|
|
names.append(n.decode("utf-8", "ignore"))
|
|
issuer = names[0] if names else None
|
|
subject = names[1] if len(names) > 1 else None
|
|
expired = (float(now_epoch) > na) if (na is not None and now_epoch is not None) else None
|
|
import datetime as _dt
|
|
fmt = lambda t: _dt.datetime.fromtimestamp(t, _dt.timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC") if t else None
|
|
return {"subject_cn": subject, "issuer_cn": issuer, "san": [],
|
|
"not_before": fmt(nb), "not_after": fmt(na),
|
|
"self_signed": (subject == issuer) if subject and issuer else None,
|
|
"expired": expired, "sig_algorithm": None, "key_type": None, "key_bits": None}
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Small helpers
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
_TLS_ALERTS = {
|
|
"0": "close_notify", "10": "unexpected_message", "20": "bad_record_mac",
|
|
"40": "handshake_failure", "42": "bad_certificate", "43": "unsupported_certificate",
|
|
"44": "certificate_revoked", "45": "certificate_expired", "46": "certificate_unknown",
|
|
"47": "illegal_parameter", "48": "unknown_ca", "49": "access_denied", "70": "protocol_version",
|
|
"71": "insufficient_security", "80": "internal_error", "112": "unrecognized_name",
|
|
}
|
|
|
|
_FILE_EXTENSIONS = {
|
|
".exe", ".dll", ".ps1", ".vbs", ".jar", ".bat", ".cmd", ".scr", ".hta", ".lnk",
|
|
".docm", ".xlsm", ".pptm", ".zip", ".rar", ".7z", ".iso", ".elf", ".msi", ".apk",
|
|
".bin", ".img", ".fw", ".ota", ".upd", ".hex", ".pkg", ".tar", ".gz", ".tgz", ".bz2",
|
|
".xz", ".rom", ".swu", ".ubi", ".squashfs", ".sh", ".py", ".pl",
|
|
}
|
|
|
|
|
|
def _looks_like_service_port(port: Optional[int]) -> bool:
|
|
return port is not None and (port < 1024 or port in CLEARTEXT_PORTS or port in (
|
|
1883, 8883, 8080, 8443, 8000, 8888, 5683, 5684, 1900, 5353, 5355, 3306, 5432, 6379,
|
|
27017, 11211, 3389, 5900, 5222, 5223, 5228, 9000, 9100))
|
|
|
|
|
|
def _orient_conversation(conv: Conversation) -> None:
|
|
"""Swap sides when no SYN was seen and the ports show src is the server."""
|
|
if conv.protocol == "tcp" and (conv.syn_count or conv.synack_count):
|
|
return
|
|
sp, dp = conv.src_port, conv.dst_port
|
|
if sp is None or dp is None:
|
|
return
|
|
if _looks_like_service_port(sp) and not _looks_like_service_port(dp):
|
|
conv.src_ip, conv.dst_ip = conv.dst_ip, conv.src_ip
|
|
conv.src_port, conv.dst_port = dp, sp
|
|
conv.bytes_src_to_dst, conv.bytes_dst_to_src = conv.bytes_dst_to_src, conv.bytes_src_to_dst
|
|
conv.packets_src_to_dst, conv.packets_dst_to_src = conv.packets_dst_to_src, conv.packets_src_to_dst
|
|
|
|
|
|
_FIELD_CACHE: dict[str, set[str]] = {}
|
|
|
|
|
|
def known_fields(tshark: str) -> set[str]:
|
|
"""All field names the installed tshark understands (cached per binary)."""
|
|
cached = _FIELD_CACHE.get(tshark)
|
|
if cached is not None:
|
|
return cached
|
|
names: set[str] = set()
|
|
try:
|
|
out = subprocess.run([tshark, "-G", "fields"], capture_output=True, text=True, timeout=120).stdout
|
|
for line in out.splitlines():
|
|
parts = line.split("\t")
|
|
if len(parts) > 2 and parts[0] == "F":
|
|
names.add(parts[2])
|
|
except Exception as exc:
|
|
logger.warning("Field list of %s could not be read (%s); using all fields", tshark, exc)
|
|
_FIELD_CACHE[tshark] = names
|
|
return names
|
|
|
|
|
|
def supported_fields(tshark: str, fields: list[str]) -> list[str]:
|
|
"""Drop fields the installed tshark does not know; names change between versions."""
|
|
known = known_fields(tshark)
|
|
if not known:
|
|
return list(fields)
|
|
missing = [f for f in fields if f not in known]
|
|
if missing:
|
|
logger.warning("Fields not supported by %s and skipped: %s", tshark, ", ".join(missing))
|
|
return [f for f in fields if f in known]
|
|
|
|
|
|
def _tshark_version(tshark: str) -> Optional[str]:
|
|
try:
|
|
out = subprocess.run([tshark, "-v"], capture_output=True, text=True, timeout=20).stdout
|
|
first = (out or "").strip().splitlines()[0] if out else ""
|
|
m = re.search(r"(\d+\.\d+\.\d+)", first)
|
|
return f"tshark {m.group(1)}" if m else (first or None)
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _float(v) -> Optional[float]:
|
|
try:
|
|
return float(v) if v is not None else None
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def _highest_layer(protos: list[str]) -> str:
|
|
skip = {"tcp.segments", "data", "data-text-lines", "ethertype", "vlan", "frame", "llc",
|
|
"image-jfif", "image-gif", "png", "json", "xml", "urlencoded-form", "media",
|
|
"mime_multipart"}
|
|
for p in reversed(protos):
|
|
if p not in skip:
|
|
return p.upper()
|
|
return (protos[-1].upper() if protos else "RAW")
|
|
|
|
|
|
def _mark_truncated(ctx, reason: str) -> None:
|
|
"""Record every distinct limit that stopped extraction from being exhaustive."""
|
|
if reason in ctx.truncation_reasons:
|
|
return
|
|
ctx.truncation_reasons.append(reason)
|
|
ctx.truncated = True
|
|
ctx.truncation_reason = " ".join(ctx.truncation_reasons)
|
|
|
|
|
|
def _add_text_payload(ctx, text, protocol, src_ip, dst_ip, packet_number, timestamp) -> None:
|
|
"""Append payload text while keeping the retained volume bounded."""
|
|
if len(ctx.text_payloads) >= _MAX_TEXT_PAYLOADS:
|
|
_mark_truncated(
|
|
ctx,
|
|
f"Retained payload text reached the cap of {_MAX_TEXT_PAYLOADS} entries; "
|
|
"later clear-text payloads were not scanned for secrets.",
|
|
)
|
|
return
|
|
ctx.text_payloads.append({
|
|
"text": str(text)[:_MAX_PAYLOAD_CHARS], "protocol": protocol,
|
|
"src_ip": src_ip, "dst_ip": dst_ip,
|
|
"packet_number": packet_number, "timestamp": timestamp,
|
|
})
|
|
|
|
|
|
def _bytes_from_hexish(raw: str) -> Optional[str]:
|
|
"""Convert a hex-formatted tshark payload to text when possible."""
|
|
s = str(raw).strip()
|
|
compact = s.replace(":", "").replace(".", "").replace(" ", "")
|
|
if compact and len(compact) % 2 == 0 and all(c in "0123456789abcdefABCDEF" for c in compact):
|
|
try:
|
|
return bytes.fromhex(compact).decode("utf-8", "ignore")
|
|
except Exception:
|
|
return None
|
|
# Some dissectors already expose decoded text.
|
|
return s
|
|
|
|
|
|
def _decode_maybe_hex(raw) -> Optional[str]:
|
|
"""``http.file_data`` is a string in tshark 4.0 and hex bytes in 4.2+."""
|
|
if raw is None:
|
|
return None
|
|
s = str(raw)
|
|
compact = s.replace(":", "")
|
|
if len(compact) >= 2 and len(compact) % 2 == 0 and all(c in "0123456789abcdefABCDEF" for c in compact):
|
|
try:
|
|
decoded = bytes.fromhex(compact).decode("utf-8", "ignore")
|
|
except Exception:
|
|
return s
|
|
if _printable_ratio(decoded) >= 0.9:
|
|
return decoded
|
|
return s
|
|
|
|
|
|
def _printable_ratio(text: str) -> float:
|
|
if not text:
|
|
return 0.0
|
|
ok = sum(1 for ch in text if ch.isprintable() or ch in "\r\n\t")
|
|
return ok / len(text)
|
|
|
|
|
|
def _guess_cleartext_proto(dst_port, src_port, text: str) -> Optional[str]:
|
|
for port in (dst_port, src_port):
|
|
if port in CLEARTEXT_PORTS:
|
|
return CLEARTEXT_PORTS[port]
|
|
head = text[:12].upper()
|
|
if head.startswith(("GET ", "POST ", "PUT ", "HEAD ", "HTTP/", "DELETE ", "OPTIONS ")):
|
|
return "http"
|
|
return None
|
|
|
|
|
|
def _parse_header_lines(lines: list[str]) -> dict[str, str]:
|
|
out: dict[str, str] = {}
|
|
for line in lines or []:
|
|
if ":" not in line:
|
|
continue
|
|
k, v = line.split(":", 1)
|
|
out[k.strip().lower()] = v.strip()
|
|
return out
|
|
|
|
|
|
def _ext_of(name) -> Optional[str]:
|
|
if not name:
|
|
return None
|
|
base = str(name).split("?")[0].split("#")[0]
|
|
ext = os.path.splitext(base)[1].lower()
|
|
return ext or None
|
|
|
|
|
|
def _telnet_login(typed: str, shown: str):
|
|
"""Recover a login exchange from reconstructed telnet text, if present."""
|
|
if not typed or not shown:
|
|
return None
|
|
low = shown.lower()
|
|
if "login:" not in low and "username:" not in low and "user:" not in low:
|
|
return None
|
|
if "password:" not in low:
|
|
return None
|
|
lines = [ln.strip() for ln in re.split(r"[\r\n]+", typed) if ln.strip()]
|
|
if len(lines) < 2:
|
|
return None
|
|
user, pw = lines[0], lines[1]
|
|
outcome = None
|
|
after = low.split("password:", 1)[1] if "password:" in low else ""
|
|
if re.search(r"incorrect|failed|invalid|denied|bad password", after):
|
|
outcome = "failure"
|
|
elif re.search(r"[\$#>]\s*$|welcome|last login|busybox", after, re.MULTILINE):
|
|
outcome = "success"
|
|
return user, pw, outcome
|
|
|
|
|
|
# tshark exposes DNS query types as numeric values.
|
|
_DNS_QTYPES = {
|
|
"1": "A", "2": "NS", "5": "CNAME", "6": "SOA", "12": "PTR", "15": "MX",
|
|
"16": "TXT", "28": "AAAA", "33": "SRV", "255": "ANY", "10": "NULL",
|
|
"43": "DS", "48": "DNSKEY", "257": "CAA", "65": "HTTPS", "64": "SVCB",
|
|
}
|
|
|
|
|
|
def _dns_qtype_name(raw) -> Optional[str]:
|
|
if raw is None:
|
|
return None
|
|
s = str(raw).strip()
|
|
return _DNS_QTYPES.get(s, s.upper())
|
|
|
|
|
|
# Content types that indicate a file rather than ordinary web content.
|
|
_DOWNLOAD_CT_MARKERS = (
|
|
"application/octet-stream", "application/x-msdownload", "application/x-dosexec",
|
|
"application/x-executable", "application/vnd.microsoft.portable-executable",
|
|
"application/zip", "application/x-rar", "application/x-7z", "application/x-tar",
|
|
"application/gzip", "application/x-gzip", "application/x-msi", "application/x-sh",
|
|
"application/java-archive", "application/x-elf", "application/vnd.android.package-archive",
|
|
"application/x-xz", "application/x-bzip2",
|
|
)
|
|
|
|
|
|
def _is_download_content_type(ct) -> bool:
|
|
if not ct:
|
|
return False
|
|
low = str(ct).lower()
|
|
return any(m in low for m in _DOWNLOAD_CT_MARKERS)
|
|
|
|
|
|
def _filename_from_disposition(disposition) -> Optional[str]:
|
|
"""Extract a filename from a Content-Disposition header (or a multipart body)."""
|
|
if not disposition:
|
|
return None
|
|
m = re.search(r'filename\*?=(?:"([^"]+)"|([^;\r\n]+))', str(disposition), re.IGNORECASE)
|
|
if not m:
|
|
return None
|
|
name = (m.group(1) or m.group(2) or "").strip().strip('"')
|
|
# Remove an optional RFC 5987 character-set prefix.
|
|
if "''" in name:
|
|
name = name.split("''", 1)[1]
|
|
return name or None
|
|
|
|
|
|
def _cert_is_self_signed(subject_cn, issuer_cn) -> Optional[bool]:
|
|
"""Return whether subject and issuer match, or None if data is incomplete."""
|
|
if not subject_cn or not issuer_cn:
|
|
return None
|
|
return str(subject_cn).strip().lower() == str(issuer_cn).strip().lower()
|
|
|
|
|
|
def _cert_is_expired(not_after, now_epoch) -> Optional[bool]:
|
|
"""Compare a parsed certificate expiry with the capture timestamp."""
|
|
if not not_after or now_epoch is None:
|
|
return None
|
|
exp = _parse_cert_time(not_after)
|
|
if exp is None:
|
|
return None
|
|
try:
|
|
return float(now_epoch) > exp
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def _parse_cert_time(raw) -> Optional[float]:
|
|
"""Convert common X.509 time formats to a UTC epoch timestamp."""
|
|
import datetime as _dt
|
|
|
|
s = str(raw).strip()
|
|
s = re.sub(r"\s*\((UTC|GMT)\)\s*$", "", s)
|
|
try:
|
|
return float(s)
|
|
except ValueError:
|
|
pass
|
|
fmts = (
|
|
"%Y-%m-%d %H:%M:%S", # 2020-01-01 00:00:00
|
|
"%Y-%m-%d %H:%M:%S UTC",
|
|
"%b %d %H:%M:%S %Y GMT", # Jan 1 00:00:00 2020 GMT
|
|
"%y%m%d%H%M%SZ", # UTCTime 200101000000Z
|
|
"%Y%m%d%H%M%SZ", # GeneralizedTime
|
|
"%Y-%m-%dT%H:%M:%SZ", # ISO 8601
|
|
)
|
|
for fmt in fmts:
|
|
try:
|
|
dt = _dt.datetime.strptime(s, fmt).replace(tzinfo=_dt.timezone.utc)
|
|
return dt.timestamp()
|
|
except ValueError:
|
|
continue
|
|
return None
|