215 lines
12 KiB
Python
215 lines
12 KiB
Python
# Test cases for SCS
|
|
# Copyright (c) 2021, Jouni Malinen <j@w1.fi>
|
|
# Copyright (c) 2021, The Linux Foundation
|
|
#
|
|
# This software may be distributed under the terms of the BSD license.
|
|
# See README for more details.
|
|
|
|
import struct
|
|
import time
|
|
|
|
import hostapd
|
|
from utils import *
|
|
|
|
def register_scs_req(hapd):
|
|
type = 0x00d0
|
|
match = "1300"
|
|
if "OK" not in hapd.request("REGISTER_FRAME %04x %s" % (type, match)):
|
|
raise Exception("Could not register frame reception for Robust AV Streaming")
|
|
|
|
def handle_scs_req(hapd, wrong_dialog=False, status_code=0, twice=False,
|
|
short=False, scsid=1):
|
|
msg = hapd.mgmt_rx()
|
|
if msg['subtype'] != 13:
|
|
logger.info("RX:" + str(msg))
|
|
raise Exception("Received unexpected Management frame")
|
|
categ, act, dialog_token = struct.unpack('BBB', msg['payload'][0:3])
|
|
if categ != 19 or act != 0:
|
|
logger.info("RX:" + str(msg))
|
|
raise Exception("Received unexpected Action frame")
|
|
|
|
if wrong_dialog:
|
|
dialog_token = (dialog_token + 1) % 256
|
|
msg['da'] = msg['sa']
|
|
msg['sa'] = hapd.own_addr()
|
|
count = 1
|
|
if short:
|
|
resp = struct.pack('BBB', 19, 1, dialog_token)
|
|
else:
|
|
resp = struct.pack('BBBB', 19, 1, dialog_token, count)
|
|
resp += struct.pack('<BH', scsid, status_code)
|
|
msg['payload'] = resp
|
|
hapd.mgmt_tx(msg)
|
|
ev = hapd.wait_event(["MGMT-TX-STATUS"], timeout=5)
|
|
if ev is None or "stype=13 ok=1" not in ev:
|
|
raise Exception("No TX status reported")
|
|
if twice:
|
|
hapd.mgmt_tx(msg)
|
|
ev = hapd.wait_event(["MGMT-TX-STATUS"], timeout=5)
|
|
if ev is None or "stype=13 ok=1" not in ev:
|
|
raise Exception("No TX status reported")
|
|
|
|
def wait_scs_result(dev, expect_status="0"):
|
|
ev = dev.wait_event(["CTRL-EVENT-SCS-RESULT"], timeout=2)
|
|
if ev is None:
|
|
raise Exception("No SCS result reported")
|
|
if "status_code=%s" % expect_status not in ev:
|
|
raise Exception("Unexpected SCS result: " + ev)
|
|
|
|
def test_scs_invalid_params(dev, apdev):
|
|
"""SCS command invalid parameters"""
|
|
tests = ["",
|
|
"scs_id=1",
|
|
"scs_id=1 foo",
|
|
"scs_id=1 add ",
|
|
"scs_id=1 add scs_up=8",
|
|
"scs_id=1 add scs_up=7",
|
|
"scs_id=1 add scs_up=7 classifier_type=1",
|
|
"scs_id=1 add scs_up=7 classifier_type=4",
|
|
"scs_id=1 add scs_up=7 classifier_type=4 ip_version=ipv4",
|
|
"scs_id=1 add scs_up=7 classifier_type=4 ip_version=ipv4 src_ip=q",
|
|
"scs_id=1 add scs_up=7 classifier_type=4 ip_version=ipv4 dst_ip=q",
|
|
"scs_id=1 add scs_up=7 classifier_type=4 ip_version=ipv4 src_port=q",
|
|
"scs_id=1 add scs_up=7 classifier_type=4 ip_version=ipv4 dst_port=q",
|
|
"scs_id=1 add scs_up=7 classifier_type=4 ip_version=ipv4 protocol=foo",
|
|
"scs_id=1 add scs_up=7 classifier_type=4 ip_version=ipv6 protocol=foo",
|
|
"scs_id=1 add scs_up=7 classifier_type=4 ip_version=ipv6 next_header=foo",
|
|
"scs_id=1 add scs_up=7 classifier_type=4 ip_version=ipv6 flow_label=ffffff",
|
|
"scs_id=1 add scs_up=7 classifier_type=10",
|
|
"scs_id=1 add scs_up=7 classifier_type=10 prot_instance=1",
|
|
"scs_id=1 add scs_up=7 classifier_type=10 prot_instance=1 prot_number=udp",
|
|
"scs_id=1 add scs_up=7 classifier_type=10 prot_instance=1 prot_number=udp filter_value=11223344",
|
|
"scs_id=1 add scs_up=7 classifier_type=10 prot_instance=1 prot_number=udp filter_value=qq",
|
|
"scs_id=1 add scs_up=7 classifier_type=10 prot_instance=1 prot_number=udp filter_value=11223344 filter_mask=ffffff",
|
|
"scs_id=1 add scs_up=7 classifier_type=10 prot_instance=1 prot_number=udp filter_value=11223344 filter_mask=qqqqqqqq",
|
|
"scs_id=1 add scs_up=7 classifier_type=10 prot_instance=1 prot_number=udp filter_value=11223344 filter_mask=ffffffff",
|
|
"scs_id=1 add scs_up=7 classifier_type=10 prot_instance=1 prot_number=foo filter_value=11223344 filter_mask=ffffffff",
|
|
"scs_id=1 add scs_up=7 classifier_type=10 prot_instance=1 prot_number=udp filter_value=11 filter_mask=ee classifier_type=10 prot_instance=2 prot_number=udp filter_value=22 filter_mask=ff",
|
|
"scs_id=1 add scs_up=7 classifier_type=10 prot_instance=1 prot_number=udp filter_value=11 filter_mask=ee classifier_type=10 prot_instance=2 prot_number=udp filter_value=22 filter_mask=ff tclas_processing=2",
|
|
"scs_id=1 add scs_up=7 classifier_type=10 prot_instance=1 prot_number=udp filter_value=11 filter_mask=ee classifier_type=10 prot_instance=2 prot_number=udp filter_value=22 filter_mask=ff tclas_processing=0",
|
|
"scs_id=1 add scs_up=6 classifier_type=4 ip_version=ipv4 src_ip=1.2.3.4 dst_ip=5.6.7.8 src_port=12345 dst_port=23456 dscp=5 protocol=udp scs_id=1 add scs_up=6 classifier_type=4 ip_version=ipv4 src_ip=1.2.3.4 dst_ip=5.6.7.8 src_port=12345 dst_port=23456 dscp=5 protocol=tcp"
|
|
"scs_id=1 remove",
|
|
"scs_id=1 change "]
|
|
for t in tests:
|
|
if "FAIL" not in dev[0].request("SCS " + t):
|
|
raise Exception("Invalid SCS parameters accepted: " + t)
|
|
|
|
def test_scs_request(dev, apdev):
|
|
"""SCS Request"""
|
|
params = {"ssid": "scs",
|
|
"ext_capa": 6*"00" + "40"}
|
|
hapd = hostapd.add_ap(apdev[0], params)
|
|
run_scs_request(dev, hapd, False)
|
|
|
|
def test_scs_request_wfa(dev, apdev):
|
|
"""SCS Request (WFA capability)"""
|
|
params = {"ssid": "scs",
|
|
"ext_capa": 6*"00" + "40",
|
|
"vendor_elements": "dd06506f9a230104"}
|
|
hapd = hostapd.add_ap(apdev[0], params)
|
|
run_scs_request(dev, hapd, True)
|
|
|
|
def run_scs_request(dev, hapd, wfa):
|
|
register_scs_req(hapd)
|
|
|
|
dev[0].connect("scs", key_mgmt="NONE", scan_freq="2412")
|
|
|
|
hapd.dump_monitor()
|
|
hapd.set("ext_mgmt_frame_handling", "1")
|
|
|
|
cmd = "SCS scs_id=1 add scs_up=7 classifier_type=10 prot_instance=1 prot_number=udp filter_value=11223344 filter_mask=ffffffff"
|
|
if "OK" not in dev[0].request(cmd):
|
|
raise Exception("SCS add failed")
|
|
|
|
handle_scs_req(hapd)
|
|
wait_scs_result(dev[0])
|
|
|
|
cmd = "SCS scs_id=2 add scs_up=5 classifier_type=10 prot_instance=1 prot_number=udp filter_value=11223344 filter_mask=ffffffff"
|
|
if "OK" not in dev[0].request(cmd):
|
|
raise Exception("SCS add failed")
|
|
|
|
handle_scs_req(hapd, wrong_dialog=True)
|
|
ev = dev[0].wait_event(["CTRL-EVENT-SCS-RESULT"], timeout=2)
|
|
if ev is None:
|
|
raise Exception("No SCS result reported")
|
|
if "status_code=timedout" not in ev:
|
|
raise Exception("Timeout not reported: " + ev)
|
|
|
|
cmd = "SCS scs_id=1 add scs_up=5 classifier_type=10 prot_instance=1 prot_number=udp filter_value=11223344 filter_mask=ffffffff"
|
|
if "FAIL" not in dev[0].request(cmd):
|
|
raise Exception("SCS add for already configured scs_id did not fail")
|
|
|
|
cmd = "SCS scs_id=1 remove"
|
|
if "OK" not in dev[0].request(cmd):
|
|
raise Exception("SCS remove failed")
|
|
handle_scs_req(hapd)
|
|
wait_scs_result(dev[0])
|
|
|
|
tests = ["scs_up=6 classifier_type=4 ip_version=ipv4 src_ip=1.2.3.4 dst_ip=5.6.7.8 src_port=12345 dst_port=23456 dscp=5 protocol=udp",
|
|
"scs_up=6 classifier_type=4 ip_version=ipv4 src_ip=1.2.3.4 dst_ip=5.6.7.8 src_port=12345 dst_port=23456 dscp=5 protocol=tcp",
|
|
"scs_up=6 classifier_type=4 ip_version=ipv4 src_ip=1.2.3.4 dst_ip=5.6.7.8 src_port=12345 dst_port=23456 dscp=5 protocol=esp",
|
|
"scs_up=6 classifier_type=4 ip_version=ipv6 src_ip=::1 dst_ip=::1 src_port=12345 dst_port=23456 dscp=5 next_header=udp",
|
|
"scs_up=6 classifier_type=4 ip_version=ipv6 src_ip=::1 dst_ip=::1 src_port=12345 dst_port=23456 dscp=5 next_header=tcp",
|
|
"scs_up=6 classifier_type=4 ip_version=ipv6 src_ip=::1 dst_ip=::1 src_port=12345 dst_port=23456 dscp=5 next_header=esp flow_label=012345",
|
|
"scs_up=6 classifier_type=10 prot_instance=1 prot_number=tcp filter_value=11223344 filter_mask=ffffffff",
|
|
"scs_up=6 classifier_type=10 prot_instance=1 prot_number=esp filter_value=11223344 filter_mask=ffffffff",
|
|
"scs_up=6 classifier_type=10 prot_instance=1 prot_number=udp filter_value=11223344 filter_mask=ffffffff",
|
|
"scs_up=6 classifier_type=4 ip_version=ipv4 src_ip=1.2.3.4 dst_ip=5.6.7.8 src_port=12345 dst_port=23456 dscp=5 protocol=udp classifier_type=4 ip_version=ipv4 src_ip=1.2.3.4 dst_ip=5.6.7.8 src_port=12345 dst_port=23456 dscp=5 protocol=tcp tclas_processing=1",
|
|
"scs_up=6 classifier_type=4 ip_version=ipv4 src_ip=1.2.3.4 dst_ip=5.6.7.8 src_port=12345 dst_port=23456 dscp=5 protocol=udp scs_id=10 add scs_up=6 classifier_type=4 ip_version=ipv4 src_ip=1.2.3.4 dst_ip=5.6.7.8 src_port=12345 dst_port=23456 dscp=5 protocol=tcp"]
|
|
qos_char = "scs_up=6 classifier_type=4 ip_version=ipv4 src_ip=1.2.3.4 dst_ip=5.6.7.8 src_port=12345 dst_port=23456 dscp=5 protocol=udp qos_characteristics=direct min_si=1 max_si=9 min_data_rate=123 delay_bound=1234 max_msdu=12345 service_start_time=123456 service_start_time_link_id=0 mean_data_rate=1 burst_size=1234 msdu_lifetime=1234 msdu_delivery_info=1 medium_time=2"
|
|
if wfa:
|
|
tests.append(qos_char)
|
|
for t in tests:
|
|
cmd = "SCS scs_id=1 change " + t
|
|
if "OK" not in dev[0].request(cmd):
|
|
raise Exception("SCS change failed: " + t)
|
|
handle_scs_req(hapd)
|
|
wait_scs_result(dev[0])
|
|
if "scs_id=" in t:
|
|
wait_scs_result(dev[0], expect_status="response_not_received")
|
|
|
|
if not wfa:
|
|
cmd = "SCS scs_id=1 change " + qos_char
|
|
if "OK" in dev[0].request(cmd):
|
|
raise Exception("SCS change with QoS characteristics accepted unexpectedly")
|
|
|
|
cmd = "SCS scs_id=1 change scs_up=6 classifier_type=4 ip_version=ipv4 src_ip=1.2.3.4 dst_ip=5.6.7.8 src_port=12345 dst_port=23456 dscp=5 protocol=udp"
|
|
if "OK" not in dev[0].request(cmd):
|
|
raise Exception("SCS change failed: " + t)
|
|
if "FAIL" not in dev[0].request(cmd):
|
|
raise Exception("SCS change failed: " + t)
|
|
handle_scs_req(hapd, twice=True)
|
|
wait_scs_result(dev[0])
|
|
ev = dev[0].wait_event(["CTRL-EVENT-SCS-RESULT"], timeout=0.1)
|
|
if ev is not None:
|
|
raise Exception("Unexpected SCS result reported(1)")
|
|
|
|
if "OK" not in dev[0].request(cmd):
|
|
raise Exception("SCS change failed: " + t)
|
|
handle_scs_req(hapd, short=True)
|
|
ev = dev[0].wait_event(["CTRL-EVENT-SCS-RESULT"], timeout=3)
|
|
if ev is not None:
|
|
raise Exception("Unexpected SCS result reported(2)")
|
|
|
|
cmd = "SCS scs_id=123 add scs_up=6 classifier_type=4 ip_version=ipv4 src_ip=1.2.3.4 dst_ip=5.6.7.8 src_port=12345 dst_port=23456 dscp=5 protocol=udp"
|
|
if "OK" not in dev[0].request(cmd):
|
|
raise Exception("SCS add failed: " + t)
|
|
handle_scs_req(hapd, scsid=34)
|
|
wait_scs_result(dev[0], expect_status="response_not_received")
|
|
|
|
cmd = "SCS scs_id=33 add scs_up=6 classifier_type=4 ip_version=ipv4 src_ip=1.2.3.4 dst_ip=5.6.7.8 src_port=12345 dst_port=23456 dscp=5 protocol=udp"
|
|
if "OK" not in dev[0].request(cmd):
|
|
raise Exception("SCS add failed: " + t)
|
|
handle_scs_req(hapd, scsid=33, status_code=123)
|
|
wait_scs_result(dev[0], expect_status="123")
|
|
|
|
def test_scs_request_without_ap_capa(dev, apdev):
|
|
"""SCS Request without AP capability"""
|
|
params = {"ssid": "scs"}
|
|
hapd = hostapd.add_ap(apdev[0], params)
|
|
|
|
dev[0].connect("scs", key_mgmt="NONE", scan_freq="2412")
|
|
|
|
cmd = "SCS scs_id=1 add scs_up=7 classifier_type=10 prot_instance=1 prot_number=udp filter_value=11223344 filter_mask=ffffffff"
|
|
if "FAIL" not in dev[0].request(cmd):
|
|
raise Exception("SCS add accepted")
|