EtherNet/IP Encapsulation#

[ODVA CIP Vol 2] EtherNet/IP encapsulation protocol codecs.

class icspacket.proto.cip.encap.EncapsulationCommand(*values)[source]#: EtherNet/IP encapsulation commands (See CIP Vol 2, clause 2-3.2, Table 2-3.2).

class icspacket.proto.cip.encap.EncapsulationPacket(*, command: EncapsulationCommand | int = 0, length: int = 0, session_handle: int = 0, status: int = 0, sender_context: bytes = b'\x00\x00\x00\x00\x00\x00\x00\x00', options: int = 0, payload_raw: bytes = b'')[source]#

Wraps a full EtherNet/IP message as sent on the wire: a constant 24-byte header followed by the raw command-specific payload bytes (See CIP Vol 2, clause 2-3.1, Table 2-3.1).

command: EncapsulationCommand | int = 0#: Which encapsulation command this message carries, determining how the payload bytes should be interpreted.

length: int = 0#: Byte count of the command-specific payload that follows this header, used to know how many trailing bytes belong to the message.

session_handle: int = 0#: Opaque session token whose meaning is left to the communicating endpoints rather than fixed by this header.

status: int = 0#: Outcome of the command, expressed as one of the EncapsulationStatus values (zero on success).

sender_context: bytes = b'\x00\x00\x00\x00\x00\x00\x00\x00'#: Free-form 8-byte value whose meaning is defined solely by whichever endpoint originated the message; other implementations do not need to interpret it.

options: int = 0#: General-purpose flag bits belonging to the encapsulation header.

payload_raw: bytes = b''#: Undecoded command-specific bytes; use payload to interpret them based on command. A StructDefMixin instance is also accepted and encoded automatically.

Command-specific payload, decoded according to command.

Falls back to the raw, undecoded bytes for a command with no known payload type.

class icspacket.proto.cip.encap.EncapsulationStatus(*values)[source]#: Common encapsulation status values (See CIP Vol 2, clause 2-3.5, Table 2-3.3).

class icspacket.proto.cip.encap.ListIdentityPayload(cpf: CPF = <factory>)[source]#: ListIdentity request/response payload (See CIP Vol 2, clause 2-4.2, Tables 2-4.2 and 2-4.3).

class icspacket.proto.cip.encap.ListInterfacesPayload(cpf: CPF = <factory>)[source]#: ListInterfaces request/response payload (See CIP Vol 2, clause 2-4.3, Tables 2-4.5 and 2-4.6).

class icspacket.proto.cip.encap.ListServicesPayload(cpf: CPF = <factory>)[source]#: ListServices request/response payload (See CIP Vol 2, clause 2-4.6, Tables 2-4.10 and 2-4.11).

class icspacket.proto.cip.encap.NOPPayload(data: bytes = b'')[source]#

NOP command-specific payload (See CIP Vol 2, clause 2-4.1, Table 2-4.1).

The command carries no required semantics, so this simply stores whatever raw bytes accompany it.

class icspacket.proto.cip.encap.RegisterSessionPayload(protocol_version: int = 1, options: int = 0)[source]#

RegisterSession payload (See CIP Vol 2, clause 2-4.4, Tables 2-4.7 and 2-4.8).

protocol_version: int = 1#: Encapsulation protocol version the client is requesting to use for this session.

options: int = 0#: Placeholder for future session option flags; none exist yet, so this is always sent as zero.

class icspacket.proto.cip.encap.SendRRDataPayload(interface_handle: int = 0, timeout: int = 0, cpf: CPF = <factory>)[source]#: SendRRData payload (See CIP Vol 2, clause 2-4.7, Tables 2-4.13 and 2-4.14).

class icspacket.proto.cip.encap.SendUnitDataPayload(interface_handle: int = 0, timeout: int = 0, cpf: CPF = <factory>)[source]#: SendUnitData payload (See CIP Vol 2, clause 2-4.8, Table 2-4.15).

class icspacket.proto.cip.encap.UnRegisterSessionPayload[source]#: UnRegisterSession has no command-specific data (See CIP Vol 2, clause 2-4.5, Table 2-4.9).

[ODVA CIP Vol 2] EtherNet/IP Common Packet Format codecs.

