Test: Implement TLS fragment support in EAP TLS authentication.
Now that EAP tls fragmentation is supported, change tls authentication tests to the standard 2048 bits authentication mode.
Change-Id: I683c9bc441d5b965415fe4c97fbf84bf4e0a05e6
diff --git a/src/test/utils/EapolAAA.py b/src/test/utils/EapolAAA.py
index 0a2f8bd..c51f111 100644
--- a/src/test/utils/EapolAAA.py
+++ b/src/test/utils/EapolAAA.py
@@ -15,6 +15,7 @@
#
#### Authentication parameters
from scapy.all import *
+from scapy_ssl_tls.ssl_tls import *
from socket import *
from struct import *
import sys
@@ -45,13 +46,14 @@
EAP_TYPE_TLS = 13
cCertMsg = '\x0b\x00\x00\x03\x00\x00\x00'
TLS_LENGTH_INCLUDED = 0x80
+TLS_MORE_FRAGMENTS = 0x40
class EapolPacket(object):
def __init__(self, intf = 'veth0'):
self.intf = intf
self.s = None
- self.max_payload_size = 1600
+ self.max_recv_size = 1600
def setup(self):
self.s = socket(AF_PACKET, SOCK_RAW, htons(ETHERTYPE_PAE))
@@ -71,6 +73,64 @@
def eap(self, code, pkt_id, req_type=0, data=""):
return EAP(code = code, id = pkt_id, type = req_type)/data
+ def eapFragmentSend(self, code, pkt_id, flags = TLS_LENGTH_INCLUDED, payload = "", fragsize = 1024):
+ req_type = EAP_TYPE_TLS
+ if code in [ EAP_SUCCESS, EAP_FAILURE ]:
+ data = pack("!BBH", code, pkt_id, 4)
+ self.eapol_send(EAPOL_EAPPACKET, data)
+ return True
+
+ if len(payload) <= fragsize:
+ if flags & TLS_LENGTH_INCLUDED:
+ flags_dlen = pack("!BL", flags, len(payload))
+ data = pack("!BBHB", code, pkt_id, 5 + len(flags_dlen) + len(payload), req_type) \
+ + flags_dlen + payload
+ self.eapol_send(EAPOL_EAPPACKET, data)
+ return True
+ flags_str = pack("!B", flags)
+ data = pack("!BBHB", code, pkt_id, 5+len(flags_str)+len(payload), req_type) + flags_str + payload
+ self.eapol_send(EAPOL_EAPPACKET, data)
+ return True
+
+ fragments = []
+ data = payload[:]
+ frag = 0
+ def eapol_frag_cb(pkt):
+ r = str(pkt)
+ tls_data = r[self.TLS_OFFSET:]
+ frag_data = fragments[frag]
+ ##change packet id in response to match request
+ eap_payload = frag_data[:1] + pack("!B", pkt[EAP].id) + frag_data[2:]
+ self.eapol_send(EAPOL_EAPPACKET, eap_payload)
+
+ while len(data) > 0:
+ data_frag = data[:fragsize]
+ data = data[fragsize:]
+ if frag == 0:
+ ##first frag, include the total length
+ flags_dlen = pack("!BL", TLS_LENGTH_INCLUDED | TLS_MORE_FRAGMENTS, len(payload))
+ fragments.append(pack("!BBHB", code, pkt_id, 5 + len(flags_dlen) + len(data_frag), req_type) \
+ + flags_dlen + data_frag)
+ else:
+ if len(data) > 0:
+ flags = TLS_MORE_FRAGMENTS
+ else:
+ flags = 0
+ flags_str = pack("!B", flags)
+ fragments.append(pack("!BBHB", code, pkt_id, 5+len(flags_str)+len(data_frag), req_type) + \
+ flags_str + data_frag)
+ frag += 1
+
+ frag = 0
+ self.eapol_send(EAPOL_EAPPACKET, fragments[frag])
+ for frag in range(len(fragments)-1):
+ frag += 1
+ r = self.eapol_scapy_recv(cb = eapol_frag_cb,
+ lfilter = lambda pkt: EAP in pkt and pkt[EAP].type == EAP_TYPE_TLS and \
+ pkt[EAP].code == EAP.REQUEST)
+
+ return True
+
def eapTLS(self, code, pkt_id, flags = TLS_LENGTH_INCLUDED, data=""):
req_type = EAP_TYPE_TLS
if code in [EAP_SUCCESS, EAP_FAILURE]:
@@ -82,11 +142,28 @@
flags_str = pack("!B", flags)
return pack("!BBHB", code, pkt_id, 5+len(flags_str)+len(data), req_type) + flags_str + data
+ def eapTLSFragment(self, code, pkt_id, frag, data="", data_len = 0):
+ req_type = EAP_TYPE_TLS
+ if frag == 0:
+ flags = TLS_LENGTH_INCLUDED | TLS_MORE_FRAGMENTS
+ elif frag > 0:
+ flags = TLS_MORE_FRAGMENTS
+ else:
+ #last fragment
+ flags = 0
+ if data_len == 0:
+ data_len = len(data)
+ if flags & TLS_LENGTH_INCLUDED:
+ flags_dlen = pack("!BL", flags, data_len)
+ return pack("!BBHB", code, pkt_id, 5+len(flags_dlen)+len(data), req_type) + flags_dlen + data
+ flags_str = pack("!B", flags)
+ return pack("!BBHB", code, pkt_id, 5+len(flags_str)+len(data), req_type) + flags_str + data
+
def eapol_send(self, eapol_type, eap_payload):
return sendp(self.llheader/self.eapol(eapol_type, eap_payload), iface=self.intf)
def eapol_recv(self):
- p = self.s.recv(self.max_payload_size)[14:]
+ p = self.s.recv(self.max_recv_size)[14:]
vers,pkt_type,eapollen = unpack("!BBH",p[:4])
print "Version %d, type %d, len %d" %(vers, pkt_type, eapollen)
assert_equal(pkt_type, EAPOL_EAPPACKET)
@@ -113,7 +190,7 @@
def eap_md5_challenge_recv(self,rad_pwd):
PASS = rad_pwd
print 'Inside EAP MD5 Challenge Exchange'
- p = self.s.recv(self.max_payload_size)[14:]
+ p = self.s.recv(self.max_recv_size)[14:]
vers,pkt_type,eapollen = unpack("!BBH",p[:4])
print "EAPOL Version %d, type %d, len %d" %(vers, pkt_type, eapollen)
code, pkt_id, eaplen = unpack("!BBH", p[4:8])
@@ -129,7 +206,7 @@
def eap_Status(self):
print 'Inside EAP Status'
- p = self.s.recv(self.max_payload_size)[14:]
+ p = self.s.recv(self.max_recv_size)[14:]
code, id, eaplen = unpack("!BBH", p[4:8])
return code