path: root/scripts/radtee

#!/usr/bin/env python2
from __future__ import with_statement 

# RADIUS comparison tee v1.0
# Sniffs local RADIUS traffic, replays incoming requests to a test
# server, and compares the sniffed responses with the responses
# generated by the test server.
#
# Copyright (c) 2009, Frontier Communications
# Copyright (c) 2010, John Morrissey <jwm@horde.net>
#
# This program is free software; you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the Free
# Software Foundation; either version 2 of the License, or (at your option)
# any later version.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
# or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
# for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program; if not, write to the Free Software Foundation, Inc., 59
# Temple Place, Suite 330, Boston, MA 02111-1307, USA.

# Requires
# ========
# - python 2.4 or newer
# - impacket
# - pcapy
# - pyrad, ideally 1.2 or newer

# Output
# ======
# - .: 50 successful, matching responses processed.
# - C=x.x.x.x: Ignored packet sniffed from unknown client.
# - D: Dropped sniffed packet due to processing bottleneck. Consider
#      increasing THREADS.
# - I: Invalid/unparseable packet sniffed.
# - Mreq: Response was sniffed without sniffing a corresponding request.
# - Mresp: Request was sniffed without sniffing a corresponding response.
# - T: Request to test server timed out.

import fcntl
from getopt import gnu_getopt, GetoptError
import os
import Queue
import re
import signal
import socket
import struct
import sys
import thread
from threading import Thread
from time import sleep, time

from impacket.ImpactDecoder import EthDecoder
import pcapy
from pyrad.client import Client
from pyrad.dictionary import Dictionary
from pyrad import packet


TEST_DEST = 'server.example.com'
TEST_SECRET = 'examplesecret'

# Dictionary to use when decoding RADIUS packets. pyrad earlier than
# v1.2 can't parse $INCLUDE directives, so you must combine FreeRADIUS'
# dictionary manually, with something like this:
#
# import re
# import sys
# 
# def combine(file):
#     for line in open(file):
#         matches = re.search(r'^\$INCLUDE\s+(.*)$', line)
#         if not matches:
#             sys.stdout.write(line)
#             continue
# 
#         combine(matches.group(1))
# 
# combine('/etc/freeradius/dictionary')
DICTIONARY = '/etc/freeradius/dictionary'

# Number of worker threads to run.
THREADS = 32

# Mapping of RADIUS request source addresses to shared secrets,
# so we can decode incoming RADIUS requests.
#
# For example:
#     '127.0.0.1': 'test',
CLIENTS = {
}

# Ignore any sniffed requests from these IP addresses.
IGNORE_CLIENTS = [
]

# Expected mismatches to ignore and consider the packet matching.
# Only the differences are compared to these items, so only the
# differing attrs need be listed in the attrs array.
#
# Examples:
# - Ignore mismatched AccessRejects whose sole difference is a
#   Reply-Message attribute with the values given.
#   {
#       'sniffed': {
#           'code': packet.AccessReject,
#           'attrs': [
#               'Reply-Message=Request Denied',
#           ],
#       },
#       'test': {
#           'code': packet.AccessReject,
#           'attrs': [
#               'Reply-Message=Account is disabled.',
#           ],
#       }
#   },
#
# - Ignore mismatched AccessRejects with Reply-Message=Request Denied
#   and arbitrary Cisco dns-servers in the sniffed packet, and
#   no Reply-Message and Cisco-AVPair attrs in the response from the
#   test RADIUS server.
#    {
#        'sniffed': {
#            'code': packet.AccessReject,
#            'attrs': [
#                'Reply-Message=Request Denied',
#                'regex:^Cisco-AVPair=ip:dns-servers=.*$',
#            ],
#        },
#        'test': {
#            'code': packet.AccessReject,
#            'attrs': [
#            ],
#        }
#    },
#
# - Only apply this stanza to sniffed requests with
#   'User-Name= user@example.com' (note the leading whitespace).
#    {
#        'check': [
#            'User-Name= user@example.com',
#        ],
#        'sniffed': {
#            'code': packet.AccessReject,
#            'attrs': [
#                'Reply-Message=Request Denied',
#            ],
#        },
#        'test': {
#            'code': packet.AccessAccept,
#            'attrs': [
#                'Service-Type=Framed-User',
#                'Framed-Protocol=PPP',
#                'Framed-IP-Address=255.255.255.255',
#                'Framed-MTU=1500',
#                'Framed-Compression=Van-Jacobson-TCP-IP',
#            ],
#        }
#    },
IGNORE = [
]


