From 8493fe34579415da546cde637c95da1cb979eb03 Mon Sep 17 00:00:00 2001 From: Jouni Malinen Date: Fri, 21 Aug 2015 01:21:18 +0300 Subject: [PATCH] eapol_test: Add an example controller script for RADIUS testing This new script can be used as an example of load testing for RADIUS authentication servers. It can control multiple eapol_test processes over the control interface to run EAP-TLS authentication. As an example, the following commands would run eight instances of eapol_test concurrently: for i in 0 1 2 3 4 5 6 7; do ./eapol_test -T /tmp/eapol_test -i $i > eapol_test.log$i & done time ./eapol_test.py --ctrl /tmp/eapol_test --num 8 --iter 100 0: PASS 100 1: PASS 100 2: PASS 100 3: PASS 100 4: PASS 100 5: PASS 100 6: PASS 100 7: PASS 100 real 0m0.339s time ./eapol_test.py --ctrl /tmp/eapol_test --num 8 --iter 100 --no-fast-reauth 0: PASS 100 1: PASS 100 2: PASS 100 3: PASS 100 4: PASS 100 5: PASS 100 6: PASS 100 7: PASS 100 real 0m1.876s Signed-off-by: Jouni Malinen --- wpa_supplicant/eapol_test.py | 142 +++++++++++++++++++++++++++++++++++ 1 file changed, 142 insertions(+) create mode 100755 wpa_supplicant/eapol_test.py diff --git a/wpa_supplicant/eapol_test.py b/wpa_supplicant/eapol_test.py new file mode 100755 index 000000000..80e7dfcf5 --- /dev/null +++ b/wpa_supplicant/eapol_test.py @@ -0,0 +1,142 @@ +#!/usr/bin/env python2 +# +# eapol_test controller +# Copyright (c) 2015, Jouni Malinen +# +# 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): + 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() + 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') + args = parser.parse_args() + + num = int(args.num) + iter = int(args.iter) + if args.ctrl: + global wpas_ctrl + wpas_ctrl = args.ctrl + + 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])) + 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()