mirror of
https://github.com/fortra/impacket
synced 2026-06-08 14:15:13 +00:00
gabrielg5
531 lines
19 KiB
Python
531 lines
19 KiB
Python
# Impacket - Collection of Python classes for working with network protocols.
|
|
#
|
|
# Copyright Fortra, LLC and its affiliated companies
|
|
#
|
|
# All rights reserved.
|
|
#
|
|
# This software is provided under a slightly modified version
|
|
# of the Apache Software License. See the accompanying LICENSE file
|
|
# for more information.
|
|
#
|
|
|
|
import array
|
|
import struct
|
|
|
|
from impacket.ImpactPacket import Header, Data, array_tobytes
|
|
from impacket.IP6_Address import IP6_Address
|
|
|
|
|
|
class ICMP6(Header):
|
|
#IP Protocol number for ICMP6
|
|
IP_PROTOCOL_NUMBER = 58
|
|
protocol = IP_PROTOCOL_NUMBER #ImpactDecoder uses the constant "protocol" as the IP Protocol Number
|
|
|
|
#Size of ICMP6 header (excluding payload)
|
|
HEADER_SIZE = 4
|
|
|
|
#ICMP6 Message Type numbers
|
|
DESTINATION_UNREACHABLE = 1
|
|
PACKET_TOO_BIG = 2
|
|
TIME_EXCEEDED = 3
|
|
PARAMETER_PROBLEM = 4
|
|
ECHO_REQUEST = 128
|
|
ECHO_REPLY = 129
|
|
ROUTER_SOLICITATION = 133
|
|
ROUTER_ADVERTISEMENT = 134
|
|
NEIGHBOR_SOLICITATION = 135
|
|
NEIGHBOR_ADVERTISEMENT = 136
|
|
REDIRECT_MESSAGE = 137
|
|
NODE_INFORMATION_QUERY = 139
|
|
NODE_INFORMATION_REPLY = 140
|
|
|
|
#Destination Unreachable codes
|
|
NO_ROUTE_TO_DESTINATION = 0
|
|
ADMINISTRATIVELY_PROHIBITED = 1
|
|
BEYOND_SCOPE_OF_SOURCE_ADDRESS = 2
|
|
ADDRESS_UNREACHABLE = 3
|
|
PORT_UNREACHABLE = 4
|
|
SOURCE_ADDRESS_FAILED_INGRESS_EGRESS_POLICY = 5
|
|
REJECT_ROUTE_TO_DESTINATION = 6
|
|
|
|
#Time Exceeded codes
|
|
HOP_LIMIT_EXCEEDED_IN_TRANSIT = 0
|
|
FRAGMENT_REASSEMBLY_TIME_EXCEEDED = 1
|
|
|
|
#Parameter problem codes
|
|
ERRONEOUS_HEADER_FIELD_ENCOUNTERED = 0
|
|
UNRECOGNIZED_NEXT_HEADER_TYPE_ENCOUNTERED = 1
|
|
UNRECOGNIZED_IPV6_OPTION_ENCOUNTERED = 2
|
|
|
|
#Node Information codes
|
|
NODE_INFORMATION_QUERY_IPV6 = 0
|
|
NODE_INFORMATION_QUERY_NAME_OR_EMPTY = 1
|
|
NODE_INFORMATION_QUERY_IPV4 = 2
|
|
NODE_INFORMATION_REPLY_SUCCESS = 0
|
|
NODE_INFORMATION_REPLY_REFUSED = 1
|
|
NODE_INFORMATION_REPLY_UNKNOWN_QTYPE = 2
|
|
|
|
#Node Information qtypes
|
|
NODE_INFORMATION_QTYPE_NOOP = 0
|
|
NODE_INFORMATION_QTYPE_UNUSED = 1
|
|
NODE_INFORMATION_QTYPE_NODENAME = 2
|
|
NODE_INFORMATION_QTYPE_NODEADDRS = 3
|
|
NODE_INFORMATION_QTYPE_IPv4ADDRS = 4
|
|
|
|
#ICMP Message semantic types (error or informational)
|
|
ERROR_MESSAGE = 0
|
|
INFORMATIONAL_MESSAGE = 1
|
|
|
|
#ICMP message dictionary - specifying text descriptions and valid message codes
|
|
#Key: ICMP message number
|
|
#Data: Tuple ( Message Type (error/informational), Text description, Codes dictionary (can be None) )
|
|
#Codes dictionary
|
|
#Key: Code number
|
|
#Data: Text description
|
|
|
|
#ICMP message dictionary tuple indexes
|
|
MSG_TYPE_INDEX = 0
|
|
DESCRIPTION_INDEX = 1
|
|
CODES_INDEX = 2
|
|
|
|
icmp_messages = {
|
|
DESTINATION_UNREACHABLE : (ERROR_MESSAGE, "Destination unreachable",
|
|
{ NO_ROUTE_TO_DESTINATION : "No route to destination",
|
|
ADMINISTRATIVELY_PROHIBITED : "Administratively prohibited",
|
|
BEYOND_SCOPE_OF_SOURCE_ADDRESS : "Beyond scope of source address",
|
|
ADDRESS_UNREACHABLE : "Address unreachable",
|
|
PORT_UNREACHABLE : "Port unreachable",
|
|
SOURCE_ADDRESS_FAILED_INGRESS_EGRESS_POLICY : "Source address failed ingress/egress policy",
|
|
REJECT_ROUTE_TO_DESTINATION : "Reject route to destination"
|
|
}),
|
|
PACKET_TOO_BIG : (ERROR_MESSAGE, "Packet too big", None),
|
|
TIME_EXCEEDED : (ERROR_MESSAGE, "Time exceeded",
|
|
{HOP_LIMIT_EXCEEDED_IN_TRANSIT : "Hop limit exceeded in transit",
|
|
FRAGMENT_REASSEMBLY_TIME_EXCEEDED : "Fragment reassembly time exceeded"
|
|
}),
|
|
PARAMETER_PROBLEM : (ERROR_MESSAGE, "Parameter problem",
|
|
{
|
|
ERRONEOUS_HEADER_FIELD_ENCOUNTERED : "Erroneous header field encountered",
|
|
UNRECOGNIZED_NEXT_HEADER_TYPE_ENCOUNTERED : "Unrecognized Next Header type encountered",
|
|
UNRECOGNIZED_IPV6_OPTION_ENCOUNTERED : "Unrecognized IPv6 Option Encountered"
|
|
}),
|
|
ECHO_REQUEST : (INFORMATIONAL_MESSAGE, "Echo request", None),
|
|
ECHO_REPLY : (INFORMATIONAL_MESSAGE, "Echo reply", None),
|
|
ROUTER_SOLICITATION : (INFORMATIONAL_MESSAGE, "Router Solicitation", None),
|
|
ROUTER_ADVERTISEMENT : (INFORMATIONAL_MESSAGE, "Router Advertisement", None),
|
|
NEIGHBOR_SOLICITATION : (INFORMATIONAL_MESSAGE, "Neighbor Solicitation", None),
|
|
NEIGHBOR_ADVERTISEMENT : (INFORMATIONAL_MESSAGE, "Neighbor Advertisement", None),
|
|
REDIRECT_MESSAGE : (INFORMATIONAL_MESSAGE, "Redirect Message", None),
|
|
NODE_INFORMATION_QUERY: (INFORMATIONAL_MESSAGE, "Node Information Query", None),
|
|
NODE_INFORMATION_REPLY: (INFORMATIONAL_MESSAGE, "Node Information Reply", None),
|
|
}
|
|
|
|
|
|
|
|
|
|
############################################################################
|
|
def __init__(self, buffer = None):
|
|
Header.__init__(self, self.HEADER_SIZE)
|
|
if (buffer):
|
|
self.load_header(buffer)
|
|
|
|
def get_header_size(self):
|
|
return self.HEADER_SIZE
|
|
|
|
def get_ip_protocol_number(self):
|
|
return self.IP_PROTOCOL_NUMBER
|
|
|
|
def __str__(self):
|
|
type = self.get_type()
|
|
code = self.get_code()
|
|
checksum = self.get_checksum()
|
|
|
|
s = "ICMP6 - Type: " + str(type) + " - " + self.__get_message_description() + "\n"
|
|
s += "Code: " + str(code)
|
|
if (self.__get_code_description() != ""):
|
|
s += " - " + self.__get_code_description()
|
|
s += "\n"
|
|
s += "Checksum: " + str(checksum) + "\n"
|
|
return s
|
|
|
|
def __get_message_description(self):
|
|
return self.icmp_messages[self.get_type()][self.DESCRIPTION_INDEX]
|
|
|
|
def __get_code_description(self):
|
|
code_dictionary = self.icmp_messages[self.get_type()][self.CODES_INDEX]
|
|
if (code_dictionary is None):
|
|
return ""
|
|
else:
|
|
return code_dictionary[self.get_code()]
|
|
|
|
############################################################################
|
|
def get_type(self):
|
|
return (self.get_byte(0))
|
|
|
|
def get_code(self):
|
|
return (self.get_byte(1))
|
|
|
|
def get_checksum(self):
|
|
return (self.get_word(2))
|
|
|
|
############################################################################
|
|
def set_type(self, type):
|
|
self.set_byte(0, type)
|
|
|
|
def set_code(self, code):
|
|
self.set_byte(1, code)
|
|
|
|
def set_checksum(self, checksum):
|
|
self.set_word(2, checksum)
|
|
|
|
############################################################################
|
|
def calculate_checksum(self):
|
|
#Initialize the checksum value to 0 to yield a correct calculation
|
|
self.set_checksum(0)
|
|
#Fetch the pseudo header from the IP6 parent packet
|
|
pseudo_header = self.parent().get_pseudo_header()
|
|
#Fetch the ICMP data
|
|
icmp_header = self.get_bytes()
|
|
#Build an array of bytes concatenating the pseudo_header, the ICMP header and the ICMP data (if present)
|
|
checksum_array = array.array('B')
|
|
checksum_array.extend(pseudo_header)
|
|
checksum_array.extend(icmp_header)
|
|
if (self.child()):
|
|
checksum_array.extend(self.child().get_bytes())
|
|
|
|
#Compute the checksum over that array
|
|
self.set_checksum(self.compute_checksum(checksum_array))
|
|
|
|
def is_informational_message(self):
|
|
return self.icmp_messages[self.get_type()][self.MSG_TYPE_INDEX] == self.INFORMATIONAL_MESSAGE
|
|
|
|
def is_error_message(self):
|
|
return self.icmp_messages[self.get_type()][self.MSG_TYPE_INDEX] == self.ERROR_MESSAGE
|
|
|
|
def is_well_formed(self):
|
|
well_formed = True
|
|
|
|
#Check that the message type is known
|
|
well_formed &= self.get_type() in self.icmp_messages.keys()
|
|
|
|
#Check that the code is known (zero, if there are no codes defined)
|
|
code_dictionary = self.icmp_messages[self.get_type()][self.CODES_INDEX]
|
|
if (code_dictionary is None):
|
|
well_formed &= self.get_code() == 0
|
|
else:
|
|
well_formed &= self.get_code() in code_dictionary.keys()
|
|
|
|
return well_formed
|
|
|
|
############################################################################
|
|
|
|
@classmethod
|
|
def Echo_Request(class_object, id, sequence_number, arbitrary_data = None):
|
|
return class_object.__build_echo_message(ICMP6.ECHO_REQUEST, id, sequence_number, arbitrary_data)
|
|
|
|
@classmethod
|
|
def Echo_Reply(class_object, id, sequence_number, arbitrary_data = None):
|
|
return class_object.__build_echo_message(ICMP6.ECHO_REPLY, id, sequence_number, arbitrary_data)
|
|
|
|
@classmethod
|
|
def __build_echo_message(class_object, type, id, sequence_number, arbitrary_data):
|
|
#Build ICMP6 header
|
|
icmp_packet = ICMP6()
|
|
icmp_packet.set_type(type)
|
|
icmp_packet.set_code(0)
|
|
|
|
#Pack ICMP payload
|
|
icmp_bytes = struct.pack('>H', id)
|
|
icmp_bytes += struct.pack('>H', sequence_number)
|
|
if (arbitrary_data is not None):
|
|
icmp_bytes += array_tobytes(array.array('B', arbitrary_data))
|
|
icmp_payload = Data()
|
|
icmp_payload.set_data(icmp_bytes)
|
|
|
|
#Link payload to header
|
|
icmp_packet.contains(icmp_payload)
|
|
|
|
return icmp_packet
|
|
|
|
|
|
############################################################################
|
|
@classmethod
|
|
def Destination_Unreachable(class_object, code, originating_packet_data = None):
|
|
unused_bytes = [0x00, 0x00, 0x00, 0x00]
|
|
return class_object.__build_error_message(ICMP6.DESTINATION_UNREACHABLE, code, unused_bytes, originating_packet_data)
|
|
|
|
@classmethod
|
|
def Packet_Too_Big(class_object, MTU, originating_packet_data = None):
|
|
MTU_bytes = struct.pack('!L', MTU)
|
|
return class_object.__build_error_message(ICMP6.PACKET_TOO_BIG, 0, MTU_bytes, originating_packet_data)
|
|
|
|
@classmethod
|
|
def Time_Exceeded(class_object, code, originating_packet_data = None):
|
|
unused_bytes = [0x00, 0x00, 0x00, 0x00]
|
|
return class_object.__build_error_message(ICMP6.TIME_EXCEEDED, code, unused_bytes, originating_packet_data)
|
|
|
|
@classmethod
|
|
def Parameter_Problem(class_object, code, pointer, originating_packet_data = None):
|
|
pointer_bytes = struct.pack('!L', pointer)
|
|
return class_object.__build_error_message(ICMP6.PARAMETER_PROBLEM, code, pointer_bytes, originating_packet_data)
|
|
|
|
@classmethod
|
|
def __build_error_message(class_object, type, code, data, originating_packet_data):
|
|
#Build ICMP6 header
|
|
icmp_packet = ICMP6()
|
|
icmp_packet.set_type(type)
|
|
icmp_packet.set_code(code)
|
|
|
|
#Pack ICMP payload
|
|
icmp_bytes = array_tobytes(array.array('B', data))
|
|
if (originating_packet_data is not None):
|
|
icmp_bytes += array_tobytes(array.array('B', originating_packet_data))
|
|
icmp_payload = Data()
|
|
icmp_payload.set_data(icmp_bytes)
|
|
|
|
#Link payload to header
|
|
icmp_packet.contains(icmp_payload)
|
|
|
|
return icmp_packet
|
|
|
|
############################################################################
|
|
|
|
@classmethod
|
|
def Neighbor_Solicitation(class_object, target_address):
|
|
return class_object.__build_neighbor_message(ICMP6.NEIGHBOR_SOLICITATION, target_address)
|
|
|
|
@classmethod
|
|
def Neighbor_Advertisement(class_object, target_address):
|
|
return class_object.__build_neighbor_message(ICMP6.NEIGHBOR_ADVERTISEMENT, target_address)
|
|
|
|
@classmethod
|
|
def __build_neighbor_message(class_object, msg_type, target_address):
|
|
#Build ICMP6 header
|
|
icmp_packet = ICMP6()
|
|
icmp_packet.set_type(msg_type)
|
|
icmp_packet.set_code(0)
|
|
|
|
# Flags + Reserved
|
|
icmp_bytes = array_tobytes(array.array('B', [0x00] * 4))
|
|
|
|
# Target Address: The IP address of the target of the solicitation.
|
|
# It MUST NOT be a multicast address.
|
|
icmp_bytes += array_tobytes(array.array('B', IP6_Address(target_address).as_bytes()))
|
|
|
|
icmp_payload = Data()
|
|
icmp_payload.set_data(icmp_bytes)
|
|
|
|
#Link payload to header
|
|
icmp_packet.contains(icmp_payload)
|
|
|
|
return icmp_packet
|
|
|
|
############################################################################
|
|
|
|
def get_target_address(self):
|
|
return IP6_Address(self.child().get_bytes()[4:20])
|
|
|
|
def set_target_address(self, target_address):
|
|
address = IP6_Address(target_address)
|
|
payload_bytes = self.child().get_bytes()
|
|
payload_bytes[4:20] = address.get_bytes()
|
|
self.child().set_bytes(payload_bytes)
|
|
|
|
# 0 1 2 3 4 5 6 7
|
|
# +-+-+-+-+-+-+-+-+
|
|
# |R|S|O|reserved |
|
|
# +-+-+-+-+-+-+-+-+
|
|
|
|
def get_neighbor_advertisement_flags(self):
|
|
return self.child().get_byte(0)
|
|
|
|
def set_neighbor_advertisement_flags(self, flags):
|
|
self.child().set_byte(0, flags)
|
|
|
|
def get_router_flag(self):
|
|
return (self.get_neighbor_advertisement_flags() & 0x80) != 0
|
|
|
|
def set_router_flag(self, flag_value):
|
|
curr_flags = self.get_neighbor_advertisement_flags()
|
|
if flag_value:
|
|
curr_flags |= 0x80
|
|
else:
|
|
curr_flags &= ~0x80
|
|
self.set_neighbor_advertisement_flags(curr_flags)
|
|
|
|
def get_solicited_flag(self):
|
|
return (self.get_neighbor_advertisement_flags() & 0x40) != 0
|
|
|
|
def set_solicited_flag(self, flag_value):
|
|
curr_flags = self.get_neighbor_advertisement_flags()
|
|
if flag_value:
|
|
curr_flags |= 0x40
|
|
else:
|
|
curr_flags &= ~0x40
|
|
self.set_neighbor_advertisement_flags(curr_flags)
|
|
|
|
def get_override_flag(self):
|
|
return (self.get_neighbor_advertisement_flags() & 0x20) != 0
|
|
|
|
def set_override_flag(self, flag_value):
|
|
curr_flags = self.get_neighbor_advertisement_flags()
|
|
if flag_value:
|
|
curr_flags |= 0x20
|
|
else:
|
|
curr_flags &= ~0x20
|
|
self.set_neighbor_advertisement_flags(curr_flags)
|
|
|
|
############################################################################
|
|
@classmethod
|
|
def Node_Information_Query(class_object, code, payload = None):
|
|
return class_object.__build_node_information_message(ICMP6.NODE_INFORMATION_QUERY, code, payload)
|
|
|
|
@classmethod
|
|
def Node_Information_Reply(class_object, code, payload = None):
|
|
return class_object.__build_node_information_message(ICMP6.NODE_INFORMATION_REPLY, code, payload)
|
|
|
|
@classmethod
|
|
def __build_node_information_message(class_object, type, code, payload = None):
|
|
#Build ICMP6 header
|
|
icmp_packet = ICMP6()
|
|
icmp_packet.set_type(type)
|
|
icmp_packet.set_code(code)
|
|
|
|
#Pack ICMP payload
|
|
qtype = 0
|
|
flags = 0
|
|
nonce = [0x00] * 8
|
|
|
|
icmp_bytes = struct.pack('>H', qtype)
|
|
icmp_bytes += struct.pack('>H', flags)
|
|
icmp_bytes += array_tobytes(array.array('B', nonce))
|
|
|
|
if payload is not None:
|
|
icmp_bytes += array_tobytes(array.array('B', payload))
|
|
|
|
icmp_payload = Data()
|
|
icmp_payload.set_data(icmp_bytes)
|
|
|
|
#Link payload to header
|
|
icmp_packet.contains(icmp_payload)
|
|
|
|
return icmp_packet
|
|
|
|
def get_qtype(self):
|
|
return self.child().get_word(0)
|
|
|
|
def set_qtype(self, qtype):
|
|
self.child().set_word(0, qtype)
|
|
|
|
def get_nonce(self):
|
|
return self.child().get_bytes()[4:12]
|
|
|
|
def set_nonce(self, nonce):
|
|
payload_bytes = self.child().get_bytes()
|
|
payload_bytes[4:12] = array.array('B', nonce)
|
|
self.child().set_bytes(payload_bytes)
|
|
|
|
# 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5
|
|
# +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|
|
# | unused |G|S|L|C|A|T|
|
|
# +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|
|
|
|
def get_flags(self):
|
|
return self.child().get_word(2)
|
|
|
|
def set_flags(self, flags):
|
|
self.child().set_word(2, flags)
|
|
|
|
def get_flag_T(self):
|
|
return (self.get_flags() & 0x0001) != 0
|
|
|
|
def set_flag_T(self, flag_value):
|
|
curr_flags = self.get_flags()
|
|
if flag_value:
|
|
curr_flags |= 0x0001
|
|
else:
|
|
curr_flags &= ~0x0001
|
|
self.set_flags(curr_flags)
|
|
|
|
def get_flag_A(self):
|
|
return (self.get_flags() & 0x0002) != 0
|
|
|
|
def set_flag_A(self, flag_value):
|
|
curr_flags = self.get_flags()
|
|
if flag_value:
|
|
curr_flags |= 0x0002
|
|
else:
|
|
curr_flags &= ~0x0002
|
|
self.set_flags(curr_flags)
|
|
|
|
def get_flag_C(self):
|
|
return (self.get_flags() & 0x0004) != 0
|
|
|
|
def set_flag_C(self, flag_value):
|
|
curr_flags = self.get_flags()
|
|
if flag_value:
|
|
curr_flags |= 0x0004
|
|
else:
|
|
curr_flags &= ~0x0004
|
|
self.set_flags(curr_flags)
|
|
|
|
def get_flag_L(self):
|
|
return (self.get_flags() & 0x0008) != 0
|
|
|
|
def set_flag_L(self, flag_value):
|
|
curr_flags = self.get_flags()
|
|
if flag_value:
|
|
curr_flags |= 0x0008
|
|
else:
|
|
curr_flags &= ~0x0008
|
|
self.set_flags(curr_flags)
|
|
|
|
def get_flag_S(self):
|
|
return (self.get_flags() & 0x0010) != 0
|
|
|
|
def set_flag_S(self, flag_value):
|
|
curr_flags = self.get_flags()
|
|
if flag_value:
|
|
curr_flags |= 0x0010
|
|
else:
|
|
curr_flags &= ~0x0010
|
|
self.set_flags(curr_flags)
|
|
|
|
def get_flag_G(self):
|
|
return (self.get_flags() & 0x0020) != 0
|
|
|
|
def set_flag_G(self, flag_value):
|
|
curr_flags = self.get_flags()
|
|
if flag_value:
|
|
curr_flags |= 0x0020
|
|
else:
|
|
curr_flags &= ~0x0020
|
|
self.set_flags(curr_flags)
|
|
|
|
def set_node_information_data(self, data):
|
|
payload_bytes = self.child().get_bytes()
|
|
payload_bytes[12:] = array.array('B', data)
|
|
self.child().set_bytes(payload_bytes)
|
|
|
|
def get_note_information_data(self):
|
|
return self.child().get_bytes()[12:]
|
|
|
|
############################################################################
|
|
def get_echo_id(self):
|
|
return self.child().get_word(0)
|
|
|
|
def get_echo_sequence_number(self):
|
|
return self.child().get_word(2)
|
|
|
|
def get_echo_arbitrary_data(self):
|
|
return self.child().get_bytes()[4:]
|
|
|
|
def get_mtu(self):
|
|
return self.child().get_long(0)
|
|
|
|
def get_parm_problem_pointer(self):
|
|
return self.child().get_long(0)
|
|
|
|
def get_originating_packet_data(self):
|
|
return self.child().get_bytes()[4:]
|