aboutsummaryrefslogtreecommitdiffstats
path: root/tests/hwsim/test_wpas_wmm_ac.py
blob: f81588a1824e402985e35a87403ef942fb71c860 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
# Test cases for wpa_supplicant WMM-AC operations
# Copyright (c) 2014, Intel Corporation
#
# This software may be distributed under the terms of the BSD license.
# See README for more details.

from remotehost import remote_compatible
import logging
logger = logging.getLogger()
import struct

import hwsim_utils
import hostapd
from utils import fail_test

def add_wmm_ap(apdev, acm_list):
    params = { "ssid": "wmm_ac",
               "hw_mode": "g",
               "channel": "11",
               "wmm_enabled" : "1"}

    for ac in acm_list:
        params["wmm_ac_%s_acm" % (ac.lower())] = "1"

    return hostapd.add_ap(apdev, params)

def test_tspec(dev, apdev):
    """Basic addts/delts tests"""
    # configure ap with VO and VI requiring admission-control
    hapd = add_wmm_ap(apdev[0], ["VO", "VI"])
    dev[0].connect("wmm_ac", key_mgmt="NONE", scan_freq="2462")
    hwsim_utils.test_connectivity(dev[0], hapd)
    status = dev[0].request("WMM_AC_STATUS")
    if "WMM AC is Enabled" not in status:
        raise Exception("WMM-AC not enabled")
    if "TSID" in status:
        raise Exception("Unexpected TSID info")
    if "BK: acm=0 uapsd=0" not in status:
        raise Exception("Unexpected BK info" + status)
    if "BE: acm=0 uapsd=0" not in status:
        raise Exception("Unexpected BE info" + status)
    if "VI: acm=1 uapsd=0" not in status:
        raise Exception("Unexpected VI info" + status)
    if "VO: acm=1 uapsd=0" not in status:
        raise Exception("Unexpected VO info" + status)

    # no tsid --> tsid out of range
    if "FAIL" not in dev[0].request("WMM_AC_ADDTS downlink"):
        raise Exception("Invalid WMM_AC_ADDTS accepted")
    # no direction
    if "FAIL" not in dev[0].request("WMM_AC_ADDTS tsid=5"):
        raise Exception("Invalid WMM_AC_ADDTS accepted")
    # param out of range
    if "FAIL" not in dev[0].request("WMM_AC_ADDTS tsid=5 downlink"):
        raise Exception("Invalid WMM_AC_ADDTS accepted")

    tsid = 5

    # make sure we fail when the ac is not configured for acm
    try:
        dev[0].add_ts(tsid, 3)
        raise Exception("ADDTS succeeded although it should have failed")
    except Exception as e:
        if not str(e).startswith("ADDTS failed"):
            raise
    status = dev[0].request("WMM_AC_STATUS")
    if "TSID" in status:
        raise Exception("Unexpected TSID info")

    # add tspec for UP=6
    dev[0].add_ts(tsid, 6)
    status = dev[0].request("WMM_AC_STATUS")
    if "TSID" not in status:
        raise Exception("Missing TSID info")

    # using the same tsid for a different ac is invalid
    try:
        dev[0].add_ts(tsid, 5)
        raise Exception("ADDTS succeeded although it should have failed")
    except Exception as e:
        if not str(e).startswith("ADDTS failed"):
            raise

    # update the tspec for a different UP of the same ac
    dev[0].add_ts(tsid, 7, extra="fixed_nominal_msdu")
    dev[0].del_ts(tsid)
    status = dev[0].request("WMM_AC_STATUS")
    if "TSID" in status:
        raise Exception("Unexpected TSID info")

    # verify failure on uplink/bidi without driver support
    tsid = 6
    try:
        dev[0].add_ts(tsid, 7, direction="uplink")
        raise Exception("ADDTS succeeded although it should have failed")
    except Exception as e:
        if not str(e).startswith("ADDTS failed"):
            raise
    try:
        dev[0].add_ts(tsid, 7, direction="bidi")
        raise Exception("ADDTS succeeded although it should have failed")
    except Exception as e:
        if not str(e).startswith("ADDTS failed"):
            raise

    # attempt to delete non-existing tsid
    try:
        dev[0].del_ts(tsid)
        raise Exception("DELTS succeeded although it should have failed")
    except Exception as e:
        if not str(e).startswith("DELTS failed"):
            raise

    # "CTRL: Invalid WMM_AC_ADDTS parameter: 'foo'
    if "FAIL" not in dev[0].request("WMM_AC_ADDTS foo"):
        raise Exception("Invalid WMM_AC_ADDTS command accepted")

