import re import traceback from typing import Any import demistomock as demisto import pyshark from CommonServerPython import * """GLOBAL VARS""" BAD_CHARS = ["[", "]", ">", "<", "'", " Layer", " ", "{", "}"] EMAIL_REGEX = r"\b[A-Za-z0-9._%=+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\b" IP_REGEX = r"\b(?:(?:25[0-5]|2[0-4][0-9]|1[0-9][0-9]|[1-9]?[0-9])\.){3}(?:25[0-5]|2[0-4][0-9]|1[0-9][0-9]|[1-9]?[0-9])\b" URL_REGEX = r"http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\(\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+" PRAGMA_REGEX = r"Pragma: ([^\\]+)" TYPE_REGEX = r"Type: (.+)" CLASS_REGEX = r"Class: (.+)" COMMAND_REGEX = r"Command: (.+)" SNAME_REGEX = r"SNameString: (.+)" MESSAGE_CODE = r"Message Code: (.+)" RESPONSE_CODE = r"Response code: (.+)" ALL_SUPPORTED_PROTOCOLS = [ "HTTP", "DNS", "LLMNR", "SYSLOG", "SMTP", "NETBIOS", "ICMP", "KERBEROS", "TELNET", "SSH", "IRC", "FTP", "SMB2", ] class PCAP: @logger def __init__(self, is_reg_extract: bool, extracted_protocols: list, homemade_regex: str, unique_ips: bool, entry_id: str): """ PCAP class. This class will contain all data and functions in order to mine the pcap. Args: is_reg_extract: Whether to extract regular expressions from the PCAP. URL, IP, Email extracted_protocols: A list of protocols to extract homemade_regex: A regex to extract from the PCAP unique_ips: Whether to export to context the unique source and dest IPs from the PCAP. """ # setup data structures self.hierarchy: dict[str, int] = {} self.num_of_packets = 0 self.tcp_streams = 0 self.udp_streams = 0 self.last_packet = 0 self.bytes_transmitted = 0 self.min_time = float("inf") self.max_time = -float("inf") self.conversations: dict[tuple, Any] = {} self.flows: dict[tuple, Any] = {} self.unique_source_ip: set = set() self.unique_dest_ip: set = set() self.ips_extracted: set = set() self.urls_extracted: set = set() self.emails_extracted: set = set() self.homemade_extracted: set = set() self.last_layer: set = set() self.irc_data: list = [] self.protocol_data: dict[str, Any] = {} self.entry_id = entry_id self.extracted_protocols = extracted_protocols self.homemade_regex = homemade_regex self.unique_ips = unique_ips for protocol in extracted_protocols: self.protocol_data[protocol] = {} # Regex compilation if "LLMNR" in extracted_protocols: self.llmnr_type = re.compile("Type: (.*)\n") self.llmnr_class = re.compile("Class: (.*)\n") self.llmnr_dict: dict = {} if is_reg_extract: self.reg_ip = re.compile(IP_REGEX) self.reg_email = re.compile(EMAIL_REGEX) self.reg_url = re.compile(URL_REGEX) if "HTTP" in extracted_protocols: self.reg_pragma = re.compile(PRAGMA_REGEX) if "ICMP" in extracted_protocols: self.icmp_data: set = set() if "DNS" in extracted_protocols or "NETBIOS" in extracted_protocols or "ICMP" in extracted_protocols: self.reg_type = re.compile(TYPE_REGEX) if "NETBIOS" in extracted_protocols: self.reg_class = re.compile(CLASS_REGEX) if "SMB2" in extracted_protocols: self.reg_cmd = re.compile(COMMAND_REGEX) if "KERBEROS" in extracted_protocols: self.reg_sname = re.compile(SNAME_REGEX) self.kerb_data: list = [] if "SSH" in extracted_protocols: self.ssh_data = { "EntryID": entry_id, "ClientProtocols": set(), "ServerProtocols": set(), "KeyExchangeMessageCode": set(), } self.reg_message_code = re.compile(MESSAGE_CODE) if "FTP" in extracted_protocols: self.reg_res_code = re.compile(RESPONSE_CODE) if "TELNET" in extracted_protocols: self.telnet_data: set = set() self.telnet_commands: set = set() if homemade_regex: self.reg_homemade = re.compile(self.homemade_regex) @logger def extract_dns(self, packet): dns_layer = packet.dns temp_dns = { "EntryID": self.entry_id, "ID": dns_layer.get("id"), "Request": dns_layer.get("qry_name"), "Response": dns_layer.get("a"), "Type": self.reg_type.findall(str(dns_layer))[0] if self.reg_type.findall(str(dns_layer)) else None, } add_to_data(self.protocol_data["DNS"], temp_dns) @logger def extract_kerberos(self, packet): kerb_layer = packet.kerberos sname_results = self.reg_sname.findall(str(kerb_layer)) self.kerb_data.append( { "EntryID": self.entry_id, "Realm": kerb_layer.get("realm"), "CName": kerb_layer.get("CNameString"), "SName": sname_results if sname_results else None, } ) @logger def extract_telnet(self, packet): telnet_layer = packet.telnet for message in telnet_layer._get_all_field_lines(): if "Data:" in message: self.telnet_data.add(message.lstrip("Data: ")) # noqa: B005 continue if ":" not in message: self.telnet_commands.add(message.lstrip("\t").rstrip("\n")) @logger def extract_llmnr(self, packet): llmnr_layer = packet.llmnr llmnr_layer_string = str(llmnr_layer) query_type_results = self.llmnr_type.findall(llmnr_layer_string) query_class_results = self.llmnr_class.findall(llmnr_layer_string) llmnr_data = { "EntryID": self.entry_id, "ID": llmnr_layer.get("dns_id"), "QueryType": None if len(query_type_results) == 0 else query_type_results[0], "QueryClass": None if len(query_class_results) == 0 else query_class_results[0], "QueryName": str(llmnr_layer.get("dns_qry_name")), "Questions": int(llmnr_layer.get("dns_count_queries")), } add_to_data(self.protocol_data["LLMNR"], llmnr_data) @logger def extract_syslog(self, packet): syslog_layer = packet.syslog syslog_data = { "EntryID": self.entry_id, "ID": syslog_layer.get("msgid"), "Message": syslog_layer.get("msg"), "Hostname": syslog_layer.get("hostname"), "Timestamp": syslog_layer.get("timestamp"), } add_to_data(self.protocol_data["SYSLOG"], syslog_data) @logger def extract_imf(self, packet): imf_layer = packet.imf imf_data = { "EntryID": self.entry_id, "ID": imf_layer.get("Message-ID", -1), "To": strip(imf_layer.get("to"), ["<", ">"]), "From": strip(imf_layer.get("from"), ["<", ">"]), "Subject": imf_layer.get("subject"), "MimeVersion": imf_layer.get("mime-version"), } add_to_data(self.protocol_data["SMTP"], imf_data) @logger def extract_smtp(self, packet): smtp_layer = packet.smtp if smtp_layer.get_field("req_parameter") is None: parameters = [] else: parameters = smtp_layer.req_parameter.split(":") smtp_data = {"EntryID": self.entry_id, "ID": packet.tcp.seq} if len(parameters) == 2: smtp_data[parameters[0].title()] = strip(parameters[1], ["<", ">"]) add_to_data(self.protocol_data["SMTP"], smtp_data, packet.tcp.nxtseq) @logger def extract_smb(self, packet): smb_layer = packet.smb2 command_results = self.reg_cmd.findall(str(smb_layer)) smb_data = { "EntryID": self.entry_id, "ID": smb_layer.get("sesid", -1), "UserName": smb_layer.get("ntlmssp_auth_username"), "Domain": smb_layer.get("ntlmssp_auth_domain"), "HostName": smb_layer.get("ntlmssp_auth_hostname"), "Command": command_results[0] if command_results else None, "FileName": smb_layer.get("smb2.filename"), "Tree": smb_layer.get("tree"), } add_to_data(self.protocol_data["SMB2"], smb_data) @logger def extract_netbios(self, packet): netbios_layer = packet.nbns type_results = self.reg_type.findall(str(netbios_layer)) class_results = self.reg_class.findall(str(netbios_layer)) netbios_data = { "EntryID": self.entry_id, "ID": netbios_layer.get("id", -1), "Name": netbios_layer.get("name"), "Type": type_results if type_results else None, "Class": class_results if class_results else None, } add_to_data(self.protocol_data["NETBIOS"], netbios_data) @logger def extract_icmp(self, packet): icmp_layer = packet.icmp type_results = self.reg_type.findall(str(icmp_layer)) for result in type_results: self.icmp_data.add(result) @logger def extract_ssh(self, packet): ssh_layer = packet.ssh protocol = ssh_layer.get("protocol") message_code_results = self.reg_message_code.findall(str(ssh_layer)) if protocol and ssh_layer.get("direction") == 1: # direction is server to client self.ssh_data["ServerProtocols"].add(protocol) # type: ignore[attr-defined] if protocol and ssh_layer.get("direction") == 0: # direction is client to server self.ssh_data["ClientProtocols"].add(protocol) # type: ignore[attr-defined] if message_code_results and message_code_results: self.ssh_data["KeyExchangeMessageCode"].add(message_code_results[0]) # type: ignore[attr-defined] @logger def extract_irc(self, packet): irc_layer = packet.irc if irc_layer.get("request"): command = irc_layer.get("request_command") trailer = irc_layer.get("request_trailer", "") prefix = irc_layer.get("request_prefix", "") parameters = irc_layer.get("request").replace(command, "").replace(trailer, "").replace(prefix, "").split(" ") parameters.remove(" ") irc_data = { "EntryID": self.entry_id, "ID": packet.tcp.get("ack"), "RequestCommand": command, "RequestTrailer": trailer, "RequestPrefix": prefix, "RequestParameters": parameters, } else: command = irc_layer.get("response_command") trailer = irc_layer.get("response_trailer", "") prefix = irc_layer.get("response_prefix", "") parameters = irc_layer.get("response").replace(command, "").replace(trailer, "").replace(prefix, "").split(" ") parameters.remove(" ") irc_data = { "EntryID": self.entry_id, "ID": packet.tcp.get("seq"), "ResponseCommand": command, "ResponseTrailer": trailer, "ResponsePrefix": prefix, "ResponseParameters": parameters, } add_to_data(self.protocol_data["IRC"], irc_data, next_id=packet.tcp.nxtseq) @logger def extract_ftp(self, packet): ftp_layer = packet.ftp res_code_results = self.reg_res_code.findall(str(ftp_layer)) ftp_data = { "EntryID": self.entry_id, "ID": packet.tcp.get("seq"), "RequestCommand": ftp_layer.get("request_command"), "ResponseArgs": ftp_layer.get("response_arg"), "ResponseCode": res_code_results[0] if res_code_results else None, } if ftp_data["ResponseCode"]: # if packet is a response, don't update ID. This is because FTP may have a lot of requests and # responses in the same conversation add_to_data(self.protocol_data["FTP"], ftp_data) else: add_to_data(self.protocol_data["FTP"], ftp_data, next_id=packet.tcp.ack) @logger def extract_context_from_packet(self, packet, layers_str: str, is_reg_extract: bool = False) -> None: """ Get a packet and extract context from it for the specified protocols specified in `self.extracted_protocols` Args: packet: The packet itself. layers_str: A comma-seperated string of the layers in the packet. is_reg_extract: Whether to export to context the unique source and dest IPs from the PCAP. """ layers = layers_str.split(",") if is_reg_extract: self.ips_extracted.update(self.reg_ip.findall(str(packet))) self.emails_extracted.update(self.reg_email.findall(str(packet))) self.urls_extracted.update(self.reg_url.findall(str(packet))) if self.homemade_regex: self.homemade_extracted.update(self.reg_homemade.findall(str(packet))) if "DNS" in self.extracted_protocols and "DNS" in layers: return self.extract_dns(packet) if "KERBEROS" in self.extracted_protocols and "KERBEROS" in layers: return self.extract_kerberos(packet) if "TELNET" in self.extracted_protocols and "TELNET" in layers: return self.extract_telnet(packet) if "LLMNR" in self.extracted_protocols and "LLMNR" in layers: return self.extract_llmnr(packet) if "SYSLOG" in self.extracted_protocols and "SYSLOG" in layers: return self.extract_syslog(packet) if "SMB2" in self.extracted_protocols and "SMB2" in layers: return self.extract_smb(packet) if "NETBIOS" in self.extracted_protocols and "NETBIOS" in layers: return self.extract_netbios(packet) if "ICMP" in self.extracted_protocols and "ICMP" in layers: return self.extract_icmp(packet) if "SSH" in self.extracted_protocols and "SSH" in layers: return self.extract_ssh(packet) if "IRC" in self.extracted_protocols and "IRC" in layers: return self.extract_irc(packet) if "FTP" in self.extracted_protocols and "FTP" in layers: return self.extract_ftp(packet) if "SMTP" in self.extracted_protocols: # noqa: RET503 if "IMF" in layers: return self.extract_imf(packet) if "SMTP" in layers: # noqa: RET503 return self.extract_smtp(packet) @logger def get_outputs(self, conversation_number_to_display=15, is_flows=False, is_reg_extract=False): if self.num_of_packets == 0: return "## No packets found.\nTry changing the filter, if applied.", {}, {} md = ( f"## PCAP Info:\n" f"Between {formatEpochDate(self.min_time)} and {formatEpochDate(self.max_time)} " f"there were {self.num_of_packets} " f"packets transmitted in {self.tcp_streams + self.udp_streams} streams.\n" ) md += "#### Protocol Breakdown\n" md += hierarchy_to_md(self.hierarchy) md += f"#### Top {conversation_number_to_display} Conversations\n" md += conversations_to_md(self.conversations, conversation_number_to_display) if is_flows: md += f"#### Top {conversation_number_to_display} Flows\n" md += flows_to_md(self.flows, conversation_number_to_display) # Entry Context general_context = { "EntryID": self.entry_id, "Bytes": self.bytes_transmitted, "Packets": self.num_of_packets, "StreamCount": self.tcp_streams + self.udp_streams, "UniqueSourceIP": len(self.unique_source_ip), "UniqueDestIP": len(self.unique_dest_ip), "StartTime": formatEpochDate(self.min_time), "EndTime": formatEpochDate(self.max_time), "Protocols": list(self.last_layer), } ec = {"PCAPResults(val.EntryID == obj.EntryID)": general_context} for protocol in self.extracted_protocols: if self.protocol_data[protocol]: ec[f"PCAPResults{protocol}"] = list(self.protocol_data[protocol].values()) # type: ignore[assignment] if "ICMP" in self.extracted_protocols and self.icmp_data: ec["PcapResultsICMP"] = list(self.icmp_data) # type: ignore[assignment] if "KERBEROS" in self.extracted_protocols and self.kerb_data: ec["PCAPResultsKERBEROS"] = self.kerb_data # type: ignore[assignment] if "SSH" in self.extracted_protocols: temp = { "EntryID": self.entry_id, "ClientProtocols": list(self.ssh_data["ClientProtocols"]), "ServerProtocols": list(self.ssh_data["ServerProtocols"]), "KeyExchangeMessageCode": list(self.ssh_data["KeyExchangeMessageCode"]), } ec["PCAPResultsSSH"] = assign_params(**temp) if "TELNET" in self.extracted_protocols: ec["PCAPResultsTelnet"] = { "Commands": list(self.telnet_commands), "Data": list(self.telnet_data), "EntryID": self.entry_id, } if is_flows: ec["PCAPResultsFlow"] = flows_to_ec(self.flows) if is_reg_extract: general_context["IP"] = list(self.ips_extracted) general_context["URL"] = list(self.urls_extracted) general_context["Email"] = list(self.emails_extracted) if self.homemade_regex: general_context["Regex"] = list(self.homemade_extracted) if self.unique_ips: all_ips = self.unique_source_ip.copy() all_ips.update(self.unique_dest_ip) if general_context.get("IP"): general_context.get("IP").append(list(all_ips)) # type: ignore[union-attr] else: general_context["IP"] = list(all_ips) general_context["SourceIP"] = list(self.unique_source_ip) general_context["DestIP"] = list(self.unique_dest_ip) return md, ec, general_context @logger def mine( self, file_path: str, wpa_password: str, rsa_key_file_path: str, is_flows: bool, is_reg_extract: bool, pcap_filter: str, pcap_filter_new_file_path: str, ) -> None: """ The main function of the script. Mines the PCAP. Args: file_path: The PCAP's file path. wpa_password: The wpa password for the decryption is_flows: Whether to extract flows. is_reg_extract: Whether to extract regexes from the PCAP. pcap_filter: A filter to apply on the PCAP. Same filter syntax as in Wireshark pcap_filter_new_file_path: The new path to save the filtered PCAP in """ cap = None try: custom_parameters = None if rsa_key_file_path: custom_parameters = {"-o": f'uat:rsa_keys:"{rsa_key_file_path}",""'} cap = pyshark.FileCapture( file_path, display_filter=pcap_filter, output_file=pcap_filter_new_file_path, decryption_key=wpa_password, encryption_type="WPA-PWD", keep_packets=False, custom_parameters=custom_parameters, ) for packet in cap: self.last_packet = int(packet.number) self.last_layer.add(packet.layers[-1].layer_name.upper()) layers_string = str(packet.layers) # remove duplicate layer names such as [ETH,DATA,DATA] -> # [ETH, DATA] layers: list = list(dict.fromkeys(layers_string.split(","))) layers = strip(str(layers)) self.hierarchy[layers] = self.hierarchy.get(layers, 0) + 1 # type: ignore[index, call-overload] # update times packet_epoch_time = float(packet.frame_info.get("time_epoch")) self.max_time = max(self.max_time, packet_epoch_time) self.min_time = min(self.min_time, packet_epoch_time) # count packets self.num_of_packets += 1 # count bytes self.bytes_transmitted += int(packet.length) # count num of streams + get src/dest ports tcp = packet.get_multiple_layers("tcp") tcp_or_udp = "Not TCP and not UDP" if tcp: self.tcp_streams = max(int(tcp[0].get("stream", 0)) + 1, self.tcp_streams) src_port = int(tcp[0].get("srcport", 0)) dest_port = int(tcp[0].get("dstport", 0)) tcp_or_udp = "TCP" udp = packet.get_multiple_layers("udp") if udp: self.udp_streams = max(int(udp[0].get("stream", 0)) + 1, self.udp_streams) src_port = int(udp[0].get("srcport", 0)) dest_port = int(udp[0].get("dstport", 0)) tcp_or_udp = "UDP" # add conversations ip_layer = packet.get_multiple_layers("ip") if ip_layer: a = ip_layer[0].get("src_host", "") b = ip_layer[0].get("dst_host") self.unique_source_ip.add(a) self.unique_dest_ip.add(b) # generate flow data if is_flows: if "src_port" not in locals(): continue if (b, dest_port, a, src_port) in self.flows: b, a, src_port, dest_port = a, b, dest_port, src_port flow = (a, src_port, b, dest_port) flow_data = self.flows.get( flow, { "EntryID": self.entry_id, "Transport": tcp_or_udp, "min_time": float("inf"), "max_time": -float("inf"), "bytes": 0, "counter": 0, }, ) flow_data["min_time"] = min(flow_data["min_time"], packet_epoch_time) flow_data["max_time"] = max(flow_data["min_time"], packet_epoch_time) flow_data["bytes"] += int(packet.length) flow_data["counter"] += 1 self.flows[flow] = flow_data # gather http data if "HTTP" in self.extracted_protocols: http_layer = packet.get_multiple_layers("http") if http_layer: http_layer = http_layer[0] all_fields = http_layer._all_fields temp_http = { "EntryID": self.entry_id, "ID": http_layer.get("request_in", packet.number), "RequestAgent": all_fields.get("http.user_agent"), "RequestHost": all_fields.get("http.host"), "RequestSourceIP": a, "RequestURI": http_layer.get("request_full_uri"), "RequestMethod": http_layer.get("request_method"), "RequestVersion": http_layer.get("request_version"), "RequestAcceptEncoding": http_layer.get("accept_encoding"), "RequestPragma": self.reg_pragma.findall(str(http_layer))[0] if self.reg_pragma.findall(str(http_layer)) else None, "RequestAcceptLanguage": http_layer.get("accept_language"), "RequestCacheControl": http_layer.get("cache_control"), } # if the packet is a response if all_fields.get("http.response"): temp_http.update( { "EntryID": self.entry_id, "ResponseStatusCode": http_layer.get("response_code"), "ResponseVersion": all_fields.get("http.response.version"), "ResponseCodeDesc": http_layer.get("response_code_desc"), "ResponseContentLength": http_layer.get("content_length"), "ResponseContentType": http_layer.get("content_type"), "ResponseDate": formatEpochDate(packet_epoch_time), } ) add_to_data(self.protocol_data["HTTP"], temp_http) if (b, a) in self.conversations: a, b = b, a hosts = (a, b) self.conversations[hosts] = self.conversations.get(hosts, 0) + 1 self.extract_context_from_packet(packet, layers, is_reg_extract) 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: if cap: cap.close() """HELPER FUNCTIONS""" @logger def strip(s: str, bad_chars=None): """ Args: s: string to strip bad_chars: all characters to remove from string. Returns: The input s without the bad_chars """ if bad_chars is None: bad_chars = BAD_CHARS temp = s for char in bad_chars: temp = temp.replace(char, "") return temp @logger def hierarchy_to_md(hierarchy: dict) -> str: """ Args: hierarchy: a dictionary of layer hierarchy for all packets Returns: A markdown string for displaying the hierarchy in a nice view. The script also counts the number of occurrences each hierarchy. """ final_dict: dict[str, Any] = {} num_of_all_packets = 0 for k in hierarchy: layer_heir = "" for layer in k.split(","): layer_heir += " -> " + layer final_dict[layer_heir] = final_dict.get(layer_heir, 0) + hierarchy[k] num_of_all_packets += hierarchy[k] md = "|Layer| # of Packets|% of Packets|\n|---|----|---|\n" for key in sorted(final_dict): md += f"|{key}|{final_dict[key]}|{round(final_dict[key] / num_of_all_packets, 3) * 100}%|\n" return md @logger def conversations_to_md(conversations: dict, disp_num: int) -> str: """ Args: conversations: a raw dictionary of conversations. disp_num: The limit of conversations to display. Returns: A mardkown of <=disp_num of conversations, ordered in descending order. """ md = "|A|B|# of Packets|\n|---|---|---|\n" ordered_conv_list = sorted(conversations.items(), key=lambda x: x[1], reverse=True) disp_num = min(disp_num, len(ordered_conv_list)) for conv in ordered_conv_list[:disp_num]: (ipA, ipB), data = conv md += f"|{ipA}|{ipB}|{data}|\n" return md @logger def flows_to_md(flows: dict, disp_num: int) -> str: """ Args: flows: a raw dictionary of flows. disp_num: The limit of flows to display. Returns: A mardkown of <=disp_num of flows, ordered in descending order. """ md = "|A|port|B|port|# of Packets|\n|---|---|---|---|---|\n" ordered_flow_list = sorted(flows.items(), key=lambda x: x[1].get("counter"), reverse=True) disp_num = min(disp_num, len(ordered_flow_list)) for flow in ordered_flow_list[:disp_num]: (ipA, portA, ipB, portB), data = flow md += f'|{ipA}|{portA}|{ipB}|{portB}|{data.get("counter", 0)}|\n' return md @logger def flows_to_ec(flows: dict) -> list: """ Args: flows: A dictionary that hold the flows data Returns: flows data in ec format. """ flows_ec = [] for flow in flows: flow_data = flows[flow] flow_ec = { "SourceIP": flow[0], "SourcePort": flow[1], "DestIP": flow[2], "DestPort": flow[3], "Duration": round(flow_data.get("max_time", 0) - flow_data.get("min_time", 0)), "StartTime": formatEpochDate(flow_data.get("min_time", 0)), "EndTime": formatEpochDate(flow_data.get("max_time", 0)), "Bytes": flow_data.get("bytes", 0), "EntryID": flow_data.get("EntryID"), "Transport": flow_data.get("Transport"), } flows_ec.append(flow_ec) return flows_ec @logger def add_to_data(d: dict, data: dict, next_id: int = None) -> None: """ updates dictionary d to include/update the data. Also removes None values. Args: d: a Dictionary of ID: data to which we want to update the data according to ID data: the data to update. data must have an "ID" field. prev_id: when there's no ID in a packet, we'd like to keep link the data by the next sequence number from TCP protocol. In order to keep this up to date we'd like to know what the next sequence ID and then update it to be the current acknowledge number. Returns: updates dictionary d to include/update the data. Also removes None values. """ data_id = data.get("ID") if not data_id: return else: if not d.get(data_id): if list(data.keys()) == ["ID"] or list(data.keys()) == ["ID", "EntryID"]: # The dictionary doesn't exist and is empty (except ID/EntryID) return if next_id: d[next_id] = assign_params(**data) else: d[data_id] = assign_params(**data) else: if next_id and next_id > data_id: # id exists but we want to keep its next seq as ID (and also next seq id is larger therefore # there's another packet in the future) temp = d.pop(data_id) temp.update(assign_params(**data)) d[next_id] = temp else: # ID exists, we just want to update it. d[data_id].update(assign_params(**data)) """MAIN""" def main(): args = demisto.args() entry_id = args.get("entry_id", "") file_path = demisto.getFilePath(entry_id).get("path") wpa_password = args.get("wpa_password", "") rsa_decrypt_key_entry_id = args.get("rsa_decrypt_key_entry_id", "") rsa_key_file_path = None if rsa_decrypt_key_entry_id: rsa_key_file_path = demisto.getFilePath(rsa_decrypt_key_entry_id).get("path") conversation_number_to_display = int(args.get("convs_to_display", "15")) extracted_protocols = argToList(args.get("protocol_output", "")) if "All" in extracted_protocols: extracted_protocols = ALL_SUPPORTED_PROTOCOLS is_flows = True is_reg_extract = args.get("extract_strings", "False") == "True" pcap_filter = args.get("pcap_filter", "") homemade_regex = args.get("custom_regex", "") # 'Layer (.+):' pcap_filter_new_file_path = "" pcap_filter_new_file_name = args.get("filtered_file_name", "") unique_ips = args.get("extract_ips", "False") == "True" if pcap_filter_new_file_name: temp = demisto.uniqueFile() pcap_filter_new_file_path = demisto.investigation()["id"] + "_" + temp try: pcap = PCAP(is_reg_extract, extracted_protocols, homemade_regex, unique_ips, entry_id) pcap.mine(file_path, wpa_password, rsa_key_file_path, is_flows, is_reg_extract, pcap_filter, pcap_filter_new_file_path) hr, ec, raw = pcap.get_outputs(conversation_number_to_display, is_flows, is_reg_extract) return_outputs(hr, ec, raw) except Exception as e: return_error(f"Unexpected error: {e!s}", error=traceback.format_exc()) if pcap_filter_new_file_name: demisto.results( {"Contents": "", "ContentsFormat": formats["text"], "Type": 3, "File": pcap_filter_new_file_name, "FileID": temp} ) if __name__ in ["__main__", "builtin", "builtins"]: main()