zte's code,first commit
Change-Id: I9a04da59e459a9bc0d67f101f700d9d7dc8d681b
diff --git a/ap/app/wpa_supplicant-2.10/wpa_supplicant/eapol_test.py b/ap/app/wpa_supplicant-2.10/wpa_supplicant/eapol_test.py
new file mode 100755
index 0000000..88c83f3
--- /dev/null
+++ b/ap/app/wpa_supplicant-2.10/wpa_supplicant/eapol_test.py
@@ -0,0 +1,159 @@
+#!/usr/bin/env python2
+#
+# eapol_test controller
+# Copyright (c) 2015, Jouni Malinen <j@w1.fi>
+#
+# This software may be distributed under the terms of the BSD license.
+# See README for more details.
+
+import argparse
+import logging
+import os
+import Queue
+import sys
+import threading
+
+logger = logging.getLogger()
+dir = os.path.dirname(os.path.realpath(sys.modules[__name__].__file__))
+sys.path.append(os.path.join(dir, '..', 'wpaspy'))
+import wpaspy
+wpas_ctrl = '/tmp/eapol_test'
+
+class eapol_test:
+ def __init__(self, ifname):
+ self.ifname = ifname
+ self.ctrl = wpaspy.Ctrl(os.path.join(wpas_ctrl, ifname))
+ if "PONG" not in self.ctrl.request("PING"):
+ raise Exception("Failed to connect to eapol_test (%s)" % ifname)
+ self.mon = wpaspy.Ctrl(os.path.join(wpas_ctrl, ifname))
+ self.mon.attach()
+
+ def add_network(self):
+ id = self.request("ADD_NETWORK")
+ if "FAIL" in id:
+ raise Exception("ADD_NETWORK failed")
+ return int(id)
+
+ def remove_network(self, id):
+ id = self.request("REMOVE_NETWORK " + str(id))
+ if "FAIL" in id:
+ raise Exception("REMOVE_NETWORK failed")
+ return None
+
+ def set_network(self, id, field, value):
+ res = self.request("SET_NETWORK " + str(id) + " " + field + " " + value)
+ if "FAIL" in res:
+ raise Exception("SET_NETWORK failed")
+ return None
+
+ def set_network_quoted(self, id, field, value):
+ res = self.request("SET_NETWORK " + str(id) + " " + field + ' "' + value + '"')
+ if "FAIL" in res:
+ raise Exception("SET_NETWORK failed")
+ return None
+
+ def request(self, cmd, timeout=10):
+ return self.ctrl.request(cmd, timeout=timeout)
+
+ def wait_event(self, events, timeout=10):
+ start = os.times()[4]
+ while True:
+ while self.mon.pending():
+ ev = self.mon.recv()
+ logger.debug(self.ifname + ": " + ev)
+ for event in events:
+ if event in ev:
+ return ev
+ now = os.times()[4]
+ remaining = start + timeout - now
+ if remaining <= 0:
+ break
+ if not self.mon.pending(timeout=remaining):
+ break
+ return None
+
+def run(ifname, count, no_fast_reauth, res, conf):
+ et = eapol_test(ifname)
+
+ et.request("AP_SCAN 0")
+ if no_fast_reauth:
+ et.request("SET fast_reauth 0")
+ else:
+ et.request("SET fast_reauth 1")
+ id = et.add_network()
+
+ if len(conf):
+ for item in conf:
+ et.set_network(id, item, conf[item])
+ else:
+ et.set_network(id, "key_mgmt", "IEEE8021X")
+ et.set_network(id, "eapol_flags", "0")
+ et.set_network(id, "eap", "TLS")
+ et.set_network_quoted(id, "identity", "user")
+ et.set_network_quoted(id, "ca_cert", 'ca.pem')
+ et.set_network_quoted(id, "client_cert", 'client.pem')
+ et.set_network_quoted(id, "private_key", 'client.key')
+ et.set_network_quoted(id, "private_key_passwd", 'whatever')
+
+ et.set_network(id, "disabled", "0")
+
+ fail = False
+ for i in range(count):
+ et.request("REASSOCIATE")
+ ev = et.wait_event(["CTRL-EVENT-CONNECTED", "CTRL-EVENT-EAP-FAILURE"])
+ if ev is None or "CTRL-EVENT-CONNECTED" not in ev:
+ fail = True
+ break
+
+ et.remove_network(id)
+
+ if fail:
+ res.put("FAIL (%d OK)" % i)
+ else:
+ res.put("PASS %d" % (i + 1))
+
+def main():
+ parser = argparse.ArgumentParser(description='eapol_test controller')
+ parser.add_argument('--ctrl', help='control interface directory')
+ parser.add_argument('--num', help='number of processes')
+ parser.add_argument('--iter', help='number of iterations')
+ parser.add_argument('--no-fast-reauth', action='store_true',
+ dest='no_fast_reauth',
+ help='disable TLS session resumption')
+ parser.add_argument('--conf', help='file of network conf items')
+ args = parser.parse_args()
+
+ num = int(args.num)
+ iter = int(args.iter)
+ if args.ctrl:
+ global wpas_ctrl
+ wpas_ctrl = args.ctrl
+
+ conf = {}
+ if args.conf:
+ f = open(args.conf, "r")
+ for line in f:
+ confitem = line.split("=")
+ if len(confitem) == 2:
+ conf[confitem[0].strip()] = confitem[1].strip()
+ f.close()
+
+ t = {}
+ res = {}
+ for i in range(num):
+ res[i] = Queue.Queue()
+ t[i] = threading.Thread(target=run, args=(str(i), iter,
+ args.no_fast_reauth, res[i],
+ conf))
+ for i in range(num):
+ t[i].start()
+ for i in range(num):
+ t[i].join()
+ try:
+ results = res[i].get(False)
+ except:
+ results = "N/A"
+ print("%d: %s" % (i, results))
+
+if __name__ == "__main__":
+ main()