| #### Authentication parameters |
| from scapy.all import * |
| from socket import * |
| from struct import * |
| import sys |
| from nose.tools import assert_equal, assert_not_equal, assert_raises, assert_true |
| |
| 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 |
| |
| class EapolPacket(object): |
| |
| def __init__(self, intf = 'veth0'): |
| self.intf = intf |
| self.s = None |
| self.max_payload_size = 1600 |
| |
| def setup(self): |
| self.s = socket(AF_PACKET, SOCK_RAW, htons(ETHERTYPE_PAE)) |
| 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) |
| |
| 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 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 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:] |
| 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_start(self): |
| eap_payload = self.eap(EAPOL_START, 2) |
| return self.eapol_send(EAPOL_START, 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_payload_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_payload_size)[14:] |
| code, id, eaplen = unpack("!BBH", p[4:8]) |
| return code |
| |