8.3. PEAT API

The general API provides a generalized interface to execute the PEAT “verbs”: scan, parse, pull, push, pillage, and heat. The purpose of this API is to “wrap” the device module API and provide a consistent and well-tested set of powerful interfaces into PEAT’s core functionality and device modules. It is used for implementing the PEAT CLI, as well as integration with other SNL-developed capabilities.

8.3.1. High-level API

scan(scan_targets, scan_type, device_types=None)[source]

Scan IP networks and hosts and/or serial ports for OT devices.

Parameters:
  • scan_targets (list[str]) -- What to scan, such as network hosts or serial ports

  • scan_type (Literal['unicast_ip', 'broadcast_ip', 'serial']) --

    Communication type of the targets. Allowed values:

    • unicast_ip

    • broadcast_ip

    • serial

  • device_types (list[str | type[DeviceModule]] | None) -- mixed device type strings, alias strings or DeviceModule classes to scan using. If None, all currently imported DeviceModule modules are used.

Return type:

dict[str, dict | list | str | float | int] | None

Returns:

Scan summary as a dict, or None if an error occurred

pull(targets, comm_type, device_types)[source]

Pull from devices.

Parameters:
  • targets (list[str]) -- Devices to pull from, such as network hosts or serial ports

  • comm_type (Literal['unicast_ip', 'broadcast_ip', 'serial']) --

    Method of communication for the pull. Allowed values:

    • unicast_ip

    • broadcast_ip

    • serial

  • device_types (list[str]) -- Names of device modules or module aliases to use

Return type:

dict[str, dict | list | str | float | int] | None

Returns:

Pull summary as a dict, or None if an error occurred

push(targets, comm_type, device_types, input_source, push_type, skip_scan=False)[source]

Push (upload) configuration or firmware to devices.

Note

By default all targets are scanned and verified, and only the devices that are successfully verified are pushed to. This enables this function to be used without requiring a scan to be done and ensuring pushes are not performed to invalid devices. This behavior can be disabled by setting PUSH_SKIP_SCAN to True or passing --push-skip-scan on the CLI.

Parameters:
  • targets (list[str]) -- Devices to push to, such as network hosts or serial ports

  • comm_type (Literal['unicast_ip', 'broadcast_ip', 'serial']) --

    Method of communication for the push. Allowed values:

    • unicast_ip

    • broadcast_ip

    • serial

  • device_types (list[str]) -- Names of device modules or module aliases to use. If scanning is disabled this should be a single device type.

  • input_source (Path | str) -- Path of the file to push, as a string

  • push_type (Literal['config', 'firmware']) -- Type of push being performed. Valid push types are “config” and “firmware”.

  • skip_scan (bool) -- If device verification (scanning) should be skipped. NOTE: this currently only applies to unicast_ip devices.

Return type:

bool

Returns:

If the push was successful

Raises:

PeatError -- If the push failed due to an issue with configuration or arguments, such as invalid device types or push type.

parse(filepaths, device_types=None, sub_dirs=True)[source]

Find and parse device and/or project files.

Parameters:
  • filepaths (str | Path | list[str | Path]) -- File or directory paths to parse, or "-" to read from standard input (stdin).

  • device_types (list[type[DeviceModule] | str] | None) -- names, aliases, or classes of PEAT device modules to use. If None, all currently imported modules are used.

  • sub_dirs (bool) -- If sub-directories of a directory path should be searched

Return type:

dict[str, dict | list | str | float | int] | None

Returns:

Pull summary as a dict, or None if an error occurred

pillage(source)[source]

peat pillage.

Parameters:
  • config_file -- Path to pillage config file

  • source (str) -- Path of directory to pillage

Return type:

bool

Returns:

If pillaging was successful

8.3.2. Configuration API

Configuration options that are accessible from anywhere in PEAT. These are currently resident in peat.settings.config variable, which is a singleton instance of the Configuration class. They can be accessed with from peat import config.

Refer to Operate for details on the available configuration options and how to set them, and the Configuration and state deep dive in the PEAT developer documentation for a deep dive into how the system works internally.

class Configuration(label, env_prefix, init_env=True)[source]

