## This file is part of Scapy
|
## See http://www.secdev.org/projects/scapy for more informations
|
## Copyright (C) Philippe Biondi <phil@secdev.org>
|
## This program is published under a GPLv2 license
|
|
"""
|
Wireless LAN according to IEEE 802.11.
|
"""
|
|
from __future__ import print_function
|
import re,struct
|
from zlib import crc32
|
|
from scapy.config import conf, crypto_validator
|
from scapy.data import *
|
from scapy.compat import *
|
from scapy.packet import *
|
from scapy.fields import *
|
from scapy.ansmachine import *
|
from scapy.plist import PacketList
|
from scapy.layers.l2 import *
|
from scapy.layers.inet import IP, TCP
|
from scapy.error import warning
|
|
|
if conf.crypto_valid:
|
from cryptography.hazmat.backends import default_backend
|
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms
|
else:
|
default_backend = Ciphers = algorithms = None
|
log_loading.info("Can't import python-cryptography v1.7+. Disabled WEP decryption/encryption.")
|
|
|
### Layers
|
|
|
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)
|
|
class RadioTap(Packet):
|
name = "RadioTap dummy"
|
fields_desc = [ ByteField('version', 0),
|
ByteField('pad', 0),
|
FieldLenField('len', None, 'notdecoded', '<H', adjust=lambda pkt,x:x+8),
|
FlagsField('present', None, -32, ['TSFT','Flags','Rate','Channel','FHSS','dBm_AntSignal',
|
'dBm_AntNoise','Lock_Quality','TX_Attenuation','dB_TX_Attenuation',
|
'dBm_TX_Power', 'Antenna', 'dB_AntSignal', 'dB_AntNoise',
|
'b14', 'b15','b16','b17','b18','b19','b20','b21','b22','b23',
|
'b24','b25','b26','b27','b28','b29','b30','Ext']),
|
StrLenField('notdecoded', "", length_from= lambda pkt:pkt.len-8) ]
|
|
class PPI(Packet):
|
name = "Per-Packet Information header (partial)"
|
fields_desc = [ ByteField("version", 0),
|
ByteField("flags", 0),
|
FieldLenField("len", None, fmt="<H", length_of="notdecoded", adjust=lambda pkt,x:x+8),
|
LEIntField("dlt", 0),
|
StrLenField("notdecoded", "", length_from = lambda pkt:pkt.len-8)
|
]
|
|
|
class Dot11(Packet):
|
name = "802.11"
|
fields_desc = [
|
BitField("subtype", 0, 4),
|
BitEnumField("type", 0, 2, ["Management", "Control", "Data",
|
"Reserved"]),
|
BitField("proto", 0, 2),
|
FlagsField("FCfield", 0, 8, ["to-DS", "from-DS", "MF", "retry",
|
"pw-mgt", "MD", "wep", "order"]),
|
ShortField("ID",0),
|
MACField("addr1", ETHER_ANY),
|
ConditionalField(
|
MACField("addr2", ETHER_ANY),
|
lambda pkt: (pkt.type != 1 or
|
pkt.subtype in [0x9, 0xb, 0xa, 0xe, 0xf]),
|
),
|
ConditionalField(
|
MACField("addr3", ETHER_ANY),
|
lambda pkt: pkt.type in [0, 2],
|
),
|
ConditionalField(LEShortField("SC", 0), lambda pkt: pkt.type != 1),
|
ConditionalField(
|
MACField("addr4", ETHER_ANY),
|
lambda pkt: (pkt.type == 2 and
|
pkt.FCfield & 3 == 3), ## from-DS+to-DS
|
),
|
]
|
def mysummary(self):
|
return self.sprintf("802.11 %Dot11.type% %Dot11.subtype% %Dot11.addr2% > %Dot11.addr1%")
|
def guess_payload_class(self, payload):
|
if self.type == 0x02 and (0x08 <= self.subtype <= 0xF and self.subtype != 0xD):
|
return Dot11QoS
|
elif self.FCfield & 0x40:
|
return Dot11WEP
|
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
|
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 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 Dot11QoS(Packet):
|
name = "802.11 QoS"
|
fields_desc = [ BitField("Reserved",None,1),
|
BitField("Ack Policy",None,2),
|
BitField("EOSP",None,1),
|
BitField("TID",None,4),
|
ByteField("TXOP",None) ]
|
def guess_payload_class(self, payload):
|
if isinstance(self.underlayer, Dot11):
|
if self.underlayer.FCfield & 0x40:
|
return Dot11WEP
|
return Packet.guess_payload_class(self, payload)
|
|
|
capability_list = [ "res8", "res9", "short-slot", "res11",
|
"res12", "DSSS-OFDM", "res14", "res15",
|
"ESS", "IBSS", "CFP", "CFP-req",
|
"privacy", "short-preamble", "PBCC", "agility"]
|
|
reason_code = {0:"reserved",1:"unspec", 2:"auth-expired",
|
3:"deauth-ST-leaving",
|
4:"inactivity", 5:"AP-full", 6:"class2-from-nonauth",
|
7:"class3-from-nonass", 8:"disas-ST-leaving",
|
9:"ST-not-auth"}
|
|
status_code = {0:"success", 1:"failure", 10:"cannot-support-all-cap",
|
11:"inexist-asso", 12:"asso-denied", 13:"algo-unsupported",
|
14:"bad-seq-num", 15:"challenge-failure",
|
16:"timeout", 17:"AP-full",18:"rate-unsupported" }
|
|
class Dot11Beacon(Packet):
|
name = "802.11 Beacon"
|
fields_desc = [ LELongField("timestamp", 0),
|
LEShortField("beacon_interval", 0x0064),
|
FlagsField("cap", 0, 16, capability_list) ]
|
|
|
class Dot11Elt(Packet):
|
name = "802.11 Information Element"
|
fields_desc = [ ByteEnumField("ID", 0, {0:"SSID", 1:"Rates", 2: "FHset", 3:"DSset", 4:"CFset", 5:"TIM", 6:"IBSSset", 16:"challenge",
|
42:"ERPinfo", 46:"QoS Capability", 47:"ERPinfo", 48:"RSNinfo", 50:"ESRates",221:"vendor",68:"reserved"}),
|
FieldLenField("len", None, "info", "B"),
|
StrLenField("info", "", length_from=lambda x:x.len) ]
|
def mysummary(self):
|
if self.ID == 0:
|
ssid = repr(self.info)
|
if ssid[:2] in ['b"', "b'"]:
|
ssid = ssid[1:]
|
return "SSID=%s" % ssid, [Dot11]
|
else:
|
return ""
|
|
class Dot11ATIM(Packet):
|
name = "802.11 ATIM"
|
|
class Dot11Disas(Packet):
|
name = "802.11 Disassociation"
|
fields_desc = [ LEShortEnumField("reason", 1, reason_code) ]
|
|
class Dot11AssoReq(Packet):
|
name = "802.11 Association Request"
|
fields_desc = [ FlagsField("cap", 0, 16, capability_list),
|
LEShortField("listen_interval", 0x00c8) ]
|
|
|
class Dot11AssoResp(Packet):
|
name = "802.11 Association Response"
|
fields_desc = [ FlagsField("cap", 0, 16, capability_list),
|
LEShortField("status", 0),
|
LEShortField("AID", 0) ]
|
|
class Dot11ReassoReq(Packet):
|
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(Packet):
|
name = "802.11 Probe Request"
|
|
class Dot11ProbeResp(Packet):
|
name = "802.11 Probe Response"
|
fields_desc = [ LELongField("timestamp", 0),
|
LEShortField("beacon_interval", 0x0064),
|
FlagsField("cap", 0, 16, capability_list) ]
|
|
class Dot11Auth(Packet):
|
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.seqnum == other.seqnum+1:
|
return 1
|
return 0
|
|
class Dot11Deauth(Packet):
|
name = "802.11 Deauthentication"
|
fields_desc = [ LEShortEnumField("reason", 1, reason_code) ]
|
|
|
|
class Dot11WEP(Packet):
|
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) ]
|
|
@crypto_validator
|
def decrypt(self, key=None):
|
if key is None:
|
key = conf.wepkey
|
if key:
|
d = Cipher(
|
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("<I", crc32(pay) & 0xffffffff)
|
icv = b""
|
else:
|
icv = p[4:8]
|
e = Cipher(
|
algorithms.ARC4(self.iv + key.encode("utf8")),
|
None,
|
default_backend(),
|
).encryptor()
|
return p[:4] + e.update(pay) + e.finalize() + icv
|
else:
|
warning("No WEP key set (conf.wepkey).. strange results expected..")
|
return b""
|
|
def post_build(self, p, pay):
|
if self.wepdata is None:
|
p = self.encrypt(p, raw(pay))
|
return p
|
|
|
class Dot11Ack(Packet):
|
name = "802.11 Ack packet"
|
|
|
bind_layers( PrismHeader, Dot11, )
|
bind_layers( RadioTap, Dot11, )
|
bind_layers( PPI, Dot11, dlt=105)
|
bind_layers( Dot11, LLC, type=2)
|
bind_layers( Dot11QoS, LLC, )
|
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, 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, )
|
|
|
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)
|
conf.l2types.register(DLT_PPI, PPI)
|
|
|
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
|
ip = pkt.getlayer(IP)
|
tcp = pkt.getlayer(TCP)
|
pay = raw(tcp.payload)
|
if not self.ptrn.match(pay):
|
return 0
|
if self.iptrn.match(pay) == True:
|
return 0
|
return True
|
|
def make_reply(self, p):
|
ip = p.getlayer(IP)
|
tcp = p.getlayer(TCP)
|
pay = raw(tcp.payload)
|
del(p.payload.payload.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)
|
|
|