class icspacket.proto.cip.cpf.CPF(*, count: int = 0, items: list[CPFItem])[source]#

Common Packet Format envelope: an item count followed by that many typed items (See CIP Volume 2, clause 2-6.1, Table 2-6.1).

count: int = 0#: Number of items encoded in items; kept in sync automatically when constructed from items directly.

items: list[CPFItem]#: Item list, each entry wrapping a type_id/decoded-value pair. Use new()/values to work with bare item values instead.

property values: list[bytes | UnconnectedDataItem | ConnectedAddressItem | ConnectedDataItem | SequencedAddressItem | SockaddrInfoItem | SockaddrInfoOTItem | SockaddrInfoTOItem | ListServicesResponseItem | ListInterfacesResponseItem | ListIdentityResponseItem | NullAddressItem]#: items, unwrapped to their bare decoded values.

classmethod new(*values: bytes | UnconnectedDataItem | ConnectedAddressItem | ConnectedDataItem | SequencedAddressItem | SockaddrInfoItem | SockaddrInfoOTItem | SockaddrInfoTOItem | ListServicesResponseItem | ListInterfacesResponseItem | ListIdentityResponseItem | NullAddressItem) → Self[source]#: Builds a CPF from bare item values, wrapping each one in a CPFItem with its type_id derived automatically.

Fallback CPF item for a type ID with no struct registered via cpf_item().

type_id: int = 0#: Numeric code identifying which CPF item type this is; left undecoded since no struct is registered for it.

value: bytes | UnconnectedDataItem | ConnectedAddressItem | ConnectedDataItem | SequencedAddressItem | SockaddrInfoItem | SockaddrInfoOTItem | SockaddrInfoTOItem | ListServicesResponseItem | ListInterfacesResponseItem | ListIdentityResponseItem | NullAddressItem = b''#: decoded item body.

classmethod new(value: bytes | UnconnectedDataItem | ConnectedAddressItem | ConnectedDataItem | SequencedAddressItem | SockaddrInfoItem | SockaddrInfoOTItem | SockaddrInfoTOItem | ListServicesResponseItem | ListInterfacesResponseItem | ListIdentityResponseItem | NullAddressItem, type_id: int | CPFItemType | None = None) → Self[source]#: Builds a CPFItem, deriving type_id from value’s TYPE_ID when not given explicitly.

class icspacket.proto.cip.cpf.CPFItemType(*values)[source]#: Common Packet Format item type identifiers (See CIP Volume 2, clause 2-6.1, Table 2-6.3).

class icspacket.proto.cip.cpf.ConnectedAddressItem(connection_id: int)[source]#

Connected Address CPF item (See CIP Volume 2, clause 2-6.2.2, Table 2-6.5).

connection_id: int#: Numeric ID naming the connection this address item is directed at.

class icspacket.proto.cip.cpf.ConnectedDataItem(data: bytes = b'')[source]#: Connected Data CPF item (See CIP Volume 2, clause 2-6.3.2, Table 2-6.8).

class icspacket.proto.cip.cpf.ListIdentityResponseItem(protocol_version: int = 1, socket_address: SockaddrInfo = <factory>, vendor_id: int = 0, device_type: int = 0, product_code: int = 0, revision_major: int = 0, revision_minor: int = 0, status: int = 0, serial_number: int = 0, product_name: str = '', state: int = 0)[source]#

ListIdentity reply CPF item (See CIP Volume 2, clause 2-4.2.3, Table 2-4.4).

protocol_version: int = 1#: Encapsulation protocol version implemented by the responding device.

socket_address: SockaddrInfo#: Address and port where the target can be reached, embedded as a SockaddrInfo structure.

vendor_id: int = 0#: Numeric identifier of the device’s manufacturer.

device_type: int = 0#: Code classifying the general product category this device belongs to.

product_code: int = 0#: Code that distinguishes this specific product within its device type.

revision_major: int = 0#: Major segment of the device’s two-part revision number.

revision_minor: int = 0#: Minor segment of the device’s two-part revision number.

status: int = 0#: Device status value captured at the moment it answered.

serial_number: int = 0#: Serial number reported by the responding device.

product_name: str = ''#: Human-readable description of the device, encoded as a length-prefixed ASCII string.

state: int = 0#: Current operating state reported by the device.