Global configuration settings used throughout PEAT.

METADATA: dict = {'author': '', 'created': '', 'description': '', 'name': '', 'updated': ''}

Metadata for the PEAT YAML configuration file.

ADDITIONAL_MODULES: list[str | Path] = []

File paths to external/third-party PEAT device modules to import.

These can be the path to a .py file with a DeviceModule subclass or a path to a folder with an __init__.py file and any number of .py files containing DeviceModule subclasses to import.

DEBUG: int = 0

DEBUG level (higher = more output, 0 = disabled).

VERBOSE: bool = False

Include DEBUG messages in the CLI terminal output.

QUIET: bool = False

Don’t write log messages to the CLI terminal.

NO_PRINT_RESULTS: bool = False

DEPRECATED. Setting this no longer has any effect, as it’s now the default behavior.

PRINT_RESULTS: bool = False

Print JSON-formatted results from the operation to the terminal (stdout). Note that log messages will still be printed unless ‘--quiet’ is specified.

NO_COLOR: bool = False

Don’t color log messages in the CLI terminal.

Don’t print the PEAT logo at startup in the CLI terminal.

DRY_RUN: bool = False

“Dry run” when running on the CLI, do everything except running commands or connecting to servers. Actions won’t be executed, but logs and state will still be written to files and saved to Elasticsearch (if enabled). This includes loading and verifying configuration and importing modules

Useful for verifying a YAML config file or other settings are correct, or that a third-party PEAT module is imported correctly.

ASSUME_ONLINE: bool = False

Skip the host online check before scan/pull/push.

MAX_THREADS: int = 260

Maximum number of threads for any concurrent operations (scanning, etc.).

DEFAULT_TIMEOUT: float = 5.0

Default timeout for sockets and potentially other things.

RESOLVE_IP: bool = True

If PEAT should attempt to resolve device IP address from a hostname or MAC address.

RESOLVE_MAC: bool = True

If PEAT should attempt to resolve device MAC address from it’s IP address.

RESOLVE_HOSTNAME: bool = True

If PEAT should attempt to resolve device hostname from it’s IP address.

FORCE_ONLINE_METHOD_PING: bool = False

Force ARP and ICMP requests to be used to check if a host is online, even if the system running PEAT isn’t able to use them.

FORCE_ONLINE_METHOD_TCP: bool = False

Force TCP SYNs to be used to check if a host is online, even if the system running PEAT is able to use ARP or ICMP to perform the checks.

ICMP_FALLBACK_TCP_SYN: bool = True

In the case of a ICMP failure, fallback to attempting a TCP SYN RST to check if the host is online. If false, then ICMP failures will result in the host being marked as down, even if they’re blocked by a firewall or gateway.

SYN_PORT: int = 80

Default port used for basic TCP SYN online checks. The default HTTP (web) port 80 is generally safe to check.

PUSH_SKIP_SCAN: bool = False

Skip scanning and verification of hosts being pushed to and assume all hosts are online and valid devices.

Note

This requires a single device type to be specified (the -d argument on the CLI)

SCAN_SWEEP: bool = False

Simple host up/down check (equivalent to nmap -Pn <hosts>).

If serial ports are targeted, this will enumerate the active serial ports on the host.

INTENSIVE_SCAN: bool = False

Force identification checks of all ports during scanning.

HASH_ALGORITHMS: list[str] = ['md5', 'sha1', 'sha256', 'sha512']

Hash algorithms to use wherever hashes are calculated. Available algorithms are any provided by hashlib.

OUT_DIR: Path

Default directory for all file output.

RUN_DIR: Path

Output directory for all files associated with a single run of PEAT.

Name format: <command>_<config-name>_<timestamp>_<run-id>

  • <command> : PEAT command, e.g. “scan”, “pull”, “parse”, etc.

  • <config-name>name of YAML config file, set in metadata: name: "name".

    If no config name is specified, then the string default-config is used.

  • <timestamp>start time of the PEAT run, e.g. 2022-06-15_13-08-59.

    This value is retrieved from consts.START_TIME.

  • <run-id>Run ID, aka agent.id, e.g. 165532013980.

    This value is retrieved from consts.RUN_ID.

