8.4. Module API¶

8.4.1. Quickstart¶

The DeviceModule class is the implementation of a PEAT “device module”, and is the core of the PEAT Device Module API. To interact with a device, users of the API instantiate a DataManager instance with basic information about the device, then call the standard API functions on the implementation (e.g. SCEPTRE) and pass them the DataManager instance. Here is a simple example demonstrating usage:

Pulling data from a SEL relay via the network¶

from pathlib import Path
from pprint import pprint
from peat import SELRelay, datastore

# Create a DataManager instance with the device's IP address
device = datastore.get("192.0.2.22")

# Pass the instance to the pull_project method on the DeviceModule
# implementation "SELRelay". The data pulled is added to the DataManager
# instance created earlier.
SELRelay.pull_project(device)

# Export the data from the DataManager as a Python dictionary
pprint(device.export())

# Export it as JSON, sorted by key
print(device.json(sorted=True))

# Export to files (location set by config.DEVICE_RESULTS_DIR)
device.export_to_files()

Parsing the configuration for a SCEPTRE virtual field device¶

from pathlib import Path
from pprint import pprint
from peat import SCEPTRE

config_path = Path("examples/devices/sceptre/config.xml")
device = SCEPTRE.parse(config_path)
pprint(device.export())

Further examples of module usage can be found in the Python examples, peat/cli_main.py, and the API implementations in peat/api/.

8.4.2. Overview¶

To implement a module, refer to the Module developer guide.

8.4.2.1. Data model¶

All device data (e.g. firmware version, logic, etc.) is stored in instances of DeviceData. This is known as the “data model”. Refer to Data model for further details.

8.4.3. API¶

8.4.3.1. DeviceModule class¶

Warning

Not all of the methods defined in the base DeviceModule class are guaranteed to be implemented by a module implementation (subclass)

class DeviceModule[source]¶

Base class for all PEAT device modules.

The methods of this class represent the core PEAT API, and should be implemented by all devices that inherit from it. Sub-classes may add their own device-specific methods and data structures as needed.

ip_methods: list[IPMethod] = []¶: Methods for identifying devices via IP or Ethernet.

serial_methods: list[SerialMethod] = []¶: Methods for identifying devices via a serial connection.

device_type: str = ''¶: Type of device, e.g “PLC”, “Relay”, “RTU”, “RTAC”, etc. Elasticsearch field: host.type.

vendor_id: str = ''¶: Short-form vendor name. Elasticsearch field: host.description.vendor.id.

vendor_name: str = ''¶: Long-form vendor name. Elasticsearch field: host.description.vendor.name.

brand: str = ''¶: Device brand. Elasticsearch field: host.description.brand.

model: str = ''¶: Device’s default model (if not known). Elasticsearch field: host.description.model.

supported_models: list[str] = []¶: Device models this module supports or is known to work with.

filename_patterns: list[str] = []¶: Patterns of files the module is capable of parsing, if any. Patterns are a literal name (*SET_ALL.TXT) or a Unix shell glob (*.rdb), are case-insensitive, and must start with a wildcard character (*). Globs can be anything accepted by glob.

can_parse_dir: bool = False¶: If the module will accept a directory as the source path for parsing. If this is True, a Path like Path("./some_files/") would be a valid target for parsing. Any handling of files in the directory will have to be handled by the module, not PEAT.

module_aliases: list[str] = []¶: Alternative names for looking up the module, e.g. for the -d CLI option. Aliases that can be used to refer to the device via the PEAT’s Module API (peat.module_manager).

annotate_fields: dict[str, Any] = {}¶: Fields that will be annotated (populated) by default for most operations, such as scan pull, parse, etc. Examples include known OS versions or hardware architecture. These fields will populated ONLY IF they are already unset on the device being annotated. Format is the path to the field to populate, e.g. os.name, os.vendor.id, etc.

default_options: dict[str, Any] = {}¶: Define module-specific options and/or override global defaults, such as default ports for protocols or default credentials.

classmethod pull(dev)[source]¶

Pull artifacts from the device, such as logic, configuration, or firmware.

