Amazing-Mirror/fragattacks
1
0
Fork 0
mirror of https://github.com/vanhoefm/fragattacks.git synced 2024-11-28 10:18:21 -05:00
Code Issues Projects Releases Wiki Activity

fragattacks: do not use format strings

Browse Source 
There are only supported on Python 3.6 and above. With openwifi we
likely need to support an older Python version.

This patch may be reverted in the future once support for older
Python versions is no longer needed.
This commit is contained in:
Mathy Vanhoef 2022-10-16 18:44:08 +02:00
parent 934878c386
commit b7a520637e
5 changed files with 69 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
research/fragattack.py
View File

@ -175,9 +175,9 @@ def get_expected_scapy_ver():
return None return None
if __name__ == "__main__": if __name__ == "__main__":
log(STATUS, f"This is FragAttack version {FRAGVERSION}.") log(STATUS, "This is FragAttack version {}.".format(FRAGVERSION))
parser = argparse.ArgumentParser(description=f"Test for fragmentation vulnerabilities (version {FRAGVERSION}).") parser = argparse.ArgumentParser(description="Test for fragmentation vulnerabilities (version {}).".format(FRAGVERSION))
parser.add_argument('iface', help="Interface to use for the tests.") parser.add_argument('iface', help="Interface to use for the tests.")
parser.add_argument('testname', help="Name or identifier of the test to run.") parser.add_argument('testname', help="Name or identifier of the test to run.")
parser.add_argument('actions', nargs='?', help="Optional textual descriptions of actions") parser.add_argument('actions', nargs='?', help="Optional textual descriptions of actions")
@ -224,7 +224,7 @@ if __name__ == "__main__":
# Check if we're using the expected scapy version # Check if we're using the expected scapy version
expected_ver = get_expected_scapy_ver() expected_ver = get_expected_scapy_ver()
if expected_ver!= None and scapy.VERSION != expected_ver: if expected_ver!= None and scapy.VERSION != expected_ver:
log(WARNING, f"You are using scapy version {scapy.VERSION} instead of the expected {expected_ver}") log(WARNING, "You are using scapy version {} instead of the expected {}".format(scapy.VERSION, expected_ver))
log(WARNING, "Are you executing the script from inside the correct python virtual environment?") log(WARNING, "Are you executing the script from inside the correct python virtual environment?")
# Default value for options that should not be command line parameters # Default value for options that should not be command line parameters
@ -234,7 +234,7 @@ if __name__ == "__main__":
options.ptype = args2ptype(options) options.ptype = args2ptype(options)
options.as_msdu = args2msdu(options) options.as_msdu = args2msdu(options)
if options.pn_per_qos and options.no_qos: if options.pn_per_qos and options.no_qos:
log(STATUS, f"Cannot specify option --pn-per-qos and --no-qos simultaneously.") log(STATUS, "Cannot specify option --pn-per-qos and --no-qos simultaneously.")
quit(1) quit(1)
# Make the --inject-test-postauth flags easier to check # Make the --inject-test-postauth flags easier to check
@ -258,7 +258,7 @@ if __name__ == "__main__":
# Construct the test # Construct the test
options.test = prepare_tests(options) options.test = prepare_tests(options)
if options.test == None: if options.test == None:
log(STATUS, f"Test name '{options.testname}' not recognized. Specify a valid test case.") log(STATUS, "Test name '{}' not recognized. Specify a valid test case.".format(options.testname))
quit(1) quit(1)
# Parse remaining options # Parse remaining options

62
research/fraginternals.py
View File

