rad_authcheck.py #2

#!/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 RAD_SERVERS 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% %clientip%"
#
# Note: The script current is set such that the Perforce user names should match the RSA
# ID's. In the case of one customer, the RSA ID's were all numeric, so we made the Perforce
# usernames be realname_RSAID and had this script just strip off the realname_ part.
# Example commented out in the main function.

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

# Configuration values
LDAP_TIMEOUT = 20
AD_HOSTS = [ "ldap://ad.company.com" ]
DOMAIN = "YOURDOMAIN\\"
RAD_SERVERS = ["server1", "server2", "server3"]
RAD_SHARED_SECRET = b"your_shared_secret"

LOCAL_PASSWD_FILE = "/p4/common/bin/triggers/local.passwd"
SVC_USER_FILE = "/p4/common/bin/triggers/serviceusers.txt"
RAD_DICTIONARY = "/p4/common/bin/triggers/dictionary"
ERRORLOG = "/p4/common/bin/triggers/rad_errors.log"
P4 = "/p4/common/bin/p4"
TWO_FACTOR_ERROR = "Invalid login, you must enter your RSA pin and token to login."
SVC_USER_ERROR = "Invalid login, please check your password."
TWO_FACTOR_SVR_ERROR = "Problem with call to RSA server, please contact the helpdesk."
blocked_users = []
blocked_ips = []
BLOCKED_USER_ERROR = "User account currently blocked. Contact the helpdesk for assistance."
BLOCKED_IP_ERROR = "Your client ip address is currently blocked, please contact the helpdesk for assistance."

# 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):
  for ad_host in AD_HOSTS:
    # 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, LDAP_TIMEOUT)
    con.set_option(ldap.OPT_TIMEOUT, LDAP_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
  # If all bind attempts fail, return 1, no access.
  return 1

# Check Radius password+token
def checkRadiusPassword(userid, password):
  try:
    for radsvr in RAD_SERVERS:
      srv = Client(server=radsvr, secret=RAD_SHARED_SECRET, dict=Dictionary(RAD_DICTIONARY))
      # create request
      req = srv.CreateAuthPacket(code=pyrad.packet.AccessRequest, User_Name=userid, 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(TWO_FACTOR_SVR_ERROR)
    errorlog = open(ERRORLOG, "a")
    errorlog.write("%s,%s,%s" % (userid, serverid, e))
    errorlog.close()
    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)

  # now check the result, returning "authentication failed" error if result is not 0
  if (result):
    if (svc_user):
      print(SVC_USER_ERROR)
    else:
      print(TWO_FACTOR_ERROR)
  # exit with the result code
  sys.exit(result)

if __name__ == '__main__':
  if len(sys.argv) < 4:
    print( "Usage: %s %user% %serverid% %clientip%" % sys.argv[0])
    sys.exit(1)
  userid = sys.argv[1]
  serverid = sys.argv[2]
  clientip = sys.argv[3]
  if userid.lower() in blocked_users:
    print( BLOCKED_USER_ERROR )
    sys.exit(1)
  if clientip in blocked_ips:
    print( BLOCKED_IP_ERROR )
    sys.exit(1)
  doAuthenticate(userid, serverid)