# SPDX-License-Identifier: GPL-2.0-or-later # This file is part of Scapy # See https://scapy.net/ for more information # Copyright (C) 2019 Jose Diogo Monteiro # Updated (C) 2023 Claire Vacherot # scapy.contrib.description = EtherNet/IP # scapy.contrib.status = loads """ EtherNet/IP (Industrial Protocol) Based on https://github.com/scy-phy/scapy-cip-enip EtherNet/IP Home: www.odva.org """ import struct from scapy.packet import Packet, bind_layers from scapy.layers.inet import TCP from scapy.fields import LEShortField, LEShortEnumField, LEIntEnumField, \ LEIntField, LELongField, FieldLenField, PacketListField, ByteField, \ StrLenField, StrFixedLenField, XLEIntField, XLEStrLenField, \ LEFieldLenField, ShortField, IPField, LongField, XLEShortField _commandIdList = { 0x0001: "UnknownCommand", 0x0004: "ListServices", # Request Struct Don't Have Command Spec Data 0x0063: "ListIdentity", # Request Struct Don't Have Command Spec Data 0x0064: "ListInterfaces", # Request Struct Don't Have Command Spec Data 0x0065: "RegisterSession", # Request Structure = Reply Structure 0x0066: "UnregisterSession", # Don't Have Command Specific Data 0x006f: "SendRRData", # Request Structure = Reply Structure 0x0070: "SendUnitData", # There is no reply 0x0072: "IndicateStatus", 0x0073: "Cancel" } _statusList = { 0: "success", 1: "invalid_cmd", 2: "no_resources", 3: "incorrect_data", 100: "invalid_session", 101: "invalid_length", 105: "unsupported_prot_rev" } _typeIdList = { 0x0000: "Null Address Item", 0x000c: "CIP Identity", 0x0086: "CIP Security Information", 0x0087: "EtherNet/IP Capability", 0x0088: "EtherNet/IP Usage", 0x00a1: "Connected Address Item", 0x00B1: "Connected Data Item", 0x00B2: "Unconnected Data Item", 0x0100: "List Services Response", 0x8000: "Socket Address Info O->T", 0x8001: "Socket Address Info T->O", 0x8002: "Sequenced Address Item", 0x8003: "Unconnected Message over UDP" } _deviceTypeList = { 0x0000: "Generic Device (deprecated)", 0x0002: "AC Drive", 0x0003: "Motor Overload", 0x0004: "Limit Switch", 0x0005: "Inductive Proximity Switch", 0x0006: "Photoelectric Sensor", 0x0007: "General Purpose Discrete I/O", 0x0009: "Resolver", 0x000C: "Communications Adapter", 0x000E: "Programmable Logic Controller", 0x0010: "Position Controller", 0x0013: "DC Drive", 0x0015: "Contactor", 0x0016: "Motor Starter", 0x0017: "Soft Start", 0x0018: "Human-Machine Interface", 0x001A: "Mass Flow Controller", 0x001B: "Pneumatic Valve", 0x001C: "Vacuum Pressure Gauge", 0x001D: "Process Control Value", 0x001E: "Residual Gas Analyzer", 0x001F: "DC Power Generator", 0x0020: "RF Power Generator", 0x0021: "Turbomolecular Vacuum Pump", 0x0022: "Encoder", 0x0023: "Safety Discrete I/O Device", 0x0024: "Fluid Flow Controller", 0x0025: "CIP Motion Drive", 0x0026: "CompoNet Repeater", 0x0027: "Mass Flow Controller, Enhanced", 0x0028: "CIP Modbus Device", 0x0029: "CIP Modbus Translator", 0x002A: "Safety Analog I/O Device", 0x002B: "Generic Device (keyable)", 0x002C: "Managed Ethernet Switch", 0x002D: "CIP Motion Safety Drive Device", 0x002E: "Safety Drive Device", 0x002F: "CIP Motion Encoder", 0x0030: "CIP Motion Converter", 0x0031: "CIP Motion I/O", 0x0032: "ControlNet Physical Layer Component", 0x0033: "Circuit Breaker", 0x0034: "HART Device", 0x0035: "CIP-HART Translator", 0x00C8: "Embedded Component", } _interfaceList = { 0x00: "CIP" } class ItemData(Packet): """Common Packet Format""" name = "Item Data" fields_desc = [ LEShortEnumField("typeId", 0, _typeIdList), LEShortField("length", 0), XLEStrLenField("data", "", length_from=lambda pkt: pkt.length), ] def extract_padding(self, s): return '', s # Unknown command (0x0001) class ENIPUnknownCommand(Packet): """Unknown Command reply""" name = "ENIPUnknownCommand" pass # List services (0x0004) class ENIPListServicesItem(Packet): """List Services Item Field""" name = "ENIPListServicesItem" fields_desc = [ LEShortEnumField("itemTypeCode", 0, _typeIdList), LEFieldLenField("itemLength", 0), LEShortField("protocolVersion", 0), XLEShortField("flag", 0), # TODO: detail with BitFields StrFixedLenField("serviceName", None, 16), ] class ENIPListServices(Packet): """List Services Command Field""" name = "ENIPListServices" fields_desc = [ FieldLenField("itemCount", 0, count_of="items"), PacketListField("items", None, ENIPListServicesItem), ] # List identity (0x0063) class ENIPListIdentityItem(Packet): """List Identity Item Fields""" name = "ENIPListIdentityReplyItem" fields_desc = [ LEShortEnumField("itemTypeCode", 0, _typeIdList), LEFieldLenField("itemLength", 0), LEShortField("protocolVersion", 0), # Socket address ShortField("sinFamily", 0), ShortField("sinPort", 0), IPField("sinAddress", None), LongField("sinZero", 0), # End socket address LEShortField("vendorId", 0), # Vendor list could be added (long list) LEShortEnumField("deviceType", 0, _deviceTypeList), LEShortField("productCode", 0), ByteField("revisionMajor", 0), ByteField("revisionMinor", 0), LEShortField("status", 0), XLEIntField("serialNumber", 0), ByteField("productNameLength", 0), StrLenField("productName", None, length_from=lambda pkt: pkt.productNameLength), ByteField("state", 0) ] class ENIPListIdentity(Packet): """List identity request and response""" name = "ENIPListIdentity" fields_desc = [ FieldLenField("itemCount", 0, count_of="items"), PacketListField("items", None, ENIPListIdentityItem) ] # List Interfaces (0x0064) class ENIPListInterfacesItem(Packet): """List Interfaces Item Fields""" name = "ENIPListInterfacesItem" fields_desc = [ LEShortEnumField("itemTypeCode", 0, _typeIdList), FieldLenField("itemLength", 0, length_of="itemData"), # TODO: Could be detailed StrLenField("itemData", "", length_from=lambda pkt: pkt.itemLength), ] class ENIPListInterfaces(Packet): """List Interfaces Command Field""" name = "ENIPListInterfaces" fields_desc = [ FieldLenField("itemCount", 0, count_of="items"), PacketListField("items", None, ENIPListInterfacesItem), ] # Register Session (0x0065) class ENIPRegisterSession(Packet): """Register Session Command Field""" name = "ENIPRegisterSession" fields_desc = [ LEShortField("protocolVersion", 1), LEShortField("options", 0) ] # Unregister Session (0x0066) -- Requires further testing class ENIPUnregisterSession(Packet): """Unregister Session Command Field""" name = "ENIPUnregisterSession" pass # Send RR Data (0x006f) class ENIPSendRRData(Packet): """Send RR Data Command Field""" name = "ENIPSendRRData" fields_desc = [ LEIntEnumField("interface", 0, _interfaceList), LEShortField("timeout", 0xff), LEFieldLenField("itemCount", 0, count_of="items"), PacketListField("items", None, ItemData) # TODO: Send RR Data is usually followed by a CIP packet ] # Send Unit Data (0x006f) class ENIPSendUnitData(Packet): """Send Unit Data Command Field""" name = "ENIPSendUnitData" fields_desc = [ LEIntEnumField("interface", 0, _interfaceList), LEShortField("timeout", 0xff), LEFieldLenField("itemCount", 0, count_of="items"), PacketListField("items", None, ItemData) ] # Main Ethernet/IP packet structure with header class ENIPTCP(Packet): """Ethernet/IP packet over TCP""" name = "ENIPTCP" fields_desc = [ LEShortEnumField("commandId", None, _commandIdList), LEShortField("length", 0), XLEIntField("session", 0), LEIntEnumField("status", None, _statusList), LELongField("senderContext", 0), LEIntField("options", 0), ] def post_build(self, pkt, pay): if self.length is None and pay: pkt = pkt[:2] + struct.pack("