@ -188,7 +188,7 @@ class Action():
def __str__(self): def __str__(self):
trigger = ["NoTigger", "StartAuth", "BeforeAuth", "AfterAuth", "Connected"][self.trigger] trigger = ["NoTigger", "StartAuth", "BeforeAuth", "AfterAuth", "Connected"][self.trigger]
action = ["NoAction", "GetIp", "Rekey", "Reconnect", "Roam", "Inject", "Func"][self.action] action = ["NoAction", "GetIp", "Rekey", "Reconnect", "Roam", "Inject", "Func"][self.action]
return f"Action({trigger}, {action})" return "Action({}, {})".format(trigger, action)
def __repr__(self): def __repr__(self):
return str(self) return str(self)
@ -469,7 +469,7 @@ class Station():
return encrypted, key return encrypted, key
def handle_connecting(self, bss): def handle_connecting(self, bss):
log(STATUS, f"Station: setting BSS MAC address {bss}") log(STATUS, "Station: setting BSS MAC address {}".format(bss))
self.bss = bss self.bss = bss
# Clear the keys on a new connection # Clear the keys on a new connection
@ -572,7 +572,7 @@ class Station():
elif act.action == Action.Inject: elif act.action == Action.Inject:
if act.delay != None and act.delay > 0: if act.delay != None and act.delay > 0:
log(STATUS, f"Sleeping {act.delay} seconds") log(STATUS, "Sleeping {} seconds".format(act.delay))
time.sleep(act.delay) time.sleep(act.delay)
if act.encrypted: if act.encrypted:
@ -721,7 +721,7 @@ class Daemon(metaclass=abc.ABCMeta):
log(ERROR, "wpa_supplicant did not recognize the command %s. Did you (re)compile wpa_supplicant/hostapd?" % cmd.split()[0]) log(ERROR, "wpa_supplicant did not recognize the command %s. Did you (re)compile wpa_supplicant/hostapd?" % cmd.split()[0])
quit(1) quit(1)
elif "FAIL" in response: elif "FAIL" in response:
log(ERROR, f"Failed to execute command {cmd}") log(ERROR, "Failed to execute command {}".format(cmd))
quit(1) quit(1)
return response[2:] return response[2:]
@ -746,8 +746,8 @@ class Daemon(metaclass=abc.ABCMeta):
time.sleep(5) time.sleep(5)
elif FRAGVERSION != open("/sys/module/mac80211/parameters/fragattack_version").read().strip(): elif FRAGVERSION != open("/sys/module/mac80211/parameters/fragattack_version").read().strip():
version = open("/sys/module/mac80211/parameters/fragattack_version").read().strip() version = open("/sys/module/mac80211/parameters/fragattack_version").read().strip()
log(ERROR, f"This script has version {FRAGVERSION} but the modified drivers are version {version}.") log(ERROR, "This script has version {} but the modified drivers are version {}.".format(FRAGVERSION, version))
log(ERROR, f"Recompile and reinstall the modified drivers or add --no-drivercheck (see the README for details).") log(ERROR, "Recompile and reinstall the modified drivers or add --no-drivercheck (see the README for details).")
quit(1) quit(1)
# 1. Assign/create interfaces according to provided options # 1. Assign/create interfaces according to provided options
@ -758,7 +758,7 @@ class Daemon(metaclass=abc.ABCMeta):
set_macaddress(self.nic_iface, get_macaddress(self.nic_mon)) set_macaddress(self.nic_iface, get_macaddress(self.nic_mon))
if not self.options.ap: if not self.options.ap:
log(WARNING, f"Note: you must manually set {self.nic_mon} on the channel of the AP") log(WARNING, "Note: you must manually set {} on the channel of the AP".format(self.nic_mon))
elif self.options.inject: elif self.options.inject:
# Use the provided interface to monitor/inject frames # Use the provided interface to monitor/inject frames
@ -784,7 +784,7 @@ class Daemon(metaclass=abc.ABCMeta):
elif driver in ["ath9k_htc", "iwlwifi"]: elif driver in ["ath9k_htc", "iwlwifi"]:
# Assure that fragmented frames are reliably injected on certain iwlwifi and ath9k_htc devices # Assure that fragmented frames are reliably injected on certain iwlwifi and ath9k_htc devices
self.options.inject_mf_workaround = True self.options.inject_mf_workaround = True
log(STATUS, f"Detected {driver}, using injection bug workarounds") log(STATUS, "Detected {}, using injection bug workarounds".format(driver))
# 2.B Check if ath9k_htc is using patched firmware # 2.B Check if ath9k_htc is using patched firmware
if not self.options.no_drivercheck and driver == "ath9k_htc": if not self.options.no_drivercheck and driver == "ath9k_htc":
@ -799,7 +799,7 @@ class Daemon(metaclass=abc.ABCMeta):
# 3. Enable monitor mode # 3. Enable monitor mode
set_monitor_mode(self.nic_mon) set_monitor_mode(self.nic_mon)
log(STATUS, f"Using interface {self.nic_mon} ({get_device_driver(self.nic_mon)}) to inject frames.") log(STATUS, "Using interface {} ({}) to inject frames.".format(self.nic_mon, get_device_driver(self.nic_mon)))
if self.nic_hwsim: if self.nic_hwsim:
set_monitor_mode(self.nic_hwsim) set_monitor_mode(self.nic_hwsim)
@ -844,17 +844,17 @@ class Daemon(metaclass=abc.ABCMeta):
# which can fail on certain devices such as the AWUS036ACH. # which can fail on certain devices such as the AWUS036ACH.
channel = self.wpaspy_command("GET_CHANNEL").strip() channel = self.wpaspy_command("GET_CHANNEL").strip()
if self.options.inject: if self.options.inject:
log(STATUS, f"{self.nic_mon}: setting to channel {channel}") log(STATUS, "{}: setting to channel {}".format(self.nic_mon, channel))
set_channel(self.nic_mon, channel) set_channel(self.nic_mon, channel)
elif self.options.hwsim: elif self.options.hwsim:
log(STATUS, f"{self.nic_hwsim}: setting to channel {channel}") log(STATUS, "{}: setting to channel {}".format(self.nic_hwsim, channel))
log(STATUS, f"{self.nic_mon}: setting to channel {channel}") log(STATUS, "{}: setting to channel {}".format(self.nic_mon, channel))
set_channel(self.nic_hwsim, channel) set_channel(self.nic_hwsim, channel)
set_channel(self.nic_mon, channel) set_channel(self.nic_mon, channel)
if self.options.inject_test != None and self.options.inject_test != "self": if self.options.inject_test != None and self.options.inject_test != "self":
# FIXME: When using 40 MHz channel this call tends to fail the first time # FIXME: When using 40 MHz channel this call tends to fail the first time
log(STATUS, f"{self.options.inject_test}: setting to channel {channel}") log(STATUS, "{}: setting to channel {}".format(self.options.inject_test, channel))
set_channel(self.options.inject_test, channel) set_channel(self.options.inject_test, channel)
# When explicitly testing we can afford a longer timeout. Otherwise we should avoid it. # When explicitly testing we can afford a longer timeout. Otherwise we should avoid it.
time.sleep(0.5) time.sleep(0.5)
@ -876,7 +876,7 @@ class Daemon(metaclass=abc.ABCMeta):
log(ERROR, "Unexpected error. Are you using the correct kernel/driver/device?") log(ERROR, "Unexpected error. Are you using the correct kernel/driver/device?")
quit(1) quit(1)
log(DEBUG, f"Passed injection self-test on interface {self.nic_mon}.") log(DEBUG, "Passed injection self-test on interface {}.".format(self.nic_mon))
quit(1) quit(1)
def forward_hwsim(self, p, s): def forward_hwsim(self, p, s):
@ -885,7 +885,7 @@ class Daemon(metaclass=abc.ABCMeta):
if p.type != 0 and p.type != 2: return if p.type != 0 and p.type != 2: return
if len(p) >= 2200: if len(p) >= 2200:
log(DEBUG, f"Cannot forward frame longer than MTU (length {len(p)}).") log(DEBUG, "Cannot forward frame longer than MTU (length {}).".format(len(p)))
return return
# Due to very strange buy in Scapy, we cannot directly forward frames with a # Due to very strange buy in Scapy, we cannot directly forward frames with a
@ -909,8 +909,8 @@ class Daemon(metaclass=abc.ABCMeta):
# Verify that hostap got recompiled on updates # Verify that hostap got recompiled on updates
version = self.wpaspy_command("GET_VERSION").strip() version = self.wpaspy_command("GET_VERSION").strip()
if version != FRAGVERSION: if version != FRAGVERSION:
log(ERROR, f"This script has version {FRAGVERSION} but compiled wpa_supplicant/hostapd is {version}.") log(ERROR, "This script has version {} but compiled wpa_supplicant/hostapd is {}.".format(FRAGVERSION, version))
log(ERROR, f"Please recompile hostapd/wpa_supplicant using `build.sh`.") log(ERROR, "Please recompile hostapd/wpa_supplicant using `build.sh`.")
quit(1) quit(1)
# Post-startup configuration of the supplicant or AP # Post-startup configuration of the supplicant or AP
@ -974,11 +974,11 @@ class Authenticator(Daemon):
station.time_tick() station.time_tick()
def get_ip(self, station): def get_ip(self, station):
log(STATUS, f"Waiting on client {station.get_peermac()} to get IP") log(STATUS, "Waiting on client {} to get IP".format(station.get_peermac()))
def rekey(self, station): def rekey(self, station):
log(STATUS, f"Starting PTK rekey with client {station.get_peermac()}", color="green") log(STATUS, "Starting PTK rekey with client {}".format(station.get_peermac()), color="green")
cmd = f"REKEY_PTK {station.get_peermac()}" cmd = "REKEY_PTK {}".format(station.get_peermac())
if self.options.rekey_early_install: if self.options.rekey_early_install:
log(STATUS, "Will install PTK during rekey after sending Msg3") log(STATUS, "Will install PTK during rekey after sending Msg3")
cmd += " early-install" cmd += " early-install"
@ -990,10 +990,10 @@ class Authenticator(Daemon):
# Takes a few seconds, and then does a full new connection: Security Camera # Takes a few seconds, and then does a full new connection: Security Camera
if self.options.full_reconnect: if self.options.full_reconnect:
log(STATUS, "Deauthentication station to make it reconnect", color="green") log(STATUS, "Deauthentication station to make it reconnect", color="green")
cmd = f"DEAUTHENTICATE {station.get_peermac()} reason={WLAN_REASON_CLASS2_FRAME_FROM_NONAUTH_STA}" cmd = "DEAUTHENTICATE {} reason={}".format(station.get_peermac(), WLAN_REASON_CLASS2_FRAME_FROM_NONAUTH_STA)
else: else:
log(STATUS, "Disassociating station to make it reconnect", color="green") log(STATUS, "Disassociating station to make it reconnect", color="green")
cmd = f"DISASSOCIATE {station.get_peermac()} reason={WLAN_REASON_CLASS3_FRAME_FROM_NONASSOC_STA}" cmd = "DISASSOCIATE {} reason={}".format(station.get_peermac(), WLAN_REASON_CLASS3_FRAME_FROM_NONASSOC_STA)
self.wpaspy_command(cmd) self.wpaspy_command(cmd)
def handle_eth_dhcp(self, p, station): def handle_eth_dhcp(self, p, station):
@ -1004,7 +1004,7 @@ class Authenticator(Daemon):
if req_type != 3: return if req_type != 3: return
peerip = self.dhcp.leases[station.get_peermac()] peerip = self.dhcp.leases[station.get_peermac()]
log(STATUS, f"Client {station.get_peermac()} with IP {peerip} has connected") log(STATUS, "Client {} with IP {} has connected".format(station.get_peermac(), peerip))
station.set_ip_addresses(self.arp_sender_ip, peerip) station.set_ip_addresses(self.arp_sender_ip, peerip)
def handle_eth(self, p): def handle_eth(self, p):
@ -1045,7 +1045,7 @@ class Authenticator(Daemon):
cmd, clientmac, source = msg.split() cmd, clientmac, source = msg.split()
self.add_station(clientmac) self.add_station(clientmac)
log(STATUS, f"Client {clientmac} is connecting") log(STATUS, "Client {} is connecting".format(clientmac))
station = self.stations[clientmac] station = self.stations[clientmac]
station.handle_connecting(self.apmac) station.handle_connecting(self.apmac)
station.set_peermac(clientmac) station.set_peermac(clientmac)
@ -1057,14 +1057,14 @@ class Authenticator(Daemon):
elif "EAPOL-TX" in msg: elif "EAPOL-TX" in msg:
cmd, clientmac, payload = msg.split() cmd, clientmac, payload = msg.split()
if not clientmac in self.stations: if not clientmac in self.stations:
log(WARNING, f"Sending EAPOL to unknown client {clientmac}.") log(WARNING, "Sending EAPOL to unknown client {}.".format(clientmac))
return return
self.stations[clientmac].handle_eapol_tx(bytes.fromhex(payload), clientmac) self.stations[clientmac].handle_eapol_tx(bytes.fromhex(payload), clientmac)
elif "AP-STA-CONNECTED" in msg: elif "AP-STA-CONNECTED" in msg:
cmd, clientmac = msg.split() cmd, clientmac = msg.split()
if not clientmac in self.stations: if not clientmac in self.stations:
log(WARNING, f"Unknown client {clientmac} finished authenticating.") log(WARNING, "Unknown client {} finished authenticating.".format(clientmac))
return return
self.stations[clientmac].handle_authenticated() self.stations[clientmac].handle_authenticated()
@ -1167,7 +1167,7 @@ class Supplicant(Daemon):
req = req/UDP(sport=68, dport=67)/BOOTP(op=1, chaddr=rawmac, xid=self.dhcp_xid) req = req/UDP(sport=68, dport=67)/BOOTP(op=1, chaddr=rawmac, xid=self.dhcp_xid)
req = req/DHCP(options=[("message-type", "discover"), "end"]) req = req/DHCP(options=[("message-type", "discover"), "end"])
log(STATUS, f"Sending DHCP discover with XID {self.dhcp_xid}") log(STATUS, "Sending DHCP discover with XID {}".format(self.dhcp_xid))
self.station.send_mon(req) self.station.send_mon(req)
def send_dhcp_request(self, offer): def send_dhcp_request(self, offer):
@ -1181,7 +1181,7 @@ class Supplicant(Daemon):
reply = reply/DHCP(options=[("message-type", "request"), ("requested_addr", myip), reply = reply/DHCP(options=[("message-type", "request"), ("requested_addr", myip),
("hostname", "fragclient"), "end"]) ("hostname", "fragclient"), "end"])
log(STATUS, f"Sending DHCP request with XID {self.dhcp_xid}") log(STATUS, "Sending DHCP request with XID {}".format(self.dhcp_xid))
self.station.send_mon(reply) self.station.send_mon(reply)
def handle_eth_dhcp(self, p): def handle_eth_dhcp(self, p):
@ -1201,14 +1201,14 @@ class Supplicant(Daemon):
clientip = p[BOOTP].yiaddr clientip = p[BOOTP].yiaddr
serverip = p[IP].src serverip = p[IP].src
self.time_retrans_dhcp = None self.time_retrans_dhcp = None
log(STATUS, f"Received DHCP ack. My ip is {clientip} and router is {serverip}.", color="green") log(STATUS, "Received DHCP ack. My ip is {} and router is {}.".format(clientip, serverip), color="green")
self.initialize_peermac(p.src) self.initialize_peermac(p.src)
self.initialize_ips(clientip, serverip) self.initialize_ips(clientip, serverip)
def initialize_peermac(self, peermac): def initialize_peermac(self, peermac):
if peermac != self.station.bss: if peermac != self.station.bss:
log(STATUS, f"Will now use peer MAC address {peermac} instead of the BSS {self.station.bss}.") log(STATUS, "Will now use peer MAC address {} instead of the BSS {}.".format(peermac, self.station.bss))
self.station.set_peermac(peermac) self.station.set_peermac(peermac)
def initialize_ips(self, clientip, serverip): def initialize_ips(self, clientip, serverip):
@ -1266,7 +1266,7 @@ class Supplicant(Daemon):
# the authentication phase (reducing the chance that packet queues are reset). # the authentication phase (reducing the chance that packet queues are reset).
optim = "0" if self.options.full_reconnect else "1" optim = "0" if self.options.full_reconnect else "1"
self.wpaspy_command(f"SET reassoc_same_bss_optim {optim}") self.wpaspy_command("SET reassoc_same_bss_optim {}".format(optim))
self.wpaspy_command("REASSOCIATE") self.wpaspy_command("REASSOCIATE")
def configure_daemon(self): def configure_daemon(self):

