256 lines
8.1 KiB
Python
Executable file
256 lines
8.1 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
|
|
import signal
|
|
import argparse
|
|
import re
|
|
import os
|
|
import pwd
|
|
import sys
|
|
import time
|
|
import dbus
|
|
import dbus.mainloop.glib
|
|
import dbus.service
|
|
from usb import core as usb_core
|
|
from binascii import hexlify, unhexlify
|
|
from threading import Thread
|
|
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
|
|
from gi.repository import GLib
|
|
import logging
|
|
import logging.handlers
|
|
|
|
from validitysensor import init
|
|
from validitysensor.tls import tls
|
|
from validitysensor.usb import usb
|
|
from validitysensor.sid import sid_from_string
|
|
from validitysensor.db import subtype_to_string, db
|
|
from validitysensor.sensor import sensor, reboot, RebootException
|
|
|
|
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')
|
|
|
|
def uid2identity(uid):
|
|
sidstr='S-1-5-21-111111111-1111111111-1111111111-%d' % uid
|
|
return sid_from_string(sidstr)
|
|
|
|
class Device(dbus.service.Object):
|
|
def __init__(self, bus_name):
|
|
dbus.service.Object.__init__(self, bus_name, '/io/github/uunicorn/Fprint/Device')
|
|
|
|
@dbus.service.method(dbus_interface=INTERFACE_NAME,
|
|
in_signature="s",
|
|
out_signature="as")
|
|
def ListEnrolledFingers(self, usr):
|
|
try:
|
|
logging.debug('In ListEnrolledFingers %s' % usr)
|
|
|
|
pw=pwd.getpwnam(usr)
|
|
uid=pw.pw_uid
|
|
usr=db.lookup_user(uid2identity(uid))
|
|
|
|
if usr == 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)
|
|
pw=pwd.getpwnam(user)
|
|
usr=db.lookup_user(uid2identity(pw.pw_uid))
|
|
|
|
if usr == 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))
|
|
|
|
pw = pwd.getpwnam(user)
|
|
usr = db.lookup_user(uid2identity(pw.pw_uid))
|
|
|
|
if usr == 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))
|
|
pw=pwd.getpwnam(user)
|
|
uid=pw.pw_uid
|
|
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(uid2identity(uid), 0xf5, update_cb) # TODO parse the finger name
|
|
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)
|
|
|
|
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')
|
|
|
|
if __name__ == '__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>')
|
|
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])
|
|
|
|
backoff()
|
|
|
|
try:
|
|
if args.devpath is None:
|
|
init.open()
|
|
else:
|
|
z = re.match('^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)
|
|
|
|
watcher = None
|
|
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('Cought 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
|
|
|