Signed-off-by: Masashi Honma <masashi.honma@xxxxxxxxx>
---
tests/hwsim/test_ap_wps.py | 491 ++++++++++++++++++++-----------------
1 file changed, 267 insertions(+), 224 deletions(-)
diff --git a/tests/hwsim/test_ap_wps.py b/tests/hwsim/test_ap_wps.py
index 7afe72c02..4044a2939 100644
--- a/tests/hwsim/test_ap_wps.py
+++ b/tests/hwsim/test_ap_wps.py
@@ -12,6 +12,7 @@ import hashlib
import hmac
import os
import time
+import sys
import stat
import subprocess
import logging
@@ -177,7 +178,7 @@ def test_ap_wps_init_through_wps_config(dev, apdev):
ssid = "test-wps-init-config"
hapd = hostapd.add_ap(apdev[0],
{ "ssid": ssid, "eap_server": "1", "wps_state": "1" })
- if "FAIL" in hapd.request("WPS_CONFIG " + ssid.encode("hex") + " WPA2PSK CCMP " + "12345678".encode("hex")):
+ if "FAIL" in hapd.request("WPS_CONFIG " + binascii.hexlify(ssid.encode()).decode() + " WPA2PSK CCMP " + binascii.hexlify(b"12345678").decode()):
raise Exception("WPS_CONFIG command failed")
ev = hapd.wait_event(["WPS-NEW-AP-SETTINGS"], timeout=5)
if ev is None:
@@ -200,7 +201,7 @@ def test_ap_wps_init_through_wps_config_2(dev, apdev):
hapd = hostapd.add_ap(apdev[0],
{ "ssid": ssid, "eap_server": "1", "wps_state": "1",
"wps_cred_processing": "2" })
- if "FAIL" in hapd.request("WPS_CONFIG " + ssid.encode("hex") + " WPA2PSK CCMP " + "12345678".encode("hex")):
+ if "FAIL" in hapd.request("WPS_CONFIG " + binascii.hexlify(ssid.encode()).decode() + " WPA2PSK CCMP " + binascii.hexlify(b"12345678").decode()):
raise Exception("WPS_CONFIG command failed")
ev = hapd.wait_event(["WPS-NEW-AP-SETTINGS"], timeout=5)
if ev is None:
@@ -214,7 +215,7 @@ def test_ap_wps_invalid_wps_config_passphrase(dev, apdev):
ssid = "test-wps-init-config"
hapd = hostapd.add_ap(apdev[0],
{ "ssid": ssid, "eap_server": "1", "wps_state": "1" })
- if "FAIL" not in hapd.request("WPS_CONFIG " + ssid.encode("hex") + " WPA2PSK CCMP " + "1234567".encode("hex")):
+ if "FAIL" not in hapd.request("WPS_CONFIG " + binascii.hexlify(ssid.encode()).decode() + " WPA2PSK CCMP " + binascii.hexlify(b"1234567").decode()):
raise Exception("Invalid WPS_CONFIG command accepted")
def test_ap_wps_conf(dev, apdev):
@@ -612,13 +613,13 @@ def test_ap_wps_reg_override_ap_settings(dev, apdev):
except:
pass
# Override AP Settings with values that point to another AP
- data = build_wsc_attr(ATTR_NETWORK_INDEX, '\x01')
- data += build_wsc_attr(ATTR_SSID, "test")
- data += build_wsc_attr(ATTR_AUTH_TYPE, '\x00\x01')
- data += build_wsc_attr(ATTR_ENCR_TYPE, '\x00\x01')
- data += build_wsc_attr(ATTR_NETWORK_KEY, '')
+ data = build_wsc_attr(ATTR_NETWORK_INDEX, b'\x01')
+ data += build_wsc_attr(ATTR_SSID, b"test")
+ data += build_wsc_attr(ATTR_AUTH_TYPE, b'\x00\x01')
+ data += build_wsc_attr(ATTR_ENCR_TYPE, b'\x00\x01')
+ data += build_wsc_attr(ATTR_NETWORK_KEY, bytes())
data += build_wsc_attr(ATTR_MAC_ADDR, binascii.unhexlify(apdev[1]['bssid'].replace(':', '')))
- with open(ap_settings, "w") as f:
+ with open(ap_settings, "wb") as f:
f.write(data)
ssid = "test-wps-reg-ap-pin"
appin = "12345670"
@@ -771,7 +772,7 @@ def test_ap_wps_reg_config_ext_processing(dev, apdev):
if "1026" not in ev:
raise Exception("AP Settings missing from event")
hapd.request("SET wps_cred_processing 0")
- if "FAIL" in hapd.request("WPS_CONFIG " + new_ssid.encode("hex") + " WPA2PSK CCMP " + new_passphrase.encode("hex")):
+ if "FAIL" in hapd.request("WPS_CONFIG " + binascii.hexlify(new_ssid.encode()).decode() + " WPA2PSK CCMP " + binascii.hexlify(new_passphrase.encode()).decode()):
raise Exception("WPS_CONFIG command failed")
dev[0].wait_connected(timeout=15)
@@ -1572,8 +1573,8 @@ def _test_ap_wps_er_config_ap(dev, apdev):
raise Exception("Expected AP UUID not found")
new_passphrase = "1234567890"
dev[0].request("WPS_ER_CONFIG " + apdev[0]['bssid'] + " " + ap_pin + " " +
- ssid.encode("hex") + " WPA2PSK CCMP " +
- new_passphrase.encode("hex"))
+ binascii.hexlify(ssid.encode()).decode() + " WPA2PSK CCMP " +
+ binascii.hexlify(new_passphrase.encode()).decode())
ev = dev[0].wait_event(["WPS-SUCCESS"])
if ev is None:
raise Exception("WPS ER configuration operation timed out")
@@ -2486,7 +2487,7 @@ def ssdp_send(msg, no_recv=False):
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 2)
sock.bind(("127.0.0.1", 0))
- sock.sendto(msg, ("239.255.255.250", 1900))
+ sock.sendto(msg.encode(), ("239.255.255.250", 1900))
if no_recv:
return None
return sock.recv(1000)
@@ -2557,7 +2558,7 @@ def test_ap_wps_ssdp_invalid_msearch(dev, apdev):
'MAN: "ssdp:discover"',
'ST: urn:schemas-wifialliance-org:device:WFADevice:1',
'', ''])
- sock.sendto(msg, ("239.255.255.250", 1900))
+ sock.sendto(msg.encode(), ("239.255.255.250", 1900))
logger.debug("Negative MX")
msg = '\r\n'.join([
@@ -2567,7 +2568,7 @@ def test_ap_wps_ssdp_invalid_msearch(dev, apdev):
'MAN: "ssdp:discover"',
'ST: urn:schemas-wifialliance-org:device:WFADevice:1',
'', ''])
- sock.sendto(msg, ("239.255.255.250", 1900))
+ sock.sendto(msg.encode(), ("239.255.255.250", 1900))
logger.debug("Invalid MX")
msg = '\r\n'.join([
@@ -2577,7 +2578,7 @@ def test_ap_wps_ssdp_invalid_msearch(dev, apdev):
'MAN: "ssdp:discover"',
'ST: urn:schemas-wifialliance-org:device:WFADevice:1',
'', ''])
- sock.sendto(msg, ("239.255.255.250", 1900))
+ sock.sendto(msg.encode(), ("239.255.255.250", 1900))
logger.debug("Missing MAN")
msg = '\r\n'.join([
@@ -2586,7 +2587,7 @@ def test_ap_wps_ssdp_invalid_msearch(dev, apdev):
'MX: 1',
'ST: urn:schemas-wifialliance-org:device:WFADevice:1',
'', ''])
- sock.sendto(msg, ("239.255.255.250", 1900))
+ sock.sendto(msg.encode(), ("239.255.255.250", 1900))
logger.debug("Invalid MAN")
msg = '\r\n'.join([
@@ -2596,7 +2597,7 @@ def test_ap_wps_ssdp_invalid_msearch(dev, apdev):
'MAN: foo',
'ST: urn:schemas-wifialliance-org:device:WFADevice:1',
'', ''])
- sock.sendto(msg, ("239.255.255.250", 1900))
+ sock.sendto(msg.encode(), ("239.255.255.250", 1900))
msg = '\r\n'.join([
'M-SEARCH * HTTP/1.1',
'HOST: 239.255.255.250:1900',
@@ -2604,7 +2605,7 @@ def test_ap_wps_ssdp_invalid_msearch(dev, apdev):
'MAN; "ssdp:discover"',
'ST: urn:schemas-wifialliance-org:device:WFADevice:1',
'', ''])
- sock.sendto(msg, ("239.255.255.250", 1900))
+ sock.sendto(msg.encode(), ("239.255.255.250", 1900))
logger.debug("Missing HOST")
msg = '\r\n'.join([
@@ -2613,7 +2614,7 @@ def test_ap_wps_ssdp_invalid_msearch(dev, apdev):
'MX: 1',
'ST: urn:schemas-wifialliance-org:device:WFADevice:1',
'', ''])
- sock.sendto(msg, ("239.255.255.250", 1900))
+ sock.sendto(msg.encode(), ("239.255.255.250", 1900))
logger.debug("Missing ST")
msg = '\r\n'.join([
@@ -2622,7 +2623,7 @@ def test_ap_wps_ssdp_invalid_msearch(dev, apdev):
'MAN: "ssdp:discover"',
'MX: 1',
'', ''])
- sock.sendto(msg, ("239.255.255.250", 1900))
+ sock.sendto(msg.encode(), ("239.255.255.250", 1900))
logger.debug("Mismatching ST")
msg = '\r\n'.join([
@@ -2632,7 +2633,7 @@ def test_ap_wps_ssdp_invalid_msearch(dev, apdev):
'MX: 1',
'ST: uuid:16d5f8a9-4ee4-4f5e-81f9-cc6e2f47f42d',
'', ''])
- sock.sendto(msg, ("239.255.255.250", 1900))
+ sock.sendto(msg.encode(), ("239.255.255.250", 1900))
msg = '\r\n'.join([
'M-SEARCH * HTTP/1.1',
'HOST: 239.255.255.250:1900',
@@ -2640,7 +2641,7 @@ def test_ap_wps_ssdp_invalid_msearch(dev, apdev):
'MX: 1',
'ST: foo:bar',
'', ''])
- sock.sendto(msg, ("239.255.255.250", 1900))
+ sock.sendto(msg.encode(), ("239.255.255.250", 1900))
msg = '\r\n'.join([
'M-SEARCH * HTTP/1.1',
'HOST: 239.255.255.250:1900',
@@ -2648,7 +2649,7 @@ def test_ap_wps_ssdp_invalid_msearch(dev, apdev):
'MX: 1',
'ST: foobar',
'', ''])
- sock.sendto(msg, ("239.255.255.250", 1900))
+ sock.sendto(msg.encode(), ("239.255.255.250", 1900))
logger.debug("Invalid ST")
msg = '\r\n'.join([
@@ -2658,7 +2659,7 @@ def test_ap_wps_ssdp_invalid_msearch(dev, apdev):
'MX: 1',
'ST; urn:schemas-wifialliance-org:device:WFADevice:1',
'', ''])
- sock.sendto(msg, ("239.255.255.250", 1900))
+ sock.sendto(msg.encode(), ("239.255.255.250", 1900))
logger.debug("Invalid M-SEARCH")
msg = '\r\n'.join([
@@ -2668,7 +2669,7 @@ def test_ap_wps_ssdp_invalid_msearch(dev, apdev):
'MX: 1',
'ST: urn:schemas-wifialliance-org:device:WFADevice:1',
'', ''])
- sock.sendto(msg, ("239.255.255.250", 1900))
+ sock.sendto(msg.encode(), ("239.255.255.250", 1900))
msg = '\r\n'.join([
'M-SEARCH-* HTTP/1.1',
'HOST: 239.255.255.250:1900',
@@ -2676,10 +2677,10 @@ def test_ap_wps_ssdp_invalid_msearch(dev, apdev):
'MX: 1',
'ST: urn:schemas-wifialliance-org:device:WFADevice:1',
'', ''])
- sock.sendto(msg, ("239.255.255.250", 1900))
+ sock.sendto(msg.encode(), ("239.255.255.250", 1900))
logger.debug("Invalid message format")
- sock.sendto("NOTIFY * HTTP/1.1", ("239.255.255.250", 1900))
+ sock.sendto(b"NOTIFY * HTTP/1.1", ("239.255.255.250", 1900))
msg = '\r'.join([
'M-SEARCH * HTTP/1.1',
'HOST: 239.255.255.250:1900',
@@ -2687,7 +2688,7 @@ def test_ap_wps_ssdp_invalid_msearch(dev, apdev):
'MX: 1',
'ST: urn:schemas-wifialliance-org:device:WFADevice:1',
'', ''])
- sock.sendto(msg, ("239.255.255.250", 1900))
+ sock.sendto(msg.encode(), ("239.255.255.250", 1900))
try:
r = sock.recv(1000)
@@ -2703,7 +2704,7 @@ def test_ap_wps_ssdp_invalid_msearch(dev, apdev):
'MX: 1',
'ST: urn:schemas-wifialliance-org:device:WFADevice:1',
'', ''])
- sock.sendto(msg, ("239.255.255.250", 1900))
+ sock.sendto(msg.encode(), ("239.255.255.250", 1900))
try:
r = sock.recv(1000)
@@ -2729,11 +2730,11 @@ def test_ap_wps_ssdp_burst(dev, apdev):
sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 2)
sock.bind(("127.0.0.1", 0))
for i in range(0, 25):
- sock.sendto(msg, ("239.255.255.250", 1900))
+ sock.sendto(msg.encode(), ("239.255.255.250", 1900))
resp = 0
while True:
try:
- r = sock.recv(1000)
+ r = sock.recv(1000).decode()
if not r.startswith("HTTP/1.1 200 OK\r\n"):
raise Exception("Unexpected message: " + r)
resp += 1
@@ -2747,10 +2748,10 @@ def test_ap_wps_ssdp_burst(dev, apdev):
sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 2)
sock.bind(("127.0.0.1", 0))
for i in range(0, 25):
- sock.sendto(msg, ("239.255.255.250", 1900))
+ sock.sendto(msg.encode(), ("239.255.255.250", 1900))
while True:
try:
- r = sock.recv(1000)
+ r = sock.recv(1000).decode()
if ap_uuid in r:
break
except socket.timeout:
@@ -2760,6 +2761,7 @@ def ssdp_get_location(uuid):
res = ssdp_send_msearch("uuid:" + uuid)
location = None
for l in res.splitlines():
+ l = l.decode()
if l.lower().startswith("location:"):
location = l.split(':', 1)[1].strip()
break
@@ -2768,7 +2770,10 @@ def ssdp_get_location(uuid):
return location
def upnp_get_urls(location):
- conn = urlopen(location, proxies={})
+ if sys.version_info[0] > 2:
+ conn = urlopen(location)
+ else:
+ conn = urlopen(location, proxies={})
tree = ET.parse(conn)
root = tree.getroot()
urn = '{urn:schemas-upnp-org:device-1-0}'
@@ -2793,23 +2798,19 @@ def upnp_soap_action(conn, path, action, include_soap_action=True,
act = ET.SubElement(body, "{%s}%s" % (wpsns, action))
if newmsg:
msg = ET.SubElement(act, "NewMessage")
- msg.text = base64.b64encode(newmsg)
+ msg.text = base64.b64encode(newmsg.encode()).decode()
if neweventtype:
msg = ET.SubElement(act, "NewWLANEventType")
msg.text = neweventtype
if neweventmac:
msg = ET.SubElement(act, "NewWLANEventMAC")
msg.text = neweventmac
- tree = ET.ElementTree(root)
- soap = StringIO()
- tree.write(soap, xml_declaration=True, encoding='utf-8')
-
headers = { "Content-type": 'text/xml; charset="utf-8"' }
if include_soap_action:
headers["SOAPAction"] = '"urn:schemas-wifialliance-org:service:WFAWLANConfig:1#%s"' % action
elif soap_action_override:
headers["SOAPAction"] = soap_action_override
- conn.request("POST", path, soap.getvalue(), headers)
+ conn.request("POST", path, ET.tostring(root), headers)
return conn.getresponse()
def test_ap_wps_upnp(dev, apdev):
@@ -2820,13 +2821,23 @@ def test_ap_wps_upnp(dev, apdev):
location = ssdp_get_location(ap_uuid)
urls = upnp_get_urls(location)
- conn = urlopen(urls['scpd_url'], proxies={})
+ if sys.version_info[0] > 2:
+ conn = urlopen(urls['scpd_url'])
+ else:
+ conn = urlopen(urls['scpd_url'], proxies={})
scpd = conn.read()
- conn = urlopen(urljoin(location, "unknown.html"),
- proxies={})
- if conn.getcode() != 404:
- raise Exception("Unexpected HTTP response to GET unknown URL")
+ if sys.version_info[0] > 2:
+ try:
+ conn = urlopen(urljoin(location, "unknown.html"))
+ raise Exception("Unexpected HTTP response to GET unknown URL")
+ except HTTPError as e:
+ if e.code != 404:
+ raise Exception("Unexpected HTTP response to GET unknown URL")
+ else:
+ conn = urlopen(urljoin(location, "unknown.html"), proxies={})
+ if conn.getcode() != 404:
+ raise Exception("Unexpected HTTP response to GET unknown URL")
url = urlparse(location)
conn = HTTPConnection(url.netloc)
@@ -2871,7 +2882,7 @@ def test_ap_wps_upnp(dev, apdev):
resp = upnp_soap_action(conn, ctrlurl.path, "GetDeviceInfo")
if resp.status != 200:
raise Exception("Unexpected HTTP response: %d" % resp.status)
- dev = resp.read()
+ dev = resp.read().decode()
if "NewDeviceInfo" not in dev:
raise Exception("Unexpected GetDeviceInfo response")
@@ -3433,10 +3444,10 @@ def test_ap_wps_upnp_http_proto_chunked(dev, apdev):
conn.putrequest("POST", "hello")
conn.putheader('Transfer-Encoding', 'chunked')
conn.endheaders()
- conn.send("a\r\nabcdefghij\r\n")
+ conn.send(b"a\r\nabcdefghij\r\n")
time.sleep(0.1)
- conn.send("2\r\nkl\r\n")
- conn.send("0\r\n\r\n")
+ conn.send(b"2\r\nkl\r\n")
+ conn.send(b"0\r\n\r\n")
resp = conn.getresponse()
if resp.status != 404:
raise Exception("Unexpected HTTP response: %d" % resp.status)
@@ -3448,8 +3459,8 @@ def test_ap_wps_upnp_http_proto_chunked(dev, apdev):
completed = False
try:
for i in range(20000):
- conn.send("1\r\nZ\r\n")
- conn.send("0\r\n\r\n")
+ conn.send(b"1\r\nZ\r\n")
+ conn.send(b"0\r\n\r\n")
resp = conn.getresponse()
completed = True
except Exception as e:
@@ -3904,7 +3915,7 @@ def test_ap_wps_wpa_cli_action(dev, apdev, test_params):
arg = [ 'ps', 'ax' ]
cmd = subprocess.Popen(arg, stdout=subprocess.PIPE)
- out = cmd.communicate()[0]
+ out = cmd.communicate()[0].decode()
cmd.wait()
logger.debug("Processes:\n" + out)
if "wpa_cli -P %s -B -i %s" % (pidfile, dev[0].ifname) not in out:
@@ -3930,7 +3941,7 @@ def test_ap_wps_wpa_cli_action(dev, apdev, test_params):
arg = [ 'ps', 'ax' ]
cmd = subprocess.Popen(arg, stdout=subprocess.PIPE)
- out = cmd.communicate()[0]
+ out = cmd.communicate()[0].decode()
cmd.wait()
logger.debug("Remaining processes:\n" + out)
if "wpa_cli -P %s -B -i %s" % (pidfile, dev[0].ifname) in out:
@@ -3956,49 +3967,51 @@ def _test_ap_wps_er_ssdp_proto(dev, apdev):
if "OK" not in dev[0].request("WPS_ER_START ifname=lo 1.2.3.4"):
raise Exception("WPS_ER_START with filter failed")
(msg,addr) = sock.recvfrom(1000)
+ msg = msg.decode()
logger.debug("Received SSDP message from %s: %s" % (str(addr), msg))
if "M-SEARCH" not in msg:
raise Exception("Not an M-SEARCH")
- sock.sendto("FOO", addr)
+ sock.sendto(b"FOO", addr)
time.sleep(0.1)
dev[0].request("WPS_ER_STOP")
dev[0].request("WPS_ER_START ifname=lo")
(msg,addr) = sock.recvfrom(1000)
+ msg = msg.decode()
logger.debug("Received SSDP message from %s: %s" % (str(addr), msg))
if "M-SEARCH" not in msg:
raise Exception("Not an M-SEARCH")
- sock.sendto("FOO", addr)
- sock.sendto("HTTP/1.1 200 OK\r\nFOO\r\n\r\n", addr)
- sock.sendto("HTTP/1.1 200 OK\r\nNTS:foo\r\n\r\n", addr)
- sock.sendto("HTTP/1.1 200 OK\r\nNTS:ssdp:byebye\r\n\r\n", addr)
- sock.sendto("HTTP/1.1 200 OK\r\ncache-control: foo=1\r\n\r\n", addr)
- sock.sendto("HTTP/1.1 200 OK\r\ncache-control: max-age=1\r\n\r\n", addr)
- sock.sendto("HTTP/1.1 200 OK\r\nusn:\r\n\r\n", addr)
- sock.sendto("HTTP/1.1 200 OK\r\nusn:foo\r\n\r\n", addr)
- sock.sendto("HTTP/1.1 200 OK\r\nusn: uuid:\r\n\r\n", addr)
- sock.sendto("HTTP/1.1 200 OK\r\nusn: uuid: \r\n\r\n", addr)
- sock.sendto("HTTP/1.1 200 OK\r\nusn: uuid: foo\r\n\r\n", addr)
- sock.sendto("HTTP/1.1 200 OK\r\nST: urn:schemas-wifialliance-org:device:WFADevice:1\r\n\r\n", addr)
- sock.sendto("HTTP/1.1 200 OK\r\nST: urn:schemas-wifialliance-org:device:WFADevice:1\r\nNTS:ssdp:byebye\r\n\r\n", addr)
- sock.sendto("HTTP/1.1 200 OK\r\nST: urn:schemas-wifialliance-org:device:WFADevice:1\r\nlocation:foo\r\n\r\n", addr)
+ sock.sendto(b"FOO", addr)
+ sock.sendto(b"HTTP/1.1 200 OK\r\nFOO\r\n\r\n", addr)
+ sock.sendto(b"HTTP/1.1 200 OK\r\nNTS:foo\r\n\r\n", addr)
+ sock.sendto(b"HTTP/1.1 200 OK\r\nNTS:ssdp:byebye\r\n\r\n", addr)
+ sock.sendto(b"HTTP/1.1 200 OK\r\ncache-control: foo=1\r\n\r\n", addr)
+ sock.sendto(b"HTTP/1.1 200 OK\r\ncache-control: max-age=1\r\n\r\n", addr)
+ sock.sendto(b"HTTP/1.1 200 OK\r\nusn:\r\n\r\n", addr)
+ sock.sendto(b"HTTP/1.1 200 OK\r\nusn:foo\r\n\r\n", addr)
+ sock.sendto(b"HTTP/1.1 200 OK\r\nusn: uuid:\r\n\r\n", addr)
+ sock.sendto(b"HTTP/1.1 200 OK\r\nusn: uuid: \r\n\r\n", addr)
+ sock.sendto(b"HTTP/1.1 200 OK\r\nusn: uuid: foo\r\n\r\n", addr)
+ sock.sendto(b"HTTP/1.1 200 OK\r\nST: urn:schemas-wifialliance-org:device:WFADevice:1\r\n\r\n", addr)
+ sock.sendto(b"HTTP/1.1 200 OK\r\nST: urn:schemas-wifialliance-org:device:WFADevice:1\r\nNTS:ssdp:byebye\r\n\r\n", addr)
+ sock.sendto(b"HTTP/1.1 200 OK\r\nST: urn:schemas-wifialliance-org:device:WFADevice:1\r\nlocation:foo\r\n\r\n", addr)
with alloc_fail(dev[0], 1, "wps_er_ap_add"):
- sock.sendto("HTTP/1.1 200 OK\r\nST: urn:schemas-wifialliance-org:device:WFADevice:1\r\nlocation:foo\r\ncache-control:max-age=1\r\n\r\n", addr)
+ sock.sendto(b"HTTP/1.1 200 OK\r\nST: urn:schemas-wifialliance-org:device:WFADevice:1\r\nlocation:foo\r\ncache-control:max-age=1\r\n\r\n", addr)
time.sleep(0.1)
with alloc_fail(dev[0], 2, "wps_er_ap_add"):
- sock.sendto("HTTP/1.1 200 OK\r\nST: urn:schemas-wifialliance-org:device:WFADevice:1\r\nlocation:foo\r\ncache-control:max-age=1\r\n\r\n", addr)
+ sock.sendto(b"HTTP/1.1 200 OK\r\nST: urn:schemas-wifialliance-org:device:WFADevice:1\r\nlocation:foo\r\ncache-control:max-age=1\r\n\r\n", addr)
time.sleep(0.1)
# Add an AP with bogus URL
- sock.sendto("HTTP/1.1 200 OK\r\nST: urn:schemas-wifialliance-org:device:WFADevice:1\r\nlocation:foo\r\ncache-control:max-age=1\r\n\r\n", addr)
+ sock.sendto(b"HTTP/1.1 200 OK\r\nST: urn:schemas-wifialliance-org:device:WFADevice:1\r\nlocation:foo\r\ncache-control:max-age=1\r\n\r\n", addr)
# Update timeout on AP without updating URL
- sock.sendto("HTTP/1.1 200 OK\r\nST: urn:schemas-wifialliance-org:device:WFADevice:1\r\nlocation:http://127.0.0.1:12345/foo.xml\r\ncache-control:max-age=1\r\n\r\n";, addr)
+ sock.sendto(b"HTTP/1.1 200 OK\r\nST: urn:schemas-wifialliance-org:device:WFADevice:1\r\nlocation:http://127.0.0.1:12345/foo.xml\r\ncache-control:max-age=1\r\n\r\n";, addr)
ev = dev[0].wait_event(["WPS-ER-AP-REMOVE"], timeout=5)
if ev is None:
raise Exception("No WPS-ER-AP-REMOVE event on max-age timeout")
# Add an AP with a valid URL (but no server listing to it)
- sock.sendto("HTTP/1.1 200 OK\r\nST: urn:schemas-wifialliance-org:device:WFADevice:1\r\nlocation:http://127.0.0.1:12345/foo.xml\r\ncache-control:max-age=1\r\n\r\n";, addr)
+ sock.sendto(b"HTTP/1.1 200 OK\r\nST: urn:schemas-wifialliance-org:device:WFADevice:1\r\nlocation:http://127.0.0.1:12345/foo.xml\r\ncache-control:max-age=1\r\n\r\n";, addr)
ev = dev[0].wait_event(["WPS-ER-AP-REMOVE"], timeout=5)
if ev is None:
raise Exception("No WPS-ER-AP-REMOVE event on max-age timeout")
@@ -4046,7 +4059,7 @@ def gen_upnp_info(eventSubURL='wps_event', controlURL='wps_control',
'Connection: close\r\n' + \
'Content-Length: ' + str(len(payload)) + '\r\n' + \
'Date: Sat, 15 Aug 2015 18:55:08 GMT\r\n\r\n'
- return hdr + payload
+ return (hdr + payload).encode()
def gen_wps_control(payload_override=None):
payload = '''<?xml version="1.0"?>
@@ -4074,7 +4087,7 @@ AAYANyoAASA=
'Connection: close\r\n' + \
'Content-Length: ' + str(len(payload)) + '\r\n' + \
'Date: Sat, 15 Aug 2015 18:55:08 GMT\r\n\r\n'
- return hdr + payload
+ return (hdr + payload).encode()
def gen_wps_event(sid='uuid:7eb3342a-8a5f-47fe-a585-0785bfec6d8a'):
payload = ""
@@ -4087,14 +4100,14 @@ def gen_wps_event(sid='uuid:7eb3342a-8a5f-47fe-a585-0785bfec6d8a'):
hdr += 'SID: ' + sid + '\r\n'
hdr += 'Timeout: Second-1801\r\n' + \
'Date: Sat, 15 Aug 2015 18:55:08 GMT\r\n\r\n'
- return hdr + payload
+ return (hdr + payload).encode()
class WPSAPHTTPServer(StreamRequestHandler):
def handle(self):
- data = self.rfile.readline().strip()
+ data = self.rfile.readline().decode().strip()
logger.info("HTTP server received: " + data)
while True:
- hdr = self.rfile.readline().strip()
+ hdr = self.rfile.readline().decode().strip()
if len(hdr) == 0:
break
logger.info("HTTP header: " + hdr)
@@ -4137,6 +4150,7 @@ def wps_er_start(dev, http_server, max_age=1, wait_m_search=False,
dev.request("WPS_ER_START ifname=lo")
for i in range(100):
(msg,addr) = sock.recvfrom(1000)
+ msg = msg.decode()
logger.debug("Received SSDP message from %s: %s" % (str(addr), msg))
if "M-SEARCH" in msg:
break
@@ -4149,7 +4163,7 @@ def wps_er_start(dev, http_server, max_age=1, wait_m_search=False,
server = MyTCPServer(("127.0.0.1", 12345), http_server)
if not location_url:
location_url = 'http://127.0.0.1:12345/foo.xml'
- sock.sendto("HTTP/1.1 200 OK\r\nST: urn:schemas-wifialliance-org:device:WFADevice:1\r\nlocation:%s\r\ncache-control:max-age=%d\r\n\r\n" % (location_url, max_age), addr)
+ sock.sendto(("HTTP/1.1 200 OK\r\nST: urn:schemas-wifialliance-org:device:WFADevice:1\r\nlocation:%s\r\ncache-control:max-age=%d\r\n\r\n" % (location_url, max_age)).encode(), addr)
server.timeout = 1
return server,sock
@@ -4200,7 +4214,7 @@ def send_wlanevent(url, uuid, data, no_response=False):
<e:property><STAStatus>1</STAStatus></e:property>
<e:property><APStatus>1</APStatus></e:property>
<e:property><WLANEvent>'''
- payload += base64.b64encode(data)
+ payload += base64.b64encode(data).decode()
payload += '</WLANEvent></e:property></e:propertyset>'
headers = { "Content-type": 'text/xml; charset="utf-8"',
"Server": "Unspecified, UPnP/1.0, Unspecified",
@@ -4318,164 +4332,164 @@ RGV2aWNlIEEQSQAGADcqAAEg
time.sleep(0.5)
logger.info("Too short WLANEvent")
- data = '\x00'
+ data = b'\x00'
send_wlanevent(url, uuid, data)
logger.info("Invalid WLANEventMAC")
- data = '\x00qwertyuiopasdfghjklzxcvbnm'
+ data = b'\x00qwertyuiopasdfghjklzxcvbnm'
send_wlanevent(url, uuid, data)
logger.info("Unknown WLANEventType")
- data = '\xff02:00:00:00:00:00'
+ data = b'\xff02:00:00:00:00:00'
send_wlanevent(url, uuid, data)
logger.info("Probe Request notification without any attributes")
- data = '\x0102:00:00:00:00:00'
+ data = b'\x0102:00:00:00:00:00'
send_wlanevent(url, uuid, data)
logger.info("Probe Request notification with invalid attribute")
- data = '\x0102:00:00:00:00:00\xff'
+ data = b'\x0102:00:00:00:00:00\xff'
send_wlanevent(url, uuid, data)
logger.info("EAP message without any attributes")
- data = '\x0202:00:00:00:00:00'
+ data = b'\x0202:00:00:00:00:00'
send_wlanevent(url, uuid, data)
logger.info("EAP message with invalid attribute")
- data = '\x0202:00:00:00:00:00\xff'
+ data = b'\x0202:00:00:00:00:00\xff'
send_wlanevent(url, uuid, data)
logger.info("EAP message from new STA and not M1")
- data = '\x0202:ff:ff:ff:ff:ff' + '\x10\x22\x00\x01\x05'
+ data = b'\x0202:ff:ff:ff:ff:ff' + b'\x10\x22\x00\x01\x05'
send_wlanevent(url, uuid, data)
logger.info("EAP message: M1")
- data = '\x0202:00:00:00:00:00'
- data += '\x10\x22\x00\x01\x04'
- data += '\x10\x47\x00\x10' + 16*'\x00'
- data += '\x10\x20\x00\x06\x02\x00\x00\x00\x00\x00'
- data += '\x10\x1a\x00\x10' + 16*'\x00'
- data += '\x10\x32\x00\xc0' + 192*'\x00'
- data += '\x10\x04\x00\x02\x00\x00'
- data += '\x10\x10\x00\x02\x00\x00'
- data += '\x10\x0d\x00\x01\x00'
- data += '\x10\x08\x00\x02\x00\x00'
- data += '\x10\x44\x00\x01\x00'
- data += '\x10\x21\x00\x00'
- data += '\x10\x23\x00\x00'
- data += '\x10\x24\x00\x00'
- data += '\x10\x42\x00\x00'
- data += '\x10\x54\x00\x08' + 8*'\x00'
- data += '\x10\x11\x00\x00'
- data += '\x10\x3c\x00\x01\x00'
- data += '\x10\x02\x00\x02\x00\x00'
- data += '\x10\x12\x00\x02\x00\x00'
- data += '\x10\x09\x00\x02\x00\x00'
- data += '\x10\x2d\x00\x04\x00\x00\x00\x00'
+ data = b'\x0202:00:00:00:00:00'
+ data += b'\x10\x22\x00\x01\x04'
+ data += b'\x10\x47\x00\x10' + 16 * b'\x00'
+ data += b'\x10\x20\x00\x06\x02\x00\x00\x00\x00\x00'
+ data += b'\x10\x1a\x00\x10' + 16 * b'\x00'
+ data += b'\x10\x32\x00\xc0' + 192 * b'\x00'
+ data += b'\x10\x04\x00\x02\x00\x00'
+ data += b'\x10\x10\x00\x02\x00\x00'
+ data += b'\x10\x0d\x00\x01\x00'
+ data += b'\x10\x08\x00\x02\x00\x00'
+ data += b'\x10\x44\x00\x01\x00'
+ data += b'\x10\x21\x00\x00'
+ data += b'\x10\x23\x00\x00'
+ data += b'\x10\x24\x00\x00'
+ data += b'\x10\x42\x00\x00'
+ data += b'\x10\x54\x00\x08' + 8 * b'\x00'
+ data += b'\x10\x11\x00\x00'
+ data += b'\x10\x3c\x00\x01\x00'
+ data += b'\x10\x02\x00\x02\x00\x00'
+ data += b'\x10\x12\x00\x02\x00\x00'
+ data += b'\x10\x09\x00\x02\x00\x00'
+ data += b'\x10\x2d\x00\x04\x00\x00\x00\x00'
m1 = data
send_wlanevent(url, uuid, data)
logger.info("EAP message: WSC_ACK")
- data = '\x0202:00:00:00:00:00' + '\x10\x22\x00\x01\x0d'
+ data = b'\x0202:00:00:00:00:00' + b'\x10\x22\x00\x01\x0d'
send_wlanevent(url, uuid, data)
logger.info("EAP message: M1")
send_wlanevent(url, uuid, m1)
logger.info("EAP message: WSC_NACK")
- data = '\x0202:00:00:00:00:00' + '\x10\x22\x00\x01\x0e'
+ data = b'\x0202:00:00:00:00:00' + b'\x10\x22\x00\x01\x0e'
send_wlanevent(url, uuid, data)
logger.info("EAP message: M1 - Too long attribute values")
- data = '\x0202:00:00:00:00:00'
- data += '\x10\x11\x00\x21' + 33*'\x00'
- data += '\x10\x45\x00\x21' + 33*'\x00'
- data += '\x10\x42\x00\x21' + 33*'\x00'
- data += '\x10\x24\x00\x21' + 33*'\x00'
- data += '\x10\x23\x00\x21' + 33*'\x00'
- data += '\x10\x21\x00\x41' + 65*'\x00'
- data += '\x10\x49\x00\x09\x00\x37\x2a\x05\x02\x00\x00\x05\x00'
+ data = b'\x0202:00:00:00:00:00'
+ data += b'\x10\x11\x00\x21' + 33 * b'\x00'
+ data += b'\x10\x45\x00\x21' + 33 * b'\x00'
+ data += b'\x10\x42\x00\x21' + 33 * b'\x00'
+ data += b'\x10\x24\x00\x21' + 33 * b'\x00'
+ data += b'\x10\x23\x00\x21' + 33 * b'\x00'
+ data += b'\x10\x21\x00\x41' + 65 * b'\x00'
+ data += b'\x10\x49\x00\x09\x00\x37\x2a\x05\x02\x00\x00\x05\x00'
send_wlanevent(url, uuid, data)
logger.info("EAP message: M1 missing UUID-E")
- data = '\x0202:00:00:00:00:00'
- data += '\x10\x22\x00\x01\x04'
+ data = b'\x0202:00:00:00:00:00'
+ data += b'\x10\x22\x00\x01\x04'
send_wlanevent(url, uuid, data)
logger.info("EAP message: M1 missing MAC Address")
- data += '\x10\x47\x00\x10' + 16*'\x00'
+ data += b'\x10\x47\x00\x10' + 16 * b'\x00'
send_wlanevent(url, uuid, data)
logger.info("EAP message: M1 missing Enrollee Nonce")
- data += '\x10\x20\x00\x06\x02\x00\x00\x00\x00\x00'
+ data += b'\x10\x20\x00\x06\x02\x00\x00\x00\x00\x00'
send_wlanevent(url, uuid, data)
logger.info("EAP message: M1 missing Public Key")
- data += '\x10\x1a\x00\x10' + 16*'\x00'
+ data += b'\x10\x1a\x00\x10' + 16 * b'\x00'
send_wlanevent(url, uuid, data)
logger.info("EAP message: M1 missing Authentication Type flags")
- data += '\x10\x32\x00\xc0' + 192*'\x00'
+ data += b'\x10\x32\x00\xc0' + 192 * b'\x00'
send_wlanevent(url, uuid, data)
logger.info("EAP message: M1 missing Encryption Type Flags")
- data += '\x10\x04\x00\x02\x00\x00'
+ data += b'\x10\x04\x00\x02\x00\x00'
send_wlanevent(url, uuid, data)
logger.info("EAP message: M1 missing Connection Type flags")
- data += '\x10\x10\x00\x02\x00\x00'
+ data += b'\x10\x10\x00\x02\x00\x00'
send_wlanevent(url, uuid, data)
logger.info("EAP message: M1 missing Config Methods")
- data += '\x10\x0d\x00\x01\x00'
+ data += b'\x10\x0d\x00\x01\x00'
send_wlanevent(url, uuid, data)
logger.info("EAP message: M1 missing Wi-Fi Protected Setup State")
- data += '\x10\x08\x00\x02\x00\x00'
+ data += b'\x10\x08\x00\x02\x00\x00'
send_wlanevent(url, uuid, data)
logger.info("EAP message: M1 missing Manufacturer")
- data += '\x10\x44\x00\x01\x00'
+ data += b'\x10\x44\x00\x01\x00'
send_wlanevent(url, uuid, data)
logger.info("EAP message: M1 missing Model Name")
- data += '\x10\x21\x00\x00'
+ data += b'\x10\x21\x00\x00'
send_wlanevent(url, uuid, data)
logger.info("EAP message: M1 missing Model Number")
- data += '\x10\x23\x00\x00'
+ data += b'\x10\x23\x00\x00'
send_wlanevent(url, uuid, data)
logger.info("EAP message: M1 missing Serial Number")
- data += '\x10\x24\x00\x00'
+ data += b'\x10\x24\x00\x00'
send_wlanevent(url, uuid, data)
logger.info("EAP message: M1 missing Primary Device Type")
- data += '\x10\x42\x00\x00'
+ data += b'\x10\x42\x00\x00'
send_wlanevent(url, uuid, data)
logger.info("EAP message: M1 missing Device Name")
- data += '\x10\x54\x00\x08' + 8*'\x00'
+ data += b'\x10\x54\x00\x08' + 8 * b'\x00'
send_wlanevent(url, uuid, data)
logger.info("EAP message: M1 missing RF Bands")
- data += '\x10\x11\x00\x00'
+ data += b'\x10\x11\x00\x00'
send_wlanevent(url, uuid, data)
logger.info("EAP message: M1 missing Association State")
- data += '\x10\x3c\x00\x01\x00'
+ data += b'\x10\x3c\x00\x01\x00'
send_wlanevent(url, uuid, data)
logger.info("EAP message: M1 missing Device Password ID")
- data += '\x10\x02\x00\x02\x00\x00'
+ data += b'\x10\x02\x00\x02\x00\x00'
send_wlanevent(url, uuid, data)
logger.info("EAP message: M1 missing Configuration Error")
- data += '\x10\x12\x00\x02\x00\x00'
+ data += b'\x10\x12\x00\x02\x00\x00'
send_wlanevent(url, uuid, data)
logger.info("EAP message: M1 missing OS Version")
- data += '\x10\x09\x00\x02\x00\x00'
+ data += b'\x10\x09\x00\x02\x00\x00'
send_wlanevent(url, uuid, data)
logger.info("Check max concurrent requests")
@@ -4487,11 +4501,11 @@ RGV2aWNlIEEQSQAGADcqAAEg
socks[i].settimeout(10)
socks[i].connect(addr)
for i in range(20):
- socks[i].send("GET / HTTP/1.1\r\n\r\n")
+ socks[i].send(b"GET / HTTP/1.1\r\n\r\n")
count = 0
for i in range(20):
try:
- res = socks[i].recv(100)
+ res = socks[i].recv(100).decode()
if "HTTP/1" in res:
count += 1
else:
@@ -4512,7 +4526,7 @@ RGV2aWNlIEEQSQAGADcqAAEg
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM,
socket.IPPROTO_TCP)
sock.connect(addr)
- sock.send("GET / HTTP/1.1\r\n\r\n")
+ sock.send(b"GET / HTTP/1.1\r\n\r\n")
try:
sock.recv(100)
except:
@@ -4535,7 +4549,7 @@ RGV2aWNlIEEQSQAGADcqAAEg
socket.IPPROTO_TCP)
sock.settimeout(0.1)
sock.connect(addr)
- sock.send(req)
+ sock.send(req.encode())
try:
sock.recv(100)
except:
@@ -4546,7 +4560,7 @@ RGV2aWNlIEEQSQAGADcqAAEg
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM,
socket.IPPROTO_TCP)
sock.connect(addr)
- sock.send("NOTIFY / HTTP/1.1\r\n\r\n" + 4500*'a')
+ sock.send(b"NOTIFY / HTTP/1.1\r\n\r\n" + 4500 * b'a')
try:
sock.recv(100)
except:
@@ -4582,10 +4596,10 @@ RGV2aWNlIEEQSQAGADcqAAEg
raise Exception("Unexpected HTTP response: %d" % resp.status)
with alloc_fail(dev[0], 1, "xml_get_first_item"):
- send_wlanevent(url, uuid, '')
+ send_wlanevent(url, uuid, bytes())
with alloc_fail(dev[0], 1, "wpabuf_alloc_ext_data;xml_get_base64_item"):
- send_wlanevent(url, uuid, 'foo')
+ send_wlanevent(url, uuid, b'foo')
for func in [ "wps_init",
"wps_process_manufacturer",
@@ -4604,28 +4618,28 @@ RGV2aWNlIEEQSQAGADcqAAEg
send_wlanevent(url2, uuid, m1, no_response=True)
logger.info("EAP message: M1")
- data = '\x0202:11:22:00:00:00'
- data += '\x10\x22\x00\x01\x04'
- data += '\x10\x47\x00\x10' + 16*'\x00'
- data += '\x10\x20\x00\x06\x02\x00\x00\x00\x00\x00'
- data += '\x10\x1a\x00\x10' + 16*'\x00'
- data += '\x10\x32\x00\xc0' + 192*'\x00'
- data += '\x10\x04\x00\x02\x00\x00'
- data += '\x10\x10\x00\x02\x00\x00'
- data += '\x10\x0d\x00\x01\x00'
- data += '\x10\x08\x00\x02\x00\x00'
- data += '\x10\x44\x00\x01\x00'
- data += '\x10\x21\x00\x00'
- data += '\x10\x23\x00\x00'
- data += '\x10\x24\x00\x00'
- data += '\x10\x42\x00\x00'
- data += '\x10\x54\x00\x08' + 8*'\x00'
- data += '\x10\x11\x00\x00'
- data += '\x10\x3c\x00\x01\x00'
- data += '\x10\x02\x00\x02\x00\x00'
- data += '\x10\x12\x00\x02\x00\x00'
- data += '\x10\x09\x00\x02\x00\x00'
- data += '\x10\x2d\x00\x04\x00\x00\x00\x00'
+ data = b'\x0202:11:22:00:00:00'
+ data += b'\x10\x22\x00\x01\x04'
+ data += b'\x10\x47\x00\x10' + 16 * b'\x00'
+ data += b'\x10\x20\x00\x06\x02\x00\x00\x00\x00\x00'
+ data += b'\x10\x1a\x00\x10' + 16 * b'\x00'
+ data += b'\x10\x32\x00\xc0' + 192 * b'\x00'
+ data += b'\x10\x04\x00\x02\x00\x00'
+ data += b'\x10\x10\x00\x02\x00\x00'
+ data += b'\x10\x0d\x00\x01\x00'
+ data += b'\x10\x08\x00\x02\x00\x00'
+ data += b'\x10\x44\x00\x01\x00'
+ data += b'\x10\x21\x00\x00'
+ data += b'\x10\x23\x00\x00'
+ data += b'\x10\x24\x00\x00'
+ data += b'\x10\x42\x00\x00'
+ data += b'\x10\x54\x00\x08' + 8 * b'\x00'
+ data += b'\x10\x11\x00\x00'
+ data += b'\x10\x3c\x00\x01\x00'
+ data += b'\x10\x02\x00\x02\x00\x00'
+ data += b'\x10\x12\x00\x02\x00\x00'
+ data += b'\x10\x09\x00\x02\x00\x00'
+ data += b'\x10\x2d\x00\x04\x00\x00\x00\x00'
dev[0].dump_monitor()
with alloc_fail(dev[0], 1, "wps_er_add_sta_data"):
send_wlanevent(url, uuid, data)
@@ -4719,7 +4733,7 @@ def test_ap_wps_er_http_proto_subscribe_failing(dev, apdev):
'Content-Length: ' + str(len(payload)) + '\r\n' + \
'Timeout: Second-1801\r\n' + \
'Date: Sat, 15 Aug 2015 18:55:08 GMT\r\n\r\n'
- self.wfile.write(hdr + payload)
+ self.wfile.write((hdr + payload).encode())
run_wps_er_proto_test(dev[0], WPSAPHTTPServer_fail_subscribe)
def test_ap_wps_er_http_proto_subscribe_invalid_response(dev, apdev):
@@ -4734,7 +4748,7 @@ def test_ap_wps_er_http_proto_subscribe_invalid_response(dev, apdev):
'Content-Length: ' + str(len(payload)) + '\r\n' + \
'Timeout: Second-1801\r\n' + \
'Date: Sat, 15 Aug 2015 18:55:08 GMT\r\n\r\n'
- self.wfile.write(hdr + payload)
+ self.wfile.write((hdr + payload).encode())
run_wps_er_proto_test(dev[0], WPSAPHTTPServer_subscribe_invalid_response)
def test_ap_wps_er_http_proto_subscribe_invalid_response(dev, apdev):
@@ -4771,7 +4785,7 @@ def test_ap_wps_er_http_proto_upnp_info_no_device(dev, apdev):
'Connection: close\r\n' + \
'Content-Length: ' + str(len(payload)) + '\r\n' + \
'Date: Sat, 15 Aug 2015 18:55:08 GMT\r\n\r\n'
- self.wfile.write(hdr + payload)
+ self.wfile.write((hdr + payload).encode())
run_wps_er_proto_test(dev[0], WPSAPHTTPServer_no_device, no_event_url=True)
def test_ap_wps_er_http_proto_upnp_info_no_device_type(dev, apdev):
@@ -4794,7 +4808,7 @@ def test_ap_wps_er_http_proto_upnp_info_no_device_type(dev, apdev):
'Connection: close\r\n' + \
'Content-Length: ' + str(len(payload)) + '\r\n' + \
'Date: Sat, 15 Aug 2015 18:55:08 GMT\r\n\r\n'
- self.wfile.write(hdr + payload)
+ self.wfile.write((hdr + payload).encode())
run_wps_er_proto_test(dev[0], WPSAPHTTPServer_no_device, no_event_url=True)
def test_ap_wps_er_http_proto_upnp_info_invalid_udn_uuid(dev, apdev):
@@ -4838,7 +4852,7 @@ def _test_ap_wps_http_timeout(dev, apdev):
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM,
socket.IPPROTO_TCP)
sock.connect(addr)
- sock.send("G")
+ sock.send(b"G")
class DummyServer(StreamRequestHandler):
def handle(self):
@@ -4857,7 +4871,7 @@ def _test_ap_wps_http_timeout(dev, apdev):
logger.debug("Complete HTTP connection with hostapd (that should have already closed the connection)")
try:
sock.send("ET / HTTP/1.1\r\n\r\n")
- res = sock.recv(100)
+ sock.recv(100)
sock.close()
except:
pass
@@ -4876,14 +4890,15 @@ def _test_ap_wps_er_url_parse(dev, apdev):
sock.bind(("239.255.255.250", 1900))
dev[0].request("WPS_ER_START ifname=lo")
(msg,addr) = sock.recvfrom(1000)
+ msg = msg.decode()
logger.debug("Received SSDP message from %s: %s" % (str(addr), msg))
if "M-SEARCH" not in msg:
raise Exception("Not an M-SEARCH")
- sock.sendto("HTTP/1.1 200 OK\r\nST: urn:schemas-wifialliance-org:device:WFADevice:1\r\nlocation:http://127.0.0.1\r\ncache-control:max-age=1\r\n\r\n";, addr)
+ sock.sendto(b"HTTP/1.1 200 OK\r\nST: urn:schemas-wifialliance-org:device:WFADevice:1\r\nlocation:http://127.0.0.1\r\ncache-control:max-age=1\r\n\r\n";, addr)
ev = dev[0].wait_event(["WPS-ER-AP-REMOVE"], timeout=2)
- sock.sendto("HTTP/1.1 200 OK\r\nST: urn:schemas-wifialliance-org:device:WFADevice:1\r\nlocation:http://127.0.0.1/:foo\r\ncache-control:max-age=1\r\n\r\n";, addr)
+ sock.sendto(b"HTTP/1.1 200 OK\r\nST: urn:schemas-wifialliance-org:device:WFADevice:1\r\nlocation:http://127.0.0.1/:foo\r\ncache-control:max-age=1\r\n\r\n";, addr)
ev = dev[0].wait_event(["WPS-ER-AP-REMOVE"], timeout=2)
- sock.sendto("HTTP/1.1 200 OK\r\nST: urn:schemas-wifialliance-org:device:WFADevice:1\r\nlocation:http://255.255.255.255:0/foo.xml\r\ncache-control:max-age=1\r\n\r\n";, addr)
+ sock.sendto(b"HTTP/1.1 200 OK\r\nST: urn:schemas-wifialliance-org:device:WFADevice:1\r\nlocation:http://255.255.255.255:0/foo.xml\r\ncache-control:max-age=1\r\n\r\n";, addr)
ev = dev[0].wait_event(["WPS-ER-AP-REMOVE"], timeout=2)
sock.close()
@@ -4915,7 +4930,7 @@ def test_ap_wps_er_http_client(dev, apdev):
class WPSAPHTTPServer_req_as_resp(WPSAPHTTPServer):
def handle_upnp_info(self):
- self.wfile.write("GET / HTTP/1.1\r\n\r\n")
+ self.wfile.write(b"GET / HTTP/1.1\r\n\r\n")
run_wps_er_proto_test(dev[0], WPSAPHTTPServer_req_as_resp,
no_event_url=True)
@@ -5973,7 +5988,7 @@ def test_ap_wps_set_selected_registrar_proto(dev, apdev):
(200, "104a000110") ]
for status,test in tests:
tlvs = binascii.unhexlify(test)
- newmsg = base64.b64encode(tlvs)
+ newmsg = base64.b64encode(tlvs).decode()
msg = '<?xml version="1.0"?>\n'
msg += '<s:Envelope xmlns:s="http://schemas.xmlsoap.org/soap/envelope/"; s:encodingStyle="http://schemas.xmlsoap.org/soap/encoding/";>'
msg += '<s:Body>'
@@ -6174,13 +6189,16 @@ def get_wsc_msg(dev):
# Parse EAP expanded header
if len(data) < 1:
raise Exception("No EAP type included")
- msg['eap_type'], = struct.unpack('B', data[0])
+ if sys.version_info[0] > 2:
+ msg['eap_type'] = data[0]
+ else:
+ msg['eap_type'], = struct.unpack('B', data[0])
data = data[1:]
if msg['eap_type'] == 254:
if len(data) < 3 + 4:
raise Exception("Truncated EAP expanded header")
- msg['eap_vendor_id'], msg['eap_vendor_type'] = struct.unpack('>LL', '\0' + data[0:7])
+ msg['eap_vendor_id'], msg['eap_vendor_type'] = struct.unpack('>LL', b'\x00' + data[0:7])
data = data[7:]
else:
raise Exception("Unexpected EAP type")
@@ -6222,7 +6240,11 @@ def recv_wsc_msg(dev, opcode, msg_type):
return msg, msg['wsc_attrs'], msg['raw_attrs']
def build_wsc_attr(attr, payload):
- return struct.pack('>HH', attr, len(payload)) + payload
+ if type(payload) == bytes:
+ _payload = payload
+ else:
+ _payload = payload.encode()
+ return struct.pack('>HH', attr, len(payload)) + _payload
def build_attr_msg_type(msg_type):
return build_wsc_attr(ATTR_MSG_TYPE, struct.pack('B', msg_type))
@@ -6240,7 +6262,10 @@ def build_eap_wsc(eap_code, eap_id, payload, opcode=WSC_MSG):
# EAP-WSC header
msg += struct.pack('BB', opcode, 0)
# WSC attributes
- msg += payload
+ if type(payload) == bytes:
+ msg += payload
+ else:
+ msg += payload.encode()
return msg
def build_eap_success(eap_id):
@@ -6260,7 +6285,7 @@ def build_eap_failure(eap_id):
return msg
def send_wsc_msg(dev, src, msg):
- res = dev.request("EAPOL_RX " + src + " " + binascii.hexlify(msg))
+ res = dev.request("EAPOL_RX " + src + " " + binascii.hexlify(msg).decode())
if "OK" not in res:
raise Exception("EAPOL_RX failed")
@@ -6268,14 +6293,14 @@ group_5_prime = 0xFFFFFFFFFFFFFFFFC90FDAA22168C234C4C6628B80DC1CD129024E088A67CC
group_5_generator = 2
def wsc_kdf(key, label, bits):
- result = ''
+ result = bytes()
i = 1
while len(result) * 8 < bits:
- data = struct.pack('>L', i) + label + struct.pack('>L', bits)
+ data = struct.pack('>L', i) + label.encode() + struct.pack('>L', bits)
m = hmac.new(key, data, hashlib.sha256)
result += m.digest()
i += 1
- return result[0:bits / 8]
+ return result[0:bits // 8]
def wsc_keys(kdk):
keys = wsc_kdf(kdk, "Wi-Fi Easy and Secure Key Derivation", 640)
@@ -6285,12 +6310,12 @@ def wsc_keys(kdk):
return authkey,keywrapkey,emsk
def wsc_dev_pw_half_psk(authkey, dev_pw):
- m = hmac.new(authkey, dev_pw, hashlib.sha256)
+ m = hmac.new(authkey, dev_pw.encode(), hashlib.sha256)
return m.digest()[0:16]
def wsc_dev_pw_psk(authkey, dev_pw):
- dev_pw_1 = dev_pw[0:len(dev_pw) / 2]
- dev_pw_2 = dev_pw[len(dev_pw) / 2:]
+ dev_pw_1 = dev_pw[0:len(dev_pw) // 2]
+ dev_pw_2 = dev_pw[len(dev_pw) // 2:]
psk1 = wsc_dev_pw_half_psk(authkey, dev_pw_1)
psk2 = wsc_dev_pw_half_psk(authkey, dev_pw_2)
return psk1,psk2
@@ -6301,15 +6326,22 @@ def build_attr_authenticator(authkey, prev_msg, curr_msg):
return build_wsc_attr(ATTR_AUTHENTICATOR, auth)
def build_attr_encr_settings(authkey, keywrapkey, data):
- m = hmac.new(authkey, data, hashlib.sha256)
+ if type(data) == bytes:
+ _data = data
+ else:
+ if len(data) > 0:
+ _data = data.encode()
+ else:
+ _data = bytes()
+ m = hmac.new(authkey, _data, hashlib.sha256)
kwa = m.digest()[0:8]
- data += build_wsc_attr(ATTR_KEY_WRAP_AUTH, kwa)
- iv = 16*'\x99'
+ _data += build_wsc_attr(ATTR_KEY_WRAP_AUTH, kwa)
+ iv = 16 * b'\x99'
aes = AES.new(keywrapkey, AES.MODE_CBC, iv)
- pad_len = 16 - len(data) % 16
+ pad_len = 16 - len(_data) % 16
ps = pad_len * struct.pack('B', pad_len)
- data += ps
- wrapped = aes.encrypt(data)
+ _data += ps
+ wrapped = aes.encrypt(_data)
return build_wsc_attr(ATTR_ENCR_SETTINGS, iv + wrapped)
def decrypt_attr_encr_settings(authkey, keywrapkey, data):
@@ -6319,7 +6351,10 @@ def decrypt_attr_encr_settings(authkey, keywrapkey, data):
encr = data[16:]
aes = AES.new(keywrapkey, AES.MODE_CBC, iv)
decrypted = aes.decrypt(encr)
- pad_len, = struct.unpack('B', decrypted[-1])
+ if sys.version_info[0] > 2:
+ pad_len = decrypted[-1]
+ else:
+ pad_len, = struct.unpack('B', decrypted[-1])
if pad_len > len(decrypted):
raise Exception("Invalid padding in Encrypted Settings")
for i in range(-pad_len, -1):
@@ -6360,7 +6395,7 @@ def wsc_dh_kdf(peer_pk, own_private, mac_addr, e_nonce, r_nonce):
peer_public = int(binascii.hexlify(peer_pk), 16)
if peer_public < 2 or peer_public >= group_5_prime:
raise Exception("Invalid peer public key")
- if pow(peer_public, (group_5_prime - 1) / 2, group_5_prime) != 1:
+ if pow(peer_public, (group_5_prime - 1) // 2, group_5_prime) != 1:
raise Exception("Unexpected Legendre symbol for peer public key")
shared_secret = pow(peer_public, own_private, group_5_prime)
@@ -6368,33 +6403,41 @@ def wsc_dh_kdf(peer_pk, own_private, mac_addr, e_nonce, r_nonce):
logger.debug("DH shared secret: " + ss)
dhkey = hashlib.sha256(binascii.unhexlify(ss)).digest()
- logger.debug("DHKey: " + binascii.hexlify(dhkey))
+ logger.debug("DHKey: " + binascii.hexlify(dhkey).decode())
- m = hmac.new(dhkey, e_nonce + mac_addr + r_nonce, hashlib.sha256)
+ if type(e_nonce) == bytes:
+ _e_nonce = e_nonce
+ else:
+ _e_nonce = e_nonce.encode()
+ if type(r_nonce) == bytes:
+ _r_nonce = r_nonce
+ else:
+ _r_nonce = r_nonce.encode()
+ m = hmac.new(dhkey, _e_nonce + mac_addr + _r_nonce, hashlib.sha256)
kdk = m.digest()
- logger.debug("KDK: " + binascii.hexlify(kdk))
+ logger.debug("KDK: " + binascii.hexlify(kdk).decode())
authkey,keywrapkey,emsk = wsc_keys(kdk)
- logger.debug("AuthKey: " + binascii.hexlify(authkey))
- logger.debug("KeyWrapKey: " + binascii.hexlify(keywrapkey))
- logger.debug("EMSK: " + binascii.hexlify(emsk))
+ logger.debug("AuthKey: " + binascii.hexlify(authkey).decode())
+ logger.debug("KeyWrapKey: " + binascii.hexlify(keywrapkey).decode())
+ logger.debug("EMSK: " + binascii.hexlify(emsk).decode())
return authkey,keywrapkey
def wsc_dev_pw_hash(authkey, dev_pw, e_pk, r_pk):
psk1,psk2 = wsc_dev_pw_psk(authkey, dev_pw)
- logger.debug("PSK1: " + binascii.hexlify(psk1))
- logger.debug("PSK2: " + binascii.hexlify(psk2))
+ logger.debug("PSK1: " + binascii.hexlify(psk1).decode())
+ logger.debug("PSK2: " + binascii.hexlify(psk2).decode())
# Note: Secret values are supposed to be random, but hardcoded values are
# fine for testing.
- s1 = 16*'\x77'
+ s1 = 16 * b'\x77'
m = hmac.new(authkey, s1 + psk1 + e_pk + r_pk, hashlib.sha256)
hash1 = m.digest()
- logger.debug("Hash1: " + binascii.hexlify(hash1))
+ logger.debug("Hash1: " + binascii.hexlify(hash1).decode())
- s2 = 16*'\x88'
+ s2 = 16 * b'\x88'
m = hmac.new(authkey, s2 + psk2 + e_pk + r_pk, hashlib.sha256)
hash2 = m.digest()
- logger.debug("Hash2: " + binascii.hexlify(hash2))
+ logger.debug("Hash2: " + binascii.hexlify(hash2).decode())
return s1,s2,hash1,hash2
def build_m1(eap_id, uuid_e, mac_addr, e_nonce, e_pk,
@@ -6582,7 +6625,7 @@ def test_wps_ext(dev, apdev):
msg, m8_attrs, raw_m8_attrs = recv_wsc_msg(hapd, WSC_MSG, WPS_M8)
m8_cred = decrypt_attr_encr_settings(authkey, keywrapkey,
m8_attrs[ATTR_ENCR_SETTINGS])
- logger.debug("M8 Credential: " + binascii.hexlify(m8_cred))
+ logger.debug("M8 Credential: " + binascii.hexlify(m8_cred).decode())
logger.debug("Prepare WSC_Done")
attrs = build_wsc_attr(ATTR_VERSION, '\x10')
@@ -6748,7 +6791,7 @@ def test_wps_ext_kwa_proto_no_kwa(dev, apdev):
r_s1,keywrapkey,authkey,raw_m3_attrs,eap_id,bssid,attrs = wps_start_kwa(dev, apdev)
data = build_wsc_attr(ATTR_R_SNONCE1, r_s1)
# Encrypted Settings without KWA
- iv = 16*'\x99'
+ iv = 16 * b'\x99'
aes = AES.new(keywrapkey, AES.MODE_CBC, iv)
pad_len = 16 - len(data) % 16
ps = pad_len * struct.pack('B', pad_len)
@@ -6766,7 +6809,7 @@ def test_wps_ext_kwa_proto_data_after_kwa(dev, apdev):
kwa = m.digest()[0:8]
data += build_wsc_attr(ATTR_KEY_WRAP_AUTH, kwa)
data += build_wsc_attr(ATTR_VENDOR_EXT, "1234567890")
- iv = 16*'\x99'
+ iv = 16 * b'\x99'
aes = AES.new(keywrapkey, AES.MODE_CBC, iv)
pad_len = 16 - len(data) % 16
ps = pad_len * struct.pack('B', pad_len)
@@ -6781,7 +6824,7 @@ def test_wps_ext_kwa_proto_kwa_mismatch(dev, apdev):
data = build_wsc_attr(ATTR_R_SNONCE1, r_s1)
# Encrypted Settings and KWA with incorrect value
data += build_wsc_attr(ATTR_KEY_WRAP_AUTH, 8*'\x00')
- iv = 16*'\x99'
+ iv = 16 * b'\x99'
aes = AES.new(keywrapkey, AES.MODE_CBC, iv)
pad_len = 16 - len(data) % 16
ps = pad_len * struct.pack('B', pad_len)
@@ -6904,7 +6947,7 @@ def wps_run_cred_proto(dev, apdev, m8_cred, connect=False, no_connect=False):
def build_cred(nw_idx='\x01', ssid='test-wps-conf', auth_type='\x00\x20',
encr_type='\x00\x08', nw_key="12345678",
mac_addr='\x00\x00\x00\x00\x00\x00'):
- attrs = ''
+ attrs = bytes()
if nw_idx is not None:
attrs += build_wsc_attr(ATTR_NETWORK_INDEX, nw_idx)
if ssid is not None:
@@ -7038,8 +7081,8 @@ def test_wps_ext_proto_m2_invalid_public_key(dev, apdev):
wps_ext_eap_wsc(dev[0], hapd, bssid, "EAP-WSC/Start")
mac_addr = binascii.unhexlify(dev[0].own_addr().replace(':', ''))
- uuid_r = 16*'\x33'
- r_nonce = 16*'\x44'
+ uuid_r = 16 * b'\x33'
+ r_nonce = 16 * b'\x44'
own_private, e_pk = wsc_dh_init()
logger.debug("Receive M1 from STA")
@@ -7055,7 +7098,7 @@ def test_wps_ext_proto_m2_invalid_public_key(dev, apdev):
logger.debug("Send M2 to STA")
m2, raw_m2_attrs = build_m2(authkey, raw_m1_attrs, eap_id,
m1_attrs[ATTR_ENROLLEE_NONCE],
- r_nonce, uuid_r, 192*'\xff')
+ r_nonce, uuid_r, 192 * b'\xff')
send_wsc_msg(dev[0], bssid, m2)
eap_id = (eap_id + 1) % 256
@@ -7557,7 +7600,7 @@ def test_wps_ext_proto_m4_invalid_pad_string(dev, apdev):
m = hmac.new(authkey, data, hashlib.sha256)
kwa = m.digest()[0:8]
data += build_wsc_attr(ATTR_KEY_WRAP_AUTH, kwa)
- iv = 16*'\x99'
+ iv = 16 * b'\x99'
aes = AES.new(keywrapkey, AES.MODE_CBC, iv)
pad_len = 16 - len(data) % 16
ps = (pad_len - 1) * struct.pack('B', pad_len) + struct.pack('B', pad_len - 1)
@@ -7594,7 +7637,7 @@ def test_wps_ext_proto_m4_invalid_pad_value(dev, apdev):
m = hmac.new(authkey, data, hashlib.sha256)
kwa = m.digest()[0:8]
data += build_wsc_attr(ATTR_KEY_WRAP_AUTH, kwa)
- iv = 16*'\x99'
+ iv = 16 * b'\x99'
aes = AES.new(keywrapkey, AES.MODE_CBC, iv)
pad_len = 16 - len(data) % 16
ps = (pad_len - 1) * struct.pack('B', pad_len) + struct.pack('B', 255)
@@ -9259,7 +9302,7 @@ def test_wps_ext_m1_workaround(dev, apdev):
logger.debug("Send M1 to AP")
m1, raw_m1_attrs = build_m1(msg['eap_identifier'], uuid_e, mac_addr,
e_nonce, e_pk, manufacturer='Apple TEST',
- model_name='AirPort', config_methods='\xff\xff')
+ model_name='AirPort', config_methods=b'\xff\xff')
send_wsc_msg(hapd, addr, m1)
logger.debug("Receive M2 from AP")
--
2.17.1
_______________________________________________
Hostap mailing list
Hostap@xxxxxxxxxxxxxxxxxxx
http://lists.infradead.org/mailman/listinfo/hostap