def test_tspec_protocol(dev, apdev):
    """Protocol tests for addts/delts"""
    # configure ap with VO and VI requiring admission-control
    hapd = add_wmm_ap(apdev[0], ["VO", "VI"])
    dev[0].connect("wmm_ac", key_mgmt="NONE", scan_freq="2462")

    dev[0].dump_monitor()
    hapd.set("ext_mgmt_frame_handling", "1")

    tsid = 6

    # timeout on ADDTS response
    dev[0].add_ts(tsid, 7, expect_failure=True)

    hapd.dump_monitor()
    req = "WMM_AC_ADDTS downlink tsid=6 up=7 nominal_msdu_size=1500 sba=9000 mean_data_rate=1500 min_phy_rate=6000000"
    if "OK" not in dev[0].request(req):
        raise Exception("WMM_AC_ADDTS failed")
    # a new request while previous is still pending
    if "FAIL" not in dev[0].request(req):
        raise Exception("WMM_AC_ADDTS accepted while oen was still pending")
    msg = hapd.mgmt_rx()
    payload = msg['payload']
    (categ, action, dialog, status) = struct.unpack('BBBB', payload[0:4])
    if action != 0:
        raise Exception("Unexpected Action code: %d" % action)

    msg['da'] = msg['sa']
    msg['sa'] = apdev[0]['bssid']

    # unexpected dialog token
    msg['payload'] = struct.pack('BBBB', 17, 1, (dialog + 1) & 0xff, 0) + payload[4:]
    hapd.mgmt_tx(msg)

    # valid response
    msg['payload'] = struct.pack('BBBB', 17, 1, dialog, 0) + payload[4:]
    hapd.mgmt_tx(msg)
    ev = dev[0].wait_event(["TSPEC-ADDED"], timeout=10)
    if ev is None:
        raise Exception("Timeout on TSPEC-ADDED")
    if "tsid=%d" % tsid not in ev:
        raise Exception("Unexpected TSPEC-ADDED contents: " + ev)

    # duplicated response
    msg['payload'] = struct.pack('BBBB', 17, 1, dialog, 0) + payload[4:]
    hapd.mgmt_tx(msg)

    # too short ADDTS
    msg['payload'] = struct.pack('BBBB', 17, 1, dialog, 0)
    hapd.mgmt_tx(msg)

    # invalid IE
    msg['payload'] = struct.pack('BBBB', 17, 1, dialog, 0) + payload[4:] + struct.pack('BB', 0xdd, 100)
    hapd.mgmt_tx(msg)

    # too short WMM element
    msg['payload'] = struct.pack('BBBB', 17, 1, dialog, 0) + payload[4:] + '\xdd\x06\x00\x50\xf2\x02\x02\x01'
    hapd.mgmt_tx(msg)

    # DELTS
    dev[0].dump_monitor()
    msg['payload'] = struct.pack('BBBB', 17, 2, 0, 0) + payload[4:]
    hapd.mgmt_tx(msg)
    ev = dev[0].wait_event(['TSPEC-REMOVED'], timeout=6)
    if ev is None:
        raise Exception("Timeout on TSPEC-REMOVED event")
    if "tsid=%d" % tsid not in ev:
        raise Exception("Unexpected TSPEC-REMOVED contents: " + ev)
    # DELTS duplicated
    msg['payload'] = struct.pack('BBBB', 17, 2, 0, 0) + payload[4:]
    hapd.mgmt_tx(msg)

    # start a new request
    hapd.dump_monitor()
    if "OK" not in dev[0].request(req):
        raise Exception("WMM_AC_ADDTS failed")
    msg = hapd.mgmt_rx()
    payload = msg['payload']
    (categ, action, dialog, status) = struct.unpack('BBBB', payload[0:4])
    if action != 0:
        raise Exception("Unexpected Action code: %d" % action)

    msg['da'] = msg['sa']
    msg['sa'] = apdev[0]['bssid']

    # modified parameters
    msg['payload'] = struct.pack('BBBB', 17, 1, dialog, 1) + payload[4:12] + struct.pack('B', ord(payload[12]) & ~0x60) + payload[13:]
    hapd.mgmt_tx(msg)

    # reject request
    msg['payload'] = struct.pack('BBBB', 17, 1, dialog, 1) + payload[4:]
    hapd.mgmt_tx(msg)
    ev = dev[0].wait_event(["TSPEC-REQ-FAILED"], timeout=10)
    if ev is None:
        raise Exception("Timeout on TSPEC-REQ-FAILED")
    if "tsid=%d" % tsid not in ev:
        raise Exception("Unexpected TSPEC-REQ-FAILED contents: " + ev)

    hapd.set("ext_mgmt_frame_handling", "0")