60
research/libwifi/injectiontest.py
View File

@ -71,45 +71,45 @@ def test_packet_injection(sout, sin, p, test_func, frametype, msgfail):
"""Check if given property holds of all injected frames""" """Check if given property holds of all injected frames"""
packets = inject_and_capture(sout, sin, p, count=1) packets = inject_and_capture(sout, sin, p, count=1)
if len(packets) < 1: if len(packets) < 1:
log(ERROR, f"[-] Unable to capture injected {frametype}.") log(ERROR, "[-] Unable to capture injected {}.".format(frametype))
return FLAG_NOCAPTURE return FLAG_NOCAPTURE
if not all([test_func(cap) for cap in packets]): if not all([test_func(cap) for cap in packets]):
log(ERROR, f"[-] " + msgfail.format(frametype=frametype)) log(ERROR, "[-] " + msgfail.format(frametype=frametype))
return FLAG_FAIL return FLAG_FAIL
log(STATUS, f" Properly captured injected {frametype}.") log(STATUS, " Properly captured injected {}.".format(frametype))
return 0 return 0
def test_injection_fields(sout, sin, ref, strtype): def test_injection_fields(sout, sin, ref, strtype):
log(STATUS, f"--- Testing injection of fields using {strtype}") log(STATUS, "--- Testing injection of fields using {}".format(strtype))
status = 0 status = 0
p = Dot11(FCfield=ref.FCfield, addr1=ref.addr1, addr2=ref.addr2, addr3=ref.addr3, type=2, SC=30<<4)/LLC()/SNAP()/EAPOL()/EAP() p = Dot11(FCfield=ref.FCfield, addr1=ref.addr1, addr2=ref.addr2, addr3=ref.addr3, type=2, SC=30<<4)/LLC()/SNAP()/EAPOL()/EAP()
status |= test_packet_injection(sout, sin, p, lambda cap: EAPOL in cap, f"EAPOL frame with {strtype}", status |= test_packet_injection(sout, sin, p, lambda cap: EAPOL in cap, "EAPOL frame with {}".format(strtype),
"Scapy thinks injected {frametype} is a different frame?") "Scapy thinks injected {frametype} is a different frame?")
p = Dot11(FCfield=ref.FCfield, addr1=ref.addr1, addr2=ref.addr2, addr3=ref.addr3, type=2, SC=31<<4) p = Dot11(FCfield=ref.FCfield, addr1=ref.addr1, addr2=ref.addr2, addr3=ref.addr3, type=2, SC=31<<4)
status |= test_packet_injection(sout, sin, p, lambda cap: cap.SC == p.SC, f"empty data frame with {strtype}", status |= test_packet_injection(sout, sin, p, lambda cap: cap.SC == p.SC, "empty data frame with {}".format(strtype),
"Sequence number of injected {frametype} is being overwritten!") "Sequence number of injected {frametype} is being overwritten!")
p = Dot11(FCfield=ref.FCfield, addr1=ref.addr1, addr2=ref.addr2, addr3=ref.addr3, type=2, SC=(32<<4)|1) p = Dot11(FCfield=ref.FCfield, addr1=ref.addr1, addr2=ref.addr2, addr3=ref.addr3, type=2, SC=(32<<4)|1)
status |= test_packet_injection(sout, sin, p, lambda cap: (cap.SC & 0xf) == 1, f"fragmented empty data frame with {strtype}", status |= test_packet_injection(sout, sin, p, lambda cap: (cap.SC & 0xf) == 1, "fragmented empty data frame with {}".format(strtype),
"Fragment number of injected {frametype} is being overwritten!") "Fragment number of injected {frametype} is being overwritten!")
p = Dot11(FCfield=ref.FCfield, addr1=ref.addr1, addr2=ref.addr2, addr3=ref.addr3, type=2, subtype=8, SC=33<<4)/Dot11QoS(TID=2) p = Dot11(FCfield=ref.FCfield, addr1=ref.addr1, addr2=ref.addr2, addr3=ref.addr3, type=2, subtype=8, SC=33<<4)/Dot11QoS(TID=2)
status |= test_packet_injection(sout, sin, p, lambda cap: cap.TID == p.TID, f"empty QoS data frame with {strtype}", status |= test_packet_injection(sout, sin, p, lambda cap: cap.TID == p.TID, "empty QoS data frame with {}".format(strtype),
"QoS TID of injected {frametype} is being overwritten!") "QoS TID of injected {frametype} is being overwritten!")
p = Dot11(FCfield=ref.FCfield, addr1=ref.addr1, addr2=ref.addr2, addr3=ref.addr3, type=2, subtype=8, SC=33<<4)/Dot11QoS(TID=2)/Raw("BBBB") p = Dot11(FCfield=ref.FCfield, addr1=ref.addr1, addr2=ref.addr2, addr3=ref.addr3, type=2, subtype=8, SC=33<<4)/Dot11QoS(TID=2)/Raw("BBBB")
set_amsdu(p[Dot11QoS]) set_amsdu(p[Dot11QoS])
status |= test_packet_injection(sout, sin, p, \ status |= test_packet_injection(sout, sin, p, \
lambda cap: cap.TID == p.TID and is_amsdu(cap) and b"BBBB" in raw(cap), \ lambda cap: cap.TID == p.TID and is_amsdu(cap) and b"BBBB" in raw(cap), \
f"A-MSDU frame with {strtype}", "A-MSDU frame is not properly injected!") "A-MSDU frame with {}".format(strtype), "A-MSDU frame is not properly injected!")
if status == 0: log(STATUS, f"[+] All tested fields are properly injected when using {strtype}.", color="green") if status == 0: log(STATUS, "[+] All tested fields are properly injected when using {}.".format(strtype), color="green")
return status return status
def test_injection_order(sout, sin, ref, strtype, retries=1): def test_injection_order(sout, sin, ref, strtype, retries=1):
log(STATUS, f"--- Testing order of injected QoS frames using {strtype}") log(STATUS, "--- Testing order of injected QoS frames using {}".format(strtype))
label = b"AAAA" + struct.pack(">II", random.randint(0, 2**32), random.randint(0, 2**32)) label = b"AAAA" + struct.pack(">II", random.randint(0, 2**32), random.randint(0, 2**32))
p2 = Dot11(FCfield=ref.FCfield, addr1=ref.addr1, addr2=ref.addr2, type=2, subtype=8, SC=33<<4)/Dot11QoS(TID=2) p2 = Dot11(FCfield=ref.FCfield, addr1=ref.addr1, addr2=ref.addr2, type=2, subtype=8, SC=33<<4)/Dot11QoS(TID=2)
@ -126,18 +126,18 @@ def test_injection_order(sout, sin, ref, strtype, retries=1):
# Sanity check the captured TIDs, and then analyze the results # Sanity check the captured TIDs, and then analyze the results
if not (2 in tids and 6 in tids): if not (2 in tids and 6 in tids):
log(STATUS, f"We didn't capture all injected QoS TID frames, retrying.") log(STATUS, "We didn't capture all injected QoS TID frames, retrying.")
else: else:
break break
if not (2 in tids and 6 in tids): if not (2 in tids and 6 in tids):
log(ERROR, f"[-] We didn't capture all injected QoS TID frames with {strtype}. Test failed.") log(ERROR, "[-] We didn't capture all injected QoS TID frames with {}. Test failed.".format(strtype))
return FLAG_NOCAPTURE return FLAG_NOCAPTURE
elif tids != sorted(tids): elif tids != sorted(tids):
log(ERROR, f"[-] Frames with different QoS TIDs are reordered during injection with {strtype}.") log(ERROR, "[-] Frames with different QoS TIDs are reordered during injection with {}.".format(strtype))
return FLAG_FAIL return FLAG_FAIL
else: else:
log(STATUS, f"[+] Frames with different QoS TIDs are not reordered during injection with {strtype}.", color="green") log(STATUS, "[+] Frames with different QoS TIDs are not reordered during injection with {}.".format(strtype), color="green")
return 0 return 0
def test_injection_ack(sout, sin, addr1, addr2): def test_injection_ack(sout, sin, addr1, addr2):
@ -147,7 +147,7 @@ def test_injection_ack(sout, sin, addr1, addr2):
# Test number of retransmissions # Test number of retransmissions
p = Dot11(FCfield="to-DS", addr1="00:11:00:00:02:01", addr2="00:11:00:00:02:01", type=2, SC=33<<4) p = Dot11(FCfield="to-DS", addr1="00:11:00:00:02:01", addr2="00:11:00:00:02:01", type=2, SC=33<<4)
num = len(inject_and_capture(sout, sin, p, retries=1)) num = len(inject_and_capture(sout, sin, p, retries=1))
log(STATUS, f"Injected frames seem to be (re)transitted {num} times") log(STATUS, "Injected frames seem to be (re)transitted {} times".format(num))
if num == 0: if num == 0:
log(ERROR, "Couldn't capture injected frame. Please restart the test.") log(ERROR, "Couldn't capture injected frame. Please restart the test.")
test_fail = True test_fail = True
@ -158,7 +158,7 @@ def test_injection_ack(sout, sin, addr1, addr2):
# Test ACK towards an unassigned MAC address # Test ACK towards an unassigned MAC address
p = Dot11(FCfield="to-DS", addr1=addr1, addr2="00:22:00:00:00:01", type=2, SC=33<<4) p = Dot11(FCfield="to-DS", addr1=addr1, addr2="00:22:00:00:00:01", type=2, SC=33<<4)
num = len(inject_and_capture(sout, sin, p, retries=1)) num = len(inject_and_capture(sout, sin, p, retries=1))
log(STATUS, f"Captured {num} (re)transmitted frames to the AP when using a spoofed sender address") log(STATUS, "Captured {} (re)transmitted frames to the AP when using a spoofed sender address".format(num))
if num == 0: if num == 0:
log(ERROR, "Couldn't capture injected frame. Please restart the test.") log(ERROR, "Couldn't capture injected frame. Please restart the test.")
test_fail = True test_fail = True
@ -168,7 +168,7 @@ def test_injection_ack(sout, sin, addr1, addr2):
# Test ACK towards an assigned MAC address # Test ACK towards an assigned MAC address
p = Dot11(FCfield="to-DS", addr1=addr1, addr2=addr2, type=2, SC=33<<4) p = Dot11(FCfield="to-DS", addr1=addr1, addr2=addr2, type=2, SC=33<<4)
num = len(inject_and_capture(sout, sin, p, retries=1)) num = len(inject_and_capture(sout, sin, p, retries=1))
log(STATUS, f"Captured {num} (re)transmitted frames to the AP when using the real sender address") log(STATUS, "Captured {} (re)transmitted frames to the AP when using the real sender address".format(num))
if num == 0: if num == 0:
log(ERROR, "Couldn't capture injected frame. Please restart the test.") log(ERROR, "Couldn't capture injected frame. Please restart the test.")
test_fail = True test_fail = True
@ -194,22 +194,22 @@ def test_injection(iface_out, iface_in=None, peermac=None, ownmac=None, testack=
# Workaround to properly inject fragmented frames (and prevent it from blocking Tx queue). # Workaround to properly inject fragmented frames (and prevent it from blocking Tx queue).
sout.mf_workaround = driver_out in ["iwlwifi", "ath9k_htc"] sout.mf_workaround = driver_out in ["iwlwifi", "ath9k_htc"]
if sout.mf_workaround: if sout.mf_workaround:
log(WARNING, f"Detected {driver_out}, using workaround to reliably inject fragmented frames.") log(WARNING, "Detected {}, using workaround to reliably inject fragmented frames.".format(driver_out))
# Print out what we are tested. Abort if the driver is known not to support a self-test. # Print out what we are tested. Abort if the driver is known not to support a self-test.
log(STATUS, f"Injection test: using {iface_out} ({driver_out}) to inject frames") log(STATUS, "Injection test: using {} ({}) to inject frames".format(iface_out, driver_out))
if iface_in == None: if iface_in == None:
log(WARNING, f"Injection selftest: also using {iface_out} to capture frames. This means the tests can detect if the kernel") log(WARNING, "Injection selftest: also using {} to capture frames. This means the tests can detect if the kernel".format(iface_out))
log(WARNING, f" interferes with injection, but it cannot check the behaviour of the device itself.") log(WARNING, " interferes with injection, but it cannot check the behaviour of the device itself.")
if driver_out in ["mt76x2u"]: if driver_out in ["mt76x2u"]:
log(WARNING, f" WARNING: self-test with the {driver_out} driver can be unreliable.") log(WARNING, " WARNING: self-test with the {} driver can be unreliable.".format(driver_out))
elif not driver_out in ["iwlwifi", "ath9k_htc"]: elif not driver_out in ["iwlwifi", "ath9k_htc"]:
log(WARNING, f" WARNING: it is unknown whether a self-test works with the {driver_out} driver.") log(WARNING, " WARNING: it is unknown whether a self-test works with the {} driver.".format(driver_out))
sin = sout sin = sout
else: else:
driver_in = get_device_driver(iface_in) driver_in = get_device_driver(iface_in)
log(STATUS, f"Injection test: using {iface_in} ({driver_in}) to capture frames") log(STATUS, "Injection test: using {} ({}) to capture frames".format(iface_in, driver_in))
sin = L2Socket(type=ETH_P_ALL, iface=iface_in) sin = L2Socket(type=ETH_P_ALL, iface=iface_in)
# Injection using the "own" MAC address is mainly a problem when using a second virtual # Injection using the "own" MAC address is mainly a problem when using a second virtual
@ -242,15 +242,15 @@ def test_injection(iface_out, iface_in=None, peermac=None, ownmac=None, testack=
# 2. If sout interface "sees" the AP this assure it will also receive its ACK frames # 2. If sout interface "sees" the AP this assure it will also receive its ACK frames
# 3. The given peermac might be a client that goes into sleep mode # 3. The given peermac might be a client that goes into sleep mode
channel = get_channel(sin.iface) channel = get_channel(sin.iface)
log(STATUS, f"--- Searching for AP on channel {channel} to test ACK behaviour.") log(STATUS, "--- Searching for AP on channel {} to test ACK behaviour.".format(channel))
apmac, ssid = get_nearby_ap_addr(sout) apmac, ssid = get_nearby_ap_addr(sout)
if apmac == None and peermac == None: if apmac == None and peermac == None:
raise IOError("Unable to find nearby AP to test injection") raise IOError("Unable to find nearby AP to test injection")
elif apmac == None: elif apmac == None:
log(WARNING, f"Unable to find AP. Try a different channel? Testing ACK behaviour with peer {peermac}.") log(WARNING, "Unable to find AP. Try a different channel? Testing ACK behaviour with peer {}.".format(peermac))
destmac = peermac destmac = peermac
else: else:
log(STATUS, f"Testing ACK behaviour by injecting frames to AP {ssid} ({apmac}).") log(STATUS, "Testing ACK behaviour by injecting frames to AP {} ({}).".format(ssid, apmac))
destmac = apmac destmac = apmac
test_injection_ack(sout, sin, addr1=destmac, addr2=ownmac) test_injection_ack(sout, sin, addr1=destmac, addr2=ownmac)
@ -259,9 +259,9 @@ def test_injection(iface_out, iface_in=None, peermac=None, ownmac=None, testack=
if status == 0: if status == 0:
log(STATUS, "==> The most important tests have been passed successfully!", color="green") log(STATUS, "==> The most important tests have been passed successfully!", color="green")
if status & FLAG_NOCAPTURE != 0: if status & FLAG_NOCAPTURE != 0:
log(WARNING, f"==> Failed to capture some frames. Try another channel or use another monitoring device.") log(WARNING, "==> Failed to capture some frames. Try another channel or use another monitoring device.")
if status & FLAG_FAIL !=0 : if status & FLAG_FAIL !=0 :
log(ERROR, f"==> Some tests failed. Are you using patched drivers/firmware?") log(ERROR, "==> Some tests failed. Are you using patched drivers/firmware?")
sout.close() sout.close()
sin.close() sin.close()

2
research/tests_common.py
View File

@ -228,7 +228,7 @@ class EapolAmsduTest(Test):
actions = self.get_actions(Action.Inject) actions = self.get_actions(Action.Inject)
if len(actions) != 1: if len(actions) != 1:
log(ERROR, f"eapol-amsdu: invalid arguments, should only give 1 inject action (gave {len(actions)}).") log(ERROR, "eapol-amsdu: invalid arguments, should only give 1 inject action (gave {}).".format(len(actions)))
quit(1) quit(1)
def prepare(self, station): def prepare(self, station):

4
research/tests_old.py
View File

@ -19,10 +19,10 @@ class BcastEapFragTest(Test):
actions = self.get_actions(Action.Inject) actions = self.get_actions(Action.Inject)
if len(actions) != 2: if len(actions) != 2:
log(ERROR, f"eapfrag: invalid arguments, should give 2 inject action (gave {len(actions)}).") log(ERROR, "eapfrag: invalid arguments, should give 2 inject action (gave {}).".format(len(actions)))
quit(1) quit(1)
elif actions[0].encrypted: elif actions[0].encrypted:
log(ERROR, f"eapfrag: first inject action should not be encrypted.") log(ERROR, "eapfrag: first inject action should not be encrypted.")
quit(1) quit(1)
def prepare(self, station): def prepare(self, station):