1#!/usr/bin/python
2#
3# Test cases for AP VLAN
4# Copyright (c) 2013-2016, Jouni Malinen <j@w1.fi>
5#
6# This software may be distributed under the terms of the BSD license.
7# See README for more details.
8
9from remotehost import remote_compatible
10import time
11import subprocess
12import logging
13logger = logging.getLogger(__name__)
14
15try:
16    import netifaces
17    netifaces_imported = True
18except ImportError:
19    netifaces_imported = False
20
21import hwsim_utils
22import hostapd
23from utils import *
24import os
25from tshark import run_tshark
26
27def test_ap_vlan_open(dev, apdev):
28    """AP VLAN with open network"""
29    filename = hostapd.acl_file(dev, apdev, 'hostapd.accept')
30    hostapd.send_file(apdev[0], filename, filename)
31    params = {"ssid": "test-vlan-open",
32              "dynamic_vlan": "1",
33              "accept_mac_file": filename}
34    hapd = hostapd.add_ap(apdev[0], params)
35
36    dev[0].connect("test-vlan-open", key_mgmt="NONE", scan_freq="2412")
37    dev[1].connect("test-vlan-open", key_mgmt="NONE", scan_freq="2412")
38    dev[2].connect("test-vlan-open", key_mgmt="NONE", scan_freq="2412")
39    hwsim_utils.test_connectivity_iface(dev[0], hapd, "brvlan1")
40    hwsim_utils.test_connectivity_iface(dev[1], hapd, "brvlan2")
41    hwsim_utils.test_connectivity(dev[2], hapd)
42    if filename.startswith('/tmp/'):
43        os.unlink(filename)
44
45def test_ap_vlan_file_open(dev, apdev):
46    """AP VLAN with open network and vlan_file mapping"""
47    filename = hostapd.acl_file(dev, apdev, 'hostapd.accept')
48    hostapd.send_file(apdev[0], filename, filename)
49    params = {"ssid": "test-vlan-open",
50              "dynamic_vlan": "1",
51              "vlan_file": "hostapd.vlan",
52              "accept_mac_file": filename}
53    hapd = hostapd.add_ap(apdev[0], params)
54
55    dev[0].connect("test-vlan-open", key_mgmt="NONE", scan_freq="2412")
56    dev[1].connect("test-vlan-open", key_mgmt="NONE", scan_freq="2412")
57    dev[2].connect("test-vlan-open", key_mgmt="NONE", scan_freq="2412")
58    hwsim_utils.test_connectivity_iface(dev[0], hapd, "brvlan1")
59    hwsim_utils.test_connectivity_iface(dev[1], hapd, "brvlan2")
60    hwsim_utils.test_connectivity(dev[2], hapd)
61    if filename.startswith('/tmp/'):
62        os.unlink(filename)
63
64def test_ap_vlan_file_open2(dev, apdev):
65    """AP VLAN with open network and vlan_file mapping (2)"""
66    filename = hostapd.acl_file(dev, apdev, 'hostapd.accept2')
67    hostapd.send_file(apdev[0], filename, filename)
68    params = {"ssid": "test-vlan-open",
69              "dynamic_vlan": "1",
70              "vlan_file": "hostapd.vlan2",
71              "accept_mac_file": filename}
72    hapd = hostapd.add_ap(apdev[0], params)
73
74    dev[0].connect("test-vlan-open", key_mgmt="NONE", scan_freq="2412")
75    dev[1].connect("test-vlan-open", key_mgmt="NONE", scan_freq="2412")
76    dev[2].connect("test-vlan-open", key_mgmt="NONE", scan_freq="2412")
77    hwsim_utils.test_connectivity_iface(dev[0], hapd, "brvlan1")
78    hwsim_utils.test_connectivity_iface(dev[1], hapd, "brvlan2")
79    hwsim_utils.test_connectivity_iface(dev[2], hapd, "hwsimbr3")
80    if filename.startswith('/tmp/'):
81        os.unlink(filename)
82
83def test_ap_vlan_file_parsing(dev, apdev, params):
84    """hostapd vlan_file/mac_file parsing"""
85    tmp = os.path.join(params['logdir'], 'ap_vlan_file_parsing.tmp')
86    params = {"ssid": "test-vlan-open", "dynamic_vlan": "1"}
87    hapd = hostapd.add_ap(apdev[0], params)
88
89    tests = ["#\n\n0\t11\n",
90             "1  netdev  br\n1",
91             "* ",
92             "1 netdev12345678901234567890"]
93    for t in tests:
94        with open(tmp, "w") as f:
95            f.write(t)
96        if "FAIL" not in hapd.request("SET vlan_file " + tmp):
97            raise Exception("Invalid vlan_file accepted")
98
99    with open(tmp, "w") as f:
100        f.write("1\tvlan\n")
101    with alloc_fail(hapd, 1, "=hostapd_config_read_vlan_file"):
102        if "FAIL" not in hapd.request("SET vlan_file " + tmp):
103            raise Exception("vlan_file accepted during OOM")
104
105    tests = ["#\n\n0\tvlan\n",
106             "4095\tvlan\n",
107             "vlan\n",
108             "1\t1234567890abcdef1234567890\n",
109             "1\n"]
110    for t in tests:
111        with open(tmp, "w") as f:
112            f.write(t)
113        if "FAIL" not in hapd.request("SET accept_mac_file " + tmp):
114            raise Exception("Invalid accept_mac_file accepted")
115
116    with open(tmp, "w") as f:
117        f.write("00:11:22:33:44:55\n")
118    with alloc_fail(hapd, 1, "hostapd_config_read_maclist"):
119        if "FAIL" not in hapd.request("SET accept_mac_file " + tmp):
120            raise Exception("accept_mac_file accepted during OOM")
121
122def test_ap_vlan_wpa2(dev, apdev):
123    """AP VLAN with WPA2-PSK"""
124    filename = hostapd.acl_file(dev, apdev, 'hostapd.accept')
125    hostapd.send_file(apdev[0], filename, filename)
126    params = hostapd.wpa2_params(ssid="test-vlan",
127                                 passphrase="12345678")
128    params['dynamic_vlan'] = "1"
129    params['accept_mac_file'] = filename
130    hapd = hostapd.add_ap(apdev[0], params)
131
132    dev[0].connect("test-vlan", psk="12345678", scan_freq="2412")
133    dev[1].connect("test-vlan", psk="12345678", scan_freq="2412")
134    dev[2].connect("test-vlan", psk="12345678", scan_freq="2412")
135    hapd.wait_sta()
136    hapd.wait_sta()
137    hapd.wait_sta()
138    hwsim_utils.test_connectivity_iface(dev[0], hapd, "brvlan1")
139    hwsim_utils.test_connectivity_iface(dev[1], hapd, "brvlan2")
140    hwsim_utils.test_connectivity(dev[2], hapd)
141    if filename.startswith('/tmp/'):
142        os.unlink(filename)
143
144def test_ap_vlan_wpa2_radius(dev, apdev):
145    """AP VLAN with WPA2-Enterprise and RADIUS attributes"""
146    params = hostapd.wpa2_eap_params(ssid="test-vlan")
147    params['dynamic_vlan'] = "1"
148    hapd = hostapd.add_ap(apdev[0], params)
149
150    dev[0].connect("test-vlan", key_mgmt="WPA-EAP", eap="PAX",
151                   identity="vlan1",
152                   password_hex="0123456789abcdef0123456789abcdef",
153                   scan_freq="2412")
154    dev[1].connect("test-vlan", key_mgmt="WPA-EAP", eap="PAX",
155                   identity="vlan2",
156                   password_hex="0123456789abcdef0123456789abcdef",
157                   scan_freq="2412")
158    dev[2].connect("test-vlan", key_mgmt="WPA-EAP", eap="PAX",
159                   identity="pax.user@example.com",
160                   password_hex="0123456789abcdef0123456789abcdef",
161                   scan_freq="2412")
162    hapd.wait_sta()
163    hapd.wait_sta()
164    hapd.wait_sta()
165    hwsim_utils.test_connectivity_iface(dev[0], hapd, "brvlan1")
166    hwsim_utils.test_connectivity_iface(dev[1], hapd, "brvlan2")
167    hwsim_utils.test_connectivity(dev[2], hapd)
168
169def test_ap_vlan_wpa2_radius_2(dev, apdev):
170    """AP VLAN with WPA2-Enterprise and RADIUS EGRESS_VLANID attributes"""
171    params = hostapd.wpa2_eap_params(ssid="test-vlan")
172    params['dynamic_vlan'] = "1"
173    hapd = hostapd.add_ap(apdev[0], params)
174
175    dev[0].connect("test-vlan", key_mgmt="WPA-EAP", eap="PAX",
176                   identity="vlan1b",
177                   password_hex="0123456789abcdef0123456789abcdef",
178                   scan_freq="2412")
179
180    hapd.wait_sta()
181    hwsim_utils.test_connectivity_iface(dev[0], hapd, "brvlan1")
182
183def test_ap_vlan_wpa2_radius_local(dev, apdev):
184    """AP VLAN with WPA2-Enterprise and local file setting VLAN IDs"""
185    filename = hostapd.acl_file(dev, apdev, 'hostapd.accept')
186    hostapd.send_file(apdev[0], filename, filename)
187    params = hostapd.wpa2_eap_params(ssid="test-vlan")
188    params['dynamic_vlan'] = "0"
189    params['vlan_file'] = "hostapd.vlan"
190    params['vlan_bridge'] = "test_br_vlan"
191    params['accept_mac_file'] = filename
192    hapd = hostapd.add_ap(apdev[0], params)
193
194    dev[0].connect("test-vlan", key_mgmt="WPA-EAP", eap="PAX",
195                   identity="pax.user@example.com",
196                   password_hex="0123456789abcdef0123456789abcdef",
197                   scan_freq="2412")
198    dev[1].connect("test-vlan", key_mgmt="WPA-EAP", eap="PAX",
199                   identity="pax.user@example.com",
200                   password_hex="0123456789abcdef0123456789abcdef",
201                   scan_freq="2412")
202    dev[2].connect("test-vlan", key_mgmt="WPA-EAP", eap="PAX",
203                   identity="pax.user@example.com",
204                   password_hex="0123456789abcdef0123456789abcdef",
205                   scan_freq="2412")
206    hapd.wait_sta()
207    hapd.wait_sta()
208    hapd.wait_sta()
209    hwsim_utils.test_connectivity_iface(dev[0], hapd, "test_br_vlan1")
210    hwsim_utils.test_connectivity_iface(dev[1], hapd, "test_br_vlan2")
211    hwsim_utils.test_connectivity(dev[2], hapd)
212    if filename.startswith('/tmp/'):
213        os.unlink(filename)
214
215def test_ap_vlan_wpa2_radius_id_change(dev, apdev):
216    """AP VLAN with WPA2-Enterprise and RADIUS attributes changing VLANID"""
217    generic_ap_vlan_wpa2_radius_id_change(dev, apdev, False)
218
219def test_ap_vlan_tagged_wpa2_radius_id_change(dev, apdev):
220    """AP tagged VLAN with WPA2-Enterprise and RADIUS attributes changing VLANID"""
221    ifname1 = 'wlan0.1'
222    ifname2 = 'wlan0.2'
223    try:
224        # Create tagged interface for wpa_supplicant
225        subprocess.call(['ip', 'link', 'add', 'link', dev[0].ifname,
226                         'name', ifname1, 'type', 'vlan', 'id', '1'])
227        subprocess.call(['ifconfig', ifname1, 'up'])
228
229        subprocess.call(['ip', 'link', 'add', 'link', dev[0].ifname,
230                         'name', ifname2, 'type', 'vlan', 'id', '2'])
231        subprocess.call(['ifconfig', ifname2, 'up'])
232
233        generic_ap_vlan_wpa2_radius_id_change(dev, apdev, True)
234    finally:
235        subprocess.call(['ifconfig', ifname1, 'down'])
236        subprocess.call(['ifconfig', ifname2, 'down'])
237        subprocess.call(['ip', 'link', 'del', ifname1])
238        subprocess.call(['ip', 'link', 'del', ifname2])
239
240def generic_ap_vlan_wpa2_radius_id_change(dev, apdev, tagged):
241    as_params = {"ssid": "as",
242                 "beacon_int": "2000",
243                 "radius_server_clients": "auth_serv/radius_clients.conf",
244                 "radius_server_auth_port": '18128',
245                 "eap_server": "1",
246                 "eap_user_file": "auth_serv/eap_user.conf",
247                 "ca_cert": "auth_serv/ca.pem",
248                 "server_cert": "auth_serv/server.pem",
249                 "private_key": "auth_serv/server.key"}
250    authserv = hostapd.add_ap(apdev[1], as_params)
251
252    params = hostapd.wpa2_eap_params(ssid="test-vlan")
253    params['dynamic_vlan'] = "1"
254    params['auth_server_port'] = "18128"
255    hapd = hostapd.add_ap(apdev[0], params)
256
257    identity = "vlan1tagged" if tagged else "vlan1"
258
259    dev[0].connect("test-vlan", key_mgmt="WPA-EAP", eap="PAX",
260                   identity=identity,
261                   password_hex="0123456789abcdef0123456789abcdef",
262                   scan_freq="2412")
263    hapd.wait_sta()
264    time.sleep(0.1)
265    if tagged:
266        hwsim_utils.run_connectivity_test(dev[0], hapd, 0, ifname1="wlan0.1",
267                                          ifname2="brvlan1")
268    else:
269        hwsim_utils.test_connectivity_iface(dev[0], hapd, "brvlan1")
270
271    logger.info("VLAN-ID -> 2")
272
273    authserv.disable()
274    authserv.set('eap_user_file', "auth_serv/eap_user_vlan.conf")
275    authserv.enable()
276
277    dev[0].dump_monitor()
278    dev[0].request("REAUTHENTICATE")
279    ev = dev[0].wait_event(["CTRL-EVENT-EAP-SUCCESS"], timeout=15)
280    if ev is None:
281        raise Exception("EAP reauthentication timed out")
282    ev = dev[0].wait_event(["WPA: Key negotiation completed"], timeout=5)
283    if ev is None:
284        raise Exception("4-way handshake after reauthentication timed out")
285    state = dev[0].get_status_field('wpa_state')
286    if state != "COMPLETED":
287        raise Exception("Unexpected state after reauth: " + state)
288    sta = hapd.get_sta(dev[0].own_addr())
289    if 'vlan_id' not in sta:
290        raise Exception("No VLAN ID in STA info")
291    if (not tagged) and (sta['vlan_id'] != '2'):
292        raise Exception("Unexpected VLAN ID: " + sta['vlan_id'])
293    ev = hapd.wait_event(["CTRL-EVENT-EAP-SUCCESS2"], timeout=1)
294    if ev is None:
295        raise Exception("EAP reauthentication timed out (AP)")
296    time.sleep(0.1)
297    if tagged:
298        hwsim_utils.run_connectivity_test(dev[0], hapd, 0, ifname1="wlan0.2",
299                                          ifname2="brvlan2")
300    else:
301        hwsim_utils.test_connectivity_iface(dev[0], hapd, "brvlan2")
302
303    logger.info("VLAN-ID -> 1")
304    time.sleep(1)
305
306    authserv.disable()
307    authserv.set('eap_user_file', "auth_serv/eap_user.conf")
308    authserv.enable()
309
310    dev[0].dump_monitor()
311    dev[0].request("REAUTHENTICATE")
312    ev = dev[0].wait_event(["CTRL-EVENT-EAP-SUCCESS"], timeout=15)
313    if ev is None:
314        raise Exception("EAP reauthentication timed out")
315    ev = dev[0].wait_event(["WPA: Key negotiation completed"], timeout=5)
316    if ev is None:
317        raise Exception("4-way handshake after reauthentication timed out")
318    state = dev[0].get_status_field('wpa_state')
319    if state != "COMPLETED":
320        raise Exception("Unexpected state after reauth: " + state)
321    ev = hapd.wait_event(["CTRL-EVENT-EAP-SUCCESS2"], timeout=1)
322    if ev is None:
323        raise Exception("EAP reauthentication timed out (AP)")
324    sta = hapd.get_sta(dev[0].own_addr())
325    if 'vlan_id' not in sta:
326        raise Exception("No VLAN ID in STA info")
327    if (not tagged) and (sta['vlan_id'] != '1'):
328        raise Exception("Unexpected VLAN ID: " + sta['vlan_id'])
329    time.sleep(0.2)
330    try:
331        if tagged:
332            hwsim_utils.run_connectivity_test(dev[0], hapd, 0,
333                                              ifname1="wlan0.1",
334                                              ifname2="brvlan1")
335        else:
336            hwsim_utils.test_connectivity_iface(dev[0], hapd, "brvlan1")
337    except Exception as e:
338        # It is possible for new bridge setup to not be ready immediately, so
339        # try again to avoid reporting issues related to that.
340        logger.info("First VLAN-ID 1 data test failed - try again")
341        time.sleep(0.1)
342        if tagged:
343            hwsim_utils.run_connectivity_test(dev[0], hapd, 0,
344                                              ifname1="wlan0.1",
345                                              ifname2="brvlan1")
346        else:
347            hwsim_utils.test_connectivity_iface(dev[0], hapd, "brvlan1")
348
349def test_ap_vlan_wpa2_radius_required(dev, apdev):
350    """AP VLAN with WPA2-Enterprise and RADIUS attributes required"""
351    params = hostapd.wpa2_eap_params(ssid="test-vlan")
352    params['dynamic_vlan'] = "2"
353    hostapd.add_ap(apdev[0], params)
354
355    dev[0].connect("test-vlan", key_mgmt="WPA-EAP", eap="PAX",
356                   identity="vlan1",
357                   password_hex="0123456789abcdef0123456789abcdef",
358                   scan_freq="2412")
359    dev[2].connect("test-vlan", key_mgmt="WPA-EAP", eap="PAX",
360                   identity="pax.user@example.com",
361                   password_hex="0123456789abcdef0123456789abcdef",
362                   scan_freq="2412", wait_connect=False)
363    ev = dev[2].wait_event(["CTRL-EVENT-CONNECTED",
364                            "CTRL-EVENT-DISCONNECTED"], timeout=20)
365    if ev is None:
366        raise Exception("Timeout on connection attempt")
367    if "CTRL-EVENT-CONNECTED" in ev:
368        raise Exception("Unexpected success without tunnel parameters")
369
370def test_ap_vlan_tagged(dev, apdev):
371    """AP VLAN with tagged interface"""
372    filename = hostapd.acl_file(dev, apdev, 'hostapd.accept')
373    hostapd.send_file(apdev[0], filename, filename)
374    params = {"ssid": "test-vlan-open",
375              "dynamic_vlan": "1",
376              "vlan_tagged_interface": "lo",
377              "accept_mac_file": filename}
378    hapd = hostapd.add_ap(apdev[0], params)
379
380    dev[0].connect("test-vlan-open", key_mgmt="NONE", scan_freq="2412")
381    dev[1].connect("test-vlan-open", key_mgmt="NONE", scan_freq="2412")
382    dev[2].connect("test-vlan-open", key_mgmt="NONE", scan_freq="2412")
383    hwsim_utils.test_connectivity_iface(dev[0], hapd, "brlo.1")
384    hwsim_utils.test_connectivity_iface(dev[1], hapd, "brlo.2")
385    hwsim_utils.test_connectivity(dev[2], hapd)
386    if filename.startswith('/tmp/'):
387        os.unlink(filename)
388
389def ap_vlan_iface_cleanup_multibss_cleanup():
390    subprocess.call(['ifconfig', 'stub0', 'down'],
391                    stderr=open('/dev/null', 'w'))
392    ifnames = ['wlan3.1', 'wlan3.2', 'wlan3-2.1', 'wlan3-2.2', 'stub0.2',
393               'stub0.1', 'stub0', 'brvlan1', 'brvlan2']
394    for ifname in ifnames:
395        subprocess.call(['ip', 'link', 'del', ifname],
396                        stderr=open('/dev/null', 'w'))
397
398def ap_vlan_iface_test_and_prepare_environ():
399    ifaces = netifaces.interfaces()
400    if "stub0" in ifaces:
401        raise Exception("stub0 already exists before")
402    ifaces = netifaces.interfaces()
403    if "stub0.1" in ifaces:
404        raise Exception("stub0.1 already exists before")
405
406    subprocess.call(['ip', 'link', 'add', 'stub0', 'type', 'dummy'])
407    subprocess.call(['ifconfig', 'stub0', 'up'])
408
409    ifaces = netifaces.interfaces()
410    if "stub0" not in ifaces:
411        raise HwsimSkip("failed to add stub0 - missing kernel config DUMMY ?")
412
413    subprocess.call(['ip', 'link', 'add', 'link', 'stub0', 'name', 'stub0.1',
414                     'type', 'vlan', 'id', '1'])
415
416    ifaces = netifaces.interfaces()
417    if "stub0.1" not in ifaces:
418        raise HwsimSkip("failed to add stub0.1 - missing kernel config VLAN_8021Q ?")
419
420    subprocess.call(['ip', 'link', 'del', 'stub0.1'])
421
422    ifaces = netifaces.interfaces()
423    if "stub0.1" in ifaces:
424        raise Exception("stub0.1 was not removed before testing")
425
426def test_ap_vlan_iface_cleanup_multibss(dev, apdev):
427    """AP VLAN operation in multi-BSS multi-VLAN case"""
428    ap_vlan_iface_cleanup_multibss(dev, apdev, 'multi-bss-iface.conf')
429
430def ap_vlan_iface_cleanup_multibss(dev, apdev, cfgfile):
431    # AP VLAN with WPA2-Enterprise and RADIUS attributes changing VLANID
432    # check that multiple bss do not interfere with each other with respect
433    # to deletion of bridge and tagged interface.
434
435    if not netifaces_imported:
436        raise HwsimSkip("python module netifaces not available")
437
438    try:
439        ap_vlan_iface_cleanup_multibss_cleanup()
440        ap_vlan_iface_test_and_prepare_environ()
441
442        as_params = {"ssid": "as",
443                     "beacon_int": "2000",
444                     "radius_server_clients": "auth_serv/radius_clients.conf",
445                     "radius_server_auth_port": '18128',
446                     "eap_server": "1",
447                     "eap_user_file": "auth_serv/eap_user.conf",
448                     "ca_cert": "auth_serv/ca.pem",
449                     "server_cert": "auth_serv/server.pem",
450                     "private_key": "auth_serv/server.key",
451                     "vlan_naming": "1"}
452        authserv = hostapd.add_ap(apdev[1], as_params)
453
454        # start the actual test
455        hapd = hostapd.add_iface(apdev[0], cfgfile)
456        hapd1 = hostapd.Hostapd("wlan3-2", 1)
457        hapd1.enable()
458
459        ifaces = netifaces.interfaces()
460        if "brvlan1" in ifaces:
461            raise Exception("bridge brvlan1 already exists before")
462        if "brvlan2" in ifaces:
463            raise Exception("bridge brvlan2 already exists before")
464
465        dev[0].connect("bss-1", key_mgmt="WPA-EAP", eap="PAX",
466                       identity="vlan1",
467                       password_hex="0123456789abcdef0123456789abcdef",
468                       scan_freq="2412")
469        hapd.wait_sta()
470
471        ifaces = netifaces.interfaces()
472        if "brvlan1" not in ifaces:
473            raise Exception("bridge brvlan1 was not created")
474
475        hwsim_utils.test_connectivity_iface(dev[0], hapd, "brvlan1")
476        if not iface_is_in_bridge("brvlan1", "stub0.1"):
477            raise Exception("stub0.1 not in brvlan1")
478
479        dev[1].connect("bss-2", key_mgmt="WPA-EAP", eap="PAX",
480                       identity="vlan1",
481                       password_hex="0123456789abcdef0123456789abcdef",
482                       scan_freq="2412")
483
484        hapd1.wait_sta()
485        hwsim_utils.test_connectivity_iface(dev[1], hapd1, "brvlan1")
486        if not iface_is_in_bridge("brvlan1", "stub0.1"):
487            raise Exception("stub0.1 not in brvlan1")
488
489        authserv.disable()
490        authserv.set('eap_user_file', "auth_serv/eap_user_vlan.conf")
491        authserv.enable()
492
493        logger.info("wlan0 -> VLAN 2")
494
495        dev[0].dump_monitor()
496        dev[0].request("REAUTHENTICATE")
497        ev = dev[0].wait_event(["CTRL-EVENT-EAP-SUCCESS"], timeout=15)
498        if ev is None:
499            raise Exception("EAP reauthentication timed out")
500        ev = dev[0].wait_event(["WPA: Key negotiation completed"], timeout=5)
501        if ev is None:
502            raise Exception("4-way handshake after reauthentication timed out")
503        state = dev[0].get_status_field('wpa_state')
504        if state != "COMPLETED":
505            raise Exception("Unexpected state after reauth: " + state)
506
507        ifaces = netifaces.interfaces()
508        if "brvlan1" not in ifaces:
509            raise Exception("bridge brvlan1 has been removed too early")
510
511        hwsim_utils.test_connectivity_iface(dev[0], hapd, "brvlan2",
512                                            max_tries=5)
513
514        if not iface_is_in_bridge("brvlan2", "stub0.2"):
515            raise Exception("stub0.2 not in brvlan2")
516
517        logger.info("test wlan1 == VLAN 1")
518        hwsim_utils.test_connectivity_iface(dev[1], hapd1, "brvlan1")
519        if not iface_is_in_bridge("brvlan1", "stub0.1"):
520            raise Exception("stub0.1 not in brvlan1")
521
522        logger.info("wlan1 -> VLAN 2")
523
524        dev[1].dump_monitor()
525        dev[1].request("REAUTHENTICATE")
526        ev = dev[1].wait_event(["CTRL-EVENT-EAP-SUCCESS"], timeout=15)
527        if ev is None:
528            raise Exception("EAP reauthentication timed out")
529        ev = dev[1].wait_event(["WPA: Key negotiation completed"], timeout=5)
530        if ev is None:
531            raise Exception("4-way handshake after reauthentication timed out")
532        state = dev[1].get_status_field('wpa_state')
533        if state != "COMPLETED":
534            raise Exception("Unexpected state after reauth: " + state)
535
536        # it can take some time for data connectivity to be updated
537        hwsim_utils.test_connectivity_iface(dev[1], hapd1, "brvlan2",
538                                            max_tries=5)
539        logger.info("test wlan0 == VLAN 2")
540        hwsim_utils.test_connectivity_iface(dev[0], hapd, "brvlan2")
541
542        if not iface_is_in_bridge("brvlan2", "stub0.2"):
543            raise Exception("stub0.2 not in brvlan2")
544
545        ifaces = netifaces.interfaces()
546        if "brvlan1" in ifaces:
547            raise Exception("bridge brvlan1 has not been cleaned up")
548
549        # disconnect dev0 first to test a corner case
550        dev[0].request("DISCONNECT")
551        dev[0].wait_disconnected()
552        dev[1].request("DISCONNECT")
553        dev[1].wait_disconnected()
554
555        # station removal needs some time
556        for i in range(15):
557            time.sleep(1)
558            ifaces = netifaces.interfaces()
559            if "brvlan2" not in ifaces:
560                break
561
562        ifaces = netifaces.interfaces()
563        if "brvlan2" in ifaces:
564            raise Exception("bridge brvlan2 has not been cleaned up")
565
566        hapd.request("DISABLE")
567    finally:
568        ap_vlan_iface_cleanup_multibss_cleanup()
569
570def test_ap_vlan_iface_cleanup_multibss_per_sta_vif(dev, apdev):
571    """AP VLAN operation in multi-BSS multi-VLAN case with per-sta-vif set"""
572
573    # AP VLAN with WPA2-Enterprise and RADIUS attributes changing VLANID
574    # check that multiple bss do not interfere with each other with respect
575    # to deletion of bridge and tagged interface. per_sta_vif is enabled.
576    ap_vlan_iface_cleanup_multibss(dev, apdev,
577                                   'multi-bss-iface-per_sta_vif.conf')
578
579def test_ap_vlan_without_station(dev, apdev, p):
580    """AP VLAN with WPA2-PSK and no station"""
581    try:
582        filename = hostapd.acl_file(dev, apdev, 'hostapd.accept')
583        hostapd.send_file(apdev[0], filename, filename)
584        subprocess.call(['brctl', 'addbr', 'brvlan1'])
585        subprocess.call(['brctl', 'setfd', 'brvlan1', '0'])
586        subprocess.call(['ifconfig', 'brvlan1', 'up'])
587        # use a passphrase wlantest does not know, so it cannot
588        # inject decrypted frames into pcap
589        params = hostapd.wpa2_params(ssid="test-vlan",
590                                     passphrase="12345678x")
591        params['dynamic_vlan'] = "1"
592        params['vlan_file'] = 'hostapd.wlan3.vlan'
593        params['accept_mac_file'] = filename
594        hapd = hostapd.add_ap(apdev[0], params)
595
596        # inject some traffic
597        sa = hapd.own_addr()
598        da = "ff:ff:ff:ff:ff:00"
599        hapd.request('DATA_TEST_CONFIG 1 ifname=brvlan1')
600        hapd.request('DATA_TEST_TX {} {} 0'.format(da, sa))
601        hapd.request('DATA_TEST_CONFIG 0')
602        time.sleep(.1)
603
604        dev[0].connect("test-vlan", psk="12345678x", scan_freq="2412")
605        hapd.wait_sta()
606
607        # inject some traffic
608        sa = hapd.own_addr()
609        da = "ff:ff:ff:ff:ff:01"
610        hapd.request('DATA_TEST_CONFIG 1 ifname=brvlan1')
611        hapd.request('DATA_TEST_TX {} {} 0'.format(da, sa))
612        hapd.request('DATA_TEST_CONFIG 0')
613
614        # let the AP send couple of Beacon frames
615        time.sleep(1)
616        out = run_tshark(os.path.join(p['logdir'], "hwsim0.pcapng"),
617                         "wlan.da == ff:ff:ff:ff:ff:00",
618                         ["wlan.fc.protected"])
619
620        if out is not None:
621            lines = out.splitlines()
622            if len(lines) < 1:
623                # Newer kernel versions filter out frames when there are no
624                # authorized stations on an AP/AP_VLAN interface, so do not
625                # trigger an error here.
626                logger.info("first frame not observed")
627            state = 1
628            for l in lines:
629                is_protected = int(l, 16)
630                if is_protected != 1:
631                    state = 0
632            if state != 1:
633                raise Exception("Broadcast packets were not encrypted when no station was connected")
634        else:
635            raise Exception("first frame not observed")
636
637        out = run_tshark(os.path.join(p['logdir'], "hwsim0.pcapng"),
638                         "wlan.da == ff:ff:ff:ff:ff:01",
639                         ["wlan.fc.protected"])
640
641        if out is not None:
642            lines = out.splitlines()
643            if len(lines) < 1:
644                raise Exception("second frame not observed")
645            state = 1
646            for l in lines:
647                if l == "True":
648                    is_protected = 1
649                elif l == "False":
650                    is_protected = 0
651                else:
652                    is_protected = int(l, 16)
653                if is_protected != 1:
654                    state = 0
655            if state != 1:
656                raise Exception("Broadcast packets were not encrypted when station was connected")
657        else:
658            raise Exception("second frame not observed")
659
660        dev[0].request("DISCONNECT")
661        dev[0].wait_disconnected()
662        if filename.startswith('/tmp/'):
663            os.unlink(filename)
664
665    finally:
666        subprocess.call(['ip', 'link', 'set', 'dev', 'brvlan1', 'down'])
667        subprocess.call(['ip', 'link', 'set', 'dev', 'wlan3.1', 'down'],
668                        stderr=open('/dev/null', 'w'))
669        subprocess.call(['brctl', 'delif', 'brvlan1', 'wlan3.1'],
670                        stderr=open('/dev/null', 'w'))
671        subprocess.call(['brctl', 'delbr', 'brvlan1'])
672
673@remote_compatible
674def test_ap_open_per_sta_vif(dev, apdev):
675    """AP VLAN with open network"""
676    params = {"ssid": "test-vlan-open",
677              "per_sta_vif": "1"}
678    hapd = hostapd.add_ap(apdev[0], params)
679
680    dev[0].connect("test-vlan-open", key_mgmt="NONE", scan_freq="2412")
681    hwsim_utils.test_connectivity_iface(dev[0], hapd,
682                                        apdev[0]['ifname'] + ".4096")
683
684@remote_compatible
685def test_ap_vlan_open_per_sta_vif(dev, apdev):
686    """AP VLAN (dynamic) with open network"""
687    params = {"ssid": "test-vlan-open",
688              "per_sta_vif": "1",
689              "dynamic_vlan": "1"}
690    hapd = hostapd.add_ap(apdev[0], params)
691
692    dev[0].connect("test-vlan-open", key_mgmt="NONE", scan_freq="2412")
693    hwsim_utils.test_connectivity_iface(dev[0], hapd,
694                                        apdev[0]['ifname'] + ".4096")
695
696def test_ap_vlan_wpa2_radius_tagged(dev, apdev):
697    """AP VLAN with WPA2-Enterprise and RADIUS EGRESS_VLANID attributes"""
698    ifname = 'wlan0.1'
699    try:
700        params = hostapd.wpa2_eap_params(ssid="test-vlan")
701        params['dynamic_vlan'] = "1"
702        params["vlan_naming"] = "1"
703        hapd = hostapd.add_ap(apdev[0], params)
704
705        dev[0].connect("test-vlan", key_mgmt="WPA-EAP", eap="PAX",
706                       identity="vlan1tagged",
707                       password_hex="0123456789abcdef0123456789abcdef",
708                       scan_freq="2412")
709
710        # Create tagged interface for wpa_supplicant
711        subprocess.call(['ip', 'link', 'add', 'link', dev[0].ifname,
712                         'name', ifname, 'type', 'vlan', 'id', '1'])
713        subprocess.call(['ifconfig', ifname, 'up'])
714
715        hwsim_utils.run_connectivity_test(dev[0], hapd, 0, ifname1=ifname,
716                                          ifname2="brvlan1")
717    finally:
718        subprocess.call(['ifconfig', ifname, 'down'])
719        subprocess.call(['ip', 'link', 'del', ifname])
720
721def test_ap_vlan_wpa2_radius_mixed(dev, apdev):
722    """AP VLAN with WPA2-Enterprise and tagged+untagged VLANs"""
723    ifname = 'wlan0.1'
724    try:
725        params = hostapd.wpa2_eap_params(ssid="test-vlan")
726        params['dynamic_vlan'] = "1"
727        params["vlan_naming"] = "1"
728        hapd = hostapd.add_ap(apdev[0], params)
729
730        dev[0].connect("test-vlan", key_mgmt="WPA-EAP", eap="PAX",
731                       identity="vlan12mixed",
732                       password_hex="0123456789abcdef0123456789abcdef",
733                       scan_freq="2412")
734
735        # Add tagged VLAN interface to wpa_supplicant interface for testing
736        subprocess.call(['ip', 'link', 'add', 'link', dev[0].ifname,
737                         'name', ifname, 'type', 'vlan', 'id', '1'])
738        subprocess.call(['ifconfig', ifname, 'up'])
739
740        logger.info("Test connectivity in untagged VLAN 2")
741        hwsim_utils.run_connectivity_test(dev[0], hapd, 0,
742                                          ifname1=dev[0].ifname,
743                                          ifname2="brvlan2")
744        logger.info("Test connectivity in tagged VLAN 1")
745        hwsim_utils.run_connectivity_test(dev[0], hapd, 0, ifname1=ifname,
746                                          ifname2="brvlan1")
747    finally:
748        subprocess.call(['ifconfig', ifname, 'down'])
749        subprocess.call(['ip', 'link', 'del', ifname])
750
751def test_ap_vlan_reconnect(dev, apdev):
752    """AP VLAN with WPA2-PSK connect, disconnect, connect"""
753    filename = hostapd.acl_file(dev, apdev, 'hostapd.accept')
754    hostapd.send_file(apdev[0], filename, filename)
755    params = hostapd.wpa2_params(ssid="test-vlan",
756                                 passphrase="12345678")
757    params['dynamic_vlan'] = "1"
758    params['accept_mac_file'] = filename
759    hapd = hostapd.add_ap(apdev[0], params)
760
761    logger.info("connect sta")
762    dev[0].connect("test-vlan", psk="12345678", scan_freq="2412")
763    hapd.wait_sta()
764    hwsim_utils.test_connectivity_iface(dev[0], hapd, "brvlan1")
765    logger.info("disconnect sta")
766    dev[0].request("REMOVE_NETWORK all")
767    dev[0].wait_disconnected(timeout=10)
768    time.sleep(1)
769    logger.info("reconnect sta")
770    dev[0].connect("test-vlan", psk="12345678", scan_freq="2412")
771    hapd.wait_sta()
772    hwsim_utils.test_connectivity_iface(dev[0], hapd, "brvlan1")
773    if filename.startswith('/tmp/'):
774        os.unlink(filename)
775
776def test_ap_vlan_psk(dev, apdev, params):
777    """AP VLAN based on PSK/passphrase"""
778    psk_file = os.path.join(params['logdir'], 'ap_vlan_psk.wpa_psk')
779    with open(psk_file, 'w') as f:
780        f.write('vlanid=1 00:00:00:00:00:00 passphrase-for-vlan-1\n')
781        f.write('vlanid=2 00:00:00:00:00:00 passphrase-for-vlan-2\n')
782        f.write('vlanid=3 00:00:00:00:00:00 passphrase-for-vlan-3\n')
783
784    ssid = 'test-vlan-rsn'
785    params = hostapd.wpa2_params(ssid=ssid)
786    params['dynamic_vlan'] = "1"
787    params['wpa_psk_file'] = psk_file
788    hapd = hostapd.add_ap(apdev[0], params)
789
790    dev[0].connect(ssid, psk="passphrase-for-vlan-1", scan_freq="2412")
791    dev[1].connect(ssid, psk="passphrase-for-vlan-2", scan_freq="2412")
792    dev[2].connect(ssid, psk="passphrase-for-vlan-3", scan_freq="2412")
793    hapd.wait_sta()
794    hapd.wait_sta()
795    hapd.wait_sta()
796    hwsim_utils.test_connectivity_iface(dev[0], hapd, "brvlan1")
797    hwsim_utils.test_connectivity_iface(dev[1], hapd, "brvlan2")
798    hwsim_utils.test_connectivity_iface(dev[2], hapd, "brvlan3")
799
800    for i in range(3):
801        sta = hapd.get_sta(dev[i].own_addr())
802        if not (sta and "vlan_id" in sta):
803            raise Exception("VLAN information not in STA output")
804        vlan_id = int(sta["vlan_id"])
805        if vlan_id != i + 1:
806            raise Exception("Unexpected vlan_id %d for dev[%d]" % (vlan_id, i))
807
808def start_ap_vlan_sae(apdev):
809    params = hostapd.wpa2_params(ssid="test-sae-vlan")
810    params['wpa_key_mgmt'] = 'SAE'
811    params['sae_password'] = ['pw1|vlanid=1|id=id1',
812                              'pw2|mac=ff:ff:ff:ff:ff:ff|vlanid=2|id=id2',
813                              'pw3|vlanid=3|id=id3']
814    params['dynamic_vlan'] = "1"
815    params['wpa_group_rekey'] = '10'
816    return hostapd.add_ap(apdev, params)
817
818def test_ap_vlan_sae(dev, apdev, params):
819    """AP VLAN based on SAE Password Identifier"""
820    for i in range(3):
821        check_sae_capab(dev[i])
822    hapd = start_ap_vlan_sae(apdev[0])
823
824    for i in range(3):
825        dev[i].request("SET sae_groups ")
826        dev[i].connect("test-sae-vlan", sae_password="pw%d" % (i + 1),
827                       sae_password_id="id%d" % (i + 1),
828                       key_mgmt="SAE", scan_freq="2412")
829        hapd.wait_sta()
830
831    hwsim_utils.test_connectivity_iface(dev[0], hapd, "brvlan1")
832    hwsim_utils.test_connectivity_iface(dev[1], hapd, "brvlan2")
833    hwsim_utils.test_connectivity_iface(dev[2], hapd, "brvlan3")
834
835    for i in range(3):
836        sta = hapd.get_sta(dev[i].own_addr())
837        if not (sta and "vlan_id" in sta):
838            raise Exception("VLAN information not in STA output")
839        vlan_id = int(sta["vlan_id"])
840        if vlan_id != i + 1:
841            raise Exception("Unexpected vlan_id %d for dev[%d]" % (vlan_id, i))
842
843        ev = dev[i].wait_event(["RSN: Group rekeying completed"], timeout=11)
844        if ev is None:
845            raise Exception("GTK rekey timed out")
846
847    time.sleep(1)
848
849    hwsim_utils.test_connectivity_iface(dev[0], hapd, "brvlan1")
850    hwsim_utils.test_connectivity_iface(dev[1], hapd, "brvlan2")
851    hwsim_utils.test_connectivity_iface(dev[2], hapd, "brvlan3")
852
853def test_ap_vlan_sae_group_rekey(dev, apdev, params):
854    """AP VLAN and group rekeying"""
855    check_sae_capab(dev[0])
856    hapd = start_ap_vlan_sae(apdev[0])
857
858    dev[0].set("sae_groups", "")
859    dev[0].connect("test-sae-vlan", sae_password="pw1",
860                       sae_password_id="id1",
861                       key_mgmt="SAE", scan_freq="2412")
862    hapd.wait_sta()
863
864    hwsim_utils.test_connectivity_iface(dev[0], hapd, "brvlan1")
865
866    hapd.set("ext_eapol_frame_io", "1")
867    ev = hapd.wait_event(["EAPOL-TX"], timeout=15)
868    if ev is None:
869        raise Exception("Timeout on EAPOL-TX from hostapd")
870    time.sleep(0.1)
871
872    hwsim_utils.test_connectivity_iface(dev[0], hapd, "brvlan1")
873    hapd.set("ext_eapol_frame_io", "0")
874
875    ev = dev[0].wait_event(["RSN: Group rekeying completed"], timeout=11)
876    if ev is None:
877        raise Exception("GTK rekey timed out")
878    time.sleep(1)
879
880    hwsim_utils.test_connectivity_iface(dev[0], hapd, "brvlan1")
881