1# Protected management frames tests
2# Copyright (c) 2013, Jouni Malinen <j@w1.fi>
3#
4# This software may be distributed under the terms of the BSD license.
5# See README for more details.
6
7from remotehost import remote_compatible
8import binascii
9import os
10import time
11import logging
12logger = logging.getLogger()
13
14import hwsim_utils
15import hostapd
16from utils import *
17from wlantest import Wlantest
18from wpasupplicant import WpaSupplicant
19
20@remote_compatible
21def test_ap_pmf_required(dev, apdev):
22    """WPA2-PSK AP with PMF required"""
23    ssid = "test-pmf-required"
24    params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
25    params["wpa_key_mgmt"] = "WPA-PSK-SHA256"
26    params["ieee80211w"] = "2"
27    hapd = hostapd.add_ap(apdev[0], params)
28    Wlantest.setup(hapd)
29    wt = Wlantest()
30    wt.flush()
31    wt.add_passphrase("12345678")
32    key_mgmt = hapd.get_config()['key_mgmt']
33    if key_mgmt.split(' ')[0] != "WPA-PSK-SHA256":
34        raise Exception("Unexpected GET_CONFIG(key_mgmt): " + key_mgmt)
35    dev[0].connect(ssid, psk="12345678", ieee80211w="1",
36                   key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
37                   scan_freq="2412")
38    if "[WPA2-PSK-SHA256-CCMP]" not in dev[0].request("SCAN_RESULTS"):
39        raise Exception("Scan results missing RSN element info")
40    hwsim_utils.test_connectivity(dev[0], hapd)
41    dev[1].connect(ssid, psk="12345678", ieee80211w="2",
42                   key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
43                   scan_freq="2412")
44    hwsim_utils.test_connectivity(dev[1], hapd)
45    if "OK" not in hapd.request("SA_QUERY " + dev[0].own_addr()):
46        raise Exception("SA_QUERY failed")
47    if "OK" not in hapd.request("SA_QUERY " + dev[1].own_addr()):
48        raise Exception("SA_QUERY failed")
49    if "FAIL" not in hapd.request("SA_QUERY foo"):
50        raise Exception("Invalid SA_QUERY accepted")
51    wt.require_ap_pmf_mandatory(apdev[0]['bssid'])
52    wt.require_sta_pmf(apdev[0]['bssid'], dev[0].p2p_interface_addr())
53    wt.require_sta_pmf_mandatory(apdev[0]['bssid'], dev[1].p2p_interface_addr())
54    time.sleep(0.1)
55    if wt.get_sta_counter("valid_saqueryresp_tx", apdev[0]['bssid'],
56                          dev[0].p2p_interface_addr()) < 1:
57        raise Exception("STA did not reply to SA Query")
58    if wt.get_sta_counter("valid_saqueryresp_tx", apdev[0]['bssid'],
59                          dev[1].p2p_interface_addr()) < 1:
60        raise Exception("STA did not reply to SA Query")
61
62def start_ocv_ap(apdev):
63    ssid = "test-pmf-required"
64    params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
65    params["wpa_key_mgmt"] = "WPA-PSK-SHA256"
66    params["ieee80211w"] = "2"
67    params["ocv"] = "1"
68    try:
69        hapd = hostapd.add_ap(apdev, params)
70    except Exception as e:
71        if "Failed to set hostapd parameter ocv" in str(e):
72            raise HwsimSkip("OCV not supported")
73        raise
74
75    Wlantest.setup(hapd)
76    wt = Wlantest()
77    wt.flush()
78    wt.add_passphrase("12345678")
79
80    return hapd, ssid, wt
81
82@remote_compatible
83def test_ocv_sa_query(dev, apdev):
84    """Test SA Query with OCV"""
85    hapd, ssid, wt = start_ocv_ap(apdev[0])
86    dev[0].connect(ssid, psk="12345678", ieee80211w="1", ocv="1",
87                   key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
88                   scan_freq="2412")
89
90    # Test that client can handle SA Query with OCI element
91    if "OK" not in hapd.request("SA_QUERY " + dev[0].own_addr()):
92        raise Exception("SA_QUERY failed")
93    ev = hapd.wait_event(["OCV-FAILURE"], timeout=0.1)
94    if ev:
95        raise Exception("Unexpected OCV failure reported")
96    if wt.get_sta_counter("valid_saqueryresp_tx", apdev[0]['bssid'],
97                          dev[0].own_addr()) < 1:
98        raise Exception("STA did not reply to SA Query")
99
100    # Test that AP can handle SA Query with OCI element
101    if "OK" not in dev[0].request("UNPROT_DEAUTH"):
102        raise Exception("Triggering SA Query from the STA failed")
103    ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=3)
104    if ev is not None:
105        raise Exception("SA Query from the STA failed")
106
107@remote_compatible
108def test_ocv_sa_query_csa(dev, apdev):
109    """Test SA Query with OCV after channel switch"""
110    hapd, ssid, wt = start_ocv_ap(apdev[0])
111    dev[0].connect(ssid, psk="12345678", ieee80211w="1", ocv="1",
112                   key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
113                   scan_freq="2412")
114
115    hapd.request("CHAN_SWITCH 5 2437")
116    time.sleep(1)
117    if wt.get_sta_counter("valid_saqueryreq_tx", apdev[0]['bssid'],
118                          dev[0].own_addr()) < 1:
119        raise Exception("STA did not start SA Query after channel switch")
120
121    ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=16)
122    if ev is not None:
123        raise Exception("Unexpected disconnection")
124
125def test_ocv_sa_query_csa_no_resp(dev, apdev):
126    """Test SA Query with OCV after channel switch getting no response"""
127    hapd, ssid, wt = start_ocv_ap(apdev[0])
128    dev[0].connect(ssid, psk="12345678", ieee80211w="1", ocv="1",
129                   key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
130                   scan_freq="2412")
131
132    hapd.set("ext_mgmt_frame_handling", "1")
133    hapd.request("CHAN_SWITCH 5 2437")
134    ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=5)
135    if ev is None:
136        raise Exception("Disconnection after CSA not reported")
137    if "locally_generated=1" not in ev:
138        raise Exception("Unexpectedly disconnected by AP: " + ev)
139
140def test_ocv_sa_query_csa_missing(dev, apdev):
141    """Test SA Query with OCV missing after channel switch"""
142    hapd, ssid, wt = start_ocv_ap(apdev[0])
143    dev[0].connect(ssid, psk="12345678", ieee80211w="1", ocv="1",
144                   key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
145                   scan_freq="2412")
146
147    hapd.set("ext_mgmt_frame_handling", "1")
148    dev[0].request("DISCONNECT")
149    dev[0].wait_disconnected()
150    ev = hapd.wait_event(['MGMT-RX'], timeout=5)
151    if ev is None:
152        raise Exception("Deauthentication frame RX not reported")
153    hapd.set("ext_mgmt_frame_handling", "0")
154    hapd.request("CHAN_SWITCH 5 2437")
155    ev = hapd.wait_event(["AP-STA-DISCONNECTED"], timeout=20)
156    if ev is None:
157        raise Exception("No disconnection event received from hostapd")
158
159@remote_compatible
160def test_ap_pmf_optional(dev, apdev):
161    """WPA2-PSK AP with PMF optional"""
162    ssid = "test-pmf-optional"
163    params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
164    params["wpa_key_mgmt"] = "WPA-PSK"
165    params["ieee80211w"] = "1"
166    hapd = hostapd.add_ap(apdev[0], params)
167    Wlantest.setup(hapd)
168    wt = Wlantest()
169    wt.flush()
170    wt.add_passphrase("12345678")
171    dev[0].connect(ssid, psk="12345678", ieee80211w="1",
172                   key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
173                   scan_freq="2412")
174    hwsim_utils.test_connectivity(dev[0], hapd)
175    dev[1].connect(ssid, psk="12345678", ieee80211w="2",
176                   key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
177                   scan_freq="2412")
178    hwsim_utils.test_connectivity(dev[1], hapd)
179    wt.require_ap_pmf_optional(apdev[0]['bssid'])
180    wt.require_sta_pmf(apdev[0]['bssid'], dev[0].p2p_interface_addr())
181    wt.require_sta_pmf_mandatory(apdev[0]['bssid'], dev[1].p2p_interface_addr())
182
183@remote_compatible
184def test_ap_pmf_optional_2akm(dev, apdev):
185    """WPA2-PSK AP with PMF optional (2 AKMs)"""
186    ssid = "test-pmf-optional-2akm"
187    params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
188    params["wpa_key_mgmt"] = "WPA-PSK WPA-PSK-SHA256"
189    params["ieee80211w"] = "1"
190    hapd = hostapd.add_ap(apdev[0], params)
191    Wlantest.setup(hapd)
192    wt = Wlantest()
193    wt.flush()
194    wt.add_passphrase("12345678")
195    dev[0].connect(ssid, psk="12345678", ieee80211w="1",
196                   key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
197                   scan_freq="2412")
198    hwsim_utils.test_connectivity(dev[0], hapd)
199    dev[1].connect(ssid, psk="12345678", ieee80211w="2",
200                   key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
201                   scan_freq="2412")
202    hwsim_utils.test_connectivity(dev[1], hapd)
203    wt.require_ap_pmf_optional(apdev[0]['bssid'])
204    wt.require_sta_pmf(apdev[0]['bssid'], dev[0].p2p_interface_addr())
205    wt.require_sta_key_mgmt(apdev[0]['bssid'], dev[0].p2p_interface_addr(),
206                            "PSK-SHA256")
207    wt.require_sta_pmf_mandatory(apdev[0]['bssid'], dev[1].p2p_interface_addr())
208    wt.require_sta_key_mgmt(apdev[0]['bssid'], dev[1].p2p_interface_addr(),
209                            "PSK-SHA256")
210
211@remote_compatible
212def test_ap_pmf_negative(dev, apdev):
213    """WPA2-PSK AP without PMF (negative test)"""
214    ssid = "test-pmf-negative"
215    params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
216    hapd = hostapd.add_ap(apdev[0], params)
217    Wlantest.setup(hapd)
218    wt = Wlantest()
219    wt.flush()
220    wt.add_passphrase("12345678")
221    dev[0].connect(ssid, psk="12345678", ieee80211w="1",
222                   key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
223                   scan_freq="2412")
224    hwsim_utils.test_connectivity(dev[0], hapd)
225    try:
226        dev[1].connect(ssid, psk="12345678", ieee80211w="2",
227                       key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
228                       scan_freq="2412")
229        hwsim_utils.test_connectivity(dev[1], hapd)
230        raise Exception("PMF required STA connected to no PMF AP")
231    except Exception as e:
232        logger.debug("Ignore expected exception: " + str(e))
233    wt.require_ap_no_pmf(apdev[0]['bssid'])
234
235@remote_compatible
236def test_ap_pmf_assoc_comeback(dev, apdev):
237    """WPA2-PSK AP with PMF association comeback"""
238    ssid = "assoc-comeback"
239    params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
240    params["wpa_key_mgmt"] = "WPA-PSK-SHA256"
241    params["ieee80211w"] = "2"
242    hapd = hostapd.add_ap(apdev[0], params)
243    Wlantest.setup(hapd)
244    wt = Wlantest()
245    wt.flush()
246    wt.add_passphrase("12345678")
247    dev[0].connect(ssid, psk="12345678", ieee80211w="1",
248                   key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
249                   scan_freq="2412")
250    hapd.set("ext_mgmt_frame_handling", "1")
251    dev[0].request("DISCONNECT")
252    dev[0].wait_disconnected(timeout=10)
253    hapd.set("ext_mgmt_frame_handling", "0")
254    dev[0].request("REASSOCIATE")
255    dev[0].wait_connected(timeout=10, error="Timeout on re-connection")
256    if wt.get_sta_counter("assocresp_comeback", apdev[0]['bssid'],
257                          dev[0].p2p_interface_addr()) < 1:
258        raise Exception("AP did not use association comeback request")
259
260@remote_compatible
261def test_ap_pmf_assoc_comeback2(dev, apdev):
262    """WPA2-PSK AP with PMF association comeback (using DROP_SA)"""
263    ssid = "assoc-comeback"
264    params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
265    params["wpa_key_mgmt"] = "WPA-PSK"
266    params["ieee80211w"] = "1"
267    hapd = hostapd.add_ap(apdev[0], params)
268    Wlantest.setup(hapd)
269    wt = Wlantest()
270    wt.flush()
271    wt.add_passphrase("12345678")
272    dev[0].connect(ssid, psk="12345678", ieee80211w="2",
273                   key_mgmt="WPA-PSK", proto="WPA2", scan_freq="2412")
274    if "OK" not in dev[0].request("DROP_SA"):
275        raise Exception("DROP_SA failed")
276    dev[0].request("REASSOCIATE")
277    dev[0].wait_connected(timeout=10, error="Timeout on re-connection")
278    if wt.get_sta_counter("reassocresp_comeback", apdev[0]['bssid'],
279                          dev[0].p2p_interface_addr()) < 1:
280        raise Exception("AP did not use reassociation comeback request")
281
282@remote_compatible
283def test_ap_pmf_assoc_comeback3(dev, apdev):
284    """WPA2-PSK AP with PMF association comeback (using radio_disabled)"""
285    drv_flags = dev[0].get_driver_status_field("capa.flags")
286    if int(drv_flags, 0) & 0x20 == 0:
287        raise HwsimSkip("Driver does not support radio_disabled")
288    ssid = "assoc-comeback"
289    params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
290    params["wpa_key_mgmt"] = "WPA-PSK"
291    params["ieee80211w"] = "1"
292    hapd = hostapd.add_ap(apdev[0], params)
293    Wlantest.setup(hapd)
294    wt = Wlantest()
295    wt.flush()
296    wt.add_passphrase("12345678")
297    dev[0].connect(ssid, psk="12345678", ieee80211w="2",
298                   key_mgmt="WPA-PSK", proto="WPA2", scan_freq="2412")
299    dev[0].set("radio_disabled", "1")
300    dev[0].set("radio_disabled", "0")
301    dev[0].request("REASSOCIATE")
302    dev[0].wait_connected(timeout=10, error="Timeout on re-connection")
303    if wt.get_sta_counter("assocresp_comeback", apdev[0]['bssid'],
304                          dev[0].own_addr()) < 1:
305        raise Exception("AP did not use reassociation comeback request")
306
307@remote_compatible
308def test_ap_pmf_assoc_comeback_wps(dev, apdev):
309    """WPA2-PSK AP with PMF association comeback (WPS)"""
310    ssid = "assoc-comeback"
311    appin = "12345670"
312    params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
313    params["wpa_key_mgmt"] = "WPA-PSK-SHA256"
314    params["ieee80211w"] = "2"
315    params["eap_server"] = "1"
316    params["wps_state"] = "2"
317    params["ap_pin"] = appin
318    hapd = hostapd.add_ap(apdev[0], params)
319    Wlantest.setup(hapd)
320    wt = Wlantest()
321    wt.flush()
322    wt.add_passphrase("12345678")
323    dev[0].connect(ssid, psk="12345678", ieee80211w="1",
324                   key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
325                   scan_freq="2412")
326    hapd.set("ext_mgmt_frame_handling", "1")
327    dev[0].request("DISCONNECT")
328    dev[0].wait_disconnected(timeout=10)
329    hapd.set("ext_mgmt_frame_handling", "0")
330    dev[0].wps_reg(apdev[0]['bssid'], appin)
331    if wt.get_sta_counter("assocresp_comeback", apdev[0]['bssid'],
332                          dev[0].p2p_interface_addr()) < 1:
333        raise Exception("AP did not use association comeback request")
334
335def test_ap_pmf_ap_dropping_sa(dev, apdev):
336    """WPA2-PSK PMF AP dropping SA"""
337    ssid = "pmf"
338    params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
339    params["wpa_key_mgmt"] = "WPA-PSK-SHA256"
340    params["ieee80211w"] = "2"
341    hapd = hostapd.add_ap(apdev[0], params)
342    bssid = hapd.own_addr()
343    Wlantest.setup(hapd)
344    wt = Wlantest()
345    wt.flush()
346    wt.add_passphrase("12345678")
347    dev[0].connect(ssid, psk="12345678", ieee80211w="2",
348                   key_mgmt="WPA-PSK-SHA256", proto="WPA2", scan_freq="2412")
349    addr0 = dev[0].own_addr()
350    dev[0].dump_monitor()
351    hapd.wait_sta()
352    # Drop SA and association at the AP locally without notifying the STA. This
353    # results in the STA getting unprotected Deauthentication frames when trying
354    # to transmit the next Class 3 frame.
355    if "OK" not in hapd.request("DEAUTHENTICATE " + addr0 + " tx=0"):
356        raise Exception("DEAUTHENTICATE command failed")
357    ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=1)
358    if ev is not None:
359        raise Exception("Unexpected disconnection event after DEAUTHENTICATE tx=0: " + ev)
360    dev[0].request("DATA_TEST_CONFIG 1")
361    dev[0].request("DATA_TEST_TX " + bssid + " " + addr0)
362    ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=5)
363    dev[0].request("DATA_TEST_CONFIG 0")
364    if ev is None or "locally_generated=1" not in ev:
365        raise Exception("Locally generated disconnection not reported")
366
367def test_ap_pmf_valid_broadcast_deauth(dev, apdev):
368    """WPA2-PSK PMF AP sending valid broadcast deauth without dropping SA"""
369    run_ap_pmf_valid(dev, apdev, False, True)
370
371def test_ap_pmf_valid_broadcast_disassoc(dev, apdev):
372    """WPA2-PSK PMF AP sending valid broadcast disassoc without dropping SA"""
373    run_ap_pmf_valid(dev, apdev, True, True)
374
375def test_ap_pmf_valid_unicast_deauth(dev, apdev):
376    """WPA2-PSK PMF AP sending valid unicast deauth without dropping SA"""
377    run_ap_pmf_valid(dev, apdev, False, False)
378
379def test_ap_pmf_valid_unicast_disassoc(dev, apdev):
380    """WPA2-PSK PMF AP sending valid unicast disassoc without dropping SA"""
381    run_ap_pmf_valid(dev, apdev, True, False)
382
383def run_ap_pmf_valid(dev, apdev, disassociate, broadcast):
384    ssid = "pmf"
385    params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
386    params["wpa_key_mgmt"] = "WPA-PSK-SHA256"
387    params["ieee80211w"] = "2"
388    hapd = hostapd.add_ap(apdev[0], params)
389    bssid = hapd.own_addr()
390    Wlantest.setup(hapd)
391    wt = Wlantest()
392    wt.flush()
393    wt.add_passphrase("12345678")
394    dev[0].connect(ssid, psk="12345678", ieee80211w="2",
395                   key_mgmt="WPA-PSK-SHA256", proto="WPA2", scan_freq="2412")
396    addr0 = dev[0].own_addr()
397    dev[0].dump_monitor()
398    hapd.wait_sta()
399    cmd = "DISASSOCIATE " if disassociate else "DEAUTHENTICATE "
400    cmd += "ff:ff:ff:ff:ff:ff" if broadcast else addr0
401    cmd += " test=1"
402    if "OK" not in hapd.request(cmd):
403        raise Exception("hostapd command failed")
404    sta = hapd.get_sta(addr0)
405    if not sta:
406        raise Exception("STA entry lost")
407    ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=5)
408    if ev is None:
409        raise Exception("Disconnection not reported")
410    if "locally_generated=1" in ev:
411        raise Exception("Unexpected locally generated disconnection")
412
413    # Wait for SA Query procedure to fail and association comeback to succeed
414    dev[0].wait_connected()
415
416def start_wpas_ap(ssid):
417    wpas = WpaSupplicant(global_iface='/tmp/wpas-wlan5')
418    wpas.interface_add("wlan5", drv_params="use_monitor=1")
419    id = wpas.add_network()
420    wpas.set_network(id, "mode", "2")
421    wpas.set_network_quoted(id, "ssid", ssid)
422    wpas.set_network(id, "proto", "WPA2")
423    wpas.set_network(id, "key_mgmt", "WPA-PSK-SHA256")
424    wpas.set_network(id, "ieee80211w", "2")
425    wpas.set_network_quoted(id, "psk", "12345678")
426    wpas.set_network(id, "pairwise", "CCMP")
427    wpas.set_network(id, "group", "CCMP")
428    wpas.set_network(id, "frequency", "2412")
429    wpas.set_network(id, "scan_freq", "2412")
430    wpas.connect_network(id)
431    wpas.dump_monitor()
432    return wpas
433
434def test_ap_pmf_sta_sa_query(dev, apdev):
435    """WPA2-PSK AP with station using SA Query"""
436    ssid = "assoc-comeback"
437    addr = dev[0].own_addr()
438
439    wpas = start_wpas_ap(ssid)
440    bssid = wpas.own_addr()
441
442    Wlantest.setup(wpas)
443    wt = Wlantest()
444    wt.flush()
445    wt.add_passphrase("12345678")
446
447    dev[0].connect(ssid, psk="12345678", ieee80211w="1",
448                   key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
449                   scan_freq="2412")
450    wpas.dump_monitor()
451    wpas.request("DEAUTHENTICATE " + addr + " test=0")
452    wpas.dump_monitor()
453    wpas.request("DISASSOCIATE " + addr + " test=0")
454    wpas.dump_monitor()
455    ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=1)
456    if ev is not None:
457        raise Exception("Unexpected disconnection")
458
459    wpas.request("DEAUTHENTICATE " + addr + " reason=6 test=0")
460    wpas.dump_monitor()
461    wpas.request("DISASSOCIATE " + addr + " reason=7 test=0")
462    wpas.dump_monitor()
463    ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=1)
464    if ev is not None:
465        raise Exception("Unexpected disconnection")
466    if wt.get_sta_counter("valid_saqueryreq_tx", bssid, addr) < 1:
467        raise Exception("STA did not send SA Query")
468    if wt.get_sta_counter("valid_saqueryresp_rx", bssid, addr) < 1:
469        raise Exception("AP did not reply to SA Query")
470    wpas.dump_monitor()
471
472def test_ap_pmf_sta_sa_query_no_response(dev, apdev):
473    """WPA2-PSK AP with station using SA Query and getting no response"""
474    ssid = "assoc-comeback"
475    addr = dev[0].own_addr()
476
477    wpas = start_wpas_ap(ssid)
478    bssid = wpas.own_addr()
479
480    dev[0].connect(ssid, psk="12345678", ieee80211w="1",
481                   key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
482                   scan_freq="2412")
483    wpas.dump_monitor()
484    wpas.request("DEAUTHENTICATE " + addr + " test=0")
485    wpas.dump_monitor()
486    wpas.request("DISASSOCIATE " + addr + " test=0")
487    wpas.dump_monitor()
488    ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=1)
489    if ev is not None:
490        raise Exception("Unexpected disconnection")
491
492    wpas.request("SET ext_mgmt_frame_handling 1")
493    wpas.request("DEAUTHENTICATE " + addr + " reason=6 test=0")
494    wpas.dump_monitor()
495    wpas.request("DISASSOCIATE " + addr + " reason=7 test=0")
496    wpas.dump_monitor()
497    dev[0].wait_disconnected()
498    wpas.dump_monitor()
499    wpas.request("SET ext_mgmt_frame_handling 0")
500    dev[0].wait_connected()
501    wpas.dump_monitor()
502
503def test_ap_pmf_sta_unprot_deauth_burst(dev, apdev):
504    """WPA2-PSK AP with station receiving burst of unprotected Deauthentication frames"""
505    ssid = "deauth-attack"
506    addr = dev[0].own_addr()
507
508    wpas = start_wpas_ap(ssid)
509    bssid = wpas.own_addr()
510
511    Wlantest.setup(wpas)
512    wt = Wlantest()
513    wt.flush()
514    wt.add_passphrase("12345678")
515
516    dev[0].connect(ssid, psk="12345678", ieee80211w="1",
517                   key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
518                   scan_freq="2412")
519
520    for i in range(0, 10):
521        wpas.request("DEAUTHENTICATE " + addr + " reason=6 test=0")
522        wpas.request("DISASSOCIATE " + addr + " reason=7 test=0")
523    ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=1)
524    if ev is not None:
525        raise Exception("Unexpected disconnection")
526    num_req = wt.get_sta_counter("valid_saqueryreq_tx", bssid, addr)
527    num_resp = wt.get_sta_counter("valid_saqueryresp_rx", bssid, addr)
528    if num_req < 1:
529        raise Exception("STA did not send SA Query")
530    if num_resp < 1:
531        raise Exception("AP did not reply to SA Query")
532    if num_req > 1:
533        raise Exception("STA initiated too many SA Query procedures (%d)" % num_req)
534
535    time.sleep(10)
536    for i in range(0, 5):
537        wpas.request("DEAUTHENTICATE " + addr + " reason=6 test=0")
538        wpas.request("DISASSOCIATE " + addr + " reason=7 test=0")
539    ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=1)
540    if ev is not None:
541        raise Exception("Unexpected disconnection")
542    num_req = wt.get_sta_counter("valid_saqueryreq_tx", bssid, addr)
543    num_resp = wt.get_sta_counter("valid_saqueryresp_rx", bssid, addr)
544    if num_req != 2 or num_resp != 2:
545        raise Exception("Unexpected number of SA Query procedures (req=%d resp=%d)" % (num_req, num_resp))
546
547def test_ap_pmf_sta_sa_query_oom(dev, apdev):
548    """WPA2-PSK AP with station using SA Query (OOM)"""
549    ssid = "assoc-comeback"
550    addr = dev[0].own_addr()
551    wpas = start_wpas_ap(ssid)
552    dev[0].connect(ssid, psk="12345678", ieee80211w="1",
553                   key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
554                   scan_freq="2412")
555    with alloc_fail(dev[0], 1, "=sme_sa_query_timer"):
556        wpas.request("DEAUTHENTICATE " + addr + " reason=6 test=0")
557        wait_fail_trigger(dev[0], "GET_ALLOC_FAIL")
558    dev[0].request("DISCONNECT")
559    wpas.request("DISCONNECT")
560    dev[0].wait_disconnected()
561
562def test_ap_pmf_sta_sa_query_local_failure(dev, apdev):
563    """WPA2-PSK AP with station using SA Query (local failure)"""
564    ssid = "assoc-comeback"
565    addr = dev[0].own_addr()
566    wpas = start_wpas_ap(ssid)
567    dev[0].connect(ssid, psk="12345678", ieee80211w="1",
568                   key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
569                   scan_freq="2412")
570    with fail_test(dev[0], 1, "os_get_random;sme_sa_query_timer"):
571        wpas.request("DEAUTHENTICATE " + addr + " reason=6 test=0")
572        wait_fail_trigger(dev[0], "GET_FAIL")
573    dev[0].request("DISCONNECT")
574    wpas.request("DISCONNECT")
575    dev[0].wait_disconnected()
576
577def test_ap_pmf_sta_sa_query_hostapd(dev, apdev):
578    """WPA2-PSK AP with station using SA Query (hostapd)"""
579    ssid = "assoc-comeback"
580    passphrase = "12345678"
581    addr = dev[0].own_addr()
582
583    params = hostapd.wpa2_params(ssid=ssid, passphrase=passphrase,
584                                 wpa_key_mgmt="WPA-PSK-SHA256",
585                                 ieee80211w="2")
586    hapd = hostapd.add_ap(apdev[0], params)
587    bssid = hapd.own_addr()
588
589    Wlantest.setup(hapd)
590    wt = Wlantest()
591    wt.flush()
592    wt.add_passphrase("12345678")
593
594    dev[0].connect(ssid, psk=passphrase, ieee80211w="2",
595                   key_mgmt="WPA-PSK-SHA256", proto="WPA2",
596                   scan_freq="2412")
597    hapd.wait_sta()
598    if "OK" not in hapd.request("DEAUTHENTICATE " + addr + " test=0") or \
599       "OK" not in hapd.request("DISASSOCIATE " + addr + " test=0"):
600        raise Exception("Failed to send unprotected disconnection messages")
601    ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=1)
602    if ev is not None:
603        raise Exception("Unexpected disconnection")
604
605    if "OK" not in hapd.request("DEAUTHENTICATE " + addr + " reason=6 test=0") or \
606       "OK" not in hapd.request("DISASSOCIATE " + addr + " reason=7 test=0"):
607        raise Exception("Failed to send unprotected disconnection messages (2)")
608    ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=1)
609    if ev is not None:
610        raise Exception("Unexpected disconnection")
611    if wt.get_sta_counter("valid_saqueryreq_tx", bssid, addr) < 1:
612        raise Exception("STA did not send SA Query")
613    if wt.get_sta_counter("valid_saqueryresp_rx", bssid, addr) < 1:
614        raise Exception("AP did not reply to SA Query")
615
616def test_ap_pmf_sta_sa_query_no_response_hostapd(dev, apdev):
617    """WPA2-PSK AP with station using SA Query and getting no response (hostapd)"""
618    ssid = "assoc-comeback"
619    passphrase = "12345678"
620    addr = dev[0].own_addr()
621
622    params = hostapd.wpa2_params(ssid=ssid, passphrase=passphrase,
623                                 wpa_key_mgmt="WPA-PSK-SHA256",
624                                 ieee80211w="2")
625    hapd = hostapd.add_ap(apdev[0], params)
626    bssid = hapd.own_addr()
627
628    Wlantest.setup(hapd)
629    wt = Wlantest()
630    wt.flush()
631    wt.add_passphrase("12345678")
632
633    dev[0].connect(ssid, psk=passphrase, ieee80211w="2",
634                   key_mgmt="WPA-PSK-SHA256", proto="WPA2",
635                   scan_freq="2412")
636    hapd.wait_sta()
637    hapd.set("ext_mgmt_frame_handling", "1")
638    if "OK" not in hapd.request("DEAUTHENTICATE " + addr + " reason=6 test=0") or \
639       "OK" not in hapd.request("DISASSOCIATE " + addr + " reason=7 test=0"):
640        raise Exception("Failed to send unprotected disconnection messages")
641    dev[0].wait_disconnected()
642    hapd.set("ext_mgmt_frame_handling", "0")
643    if wt.get_sta_counter("valid_saqueryreq_tx", bssid, addr) < 1:
644        raise Exception("STA did not send SA Query")
645    if wt.get_sta_counter("valid_saqueryresp_rx", bssid, addr) > 0:
646        raise Exception("AP replied to SA Query")
647    dev[0].wait_connected()
648
649def test_ap_pmf_sta_unprot_deauth_burst_hostapd(dev, apdev):
650    """WPA2-PSK AP with station receiving burst of unprotected Deauthentication frames (hostapd)"""
651    ssid = "deauth-attack"
652    passphrase = "12345678"
653    addr = dev[0].own_addr()
654
655    params = hostapd.wpa2_params(ssid=ssid, passphrase=passphrase,
656                                 wpa_key_mgmt="WPA-PSK-SHA256",
657                                 ieee80211w="2")
658    hapd = hostapd.add_ap(apdev[0], params)
659    bssid = hapd.own_addr()
660
661    Wlantest.setup(hapd)
662    wt = Wlantest()
663    wt.flush()
664    wt.add_passphrase("12345678")
665
666    dev[0].connect(ssid, psk=passphrase, ieee80211w="2",
667                   key_mgmt="WPA-PSK-SHA256", proto="WPA2",
668                   scan_freq="2412")
669    hapd.wait_sta()
670    for i in range(10):
671        if "OK" not in hapd.request("DEAUTHENTICATE " + addr + " reason=6 test=0") or \
672           "OK" not in hapd.request("DISASSOCIATE " + addr + " reason=7 test=0"):
673            raise Exception("Failed to send unprotected disconnection messages")
674    ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=1)
675    if ev is not None:
676        raise Exception("Unexpected disconnection")
677    num_req = wt.get_sta_counter("valid_saqueryreq_tx", bssid, addr)
678    num_resp = wt.get_sta_counter("valid_saqueryresp_rx", bssid, addr)
679    if num_req < 1:
680        raise Exception("STA did not send SA Query")
681    if num_resp < 1:
682        raise Exception("AP did not reply to SA Query")
683    if num_req > 1:
684        raise Exception("STA initiated too many SA Query procedures (%d)" % num_req)
685
686    time.sleep(10)
687    for i in range(5):
688        if "OK" not in hapd.request("DEAUTHENTICATE " + addr + " reason=6 test=0") or \
689           "OK" not in hapd.request("DISASSOCIATE " + addr + " reason=7 test=0"):
690            raise Exception("Failed to send unprotected disconnection messages")
691    ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=1)
692    if ev is not None:
693        raise Exception("Unexpected disconnection")
694    num_req = wt.get_sta_counter("valid_saqueryreq_tx", bssid, addr)
695    num_resp = wt.get_sta_counter("valid_saqueryresp_rx", bssid, addr)
696    if num_req != 2 or num_resp != 2:
697        raise Exception("Unexpected number of SA Query procedures (req=%d resp=%d)" % (num_req, num_resp))
698
699def test_ap_pmf_required_eap(dev, apdev):
700    """WPA2-EAP AP with PMF required"""
701    ssid = "test-pmf-required-eap"
702    params = hostapd.wpa2_eap_params(ssid=ssid)
703    params["wpa_key_mgmt"] = "WPA-EAP-SHA256"
704    params["ieee80211w"] = "2"
705    hapd = hostapd.add_ap(apdev[0], params)
706    key_mgmt = hapd.get_config()['key_mgmt']
707    if key_mgmt.split(' ')[0] != "WPA-EAP-SHA256":
708        raise Exception("Unexpected GET_CONFIG(key_mgmt): " + key_mgmt)
709    dev[0].connect("test-pmf-required-eap", key_mgmt="WPA-EAP-SHA256",
710                   ieee80211w="2", eap="PSK", identity="psk.user@example.com",
711                   password_hex="0123456789abcdef0123456789abcdef",
712                   scan_freq="2412")
713    dev[1].connect("test-pmf-required-eap", key_mgmt="WPA-EAP WPA-EAP-SHA256",
714                   ieee80211w="1", eap="PSK", identity="psk.user@example.com",
715                   password_hex="0123456789abcdef0123456789abcdef",
716                   scan_freq="2412")
717
718def test_ap_pmf_optional_eap(dev, apdev):
719    """WPA2EAP AP with PMF optional"""
720    params = hostapd.wpa2_eap_params(ssid="test-wpa2-eap")
721    params["ieee80211w"] = "1"
722    hapd = hostapd.add_ap(apdev[0], params)
723    dev[0].connect("test-wpa2-eap", key_mgmt="WPA-EAP", eap="TTLS",
724                   identity="pap user", anonymous_identity="ttls",
725                   password="password",
726                   ca_cert="auth_serv/ca.pem", phase2="auth=PAP",
727                   ieee80211w="1", scan_freq="2412")
728    dev[1].connect("test-wpa2-eap", key_mgmt="WPA-EAP WPA-EAP-SHA256",
729                   eap="TTLS", identity="pap user", anonymous_identity="ttls",
730                   password="password",
731                   ca_cert="auth_serv/ca.pem", phase2="auth=PAP",
732                   ieee80211w="2", scan_freq="2412")
733
734@remote_compatible
735def test_ap_pmf_required_sha1(dev, apdev):
736    """WPA2-PSK AP with PMF required with SHA1 AKM"""
737    ssid = "test-pmf-required-sha1"
738    params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
739    params["wpa_key_mgmt"] = "WPA-PSK"
740    params["ieee80211w"] = "2"
741    hapd = hostapd.add_ap(apdev[0], params)
742    Wlantest.setup(hapd)
743    wt = Wlantest()
744    wt.flush()
745    wt.add_passphrase("12345678")
746    key_mgmt = hapd.get_config()['key_mgmt']
747    if key_mgmt.split(' ')[0] != "WPA-PSK":
748        raise Exception("Unexpected GET_CONFIG(key_mgmt): " + key_mgmt)
749    dev[0].connect(ssid, psk="12345678", ieee80211w="2",
750                   key_mgmt="WPA-PSK", proto="WPA2", scan_freq="2412")
751    if "[WPA2-PSK-CCMP]" not in dev[0].request("SCAN_RESULTS"):
752        raise Exception("Scan results missing RSN element info")
753    hwsim_utils.test_connectivity(dev[0], hapd)
754
755@remote_compatible
756def test_ap_pmf_toggle(dev, apdev):
757    """WPA2-PSK AP with PMF optional and changing PMF on reassociation"""
758    try:
759        _test_ap_pmf_toggle(dev, apdev)
760    finally:
761        dev[0].request("SET reassoc_same_bss_optim 0")
762
763def _test_ap_pmf_toggle(dev, apdev):
764    ssid = "test-pmf-optional"
765    params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
766    params["wpa_key_mgmt"] = "WPA-PSK"
767    params["ieee80211w"] = "1"
768    params["assoc_sa_query_max_timeout"] = "1"
769    params["assoc_sa_query_retry_timeout"] = "1"
770    hapd = hostapd.add_ap(apdev[0], params)
771    Wlantest.setup(hapd)
772    wt = Wlantest()
773    wt.flush()
774    wt.add_passphrase("12345678")
775    bssid = apdev[0]['bssid']
776    addr = dev[0].own_addr()
777    dev[0].request("SET reassoc_same_bss_optim 1")
778    id = dev[0].connect(ssid, psk="12345678", ieee80211w="1",
779                        key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
780                        scan_freq="2412")
781    wt.require_ap_pmf_optional(bssid)
782    wt.require_sta_pmf(bssid, addr)
783    sta = hapd.get_sta(addr)
784    if '[MFP]' not in sta['flags']:
785        raise Exception("MFP flag not present for STA")
786
787    dev[0].set_network(id, "ieee80211w", "0")
788    dev[0].request("REASSOCIATE")
789    dev[0].wait_connected()
790    wt.require_sta_no_pmf(bssid, addr)
791    sta = hapd.get_sta(addr)
792    if '[MFP]' in sta['flags']:
793        raise Exception("MFP flag unexpectedly present for STA")
794    err, data = hapd.cmd_execute(['iw', 'dev', apdev[0]['ifname'], 'station',
795                                  'get', addr])
796    if "yes" in [l for l in data.splitlines() if "MFP" in l][0]:
797        raise Exception("Kernel STA entry had MFP enabled")
798
799    dev[0].set_network(id, "ieee80211w", "1")
800    dev[0].request("REASSOCIATE")
801    dev[0].wait_connected()
802    wt.require_sta_pmf(bssid, addr)
803    sta = hapd.get_sta(addr)
804    if '[MFP]' not in sta['flags']:
805        raise Exception("MFP flag not present for STA")
806    err, data = hapd.cmd_execute(['iw', 'dev', apdev[0]['ifname'], 'station',
807                                  'get', addr])
808    if "yes" not in [l for l in data.splitlines() if "MFP" in l][0]:
809        raise Exception("Kernel STA entry did not have MFP enabled")
810
811@remote_compatible
812def test_ap_pmf_required_sta_no_pmf(dev, apdev):
813    """WPA2-PSK AP with PMF required and PMF disabled on STA"""
814    ssid = "test-pmf-required"
815    params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
816    params["wpa_key_mgmt"] = "WPA-PSK-SHA256"
817    params["ieee80211w"] = "2"
818    hapd = hostapd.add_ap(apdev[0], params)
819
820    # Disable PMF on the station and try to connect
821    dev[0].connect(ssid, psk="12345678", ieee80211w="0",
822                   key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
823                   scan_freq="2412", wait_connect=False)
824    ev = dev[0].wait_event(["CTRL-EVENT-NETWORK-NOT-FOUND",
825                            "CTRL-EVENT-ASSOC-REJECT"], timeout=2)
826    if ev is None:
827        raise Exception("No connection result")
828    if "CTRL-EVENT-ASSOC-REJECT" in ev:
829        raise Exception("Tried to connect to PMF required AP without PMF enabled")
830    dev[0].request("REMOVE_NETWORK all")
831
832def test_ap_pmf_inject_auth(dev, apdev):
833    """WPA2-PSK AP with PMF and Authentication frame injection"""
834    ssid = "test-pmf"
835    params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
836    params["wpa_key_mgmt"] = "WPA-PSK-SHA256"
837    params["ieee80211w"] = "2"
838    hapd = hostapd.add_ap(apdev[0], params)
839    dev[0].connect(ssid, psk="12345678", ieee80211w="2",
840                   key_mgmt="WPA-PSK-SHA256", proto="WPA2",
841                   scan_freq="2412")
842    hapd.wait_sta()
843    hwsim_utils.test_connectivity(dev[0], hapd)
844
845    bssid = hapd.own_addr().replace(':', '')
846    addr = dev[0].own_addr().replace(':', '')
847
848    # Inject an unprotected Authentication frame claiming to be from the
849    # associated STA, from another STA, from the AP's own address, from all
850    # zeros and all ones addresses, and from a multicast address.
851    hapd.request("SET ext_mgmt_frame_handling 1")
852    failed = False
853    addresses = [ addr, "021122334455", bssid, 6*"00", 6*"ff", 6*"01" ]
854    for a in addresses:
855        auth = "b0003a01" + bssid + a + bssid + '1000000001000000'
856        res = hapd.request("MGMT_RX_PROCESS freq=2412 datarate=0 ssi_signal=-30 frame=%s" % auth)
857        if "OK" not in res:
858            failed = True
859    hapd.request("SET ext_mgmt_frame_handling 0")
860    if failed:
861        raise Exception("MGMT_RX_PROCESS failed")
862    time.sleep(0.1)
863
864    ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=0.1)
865    if ev:
866        raise Exception("Unexpected disconnection reported on the STA")
867
868    # Verify that original association is still functional.
869    hwsim_utils.test_connectivity(dev[0], hapd)
870
871    # Inject an unprotected Association Request frame (with and without RSNE)
872    # claiming to be from the set of test addresses.
873    hapd.request("SET ext_mgmt_frame_handling 1")
874    for a in addresses:
875        assoc = "00003a01" + bssid + a + bssid + '2000' + '31040500' + '0008746573742d706d66' + '010802040b160c121824' + '301a0100000fac040100000fac040100000fac06c0000000000fac06'
876        res = hapd.request("MGMT_RX_PROCESS freq=2412 datarate=0 ssi_signal=-30 frame=%s" % assoc)
877        if "OK" not in res:
878            failed = True
879
880        assoc = "00003a01" + bssid + a + bssid + '2000' + '31040500' + '0008746573742d706d66' + '010802040b160c121824' + '3000'
881        res = hapd.request("MGMT_RX_PROCESS freq=2412 datarate=0 ssi_signal=-30 frame=%s" % assoc)
882        if "OK" not in res:
883            failed = True
884
885        assoc = "00003a01" + bssid + a + bssid + '2000' + '31040500' + '0008746573742d706d66' + '010802040b160c121824'
886        res = hapd.request("MGMT_RX_PROCESS freq=2412 datarate=0 ssi_signal=-30 frame=%s" % assoc)
887        if "OK" not in res:
888            failed = True
889    hapd.request("SET ext_mgmt_frame_handling 0")
890    if failed:
891        raise Exception("MGMT_RX_PROCESS failed")
892    time.sleep(5)
893
894    ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=0.1)
895    if ev:
896        raise Exception("Unexpected disconnection reported on the STA")
897
898    # Verify that original association is still functional.
899    hwsim_utils.test_connectivity(dev[0], hapd)
900
901def test_ap_pmf_inject_data(dev, apdev):
902    """WPA2-PSK AP with PMF and Data frame injection"""
903    try:
904        run_ap_pmf_inject_data(dev, apdev)
905    finally:
906        stop_monitor(apdev[1]["ifname"])
907
908def run_ap_pmf_inject_data(dev, apdev):
909    ssid = "test-pmf"
910    params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
911    params["wpa_key_mgmt"] = "WPA-PSK-SHA256"
912    params["ieee80211w"] = "2"
913    hapd = hostapd.add_ap(apdev[0], params)
914    dev[0].connect(ssid, psk="12345678", ieee80211w="2",
915                   key_mgmt="WPA-PSK-SHA256", proto="WPA2",
916                   scan_freq="2412")
917    hapd.wait_sta()
918    hwsim_utils.test_connectivity(dev[0], hapd)
919
920    sock = start_monitor(apdev[1]["ifname"])
921    radiotap = radiotap_build()
922
923    bssid = hapd.own_addr().replace(':', '')
924    addr = dev[0].own_addr().replace(':', '')
925
926    # Inject Data frame with A2=broadcast, A2=multicast, A2=BSSID, A2=STA, and
927    # A2=unknown unicast
928    addresses = [ 6*"ff", 6*"01", bssid, addr, "020102030405" ]
929    for a in addresses:
930        frame = binascii.unhexlify("48010000" + bssid + a + bssid + "0000")
931        sock.send(radiotap + frame)
932
933    time.sleep(0.1)
934    ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=0.1)
935    if ev:
936        raise Exception("Unexpected disconnection reported on the STA")
937    hwsim_utils.test_connectivity(dev[0], hapd)
938
939def test_ap_pmf_tkip_reject(dev, apdev):
940    """Mixed mode BSS and MFP-enabled AP rejecting TKIP"""
941    skip_without_tkip(dev[0])
942    params = hostapd.wpa2_params(ssid="test-pmf", passphrase="12345678")
943    params['wpa'] = '3'
944    params["ieee80211w"] = "1"
945    params["wpa_pairwise"] = "TKIP CCMP"
946    params["rsn_pairwise"] = "TKIP CCMP"
947    hostapd.add_ap(apdev[0], params)
948
949    dev[0].connect("test-pmf", psk="12345678", pairwise="CCMP", ieee80211w="2",
950                   scan_freq="2412")
951    dev[0].dump_monitor()
952
953    dev[1].connect("test-pmf", psk="12345678", proto="WPA", pairwise="TKIP",
954                   ieee80211w="0", scan_freq="2412")
955    dev[1].dump_monitor()
956
957    dev[2].connect("test-pmf", psk="12345678", pairwise="TKIP",
958                   ieee80211w="2", scan_freq="2412", wait_connect=False)
959    ev = dev[2].wait_event(["CTRL-EVENT-CONNECTED",
960                            "CTRL-EVENT-ASSOC-REJECT"], timeout=10)
961    if ev is None:
962        raise Exception("No connection result reported")
963    if "CTRL-EVENT-ASSOC-REJECT" not in ev:
964        raise Exception("MFP + TKIP connection was not rejected")
965    if "status_code=31" not in ev:
966        raise Exception("Unexpected status code in rejection: " + ev)
967    dev[2].request("DISCONNECT")
968    dev[2].dump_monitor()
969
970def test_ap_pmf_sa_query_timeout(dev, apdev):
971    """SA Query timeout"""
972    ssid = "test-pmf-required"
973    params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
974    params["wpa_key_mgmt"] = "WPA-PSK-SHA256"
975    params["ieee80211w"] = "2"
976    hapd = hostapd.add_ap(apdev[0], params)
977    dev[0].connect(ssid, psk="12345678", ieee80211w="2",
978                   key_mgmt="WPA-PSK-SHA256", proto="WPA2",
979                   scan_freq="2412")
980
981    hapd.set("ext_mgmt_frame_handling", "1")
982    if "OK" not in dev[0].request("UNPROT_DEAUTH"):
983        raise Exception("Triggering SA Query from the STA failed")
984    ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=2)
985    if ev is None:
986        raise Exception("No disconnection on SA Query timeout seen")
987    hapd.set("ext_mgmt_frame_handling", "0")
988    dev[0].wait_connected()
989    dev[0].dump_monitor()
990
991    hapd.set("ext_mgmt_frame_handling", "1")
992    if "OK" not in dev[0].request("UNPROT_DEAUTH"):
993        raise Exception("Triggering SA Query from the STA failed")
994    ev = hapd.mgmt_rx()
995    hapd.set("ext_mgmt_frame_handling", "0")
996    dev[0].request("DISCONNECT")
997    dev[0].wait_disconnected()
998    dev[0].request("RECONNECT")
999    dev[0].wait_connected()
1000    hapd.set("ext_mgmt_frame_handling", "1")
1001    ev = dev[0].wait_event(["CTRL-EVENT-DISCONNECTED"], timeout=1.5)
1002    if ev is not None:
1003        raise Exception("Unexpected disconnection after reconnection seen")
1004
1005def mac80211_read_key(keydir):
1006    vals = {}
1007    for name in os.listdir(keydir):
1008        try:
1009            with open(os.path.join(keydir, name)) as f:
1010                vals[name] = f.read().strip()
1011        except OSError as e:
1012            pass
1013    return vals
1014
1015def check_mac80211_bigtk(dev, hapd):
1016    sta_key = None
1017    ap_key = None
1018
1019    phy = dev.get_driver_status_field("phyname")
1020    keys = "/sys/kernel/debug/ieee80211/%s/keys" % phy
1021    try:
1022        for key in os.listdir(keys):
1023            keydir = os.path.join(keys, key)
1024            vals = mac80211_read_key(keydir)
1025            keyidx = int(vals['keyidx'])
1026            if keyidx == 6 or keyidx == 7:
1027                sta_key = vals;
1028                break
1029    except OSError as e:
1030        raise HwsimSkip("debugfs not supported in mac80211 (STA)")
1031
1032    phy = hapd.get_driver_status_field("phyname")
1033    keys = "/sys/kernel/debug/ieee80211/%s/keys" % phy
1034    try:
1035        for key in os.listdir(keys):
1036            keydir = os.path.join(keys, key)
1037            vals = mac80211_read_key(keydir)
1038            keyidx = int(vals['keyidx'])
1039            if keyidx == 6 or keyidx == 7:
1040                ap_key = vals;
1041                break
1042    except OSError as e:
1043        raise HwsimSkip("debugfs not supported in mac80211 (AP)")
1044
1045    if not sta_key:
1046        raise Exception("Could not find STA key information from debugfs")
1047    logger.info("STA key: " + str(sta_key))
1048
1049    if not ap_key:
1050        raise Exception("Could not find AP key information from debugfs")
1051    logger.info("AP key: " + str(ap_key))
1052
1053    if sta_key['key'] != ap_key['key']:
1054        raise Exception("AP and STA BIGTK mismatch")
1055
1056    if sta_key['keyidx'] != ap_key['keyidx']:
1057        raise Exception("AP and STA BIGTK keyidx mismatch")
1058
1059    if sta_key['algorithm'] != ap_key['algorithm']:
1060        raise Exception("AP and STA BIGTK algorithm mismatch")
1061
1062    replays = int(sta_key['replays'])
1063    icverrors = int(sta_key['icverrors'])
1064    if replays > 0 or icverrors > 0:
1065        raise Exception("STA reported errors: replays=%d icverrors=%d" % replays, icverrors)
1066
1067    rx_spec = int(sta_key['rx_spec'], base=16)
1068    if rx_spec < 3:
1069        raise Exception("STA did not update BIGTK receive counter sufficiently")
1070
1071    tx_spec = int(ap_key['tx_spec'], base=16)
1072    if tx_spec < 3:
1073        raise Exception("AP did not update BIGTK BIPN sufficiently")
1074
1075def test_ap_pmf_beacon_protection_bip(dev, apdev):
1076    """WPA2-PSK Beacon protection (BIP)"""
1077    run_ap_pmf_beacon_protection(dev, apdev, "AES-128-CMAC")
1078
1079def test_ap_pmf_beacon_protection_bip_cmac_256(dev, apdev):
1080    """WPA2-PSK Beacon protection (BIP-CMAC-256)"""
1081    run_ap_pmf_beacon_protection(dev, apdev, "BIP-CMAC-256")
1082
1083def test_ap_pmf_beacon_protection_bip_gmac_128(dev, apdev):
1084    """WPA2-PSK Beacon protection (BIP-GMAC-128)"""
1085    run_ap_pmf_beacon_protection(dev, apdev, "BIP-GMAC-128")
1086
1087def test_ap_pmf_beacon_protection_bip_gmac_256(dev, apdev):
1088    """WPA2-PSK Beacon protection (BIP-GMAC-256)"""
1089    run_ap_pmf_beacon_protection(dev, apdev, "BIP-GMAC-256")
1090
1091def run_ap_pmf_beacon_protection(dev, apdev, cipher):
1092    ssid = "test-beacon-prot"
1093    params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
1094    params["wpa_key_mgmt"] = "WPA-PSK-SHA256"
1095    params["ieee80211w"] = "2"
1096    params["beacon_prot"] = "1"
1097    params["group_mgmt_cipher"] = cipher
1098    try:
1099        hapd = hostapd.add_ap(apdev[0], params)
1100    except Exception as e:
1101        if "Failed to enable hostapd interface" in str(e):
1102            raise HwsimSkip("Beacon protection not supported")
1103        raise
1104
1105    bssid = hapd.own_addr()
1106
1107    Wlantest.setup(hapd)
1108    wt = Wlantest()
1109    wt.flush()
1110    wt.add_passphrase("12345678")
1111
1112    # STA with Beacon protection enabled
1113    dev[0].connect(ssid, psk="12345678", ieee80211w="2", beacon_prot="1",
1114                   key_mgmt="WPA-PSK-SHA256", proto="WPA2", scan_freq="2412")
1115
1116    # STA with Beacon protection disabled
1117    dev[1].connect(ssid, psk="12345678", ieee80211w="2",
1118                   key_mgmt="WPA-PSK-SHA256", proto="WPA2", scan_freq="2412")
1119
1120    time.sleep(1)
1121    check_mac80211_bigtk(dev[0], hapd)
1122
1123    valid_bip = wt.get_bss_counter('valid_bip_mmie', bssid)
1124    invalid_bip = wt.get_bss_counter('invalid_bip_mmie', bssid)
1125    missing_bip = wt.get_bss_counter('missing_bip_mmie', bssid)
1126    logger.info("wlantest BIP counters: valid=%d invalid=%d missing=%d" % (valid_bip, invalid_bip, missing_bip))
1127    if valid_bip < 0 or invalid_bip > 0 or missing_bip > 0:
1128        raise Exception("Unexpected wlantest BIP counters: valid=%d invalid=%d missing=%d" % (valid_bip, invalid_bip, missing_bip))
1129
1130def test_ap_pmf_beacon_protection_mismatch(dev, apdev):
1131    """WPA2-PSK Beacon protection MIC mismatch"""
1132    run_ap_pmf_beacon_protection_mismatch(dev, apdev, False)
1133
1134def test_ap_pmf_beacon_protection_missing(dev, apdev):
1135    """WPA2-PSK Beacon protection MME missing"""
1136    run_ap_pmf_beacon_protection_mismatch(dev, apdev, True)
1137
1138def run_ap_pmf_beacon_protection_mismatch(dev, apdev, clear):
1139    ssid = "test-beacon-prot"
1140    params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
1141    params["wpa_key_mgmt"] = "WPA-PSK-SHA256"
1142    params["ieee80211w"] = "2"
1143    params["beacon_prot"] = "1"
1144    params["group_mgmt_cipher"] = "AES-128-CMAC"
1145    try:
1146        hapd = hostapd.add_ap(apdev[0], params)
1147    except Exception as e:
1148        if "Failed to enable hostapd interface" in str(e):
1149            raise HwsimSkip("Beacon protection not supported")
1150        raise
1151
1152    bssid = hapd.own_addr()
1153
1154    Wlantest.setup(hapd)
1155    wt = Wlantest()
1156    wt.flush()
1157    wt.add_passphrase("12345678")
1158
1159    dev[0].connect(ssid, psk="12345678", ieee80211w="2", beacon_prot="1",
1160                   key_mgmt="WPA-PSK-SHA256", proto="WPA2", scan_freq="2412")
1161
1162    WPA_ALG_NONE = 0
1163    WPA_ALG_IGTK = 4
1164    KEY_FLAG_DEFAULT = 0x02
1165    KEY_FLAG_TX = 0x08
1166    KEY_FLAG_GROUP = 0x10
1167    KEY_FLAG_GROUP_TX_DEFAULT = KEY_FLAG_GROUP | KEY_FLAG_TX | KEY_FLAG_DEFAULT
1168
1169    addr = "ff:ff:ff:ff:ff:ff"
1170
1171    if clear:
1172        res = hapd.request("SET_KEY %d %s %d %d %s %s %d" % (WPA_ALG_NONE, addr, 6, 1, 6*"00", "", KEY_FLAG_GROUP))
1173    else:
1174        res = hapd.request("SET_KEY %d %s %d %d %s %s %d" % (WPA_ALG_IGTK, addr, 6, 1, 6*"00", 16*"00", KEY_FLAG_GROUP_TX_DEFAULT))
1175    if "OK" not in res:
1176        raise Exception("SET_KEY failed")
1177
1178    ev = dev[0].wait_event(["CTRL-EVENT-UNPROT-BEACON"], timeout=5)
1179    if ev is None:
1180        raise Exception("Unprotected Beacon frame not reported")
1181
1182    ev = dev[0].wait_event(["CTRL-EVENT-BEACON-LOSS"], timeout=5)
1183    if ev is None:
1184        raise Exception("Beacon loss not reported")
1185
1186    ev = hapd.wait_event(["CTRL-EVENT-UNPROT-BEACON"], timeout=5)
1187    if ev is None:
1188        raise Exception("WNM-Notification Request frame not reported")
1189
1190def test_ap_pmf_sta_global_require(dev, apdev):
1191    """WPA2-PSK AP with PMF optional and wpa_supplicant pmf=2"""
1192    ssid = "test-pmf-optional"
1193    params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
1194    params["wpa_key_mgmt"] = "WPA-PSK"
1195    params["ieee80211w"] = "1"
1196    hapd = hostapd.add_ap(apdev[0], params)
1197    try:
1198        dev[0].set("pmf", "2")
1199        dev[0].connect(ssid, psk="12345678",
1200                       key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
1201                       scan_freq="2412")
1202        pmf = dev[0].get_status_field("pmf")
1203        if pmf != "1":
1204            raise Exception("Unexpected PMF state: " + str(pmf))
1205    finally:
1206        dev[0].set("pmf", "0")
1207
1208def test_ap_pmf_sta_global_require2(dev, apdev):
1209    """WPA2-PSK AP with PMF optional and wpa_supplicant pmf=2 (2)"""
1210    ssid = "test-pmf-optional"
1211    params = hostapd.wpa2_params(ssid=ssid, passphrase="12345678")
1212    params["wpa_key_mgmt"] = "WPA-PSK"
1213    params["ieee80211w"] = "0"
1214    hapd = hostapd.add_ap(apdev[0], params)
1215    bssid = hapd.own_addr()
1216    try:
1217        dev[0].scan_for_bss(bssid, freq=2412)
1218        dev[0].set("pmf", "2")
1219        dev[0].connect(ssid, psk="12345678",
1220                       key_mgmt="WPA-PSK WPA-PSK-SHA256", proto="WPA2",
1221                       scan_freq="2412", wait_connect=False)
1222        ev = dev[0].wait_event(["CTRL-EVENT-CONNECTED",
1223                                "CTRL-EVENT-NETWORK-NOT-FOUND"], timeout=10)
1224        if ev is None:
1225            raise Exception("Connection result not reported")
1226        if "CTRL-EVENT-CONNECTED" in ev:
1227            raise Exception("Unexpected connection")
1228    finally:
1229        dev[0].set("pmf", "0")
1230