ScanPcap

Extract files from pcap/pcapng files, use Suricata to match alert signatures.

Options

limit: Maximum number of files to extract. Defaults to 1000.

Source code in strelka/src/python/strelka/scanners/scan_pcap.py

class ScanPcap(strelka.Scanner):
    """Extract files from pcap/pcapng files, use Suricata to match alert signatures.

    Options:
        limit: Maximum number of files to extract.
            Defaults to 1000.
    """

    def scan(
        self, data: bytes, file: Dict[str, Any], options: Dict[str, Any], expire_at: int
    ) -> None:
        """Process pcap/pcapng data and extract files using Zeek and optionally run Suricata.

        Args:
            data: Raw pcap/pcapng data.
            file: File metadata.
            options: Scanner options.
            expire_at: Expiration timestamp.
        """
        self.expire_at = expire_at
        self.file_limit: int = options.get("limit", 1000)
        self.tmp_directory: str = options.get("tmp_file_directory", "/tmp/")
        self.zeek_conn: Optional[bool] = options.get("zeek_conn", True)
        self.zeek_conn_limit: Optional[int] = options.get("zeek_conn_limit", 100)
        self.zeek_proto: Optional[bool] = options.get("zeek_proto", True)
        self.suricata_config: Optional[str] = options.get("suricata_config", None)
        self.suricata_rules: Optional[str] = options.get("suricata_rules", None)
        self.suricata_alert_dedupe: bool = options.get("suricata_alert_dedupe", False)
        self.scanner_timeout: int = options.get("scanner_timeout", 120)

        self.event["total"] = {"files": 0, "extracted": 0}
        self.event["files"] = []

        try:
            # Check if zeek package is installed
            if not shutil.which("zeek"):
                self.flags.append("zeek_not_installed_error")
                return
        except Exception as e:
            self.flags.append(str(e))
            return

        with tempfile.NamedTemporaryFile(dir=self.tmp_directory, mode="wb") as tmp_data:
            tmp_data.write(data)
            tmp_data.flush()
            tmp_data.seek(0)

            self.zeek_extract(tmp_data.name)

            if self.suricata_rules:
                try:
                    # Check if suricata package is installed
                    if not shutil.which("suricata"):
                        self.flags.append("suricata_not_installed_error")
                        return

                    self.suricata(tmp_data.name)
                except Exception as e:
                    self.flags.append(str(e))

    def zeek_extract(self, pcap_path: str) -> None:
        """Extract files from pcap using Zeek.

        Args:
            pcap_path: Path to the pcap file.
        """
        with tempfile.TemporaryDirectory() as tmp_zeek:
            try:

                # Zeek outputs to the current working directory by default
                os.chdir(tmp_zeek)

                stdout, stderr = subprocess.Popen(
                    [
                        "zeek",
                        "-r",
                        pcap_path,
                        "/opt/zeek/share/zeek/policy/frameworks/files/extract-all-files.zeek",
                        f"FileExtract::prefix={os.path.join(tmp_zeek, 'files')}",
                        "LogAscii::use_json=T",
                    ],
                    stdout=subprocess.PIPE,
                    stderr=subprocess.PIPE,
                ).communicate(timeout=self.scanner_timeout)

                files_log_path = os.path.join(tmp_zeek, "files.log")

                if os.path.exists(files_log_path):
                    self._process_files_log(
                        files_log_path, os.path.join(tmp_zeek, "files")
                    )
                else:
                    self.flags.append("zeek_no_file_log")

                if self.zeek_conn:
                    # Parse the conn.log file to get basic connection information
                    connections = []
                    total_connections = 0
                    conn_log = os.path.join(tmp_zeek, "conn.log")

                    if os.path.exists(conn_log):
                        with open(conn_log, "r") as f:
                            for line in f:
                                if line.startswith("#"):
                                    continue

                                total_connections += 1

                                if (
                                    self.zeek_conn_limit > 0
                                    and len(connections) >= self.zeek_conn_limit
                                ):
                                    if "zeek_conn_limit_reached" not in self.flags:
                                        self.flags.append("zeek_conn_limit_reached")
                                    continue

                                try:
                                    conn_data = json.loads(line)
                                    connection = {
                                        "uid": conn_data.get("uid"),
                                        "source_ip": conn_data.get("id.orig_h"),
                                        "source_port": conn_data.get("id.orig_p"),
                                        "dest_ip": conn_data.get("id.resp_h"),
                                        "dest_port": conn_data.get("id.resp_p"),
                                        "protocol": conn_data.get("proto"),
                                        "service": conn_data.get("service"),
                                        "duration": conn_data.get("duration"),
                                        "bytes_orig": conn_data.get("orig_bytes"),
                                        "bytes_resp": conn_data.get("resp_bytes"),
                                        "conn_state": conn_data.get("conn_state"),
                                    }

                                    connections.append(connection)
                                except json.JSONDecodeError:
                                    continue

                    if self.zeek_proto:
                        # Check for protocol-specific logs to enhance connection details
                        protocol_logs = [
                            "http.log",
                            "dns.log",
                            "ssl.log",
                            "ssh.log",
                            "smtp.log",
                            "ftp.log",
                            "rdp.log",
                            "sip.log",
                            "smb_mapping.log",
                        ]

                        # Map to store protocol details by connection UID
                        protocol_details = defaultdict(dict)

                        for log_name in protocol_logs:
                            log_path = os.path.join(tmp_zeek, log_name)
                            if os.path.exists(log_path):
                                protocol = log_name.split(".")[0]

                                with open(log_path, "r") as f:
                                    for line in f:
                                        if line.startswith("#"):
                                            continue
                                        try:
                                            log_data = json.loads(line)
                                            uid = log_data.get("uid")
                                            if uid:
                                                # Extract relevant details based on protocol
                                                if protocol == "http":
                                                    protocol_details[uid][protocol] = {
                                                        "method": log_data.get(
                                                            "method"
                                                        ),
                                                        "host": log_data.get("host"),
                                                        "uri": log_data.get("uri"),
                                                        "user_agent": log_data.get(
                                                            "user_agent"
                                                        ),
                                                        "status_code": log_data.get(
                                                            "status_code"
                                                        ),
                                                    }
                                                elif protocol == "dns":
                                                    protocol_details[uid][protocol] = {
                                                        "query": log_data.get("query"),
                                                        "qtype": log_data.get(
                                                            "qtype_name"
                                                        ),
                                                        "answers": log_data.get(
                                                            "answers"
                                                        ),
                                                    }
                                                elif protocol == "ssl":
                                                    protocol_details[uid][protocol] = {
                                                        "server_name": log_data.get(
                                                            "server_name"
                                                        ),
                                                        "subject": log_data.get(
                                                            "subject"
                                                        ),
                                                        "issuer": log_data.get(
                                                            "issuer"
                                                        ),
                                                        "version": log_data.get(
                                                            "version"
                                                        ),
                                                    }
                                                else:
                                                    # Generic extraction for other protocols
                                                    protocol_details[uid][protocol] = {
                                                        k: v
                                                        for k, v in log_data.items()
                                                        if k != "uid" and v is not None
                                                    }
                                        except json.JSONDecodeError:
                                            continue

                        # Merge protocol details with the basic connection information
                        for connection in connections:
                            uid = connection.get("uid")
                            if uid in protocol_details:
                                connection["protocol_details"] = protocol_details[uid]

                    self.event["total"]["connections"] = total_connections
                    self.event["connections"] = connections

            except strelka.ScannerTimeout:
                raise
            except Exception:
                self.flags.append("zeek_extract_process_error")
                raise

    def _process_files_log(self, log_path: str, extract_dir: str) -> None:
        """Process Zeek's files.log and upload extracted files.

        Args:
            log_path: Path to the files.log file.
            extract_dir: Directory where extracted files are stored.
        """
        try:
            with open(log_path, "r") as json_file:
                # files.log is one JSON object per line, convert to array
                file_events = json.loads(
                    "[" + ",".join(json_file.read().splitlines()) + "]"
                )

                for file_event in file_events:
                    if self.event["total"]["extracted"] >= self.file_limit:
                        self.flags.append("pcap_file_limit_error")
                        break

                    self.event["total"]["files"] += 1
                    self.event["files"].append(file_event)

                    extracted_file_path = os.path.join(
                        extract_dir, file_event["extracted"]
                    )

                    try:
                        if os.path.exists(extracted_file_path):
                            self.upload(extracted_file_path, self.expire_at)
                            self.event["total"]["extracted"] += 1
                    except strelka.ScannerTimeout:
                        raise
                    except Exception:
                        self.flags.append("zeek_file_upload_error")
        except Exception:
            self.flags.append("zeek_file_log_parse_error")

    def suricata(self, pcap_file: str) -> None:
        """Run Suricata on pcap file to generate alerts.

        Args:
            pcap_file: Path to the pcap file.
        """
        self.event["suricata"] = {}

        with tempfile.TemporaryDirectory() as tmp_suricata:
            try:
                self._scan_with_suricata(pcap_file, log_dir=tmp_suricata)

                # Paths to log files
                eve_log_file = os.path.join(tmp_suricata, "eve.json")
                suricata_log_file = os.path.join(tmp_suricata, "suricata.log")

                # Get matching alerts
                if os.path.exists(eve_log_file):
                    alerts = self._get_matching_alerts(eve_log_file)
                else:
                    alerts = []

                if len(alerts) == 0:
                    self.flags.append("suricata_no_alerts")

                if self.suricata_alert_dedupe:
                    alerts = self._deduplicate_alerts(alerts)

                self.event["suricata"]["alerts"] = alerts

                # Parse Suricata log file for additional statistics
                if os.path.exists(suricata_log_file):
                    self._parse_suricata_log(suricata_log_file)

            except strelka.ScannerTimeout:
                raise
            except Exception:
                self.flags.append("suricata_error")
                raise

    def _parse_suricata_log(self, log_file: str) -> None:
        """Parse Suricata log file to extract statistics.

        Extracts information about rules loaded and pcap processing stats.

        Args:
            log_file: Path to the Suricata log file.
        """
        try:
            with open(log_file, "r") as file:
                log_content = file.read()

                # Extract rule loading statistics
                rule_pattern = r"<Info> - (\d+) rule files processed\. (\d+) rules successfully loaded, (\d+) rules failed"
                rule_match = re.search(rule_pattern, log_content)

                if rule_match:
                    self.event["suricata"]["rules_stats"] = {
                        "rules_loaded": int(rule_match.group(2)),
                        "rules_failed": int(rule_match.group(3)),
                    }

                # Extract pcap processing statistics
                pcap_pattern = r"<Notice> - Pcap-file module read (\d+) files, (\d+) packets, (\d+) bytes"
                pcap_match = re.search(pcap_pattern, log_content)

                if pcap_match:
                    self.event["suricata"]["pcap_stats"] = {
                        "packets_read": int(pcap_match.group(2)),
                        "bytes_read": int(pcap_match.group(3)),
                    }

        except Exception:
            self.flags.append("suricata_log_parse_error")
            raise

    def _scan_with_suricata(self, file_path: str, log_dir: str) -> None:
        """Run Suricata on the pcap file.

        Args:
            file_path: Path to the pcap file.
            log_dir: Directory to store Suricata logs.
        """
        # Build the command
        cmd: List[str] = ["suricata", "-r", file_path, "-l", log_dir]

        # Add custom config if provided
        if self.suricata_config:
            cmd.extend(["-c", self.suricata_config])
        else:
            # Default Suricata config options
            cmd.extend(
                [
                    "--set",
                    "outputs.1.eve-log.types.1.alert.flow=false",
                    "--set",
                    "port-groups.HTTP_PORTS=[80,8080]",
                    "--set",
                    "unix-command.enabled=false",
                ]
            )

        # Add custom rules if provided
        if self.suricata_rules:
            cmd.extend(["-S", self.suricata_rules])

        # Run Suricata
        result = subprocess.run(cmd, capture_output=True, text=True)

        if result.returncode != 0:
            self.flags.append("suricata_process_error")

    def _get_matching_alerts(self, log_file: str) -> List[Dict[str, Any]]:
        """Parse Suricata log file and extract alerts.

        Args:
            log_file: Path to the Suricata log file.

        Returns:
            List of alert dictionaries.
        """
        matching_alerts: List[Dict[str, Any]] = []
        with open(log_file, "r") as file:
            for line in file:
                log_entry = json.loads(line)
                if log_entry.get("event_type") == "alert":
                    matching_alerts.append(log_entry)
        return matching_alerts

    def _deduplicate_alerts(self, alerts: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        """Deduplicate alerts based on signature ID.

        Args:
            alerts: List of alert dictionaries.

        Returns:
            Deduplicated list of alert dictionaries.
        """
        deduplicated_alerts: List[Dict[str, Any]] = []
        seen_signature_ids: Set[int] = set()
        for alert in alerts:
            signature_id = alert["alert"]["signature_id"]
            if signature_id not in seen_signature_ids:
                deduplicated_alerts.append(alert)
                seen_signature_ids.add(signature_id)
        return deduplicated_alerts

    def upload(self, name: str, expire_at: int) -> None:
        """Send extracted file to coordinator.

        Args:
            name: Path to the file to upload.
            expire_at: Expiration timestamp.
        """
        with open(name, "rb") as extracted_file:
            # Send extracted file back to Strelka
            self.emit_file(extracted_file.read())

scan(data, file, options, expire_at)

Process pcap/pcapng data and extract files using Zeek and optionally run Suricata.

Parameters:

Name	Type	Description	Default
data	bytes	Raw pcap/pcapng data.	required
file	Dict[str, Any]	File metadata.	required
options	Dict[str, Any]	Scanner options.	required
expire_at	int	Expiration timestamp.	required

Source code in strelka/src/python/strelka/scanners/scan_pcap.py

def scan(
    self, data: bytes, file: Dict[str, Any], options: Dict[str, Any], expire_at: int
) -> None:
    """Process pcap/pcapng data and extract files using Zeek and optionally run Suricata.

    Args:
        data: Raw pcap/pcapng data.
        file: File metadata.
        options: Scanner options.
        expire_at: Expiration timestamp.
    """
    self.expire_at = expire_at
    self.file_limit: int = options.get("limit", 1000)
    self.tmp_directory: str = options.get("tmp_file_directory", "/tmp/")
    self.zeek_conn: Optional[bool] = options.get("zeek_conn", True)
    self.zeek_conn_limit: Optional[int] = options.get("zeek_conn_limit", 100)
    self.zeek_proto: Optional[bool] = options.get("zeek_proto", True)
    self.suricata_config: Optional[str] = options.get("suricata_config", None)
    self.suricata_rules: Optional[str] = options.get("suricata_rules", None)
    self.suricata_alert_dedupe: bool = options.get("suricata_alert_dedupe", False)
    self.scanner_timeout: int = options.get("scanner_timeout", 120)

    self.event["total"] = {"files": 0, "extracted": 0}
    self.event["files"] = []

    try:
        # Check if zeek package is installed
        if not shutil.which("zeek"):
            self.flags.append("zeek_not_installed_error")
            return
    except Exception as e:
        self.flags.append(str(e))
        return

    with tempfile.NamedTemporaryFile(dir=self.tmp_directory, mode="wb") as tmp_data:
        tmp_data.write(data)
        tmp_data.flush()
        tmp_data.seek(0)

        self.zeek_extract(tmp_data.name)

        if self.suricata_rules:
            try:
                # Check if suricata package is installed
                if not shutil.which("suricata"):
                    self.flags.append("suricata_not_installed_error")
                    return

                self.suricata(tmp_data.name)
            except Exception as e:
                self.flags.append(str(e))

zeek_extract(pcap_path)

Extract files from pcap using Zeek.

Parameters:

Name	Type	Description	Default
pcap_path	str	Path to the pcap file.	required

Source code in strelka/src/python/strelka/scanners/scan_pcap.py

def zeek_extract(self, pcap_path: str) -> None:
    """Extract files from pcap using Zeek.

    Args:
        pcap_path: Path to the pcap file.
    """
    with tempfile.TemporaryDirectory() as tmp_zeek:
        try:

            # Zeek outputs to the current working directory by default
            os.chdir(tmp_zeek)

            stdout, stderr = subprocess.Popen(
                [
                    "zeek",
                    "-r",
                    pcap_path,
                    "/opt/zeek/share/zeek/policy/frameworks/files/extract-all-files.zeek",
                    f"FileExtract::prefix={os.path.join(tmp_zeek, 'files')}",
                    "LogAscii::use_json=T",
                ],
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
            ).communicate(timeout=self.scanner_timeout)

            files_log_path = os.path.join(tmp_zeek, "files.log")

            if os.path.exists(files_log_path):
                self._process_files_log(
                    files_log_path, os.path.join(tmp_zeek, "files")
                )
            else:
                self.flags.append("zeek_no_file_log")

            if self.zeek_conn:
                # Parse the conn.log file to get basic connection information
                connections = []
                total_connections = 0
                conn_log = os.path.join(tmp_zeek, "conn.log")

                if os.path.exists(conn_log):
                    with open(conn_log, "r") as f:
                        for line in f:
                            if line.startswith("#"):
                                continue

                            total_connections += 1

                            if (
                                self.zeek_conn_limit > 0
                                and len(connections) >= self.zeek_conn_limit
                            ):
                                if "zeek_conn_limit_reached" not in self.flags:
                                    self.flags.append("zeek_conn_limit_reached")
                                continue

                            try:
                                conn_data = json.loads(line)
                                connection = {
                                    "uid": conn_data.get("uid"),
                                    "source_ip": conn_data.get("id.orig_h"),
                                    "source_port": conn_data.get("id.orig_p"),
                                    "dest_ip": conn_data.get("id.resp_h"),
                                    "dest_port": conn_data.get("id.resp_p"),
                                    "protocol": conn_data.get("proto"),
                                    "service": conn_data.get("service"),
                                    "duration": conn_data.get("duration"),
                                    "bytes_orig": conn_data.get("orig_bytes"),
                                    "bytes_resp": conn_data.get("resp_bytes"),
                                    "conn_state": conn_data.get("conn_state"),
                                }

                                connections.append(connection)
                            except json.JSONDecodeError:
                                continue

                if self.zeek_proto:
                    # Check for protocol-specific logs to enhance connection details
                    protocol_logs = [
                        "http.log",
                        "dns.log",
                        "ssl.log",
                        "ssh.log",
                        "smtp.log",
                        "ftp.log",
                        "rdp.log",
                        "sip.log",
                        "smb_mapping.log",
                    ]

                    # Map to store protocol details by connection UID
                    protocol_details = defaultdict(dict)

                    for log_name in protocol_logs:
                        log_path = os.path.join(tmp_zeek, log_name)
                        if os.path.exists(log_path):
                            protocol = log_name.split(".")[0]

                            with open(log_path, "r") as f:
                                for line in f:
                                    if line.startswith("#"):
                                        continue
                                    try:
                                        log_data = json.loads(line)
                                        uid = log_data.get("uid")
                                        if uid:
                                            # Extract relevant details based on protocol
                                            if protocol == "http":
                                                protocol_details[uid][protocol] = {
                                                    "method": log_data.get(
                                                        "method"
                                                    ),
                                                    "host": log_data.get("host"),
                                                    "uri": log_data.get("uri"),
                                                    "user_agent": log_data.get(
                                                        "user_agent"
                                                    ),
                                                    "status_code": log_data.get(
                                                        "status_code"
                                                    ),
                                                }
                                            elif protocol == "dns":
                                                protocol_details[uid][protocol] = {
                                                    "query": log_data.get("query"),
                                                    "qtype": log_data.get(
                                                        "qtype_name"
                                                    ),
                                                    "answers": log_data.get(
                                                        "answers"
                                                    ),
                                                }
                                            elif protocol == "ssl":
                                                protocol_details[uid][protocol] = {
                                                    "server_name": log_data.get(
                                                        "server_name"
                                                    ),
                                                    "subject": log_data.get(
                                                        "subject"
                                                    ),
                                                    "issuer": log_data.get(
                                                        "issuer"
                                                    ),
                                                    "version": log_data.get(
                                                        "version"
                                                    ),
                                                }
                                            else:
                                                # Generic extraction for other protocols
                                                protocol_details[uid][protocol] = {
                                                    k: v
                                                    for k, v in log_data.items()
                                                    if k != "uid" and v is not None
                                                }
                                    except json.JSONDecodeError:
                                        continue

                    # Merge protocol details with the basic connection information
                    for connection in connections:
                        uid = connection.get("uid")
                        if uid in protocol_details:
                            connection["protocol_details"] = protocol_details[uid]

                self.event["total"]["connections"] = total_connections
                self.event["connections"] = connections

        except strelka.ScannerTimeout:
            raise
        except Exception:
            self.flags.append("zeek_extract_process_error")
            raise

suricata(pcap_file)

Run Suricata on pcap file to generate alerts.

Parameters:

Name	Type	Description	Default
pcap_file	str	Path to the pcap file.	required

Source code in strelka/src/python/strelka/scanners/scan_pcap.py

def suricata(self, pcap_file: str) -> None:
    """Run Suricata on pcap file to generate alerts.

    Args:
        pcap_file: Path to the pcap file.
    """
    self.event["suricata"] = {}

    with tempfile.TemporaryDirectory() as tmp_suricata:
        try:
            self._scan_with_suricata(pcap_file, log_dir=tmp_suricata)

            # Paths to log files
            eve_log_file = os.path.join(tmp_suricata, "eve.json")
            suricata_log_file = os.path.join(tmp_suricata, "suricata.log")

            # Get matching alerts
            if os.path.exists(eve_log_file):
                alerts = self._get_matching_alerts(eve_log_file)
            else:
                alerts = []

            if len(alerts) == 0:
                self.flags.append("suricata_no_alerts")

            if self.suricata_alert_dedupe:
                alerts = self._deduplicate_alerts(alerts)

            self.event["suricata"]["alerts"] = alerts

            # Parse Suricata log file for additional statistics
            if os.path.exists(suricata_log_file):
                self._parse_suricata_log(suricata_log_file)

        except strelka.ScannerTimeout:
            raise
        except Exception:
            self.flags.append("suricata_error")
            raise

upload(name, expire_at)

Send extracted file to coordinator.

Parameters:

Name	Type	Description	Default
name	str	Path to the file to upload.	required
expire_at	int	Expiration timestamp.	required

Source code in strelka/src/python/strelka/scanners/scan_pcap.py

def upload(self, name: str, expire_at: int) -> None:
    """Send extracted file to coordinator.

    Args:
        name: Path to the file to upload.
        expire_at: Expiration timestamp.
    """
    with open(name, "rb") as extracted_file:
        # Send extracted file back to Strelka
        self.emit_file(extracted_file.read())

Features

The features of this scanner are detailed below. These features represent the capabilities and the type of analysis the scanner can perform. This may include support for Indicators of Compromise (IOC), the ability to emit files for further analysis, and the presence of extended documentation for complex analysis techniques.

Feature	Support
IOC Support
Emit Files
Extended Docs
Malware Scanner
Image Thumbnails

Tastes

Strelka's file distribution system assigns scanners to files based on 'flavors' and 'tastes'. Flavors describe the type of file, typically determined by MIME types from libmagic, matches from YARA rules, or characteristics of parent files. Tastes are the criteria used within Strelka to determine which scanners are applied to which files, with positive and negative tastes defining files to be included or excluded respectively.

Source Filetype	Include / Exclude
pcap_file
pcapng_file

Scanner Fields

This section provides a list of fields that are extracted from the files processed by this scanner. These fields include the data elements that the scanner extracts from each file, representing the analytical results produced by the scanner. If the test file is missing or cannot be parsed, this section will not contain any data.

Field Name	Field Type
connections	list
connections.bytes_orig	int
connections.bytes_resp	int
connections.conn_state	str
connections.dest_ip	str
connections.dest_port	int
connections.duration	float
connections.protocol	str
connections.protocol_details	dict
connections.protocol_details.http	dict
connections.protocol_details.http.host	str
connections.protocol_details.http.method	str
connections.protocol_details.http.status_code	int
connections.protocol_details.http.uri	str
connections.protocol_details.http.user_agent	str
connections.service	str
connections.source_ip	str
connections.source_port	int
connections.uid	str
elapsed	str
files	list
files.analyzers	str
files.depth	int
files.duration	str
files.duration	float
files.extracted	str
files.extracted_cutoff	bool
files.fuid	str
files.id.orig_h	str
files.id.orig_p	int
files.id.resp_h	str
files.id.resp_p	int
files.is_orig	bool
files.local_orig	bool
files.mime_type	str
files.missing_bytes	int
files.overflow_bytes	int
files.seen_bytes	int
files.source	str
files.timedout	bool
files.total_bytes	int
files.ts	float
files.uid	str
flags	list
suricata	dict
suricata.alerts	str
suricata.pcap_stats	dict
suricata.pcap_stats.bytes_read	int
suricata.pcap_stats.packets_read	int
suricata.rules_stats	dict
suricata.rules_stats.rules_failed	int
suricata.rules_stats.rules_loaded	int
total	dict
total.connections	int
total.extracted	int
total.files	int

Sample Event

Below is a sample event generated by this scanner, demonstrating the kind of output that can be expected when it processes a file. This sample is derived from a mock scan event configured in the scanner's test file. If no test file is available, this section will not display a sample event.

    test_scan_event = {
        "elapsed": 0.001,
        "flags": ["zeek_conn_limit_reached"],
        "total": {"connections": 3, "files": 3, "extracted": 3},
        "files": [
            {
                "analyzers": unordered(["PE", "EXTRACT"]),
                "depth": 0,
                "duration": 0.00018906593322753906,
                "extracted": "extract-1673576655.41892-HTTP-FOxTJwn9u5H1hBXn1",
                "extracted_cutoff": False,
                "fuid": "FOxTJwn9u5H1hBXn1",
                "id.orig_h": "192.168.174.1",
                "id.orig_p": 13147,
                "id.resp_h": "192.168.174.131",
                "id.resp_p": 8080,
                "is_orig": False,
                "local_orig": True,
                "mime_type": "application/x-dosexec",
                "missing_bytes": 0,
                "overflow_bytes": 0,
                "seen_bytes": 4096,
                "source": "HTTP",
                "timedout": False,
                "total_bytes": 4096,
                "ts": 1673576655.41892,
                "uid": 0.001,
            },
            {
                "analyzers": unordered(["EXTRACT"]),
                "depth": 0,
                "duration": 0.007551908493041992,
                "extracted": "extract-1673576666.163778-HTTP-FxYAi61ktBsEM4hpNd",
                "extracted_cutoff": False,
                "fuid": "FxYAi61ktBsEM4hpNd",
                "id.orig_h": "192.168.174.1",
                "id.orig_p": 13162,
                "id.resp_h": "192.168.174.131",
                "id.resp_p": 8080,
                "is_orig": False,
                "local_orig": True,
                "mime_type": "image/jpeg",
                "missing_bytes": 0,
                "overflow_bytes": 0,
                "seen_bytes": 308566,
                "source": "HTTP",
                "timedout": False,
                "total_bytes": 308566,
                "ts": 1673576666.163778,
                "uid": 0.001,
            },
            {
                "analyzers": unordered(["EXTRACT"]),
                "depth": 0,
                "duration": 0.0,
                "extracted": "extract-1673576677.801391-HTTP-FoNGFk1uRR9pVo9XKi",
                "extracted_cutoff": False,
                "fuid": "FoNGFk1uRR9pVo9XKi",
                "id.orig_h": "192.168.174.1",
                "id.orig_p": 13176,
                "id.resp_h": "192.168.174.131",
                "id.resp_p": 8080,
                "is_orig": False,
                "local_orig": True,
                "mime_type": "application/xml",
                "missing_bytes": 0,
                "overflow_bytes": 0,
                "seen_bytes": 620,
                "source": "HTTP",
                "timedout": False,
                "total_bytes": 620,
                "ts": 1673576677.801391,
                "uid": 0.001,
            },
        ],
        "suricata": {
            "pcap_stats": {"bytes_read": 334443, "packets_read": 352},
            "rules_stats": {"rules_failed": 0, "rules_loaded": 1},
            "alerts": unordered(
                [
                    {
                        "alert": {
                            "action": "allowed",
                            "category": "",
                            "gid": 1,
                            "rev": 1,
                            "severity": 3,
                            "signature": "HTTP Request Detected",
                            "signature_id": 1000001,
                        },
                        "dest_ip": "192.168.174.131",
                        "dest_port": 8080,
                        "event_type": "alert",
                        "flow_id": 0.001,
                        "pcap_cnt": 4,
                        "proto": "6",
                        "src_ip": "192.168.174.1",
                        "src_port": 13147,
                        "timestamp": "2023-01-13T02:24:15.414361+0000",
                    },
                    {
                        "alert": {
                            "action": "allowed",
                            "category": "",
                            "gid": 1,
                            "rev": 1,
                            "severity": 3,
                            "signature": "HTTP Request Detected",
                            "signature_id": 1000001,
                        },
                        "dest_ip": "192.168.174.131",
                        "dest_port": 8080,
                        "event_type": "alert",
                        "flow_id": 0.001,
                        "pcap_cnt": 19,
                        "proto": "6",
                        "src_ip": "192.168.174.1",
                        "src_port": 13162,
                        "timestamp": "2023-01-13T02:24:26.158751+0000",
                    },
                    {
                        "alert": {
                            "action": "allowed",
                            "category": "",
                            "gid": 1,
                            "rev": 1,
                            "severity": 3,
                            "signature": "HTTP Request Detected",
                            "signature_id": 1000001,
                        },
                        "app_proto": "http",
                        "dest_ip": "192.168.174.131",
                        "dest_port": 8080,
                        "event_type": "alert",
                        "flow_id": 0.001,
                        "pcap_cnt": 8,
                        "proto": "6",
                        "src_ip": "192.168.174.1",
                        "src_port": 13147,
                        "timestamp": "2023-01-13T02:24:15.418972+0000",
                    },
                    {
                        "alert": {
                            "action": "allowed",
                            "category": "",
                            "gid": 1,
                            "rev": 1,
                            "severity": 3,
                            "signature": "HTTP Request Detected",
                            "signature_id": 1000001,
                        },
                        "app_proto": "http",
                        "dest_ip": "192.168.174.131",
                        "dest_port": 8080,
                        "event_type": "alert",
                        "flow_id": 0.001,
                        "pcap_cnt": 23,
                        "proto": "6",
                        "src_ip": "192.168.174.1",
                        "src_port": 13162,
                        "timestamp": "2023-01-13T02:24:26.163830+0000",
                    },
                    {
                        "alert": {
                            "action": "allowed",
                            "category": "",
                            "gid": 1,
                            "rev": 1,
                            "severity": 3,
                            "signature": "HTTP Request Detected",
                            "signature_id": 1000001,
                        },
                        "dest_ip": "192.168.174.131",
                        "dest_port": 8080,
                        "event_type": "alert",
                        "flow_id": 0.001,
                        "pcap_cnt": 346,
                        "proto": "6",
                        "src_ip": "192.168.174.1",
                        "src_port": 13176,
                        "timestamp": "2023-01-13T02:24:37.797451+0000",
                    },
                    {
                        "alert": {
                            "action": "allowed",
                            "category": "",
                            "gid": 1,
                            "rev": 1,
                            "severity": 3,
                            "signature": "HTTP Request Detected",
                            "signature_id": 1000001,
                        },
                        "app_proto": "http",
                        "dest_ip": "192.168.174.131",
                        "dest_port": 8080,
                        "event_type": "alert",
                        "flow_id": 0.001,
                        "pcap_cnt": 350,
                        "proto": "6",
                        "src_ip": "192.168.174.1",
                        "src_port": 13176,
                        "timestamp": "2023-01-13T02:24:37.801425+0000",
                    },
                ]
            ),
        },
        "connections": [
            {
                "bytes_orig": 482,
                "bytes_resp": 4301,
                "conn_state": "SF",
                "dest_ip": "192.168.174.131",
                "dest_port": 8080,
                "duration": 0.005362987518310547,
                "protocol": "tcp",
                "protocol_details": {
                    "http": {
                        "host": "192.168.174.131:8080",
                        "method": "GET",
                        "status_code": 200,
                        "uri": "/test.exe",
                        "user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36",
                    }