I was looking for specific testing scenario and I need to create a simple Radius accounting server which is connected to LDAP. This server should do :
Upon receiving client Radius accounting start packet , server parses Framed-IP-Address, Calling-Station-Id and many other 3GPP- vendor and custom attributes and update this information in LDAP DB and respond with ACK.
At the beginning I was experimented with pyrad, but spent some time to create and attach customized dictionary properly and also for 3GPP attributes and later had difficulties to parse these attributes , but chose pyprotosim software (http://sourceforge.net/projects/pyprotosim/) and documentation for python sockets to build small udp server and clients. The dictionary is stored in xml format and easy can be exported and added your own attributes when needed.
So, here is the an example of simple Radius accounting server which is using python-ldap module for LDAP connectivity.
In Ubuntu :
sudo apt-get install python-ldap will install it on the system.
and unpack pyprotosim and create some directory under it.
$demo_radius_acct_server.py
#!/usr/bin/python
# Simple example of small Radius Accounting Server on basis of pyprotosim #software
# which parses incoming message and searches Calling-Station-Id and Framed-IP-Address AVPs
# (which are IDENTITY AND IPADDRESS of user in LDAP).
# If the accounting packet is of type = Start, then it connects to LDAP DB, where
# this user is stored and calls function to update received from radius packet new IP address in
# LDAP DB for that user (identity).
# Please refer to BSD license and copyright information for pyprotosim #software to (http://sourceforge.net/projects/pyprotosim/)
##############################################
import sys
import ldap
#Next line is to include parent directory in PATH where libraries are
sys.path.append("../")
# Remove it normally
from libRadius import *
######################################################################
# Function to update new IP address for this subscriber in LDAP DB
######################################################################
def update_user(identity,ipaddress):
try:
print "UPDATING IP ADDRESS in LDAP DB"
IDENTITY=identity
IPADDRESS=ipaddress
l = ldap.initialize(LDAP_HOST_URL)
ldap_username = LDAP_USERNAME
ldap_password = LDAP_PASSWORD
l.simple_bind(ldap_username, ldap_password)
baseDN = "cn=users,o=mycompany,o=org"
searchScope = ldap.SCOPE_SUBTREE
retrieveAttributes = None
except:
# NO LDAP CONNECTIVITY - RAISE an ERROR
print "THERE IS NO LDAP CONNECTIVITY!!!! PLEASE CHECK LDAP CONNECTION"
try:
from ldap import modlist
# Here is your ldap filter string
ldap_filter='identity=' + str(IDENTITY)
dn=ldap_filter + "," + baseDN
mod_attrs = [( ldap.MOD_REPLACE, 'ipaddress', str(IPADDRESS) )]
l.modify_s(dn,mod_attrs)
print "successfully modified ipaddress for user:", IDENTITY, "in LDAB DB"
except ldap.LDAPError, error_message:
print error_message
# Its nice to the server to disconnect and free resources when done
l.unbind_s()
##############################################################################
# Function to search IP address of client for this subscriber in LDAP DB
##############################################################################
# That's the main function that updates IP address value in LDAP DB accordingly
# for the given user identity and creates Accounting-Response
def create_Acct_Response():
# Create message header (empty)
RES=HDRItem()
stripHdr(RES,data.encode("hex"))
RID=RES.Identifier
RES.Code=dictCOMMANDname2code("Accounting-Response")
RES.Identifier=RID
REQ_avps=splitMsgAVPs(RES.msg)
STATE=findAVP("Acct-Status-Type",REQ_avps)
try:
IDENTITY=findAVP("Calling-Station-Id",REQ_avps)
IPADDR=findAVP("Framed-IP-Address",REQ_avps)
except:
pass
print "Found IDENTITY/IP:", str(IDENTITY) + ", " + str(IPADDR)
STATE=STATE.encode("hex")
if STATE == '00000001':
print "THIS IS ACCOUNTING START"
# updating new IP address in LDAB DB for given user:
update_user(IDENTITY,IPADDR)
elif STATE == '00000002':
print "THIS IS ACCOUNTING STOP"
# do some code here
elif STATE == '00000003':
print "THIS IS ACCOUNTING INTERIM"
# do some code here
elif STATE == '00000007':
print "THIS IS ACCOUNTING ON"
# add code here if needed
elif STATE == '00000008':
print "THIS IS ACCOUNTING OFF"
# add code here if needed
else:
print "UNKNOWN PACKET RECEIVED"
# The RES_avps are left empty here because function createWithAuthenticator must be constructed with RES_avps according libRadius.xml
# But according RFC for Radius protocol, accounting response could have no any AVPs in it, therefore leaving RES_avps as empty array
RES_avps=[]
auth=createZeroAuthenticator()
# Accounting-Response has no need to have attributes in it
msg=createWithAuthenticator(RES,auth,RES_avps,SECRET)
# msg now contains Accounting-Response as hex string
return msg
if __name__ == "__main__":
#logging.basicConfig(level=logging.DEBUG)
#logging.basicConfig(level=logging.INFO)
LoadDictionary("../dictRadius.xml")
#Change your LDAP IP and Port here:
LDAP_HOST_URL='ldap://127.0.0.1:389'
#Change to your ldap user
LDAP_USERNAME="cn=admin,o=mycompany,o=org"
LDAP_PASSWORD="secret"
# Set up here IP and Port for RADIUS ACCOUNTING server
RADIUS_IP = "127.0.0.1"
RADIUS_PORT = 1813
BUFFER_SIZE=4096
# Set up shared secret here
SECRET="SOMEPASSWORD"
# Now creating simple udp socket
RADIUS_server = socket.socket(socket.AF_INET,socket.SOCK_DGRAM)
RADIUS_server.bind((RADIUS_IP, RADIUS_PORT))
# Looping server until user sends CTRL+C or kill to stop it.
while True:
data, addr = RADIUS_server.recvfrom(BUFFER_SIZE)
if (data != ""):
msg=create_Acct_Response()
dbg="Sending response"
logging.info(dbg)
RADIUS_server.sendto(msg.decode("hex"),addr)
#End of code
./demo_radius_acct_server.py
Found IDENTITY/IP: 123456789012346, 192.168.0.220
THIS IS ACCOUNTING START
UPDATING IP ADDRESS in LDAP DB
successfully modified ipaddress for user: 123456789012346 in LDAB DB
No comments:
Post a Comment