QUEUE = Queue.Queue(maxsize=25000)
DICT = Dictionary(DICTIONARY)

def code2str(code):
	if code == packet.AccessRequest:
		return "Access-Request"
	elif code == packet.AccessAccept:
		return "Access-Accept"
	elif code == packet.AccessReject:
		return "Access-Reject"
	elif code == packet.AccountingRequest:
		return "Accounting-Request"
	elif code == packet.AccountingResponse:
		return "Accounting-Response"
	elif code == packet.AccessChallenge:
		return "Access-Challenge"
	elif code == packet.StatusServer:
		return "Status-Server"
	elif code == packet.StatusClient:
		return "Status-Client"
	elif code == packet.DisconnectRequest:
		return "Disconnect-Request"
	elif code == packet.DisconnectACK:
		return "Disconnect-ACK"
	elif code == packet.DisconnectNAK:
		return "Disconnect-NAK"
	elif code == packet.CoARequest:
		return "CoA-Request"
	elif code == packet.CoAACK:
		return "CoA-ACK"
	elif code == packet.CoANAK:
		return "CoA-NAK"

def handlePacket(header, data):
	"""Place captured packets in the queue to be picked up
	by worker threads."""

	global QUEUE

	try:
		QUEUE.put_nowait(data)
	except Queue.Full:
		sys.stdout.write('D')
		sys.stdout.flush()

def ignore_applies(pkt, ignore):
	"""Determine whether an ignore stanza (based on its check
	items) applies to a packet."""

	# All check items must match for this ignore stanza to apply.
	stanza_applies = True
	for pair in ignore.get('check', []):
		attr, value = pair.split('=')

		if attr not in pkt:
			return False
		if value.startswith('regex:'):
			if not re.search(value.replace('regex:', '', 1), value):
				return False
		elif pkt[attr] != value:
			return False

	return True

def ignores_match(pkt, mismatched, ignore):
	"""Determine whether mismatched AV pairs remain after accounting
	for ignored differences."""

	non_regex_ignore = [
		q
		for q
		 in ignore['attrs']
		 if not q.startswith('regex:')
	]
	regex_ignore = [
		q
		for q
		 in ignore['attrs']
		 if q.startswith('regex:')
	]

	unmatched_av = mismatched[:]
	unmatched_rules = ignore['attrs'][:]
	for av in mismatched:
		if av in non_regex_ignore:
			unmatched_av.remove(av)
			unmatched_rules.remove(av)
			continue
		for regex in regex_ignore:
			if re.search(regex.replace('regex:', '', 1), av):
				unmatched_av.remove(av)
				if regex in unmatched_rules:
					unmatched_rules.remove(regex)
				break

	if unmatched_av or unmatched_rules:
		return False
	return True

def matches(req, sniffed_pkt, test_pkt):
	"""Determine whether a response from the test server matches
	the response sniffed from the wire, accounting for ignored
	differences."""

	global IGNORE

	mis_attrs_sniffed = []
	for k in sniffed_pkt.keys():
		if sorted(sniffed_pkt[k]) == sorted(test_pkt.get(k, [])):
			continue
		mis_attrs_sniffed.append('%s=%s' % (
			k, ', '.join([str(v) for v in sorted(sniffed_pkt[k])])))

	mis_attrs_test = []
	for k in test_pkt.keys():
		if sorted(test_pkt[k]) == sorted(sniffed_pkt.get(k, [])):
			continue
		mis_attrs_test.append('%s=%s' % (
			k, ', '.join([str(v) for v in sorted(test_pkt[k])])))

	# The packets match without having to consider any ignores.
	if sniffed_pkt.code == test_pkt.code and \
	   not mis_attrs_sniffed and not mis_attrs_test:
		return True

	for ignore in IGNORE:
		if not ignore_applies(req, ignore):
			continue

		if ignore['sniffed']['code'] != sniffed_pkt.code or \
		   ignore['test']['code'] != test_pkt.code:
			continue

		if ignores_match(sniffed_pkt, mis_attrs_sniffed, i['sniffed']):
			return True
		if ignores_match(test_pkt, mis_attrs_test, i['test']):
			return True

	return False

