Introduced eapol receive to use scapy with packet lambda filters.
Changed the tls/pap tests to use the eapol scapy receive functions.
diff --git a/src/test/pap/papTest.py b/src/test/pap/papTest.py
index 3ce25f9..5c9018a 100644
--- a/src/test/pap/papTest.py
+++ b/src/test/pap/papTest.py
@@ -1,8 +1,6 @@
import unittest
import os,sys
-CORD_TEST_UTILS = 'utils'
-test_root = os.getenv('CORD_TEST_ROOT') or './'
-sys.path.append(test_root + CORD_TEST_UTILS)
+from nose.tools import assert_equal
from EapPAP import PAPAuthTest
class eap_auth_exchange(unittest.TestCase):
diff --git a/src/test/utils/EapPAP.py b/src/test/utils/EapPAP.py
index 48b75a8..a128f05 100644
--- a/src/test/utils/EapPAP.py
+++ b/src/test/utils/EapPAP.py
@@ -4,11 +4,12 @@
import nosePAPAuthHolder as PAPAuthHolder
from socket import *
from struct import *
-import scapy
+from scapy.all import *
from nose.tools import *
from CordTestBase import CordTester
PAP_USER = "raduser"
PAP_PASSWD = "radpass"
+log.setLevel('INFO')
class PAPAuthTest(EapolPacket, CordTester):
@@ -49,37 +50,37 @@
self.nextEvent = self.PAPEventTable.EVT_EAP_ID_REQ
def _eapIdReq(self):
- print 'Inside EAP ID Req'
- p = self.eapol_recv()
- code, pkt_id, eaplen = unpack("!BBH", p[0:4])
- print "Code %d, id %d, len %d" %(code, pkt_id, eaplen)
- assert_equal(code, EAP_REQUEST)
- reqtype = unpack("!B", p[4:5])[0]
- reqdata = p[5:4+eaplen]
- assert_equal(reqtype, EAP_TYPE_ID)
- print "<====== Send EAP Response with identity = %s ================>" % PAP_USER
- self.eapol_id_req(pkt_id, PAP_USER)
+ log.info( 'Inside EAP ID Req' )
+ def eapol_cb(pkt):
+ log.info('Got EAPOL packet with type id and code request')
+ log.info('Packet code: %d, type: %d, id: %s', pkt[EAP].code, pkt[EAP].type, pkt[EAP].id)
+ log.info("<====== Send EAP Response with identity = %s ================>" % PAP_USER)
+ self.eapol_id_req(pkt[EAP].id, PAP_USER)
+
+ self.eapol_scapy_recv(cb = eapol_cb,
+ lfilter = lambda pkt: pkt[EAP].type == EAP.TYPE_ID and pkt[EAP].code == EAP.REQUEST)
self.nextEvent = self.PAPEventTable.EVT_EAP_PAP_USER_REQ
def _eapPAPUserReq(self):
- print 'UserReq Inside Challenge'
- p = self.eapol_recv()
- code, pkt_id, eaplen = unpack("!BBH", p[0:4])
- print "Code %d, id %d, len %d" %(code, pkt_id, eaplen)
- assert_equal(code, EAP_REQUEST)
- reqtype = unpack("!B", p[4:5])[0]
- reqdata = p[5:4+eaplen]
- assert_equal(reqtype, EAP_TYPE_TLS)
- print "<====== Send EAP Response with Password = %s ================>" % PAP_PASSWD
- self.eapol_id_req(pkt_id, PAP_PASSWD)
+ log.info('UserReq Inside Challenge')
+ def eapol_cb(pkt):
+ log.info('Got EAPOL packet with type id and code request')
+ log.info('Packet code: %d, id: %s', pkt[EAP].code, pkt[EAP].id)
+ log.info('Send EAP Response for id %s with Password = %s' %(pkt[EAP].id, PAP_PASSWD) )
+ self.eapol_id_req(pkt[EAP].id, PAP_PASSWD)
+
+ self.eapol_scapy_recv(cb = eapol_cb,
+ lfilter = lambda pkt: pkt[EAP].type == EAP_TYPE_TLS and pkt[EAP].code == EAP.REQUEST)
#self.nextEvent = self.PAPEventTable.EVT_EAP_PAP_PASSWD_REQ
- self.nextEvent = self.PAPEventTable.EVT_EAP_PAP_DONE
-
+ self.nextEvent = None
+
def _eapPAPPassReq(self):
- print 'PassReq Inside Challenge'
- p = self.eapol_recv()
- code, pkt_id, eaplen = unpack("!BBH", p[0:4])
- print "Code %d, id %d, len %d" %(code, pkt_id, eaplen)
- assert_equal(code, EAP_SUCCESS)
+ log.info('PassReq Inside Challenge')
+ def eapol_cb(pkt):
+ log.info('Got EAPOL packet with type id and code request')
+ log.info('Packet code: %d, type: %d', pkt[EAP].code, pkt[EAP].type)
+
+ self.eapol_scapy_recv(cb = eapol_cb,
+ lfilter = lambda pkt: pkt[EAP].code == EAP.SUCCESS)
self.nextEvent = self.PAPEventTable.EVT_EAP_PAP_DONE
diff --git a/src/test/utils/EapTLS.py b/src/test/utils/EapTLS.py
index dfd6f88..f3cb66b 100644
--- a/src/test/utils/EapTLS.py
+++ b/src/test/utils/EapTLS.py
@@ -9,7 +9,7 @@
from nose.tools import *
from CordTestBase import CordTester
import re
-log.setLevel('DEBUG')
+log.setLevel('INFO')
class TLSAuthTest(EapolPacket, CordTester):
tlsStateTable = Enumeration("TLSStateTable", ("ST_EAP_SETUP",
@@ -51,23 +51,22 @@
self.nextEvent = self.tlsEventTable.EVT_EAP_ID_REQ
def _eapIdReq(self):
- p = self.eapol_recv()
- code, pkt_id, eaplen = unpack("!BBH", p[0:4])
- assert_equal(code, EAP_REQUEST)
- reqtype = unpack("!B", p[4:5])[0]
- reqdata = p[5:4+eaplen]
- assert_equal(reqtype, EAP_TYPE_ID)
- log.debug("<====== Send EAP Response with identity = %s ================>" % USER)
- self.eapol_id_req(pkt_id, USER)
+ log.info( 'Inside EAP ID Req' )
+ def eapol_cb(pkt):
+ log.info('Got EAPOL packet with type id and code request')
+ log.info('Packet code: %d, type: %d, id: %d', pkt[EAP].code, pkt[EAP].type, pkt[EAP].id)
+ log.info("<====== Send EAP Response with identity = %s ================>" % USER)
+ self.eapol_id_req(pkt[EAP].id, USER)
+
+ self.eapol_scapy_recv(cb = eapol_cb,
+ lfilter = lambda pkt: pkt[EAP].type == EAP.TYPE_ID and pkt[EAP].code == EAP.REQUEST)
self.nextEvent = self.tlsEventTable.EVT_EAP_TLS_HELLO_REQ
def _eapTlsHelloReq(self):
- p = self.eapol_recv()
- code, pkt_id, eaplen = unpack("!BBH", p[0:4])
- assert_equal(code, EAP_REQUEST)
- reqtype = unpack("!B", p[4:5])[0]
- assert_equal(reqtype, EAP_TYPE_TLS)
- reqdata = TLSRecord(version="TLS_1_0")/TLSHandshake()/TLSClientHello(version="TLS_1_0",
+
+ def eapol_cb(pkt):
+ log.info('Got hello request for id %d', pkt[EAP].id)
+ reqdata = TLSRecord(version="TLS_1_0")/TLSHandshake()/TLSClientHello(version="TLS_1_0",
gmt_unix_time=1234,
random_bytes="A" * 28,
session_id='',
@@ -75,20 +74,21 @@
cipher_suites=[TLSCipherSuite.RSA_WITH_AES_128_CBC_SHA]
)
- #reqdata.show()
- log.debug("------> Sending Client Hello TLS payload of len %d ----------->" %len(reqdata))
- eap_payload = self.eapTLS(EAP_RESPONSE, pkt_id, TLS_LENGTH_INCLUDED, str(reqdata))
- self.eapol_send(EAPOL_EAPPACKET, eap_payload)
+ #reqdata.show()
+ log.debug("Sending Client Hello TLS payload of len %d, id %d" %(len(reqdata),pkt[EAP].id))
+ eap_payload = self.eapTLS(EAP_RESPONSE, pkt[EAP].id, TLS_LENGTH_INCLUDED, str(reqdata))
+ self.eapol_send(EAPOL_EAPPACKET, eap_payload)
+
+ self.eapol_scapy_recv(cb = eapol_cb,
+ lfilter = lambda pkt: pkt[EAP].type == EAP_TYPE_TLS and pkt[EAP].code == EAP.REQUEST)
self.nextEvent = self.tlsEventTable.EVT_EAP_TLS_CERT_REQ
def _eapTlsCertReq(self):
- p = self.eapol_recv()
- code, pkt_id, eaplen = unpack("!BBH", p[0:4])
- assert_equal(code, EAP_REQUEST)
- reqtype = unpack("!B", p[4:5])[0]
- assert_equal(reqtype, EAP_TYPE_TLS)
- rex_pem = re.compile(r'\-+BEGIN[^\-]+\-+(.*?)\-+END[^\-]+\-+', re.DOTALL)
- self.pem_cert = """-----BEGIN CERTIFICATE-----
+
+ def eapol_cb(pkt):
+ log.info('Got cert request')
+ rex_pem = re.compile(r'\-+BEGIN[^\-]+\-+(.*?)\-+END[^\-]+\-+', re.DOTALL)
+ self.pem_cert = """-----BEGIN CERTIFICATE-----
MIIDvTCCAqWgAwIBAgIBAjANBgkqhkiG9w0BAQUFADCBizELMAkGA1UEBhMCVVMx
CzAJBgNVBAgTAkNBMRIwEAYDVQQHEwlTb21ld2hlcmUxEzARBgNVBAoTCkNpZW5h
IEluYy4xHjAcBgkqhkiG9w0BCQEWD2FkbWluQGNpZW5hLmNvbTEmMCQGA1UEAxMd
@@ -111,32 +111,34 @@
T1tJBrgI7/WI+dqhKBFolKGKTDWIHsZXQvZ1snGu/FRYzg1l+R/jT8cRB9BDwhUt
yg==
-----END CERTIFICATE-----"""
- self.der_cert = rex_pem.findall(self.pem_cert)[0].decode("base64")
- reqdata = TLSRecord(version="TLS_1_0")/TLSHandshake()/TLSCertificateList(
- certificates=[TLSCertificate(data=x509.X509Cert(self.der_cert))])
- #reqdata.show()
- log.info("------> Sending Client Hello TLS Certificate payload of len %d ----------->" %len(reqdata))
- eap_payload = self.eapTLS(EAP_RESPONSE, pkt_id, TLS_LENGTH_INCLUDED, str(reqdata))
- self.eapol_send(EAPOL_EAPPACKET, eap_payload)
+ self.der_cert = rex_pem.findall(self.pem_cert)[0].decode("base64")
+ reqdata = TLSRecord(version="TLS_1_0")/TLSHandshake()/TLSCertificateList(
+ certificates=[TLSCertificate(data=x509.X509Cert(self.der_cert))])
+ #reqdata.show()
+ log.info("------> Sending Client Hello TLS Certificate payload of len %d ----------->" %len(reqdata))
+ eap_payload = self.eapTLS(EAP_RESPONSE, pkt[EAP].id, TLS_LENGTH_INCLUDED, str(reqdata))
+ self.eapol_send(EAPOL_EAPPACKET, eap_payload)
+
+ self.eapol_scapy_recv(cb = eapol_cb,
+ lfilter = lambda pkt: pkt[EAP].type == EAP_TYPE_TLS and pkt[EAP].code == EAP.REQUEST)
self.nextEvent = self.tlsEventTable.EVT_EAP_TLS_CHANGE_CIPHER_SPEC
def _eapTlsChangeCipherSpec(self):
- p = self.eapol_recv()
- code, pkt_id, eaplen = unpack("!BBH", p[0:4])
- assert_equal(code, EAP_REQUEST)
- reqtype = unpack("!B", p[4:5])[0]
- assert_equal(reqtype, EAP_TYPE_TLS)
- reqdata = TLSFinished(data="")
- eap_payload = self.eapTLS(EAP_RESPONSE, pkt_id, TLS_LENGTH_INCLUDED, str(reqdata))
- self.eapol_send(EAPOL_EAPPACKET, eap_payload)
+ def eapol_cb(pkt):
+ log.info('Got change cipher request')
+ reqdata = TLSFinished(data="")
+ eap_payload = self.eapTLS(EAP_RESPONSE, pkt[EAP].id, TLS_LENGTH_INCLUDED, str(reqdata))
+ self.eapol_send(EAPOL_EAPPACKET, eap_payload)
+
+ self.eapol_scapy_recv(cb = eapol_cb,
+ lfilter = lambda pkt: pkt[EAP].type == EAP_TYPE_TLS and pkt[EAP].code == EAP.REQUEST)
self.nextEvent = self.tlsEventTable.EVT_EAP_TLS_FINISHED
def _eapTlsFinished(self):
- p = self.eapol_recv()
- code, pkt_id, eaplen = unpack("!BBH", p[0:4])
- log.debug("Code %d, id %d, len %d" %(code, pkt_id, eaplen))
- assert_equal(code, EAP_REQUEST)
- reqtype = unpack("!B", p[4:5])[0]
- assert_equal(reqtype, EAP_TYPE_TLS)
+ def eapol_cb(pkt):
+ log.info('Got tls finished request')
+
+ self.eapol_scapy_recv(cb = eapol_cb,
+ lfilter = lambda pkt: pkt[EAP].type == EAP_TYPE_TLS and pkt[EAP].code == EAP.REQUEST)
#We stop here as certification validation success implies auth success
self.nextEvent = None
diff --git a/src/test/utils/EapolAAA.py b/src/test/utils/EapolAAA.py
index 6c477f0..31be259 100644
--- a/src/test/utils/EapolAAA.py
+++ b/src/test/utils/EapolAAA.py
@@ -43,6 +43,7 @@
self.s.bind((self.intf, ETHERTYPE_PAE))
self.mymac = self.s.getsockname()[4]
self.llheader = Ether(dst = PAE_GROUP_ADDR, src = self.mymac, type = ETHERTYPE_PAE)
+ self.recv_sock = L2Socket(iface = self.intf, type = ETHERTYPE_PAE)
def cleanup(self):
if self.s is not None:
@@ -76,6 +77,12 @@
assert_equal(pkt_type, EAPOL_EAPPACKET)
return p[4:]
+ def eapol_scapy_recv(self, cb = None, lfilter = None, count = 1):
+ def eapol_default_cb(pkt): pass
+ if cb is None:
+ cb = eapol_default_cb
+ sniff(prn = cb, lfilter = lfilter, count = count, opened_socket = self.recv_sock)
+
def eapol_start(self):
eap_payload = self.eap(EAPOL_START, 2)
return self.eapol_send(EAPOL_START, eap_payload)