src/test/utils/EapolAAA.py

# 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 os
import sys
import binascii
import shutil
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
RADIUS_USER_MAC_START = (0x02 << 40) | (0x03 << 32) | (0x04 << 24) | 1
RADIUS_USER_MAC_END =  (0x02 << 40) | (0x03 << 32) | (0x04 << 24) | (0xff << 16) | ( 0xff << 8 ) | 0xff

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, start = 0, end = 0):
    """Generate radius server mac addresses"""
    """Scope to generate 256*256*256 mac addresses"""
    if start == 0 or end == 0:
        s = (0x00 << 40) | (0x02 << 32) | ( 0x03 << 24) | (1)
        e = (0x00 << 40) | (0x02 << 32) | ( 0x03 << 24) | (0xff << 16) | (0xff << 8) | (0xff)
    else:
        s = start
        e = end
    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)

def get_radius_networks(num):
    PORT_SUBNET_START = '12.0.0.0'
    PORT_SUBNET_MASK = '/24'
    PORT_SUBNET_END = '220.0.0.0'
    port_start_list = map(lambda ip: int(ip), PORT_SUBNET_START.split('.'))
    port_end_list = map(lambda ip: int(ip), PORT_SUBNET_END.split('.'))
    port_subnet_start = (port_start_list[0] << 24) | ( port_start_list[1] << 16 ) | ( port_start_list[2] << 8 ) | 0
    port_subnet_end = (port_end_list[0] << 24) | ( port_end_list[1] << 16 ) | ( port_end_list[2] << 8 ) | 0
    mask = int(PORT_SUBNET_MASK[1:])
    net_list = []
    for n in xrange(port_subnet_start, port_subnet_end, 256):
        subnet = ((n >> 24) & 0xff, (n >> 16) & 0xff, (n >> 8) & 0xff, 0, mask)
        prefix = subnet[:3]
        gw = prefix + (1,)
        subnet_s = '{}.{}.{}.{}/{}'.format(*subnet)
        prefix_s = '{}.{}.{}'.format(*prefix)
        gw_s = '{}.{}.{}.{}'.format(*gw)
        net_list.append((prefix_s, subnet_s, gw_s))
        if len(net_list) >= num:
            break

    return net_list

def get_radius_user_file():
    cur_dir = os.path.dirname(os.path.realpath(__file__))
    radius_authorize = 'setup/radius-config/freeradius/mods-config/files/authorize'
    radius_user_file = os.path.join(cur_dir, '..', *radius_authorize.split('/'))
    return radius_user_file

def radius_add_users(num):
    global RADIUS_USER_MAC_START, RADIUS_USER_MAC_END
    template = '''
%s Cleartext-Password := "radpass"
\tReply-Message := "Hello, %%{User-Name}"
'''
    radius_user_file = get_radius_user_file()
    if not os.access(radius_user_file, os.F_OK):
        return False
    mac_start = RADIUS_USER_MAC_START
    mac_end =   RADIUS_USER_MAC_END
    macs = get_radius_macs(num, start = mac_start, end = mac_end)
    save_file = '{}.save'.format(radius_user_file)
    new_file = '{}.new'.format(radius_user_file)
    shutil.copy(radius_user_file, save_file)
    with open(radius_user_file, 'r') as f:
        lines = f.readlines()
    for m in macs:
        lines.append(template %(m))
    with open(new_file, 'w') as f:
        f.writelines(lines)
    os.rename(new_file, radius_user_file)
    return True

def radius_restore_users():
    radius_user_file = get_radius_user_file()
    save_file = '{}.save'.format(radius_user_file)
    if not os.access(save_file, os.F_OK):
        return False
    os.rename(save_file, radius_user_file)
    return True