# SPDX-License-Identifier: GPL-2.0-only # This file is part of Scapy # See https://scapy.net/ for more information # Copyright (C) Philippe Biondi """ Wireless LAN according to IEEE 802.11. This file contains bindings for 802.11 layers and some usual linklayers: - PRISM - RadioTap """ import re import struct from zlib import crc32 from scapy.config import conf, crypto_validator from scapy.data import ETHER_ANY, DLT_IEEE802_11, DLT_PRISM_HEADER, \ DLT_IEEE802_11_RADIO from scapy.compat import raw, plain_str, orb, chb from scapy.packet import Packet, bind_layers, bind_top_down, NoPayload from scapy.fields import ( BitEnumField, BitField, BitMultiEnumField, ByteEnumField, ByteField, ConditionalField, FCSField, FieldLenField, FieldListField, FlagsField, IntField, LEFieldLenField, LEIntField, LELongField, LEShortEnumField, LEShortField, LESignedIntField, MayEnd, MultipleTypeField, OUIField, PacketField, PacketListField, ReversePadField, ScalingField, ShortField, StrField, StrFixedLenField, StrLenField, XByteField, XStrFixedLenField, ) from scapy.ansmachine import AnsweringMachine from scapy.plist import PacketList from scapy.layers.l2 import Ether, LLC, MACField from scapy.layers.inet import IP, TCP from scapy.error import warning, log_loading from scapy.sendrecv import sniff, sendp if conf.crypto_valid: from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives.ciphers import Cipher, algorithms try: # cryptography > 43.0 from cryptography.hazmat.decrepit.ciphers import ( algorithms as decrepit_algorithms, ) except ImportError: decrepit_algorithms = algorithms else: default_backend = Ciphers = algorithms = decrepit_algorithms = None log_loading.info("Can't import python-cryptography v1.7+. Disabled WEP decryption/encryption. (Dot11)") # noqa: E501 ######### # Prism # ######### # http://www.martin.cc/linux/prism class PrismHeader(Packet): """ iwpriv wlan0 monitor 3 """ name = "Prism header" fields_desc = [LEIntField("msgcode", 68), LEIntField("len", 144), StrFixedLenField("dev", "", 16), LEIntField("hosttime_did", 0), LEShortField("hosttime_status", 0), LEShortField("hosttime_len", 0), LEIntField("hosttime", 0), LEIntField("mactime_did", 0), LEShortField("mactime_status", 0), LEShortField("mactime_len", 0), LEIntField("mactime", 0), LEIntField("channel_did", 0), LEShortField("channel_status", 0), LEShortField("channel_len", 0), LEIntField("channel", 0), LEIntField("rssi_did", 0), LEShortField("rssi_status", 0), LEShortField("rssi_len", 0), LEIntField("rssi", 0), LEIntField("sq_did", 0), LEShortField("sq_status", 0), LEShortField("sq_len", 0), LEIntField("sq", 0), LEIntField("signal_did", 0), LEShortField("signal_status", 0), LEShortField("signal_len", 0), LESignedIntField("signal", 0), LEIntField("noise_did", 0), LEShortField("noise_status", 0), LEShortField("noise_len", 0), LEIntField("noise", 0), LEIntField("rate_did", 0), LEShortField("rate_status", 0), LEShortField("rate_len", 0), LEIntField("rate", 0), LEIntField("istx_did", 0), LEShortField("istx_status", 0), LEShortField("istx_len", 0), LEIntField("istx", 0), LEIntField("frmlen_did", 0), LEShortField("frmlen_status", 0), LEShortField("frmlen_len", 0), LEIntField("frmlen", 0), ] def answers(self, other): if isinstance(other, PrismHeader): return self.payload.answers(other.payload) else: return self.payload.answers(other) ############ # RadioTap # ############ # https://www.radiotap.org/ # Note: Radiotap alignment is crazy. See the doc: # https://www.radiotap.org/#alignment-in-radiotap # RadioTap constants _rt_present = ['TSFT', 'Flags', 'Rate', 'Channel', 'FHSS', 'dBm_AntSignal', 'dBm_AntNoise', 'Lock_Quality', 'TX_Attenuation', 'dB_TX_Attenuation', 'dBm_TX_Power', 'Antenna', 'dB_AntSignal', 'dB_AntNoise', 'RXFlags', 'TXFlags', 'b17', 'b18', 'ChannelPlus', 'MCS', 'A_MPDU', 'VHT', 'timestamp', 'HE', 'HE_MU', 'HE_MU_other_user', 'zero_length_psdu', 'L_SIG', 'TLV', 'RadiotapNS', 'VendorNS', 'Ext'] # Note: Inconsistencies with wireshark # Wireshark ignores the suggested fields, whereas we implement some of them # (some are well-used even though not accepted) # However, flags that conflicts with Wireshark are not and MUST NOT be # implemented -> b17, b18 _rt_flags = ['CFP', 'ShortPreamble', 'wep', 'fragment', 'FCS', 'pad', 'badFCS', 'ShortGI'] _rt_channelflags = ['res1', 'res2', 'res3', 'res4', 'Turbo', 'CCK', 'OFDM', '2GHz', '5GHz', 'Passive', 'Dynamic_CCK_OFDM', 'GFSK', 'GSM', 'StaticTurbo', '10MHz', '5MHz'] _rt_rxflags = ["res1", "BAD_PLCP", "res2"] _rt_txflags = ["TX_FAIL", "CTS", "RTS", "NOACK", "NOSEQ", "ORDER"] _rt_channelflags2 = ['res1', 'res2', 'res3', 'res4', 'Turbo', 'CCK', 'OFDM', '2GHz', '5GHz', 'Passive', 'Dynamic_CCK_OFDM', 'GFSK', 'GSM', 'StaticTurbo', '10MHz', '5MHz', '20MHz', '40MHz_ext_channel_above', '40MHz_ext_channel_below', 'res5', 'res6', 'res7', 'res8', 'res9'] _rt_tsflags = ['32-bit_counter', 'Accuracy', 'res1', 'res2', 'res3', 'res4', 'res5', 'res6'] _rt_knownmcs = ['MCS_bandwidth', 'MCS_index', 'guard_interval', 'HT_format', 'FEC_type', 'STBC_streams', 'Ness', 'Ness_MSB'] _rt_bandwidth = {0: "20MHz", 1: "40MHz", 2: "ht40Mhz-", 3: "ht40MHz+"} _rt_a_mpdu_flags = ['Report0Subframe', 'Is0Subframe', 'KnownLastSubframe', 'LastSubframe', 'CRCerror', 'EOFsubframe', 'KnownEOF', 'res1', 'res2', 'res3', 'res4', 'res5', 'res6', 'res7', 'res8'] _rt_vhtbandwidth = { 0: "20MHz", 1: "40MHz", 2: "40MHz", 3: "40MHz", 4: "80MHz", 5: "80MHz", 6: "80MHz", 7: "80MHz", 8: "80MHz", 9: "80MHz", 10: "80MHz", 11: "160MHz", 12: "160MHz", 13: "160MHz", 14: "160MHz", 15: "160MHz", 16: "160MHz", 17: "160MHz", 18: "160MHz", 19: "160MHz", 20: "160MHz", 21: "160MHz", 22: "160MHz", 23: "160MHz", 24: "160MHz", 25: "160MHz" } _rt_knownvht = ['STBC', 'TXOP_PS_NOT_ALLOWED', 'GuardInterval', 'SGINsysmDis', 'LDPCextraOFDM', 'Beamformed', 'Bandwidth', 'GroupID', 'PartialAID', 'res1', 'res2', 'res3', 'res4', 'res5', 'res6', 'res7'] _rt_presentvht = ['STBC', 'TXOP_PS_NOT_ALLOWED', 'GuardInterval', 'SGINsysmDis', 'LDPCextraOFDM', 'Beamformed', 'res1', 'res2'] _rt_hemuother_per_user_known = [ 'user field position', 'STA-ID', 'NSTS', 'Tx Beamforming', 'Spatial Configuration', 'MCS', 'DCM', 'Coding', ] # Radiotap utils # Note: extended presence masks are dissected pretty dumbly by # Wireshark. def _next_radiotap_extpm(pkt, lst, cur, s): """Generates the next RadioTapExtendedPresenceMask""" if cur is None or (cur.present and cur.present.Ext): st = len(lst) + (cur is not None) return lambda *args: RadioTapExtendedPresenceMask(*args, index=st) return None class RadioTapExtendedPresenceMask(Packet): """RadioTapExtendedPresenceMask should be instantiated by passing an `index=` kwarg, stating which place the item has in the list. Passing index will update the b[x] fields accordingly to the index. e.g. >>> a = RadioTapExtendedPresenceMask(present="b0+b12+b29+Ext") >>> b = RadioTapExtendedPresenceMask(index=1, present="b33+b45+b59+b62") >>> pkt = RadioTap(present="Ext", Ext=[a, b]) """ name = "RadioTap Extended presence mask" fields_desc = [FlagsField('present', None, -32, ["b%s" % i for i in range(0, 31)] + ["Ext"])] def __init__(self, _pkt=None, index=0, **kwargs): self._restart_indentation(index) Packet.__init__(self, _pkt, **kwargs) def _restart_indentation(self, index): st = index * 32 self.fields_desc[0].names = ["b%s" % (i + st) for i in range(0, 31)] + ["Ext"] # noqa: E501 def guess_payload_class(self, pay): return conf.padding_layer # This is still unimplemented in Wireshark # https://www.radiotap.org/fields/TLV.html class RadioTapTLV(Packet): fields_desc = [ LEShortEnumField("type", 0, _rt_present), LEShortField("length", None), ConditionalField( OUIField("oui", 0), lambda pkt: pkt.type == 30 # VendorNS ), ConditionalField( ByteField("subtype", 0), lambda pkt: pkt.type == 30 ), ConditionalField( LEShortField("presence_type", 0), lambda pkt: pkt.type == 30 ), ConditionalField( LEShortField("reserved", 0), lambda pkt: pkt.type == 30 ), StrLenField("data", b"", length_from=lambda pkt: pkt.length), StrLenField("pad", None, length_from=lambda pkt: -pkt.length % 4) ] def post_build(self, pkt, pay): if self.length is None: pkt = pkt[:2] + struct.pack(" %%%s.addr1%%" % ((self.__class__.__name__,) * 4)) # noqa: E501 def guess_payload_class(self, payload): if self.type == 0x02 and ( 0x08 <= self.subtype <= 0xF and self.subtype != 0xD): return Dot11QoS elif self.FCfield.protected: # When a frame is handled by encryption, the Protected Frame bit # (previously called WEP bit) is set to 1, and the Frame Body # begins with the appropriate cryptographic header. return Dot11Encrypted else: return Packet.guess_payload_class(self, payload) def answers(self, other): if isinstance(other, Dot11): if self.type == 0: # management if self.addr1.lower() != other.addr2.lower(): # check resp DA w/ req SA # noqa: E501 return 0 if (other.subtype, self.subtype) in [(0, 1), (2, 3), (4, 5)]: return 1 if self.subtype == other.subtype == 11: # auth return self.payload.answers(other.payload) elif self.type == 1: # control return 0 elif self.type == 2: # data return self.payload.answers(other.payload) elif self.type == 3: # reserved return 0 return 0 def address_meaning(self, index): """ Return the meaning of the address[index] considering the context """ if index not in [1, 2, 3, 4]: raise ValueError("Wrong index: should be [1, 2, 3, 4]") index = index - 1 if self.type == 0: # Management return _dot11_addr_meaning[0][index] elif self.type == 1: # Control if (self.type, self.subtype) == (1, 6) and self.cfe == 6: return ["RA", "NAV-SA", "NAV-DA"][index] return _dot11_addr_meaning[1][index] elif self.type == 2: # Data meaning = _dot11_addr_meaning[2][index][ self.FCfield.to_DS ][self.FCfield.from_DS] if meaning and index in [2, 3]: # Address 3-4 if isinstance(self.payload, Dot11QoS): # MSDU and Short A-MSDU if self.payload.A_MSDU_Present: meaning = "BSSID" return meaning elif self.type == 3: # Extension return _dot11_addr_meaning[3][index] return None def unwep(self, key=None, warn=1): if self.FCfield & 0x40 == 0: if warn: warning("No WEP to remove") return if isinstance(self.payload.payload, NoPayload): if key or conf.wepkey: self.payload.decrypt(key) if isinstance(self.payload.payload, NoPayload): if warn: warning("Dot11 can't be decrypted. Check conf.wepkey.") return self.FCfield &= ~0x40 self.payload = self.payload.payload class Dot11FCS(Dot11): name = "802.11-FCS" match_subclass = True fields_desc = Dot11.fields_desc + [FCSField("fcs", None, fmt="= 3: length = orb(s[1]) if length > 0 and length <= 255: self.info = s[2:2 + length] return s def post_build(self, p, pay): if self.len is None: p = p[:1] + chb(len(p) - 2) + p[2:] return p + pay # 802.11-2020 9.4.2.4 class Dot11EltDSSSet(Dot11Elt): name = "802.11 DSSS Parameter Set" match_subclass = True fields_desc = [ ByteEnumField("ID", 3, _dot11_id_enum), ByteField("len", 1), ByteField("channel", 0), ] # 802.11-2020 9.4.2.11 class Dot11EltERP(Dot11Elt): name = "802.11 ERP" match_subclass = True fields_desc = [ ByteEnumField("ID", 42, _dot11_id_enum), ByteField("len", 1), BitField("NonERP_Present", 0, 1), BitField("Use_Protection", 0, 1), BitField("Barker_Preamble_Mode", 0, 1), BitField("res", 0, 5), ] # 802.11-2020 9.4.2.24.2 class RSNCipherSuite(Packet): name = "Cipher suite" fields_desc = [ OUIField("oui", 0x000fac), ByteEnumField("cipher", 0x04, { 0x00: "Use group cipher suite", 0x01: "WEP-40", 0x02: "TKIP", 0x03: "OCB", 0x04: "CCMP-128", 0x05: "WEP-104", 0x06: "BIP-CMAC-128", 0x07: "Group addressed traffic not allowed", 0x08: "GCMP-128", 0x09: "GCMP-256", 0x0A: "CCMP-256", 0x0B: "BIP-GMAC-128", 0x0C: "BIP-GMAC-256", 0x0D: "BIP-CMAC-256" }) ] def extract_padding(self, s): return "", s # 802.11-2020 9.4.2.24.3 class AKMSuite(Packet): name = "AKM suite" fields_desc = [ OUIField("oui", 0x000fac), ByteEnumField("suite", 0x01, { 0x00: "Reserved", 0x01: "802.1X", 0x02: "PSK", 0x03: "FT-802.1X", 0x04: "FT-PSK", 0x05: "WPA-SHA256", 0x06: "PSK-SHA256", 0x07: "TDLS", 0x08: "SAE", 0x09: "FT-SAE", 0x0A: "AP-PEER-KEY", 0x0B: "WPA-SHA256-SUITE-B", 0x0C: "WPA-SHA384-SUITE-B", 0x0D: "FT-802.1X-SHA384", 0x0E: "FILS-SHA256", 0x0F: "FILS-SHA384", 0x10: "FT-FILS-SHA256", 0x11: "FT-FILS-SHA384", 0x12: "OWE" }) ] def extract_padding(self, s): return "", s # 802.11-2020 9.4.2.24.5 class PMKIDListPacket(Packet): name = "PMKIDs" fields_desc = [ LEFieldLenField("nb_pmkids", None, count_of="pmkid_list"), FieldListField( "pmkid_list", None, XStrFixedLenField("", "", length=16), count_from=lambda pkt: pkt.nb_pmkids ) ] def extract_padding(self, s): return "", s # 802.11-2020 9.4.2.24.1 class Dot11EltRSN(Dot11Elt): name = "802.11 RSN information" match_subclass = True fields_desc = [ ByteEnumField("ID", 48, _dot11_id_enum), ByteField("len", None), LEShortField("version", 1), PacketField("group_cipher_suite", RSNCipherSuite(), RSNCipherSuite), LEFieldLenField( "nb_pairwise_cipher_suites", None, count_of="pairwise_cipher_suites" ), PacketListField( "pairwise_cipher_suites", [RSNCipherSuite()], RSNCipherSuite, count_from=lambda p: p.nb_pairwise_cipher_suites ), LEFieldLenField( "nb_akm_suites", None, count_of="akm_suites" ), PacketListField( "akm_suites", [AKMSuite()], AKMSuite, count_from=lambda p: p.nb_akm_suites ), # RSN Capabilities # 802.11-2020 9.4.2.24.4 BitField("mfp_capable", 1, 1), BitField("mfp_required", 1, 1), BitField("gtksa_replay_counter", 0, 2), BitField("ptksa_replay_counter", 0, 2), BitField("no_pairwise", 0, 1), BitField("pre_auth", 0, 1), BitField("reserved", 0, 1), BitField("ocvc", 0, 1), BitField("extended_key_id", 0, 1), BitField("pbac", 0, 1), BitField("spp_a_msdu_required", 0, 1), BitField("spp_a_msdu_capable", 0, 1), BitField("peer_key_enabled", 0, 1), BitField("joint_multiband_rsna", 0, 1), # Theoretically we could use mfp_capable/mfp_required to know if those # fields are present, but some implementations poorly implement it. # In practice, do as wireshark: guess using offset. ConditionalField( PacketField("pmkids", PMKIDListPacket(), PMKIDListPacket), lambda pkt: ( True if pkt.len is None else pkt.len - ( 12 + (pkt.nb_pairwise_cipher_suites or 0) * 4 + (pkt.nb_akm_suites or 0) * 4 ) >= 2 ) ), ConditionalField( PacketField("group_management_cipher_suite", RSNCipherSuite(cipher=0x6), RSNCipherSuite), lambda pkt: ( True if pkt.len is None else pkt.len - ( 12 + (pkt.nb_pairwise_cipher_suites or 0) * 4 + (pkt.nb_akm_suites or 0) * 4 + (2 if pkt.pmkids else 0) + (pkt.pmkids and pkt.pmkids.nb_pmkids or 0) * 16 ) >= 4 ) ) ] class Dot11EltCountryConstraintTriplet(Packet): name = "802.11 Country Constraint Triplet" fields_desc = [ ByteField("first_channel_number", 1), ByteField("num_channels", 24), ByteField("mtp", 0) ] def extract_padding(self, s): return b"", s class Dot11EltCountry(Dot11Elt): name = "802.11 Country" match_subclass = True fields_desc = [ ByteEnumField("ID", 7, _dot11_id_enum), ByteField("len", None), StrFixedLenField("country_string", b"\0\0\0", length=3), MayEnd(PacketListField( "descriptors", [], Dot11EltCountryConstraintTriplet, length_from=lambda pkt: ( pkt.len - 3 - (pkt.len % 3) ) )), # When this extension is last, padding appears to be omitted ConditionalField( ByteField("pad", 0), # The length should be 3 bytes per each triplet, and 3 bytes for the # country_string field. The standard dictates that the element length # must be even, so if the result is odd, add a padding byte. # Some transmitters don't comply with the standard, so instead of assuming # the length, we test whether there is a padding byte. # Some edge cases are still not covered, for example, if the tag length # (pkt.len) is an arbitrary number. lambda pkt: ((len(pkt.descriptors) + 1) % 2) if pkt.len is None else (pkt.len % 3) # noqa: E501 ) ] class _RateField(ByteField): def i2repr(self, pkt, val): if val is None: return "" s = str((val & 0x7f) / 2.) if val & 0x80: s += "(B)" return s + " Mbps" class Dot11EltRates(Dot11Elt): name = "802.11 Rates" match_subclass = True fields_desc = [ ByteEnumField("ID", 1, _dot11_id_enum), ByteField("len", None), FieldListField( "rates", [0x82], _RateField("", 0), length_from=lambda p: p.len ) ] Dot11EltRates.register_variant(50) # Extended rates class Dot11EltHTCapabilities(Dot11Elt): name = "802.11 HT Capabilities" match_subclass = True fields_desc = [ ByteEnumField("ID", 45, _dot11_id_enum), ByteField("len", None), # HT Capabilities Info: 2B BitField("L_SIG_TXOP_Protection", 0, 1, tot_size=-2), BitField("Forty_Mhz_Intolerant", 0, 1), BitField("PSMP", 0, 1), BitField("DSSS_CCK", 0, 1), BitEnumField("Max_A_MSDU", 0, 1, {0: "3839 o", 1: "7935 o"}), BitField("Delayed_BlockAck", 0, 1), BitField("Rx_STBC", 0, 2), BitField("Tx_STBC", 0, 1), BitField("Short_GI_40Mhz", 0, 1), BitField("Short_GI_20Mhz", 0, 1), BitField("Green_Field", 0, 1), BitEnumField("SM_Power_Save", 0, 2, {0: "static SM", 1: "dynamic SM", 3: "disabled"}), BitEnumField("Supported_Channel_Width", 0, 1, {0: "20Mhz", 1: "20Mhz+40Mhz"}), BitField("LDPC_Coding_Capability", 0, 1, end_tot_size=-2), # A-MPDU Parameters: 1B BitField("res1", 0, 3, tot_size=-1), BitField("Min_MPDCU_Start_Spacing", 8, 3), BitField("Max_A_MPDU_Length_Exponent", 3, 2, end_tot_size=-1), # Supported MCS set: 16B BitField("res2", 0, 27, tot_size=-16), BitField("TX_Unequal_Modulation", 0, 1), BitField("TX_Max_Spatial_Streams", 0, 2), BitField("TX_RX_MCS_Set_Not_Equal", 0, 1), BitField("TX_MCS_Set_Defined", 0, 1), BitField("res3", 0, 6), BitField("RX_Highest_Supported_Data_Rate", 0, 10), BitField("res4", 0, 3), BitField("RX_MSC_Bitmask", 0, 77, end_tot_size=-16), # HT Extended capabilities: 2B BitField("res5", 0, 4, tot_size=-2), BitField("RD_Responder", 0, 1), BitField("HTC_HT_Support", 0, 1), BitField("MCS_Feedback", 0, 2), BitField("res6", 0, 5), BitField("PCO_Transition_Time", 0, 2), BitField("PCO", 0, 1, end_tot_size=-2), # TX Beamforming Capabilities TxBF: 4B BitField("res7", 0, 3, tot_size=-4), BitField("Channel_Estimation_Capability", 0, 2), BitField("CSI_max_n_Rows_Beamformer_Supported", 0, 2), BitField("Compressed_Steering_n_Beamformer_Antennas_Supported", 0, 2), BitField("Noncompressed_Steering_n_Beamformer_Antennas_Supported", 0, 2), BitField("CSI_n_Beamformer_Antennas_Supported", 0, 2), BitField("Minimal_Grouping", 0, 2), BitField("Explicit_Compressed_Beamforming_Feedback", 0, 2), BitField("Explicit_Noncompressed_Beamforming_Feedback", 0, 2), BitField("Explicit_Transmit_Beamforming_CSI_Feedback", 0, 2), BitField("Explicit_Compressed_Steering", 0, 1), BitField("Explicit_Noncompressed_Steering", 0, 1), BitField("Explicit_CSI_Transmit_Beamforming", 0, 1), BitField("Calibration", 0, 2), BitField("Implicit_Trasmit_Beamforming", 0, 1), BitField("Transmit_NDP", 0, 1), BitField("Receive_NDP", 0, 1), BitField("Transmit_Staggered_Sounding", 0, 1), BitField("Receive_Staggered_Sounding", 0, 1), BitField("Implicit_Transmit_Beamforming_Receiving", 0, 1, end_tot_size=-4), # ASEL Capabilities: 1B FlagsField("ASEL", 0, 8, [ "res", "Transmit_Sounding_PPDUs", "Receive_ASEL", "Antenna_Indices_Feedback", "Explicit_CSI_Feedback", "Explicit_CSI_Feedback_Based_Transmit_ASEL", "Antenna_Selection", ]) ] class Dot11EltVendorSpecific(Dot11Elt): name = "802.11 Vendor Specific" match_subclass = True fields_desc = [ ByteEnumField("ID", 221, _dot11_id_enum), ByteField("len", None), OUIField("oui", 0x000000), StrLenField("info", "", length_from=lambda x: x.len - 3) ] @classmethod def dispatch_hook(cls, _pkt=None, *args, **kargs): if _pkt: oui = struct.unpack("!I", b"\x00" + _pkt[2:5])[0] if oui == 0x0050f2: # Microsoft type_ = orb(_pkt[5]) if type_ == 0x01: # MS WPA IE return Dot11EltMicrosoftWPA elif type_ == 0x02: # MS WME IE TODO # return Dot11EltMicrosoftWME pass elif type_ == 0x04: # MS WPS IE TODO # return Dot11EltWPS pass return Dot11EltVendorSpecific return cls class Dot11EltMicrosoftWPA(Dot11EltVendorSpecific): name = "802.11 Microsoft WPA" match_subclass = True ID = 221 oui = 0x0050f2 # It appears many WPA implementations ignore the fact # that this IE should only have a single cipher and auth suite fields_desc = Dot11EltVendorSpecific.fields_desc[:3] + [ XByteField("type", 0x01) ] + Dot11EltRSN.fields_desc[2:8] # 802.11-2016 9.4.2.19 class Dot11EltCSA(Dot11Elt): name = "802.11 CSA Element" match_subclass = True fields_desc = [ ByteEnumField("ID", 37, _dot11_id_enum), ByteField("len", 3), ByteField("mode", 0), ByteField("new_channel", 0), ByteField("channel_switch_count", 0) ] # 802.11-2016 9.4.2.59 class Dot11EltOBSS(Dot11Elt): name = "802.11 OBSS Scan Parameters Element" match_subclass = True fields_desc = [ ByteEnumField("ID", 74, _dot11_id_enum), ByteField("len", 14), LEShortField("Passive_Dwell", 0), LEShortField("Active_Dwell", 0), LEShortField("Scan_Interval", 0), LEShortField("Passive_Total_Per_Channel", 0), LEShortField("Active_Total_Per_Channel", 0), LEShortField("Delay", 0), LEShortField("Activity_Threshold", 0), ] # 802.11-2016 9.4.2.159 class Dot11VHTOperationInfo(Packet): name = "802.11 VHT Operation Information" fields_desc = [ ByteField("channel_width", 0), ByteField("channel_center0", 36), ByteField("channel_center1", 0), ] def extract_padding(self, s): return "", s class Dot11EltVHTOperation(Dot11Elt): name = "802.11 VHT Operation Element" match_subclass = True fields_desc = [ ByteEnumField("ID", 192, _dot11_id_enum), ByteField("len", 5), PacketField( "VHT_Operation_Info", Dot11VHTOperationInfo(), Dot11VHTOperationInfo ), FieldListField( "mcs_set", [0x00], BitField('SS', 0x00, size=2), count_from=lambda x: 8 ) ] ###################### # 802.11 Frame types # ###################### # 802.11-2016 9.3 class Dot11Beacon(_Dot11EltUtils): name = "802.11 Beacon" fields_desc = [LELongField("timestamp", 0), LEShortField("beacon_interval", 0x0064), FlagsField("cap", 0, 16, capability_list)] class Dot11ATIM(Packet): name = "802.11 ATIM" class Dot11Disas(Packet): name = "802.11 Disassociation" fields_desc = [LEShortEnumField("reason", 1, reason_code)] class Dot11AssoReq(_Dot11EltUtils): name = "802.11 Association Request" fields_desc = [FlagsField("cap", 0, 16, capability_list), LEShortField("listen_interval", 0x00c8)] class Dot11AssoResp(_Dot11EltUtils): name = "802.11 Association Response" fields_desc = [FlagsField("cap", 0, 16, capability_list), LEShortField("status", 0), LEShortField("AID", 0)] class Dot11ReassoReq(_Dot11EltUtils): name = "802.11 Reassociation Request" fields_desc = [FlagsField("cap", 0, 16, capability_list), LEShortField("listen_interval", 0x00c8), MACField("current_AP", ETHER_ANY)] class Dot11ReassoResp(Dot11AssoResp): name = "802.11 Reassociation Response" class Dot11ProbeReq(_Dot11EltUtils): name = "802.11 Probe Request" class Dot11ProbeResp(_Dot11EltUtils): name = "802.11 Probe Response" fields_desc = [LELongField("timestamp", 0), LEShortField("beacon_interval", 0x0064), FlagsField("cap", 0, 16, capability_list)] class Dot11Auth(_Dot11EltUtils): name = "802.11 Authentication" fields_desc = [LEShortEnumField("algo", 0, ["open", "sharedkey"]), LEShortField("seqnum", 0), LEShortEnumField("status", 0, status_code)] def answers(self, other): if self.algo != other.algo: return 0 if ( self.seqnum == other.seqnum + 1 or (self.algo == 3 and self.seqnum == other.seqnum) ): return 1 return 0 class Dot11Deauth(Packet): name = "802.11 Deauthentication" fields_desc = [LEShortEnumField("reason", 1, reason_code)] class Dot11Ack(Packet): name = "802.11 Ack packet" # 802.11-2016 9.4.1.11 class Dot11Action(Packet): name = "802.11 Action" fields_desc = [ ByteEnumField("category", 0x00, { 0x00: "Spectrum Management", 0x01: "QoS", 0x02: "DLS", 0x03: "Block", 0x04: "Public", 0x05: "Radio Measurement", 0x06: "Fast BSS Transition", 0x07: "HT", 0x08: "SA Query", 0x09: "Protected Dual of Public Action", 0x0A: "WNM", 0x0B: "Unprotected WNM", 0x0C: "TDLS", 0x0D: "Mesh", 0x0E: "Multihop", 0x0F: "Self-protected", 0x10: "DMG", 0x11: "Reserved Wi-Fi Alliance", 0x12: "Fast Session Transfer", 0x13: "Robust AV Streaming", 0x14: "Unprotected DMG", 0x15: "VHT" }) ] # 802.11-2016 9.6.14.1 class Dot11WNM(Packet): name = "802.11 WNM Action" fields_desc = [ ByteEnumField("action", 0x00, { 0x00: "Event Request", 0x01: "Event Report", 0x02: "Diagnostic Request", 0x03: "Diagnostic Report", 0x04: "Location Configuration Request", 0x05: "Location Configuration Response", 0x06: "BSS Transition Management Query", 0x07: "BSS Transition Management Request", 0x08: "BSS Transition Management Response", 0x09: "FMS Request", 0x0A: "FMS Response", 0x0B: "Collocated Interference Request", 0x0C: "Collocated Interference Report", 0x0D: "TFS Request", 0x0E: "TFS Response", 0x0F: "TFS Notify", 0x10: "WNM Sleep Mode Request", 0x11: "WNM Sleep Mode Response", 0x12: "TIM Broadcast Request", 0x13: "TIM Broadcast Response", 0x14: "QoS Traffic Capability Update", 0x15: "Channel Usage Request", 0x16: "Channel Usage Response", 0x17: "DMS Request", 0x18: "DMS Response", 0x19: "Timing Measurement Request", 0x1A: "WNM Notification Request", 0x1B: "WNM Notification Response", 0x1C: "WNM-Notify Response" }) ] # 802.11-2016 9.4.2.37 class SubelemTLV(Packet): fields_desc = [ ByteField("type", 0), LEFieldLenField("len", None, fmt="B", length_of="value"), FieldListField( "value", [], ByteField('', 0), length_from=lambda p: p.len ) ] class BSSTerminationDuration(Packet): name = "BSS Termination Duration" fields_desc = [ ByteField("id", 4), ByteField("len", 10), LELongField("TSF", 0), LEShortField("duration", 0) ] def extract_padding(self, s): return "", s class NeighborReport(Packet): name = "Neighbor Report" fields_desc = [ ByteField("type", 0), ByteField("len", 13), MACField("BSSID", ETHER_ANY), # BSSID Information BitField("AP_reach", 0, 2, tot_size=-4), BitField("security", 0, 1), BitField("key_scope", 0, 1), BitField("capabilities", 0, 6), BitField("mobility", 0, 1), BitField("HT", 0, 1), BitField("VHT", 0, 1), BitField("FTM", 0, 1), BitField("reserved", 0, 18, end_tot_size=-4), # BSSID Information end ByteField("op_class", 0), ByteField("channel", 0), ByteField("phy_type", 0), ConditionalField( PacketListField( "subelems", SubelemTLV(), SubelemTLV, length_from=lambda p: p.len - 13 ), lambda p: p.len > 13 ) ] # 802.11-2016 9.6.14.9 btm_request_mode = [ "Preferred_Candidate_List_Included", "Abridged", "Disassociation_Imminent", "BSS_Termination_Included", "ESS_Disassociation_Imminent" ] class Dot11BSSTMRequest(Packet): name = "BSS Transition Management Request" fields_desc = [ ByteField("token", 0), FlagsField("mode", 0, 8, btm_request_mode), LEShortField("disassociation_timer", 0), ByteField("validity_interval", 0), ConditionalField( PacketField( "termination_duration", BSSTerminationDuration(), BSSTerminationDuration ), lambda p: p.mode and p.mode.BSS_Termination_Included ), ConditionalField( ByteField("url_len", 0), lambda p: p.mode and p.mode.ESS_Disassociation_Imminent ), ConditionalField( StrLenField("url", "", length_from=lambda p: p.url_len), lambda p: p.mode and p.mode.ESS_Disassociation_Imminent != 0 ), ConditionalField( PacketListField( "neighbor_report", NeighborReport(), NeighborReport ), lambda p: p.mode and p.mode.Preferred_Candidate_List_Included ) ] # 802.11-2016 9.6.14.10 btm_status_code = [ "Accept", "Reject-Unspecified_reject_reason", "Reject-Insufficient_Beacon_or_Probe_Response_frames", "Reject-Insufficient_available_capacity_from_all_candidates", "Reject-BSS_termination_undesired", "Reject-BSS_termination_delay_requested", "Reject-STA_BSS_Transition_Candidate_List_provided", "Reject-No_suitable_BSS_transition_candidates", "Reject-Leaving_ESS" ] class Dot11BSSTMResponse(Packet): name = "BSS Transition Management Response" fields_desc = [ ByteField("token", 0), ByteEnumField("status", 0, btm_status_code), ByteField("termination_delay", 0), ConditionalField( MACField("target", ETHER_ANY), lambda p: p.status == 0 ), ConditionalField( PacketListField( "neighbor_report", NeighborReport(), NeighborReport ), lambda p: p.status == 6 ) ] # 802.11-2016 9.6.2.1 class Dot11SpectrumManagement(Packet): name = "802.11 Spectrum Management Action" fields_desc = [ ByteEnumField("action", 0x00, { 0x00: "Measurement Request", 0x01: "Measurement Report", 0x02: "TPC Request", 0x03: "TPC Report", 0x04: "Channel Switch Announcement", }) ] # 802.11-2016 9.6.2.6 class Dot11CSA(Packet): name = "Channel Switch Announcement Frame" fields_desc = [ PacketField("CSA", Dot11EltCSA(), Dot11EltCSA), ] ################### # 802.11 Security # ################### # 802.11-2016 12 class Dot11Encrypted(Packet): name = "802.11 Encrypted (unknown algorithm)" fields_desc = [StrField("data", None)] @classmethod def dispatch_hook(cls, _pkt=None, *args, **kargs): # Extracted from # https://github.com/wireshark/wireshark/blob/master/epan/dissectors/packet-ieee80211.c # noqa: E501 KEY_EXTIV = 0x20 EXTIV_LEN = 8 if _pkt and len(_pkt) >= 3: if (orb(_pkt[3]) & KEY_EXTIV) and (len(_pkt) >= EXTIV_LEN): if orb(_pkt[1]) == ((orb(_pkt[0]) | 0x20) & 0x7f): # IS_TKIP return Dot11TKIP elif orb(_pkt[2]) == 0: # IS_CCMP return Dot11CCMP else: # Unknown encryption algorithm return Dot11Encrypted else: return Dot11WEP return conf.raw_layer # 802.11-2016 12.3.2 class Dot11WEP(Dot11Encrypted): name = "802.11 WEP packet" fields_desc = [StrFixedLenField("iv", b"\0\0\0", 3), ByteField("keyid", 0), StrField("wepdata", None, remain=4), IntField("icv", None)] def decrypt(self, key=None): if key is None: key = conf.wepkey if key and conf.crypto_valid: d = Cipher( decrepit_algorithms.ARC4(self.iv + key.encode("utf8")), None, default_backend(), ).decryptor() self.add_payload(LLC(d.update(self.wepdata) + d.finalize())) def post_dissect(self, s): self.decrypt() def build_payload(self): if self.wepdata is None: return Packet.build_payload(self) return b"" @crypto_validator def encrypt(self, p, pay, key=None): if key is None: key = conf.wepkey if key: if self.icv is None: pay += struct.pack(" LE = reversed order BitField("res", 0, 5), # # ext_iv - 4 bytes ConditionalField(ByteField("TSC2", 0), lambda pkt: pkt.ext_iv), ConditionalField(ByteField("TSC3", 0), lambda pkt: pkt.ext_iv), ConditionalField(ByteField("TSC4", 0), lambda pkt: pkt.ext_iv), ConditionalField(ByteField("TSC5", 0), lambda pkt: pkt.ext_iv), # data StrField("data", None), ] # 802.11-2016 12.5.3.2 class Dot11CCMP(Dot11Encrypted): name = "802.11 CCMP packet" fields_desc = [ # iv - 8 bytes ByteField("PN0", 0), ByteField("PN1", 0), ByteField("res0", 0), BitField("key_id", 0, 2), # BitField("ext_iv", 0, 1), # => LE = reversed order BitField("res1", 0, 5), # ByteField("PN2", 0), ByteField("PN3", 0), ByteField("PN4", 0), ByteField("PN5", 0), # data StrField("data", None), ] ############ # Bindings # ############ bind_top_down(RadioTap, Dot11FCS, present=2, Flags=16) bind_top_down(Dot11, Dot11QoS, type=2, subtype=0xc) bind_layers(PrismHeader, Dot11,) bind_layers(Dot11, LLC, type=2) bind_layers(Dot11QoS, LLC,) # 802.11-2016 9.2.4.1.3 Type and Subtype subfields bind_layers(Dot11, Dot11AssoReq, subtype=0, type=0) bind_layers(Dot11, Dot11AssoResp, subtype=1, type=0) bind_layers(Dot11, Dot11ReassoReq, subtype=2, type=0) bind_layers(Dot11, Dot11ReassoResp, subtype=3, type=0) bind_layers(Dot11, Dot11ProbeReq, subtype=4, type=0) bind_layers(Dot11, Dot11ProbeResp, subtype=5, type=0) bind_layers(Dot11, Dot11Beacon, subtype=8, type=0) bind_layers(Dot11, Dot11ATIM, subtype=9, type=0) bind_layers(Dot11, Dot11Disas, subtype=10, type=0) bind_layers(Dot11, Dot11Auth, subtype=11, type=0) bind_layers(Dot11, Dot11Deauth, subtype=12, type=0) bind_layers(Dot11, Dot11Action, subtype=13, type=0) bind_layers(Dot11, Dot11Ack, subtype=13, type=1) bind_layers(Dot11Beacon, Dot11Elt,) bind_layers(Dot11AssoReq, Dot11Elt,) bind_layers(Dot11AssoResp, Dot11Elt,) bind_layers(Dot11ReassoReq, Dot11Elt,) bind_layers(Dot11ReassoResp, Dot11Elt,) bind_layers(Dot11ProbeReq, Dot11Elt,) bind_layers(Dot11ProbeResp, Dot11Elt,) bind_layers(Dot11Auth, Dot11Elt,) bind_layers(Dot11Elt, Dot11Elt,) bind_layers(Dot11TKIP, conf.raw_layer) bind_layers(Dot11CCMP, conf.raw_layer) bind_layers(Dot11Action, Dot11SpectrumManagement, category=0x00) bind_layers(Dot11SpectrumManagement, Dot11CSA, action=4) bind_layers(Dot11Action, Dot11WNM, category=0x0A) bind_layers(Dot11WNM, Dot11BSSTMRequest, action=7) bind_layers(Dot11WNM, Dot11BSSTMResponse, action=8) conf.l2types.register(DLT_IEEE802_11, Dot11) conf.l2types.register_num2layer(801, Dot11) conf.l2types.register(DLT_PRISM_HEADER, PrismHeader) conf.l2types.register_num2layer(802, PrismHeader) conf.l2types.register(DLT_IEEE802_11_RADIO, RadioTap) conf.l2types.register_num2layer(803, RadioTap) #################### # Other WiFi utils # #################### class WiFi_am(AnsweringMachine): """Before using this, initialize "iffrom" and "ifto" interfaces: iwconfig iffrom mode monitor iwpriv orig_ifto hostapd 1 ifconfig ifto up note: if ifto=wlan0ap then orig_ifto=wlan0 note: ifto and iffrom must be set on the same channel ex: ifconfig eth1 up iwconfig eth1 mode monitor iwconfig eth1 channel 11 iwpriv wlan0 hostapd 1 ifconfig wlan0ap up iwconfig wlan0 channel 11 iwconfig wlan0 essid dontexist iwconfig wlan0 mode managed """ function_name = "airpwn" filter = None def parse_options(self, iffrom=conf.iface, ifto=conf.iface, replace="", pattern="", ignorepattern=""): self.iffrom = iffrom self.ifto = ifto self.ptrn = re.compile(pattern.encode()) self.iptrn = re.compile(ignorepattern.encode()) self.replace = replace def is_request(self, pkt): if not isinstance(pkt, Dot11): return 0 if not pkt.FCfield & 1: return 0 if not pkt.haslayer(TCP): return 0 tcp = pkt.getlayer(TCP) pay = raw(tcp.payload) if not self.ptrn.match(pay): return 0 if self.iptrn.match(pay) is True: return 0 return True def make_reply(self, p): ip = p.getlayer(IP) tcp = p.getlayer(TCP) pay = raw(tcp.payload) p[IP].underlayer.remove_payload() p.FCfield = "from-DS" p.addr1, p.addr2 = p.addr2, p.addr1 p /= IP(src=ip.dst, dst=ip.src) p /= TCP(sport=tcp.dport, dport=tcp.sport, seq=tcp.ack, ack=tcp.seq + len(pay), flags="PA") q = p.copy() p /= self.replace q.ID += 1 q.getlayer(TCP).flags = "RA" q.getlayer(TCP).seq += len(self.replace) return [p, q] def print_reply(self, query, *reply): p = reply[0][0] print(p.sprintf("Sent %IP.src%:%IP.sport% > %IP.dst%:%TCP.dport%")) def send_reply(self, reply): sendp(reply, iface=self.ifto, **self.optsend) def sniff(self): sniff(iface=self.iffrom, **self.optsniff) conf.stats_dot11_protocols += [Dot11WEP, Dot11Beacon, ] class Dot11PacketList(PacketList): def __init__(self, res=None, name="Dot11List", stats=None): if stats is None: stats = conf.stats_dot11_protocols PacketList.__init__(self, res, name, stats) def toEthernet(self): data = [x[Dot11] for x in self.res if Dot11 in x and x.type == 2] r2 = [] for p in data: q = p.copy() q.unwep() r2.append(Ether() / q.payload.payload.payload) # Dot11/LLC/SNAP/IP return PacketList(r2, name="Ether from %s" % self.listname)