def log_mismatch(nas, req, passwd, expected, got):
	"""Emit notification that the test server has returned a response
	that differs from the sniffed response."""

	print 'Mismatch: %s' % nas

	print 'Request: %s' % code2str(req.code)
	for key in req.keys():
		if key == 'User-Password':
			print '\t%s: %s' % (key, passwd)
			continue
		print '\t%s: %s' % (
			key, ', '.join([str(v) for v in req[key]]))

	print 'Expected: %s' % code2str(expected.code)
	for key in expected.keys():
		print '\t%s: %s' % (
			key, ', '.join([str(v) for v in expected[key]]))

	print 'Got: %s' % code2str(got.code)
	for key in got.keys():
		print '\t%s: %s' % (
			key, ', '.join([str(v) for v in got[key]]))

	print

REQUESTS = {}
REQUESTS_LOCK = thread.allocate_lock()
NUM_SUCCESSFUL = 0
def check_for_match(key, req_resp):
	"""Send a copy of the original request to the test server and
	determine whether the response matches the response sniffed from
	the wire."""

	global DICT, NUM_SUCCESSFUL, TEST_DEST, TEST_SECRET
	global REQUESTS, REQUESTS_LOCK

	client = Client(server=TEST_DEST,
		secret=TEST_SECRET, dict=DICT)
	fwd_req = client.CreateAuthPacket(code=packet.AccessRequest)
	fwd_req.authenticator = req_resp['req']['pkt'].authenticator

	keys = req_resp['req']['pkt'].keys()
	for k in keys:
		for value in req_resp['req']['pkt'][k]:
			fwd_req.AddAttribute(k, value)
	if 'User-Password' in keys:
		fwd_req['User-Password'] = fwd_req.PwCrypt(req_resp['req']['passwd'])
	if 'NAS-IP-Address' in fwd_req:
		del fwd_req['NAS-IP-Address']
	fwd_req.AddAttribute('NAS-IP-Address', req_resp['req']['ip'])

	try:
		test_reply = client.SendPacket(fwd_req)
	except:
		# Request to test server timed out.
		sys.stdout.write('T')
		sys.stdout.flush()
		with REQUESTS_LOCK:
			del REQUESTS[key]
		return

	if not matches(req_resp['req']['pkt'],
		req_resp['response']['pkt'], test_reply):

		print
		log_mismatch(req_resp['req']['ip'],
			req_resp['req']['pkt'],
			req_resp['req']['passwd'],
			req_resp['response']['pkt'], test_reply)

	with REQUESTS_LOCK:
		# Occasionally, this key isn't present. Maybe retransmissions
		# due to a short timeout on the remote RADIUS client's end
		# and a subsequent race?
		if key in REQUESTS:
			del REQUESTS[key]

	NUM_SUCCESSFUL += 1
	if NUM_SUCCESSFUL % 50 == 0:
		sys.stdout.write('.')
		sys.stdout.flush()