This wraps and calls _pull(). PEAT modules implementing the pull interface should implement _pull(), instead of this method.

Parameters:: dev (DeviceData) -- existing DeviceData object representing the device to pull from.
Return type:: bool
Returns:: If the pull was successful, as a bool
Raises:: DeviceError -- If a critical error occurred

classmethod _pull(dev)[source]¶

Implemented by modules. Subclass DeviceModule and override this method.

Return type:: bool

classmethod push(dev, to_push, push_type)[source]¶

Upload (push) configuration or firmware to a device.

This wraps and calls _push(). PEAT modules implementing the push interface should implement _push(), instead of this method.

Parameters:

dev (DeviceData) -- existing DeviceData object representing the device to push to.
to_push (str | bytes | Path) -- the information to push, either as a Path object pointing to a file or directory with config files to upload, or a raw string or bytes of file to upload.
push_type (Literal['config', 'firmware']) -- What information is being pushed, either ‘config’ or ‘firmware’. This comes from the -t command line argument.

Return type:

bool

Returns:

If the push was successful, as a bool

Raises:

DeviceError -- If a critical error occurred

classmethod _push(dev, to_push, push_type)[source]¶

Implemented by modules. Subclass DeviceModule and override this method.

Return type:: bool

classmethod parse(to_parse, dev=None)[source]¶

Parse device information from collected data or file artifacts.

Parameters:

to_parse (str | bytes | Path | IOBase) -- Data to be parsed. This can either be the Path of a file or the raw data to parse.
dev (DeviceData | None) -- existing DeviceData object to use instead of the one created by the module

Return type:

DeviceData | None

Returns:

Exported version of the parsed data object

Raises:

DeviceError -- If a critical error occurred

classmethod _parse(file, dev=None)[source]¶

Implemented by modules. Subclass DeviceModule and override this method.

Return type:: DeviceData | None

classmethod update_dev(dev)[source]¶

Update the device’s data with metadata, inferences, and lookups.

Note

Data values are only changed if they’re unset. In other words, existing values will NOT be overwritten.

What’s populated:

Basic Module attributes, e.g. cls.vendor_name => dev.description.vendor.name.
Any Module-defined fields in cls.annotate_fields, if present.
Calls populate_fields(), which populates fields such as adding description values, network interfaces, and other values. This call will also implicitly lookup MAC addresses, IP addresses, and/or hostnames, unless disabled with the appropriate PEAT global configuration options.

Parameters:: dev (DeviceData) -- DeviceData instance to annotate.
Return type:: None

classmethod method_implemented(method_name)[source]¶

Checks if a method of a subclass of DeviceModule is implemented and overrides the method in DeviceModule).

Return type:: bool

8.4.3.2. Identify methods¶

8.4.3.2.1. IPMethod¶

IPMethod[source]¶

Method for identifying devices via networking protocols (IP, Ethernet).

Show JSON schema

{
   "title": "IPMethod",
   "description": "Method for identifying devices via networking protocols (IP, Ethernet).",
   "type": "object",
   "properties": {
      "name": {
         "title": "Name",
         "type": "string"
      },
      "description": {
         "title": "Description",
         "type": "string"
      },
      "reliability": {
         "title": "Reliability",
         "minimum": 0,
         "maximum": 10,
         "type": "integer"
      },
      "protocol": {
         "title": "Protocol",
         "type": "string"
      },
      "transport": {
         "title": "Transport",
         "enum": [
            "tcp",
            "udp",
            "other"
         ],
         "type": "string"
      },
      "type": {
         "title": "Type",
         "enum": [
            "unicast_ip",
            "broadcast_ip"
         ],
         "type": "string"
      },
      "default_port": {
         "title": "Default Port",
         "minimum": 1,
         "maximum": 65535,
         "type": "integer"
      }
   },
   "required": [
      "name",
      "description",
      "reliability",
      "protocol",
      "transport",
      "type",
      "default_port"
   ],
   "additionalProperties": false
}

Fields:

