rad_authcheck.py #1

#!/usr/bin/env python
#==============================================================================
# Copyright and license info is available in the LICENSE file included with
# the Server Deployment Package (SDP), and also available online:
# https://swarm.workshop.perforce.com/projects/perforce-software-sdp/view/main/LICENSE
#------------------------------------------------------------------------------
#
# This trigger will check to see if the userid is in a the LOCL_PASSWD_FILE first and
# authenticate with that if found. If the users isn't in the local file, it checks to 
# see if the user is a service user, and if so, it will authenticate against LDAP.
# Finally, it will check the user against the Radius server if the other two conditions
# don't match.
#
# You need to install the python-pyrad package and python-six package for it to work.
# It also needs the file named dictionary in the same directory as the script.
#
# Set the Radius servers in radsvrs below
# Set the shared secret
# Pass in the user name as a parameter and the password from stdin.
#
# The trigger table entry is:
# authtrigger auth-check auth "/p4/common/bin/triggers/rad_authcheck.py %user% %serverport%"

import os
import re
import ldap
import sys
from pyrad.client import Client
from pyrad.dictionary import Dictionary
import pyrad.packet

# Configuration values
TIMEOUT = 20
AD_HOST = "ldap://ad.company.com"
DOMAIN = "YOURDOMAIN\\"
LOCAL_PASSWD_FILE = "/p4/common/bin/triggers/local.passwd"
SVC_USER_FILE = "/p4/common/bin/triggers/serviceusers.txt"
P4 = "/p4/common/bin/p4"
radsvrs = ["server1", "server2", "server3"]
sharedsecret = b"your_shared_secret"
user = sys.argv[1]

# localUserExists
#      checks to see if the user exists in the local password file. returns True or False
def localUserExists(username):
    exists = False
    if LOCAL_PASSWD_FILE is not None and os.path.isfile(LOCAL_PASSWD_FILE):
        f = open(LOCAL_PASSWD_FILE)
        for line in f:
            if line.startswith("%s:" % username):
                exists = True
                break
        f.close()
    return exists

# getLocalPassword
#      retrieves the local password entry (if there is one) for the specified user
def getLocalPassword(username):
    password = None
    if LOCAL_PASSWD_FILE is not None and os.path.isfile(LOCAL_PASSWD_FILE):
        f = open(LOCAL_PASSWD_FILE)
        for line in f:
            line = line.strip()
            if line.startswith("%s:" % username):
                parts = line.split(":", 2)
                password = parts[1]
                break
        f.close()
    return password

# checkLDAPPassword
#      checks the user and password against the LDAP server
def checkLDAPPassword(userid, password):
    # the following is needed to allow Python to accept a non-CA cert
    # ldap.set_option(ldap.OPT_X_TLS_REQUIRE_CERT, ldap.OPT_X_TLS_NEVER)
    con = ldap.initialize(AD_HOST)
    con.set_option(ldap.OPT_NETWORK_TIMEOUT, TIMEOUT)
    con.set_option(ldap.OPT_TIMEOUT, TIMEOUT)
    try:
        # try to bind
        aduser = con.simple_bind_s(DOMAIN + userid, password)
        return 0
    except Exception as e:
        # if bind throws an exception, then access is DENIED
        return 1

# Check Radius password+token
def checkRadiusPassword(userid, password):
    try:
        for radsvr in radsvrs:
            srv = Client(server=radsvr, secret=sharedsecret, dict=Dictionary("dictionary"))
            # create request
            req = srv.CreateAuthPacket(code=pyrad.packet.AccessRequest, User_Name=user, NAS_Identifier="localhost")
            req["User-Password"] = req.PwCrypt(password)
            # send request
            reply = srv.SendPacket(req)
            if reply.code == pyrad.packet.AccessAccept:
                return 0
        return 1
    except Exception as e:
        print("Problem with call to RSA server, please contact the helpdesk.")
        return 1

def getsvcusers(serviceusers):
    svcuserfile = open(SVC_USER_FILE, "r")
    for line in svcuserfile.readlines():
        line = line.lower()
        serviceusers.append(line.strip())
    svcuserfile.close()

# doAuthenticate
#    main routine for performing the authentication logic
def doAuthenticate(userid, serverid):
    svcusers = []
    # read password from STDIN -- this is passed by a perforce client when the user does a 'p4 login' and is prompted for a password
    # default behavior is to deny access (return code 1)
    result = 1
    svc_user = 0
    password = sys.stdin.read().strip()
    if (password == ""):
        print("Blank password not allowed.")
        sys.exit(1)

    # check to see if the user account exists in the local password file
    # if user exists in local password file, we will not consult LDAP
    if (localUserExists(userid)):
        # get the local password, if there is one
        localPassword = getLocalPassword(userid)
        # user exists in local password file
        if (localPassword is None or len(localPassword.strip()) == 0):
            print("Local password is missing.")
        else:
            # retrieved from the local password file
            if (password == localPassword):
                result = 0
    else:
        getsvcusers(svcusers)
        if (userid.lower() in svcusers):
            result = checkLDAPPassword(userid, password)
            #      pass_file = open("/p4/common/bin/triggers/svc_authlog.txt", "a")
            #      pass_file.write("%s:%s,%s\n" % (userid, password,result))
            #      pass_file.close()
            svc_user = 1
        else:
            result = checkRadiusPassword(userid, password)
        #      pass_file = open("/p4/common/bin/triggers/authlog.txt", "a")
        #      pass_file.write("%s,%s,%s,%s\n" % (serverid, userid, password,result))
        #      pass_file.close()

    # now check the result, returning "authentication failed" error if result is not 0
    if (result):
        if (svc_user):
            print("Invalid login, please check your password.")
        else:
            print("Invalid login, you must enter your RSA pin and token to login.")
    # exit with the result code
    sys.exit(result)

if __name__ == '__main__':
    if len(sys.argv) < 2:
        print("Usage: %s username" % sys.argv[0])
        sys.exit(1)
    userid = sys.argv[1]
    serverid = sys.argv[2]
    doAuthenticate(userid, serverid)