forked from scy-phy/scapy-cip-enip
-
Notifications
You must be signed in to change notification settings - Fork 2
/
Copy pathenip_cpf.py
123 lines (107 loc) · 5.88 KB
/
enip_cpf.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
#!/usr/bin/env python2
# -*- coding: utf-8 -*-
# Copyright (c) 2015 David I. Urbina, [email protected]
# Copyright (c) 2016 Andrey Dolgikh, Binghamton University
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
"""Ethernet/IP Common Packet Format Scapy dissector."""
import struct
from scapy import all as scapy_all
import utils
ITEM_ID_NUMBERS = utils.merge_dicts({
0x0000 : "Null", #(used for UCMM messages). Indicates that encapsulation routing is NOT needed. Target is either local (ethernet) or routing info is in a data Item.
0x000C : "ListIdentity Response", #
0x0091 : "Reserved", # for legacy (RA)
0x00A1 : "Connected Address Item", # (used for connected messages)
0x00B1 : "Connected Data Item", # Connected Transport packet
0x00B2 : "Unconnected Data Item", # Unconnected Messages (eg. used within CIP command SendRRData)
0x0100 : "ListServices response", #
0x8000 : "Sockaddr Info, originator-to-target", #
0x8001 : "Sockaddr Info, target-to-originator", #
0x8002 : "Sequenced Address item", #
},
{k: "Reserved for legacy (RA)" for k in range(0x0001, 0x000B + 1)}, # 0x0001 – 0x000B Reserved for legacy (RA)
{k: "Reserved for legacy (RA)" for k in range(0x000D, 0x0083 + 1)}, # 0x000D – 0x0083 Reserved for legacy (RA)
{k: "Reserved for future expansion" for k in range(0x0084, 0x0090 + 1)}, # 0x0084 – 0x0090 Reserved for future expansion
{k: "Reserved for future expansion" for k in range(0x0092, 0x00A0 + 1)}, # 0x0092 – 0x00A0 Reserved for future expansion
{k: "Reserved for legacy (RA)" for k in range(0x00A2, 0x00A4 + 1)}, # 0x00A2 – 0x00A4 Reserved for legacy (RA)
{k: "Reserved for future expansion" for k in range(0x00A5, 0x00B0 + 1)}, # 0x00A5 – 0x00B0 Reserved for future expansion
{k: "Reserved for future expansion" for k in range(0x00B3, 0x00FF + 1)}, # 0x00B3 – 0x00FF Reserved for future expansion
{k: "Reserved for legacy (RA)" for k in range(0x0101, 0x010F + 1)}, # 0x0101 – 0x010F Reserved for legacy (RA)
{k: "Reserved for future expansion" for k in range(0x0110, 0x7FFF + 1)}, # 0x0110 – 0x7FFF Reserved for future expansion
{k: "Reserved for future expansion" for k in range(0x8003, 0xFFFF + 1)}, # 0x8003 – 0xFFFF Reserved for future expansion
#regexps to produce dicts above
#(0x[\d|{ABCDF}]{4}).{3}(0x[\d|{ABCDF}]{4}) (.*)
#\{k\: \"$3\" for k in range\($1\, $2 \+ 1\)\}\, \#
)
class CPF_SequencedAddressItem(scapy_all.Packet):
name = "CPF_SequencedAddressItem"
fields_desc = [
scapy_all.LEIntField("connection_id", 0),
scapy_all.LEIntField("sequence_number", 0),
]
# Renamed so that Address Item and Data Item have different names - MED
class CPF_Item(scapy_all.Packet):
name = "CPF_Item"
fields_desc = [
scapy_all.LEShortEnumField('type_id', 0, ITEM_ID_NUMBERS),
scapy_all.LEShortField("length", None),
scapy_all.ConditionalField(
scapy_all.FieldListField(
'data',0,scapy_all.XByteField('', 0),
count_from = lambda p: p.length
),
lambda p: p.type_id==0xB1 # or p.type_id==0xB2
)
]
def extract_padding(self, p):
return p[:self.length], p[self.length:]
def post_build(self, p, pay):
if self.length is None and pay:
l = len(pay)
p = p[:2] + struct.pack("<H", l) + p[4:]
return p + pay
class CPF_AddressDataItem(CPF_Item):
name = "CPF_AddressDataItem"
class CPF_DataItem(CPF_Item):
name = "CPF_DataItem"
class ENIP_CPF(scapy_all.Packet):
name = "ENIP_CPF"
fields_desc = [
utils.LEShortLenField("count", 2, count_of="items"),
# Changed implementation to reflect use of CIP_Item above
scapy_all.PacketListField("items", [], CPF_Item,
count_from=lambda p: p.count),
# scapy_all.PacketField("Address_Item", CPF_AddressDataItem('', type_id=0x0, length=0), CPF_AddressDataItem),
# scapy_all.PacketField("Data_Item", CPF_DataItem('', type_id=0x0, length=0), CPF_DataItem),
# scapy_all.PacketField("Address_Item", CPF_AddressDataItem('', type_id=0xA1, length=None), CPF_AddressDataItem),
# scapy_all.PacketField("Data_Item", CPF_DataItem('', type_id=0xB1, length=None), CPF_DataItem),
# Due to potential 'optional' packet components at end in protocol
scapy_all.ConditionalField(
scapy_all.PacketListField("optional_items", None, CPF_Item, count_from=lambda p: p.count-2),
lambda p: p.count>2
),
]
def extract_padding(self, p):
return '', p
# Added additional binds - MED
scapy_all.bind_layers(CPF_AddressDataItem, CPF_SequencedAddressItem, type_id=0x8002)
scapy_all.bind_layers(CPF_Item, CPF_SequencedAddressItem, type_id=0x8002)
scapy_all.bind_layers(ENIP_CPF, CPF_DataItem, type_id=0x8002)
scapy_all.bind_layers(CPF_DataItem, CPF_SequencedAddressItem, type_id=0x00B2)