default_port (peat.api.identify_methods.ConstrainedIntValue)
description (str)
identify_function (collections.abc.Callable | None)
name (str)
port_function (collections.abc.Callable | None)
protocol (peat.api.identify_methods.ConstrainedStrValue)
reliability (peat.api.identify_methods.ConstrainedIntValue)
transport (Literal['tcp', 'udp', 'other'])
type (Literal['unicast_ip', 'broadcast_ip'])

protocol: ConstrainedStrValue [Required]¶: Lowercase name of the protocol used.

Note

Similar protocols, such as http and https, should be two separate IPMethod instances with nearly identical attributes

transport: Literal['tcp', 'udp', 'other'] [Required]¶

Network transport protocol.

Allowed values:

tcp
udp
other

name: str [Required]¶: Human-friendly name for the method.

description: str [Required]¶: Human-friendly description of the method.

identify_function: Callable | None¶: Python function used to perform the verification.

Note

This is the identification function itself, since functions are first-class objects in Python and can be treated like classes

reliability: ConstrainedIntValue [Required]¶

Reliability of a method. Value ranges from from 0 (unknown reliability) to 10 (very reliable). Values of 5 or below are considored to have at least some degree of inconsistency or “flakiness” (e.g. Telnet user interface), while 6 and up are considored to be fairly reliable (e.g. HTTP). This is used by PEAT to sort methods during discovery and other similar contexts, as well as for future features.

Constraints:

minimum = 0
maximum = 10

type: Literal['unicast_ip', 'broadcast_ip'] [Required]¶

The type of IP method this is.

Allowed values for this method (IPMethod):

unicast_ip
broadcast_ip

default_port: ConstrainedIntValue [Required]¶

Default protocol port used by the service the method interacts with. Note that a different port may be used if configured at runtime.

Constraints:

minimum = 1
maximum = 65535

port_function: Callable | None¶: Python function used to check if the port is open. Defaults to standard TCP/UDP check based on the value of transport (e.g. a transport of tcp will cause PEAT to use a TCP SYN-RST method to check if the port is open).

8.4.3.2.2. SerialMethod¶

SerialMethod[source]¶

Method for identifying devices via Serial interfaces and protocols (e.g. RS-232).

Show JSON schema

{
   "title": "SerialMethod",
   "description": "Method for identifying devices via Serial interfaces and protocols (e.g. RS-232).",
   "type": "object",
   "properties": {
      "name": {
         "title": "Name",
         "type": "string"
      },
      "description": {
         "title": "Description",
         "type": "string"
      },
      "reliability": {
         "title": "Reliability",
         "minimum": 0,
         "maximum": 10,
         "type": "integer"
      },
      "type": {
         "title": "Type",
         "enum": [
            "direct"
         ],
         "type": "string"
      }
   },
   "required": [
      "name",
      "description",
      "reliability",
      "type"
   ],
   "additionalProperties": false
}

Fields:

description (str)
identify_function (collections.abc.Callable | None)
name (str)
reliability (peat.api.identify_methods.ConstrainedIntValue)
type (Literal['direct'])

name: str [Required]¶: Human-friendly name for the method.

description: str [Required]¶: Human-friendly description of the method.

identify_function: Callable | None¶: Python function used to perform the verification.

Note

This is the identification function itself, since functions are first-class objects in Python and can be treated like classes

reliability: ConstrainedIntValue [Required]¶

Constraints:

minimum = 0
maximum = 10

type: Literal['direct'] [Required]¶

The type of method, is it direct or broadcast?

Currently, only “direct” is used, but “broadcast” may be added in the future for transports like CANbus, RS485, etc.

Allowed values for this method (SerialMethod):

direct

8.4.3.3. Module manager¶

class ModuleManager[source]¶

Manager for PEAT modules that implement functionality of a device.

modules¶: All currently imported device modules, keyed by module name

module_aliases¶: Mapping of alias keys to device modules

runtime_imports¶: Modules that were imported at runtime

runtime_paths¶: Paths of modules imported at runtime, if they were imported from a path.

property names: list[str]¶: Sorted list of class names of all imported modules.

property classes: list[type[DeviceModule]]¶: Sorted list of classes of all imported modules.

property aliases: list[str]¶: Sorted list of all currently registered module aliases.