class icspacket.proto.cip.cpf.ListInterfacesResponseItem(protocol_version: int = 1, capability_flags: int = 0)[source]#

Placeholder ListInterfaces reply item.

CIP Volume 2, clause 2-4.3 defines the command but no standard item body, so this only carries the protocol-version/capability-flags prefix any future item would share.

protocol_version: int = 1#: Version of the interface information the target claims to support.

capability_flags: int = 0#: Bit flags describing the target’s interface-related capabilities.

class icspacket.proto.cip.cpf.ListServicesResponseItem(protocol_version: int = 1, capability_flags: int = 0, service_name: str = '')[source]#

ListServices reply CPF item (See CIP Volume 2, clause 2-4.6, Table 2-4.11).

protocol_version: int = 1#: Version number of the CIP encapsulation service that the responding target implements.

capability_flags: int = 0#: Bit flags describing which optional capabilities the target’s service supports.

service_name: str = ''#: Human-readable service name; encoded as up to 16 bytes of ASCII, right-padded with NUL bytes.

class icspacket.proto.cip.cpf.NullAddressItem[source]#: Null Address CPF item (See CIP Volume 2, clause 2-6.2.1, Table 2-6.4).

class icspacket.proto.cip.cpf.SequencedAddressItem(connection_id: int, sequence_number: int)[source]#

Sequenced Address CPF item (See CIP Volume 2, clause 2-6.2.3, Table 2-6.6).

connection_id: int#: Numeric ID naming which open connection this sequenced item belongs to.

sequence_number: int#: Running counter carried alongside the connection ID so the receiver can order this connection’s I/O messages.

class icspacket.proto.cip.cpf.SockaddrInfo(family: int = 2, port: int = 0, address: IPv4Address = IPv4Address('0.0.0.0'), zero: bytes = b'\x00\x00\x00\x00\x00\x00\x00\x00')[source]#

Address and port pair embedded in CPF discovery items so a target device can advertise how it can be reached (See CIP Volume 2, clause 2-6.3.3, Table 2-6.9).

family: int = 2#: Selects the addressing scheme in use; this codec only builds and expects the IPv4 form, so the value is always 2 (AF_INET).

port: int = 0#: UDP port number the advertised CIP endpoint listens on.

address: IPv4Address = IPv4Address('0.0.0.0')#: IPv4 address of the advertised CIP endpoint.

zero: bytes = b'\x00\x00\x00\x00\x00\x00\x00\x00'#: Unused 8-byte filler with no defined meaning; callers should leave it zeroed.

class icspacket.proto.cip.cpf.SockaddrInfoItem(address_info: SockaddrInfo)[source]#

Base CPF item for originator/target socket-address information (See CIP Volume 2, clause 2-6.3.3, Table 2-6.9).

property family: int#: Address family from the embedded socket address.

property port: int#: TCP or UDP port from the embedded socket address.

property address: IPv4Address#: IPv4 address from the embedded socket address.

class icspacket.proto.cip.cpf.SockaddrInfoOTItem(address_info: SockaddrInfo)[source]#: Originator-to-target Sockaddr Info CPF item (See CIP Volume 2, clause 2-6.3.3, Table 2-6.9).

class icspacket.proto.cip.cpf.SockaddrInfoTOItem(address_info: SockaddrInfo)[source]#: Target-to-originator Sockaddr Info CPF item (See CIP Volume 2, clause 2-6.3.3, Table 2-6.9).

class icspacket.proto.cip.cpf.UnconnectedDataItem(data: bytes = b'')[source]#: Unconnected Data CPF item (See CIP Volume 2, clause 2-6.3.1, Table 2-6.7).

icspacket.proto.cip.cpf.cpf_item(cls: type[_T]) → type[_T][source]#
icspacket.proto.cip.cpf.cpf_item(cls: None = None) → Callable[[type[_T]], type[_T]]: Decorator to register a CPF item struct for its TYPE_ID.

icspacket.proto.cip.cpf.register_cpf_item(type_id: int | CPFItemType, item_cls: type[_T]) → type[_T][source]#

Most callers should use the cpf_item() decorator instead, which derives type_id from the class automatically; this is exposed directly for the rarer case of registering a struct that isn’t itself decorated (e.g. reusing one struct across several item types).