"""
Internet Control Message Protocol for IPv4 (TCP/IP protocol stack)
"""
from construct import *
from ipv4 import IpAddress
from binascii import unhexlify
import six
echo_payload = Struct("echo_payload",
UBInt16("identifier"),
UBInt16("sequence"),
Bytes("data", 32), # length is implementation dependent...
# is anyone using more than 32 bytes?
)
dest_unreachable_payload = Struct("dest_unreachable_payload",
Padding(2),
UBInt16("next_hop_mtu"),
IpAddress("host"),
Bytes("echo", 8),
)
dest_unreachable_code = Enum(Byte("code"),
Network_unreachable_error = 0,
Host_unreachable_error = 1,
Protocol_unreachable_error = 2,
Port_unreachable_error = 3,
The_datagram_is_too_big = 4,
Source_route_failed_error = 5,
Destination_network_unknown_error = 6,
Destination_host_unknown_error = 7,
Source_host_isolated_error = 8,
Desination_administratively_prohibited = 9,
Host_administratively_prohibited2 = 10,
Network_TOS_unreachable = 11,
Host_TOS_unreachable = 12,
)
icmp_header = Struct("icmp_header",
Enum(Byte("type"),
Echo_reply = 0,
Destination_unreachable = 3,
Source_quench = 4,
Redirect = 5,
Alternate_host_address = 6,
Echo_request = 8,
Router_advertisement = 9,
Router_solicitation = 10,
Time_exceeded = 11,
Parameter_problem = 12,
Timestamp_request = 13,
Timestamp_reply = 14,
Information_request = 15,
Information_reply = 16,
Address_mask_request = 17,
Address_mask_reply = 18,
_default_ = Pass,
),
Switch("code", lambda ctx: ctx.type,
{
"Destination_unreachable" : dest_unreachable_code,
},
default = Byte("code"),
),
UBInt16("crc"),
Switch("payload", lambda ctx: ctx.type,
{
"Echo_reply" : echo_payload,
"Echo_request" : echo_payload,
"Destination_unreachable" : dest_unreachable_payload,
},
default = Pass
)
)
if __name__ == "__main__":
cap1 = unhexlify(six.b("0800305c02001b006162636465666768696a6b6c6d6e6f70717273747576776162"
"63646566676869"))
cap2 = unhexlify(six.b("0000385c02001b006162636465666768696a6b6c6d6e6f70717273747576776162"
"63646566676869"))
cap3 = unhexlify(six.b("0301000000001122aabbccdd0102030405060708"))
print (icmp_header.parse(cap1))
print (icmp_header.parse(cap2))
print (icmp_header.parse(cap3))