class RadiusComparer(Thread):
	def run(self):
		global DICT, IGNORE_CLIENTS, QUEUE, REQUESTS, REQUESTS_LOCK

		while True:
			data = QUEUE.get()
			if not data:
				return

			frame = EthDecoder().decode(data)
			ip = frame.child()
			udp = ip.child()
			rad_raw = udp.child().get_buffer_as_string()

			try:
				pkt = packet.Packet(dict=DICT, packet=rad_raw)
			except packet.PacketError:
				sys.stdout.write('I')
				sys.stdout.flush()
				continue

			if ip.get_ip_src() in IGNORE_CLIENTS:
				continue

			if pkt.code == packet.AccessRequest:
				auth = packet.AuthPacket(data[42:])
				auth.authenticator = pkt.authenticator
				auth.secret = clients.CLIENTS.get(ip.get_ip_src(), None)
				if not auth.secret:
					# No configuration for this client.
					sys.stdout.write('C=%s' % ip.get_ip_src())
					sys.stdout.flush()
					continue
				passwd = None
				if 'User-Password' in pkt.keys():
					passwd = auth.PwDecrypt(pkt['User-Password'][0])

				key = '%s:%d:%d' % (ip.get_ip_src(),
					udp.get_uh_sport(), pkt.id)
				do_compare = None
				with REQUESTS_LOCK:
					if key not in REQUESTS:
						REQUESTS[key] = {}
					REQUESTS[key]['req'] = {
						'ip': ip.get_ip_src(),
						'port': udp.get_uh_sport(),
						'pkt': pkt,
						'passwd': passwd,
					}
					REQUESTS[key]['tstamp'] = time()
					if 'response' in REQUESTS[key]:
						do_compare = REQUESTS[key]

				if do_compare:
					check_for_match(key, do_compare)
			elif pkt.code in [packet.AccessAccept, packet.AccessReject]:
				key = '%s:%d:%d' % (ip.get_ip_dst(),
					udp.get_uh_dport(), pkt.id)
				do_compare = None
				with REQUESTS_LOCK:
					if key not in REQUESTS:
						REQUESTS[key] = {}
					REQUESTS[key]['response'] = {
						'ip': ip.get_ip_src(),
						'port': udp.get_uh_sport(),
						'pkt': pkt,
					}
					REQUESTS[key]['tstamp'] = time()
					if 'req' in REQUESTS[key]:
						do_compare = REQUESTS[key]

				if do_compare:
					check_for_match(key, do_compare)
			else:
				print >>sys.stderr, \
					'Unsupported packet type received: %d' % pkt.code

class RequestsPruner(Thread):
	"""Prune stale request state periodically."""

	def run(self):
		global REQUESTS, REQUESTS_LOCK

		while True:
			sleep(30)

			now = time()
			with REQUESTS_LOCK:
				keys = REQUESTS.keys()
				for key in keys:
					if REQUESTS[key]['tstamp'] + 60 >= now:
						continue

					if 'req' not in REQUESTS[key]:
						sys.stdout.write('Mreq')
						sys.stdout.flush()
					if 'response' not in REQUESTS[key]:
						sys.stdout.write('Mresp')
						sys.stdout.flush()

					del REQUESTS[key]

def usage():
	print 'Usage: %s INTERFACE' % os.path.basename(sys.argv[0])
	print ''
	print '    -h, --help  display this help and exit'

if __name__ == '__main__':
	global PID_FILE

	progname = os.path.basename(sys.argv[0])

	try:
		options, iface = gnu_getopt(sys.argv[1:], 'h', ['help'])
	except GetoptError, e:
		print '%s: %s' % (progname, str(e))
		usage()
		sys.exit(1)

	for option in options:
		if option[0] == '-h' or option[0] == '--help':
			usage()
			sys.exit(0)

	if len(iface) != 1:
		usage()
		sys.exit(1)
	iface = iface[0]

	if os.geteuid() != 0:
		print >>sys.stderr, '%s: must be run as root.' % progname
		sys.exit(1)

	for i in range(0, THREADS):
		RadiusComparer().start()
	RequestsPruner().start()

	s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

	# This is Linux-specific, and there's no tenable way to make
	# it portable.
	# 
	# Unfortunately, we need the interface's IP address to filter out
	# only RADIUS traffic destined for this host (avoiding traffic sent
	# *by* this host, such as proxied requests or our own traffic) to
	# avoid replaying requests not directed to the local radiusd.
	#
	# Furthermore, this only obtains the interface's *first* IP address,
	# so we won't notice traffic sent to additional IP addresses on
	# the given interface.
	#
	# This is Good Enough For Me given the effort I care to invest.
	# Of course, patches enhancing this are welcome.
	if os.uname()[0] == 'Linux':
		local_ipaddr = socket.inet_ntoa(fcntl.ioctl(
			s.fileno(),
			0x8915,  # SIOCGIFADDR
			struct.pack('256s', iface[:15])
		)[20:24])
	else:
		raise Exception('Only the Linux operating system is currently supported.')

	p = pcapy.open_live(iface, 1600, 0, 100)
	p.setfilter('''
		(dst host %s and udp and dst port 1812) or
		(src host %s and udp and src port 1812)''' % \
		(local_ipaddr, local_ipaddr))
	while True:
		try:
			p.dispatch(1, handlePacket)
		except KeyboardInterrupt:
			sys.exit(0)