pam_usb/tools/pamusb-agent

#!/usr/bin/env python
#
# Copyright (c) 2003-2007 Andrea Luzzardi <scox@sig11.org>
#
# This file is part of the pam_usb project. pam_usb is free software;
# you can redistribute it and/or modify it under the terms of the GNU General
# Public License version 2, as published by the Free Software Foundation.
#
# pam_usb 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., 51 Franklin
# Street, Fifth Floor, Boston, MA 02110-1301 USA.

import os
import sys
import pwd
import getopt
import signal
import re
import subprocess
import syslog
import gi
import threading

gi.require_version('UDisks', '2.0')

from gi.repository import GLib
from gi.repository import UDisks

import xml.etree.ElementTree as et

class HotPlugDevice:
	def __init__(self, serial):
		self.__udi = None
		self.__serial = serial
		self.__callbacks = []
		self.__running = False

	def run(self):
		self.__scanDevices()
		self.__registerSignals()
		self.__running = True
		GLib.MainLoop().run()
		print('signals registered')

	def addCallback(self, callback):
		self.__callbacks.append(callback)

	def __scanDevices(self):
		for udi in udisksObjectManager.get_objects():
			if udi.get_block():
				device = udisks.get_drive_for_block(udi.get_block())
				if device:
					self.__deviceAdded(device)

	def __registerSignals(self):
		for signal, callback in (('object-added', self.__objectAdded),
				('object-removed', self.__objectRemoved)):
			udisksObjectManager.connect(signal, callback)

	def __objectAdded(self, _, udi):
		if udi.get_block():
			device = udisks.get_drive_for_block(udi.get_block())
			if device:
				self.__deviceAdded(device)

	def __objectRemoved(self, _, udi):
		if udi.get_block():
			device = udisks.get_drive_for_block(udi.get_block())
			if device:
				self.__deviceRemoved(device)

	def __deviceAdded(self, udi):
		if self.__udi is not None:
			return
		if udi.get_property('serial') != self.__serial:
			return
		self.__udi = udi
		if self.__running:
			[ cb('added') for cb in self.__callbacks ]

	def __deviceRemoved(self, udi):
		if self.__udi is None:
			return
		if self.__udi != udi:
			return
		self.__udi = None
		if self.__running:
			[ cb('removed') for cb in self.__callbacks ]

class Log:
	def __init__(self):
		syslog.openlog('pamusb-agent', syslog.LOG_PID | syslog.LOG_PERROR,
				syslog.LOG_AUTH)

	def info(self, message):
		self.__logMessage(syslog.LOG_NOTICE, message)

	def error(self, message):
		self.__logMessage(syslog.LOG_ERR, message)

	def __logMessage(self, priority, message):
		syslog.syslog(priority, message)

def usage():
	print('Usage: %s [--help] [--config=path] [--daemon] [--check=path]' % \
			os.path.basename(__file__))
	sys.exit(1)

def runAs(uid, gid):
	def set_id():
		os.setuid(uid)
		os.setgid(gid)
	return set_id

import getopt

try:
	opts, args = getopt.getopt(sys.argv[1:], "hc:dc:",
			["help", "config=", "daemon", "check="])
except getopt.GetoptError:
	usage()

options = {'configFile' : '/etc/security/pam_usb.conf',
		'daemon' : False,
		'check' : '/usr/bin/pamusb-check'}

if len(args) != 0:
	usage()

for o, a in opts:
	if o in ('-h', '--help'):
		usage()
	if o in ('-c', '--config'):
		options['configFile'] = a
	if o in ('-d', '--daemon'):
		options['daemon'] = True
	if o in ('-c', '--check'):
		options['check'] = a


if not os.path.exists(options['check']):
	print('%s not found.' % options['check'])
	print("You might specify manually pamusb-check's location using --check.")
	usage()

logger = Log()

doc = et.parse(options['configFile'])
users = doc.findall('users/user')

def userDeviceThread(user):

	userName = user.get('id')
	uid = pwd.getpwnam(userName)[2]
	gid = pwd.getpwnam(userName)[3]
	os.environ = None

	events = {
		'lock' : [],
		'unlock' : []
	}

	for hotplug in user.findall('agent'):
		henvs = {}

		for henv in hotplug.findall('env'):
			henv_var = re.sub(r'^(.*?)=.*$', '\\1', henv.text)
			henv_arg = re.sub(r'^.*?=(.*)$', '\\1', henv.text)
			henvs[henv_var] = henv_arg

		events[hotplug.get('event')].append(
			{
				'env': henvs,
				'cmd': hotplug.find('cmd').text
			}
		)

	deviceName = user.find('device').text.strip()

	devices = doc.findall("devices/device")
	for device in devices:
		if device.get('id') == deviceName:
			break

		logger.error('Device %s not found in configuration file' % deviceName)
		sys.exit(1)

	serial = device.find('serial').text.strip()

	def authChangeCallback(event):
		if event == 'removed':
			logger.info('Device "%s" has been removed, ' \
					'locking down user "%s"...' % (deviceName, userName))

			for l in events['lock']:
				cmd = l['cmd']

				logger.info('Running "%s"' % cmd)
				subprocess.run(cmd.split(), env=l['env'], preexec_fn=runAs(uid, gid))

			logger.info('Locked.')
			return

		logger.info('Device "%s" has been inserted. ' \
				'Performing verification...' % deviceName)
		cmdLine = "%s --debug --config=%s --service=pamusb-agent %s" % (
				options['check'], options['configFile'], userName)
		logger.info('Executing "%s"' % cmdLine)
		if not os.system(cmdLine):
			logger.info('Authentication succeeded. ' \
					'Unlocking user "%s"...' % userName)

			for l in events['unlock']:
				cmd = l['cmd']

				logger.info('Running "%s"' % cmd)
				subprocess.run(cmd.split(), env=l['env'], preexec_fn=runAs(uid, gid))

			logger.info('Unlocked.')

		else:
			logger.info('Authentication failed for device %s. ' \
				'Keeping user "%s" locked down.' % (deviceName, userName))

	hpDev = HotPlugDevice(serial)
	hpDev.addCallback(authChangeCallback)

	logger.info('Watching device "%s" for user "%s"' % (deviceName, userName))
	hpDev.run()

udisks = UDisks.Client.new_sync()
udisksObjectManager = udisks.get_object_manager()

sysUsers= []
validUsers = []

with open('/etc/passwd', 'r') as f:
	for line in f.readlines():
		sysUser = re.sub(r'^(.*?):.*', '\\1', line[:-1])
		sysUsers.append(sysUser)
	f.close()

logger.info('pamusb-agent up and running.')

for userObj in users:
	userId = userObj.get('id')

	for sysUser_ in sysUsers:
		if (userId == sysUser_ and
		userObj not in validUsers):
			validUsers.append(userObj)

# logger.error('User %s not found in configuration file' % username)

for user in validUsers:
	threading.Thread(
		target=userDeviceThread,
		args=(user,)
	).start()

if options['daemon'] and os.fork():
	sys.exit(0)

def sig_handler(sig, frame):
    logger.info('Stopping agent.')
    sys.exit(0)

sys_signals = ['SIGINT', 'SIGTERM', 'SIGTSTP', 'SIGTTIN', 'SIGTTOU']

for i in sys_signals:
    signal.signal(getattr(signal, i), sig_handler)