import base64 import codecs import logging import unicodedata from typing import Any import demistomock as demisto # noqa: F401 import pyshark from CommonServerPython import * # noqa: F401 TCP_FLAG_FIN = 0x01 TCP_FLAG_SYN = 0x02 TCP_FLAG_ACK = 0x10 codecs.register_error("replace_with_space", lambda x: (" ", x.start + 1)) # type: ignore[attr-defined] def from_bytes_to_text(mode: str, binary: bytes) -> str: """ Make a text from a binary. :param mode: How to convert the binary to text. :return: A text converted from the binary. """ if mode == "text-based-protocol": # Keep all the characters used in text based protocols # * The unicodedata category names of control code start with C return "".join( " " if c == "\ufffd" or (c not in ("\n", "\r", "\t") and unicodedata.category(c)[0] == "C") else c for c in binary.decode("utf-8", errors="replace") ) elif mode == "human-readable": return binary.decode("utf-8", errors="replace_with_space") else: raise ValueError(f"Unknown text conversion mode: {mode}") class Streams: def __init__(self, server_ports: list[tuple[int, int]]): self.__tcp_streams: dict[tuple, Any] = {} self.__udp_streams: dict[tuple, Any] = {} self.__server_ports = server_ports def __make_item( self, bin2txt_mode: str, filter_keys: list[str] | None, protocol: str, sender_ip: str, sender_port: int, recipient_ip: str, recipient_port: int, stream_bytes: bytes, outgoing_bytes: bytes, incoming_bytes: bytes, ) -> dict[str, Any]: return { k: v for k, v in { "protocol": protocol, "client_ip": sender_ip, "client_port": sender_port, "server_ip": recipient_ip, "server_port": recipient_port, "stream_size": len(stream_bytes), "stream_text": from_bytes_to_text(bin2txt_mode, stream_bytes), "stream_base64": base64.b64encode(stream_bytes).decode(), "outgoing_size": len(outgoing_bytes), "outgoing_text": from_bytes_to_text(bin2txt_mode, outgoing_bytes), "outgoing_base64": base64.b64encode(outgoing_bytes).decode(), "incoming_size": len(incoming_bytes), "incoming_text": from_bytes_to_text(bin2txt_mode, incoming_bytes), "incoming_base64": base64.b64encode(incoming_bytes).decode(), }.items() if not filter_keys or (k in filter_keys) } def add_tcp( self, stream_index: int, tcp_flags: int, sender_ip: str, sender_port: int, recipient_ip: str, recipient_port: int, payload: bytes | None, ) -> None: """ Add a tcp packet to the streams. :param stream_index: The index of TCP streams. :param tcp_flags: The TCP flags. :param sender_ip: The IP address of a sender. :param sender_port: The port number of a sender. :param recipient_ip: The IP address of a recipient. :param recipient_port: The port number of a recipient. :param payload: The payload transmitted. """ key1 = (sender_ip, sender_port, recipient_ip, recipient_port, stream_index) key2 = (recipient_ip, recipient_port, sender_ip, sender_port, stream_index) if key1 in self.__tcp_streams: if payload: self.__tcp_streams[key1]["outgoing"] += payload self.__tcp_streams[key1]["stream"] += payload elif key2 in self.__tcp_streams: if payload: self.__tcp_streams[key2]["incoming"] += payload self.__tcp_streams[key2]["stream"] += payload else: mask = TCP_FLAG_SYN | TCP_FLAG_FIN | TCP_FLAG_ACK payload = payload if payload else b"" outgoing = None if (tcp_flags & mask) == TCP_FLAG_SYN: outgoing = True elif (tcp_flags & mask) == (TCP_FLAG_SYN | TCP_FLAG_ACK): outgoing = False else: matchs = any(port_range[0] < sender_port < port_range[1] for port_range in self.__server_ports) matchr = any(port_range[0] < recipient_port < port_range[1] for port_range in self.__server_ports) if matchs and (not matchr or (sender_port < recipient_port)): outgoing = False elif matchr and (not matchs or (recipient_port < sender_port)): outgoing = True if outgoing is None or outgoing is True: self.__tcp_streams[key1] = {"outgoing": payload, "incoming": b"", "stream": payload} else: self.__tcp_streams[key2] = {"outgoing": b"", "incoming": payload, "stream": payload} def add_udp(self, sender_ip: str, sender_port: int, recipient_ip: str, recipient_port: int, payload: bytes | None) -> None: """ Add a udp packet to the streams. :param sender_ip: The IP address of a sender. :param sender_port: The port number of a sender. :param recipient_ip: The IP address of a recipient. :param recipient_port: The port number of a recipient. :param payload: The payload transmitted. """ key1 = (sender_ip, sender_port, recipient_ip, recipient_port) key2 = (recipient_ip, recipient_port, sender_ip, sender_port) if key1 in self.__udp_streams: if payload: self.__udp_streams[key1]["outgoing"] += payload self.__udp_streams[key1]["stream"] += payload elif key2 in self.__udp_streams: if payload: self.__udp_streams[key2]["incoming"] += payload self.__udp_streams[key2]["stream"] += payload else: payload = payload if payload else b"" self.__udp_streams[key1] = {"outgoing": payload, "incoming": b"", "stream": payload} def build(self, bin2txt_mode: str, filter_keys: list[str] | None = None) -> list[dict[str, Any]]: """ Build the streams :param bin2txt_mode: How to convert the binary to text. :param filter_keys: Keys of the values to filter the stream information. :return: List of the streams. """ output = [] for key, stream in self.__tcp_streams.items(): sender_ip, sender_port, recipient_ip, recipient_port, stream_index = key output.append( self.__make_item( bin2txt_mode=bin2txt_mode, filter_keys=filter_keys, protocol="tcp", sender_ip=sender_ip, sender_port=sender_port, recipient_ip=recipient_ip, recipient_port=recipient_port, stream_bytes=stream.get("stream", b""), outgoing_bytes=stream.get("outgoing", b""), incoming_bytes=stream.get("incoming", b""), ) ) for key, stream in self.__udp_streams.items(): sender_ip, sender_port, recipient_ip, recipient_port = key output.append( self.__make_item( bin2txt_mode=bin2txt_mode, filter_keys=filter_keys, protocol="udp", sender_ip=sender_ip, sender_port=sender_port, recipient_ip=recipient_ip, recipient_port=recipient_port, stream_bytes=stream.get("stream", b""), outgoing_bytes=stream.get("outgoing", b""), incoming_bytes=stream.get("incoming", b""), ) ) return output class PcapParser: def __init__(self, server_ports: list[tuple[int, int]]): self.__server_ports = server_ports def __parse(self, pcap: pyshark.capture.capture.Capture) -> Streams: """ Parse a pcap :param pcap: The packet capture instance. :return: The stream data extracted. """ streams = Streams(server_ports=self.__server_ports) for packet in pcap: ip_layer = packet.get_multiple_layers("ip") if not ip_layer: continue sender_ip = ip_layer[0].get("src_host") recipient_ip = ip_layer[0].get("dst_host") if tcp_layer := packet.get_multiple_layers("tcp"): tcp = tcp_layer[0] payload = tcp.get("payload") streams.add_tcp( stream_index=int(tcp.get("stream", 0)), tcp_flags=int(tcp.flags, 0), sender_ip=str(sender_ip), sender_port=int(tcp.get("srcport")), recipient_ip=str(recipient_ip), recipient_port=int(tcp.get("dstport")), payload=payload.binary_value if payload else None, ) if udp_layer := packet.get_multiple_layers("udp"): udp = udp_layer[0] payload = udp.get("payload") streams.add_udp( sender_ip=str(sender_ip), sender_port=int(udp.get("srcport")), recipient_ip=str(recipient_ip), recipient_port=int(udp.get("dstport")), payload=payload.binary_value if payload else None, ) return streams def parse_file(self, pcap_file_path: str, wpa_password: str, rsa_key_file_path: str | None, pcap_filter: str) -> Streams: """ Parse a pcap file :param pcap_file_path: The file of the pcap to parse. :param wpa_password: The wpa password for the decryption :param rsa_key_file_path: The file of the decryption key. :param pcap_filter: A filter to apply on the PCAP. Same filter syntax as in Wireshark :return: The stream data extracted. """ # Create custom parameters custom_parameters = None if rsa_key_file_path: custom_parameters = {"-o": f'uat:rsa_keys:"{rsa_key_file_path}",""'} # Parse the pcap logging.getLogger("asyncio").setLevel(logging.ERROR) with open(os.devnull, "w") as devnull: sys.stderr = devnull try: with pyshark.FileCapture( pcap_file_path, display_filter=pcap_filter, decryption_key=wpa_password, encryption_type="WPA-PWD", keep_packets=False, custom_parameters=custom_parameters, ) as cap: return self.__parse(cap) except pyshark.capture.capture.TSharkCrashException: raise ValueError( "Could not find packets. Make sure that the file is a .cap/.pcap/.pcapng file, " "the filter is of the correct syntax and that the rsa key is added correctly." ) finally: sys.stderr = sys.__stderr__ def main(): args = demisto.args() try: pcap_entry_id = args.get("entry_id", "") bin2txt_mode = args.get("bin2txt_mode") or "text-based-protocol" rsa_decrypt_key_entry_id = args.get("rsa_decrypt_key_entry_id", "") wpa_password = args.get("wpa_password", "") pcap_filter = args.get("pcap_filter", "") filter_keys = argToList(args.get("filter_keys", "")) verbose = argToBoolean(args.get("verbose") or False) server_ports: list[tuple[int, int]] = [] for range_str in argToList(args.get("server_ports") or "1-49151"): port_range = range_str.split("-") if len(port_range) == 1: server_ports.append(((int(port_range[0].strip())), (int(port_range[0].strip())))) elif len(port_range) == 2: server_ports.append(((int(port_range[0].strip())), (int(port_range[1].strip())))) else: raise ValueError(f"Invalid port range: {range_str}") rsa_key_file_path = None if rsa_decrypt_key_entry_id: rsa_key_file_path = demisto.getFilePath(rsa_decrypt_key_entry_id).get("path") parser = PcapParser(server_ports=server_ports) streams = parser.parse_file( pcap_file_path=demisto.getFilePath(pcap_entry_id).get("path"), wpa_password=wpa_password, rsa_key_file_path=rsa_key_file_path, pcap_filter=pcap_filter, ) contents = streams.build(bin2txt_mode=bin2txt_mode, filter_keys=filter_keys) contents = [dict(entry_id=pcap_entry_id, **content) for content in contents] # Create a summary entry table_headers = [ "protocol", "client_ip", "client_port", "server_ip", "server_port", "stream_size", "outgoing_size", "incoming_size", ] if filter_keys: table_headers = [k for k in table_headers if k in filter_keys] if not table_headers: readable_output = "No summary" else: readable_output = tableToMarkdown( "PCAP Stream Summary", [{k: stringEscapeMD(str(ent[k]), True, True) for k in ent.keys() & table_headers} for ent in contents], headers=table_headers, headerTransform=lambda s: stringEscapeMD(string_to_table_header(s), True, True), ) return_results( CommandResults(outputs_prefix="PCAPStream", outputs=contents, readable_output=readable_output, raw_response=contents) ) # Create stream entries if ( verbose and contents and (not filter_keys or ({"stream_text", "outgoing_text", "incoming_text"}.intersection(filter_keys))) ): table_headers = [ "protocol", "client_ip", "client_port", "server_ip", "server_port", "stream_size", "outgoing_size", "incoming_size", "stream_text", "outgoing_text", "incoming_text", ] if filter_keys: table_headers = [k for k in table_headers if k in filter_keys] for ent in contents: readable_output = tableToMarkdown( "PCAP Stream", {k: stringEscapeMD(str(ent[k]), True, True) for k in ent.keys() & table_headers}, headers=table_headers, headerTransform=lambda s: stringEscapeMD(string_to_table_header(s), True, True), ) return_results(CommandResults(readable_output=readable_output)) except Exception as e: return_error(f"Failed to extract streams. Error: {e!s}") if __name__ in ("__main__", "builtin", "builtins"): main()