Examples:

  • pull_sceptre-test-config_2022-06-17_165532013980

  • scan_default-config_2022-09-27_165532013980

DEVICE_DIR: Path

Output directory for files generated by PEAT device modules, e.g. pulled config files, firmware images, etc. This is the most relevant directory for the majority of PEAT users.

Note

Device file output can be disabled entirely by setting this to a empty string ("") or None. This can be useful for low-footprint use cases.

ELASTIC_DIR: Path

Directory where raw documents pushed to Elasticsearch are saved in JSON format. These can be used to rebuild the Elasticsearch indices if needed.

Note

Elastic doc saving can be disabled entirely by setting this to a empty string ("") or None.

LOG_DIR: Path

Directory for PEAT’s log files. Setting this to an empty string will disable output of these files.

JSON formatted log files are also saved here. These can be used to rebuild the vedar-logs Elasticsearch index.

Files that may be created in this directory include:

  • enip/* (Rockwell-specific)

  • peat.log

  • json-log.jsonl

  • elasticsearch.log

  • telnet.log

META_DIR: Path

Directory for PEAT’s run metadata, notably dumps of PEAT’s configuration and internal state.

Setting this to an empty string will disable output of these files.

Note

Log files are stored separately, in logs/

Files that may be created in this directory include:

  • peat_configuration.yaml

  • peat_configuration.json

  • peat_state.json

  • Copy of the config file used for the run,

    if applicable (e.g. -c my-config.yaml)

SUMMARIES_DIR: Path

Directory for summary results for PEAT commands.

Files that may be created in this directory include:

  • scan-summary.json

  • pull-summary.json

  • parse-summary.json

TEMP_DIR: Path

Working directory for file artifacts. May be deleted when PEAT exits.

ZEEK_LOGDIR: Path = PosixPath('/home/runner/work/PEAT/PEAT/peat_results/default-run-dir_2026-02-07_05-32-29_177044234927/zeek_logs')

Directory where Zeek logs will be saved. Also contains Zeek artifacts.

ELASTIC_SERVER: str = None

URL of the Elasticsearch server.

ELASTIC_SAVE_LOGS: bool = True

If PEAT logs should be sent to Elasticsearch.

ELASTIC_SAVE_CONFIG: bool = True

If PEAT Configuration dumps should be sent to Elasticsearch on exit.

ELASTIC_SAVE_STATE: bool = True

If PEAT State dumps should be sent to Elasticsearch on exit.

ELASTIC_SAVE_BLOBS: bool = False

If large binary blobs should be stored in Elasticsearch (e.g. firmware images).

ELASTIC_TIMEOUT: float = 10.0

Timeout to connect to the Elasticsearch server.

ELASTIC_DISABLE_DATED_INDICES: bool = False

Append the current date to Elasticsearch index names.

Example: when false, a push to peat-configs on Feb 25 2021 will actually push to an index named peat-configs-2021.02.25. When this option disabled (set to True), the push will instead go to peat-configs (as above, without the date).

ELASTIC_ADDITIONAL_TAGS: list[str] = []

Additional tag strings to add to the tags field for each document pushed to Elasticsearch.

ELASTIC_LOG_INDEX: str = 'vedar-logs'

Base name of Elasticsearch index to use PEAT logging events.

ELASTIC_SCAN_INDEX: str = 'peat-scan-summaries'

Base name of Elasticsearch index to use for scan result summaries, e.g. what normally gets written in peat_results/<run-dir>/summaries/.

ELASTIC_PULL_INDEX: str = 'peat-pull-summaries'

Base name of Elasticsearch index to use for pull result summaries, e.g. what normally gets written in peat_results/<run-dir>/summaries/.

ELASTIC_PARSE_INDEX: str = 'peat-parse-summaries'

Base name of Elasticsearch index to use for parse result summaries, e.g. what normally gets written in peat_results/<run-dir>/summaries/.

ELASTIC_CONFIG_INDEX: str = 'peat-configs'

Base name of Elasticsearch index to use for PEAT configuration dumps from runs of PEAT, e.g. what normally gets written in peat_results/<run-dir>/metadata/.

ELASTIC_STATE_INDEX: str = 'peat-state'

Base name of Elasticsearch index to use for PEAT state dumps from runs of PEAT, e.g. what normally gets written in peat_results/<run-dir>/metadata/.

ELASTIC_HOSTS_INDEX: str = 'ot-device-hosts-timeseries'

Base name of Elasticsearch index to use for DeviceData data in ‘timeseries’ form (new entries are created for every collection performed or data parsed).

Information collected by PEAT from field devices or parsed files. A new Elasticsearch document is created for every pull of datafrom a device (the data is ‘timeseries’, with differences visible between pulls over time).

ELASTIC_FILES_INDEX: str = 'ot-device-files'

Base name of Elasticsearch index to use for File data.

Information about files present on the device, or that were present on the device at one point in time.

ELASTIC_REGISTERS_INDEX: str = 'ot-device-registers'

Base name of Elasticsearch index to use for Register data.

Information about individual communication ‘registers’ (e.g. Modbus registers/coils, DNP3 data points, BACNet objects, etc.) that are configured on devices, as extracted from device configuration information.

ELASTIC_TAGS_INDEX: str = 'ot-device-tags'

Base name of Elasticsearch index to use for Tag data.

Information about tag variables that are configured on devices, as extracted from device configuration information.

ELASTIC_IO_INDEX: str = 'ot-device-io'

Base name of Elasticsearch index to use for IO data.

Information about I/O (Input/Output) available and/or configured on a device, as extracted from device configuration information.

ELASTIC_EVENTS_INDEX: str = 'ot-device-events'

Base name of Elasticsearch index to use for Event data from devices.

Logging and other event history as extracted from devices. Examples include access logs, system logs, or protection history

ELASTIC_MEMORY_INDEX: str = 'ot-device-memory'

Base name of Elasticsearch index to use for Memory data from devices.

Memory reads from devices, including address in memory, the value read, and information about where it came from and when the read occurred.

ELASTIC_UEFI_FILES_INDEX: str = 'uefi-files'

Base name of elasticsearch index to use for peat.data.models.UEFIFiles

Information is parsed from a file

ELASTIC_UEFI_HASHES_INDEX: str = 'uefi-hashes'

Base name of elasticsearch index to use for peat.data.models.UEFIHash

HEAT_ELASTIC_SERVER: str = None

Elasticsearch server to pull HEAT data from.

HEAT_INDEX_NAMES: str = 'packets-*'

Names and/or patterns of Elasticsearch indices with Packetbeat data for HEAT.

HEAT_DATE_RANGE: str = None

Date range to filter HEAT extraction to.

HEAT_EXCLUDE_IPS: list[str] = []

IP addresses to exclude from packetbeat search (source and/or destination IP).

HEAT_ONLY_IPS: list[str] = []

IP address to limit packetbeat search to (source or destination IP).

HEAT_FILE_ONLY: bool = False

Extract the file(s) but don’t parse them using PEAT.

HEAT_ARTIFACTS_DIR: Path

Directory where HEAT artifacts should be saved.

PILLAGE: dict = {}

Configuration for peat pillage (Pillage).

DEVICE_OPTIONS: dict = {}

Protocol and module configuration options that apply to all hosts.

HOSTS: list[dict] = []

Specify or override options for specific hosts.

PCAPS: Path | None

Specify folder that contains PCAPS for processing by HEAT.

HEAT_PROTOCOLS: list = []

Specify protocols for HEAT to use.

NO_RUN_ZEEK: bool = False

Don’t have PEAT run Zeek on a PCAP, instead run it on a PCAP file, then process the output of the PCAP.

NOTE: zeek_dir must be specified if no_run_zeek is true.

ZEEK_DIR: Path | None

Directory with existing Zeek output to use as input.

This is an alternative to PEAT running Zeek on the PCAPs itself.

NOTE: This argument is required if no_run_zeek is true.

8.3.3. Constants

Constants that are determined at program start, such as platform information, timezone, time of start, and time format.

Variable

Type

Description

WINDOWS

bool

If running on Windows (not in WSL)

POSIX

bool

If running on a POSIX OS, including Linux and OSX

LINUX

bool

If running on Linux

WSL

bool

If running in Windows Subsystem for Linux (WSL)

TIME_FMT

str

Format used for most string timestamps in PEAT

LOG_TIME_FMT

str

Time format used for most logging handlers

LOG_MESG_FMT

str

Message format used for all logging handlers

START_TIME_UTC

datetime

Time PEAT started in UTC timezone

START_TIME_LOCAL

datetime

Time PEAT started, as a local timezone datetime object

START_TIME

str

Time PEAT started, as a human-readable formatted string

TIMEZONE

str

Timezone of the system, e.g America/Denver

RUN_ID

int

Unique integer ID used to distinguish different PEAT runs

LOGO

str

The PEAT logo which is printed to the terminal at startup

SYSINFO

dict

Information about the system PEAT is running on, such as the hostname, username, and OS

8.3.4. State API

Runtime values that are used to preserve or share state across PEAT. Currently stored in peat.settings.state variable, which is a singleton instance of the State class. They can be accessed with from peat import state. The state values are saved to a file when PEAT finishes executing in peat_results/metadata/state/peat_state_***.json.

Note

It is possible to modify the starting state via environment variables. Environment variables beginning with PEAT_STATE_ will be loaded to their corresponding variables in the global state registry. This can be useful if debugging or as a temporary patch for a issue. Modifying the state in this manner should be avoided if possible, as it could cause undefined behavior or a crash.

class State(label, env_prefix, init_env=True)[source]

Global persistent state singleton, e.g. caching results of expensive operations.

arp_table: str = ''

The host’s ARP table. This is updated if a MAC address lookup is made and the ARP table is queried by PEAT. The ARP table is not always queried during lookups, so this variable may not always be populated. It also won’t be populated if RESOLVE_MAC is False.

comm_type: AllowedCommTypes | None

Communication method being used for an active command (scan, pull, or push). This is a bit of a hack to allow leaking a bit of CLI-specific info to device modules that need it.

Allowed values: unicast_ip, broadcast_ip, serial

elastic: peat.Elastic | None

Elastic instance used to interact with a Elasticsearch database, such as the one used by Vedizar.

local_interface_names: list[str]

Names of local network interfaces on the current system, e.g. eth0.

local_interface_details: dict[str, dict[str, str]]

Detailed local interface IPv4 and link configuration information.

local_interface_ips: list[str]

IP addresses of local interfaces.

local_interface_macs: list[str]

MAC addresses of local interfaces.

local_interface_objects: list[IPv4Interface]

IPv4Interface objects of the local interfaces.

local_interface_networks: dict[str, list[IPv4Network]]

Mapping of local interface names to their associated subnets.

local_networks: list[IPv4Network]

Networks the local system is connected to.

default_interface_name: str

The name of the default network interface of the local system.

default_interface_ip: str

The IP address of the default network interface of the local system.

default_interface_mac: str

The MAC address of the default network interface of the local system.

default_gateway: str

The IP address of the default gateway of the local system.

raw_socket_capable: bool

If RAW sockets are available (e.g. for ARP or ICMP requests).

entrypoint: EntrypointType

How PEAT is being used (PEAT’s “entrypoint”).

Allowed values:

  • CLI: PEAT’s CLI internals.

  • Server: if being run as a server, e.g. web UI.

  • Package: any other use of PEAT’s API.

peat_initialized: bool

If PEAT has been initialized, in other words peat.init.initialize_peat() has been called at least once.

written_files: set[str]

Paths of files written by PEAT during the current run. Only includes files that were written using PEAT’s APIs.

Note

These are NOT ordered by when they were written! (it’s a set)

superuser_privs: bool | None

Is PEAT running with permissions of root (POSIX) or Administrator (Windows).

error: bool

If any critical errors occurred during the run of PEAT.

8.3.5. Exceptions

class PeatError[source]

Generic class for any error raised by PEAT.

class ParseError[source]

Parsing errors.

class CommError[source]

Communication errors.

class DeviceError[source]

Errors related to a device or DeviceModule module.