"""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