blob: 8219e27e0301d7ffaa32aac57b534ecf04dd3476 [file] [log] [blame]
# Copyright 2017-present Open Networking Foundation
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
# Copyright 2016-present Ciena Corporation
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
#### Authentication parameters
from scapy.all import *
from scapy_ssl_tls.ssl_tls import *
from socket import *
from struct import *
import sys
import binascii
from nose.tools import assert_equal, assert_not_equal, assert_raises, assert_true
from CordTestUtils import log_test
USER = "raduser"
PASS = "radpass"
WRONG_USER = "XXXX"
WRONG_PASS = "XXXX"
NO_USER = ""
NO_PASS = ""
DEV = "tap0"
ETHERTYPE_PAE = 0x888e
PAE_GROUP_ADDR = "\xff\xff\xff\xff\xff\xff"
EAPOL_VERSION = 1
EAPOL_EAPPACKET = 0
EAPOL_START = 1
EAPOL_LOGOFF = 2
EAPOL_KEY = 3
EAPOL_ASF = 4
EAP_REQUEST = 1
EAP_RESPONSE = 2
EAP_SUCCESS = 3
EAP_FAILURE = 4
EAP_TYPE_ID = 1
EAP_TYPE_MD5 = 4
EAP_TYPE_MSCHAP = 26
EAP_TYPE_TLS = 13
cCertMsg = '\x0b\x00\x00\x03\x00\x00\x00'
TLS_LENGTH_INCLUDED = 0x80
TLS_MORE_FRAGMENTS = 0x40
class EapolPacket(object):
src_mac_map = { 'bcast': 'ff:ff:ff:ff:ff:ff',
'mcast': '01:80:C2:00:00:03',
'zeros': '00:00:00:00:00:00',
'default': None
}
def __init__(self, intf = 'veth0'):
self.intf = intf
self.s = None
self.max_recv_size = 1600
def setup(self, src_mac = 'default'):
self.s = socket(AF_PACKET, SOCK_RAW, htons(ETHERTYPE_PAE))
self.s.bind((self.intf, ETHERTYPE_PAE))
self.mymac = self.s.getsockname()[4]
mac = None
mac_str = None
if src_mac == 'random':
mac = RandMAC()._fix()
elif src_mac in self.src_mac_map:
mac = self.src_mac_map[src_mac]
if mac is None:
mac = self.mymac
mac_str = binascii.hexlify(mac)
if mac_str is None:
mac_str = mac
self.llheader = Ether(dst = PAE_GROUP_ADDR, src = mac, type = ETHERTYPE_PAE)
log_test.info('llheader packet is %s'%self.llheader.show())
log_test.info('source mac of packet is %s'%mac_str)
self.recv_sock = L2Socket(iface = self.intf, type = ETHERTYPE_PAE)
def cleanup(self):
if self.s is not None:
self.s.close()
self.s = None
def eapol(self, req_type, payload=""):
return EAPOL(version = EAPOL_VERSION, type = req_type)/payload
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]:
return pack("!BBH", code, pkt_id, 4)
else:
if flags & TLS_LENGTH_INCLUDED:
flags_dlen = pack("!BL", flags, len(data))
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 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_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)
return p[4:]
def eapol_scapy_recv(self, cb = None, lfilter = None, count = 1, timeout = 10):
def eapol_default_cb(pkt): pass
if cb is None:
cb = eapol_default_cb
return sniff(prn = cb, lfilter = lfilter, count = count, timeout = timeout, opened_socket = self.recv_sock)
def eapol_start(self):
eap_payload = self.eap(EAPOL_START, 2)
return self.eapol_send(EAPOL_START, eap_payload)
def eapol_logoff(self):
eap_payload = self.eap(EAPOL_LOGOFF, 2)
return self.eapol_send(EAPOL_LOGOFF, eap_payload)
def eapol_id_req(self, pkt_id = 0, user = USER):
eap_payload = self.eap(EAP_RESPONSE, pkt_id, EAP_TYPE_ID, user)
return self.eapol_send(EAPOL_EAPPACKET, eap_payload)
def eap_md5_challenge_recv(self,rad_pwd):
PASS = rad_pwd
print 'Inside EAP MD5 Challenge Exchange'
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])
print "EAP Code %d, id %d, len %d" %(code, pkt_id, eaplen)
assert_equal(code, EAP_REQUEST)
reqtype = unpack("!B", p[8:9])[0]
reqdata = p[9:4+eaplen]
print 'Request type is %d' %(reqtype)
assert_equal(reqtype, EAP_TYPE_MD5)
challenge=pack("!B",pkt_id)+PASS+reqdata[1:]
print "Generating md5 challenge for %s" % challenge
return (challenge,pkt_id)
def eap_Status(self):
print 'Inside EAP Status'
p = self.s.recv(self.max_recv_size)[14:]
code, id, eaplen = unpack("!BBH", p[4:8])
return code
@classmethod
def eap_invalid_tls_packets_info(self, invalid_field_name = None, invalid_field_value = None):
log_test.info( 'Changing invalid field values in tls auth packets' )
if invalid_field_name == 'eapolTlsVersion':
global EAPOL_VERSION
log_test.info( 'Changing invalid field values in tls auth packets====== version changing' )
EAPOL_VERSION = invalid_field_value
if invalid_field_name == 'eapolTlsType':
global EAP_TYPE_TLS
log_test.info( 'Changing invalid field values in tls auth packets====== EAP TYPE TLS changing' )
EAP_TYPE_TLS = invalid_field_value
if invalid_field_name == 'eapolTypeID':
global EAP_TYPE_ID
log_test.info( 'Changing invalid field values in tls auth packets====== EAP TYPE TLS changing' )
EAP_TYPE_ID = invalid_field_value
if invalid_field_name == 'eapolResponse':
global EAP_RESPONSE
log_test.info( 'Changing invalid field values in tls auth packets====== EAP TYPE TLS changing' )
EAP_RESPONSE = invalid_field_value
@classmethod
def eap_tls_packets_field_value_replace(self, invalid_field_name = None):
log_test.info( 'Changing invalid field values in tls auth packets' )
if invalid_field_name == 'eapolTlsVersion':
global EAPOL_VERSION
EAPOL_VERSION = 1
log_test.info( 'Changing invalid field values in tls auth packets====== version changing' )
if invalid_field_name == 'eapolTlsType':
global EAP_TYPE_TLS
EAP_TYPE_TLS = 13
log_test.info( 'Changing invalid field values in tls auth packets====== version changing' )
if invalid_field_name == 'eapolTypeID':
global EAP_TYPE_ID
EAP_TYPE_ID = 1
log_test.info( 'Changing invalid field values in tls auth packets====== version changing' )
if invalid_field_name == 'eapolResponse':
global EAP_RESPONSE
EAP_RESPONSE = 2
log_test.info( 'Changing invalid field values in tls auth packets====== version changing' )
def get_radius_macs(num):
"""Generate radius server mac addresses"""
"""Scope to generate 256*256*256 mac addresses"""
s = (0x00 << 40) | (0x02 << 32) | ( 0x03 << 24) | (1)
e = (0x00 << 40) | (0x02 << 32) | ( 0x03 << 24) | (0xff << 16) | (0xff << 8) | (0xff)
n_macs = []
for v in xrange(s, e):
mask = (v & 0xff0000) == 0xff0000 or \
(v & 0x00ff00) == 0x00ff00 or \
(v & 0x0000ff) == 0x0000ff
if mask:
continue
n_macs.append(v)
if len(n_macs) == num:
break
def n_to_mac(n):
n_tuple = ( (n >> 40) & 0xff,
(n >> 32) & 0xff,
(n >> 24) & 0xff,
(n >> 16) & 0xff,
(n >> 8) & 0xff,
n & 0xff,
)
return '%02x:%02x:%02x:%02x:%02x:%02x' %(n_tuple)
#convert the number to macs
return map(n_to_mac, n_macs)