Python3 class for RADIUS authentication

p3-radius.py

# Python3 class for RADIUS authentication
# Based loosely on code found at: http://github.com/btimby/py-radius/

# Copyright (c) 2015, Justin W. Smith <[email protected]> 
# Copyright (c) 1999, Stuart Bishop <[email protected]> 
# All rights reserved.
# 
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are
# met:
# 
#     Redistributions of source code must retain the above copyright
#     notice, this list of conditions and the following disclaimer.
# 
#     Redistributions in binary form must reproduce the above copyright
#     notice, this list of conditions and the following disclaimer in the
#     documentation and/or other materials provided with the
#     distribution.
# 
#     The names of Stuart Bishop and Justin Smith may not be used to endorse 
#     or promote products derived from this software without specific prior
#     written permission.
# 
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
# ``AS IS'' AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A
# PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE REGENTS OR
# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
# EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
# PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
# PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
# LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
# NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
# SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

import os
from select import select
from struct import pack
from hashlib import md5
import socket
import logging


class RadiusAuthenticator:

    ACCESS_ACCEPT = 2
    ACCESS_CHALLENGE = 11

    radius_codes = {
        1: 'Access-Request',
        2: 'Access-Accept',
        3: 'Access-Reject',
        4: 'Accounting-Request',
        5: 'Accounting-Response',
        11: 'Access-Challenge',
        12: 'Status-Server',
        13: 'Status-Client'
    }

    radius_ids = {
        1: 'User-Name',
        2: 'User-Password',
        3: 'CHAP-Password',
        4: 'NAS-IP-Address',
        5: 'NAS-Port',
        6: 'Service-Type',
        7: 'Framed-Protocol',
        8: 'Framed-IP-Address',
        9: 'Framed-IP-Netmask',
        10: 'Framed-Routing',
        11: 'Filter-Id',
        12: 'Framed-MTU',
        13: 'Framed-Compression',
        14: 'Login-IP-Host',
        15: 'Login-Service',
        16: 'Login-TCP-Port',
        18: 'Reply-Message',
        19: 'Callback-Number',
        20: 'Callback-Id',
        22: 'Framed-Route',
        23: 'Framed-IPX-Network',
        24: 'State',
        25: 'Class',
        26: 'Vendor-Specific',
        27: 'Session-Timeout',
        28: 'Idle-Timeout',
        29: 'Termination-Action',
        30: 'Called-Station-Id',
        31: 'Calling-Station-Id',
        32: 'NAS-Identifier',
        33: 'Proxy-State',
        34: 'Login-LAT-Service',
        35: 'Login-LAT-Node',
        36: 'Login-LAT-Group',
        37: 'Framed-AppleTalk-Link',
        38: 'Framed-AppleTalk-Network',
        39: 'Framed-AppleTalk-Zone',
        40: 'Acct-Status-Type',
        41: 'Acct-Delay-Time',
        42: 'Acct-Input-Octets',
        43: 'Acct-Output-Octets',
        44: 'Acct-Session-Id',
        45: 'Acct-Authentic',
        46: 'Acct-Session-Time',
        47: 'Acct-Input-Packets',
        48: 'Acct-Output-Packets',
        49: 'Acct-Terminate-Cause',
        50: 'Acct-Multi-Session-Id',
        51: 'Acct-Link-Count',
        60: 'CHAP-Challenge',
        61: 'NAS-Port-Type',
        62: 'Port-Limit',
        63: 'Login-LAT-Port'
    }

    def __init__(self, host, port=1645, secret=None, retries=3, timeout=5):
        if not isinstance(secret, bytes):
            raise Exception("secret must be encoded.")

        self._secret = secret
        self._host = host
        self._port = port
        self._socket = None

        self.retries = retries
        self.timeout = timeout

    def __del__(self):
        self.closesocket()

    def opensocket(self):
        if self._socket is None:
            self._socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            self._socket.connect((self._host, self._port))

    def closesocket(self):
        if not hasattr(self, '_socket'):
            return
        if self._socket is not None:
            try:
                self._socket.close()
            finally:
                self._socket = None

    def generate_authenticator(self):
        """A 16 byte random string"""
        return os.urandom(16)

    def radius_hash(self, authenticator, text):
        """Encrypt a password with the secret"""
        if not isinstance(authenticator, bytes) or not isinstance(text, bytes):
            raise Exception("authenticator and text must be encoded.")

        # Pad the password to next multiple of 16 octets.
        text += (b'\x00' * (16 - (len(text) % 16)))
        if len(text) > 128:
            raise Exception('Password exceeds maximum of 128 bytes')
        result = b''
        last = authenticator
        while text:
            # First iteration uses an md5 of secret plus authenticator
            # Subsequent iterations use the md5 of secret plus previous result
            md5_hash = md5(self._secret + last).digest()
            for i in range(16):
                result += pack('B', md5_hash[i] ^ text[i])
            last, text = result[-16:], text[16:]
        return result

    def create_auth_payload(self, uname, passwd, identifier, authenticator):
        """ Creates payload for RADIUS authentication UDP packet """
        encpass = self.radius_hash(authenticator, passwd)

        return pack('!BBH16sBB%dsBB%ds' % (len(uname), len(encpass)),
                    1,  # B = Code
                    identifier,  # B = Identifier
                    len(uname) + len(encpass) + 24,  # H  =  Length of entire message
                    authenticator,  # 16s
                    1,  # B
                    len(uname) + 2,  # B
                    uname,  # %ds
                    2,  # B
                    len(encpass) + 2,  # B
                    encpass  # %ds
                    )

    def authenticate(self, uname, passwd, callback=None):
        """
           Attempt t authenticate with the given username and password.
           Returns False on failure
           Returns True on success
           Raises an exception if no responses or no valid responses are received.
        """

        if not callback:
            def default_callBack(resp_code):
                if resp_code == RadiusAuthenticator.ACCESS_ACCEPT:
                    logging.info("RADIUS User: '%s' successfully authenticated." % uname.decode())
                    return True
                else:
                    logging.info("RADIUS User: '%s' failed authentication. Response code: %d='%s'" %
                                 (uname.decode(), resp_code,
                                  RadiusAuthenticator.radius_codes.get(resp_code, 'UNKNOWN')))
                    return False
            callback = default_callBack

        if not callable(callback):
            raise Exception("Callback function must be callable! %s" % repr(callback))

        if not isinstance(uname, bytes) or not isinstance(passwd, bytes):
            raise Exception("Username and password must be encoded.")

        identifier = os.urandom(1)[0]
        authenticator = self.generate_authenticator()
        msg = self.create_auth_payload(uname, passwd, identifier, authenticator)

        try:
            self.opensocket()

            for i in range(0, self.retries):
                self._socket.send(msg)

                t = select([self._socket, ], [], [], self.timeout)
                if len(t[0]) > 0:
                    response = self._socket.recv(4096)
                    resp_code = response[0]
                    resp_ident = response[1]
                    if resp_ident == identifier:
                        checkauth = response[0:4] + authenticator + response[20:] + self._secret
                        if md5(checkauth).digest() == response[4:20]:
                            return callback(resp_code)
                        else:
                            logging.debug("Mismatched RADIUS response authenticator.")
                    else:
                        logging.debug("Mismatched RADIUS response identifier %d != %d" % (resp_ident, identifier))
                else:
                    logging.debug("Empty RADIUS response.")

        finally:
            self.closesocket()

        raise Exception("No response")