@remote_compatible
def test_tspec_not_enabled(dev, apdev):
    """addts failing if AP does not support WMM"""
    params = { "ssid": "wmm_no_ac",
               "hw_mode": "g",
               "channel": "11",
               "wmm_enabled" : "0" }
    hapd = hostapd.add_ap(apdev[0], params)
    dev[0].connect("wmm_no_ac", key_mgmt="NONE", scan_freq="2462")
    status = dev[0].request("WMM_AC_STATUS")
    if "Not associated to a WMM AP, WMM AC is Disabled" not in status:
        raise Exception("Unexpected WMM_AC_STATUS: " + status)

    try:
        dev[0].add_ts(5, 6)
        raise Exception("ADDTS succeeded although it should have failed")
    except Exception as e:
        if not str(e).startswith("ADDTS failed"):
            raise

    # attempt to delete non-existing tsid
    try:
        dev[0].del_ts(5)
        raise Exception("DELTS succeeded although it should have failed")
    except Exception as e:
        if not str(e).startswith("DELTS failed"):
            raise

    # unexpected Action frame when WMM is disabled
    MGMT_SUBTYPE_ACTION = 13
    msg = {}
    msg['fc'] = MGMT_SUBTYPE_ACTION << 4
    msg['da'] = dev[0].p2p_interface_addr()
    msg['sa'] = apdev[0]['bssid']
    msg['bssid'] = apdev[0]['bssid']
    msg['payload'] = struct.pack('BBBB', 17, 2, 0, 0)
    hapd.mgmt_tx(msg)

@remote_compatible
def test_tspec_ap_roam_open(dev, apdev):
    """Roam between two open APs while having tspecs"""
    hapd0 = add_wmm_ap(apdev[0], ["VO", "VI"])
    dev[0].connect("wmm_ac", key_mgmt="NONE")
    hwsim_utils.test_connectivity(dev[0], hapd0)
    dev[0].add_ts(5, 6)

    hapd1 = add_wmm_ap(apdev[1], ["VO", "VI"])
    dev[0].scan_for_bss(apdev[1]['bssid'], freq=2462)
    dev[0].roam(apdev[1]['bssid'])
    hwsim_utils.test_connectivity(dev[0], hapd1)
    if dev[0].tspecs():
        raise Exception("TSPECs weren't deleted on roaming")

    dev[0].scan_for_bss(apdev[0]['bssid'], freq=2462)
    dev[0].roam(apdev[0]['bssid'])
    hwsim_utils.test_connectivity(dev[0], hapd0)