property alias_mappings: dict[str, list[str]]¶: Dict of aliases and the module names they map to.

filter_names(filter_attr)[source]¶

Return module names for which the attribute exists and is truthy.

Example: set filter_attr to "ip_methods" to get all of the modules that support IP identification (ip_methods).

Parameters:: filter_attr (str) -- Class attribute string to filter using
Return type:: list[str]
Returns:: Module names for which the attribute exists and is truthy.

import_module(module, remove_aliases=True)[source]¶

Import a PEAT module.

Parameters:

module (Any) -- module to import. Can be a path (str or Path, class, or list of paths.
remove_aliases (bool) -- if aliases for an existing module should be removed

Return type:

bool

Returns:

If the import was successful

import_module_path(path, remove_aliases=True)[source]¶

Import all modules from a directory.

Note

The module CANNOT be “hidden” by placing the contents in an __init__.py file, and must reside in it’s own .py file (e.g. mymodule.py).

Parameters:

path (Path) -- Path to the directory containing modules to import
remove_aliases (bool) -- If aliases for an existing module should be removed

Return type:

bool

Returns:

If the import was successful

import_module_cls(module, remove_aliases=True)[source]¶

Adds the module object to the registered PEAT device modules.

Parameters:

module (type[DeviceModule]) -- The module class to import
remove_aliases (bool) -- If aliases for an existing module should be removed

Return type:

bool

Returns:

If the import was successful

get_module(name)[source]¶

Get module by name.

Parameters:: name (str) -- Exact name of a PEAT module
Return type:: type[DeviceModule] | None
Returns:: The PEAT module object (subclass of DeviceModule), or None if the module isn’t imported or doesn’t exist.

get_modules(name, filter_attr=None)[source]¶

Get PEAT device module classes.

Parameters:

name (str) -- Module name or alias
filter_attr (str | None) -- Only return modules for which this attribute is true

Return type:

list[type[DeviceModule]]

Returns:

List of module classes (subclasses of DeviceModule)

lookup_types(dev_types=None, filter_attr=None, subclass_method=None, filter_values=None)[source]¶

Process strings and classes into a sorted list of module classes.

Parameters:

dev_types (Any | None | list[str | type[DeviceModule] | DeviceModule]) -- DeviceModule names, aliases, classes, or instances to use and resolve into a list of module classes. If None, all currently imported modules searched.
filter_attr (str | None) -- Only return modules for which this attribute is true
subclass_method (str | None) -- Filter modules that have implemented a method from the base DeviceModule class, e.g. identify_ip().
filter_values (dict | None) -- Values to filter modules by, with dict keys being names of module class attributes and values being the values to compare. Comparisons must be exact matches and strings are therefore case-sensitive. Example: {"device_type": "PLC"}

Return type:

list[type[DeviceModule]]

Returns:

List of module classes (subclasses of DeviceModule)

lookup_names(dev_types, filter_attr=None, subclass_method=None, filter_values=None)[source]¶

Process strings and classes into a sorted list of module names.

This is a thin wrapper around lookup_types().

Parameters:

dev_types (Any | None | list[str | type[DeviceModule] | DeviceModule]) -- DeviceModule names, aliases, classes, or instances to use and resolve into a list of module classes. If None, all currently imported modules searched.
filter_attr (str | None) -- Only return modules for which this attribute is true
subclass_method (str | None) -- Filter modules that have implemented a method from the base DeviceModule class, e.g. identify_ip().
filter_values (dict | None) -- Values to filter modules by, with dict keys being names of module class attributes and values being the values to compare. Comparisons must be exact matches and strings are therefore case-sensitive. Example: {"device_type": "PLC"}

Return type:

list[str]

Returns:

List of module names

alias_to_names(alias)[source]¶

Resolve an alias into names of modules it corresponds to.

Parameters:: alias (str) -- The alias to resolve
Return type:: list[str]
Returns:: List of names of modules that the alias resolved to, or an empty list if it didn’t resolve to anything.

classmethod is_valid_module(module)[source]¶

Checks if a Python object is a PEAT device module.

Return type:: bool