Signed-off-by: Masashi Honma <masashi.honma@xxxxxxxxx>
---
tests/hwsim/test_ap_wps.py | 614 ++++++++++++++++++++-----------------
1 file changed, 325 insertions(+), 289 deletions(-)
diff --git a/tests/hwsim/test_ap_wps.py b/tests/hwsim/test_ap_wps.py
index b41a53be9..466c1eeaa 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
@@ -21,15 +22,14 @@ import socket
import struct
try:
from http.client import HTTPConnection
+ from urllib.request import urlopen
from urllib.parse import urlparse, urljoin
from urllib.error import HTTPError
- from io import StringIO
from socketserver import StreamRequestHandler, TCPServer
except ImportError:
from httplib import HTTPConnection
+ from urllib import urlopen
from urlparse import urlparse, urljoin
- from urllib2 import build_opener, ProxyHandler, HTTPError
- from StringIO import StringIO
from SocketServer import StreamRequestHandler, TCPServer
import urllib
import xml.etree.ElementTree as ET
@@ -175,7 +175,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("12345678".encode()).decode()):
raise Exception("WPS_CONFIG command failed")
ev = hapd.wait_event(["WPS-NEW-AP-SETTINGS"], timeout=5)
if ev is None:
@@ -198,7 +198,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("12345678".encode()).decode()):
raise Exception("WPS_CONFIG command failed")
ev = hapd.wait_event(["WPS-NEW-AP-SETTINGS"], timeout=5)
if ev is None:
@@ -212,7 +212,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("1234567".encode()).decode()):
raise Exception("Invalid WPS_CONFIG command accepted")
def test_ap_wps_conf(dev, apdev):
@@ -610,13 +610,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, "test".encode())
+ 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"
@@ -769,7 +769,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)
@@ -1570,8 +1570,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")
@@ -2359,13 +2359,13 @@ def test_ap_wps_pbc_timeout(dev, apdev, params):
location = ssdp_get_location(ap_uuid)
urls = upnp_get_urls(location)
- eventurl = urllib.parse.urlparse(urls['event_sub_url'])
- ctrlurl = urllib.parse.urlparse(urls['control_url'])
+ eventurl = urlparse(urls['event_sub_url'])
+ ctrlurl = urlparse(urls['control_url'])
- url = urllib.parse.urlparse(location)
- conn = http.client.HTTPConnection(url.netloc)
+ url = urlparse(location)
+ conn = HTTPConnection(url.netloc)
- class WPSERHTTPServer(socketserver.StreamRequestHandler):
+ class WPSERHTTPServer(StreamRequestHandler):
def handle(self):
data = self.rfile.readline().strip()
logger.debug(data)
@@ -2484,7 +2484,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)
@@ -2555,7 +2555,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([
@@ -2565,7 +2565,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([
@@ -2575,7 +2575,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([
@@ -2584,7 +2584,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([
@@ -2594,7 +2594,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',
@@ -2602,7 +2602,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([
@@ -2611,7 +2611,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([
@@ -2620,7 +2620,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([
@@ -2630,7 +2630,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',
@@ -2638,7 +2638,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',
@@ -2646,7 +2646,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([
@@ -2656,7 +2656,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([
@@ -2666,7 +2666,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',
@@ -2674,10 +2674,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("NOTIFY * HTTP/1.1".encode(), ("239.255.255.250", 1900))
msg = '\r'.join([
'M-SEARCH * HTTP/1.1',
'HOST: 239.255.255.250:1900',
@@ -2685,7 +2685,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)
@@ -2701,7 +2701,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)
@@ -2727,11 +2727,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
@@ -2745,10 +2745,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:
@@ -2758,6 +2758,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
@@ -2766,15 +2767,15 @@ def ssdp_get_location(uuid):
return location
def upnp_get_urls(location):
- conn = urllib.urlopen(location, proxies={})
+ conn = urlopen(location)
tree = ET.parse(conn)
root = tree.getroot()
urn = '{urn:schemas-upnp-org:device-1-0}'
service = root.find("./" + urn + "device/" + urn + "serviceList/" + urn + "service")
res = {}
- res['scpd_url'] = urllib.parse.urljoin(location, service.find(urn + 'SCPDURL').text)
- res['control_url'] = urllib.parse.urljoin(location, service.find(urn + 'controlURL').text)
- res['event_sub_url'] = urllib.parse.urljoin(location, service.find(urn + 'eventSubURL').text)
+ res['scpd_url'] = urljoin(location, service.find(urn + 'SCPDURL').text)
+ res['control_url'] = urljoin(location, service.find(urn + 'controlURL').text)
+ res['event_sub_url'] = urljoin(location, service.find(urn + 'eventSubURL').text)
return res
def upnp_soap_action(conn, path, action, include_soap_action=True,
@@ -2791,23 +2792,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 = io.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):
@@ -2818,16 +2815,23 @@ def test_ap_wps_upnp(dev, apdev):
location = ssdp_get_location(ap_uuid)
urls = upnp_get_urls(location)
- conn = urllib.urlopen(urls['scpd_url'], proxies={})
+ conn = urlopen(urls['scpd_url'])
scpd = conn.read()
- conn = urllib.urlopen(urllib.parse.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 = urllib.parse.urlparse(location)
- conn = http.client.HTTPConnection(url.netloc)
+ url = urlparse(location)
+ conn = HTTPConnection(url.netloc)
#conn.set_debuglevel(1)
headers = { "Content-type": 'text/xml; charset="utf-8"',
"SOAPAction": '"urn:schemas-wifialliance-org:service:WFAWLANConfig:1#GetDeviceInfo"' }
@@ -2843,7 +2847,7 @@ def test_ap_wps_upnp(dev, apdev):
headers = { "Content-type": 'text/xml; charset="utf-8"',
"SOAPAction": '"urn:some-unknown-action#GetDeviceInfo"' }
- ctrlurl = urllib.parse.urlparse(urls['control_url'])
+ ctrlurl = urlparse(urls['control_url'])
conn.request("POST", ctrlurl.path, "\r\n\r\n", headers)
resp = conn.getresponse()
if resp.status != 401:
@@ -2869,7 +2873,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")
@@ -2900,10 +2904,10 @@ def test_ap_wps_upnp_subscribe(dev, apdev):
location = ssdp_get_location(ap_uuid)
urls = upnp_get_urls(location)
- eventurl = urllib.parse.urlparse(urls['event_sub_url'])
+ eventurl = urlparse(urls['event_sub_url'])
- url = urllib.parse.urlparse(location)
- conn = http.client.HTTPConnection(url.netloc)
+ url = urlparse(location)
+ conn = HTTPConnection(url.netloc)
#conn.set_debuglevel(1)
headers = { "callback": '<http://127.0.0.1:12345/event>',
"timeout": "Second-1234" }
@@ -3249,9 +3253,9 @@ def test_ap_wps_upnp_subscribe_events(dev, apdev):
location = ssdp_get_location(ap_uuid)
urls = upnp_get_urls(location)
- eventurl = urllib.parse.urlparse(urls['event_sub_url'])
+ eventurl = urlparse(urls['event_sub_url'])
- class WPSERHTTPServer(socketserver.StreamRequestHandler):
+ class WPSERHTTPServer(StreamRequestHandler):
def handle(self):
data = self.rfile.readline().strip()
logger.debug(data)
@@ -3260,8 +3264,8 @@ def test_ap_wps_upnp_subscribe_events(dev, apdev):
server = MyTCPServer(("127.0.0.1", 12345), WPSERHTTPServer)
server.timeout = 1
- url = urllib.parse.urlparse(location)
- conn = http.client.HTTPConnection(url.netloc)
+ url = urlparse(location)
+ conn = HTTPConnection(url.netloc)
headers = { "callback": '<http://127.0.0.1:12345/event>',
"NT": "upnp:event",
@@ -3316,8 +3320,8 @@ def test_ap_wps_upnp_http_proto(dev, apdev):
location = ssdp_get_location(ap_uuid)
- url = urllib.parse.urlparse(location)
- conn = http.client.HTTPConnection(url.netloc, timeout=0.2)
+ url = urlparse(location)
+ conn = HTTPConnection(url.netloc, timeout=0.2)
#conn.set_debuglevel(1)
conn.request("HEAD", "hello")
@@ -3415,8 +3419,8 @@ def test_ap_wps_upnp_http_proto_chunked(dev, apdev):
location = ssdp_get_location(ap_uuid)
- url = urllib.parse.urlparse(location)
- conn = http.client.HTTPConnection(url.netloc)
+ url = urlparse(location)
+ conn = HTTPConnection(url.netloc)
#conn.set_debuglevel(1)
headers = { "Transfer-Encoding": 'chunked' }
@@ -3431,10 +3435,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("a\r\nabcdefghij\r\n".encode())
time.sleep(0.1)
- conn.send("2\r\nkl\r\n")
- conn.send("0\r\n\r\n")
+ conn.send("2\r\nkl\r\n".encode())
+ conn.send("0\r\n\r\n".encode())
resp = conn.getresponse()
if resp.status != 404:
raise Exception("Unexpected HTTP response: %d" % resp.status)
@@ -3446,8 +3450,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("1\r\nZ\r\n".encode())
+ conn.send("0\r\n\r\n".encode())
resp = conn.getresponse()
completed = True
except Exception as e:
@@ -3902,7 +3906,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:
@@ -3928,7 +3932,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:
@@ -3954,49 +3958,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("FOO".encode(), 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("FOO".encode(), addr)
+ sock.sendto("HTTP/1.1 200 OK\r\nFOO\r\n\r\n".encode(), addr)
+ sock.sendto("HTTP/1.1 200 OK\r\nNTS:foo\r\n\r\n".encode(), addr)
+ sock.sendto("HTTP/1.1 200 OK\r\nNTS:ssdp:byebye\r\n\r\n".encode(), addr)
+ sock.sendto("HTTP/1.1 200 OK\r\ncache-control: foo=1\r\n\r\n".encode(), addr)
+ sock.sendto("HTTP/1.1 200 OK\r\ncache-control: max-age=1\r\n\r\n".encode(), addr)
+ sock.sendto("HTTP/1.1 200 OK\r\nusn:\r\n\r\n".encode(), addr)
+ sock.sendto("HTTP/1.1 200 OK\r\nusn:foo\r\n\r\n".encode(), addr)
+ sock.sendto("HTTP/1.1 200 OK\r\nusn: uuid:\r\n\r\n".encode(), addr)
+ sock.sendto("HTTP/1.1 200 OK\r\nusn: uuid: \r\n\r\n".encode(), addr)
+ sock.sendto("HTTP/1.1 200 OK\r\nusn: uuid: foo\r\n\r\n".encode(), addr)
+ sock.sendto("HTTP/1.1 200 OK\r\nST: urn:schemas-wifialliance-org:device:WFADevice:1\r\n\r\n".encode(), 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".encode(), addr)
+ sock.sendto("HTTP/1.1 200 OK\r\nST: urn:schemas-wifialliance-org:device:WFADevice:1\r\nlocation:foo\r\n\r\n".encode(), 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("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".encode(), 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("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".encode(), 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("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".encode(), 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("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".encode(), 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("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".encode(), 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")
@@ -4044,7 +4050,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"?>
@@ -4072,7 +4078,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 = ""
@@ -4085,14 +4091,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)
@@ -4124,7 +4130,7 @@ class WPSAPHTTPServer(StreamRequestHandler):
class MyTCPServer(TCPServer):
def __init__(self, addr, handler):
self.allow_reuse_address = True
- socketserver.TCPServer.__init__(self, addr, handler)
+ TCPServer.__init__(self, addr, handler)
def wps_er_start(dev, http_server, max_age=1, wait_m_search=False,
location_url=None):
@@ -4135,6 +4141,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
@@ -4147,7 +4154,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
@@ -4192,13 +4199,13 @@ def run_wps_er_proto_test(dev, handler, no_event_url=False, location_url=None):
dev.request("WPS_ER_STOP")
def send_wlanevent(url, uuid, data, no_response=False):
- conn = http.client.HTTPConnection(url.netloc)
+ conn = HTTPConnection(url.netloc)
payload = '''<?xml version="1.0" encoding="utf-8"?>
<e:propertyset xmlns:e="urn:schemas-upnp-org:event-1-0">
<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",
@@ -4247,8 +4254,8 @@ def _test_ap_wps_er_http_proto(dev, apdev):
sock.close()
logger.info("Valid Probe Request notification")
- url = urllib.parse.urlparse(wps_event_url)
- conn = http.client.HTTPConnection(url.netloc)
+ url = urlparse(wps_event_url)
+ conn = HTTPConnection(url.netloc)
payload = '''<?xml version="1.0" encoding="utf-8"?>
<e:propertyset xmlns:e="urn:schemas-upnp-org:event-1-0">
<e:property><STAStatus>1</STAStatus></e:property>
@@ -4278,32 +4285,32 @@ RGV2aWNlIEEQSQAGADcqAAEg
raise Exception("No Enrollee UUID match")
logger.info("Incorrect event URL AP id")
- conn = http.client.HTTPConnection(url.netloc)
+ conn = HTTPConnection(url.netloc)
conn.request("NOTIFY", url.path + '123', payload, headers)
resp = conn.getresponse()
if resp.status != 404:
raise Exception("Unexpected HTTP response: %d" % resp.status)
logger.info("Missing AP id")
- conn = http.client.HTTPConnection(url.netloc)
+ conn = HTTPConnection(url.netloc)
conn.request("NOTIFY", '/event/' + url.path.split('/')[2],
payload, headers)
time.sleep(0.1)
logger.info("Incorrect event URL event id")
- conn = http.client.HTTPConnection(url.netloc)
+ conn = HTTPConnection(url.netloc)
conn.request("NOTIFY", '/event/123456789/123', payload, headers)
time.sleep(0.1)
logger.info("Incorrect event URL prefix")
- conn = http.client.HTTPConnection(url.netloc)
+ conn = HTTPConnection(url.netloc)
conn.request("NOTIFY", '/foobar/123456789/123', payload, headers)
resp = conn.getresponse()
if resp.status != 404:
raise Exception("Unexpected HTTP response: %d" % resp.status)
logger.info("Unsupported request")
- conn = http.client.HTTPConnection(url.netloc)
+ conn = HTTPConnection(url.netloc)
conn.request("FOOBAR", '/foobar/123456789/123', payload, headers)
resp = conn.getresponse()
if resp.status != 501:
@@ -4311,169 +4318,169 @@ RGV2aWNlIEEQSQAGADcqAAEg
logger.info("Unsupported request and OOM")
with alloc_fail(dev[0], 1, "wps_er_http_req"):
- conn = http.client.HTTPConnection(url.netloc)
+ conn = HTTPConnection(url.netloc)
conn.request("FOOBAR", '/foobar/123456789/123', payload, headers)
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")
@@ -4485,11 +4492,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("GET / HTTP/1.1\r\n\r\n".encode())
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:
@@ -4510,7 +4517,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("GET / HTTP/1.1\r\n\r\n".encode())
try:
sock.recv(100)
except:
@@ -4533,7 +4540,7 @@ RGV2aWNlIEEQSQAGADcqAAEg
socket.IPPROTO_TCP)
sock.settimeout(0.1)
sock.connect(addr)
- sock.send(req)
+ sock.send(req.encode())
try:
sock.recv(100)
except:
@@ -4544,14 +4551,14 @@ 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(("NOTIFY / HTTP/1.1\r\n\r\n" + 4500 * 'a').encode())
try:
sock.recv(100)
except:
pass
sock.close()
- conn = http.client.HTTPConnection(url.netloc)
+ conn = HTTPConnection(url.netloc)
payload = '<foo'
headers = { "Content-type": 'text/xml; charset="utf-8"',
"Server": "Unspecified, UPnP/1.0, Unspecified",
@@ -4565,7 +4572,7 @@ RGV2aWNlIEEQSQAGADcqAAEg
if resp.status != 200:
raise Exception("Unexpected HTTP response: %d" % resp.status)
- conn = http.client.HTTPConnection(url.netloc)
+ conn = HTTPConnection(url.netloc)
payload = '<WLANEvent foo></WLANEvent>'
headers = { "Content-type": 'text/xml; charset="utf-8"',
"Server": "Unspecified, UPnP/1.0, Unspecified",
@@ -4580,10 +4587,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, 'foo'.encode())
for func in [ "wps_init",
"wps_process_manufacturer",
@@ -4598,32 +4605,32 @@ RGV2aWNlIEEQSQAGADcqAAEg
send_wlanevent(url, uuid, m1, no_response=True)
with alloc_fail(dev[0], 1, "wps_er_http_resp_not_found"):
- url2 = urllib.parse.urlparse(wps_event_url.replace('/event/', '/notfound/'))
+ url2 = urlparse(wps_event_url.replace('/event/', '/notfound/'))
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)
@@ -4717,7 +4724,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):
@@ -4732,7 +4739,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):
@@ -4769,7 +4776,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):
@@ -4792,7 +4799,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):
@@ -4830,15 +4837,15 @@ def _test_ap_wps_http_timeout(dev, apdev):
add_ssdp_ap(apdev[0], ap_uuid)
location = ssdp_get_location(ap_uuid)
- url = urllib.parse.urlparse(location)
+ url = urlparse(location)
addr = (url.hostname, url.port)
logger.debug("Open HTTP connection to hostapd, but do not complete request")
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM,
socket.IPPROTO_TCP)
sock.connect(addr)
- sock.send("G")
+ sock.send("G".encode())
- class DummyServer(socketserver.StreamRequestHandler):
+ class DummyServer(StreamRequestHandler):
def handle(self):
logger.debug("DummyServer - start 31 sec wait")
time.sleep(31)
@@ -4855,7 +4862,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
@@ -4874,14 +4881,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("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".encode(), 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("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".encode(), 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("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".encode(), addr)
ev = dev[0].wait_event(["WPS-ER-AP-REMOVE"], timeout=2)
sock.close()
@@ -4913,7 +4921,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("GET / HTTP/1.1\r\n\r\n".encode())
run_wps_er_proto_test(dev[0], WPSAPHTTPServer_req_as_resp,
no_event_url=True)
@@ -5935,12 +5943,12 @@ def test_ap_wps_set_selected_registrar_proto(dev, apdev):
location = ssdp_get_location(ap_uuid)
urls = upnp_get_urls(location)
- eventurl = urllib.parse.urlparse(urls['event_sub_url'])
- ctrlurl = urllib.parse.urlparse(urls['control_url'])
- url = urllib.parse.urlparse(location)
- conn = http.client.HTTPConnection(url.netloc)
+ eventurl = urlparse(urls['event_sub_url'])
+ ctrlurl = urlparse(urls['control_url'])
+ url = urlparse(location)
+ conn = HTTPConnection(url.netloc)
- class WPSERHTTPServer(socketserver.StreamRequestHandler):
+ class WPSERHTTPServer(StreamRequestHandler):
def handle(self):
data = self.rfile.readline().strip()
logger.debug(data)
@@ -5971,7 +5979,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>'
@@ -6172,13 +6180,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")
@@ -6220,7 +6231,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))
@@ -6238,7 +6253,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):
@@ -6258,7 +6276,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")
@@ -6266,14 +6284,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)
@@ -6283,12 +6301,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
@@ -6299,15 +6317,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):
@@ -6317,7 +6342,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):
@@ -6358,7 +6386,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)
@@ -6366,33 +6394,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,
@@ -6580,7 +6616,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')
@@ -6746,7 +6782,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)
@@ -6764,7 +6800,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)
@@ -6779,7 +6815,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)
@@ -6902,7 +6938,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:
@@ -7036,8 +7072,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")
@@ -7053,7 +7089,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
@@ -7555,7 +7591,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)
@@ -7592,7 +7628,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)
@@ -9257,7 +9293,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")
@@ -9277,12 +9313,12 @@ def test_ap_wps_upnp_web_oom(dev, apdev, params):
hapd = add_ssdp_ap(apdev[0], ap_uuid)
location = ssdp_get_location(ap_uuid)
- url = urllib.parse.urlparse(location)
+ url = urlparse(location)
urls = upnp_get_urls(location)
- eventurl = urllib.parse.urlparse(urls['event_sub_url'])
- ctrlurl = urllib.parse.urlparse(urls['control_url'])
+ eventurl = urlparse(urls['event_sub_url'])
+ ctrlurl = urlparse(urls['control_url'])
- conn = http.client.HTTPConnection(url.netloc)
+ conn = HTTPConnection(url.netloc)
with alloc_fail(hapd, 1, "web_connection_parse_get"):
conn.request("GET", "/wps_device.xml")
try:
@@ -9290,7 +9326,7 @@ def test_ap_wps_upnp_web_oom(dev, apdev, params):
except:
pass
- conn = http.client.HTTPConnection(url.netloc)
+ conn = HTTPConnection(url.netloc)
conn.request("GET", "/unknown")
resp = conn.getresponse()
if resp.status != 404:
@@ -9304,56 +9340,56 @@ def test_ap_wps_upnp_web_oom(dev, apdev, params):
except:
pass
- conn = http.client.HTTPConnection(url.netloc)
+ conn = HTTPConnection(url.netloc)
conn.request("GET", "/wps_device.xml")
resp = conn.getresponse()
if resp.status != 200:
raise Exception("GET /wps_device.xml failed")
- conn = http.client.HTTPConnection(url.netloc)
+ conn = HTTPConnection(url.netloc)
resp = upnp_soap_action(conn, ctrlurl.path, "GetDeviceInfo")
if resp.status != 200:
raise Exception("GetDeviceInfo failed")
with alloc_fail(hapd, 1, "web_process_get_device_info"):
- conn = http.client.HTTPConnection(url.netloc)
+ conn = HTTPConnection(url.netloc)
resp = upnp_soap_action(conn, ctrlurl.path, "GetDeviceInfo")
if resp.status != 500:
raise Exception("Internal error not reported from GetDeviceInfo OOM")
with alloc_fail(hapd, 1, "wps_build_m1;web_process_get_device_info"):
- conn = http.client.HTTPConnection(url.netloc)
+ conn = HTTPConnection(url.netloc)
resp = upnp_soap_action(conn, ctrlurl.path, "GetDeviceInfo")
if resp.status != 500:
raise Exception("Internal error not reported from GetDeviceInfo OOM")
with alloc_fail(hapd, 1, "wpabuf_alloc;web_connection_send_reply"):
- conn = http.client.HTTPConnection(url.netloc)
+ conn = HTTPConnection(url.netloc)
try:
resp = upnp_soap_action(conn, ctrlurl.path, "GetDeviceInfo")
except:
pass
- conn = http.client.HTTPConnection(url.netloc)
+ conn = HTTPConnection(url.netloc)
resp = upnp_soap_action(conn, ctrlurl.path, "GetDeviceInfo")
if resp.status != 200:
raise Exception("GetDeviceInfo failed")
# No NewWLANEventType in PutWLANResponse NewMessage
- conn = http.client.HTTPConnection(url.netloc)
+ conn = HTTPConnection(url.netloc)
resp = upnp_soap_action(conn, ctrlurl.path, "PutWLANResponse", newmsg="foo")
if resp.status != 600:
raise Exception("Unexpected HTTP response: %d" % resp.status)
# No NewWLANEventMAC in PutWLANResponse NewMessage
- conn = http.client.HTTPConnection(url.netloc)
+ conn = HTTPConnection(url.netloc)
resp = upnp_soap_action(conn, ctrlurl.path, "PutWLANResponse",
newmsg="foo", neweventtype="1")
if resp.status != 600:
raise Exception("Unexpected HTTP response: %d" % resp.status)
# Invalid NewWLANEventMAC in PutWLANResponse NewMessage
- conn = http.client.HTTPConnection(url.netloc)
+ conn = HTTPConnection(url.netloc)
resp = upnp_soap_action(conn, ctrlurl.path, "PutWLANResponse",
newmsg="foo", neweventtype="1",
neweventmac="foo")
@@ -9362,7 +9398,7 @@ def test_ap_wps_upnp_web_oom(dev, apdev, params):
# Workaround for NewWLANEventMAC in PutWLANResponse NewMessage
# Ignored unexpected PutWLANResponse WLANEventType 1
- conn = http.client.HTTPConnection(url.netloc)
+ conn = HTTPConnection(url.netloc)
resp = upnp_soap_action(conn, ctrlurl.path, "PutWLANResponse",
newmsg="foo", neweventtype="1",
neweventmac="00.11.22.33.44.55")
@@ -9370,7 +9406,7 @@ def test_ap_wps_upnp_web_oom(dev, apdev, params):
raise Exception("Unexpected HTTP response: %d" % resp.status)
# PutWLANResponse NewMessage with invalid EAP message
- conn = http.client.HTTPConnection(url.netloc)
+ conn = HTTPConnection(url.netloc)
resp = upnp_soap_action(conn, ctrlurl.path, "PutWLANResponse",
newmsg="foo", neweventtype="2",
neweventmac="00:11:22:33:44:55")
@@ -9378,7 +9414,7 @@ def test_ap_wps_upnp_web_oom(dev, apdev, params):
raise Exception("Unexpected HTTP response: %d" % resp.status)
with alloc_fail(hapd, 1, "web_connection_parse_subscribe"):
- conn = http.client.HTTPConnection(url.netloc)
+ conn = HTTPConnection(url.netloc)
headers = { "callback": '<http://127.0.0.1:12345/event>',
"NT": "upnp:event",
"timeout": "Second-1234" }
@@ -9389,7 +9425,7 @@ def test_ap_wps_upnp_web_oom(dev, apdev, params):
pass
with alloc_fail(hapd, 1, "dup_binstr;web_connection_parse_subscribe"):
- conn = http.client.HTTPConnection(url.netloc)
+ conn = HTTPConnection(url.netloc)
headers = { "callback": '<http://127.0.0.1:12345/event>',
"NT": "upnp:event",
"timeout": "Second-1234" }
@@ -9399,7 +9435,7 @@ def test_ap_wps_upnp_web_oom(dev, apdev, params):
raise Exception("Unexpected HTTP response: %d" % resp.status)
with alloc_fail(hapd, 1, "wpabuf_alloc;web_connection_parse_unsubscribe"):
- conn = http.client.HTTPConnection(url.netloc)
+ conn = HTTPConnection(url.netloc)
headers = { "callback": '<http://127.0.0.1:12345/event>',
"NT": "upnp:event",
"timeout": "Second-1234" }
@@ -9410,7 +9446,7 @@ def test_ap_wps_upnp_web_oom(dev, apdev, params):
pass
with alloc_fail(hapd, 1, "web_connection_unimplemented"):
- conn = http.client.HTTPConnection(url.netloc)
+ conn = HTTPConnection(url.netloc)
conn.request("HEAD", "/wps_device.xml")
try:
resp = conn.getresponse()
--
2.17.1
_______________________________________________
Hostap mailing list
Hostap@xxxxxxxxxxxxxxxxxxx
http://lists.infradead.org/mailman/listinfo/hostap