@remote_compatible
def test_tspec_reassoc(dev, apdev):
    """Reassociation to same BSS while having tspecs"""
    hapd0 = add_wmm_ap(apdev[0], ["VO", "VI"])
    dev[0].connect("wmm_ac", key_mgmt="NONE")
    hwsim_utils.test_connectivity(dev[0], hapd0)
    dev[0].add_ts(5, 6)
    last_tspecs = dev[0].tspecs()

    dev[0].request("REASSOCIATE")
    dev[0].wait_connected()

    hwsim_utils.test_connectivity(dev[0], hapd0)
    if dev[0].tspecs() != last_tspecs:
        raise Exception("TSPECs weren't saved on reassociation")

def test_wmm_element(dev, apdev):
    """hostapd FTM range request timeout"""
    try:
        run_wmm_element(dev, apdev)
    finally:
        dev[0].request("VENDOR_ELEM_REMOVE 13 *")

def run_wmm_element(dev, apdev):
    params = { "ssid": "wmm" }
    hapd = hostapd.add_ap(apdev[0]['ifname'], params)
    bssid = hapd.own_addr()

    # Too short WMM IE
    dev[0].request("VENDOR_ELEM_ADD 13 dd060050f2020001")
    dev[0].scan_for_bss(bssid, freq=2412)
    dev[0].connect("wmm", key_mgmt="NONE", scan_freq="2412", wait_connect=False)
    ev = dev[0].wait_event(["CTRL-EVENT-ASSOC-REJECT"], timeout=10)
    if ev is None:
        raise Exception("Association not rejected")
    dev[0].request("REMOVE_NETWORK all")

    # Unsupported WMM IE Subtype/Version
    dev[0].request("VENDOR_ELEM_ADD 13 dd070050f202000000")
    dev[0].connect("wmm", key_mgmt="NONE", scan_freq="2412", wait_connect=False)
    ev = dev[0].wait_event(["CTRL-EVENT-ASSOC-REJECT"], timeout=10)
    if ev is None:
        raise Exception("Association not rejected")
    dev[0].request("REMOVE_NETWORK all")

    # Unsupported WMM IE Subtype/Version
    dev[0].request("VENDOR_ELEM_ADD 13 dd070050f202010100")
    dev[0].connect("wmm", key_mgmt="NONE", scan_freq="2412", wait_connect=False)
    ev = dev[0].wait_event(["CTRL-EVENT-ASSOC-REJECT"], timeout=10)
    if ev is None:
        raise Exception("Association not rejected")
    dev[0].request("REMOVE_NETWORK all")

def test_tspec_ap_fail(dev, apdev):
    """AP failing to send tspec response"""
    # configure ap with VO and VI requiring admission-control
    hapd = add_wmm_ap(apdev[0], ["VO", "VI"])
    dev[0].connect("wmm_ac", key_mgmt="NONE", scan_freq="2462")
    tsid = 5

    with fail_test(hapd, 1, "wmm_send_action"):
        try:
            # add tspec for UP=6
            dev[0].add_ts(tsid, 6)
        except:
            pass

