297 lines
9.7 KiB
Python
Executable file
297 lines
9.7 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
|
|
import argparse
|
|
import logging
|
|
import logging.handlers
|
|
import os
|
|
import pwd
|
|
import re
|
|
import signal
|
|
import sys
|
|
import time
|
|
import typing
|
|
from binascii import hexlify, unhexlify
|
|
from pathlib import Path
|
|
from threading import Thread
|
|
|
|
import dbus
|
|
import dbus.mainloop.glib
|
|
import dbus.service
|
|
import yaml
|
|
from gi.repository import GLib
|
|
from usb import core as usb_core
|
|
|
|
from validitysensor import init
|
|
from validitysensor.db import subtype_to_string, db, SidIdentity, User
|
|
from validitysensor.sensor import sensor, RebootException
|
|
from validitysensor.sid import sid_from_string
|
|
from validitysensor.tls import tls
|
|
from validitysensor.usb import usb
|
|
from validitysensor.winbio_constants import finger_ids
|
|
|
|
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
|
|
GLib.threads_init()
|
|
|
|
INTERFACE_NAME = 'io.github.uunicorn.Fprint.Device'
|
|
|
|
loop = GLib.MainLoop()
|
|
|
|
|
|
class NoEnrolledPrints(dbus.DBusException):
|
|
_dbus_error_name = 'net.reactivated.Fprint.Error.NoEnrolledPrints'
|
|
|
|
def __init__(self):
|
|
super().__init__('No enrolled prints found')
|
|
|
|
|
|
class Device(dbus.service.Object):
|
|
def __init__(self, bus_name: dbus.Bus, config: typing.Dict[str, typing.Any]):
|
|
dbus.service.Object.__init__(self, bus_name, '/io/github/uunicorn/Fprint/Device')
|
|
self.config = config
|
|
|
|
def user2identity(self, user: str) -> SidIdentity:
|
|
"""Compute UID to SID mapping"""
|
|
pw = pwd.getpwnam(user)
|
|
uid = pw.pw_uid
|
|
if user in self.config['user_to_sid']:
|
|
sidstr = self.config['user_to_sid'][user]
|
|
else:
|
|
sidstr = 'S-1-5-21-111111111-1111111111-1111111111-%d' % uid
|
|
return sid_from_string(sidstr)
|
|
|
|
def user2record(self, user: str) -> User:
|
|
"""Resolve user name to database object"""
|
|
return db.lookup_user(self.user2identity(user))
|
|
|
|
@dbus.service.method(dbus_interface=INTERFACE_NAME, in_signature="s", out_signature="as")
|
|
def ListEnrolledFingers(self, user):
|
|
try:
|
|
logging.debug('In ListEnrolledFingers %s' % user)
|
|
|
|
usr = self.user2record(user)
|
|
|
|
if usr is None:
|
|
return []
|
|
|
|
rc = [subtype_to_string(f['subtype']) for f in usr.fingers]
|
|
logging.debug(repr(rc))
|
|
return rc
|
|
except Exception as e:
|
|
raise e
|
|
|
|
@dbus.service.method(dbus_interface=INTERFACE_NAME, in_signature='s', out_signature='')
|
|
def DeleteEnrolledFingers(self, user):
|
|
logging.debug('In DeleteEnrolledFingers %s' % user)
|
|
usr = self.user2record(user)
|
|
|
|
if usr is None:
|
|
return
|
|
|
|
db.del_record(usr.dbid)
|
|
|
|
@dbus.service.method(dbus_interface=INTERFACE_NAME, in_signature='ss', out_signature='')
|
|
def VerifyStart(self, user, finger):
|
|
logging.debug('In VerifyStart for %s, %s' % (user, finger))
|
|
|
|
usr = self.user2record(user)
|
|
|
|
if usr is None:
|
|
raise NoEnrolledPrints()
|
|
|
|
self.VerifyFingerSelected('any')
|
|
|
|
def update_cb(e):
|
|
self.VerifyStatus('verify-retry-scan', False)
|
|
|
|
def run():
|
|
try:
|
|
# TODO: pass down the user db record id and implement a proper Sensor.verify() method
|
|
usrid, subtype, hsh = sensor.identify(update_cb)
|
|
if usr.dbid == usrid:
|
|
self.VerifyStatus('verify-match', True)
|
|
else:
|
|
self.VerifyStatus('verify-no-match', True)
|
|
except usb_core.USBError as e:
|
|
logging.exception(e)
|
|
self.VerifyStatus('verify-no-match', True)
|
|
loop.quit()
|
|
except Exception as e:
|
|
logging.exception(e)
|
|
self.VerifyStatus('verify-no-match', True)
|
|
|
|
thread = Thread(target=run)
|
|
thread.daemon = True
|
|
thread.start()
|
|
|
|
@dbus.service.method(dbus_interface=INTERFACE_NAME, in_signature='', out_signature='')
|
|
def Cancel(self):
|
|
sensor.cancel()
|
|
|
|
@dbus.service.method(dbus_interface=INTERFACE_NAME, in_signature='ss', out_signature='')
|
|
def EnrollStart(self, user, finger_name):
|
|
logging.debug('In EnrollStart %s for %s' % (finger_name, user))
|
|
|
|
# left-ring-finger => LH
|
|
hand = 'LH' if finger_name[0] == 'l' else 'RH'
|
|
# left-ring-finger => RING_FINGER
|
|
generic_finger = '_'.join(finger_name.split('-')[1:]).upper()
|
|
winbio_name = 'WINBIO_ANSI_381_POS_' + hand + '_' + generic_finger
|
|
|
|
usr = self.user2identity(user)
|
|
index = finger_ids.get(winbio_name, None)
|
|
|
|
def update_cb(rsp, e):
|
|
if e is not None:
|
|
self.EnrollStatus('enroll-retry-scan', False)
|
|
else:
|
|
self.EnrollStatus('enroll-stage-passed', False)
|
|
|
|
def run():
|
|
try:
|
|
sensor.enroll(usr, index, update_cb)
|
|
self.EnrollStatus('enroll-completed', True)
|
|
except usb_core.USBError as e:
|
|
logging.exception(e)
|
|
self.EnrollStatus('enroll-failed', True)
|
|
loop.quit()
|
|
except Exception as e:
|
|
logging.exception(e)
|
|
self.EnrollStatus('enroll-failed', True)
|
|
|
|
if index is None:
|
|
logging.exception('Unknown finger name passed to enroll? ' + finger_name + ' (' +
|
|
winbio_name + ')')
|
|
self.EnrollStatus('enroll-failed', True)
|
|
else:
|
|
thread = Thread(target=run)
|
|
thread.daemon = True
|
|
thread.start()
|
|
|
|
@dbus.service.signal(dbus_interface=INTERFACE_NAME, signature='sb')
|
|
def VerifyStatus(self, result, done):
|
|
logging.debug('VerifyStatus')
|
|
|
|
@dbus.service.signal(dbus_interface=INTERFACE_NAME, signature='s')
|
|
def VerifyFingerSelected(self, finger):
|
|
logging.debug('VerifyFingerSelected')
|
|
|
|
@dbus.service.signal(dbus_interface=INTERFACE_NAME, signature='sb')
|
|
def EnrollStatus(self, result, done):
|
|
logging.debug('EnrollStatus')
|
|
|
|
@dbus.service.method(dbus_interface=INTERFACE_NAME, in_signature='s', out_signature='s')
|
|
def RunCmd(self, cmd):
|
|
logging.debug('RunCmd')
|
|
return hexlify(tls.app(unhexlify(cmd))).decode()
|
|
|
|
|
|
backoff_file = '/usr/share/python-validity/backoff'
|
|
|
|
|
|
# I don't know how to tell systemd to backoff in case of multiple instance of the same template service, help!
|
|
def backoff():
|
|
if os.path.isfile(backoff_file):
|
|
with open(backoff_file, 'r') as f:
|
|
lines = list(map(float, f.readlines()))
|
|
else:
|
|
lines = []
|
|
|
|
while len(lines) > 0 and lines[0] + 60 < time.time():
|
|
lines = lines[1:]
|
|
|
|
lines += [time.time()]
|
|
|
|
with open(backoff_file, 'w') as f:
|
|
for l in lines:
|
|
f.write('%f\n' % l)
|
|
|
|
if len(lines) > 10:
|
|
raise Exception('Refusing to start more than 10 times per minute')
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser('Open fprintd DBus service')
|
|
parser.add_argument('--debug', help='Enable tracing', action='store_true')
|
|
parser.add_argument('--devpath', help='USB device path: usb-<busnum>-<address>')
|
|
parser.add_argument('--configpath',
|
|
default='/etc/python-validity',
|
|
type=Path,
|
|
help='Path to config files')
|
|
args = parser.parse_args()
|
|
|
|
if args.debug:
|
|
level = logging.DEBUG
|
|
usb.trace_enabled = True
|
|
tls.trace_enabled = True
|
|
else:
|
|
level = logging.INFO
|
|
|
|
handler = logging.handlers.SysLogHandler(address='/dev/log')
|
|
logging.basicConfig(level=level, handlers=[handler])
|
|
|
|
# Load and perform basic validation of config file.
|
|
try:
|
|
with (args.configpath / 'dbus-service.yaml').open(mode='rt') as configfd:
|
|
config = yaml.load(configfd)
|
|
except FileNotFoundError:
|
|
# No configuration file. Create default
|
|
config = {'user_to_sid': {}}
|
|
|
|
if 'user_to_sid' not in config:
|
|
raise Exception("No user to SID mapping in config file!")
|
|
# Be kind to user, and allow empty mapping that got turned into a null by mistake.
|
|
if config['user_to_sid'] is None:
|
|
config['user_to_sid'] = {}
|
|
if not all(isinstance(e, str) for e in config['user_to_sid'].keys()):
|
|
raise Exception("Keys in user to SID map must be strings")
|
|
if not all(isinstance(e, str) for e in config['user_to_sid'].values()):
|
|
raise Exception("Values in user to SID map must be strings")
|
|
|
|
backoff()
|
|
|
|
try:
|
|
if args.devpath is None:
|
|
init.open()
|
|
else:
|
|
z = re.match(r'^usb-(\d+)-(\d+)$', args.devpath)
|
|
if not z:
|
|
parser.error('Option --devpath should look like this: usb-<busnum>-<address>')
|
|
|
|
init.open_devpath(*map(int, z.groups()))
|
|
except RebootException:
|
|
logging.debug('Initialization ended up in rebooting the sensor. Normal exit.')
|
|
sys.exit(0)
|
|
|
|
bus = dbus.SystemBus()
|
|
|
|
svc = Device(bus, config)
|
|
|
|
def watch_cb(name):
|
|
if name == '':
|
|
logging.debug('Manager is offline')
|
|
else:
|
|
logging.debug('Manager is back online, registering')
|
|
mgr = bus.get_object(name, '/net/reactivated/Fprint/Manager')
|
|
mgr = dbus.Interface(mgr, 'net.reactivated.Fprint.Manager')
|
|
mgr.RegisterDevice(svc)
|
|
|
|
watcher = bus.watch_name_owner('net.reactivated.Fprint', watch_cb)
|
|
|
|
# Kick off the open-fprintd if it was not started yet
|
|
bus.get_object('net.reactivated.Fprint', '/net/reactivated/Fprint/Manager')
|
|
|
|
def die(x):
|
|
logging.info('Caught signal %d. Stopping...' % x)
|
|
loop.quit()
|
|
|
|
GLib.unix_signal_add(GLib.PRIORITY_DEFAULT, signal.SIGINT, die, signal.SIGINT)
|
|
GLib.unix_signal_add(GLib.PRIORITY_DEFAULT, signal.SIGTERM, die, signal.SIGINT)
|
|
try:
|
|
loop.run()
|
|
except Exception as e:
|
|
logging.exception(e)
|
|
raise e
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|