def test_tspec_ap_parsing(dev, apdev):
    """TSPEC AP parsing tests"""
    # configure ap with VO and VI requiring admission-control
    hapd = add_wmm_ap(apdev[0], ["VO", "VI"])
    bssid = hapd.own_addr()
    dev[0].connect("wmm_ac", key_mgmt="NONE", scan_freq="2462")
    addr = dev[0].own_addr()

    tests = [ "WMM_AC_ADDTS downlink tsid=5 up=6 nominal_msdu_size=1500 sba=9000 mean_data_rate=1500 min_phy_rate=600000",
              "WMM_AC_ADDTS downlink tsid=5 up=6 nominal_msdu_size=1500 sba=8192 mean_data_rate=1500 min_phy_rate=6000000",
              "WMM_AC_ADDTS downlink tsid=5 up=6 nominal_msdu_size=32767 sba=65535 mean_data_rate=1500 min_phy_rate=1000000",
              "WMM_AC_ADDTS downlink tsid=5 up=6 nominal_msdu_size=10000 sba=65535 mean_data_rate=2147483647 min_phy_rate=1000000" ]
    for t in tests:
        if "OK" not in dev[0].request(t):
            raise Exception("WMM_AC_ADDTS failed")
        ev = dev[0].wait_event(["TSPEC-REQ-FAILED"], timeout=1)
        if ev is None:
            raise Exception("No response")

    tests = []
    # WMM: Invalid Nominal MSDU Size (0)
    tests += [ "11000400dd3d0050f2020201aa300000000000000000000000000000000000000000000000000000000000ffffff7f00000000000000000000000040420f00ffff0000" ]
    # hostapd_wmm_action - missing or wrong length tspec
    tests += [ "11000400dd3e0050f2020201aa300010270000000000000000000000000000000000000000000000000000ffffff7f00000000000000000000000040420f00ffff000000" ]
    # hostapd_wmm_action - could not parse wmm action
    tests += [ "11000400dd3d0050f2020201aa300010270000000000000000000000000000000000000000000000000000ffffff7f00000000000000000000000040420f00ffff00" ]
    # valid form
    tests += [ "11000400dd3d0050f2020201aa300010270000000000000000000000000000000000000000000000000000ffffff7f00000000000000000000000040420f00ffff0000" ]

    hdr = "d0003a01" + bssid.replace(':', '') + addr.replace(':', '') + bssid.replace(':', '') + "1000"
    hapd.set("ext_mgmt_frame_handling", "1")
    for t in tests:
        if "OK" not in hapd.request("MGMT_RX_PROCESS freq=2412 datarate=0 ssi_signal=-30 frame=" + hdr + t):
            raise Exception("MGMT_RX_PROCESS failed")

    hapd.set("ext_mgmt_frame_handling", "0")

def test_wmm_disabled(dev, apdev):
    """WMM disabled and unexpected TSPEC"""
    params = { "ssid": "no-wmm", "ieee80211n": "0", "wmm_enabled": "0" }
    hapd = hostapd.add_ap(apdev[0]['ifname'], params)
    bssid = hapd.own_addr()
    dev[0].connect("no-wmm", key_mgmt="NONE", scan_freq="2412")
    addr = dev[0].own_addr()

    # wmm action received is not from associated wmm station
    hdr = "d0003a01" + bssid.replace(':', '') + addr.replace(':', '') + bssid.replace(':', '') + "1000"
    hapd.set("ext_mgmt_frame_handling", "1")
    if "OK" not in hapd.request("MGMT_RX_PROCESS freq=2412 datarate=0 ssi_signal=-30 frame=" + hdr + "11000400dd3d0050f2020201aa300010270000000000000000000000000000000000000000000000000000ffffff7f00000000000000000000000040420f00ffff0000"):
        raise Exception("MGMT_RX_PROCESS failed")

    # IEEE 802.11: Ignored Action frame (category=17) from unassociated STA
    hdr = "d0003a01" + bssid.replace(':', '') + "112233445566" + bssid.replace(':', '') + "1000"
    if "OK" not in hapd.request("MGMT_RX_PROCESS freq=2412 datarate=0 ssi_signal=-30 frame=" + hdr + "11000400dd3d0050f2020201aa300010270000000000000000000000000000000000000000000000000000ffffff7f00000000000000000000000040420f00ffff0000"):
        raise Exception("MGMT_RX_PROCESS failed")

    hapd.set("ext_mgmt_frame_handling", "0")