danvics/python-validity
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

* Use single instance of a systemd service (at least for now)

Browse source 
* Switch to use logger instead of prints
* Catch signals to enforce graceful shutdown
This commit is contained in:
Viktor Dragomiretskyy 2020-07-13 17:48:00 +12:00
parent 886e19d3d6
commit 74598ad291
15 changed files with 78 additions and 48 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
README.md
View file

@ -96,6 +96,7 @@ If you are curious you can enable tracing to see what flows in and out of device
```
>>> tls.trace_enabled=True
>>> usb.trace_enabled=True
>>> logging.basicConfig(level=logging.DEBUG)
>>> db.dump_all()
>tls> 17: 4b00000b0053746757696e64736f7200
>cmd> 1703030050c00a7ff1cf76e90f168141b4bc519ca9598eacb575ff01b7552a3707be8506b246d5272cb119e7b8b3eccd991cb7d8387245953ff1da62cebfb07fae7e47b9b536fb1a82185cc9399d30625ee3c1451f

31
dbus_service/dbus-service
View file

@ -1,5 +1,6 @@
#!/usr/bin/env python3
import signal
import argparse
import re
import os
@ -10,8 +11,9 @@ import dbus.mainloop.glib
import dbus.service
from binascii import hexlify, unhexlify
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
from gi.repository import GObject, GLib
from gi.repository import GLib
import logging
import logging.handlers
from validitysensor import init
from validitysensor.tls import tls
@ -25,7 +27,7 @@ GLib.threads_init()
INTERFACE_NAME='io.github.uunicorn.Fprint.Device'
loop = GObject.MainLoop()
loop = GLib.MainLoop()
usb.quit = lambda e: loop.quit()
@ -53,7 +55,7 @@ class Device(dbus.service.Object):
return []
rc = [subtype_to_string(f['subtype']) for f in usr.fingers]
print(repr(rc))
logging.debug(repr(rc))
return rc
except Exception as e:
raise e
@ -115,7 +117,7 @@ class Device(dbus.service.Object):
try:
# TODO hardcode the username and finger for now
z=enroll(uid2identity(uid), 0xf5)
print('Enroll was successfull')
logging.debug('Enroll was successfull')
self.EnrollStatus('enroll-completed', True)
except Exception as e:
self.EnrollStatus('enroll-failed', True)
@ -187,9 +189,14 @@ if __name__ == '__main__':
args = parser.parse_args()
if args.debug:
logging.basicConfig(level=logging.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()
@ -203,7 +210,7 @@ if __name__ == '__main__':
init.open_devpath(*map(int, z.groups()))
except RebootException:
print('Initialization ended up in rebooting the sensor. Normal exit.')
logging.debug('Initialization ended up in rebooting the sensor. Normal exit.')
sys.exit(0)
bus = dbus.SystemBus()
@ -224,5 +231,15 @@ if __name__ == '__main__':
# Kick off the open-fprintd if it was not started yet
bus.get_object('net.reactivated.Fprint', '/net/reactivated/Fprint/Manager')
loop.run()
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('Oops', e)
raise e

3
debian/python3-validity.prerm vendored
View file

@ -2,9 +2,8 @@
set -e
# Stop the socket on remove so syslog-ng is not restarted via socket activation
if [ -x "/usr/bin/deb-systemd-invoke" ] && [ "$1" = remove ]; then
deb-systemd-invoke stop 'python3-validity@*' || true
deb-systemd-invoke stop 'python3-validity.service' || true
fi
#DEBHELPER#

8
debian/python3-validity.service vendored Normal file
View file

@ -0,0 +1,8 @@
[Unit]
Description=python-validity driver dbus service
[Service]
Type=simple
ExecStart=/usr/lib/python-validity/dbus-service --debug
Restart=no

7
debian/python3-validity.udev vendored
View file

@ -11,8 +11,11 @@ LABEL="python_validity_match"
#TAG+="validity"
ACTION=="add|change", RUN+="/bin/systemctl --no-block start python3-validity@usb-$env{BUSNUM}-$env{DEVNUM}.service"
ACTION=="remove", RUN+="/bin/systemctl --no-block stop python3-validity@usb-$env{BUSNUM}-$env{DEVNUM}.service"
ACTION=="add|change", ATTR{power/control}="auto", RUN+="/bin/systemctl --no-block start python3-validity.service"
ACTION=="remove", RUN+="/bin/systemctl --no-block stop python3-validity.service"
#ACTION=="add|change", RUN+="/bin/systemctl --no-block start python3-validity@usb-$env{BUSNUM}-$env{DEVNUM}.service"
#ACTION=="remove", RUN+="/bin/systemctl --no-block stop python3-validity@usb-$env{BUSNUM}-$env{DEVNUM}.service"
# TODO: Figure out why the following is not working properly instead:
#TAG+="systemd", ENV{SYSTEMD_WANTS}+="python3-validity@usb-$env{BUSNUM}-$env{DEVNUM}.service"

10
debian/python3-validity@.service vendored
View file

@ -1,10 +0,0 @@
[Unit]
Description=python-validity driver dbus service for %I
#StopWhenUnneeded=true
[Service]
Type=simple
ExecStart=/usr/lib/python-validity/dbus-service --devpath %i
StandardOutput=syslog
Restart=no

1
scripts/prototype.py
View file

@ -7,6 +7,7 @@ from validitysensor.sid import *
from validitysensor.init import open as open9x
from threading import Condition
from time import sleep
import logging
import code
#usb.trace_enabled = True

1
validitysensor/db.py
View file

@ -1,4 +1,5 @@
import logging
from .util import unhex
from .tls import tls
from .util import assert_status

5
validitysensor/init.py
View file

@ -1,5 +1,7 @@
import atexit
import signal
import logging
from validitysensor.usb import usb
from validitysensor.tls import tls
@ -10,11 +12,13 @@ from validitysensor.upload_fwext import upload_fwext
from validitysensor.init_db import init_db
def close():
logging.debug('In atexit handler')
if usb.dev is not None:
# Send the reboot command before closing the device.
# Without it the sensor seems to keep creating new TLS sessions and eventually runs out of memory.
# The reboot command may fail if we're shutting down because the device is already gone.
try:
logging.debug('atexit: forcing reboot')
reboot()
except RebootException:
pass
@ -36,6 +40,7 @@ def open_common():
# Don't attempt to autoreboot unless we finished initializing the sensor successfully.
# Otherwise if there is a bug in initialization, we may end up in a loop constantly rebooting the device.
atexit.register(close)
logging.debug('atexit registered')
def open():
usb.open()

5
validitysensor/init_db.py
View file

@ -1,5 +1,6 @@
from struct import pack, unpack
import logging
from .db import db
from .usb import usb
@ -15,13 +16,13 @@ def machine_id_rec_value(b):
def init_db(machine_guid='e7260876-58db-4d27-8c40-8d13110d6a71'):
stg = db.get_user_storage(name='StgWindsor')
if stg == None:
print('Creating a new user storage object')
logging.info('Creating a new user storage object')
db.new_user_storate()
rc = db.get_storage_data()
if rc == []:
print('Creating a host machine GUID record')
logging.info('Creating a host machine GUID record')
stg = db.get_user_storage(name='StgWindsor')
db.new_record(stg.dbid, 0x8, stg.dbid, machine_id_rec_value(machine_guid))
rc = db.get_storage_data()

7
validitysensor/init_flash.py
View file

@ -2,6 +2,7 @@
import os
from struct import pack, unpack
from binascii import hexlify, unhexlify
import logging
from hashlib import sha256
import hmac
@ -86,7 +87,7 @@ def serialize_partition(p):
return b
def partition_flash(info, layout, client_public):
print('Detected Flash IC: %s, %d bytes' % (info.ic.name, info.ic.size))
logging.info('Detected Flash IC: %s, %d bytes' % (info.ic.name, info.ic.size))
cmd = unhex('4f 0000 0000')
cmd += with_hdr(0, serialize_flash_params(info.ic))
@ -106,10 +107,10 @@ def init_flash():
info = get_flash_info()
if len(info.partitions) > 0:
print('Flash has %d partitions.' % len(info.partitions))
logging.info('Flash has %d partitions.' % len(info.partitions))
return
else:
print('Flash was not initialized yet. Formatting...')
logging.info('Flash was not initialized yet. Formatting...')
assert_status(usb.cmd(reset_blob))

24
validitysensor/sensor.py
View file

@ -2,6 +2,7 @@
from enum import Enum
from hashlib import sha256
import os.path
import logging
from .tls import tls
from .usb import usb
from .db import db, subtype_to_string
@ -199,7 +200,7 @@ class Sensor():
self.interrupt_cb = None
self.device_info = identify_sensor()
print('Opening sensor: %s' % self.device_info.name)
logging.info('Opening sensor: %s' % self.device_info.name)
self.type_info = SensorTypeInfo.get_by_type(self.device_info.type)
if self.device_info.type == 0x199:
@ -558,10 +559,10 @@ class Sensor():
if start != b'\xff' * 0x44:
if clean_slate[:0x44] == start:
print('Calibration data already matches the data on the flash.')
logging.info('Calibration data already matches the data on the flash.')
return
else:
print('Calibration flash already written. Erasing.')
logging.info('Calibration flash already written. Erasing.')
erase_flash(6)
write_flash_all(6, 0, clean_slate)
@ -577,13 +578,12 @@ class Sensor():
hs, zeroes = start[0:0x20], start[0x20:0x40]
if zeroes != b'\0'*0x20:
print('Unexpected contents in calibration flash partition')
logging.warning('Unexpected contents in calibration flash partition')
return False
img = read_flash_all(6, 0x44, l)
if hs != sha256(img).digest():
print('Calibration flash hash mismatch')
print(hexlify(hs), hexlify(img))
logging.warning('Calibration flash hash mismatch')
return False
return True
@ -593,23 +593,23 @@ class Sensor():
if os.path.isfile(calib_data_path):
with open(calib_data_path, 'rb') as f:
self.calib_data = f.read()
print('Calibration data loaded from a file.')
logging.info('Calibration data loaded from a file.')
if self.check_clean_slate():
return
else:
print('No calibration data on the flash. Calibrating...')
logging.info('No calibration data on the flash. Calibrating...')
else:
self.calib_data = b''
print('No calibration data was loaded. Calibrating...')
logging.info('No calibration data was loaded. Calibrating...')
for i in range(0, self.calibration_iterations):
print('Calibration iteration %d...' % i)
logging.debug('Calibration iteration %d...' % i)
rsp = tls.cmd(self.build_cmd_02(CaptureMode.CALIBRATE))
assert_status(rsp)
self.process_calibration_results(self.average(usb.read_82()))
print('Requesting a blank image...')
logging.debug('Requesting a blank image...')
# Get the "clean slate" image to store on the flash for fine-grained after-capture adjustments
rsp = tls.cmd(self.build_cmd_02(CaptureMode.CALIBRATE))
@ -753,7 +753,7 @@ class Sensor():
elif tag == 3:
tid = res[magic_len:magic_len+l]
else:
print('Ignoring unknown tag %x' % tag)
logging.warning('Ignoring unknown tag %x' % tag)
res=res[magic_len+l:]

3
validitysensor/tls.py
View file

@ -3,6 +3,7 @@ import hmac
import sys
import os
import pickle
import logging
from struct import pack, unpack
from binascii import *
@ -143,7 +144,7 @@ class Tls():
def trace(self, s):
if self.trace_enabled:
print(s)
logging.debug(s)
def app(self, b):
b = b() if callable(b) else b

11
validitysensor/upload_fwext.py
View file

@ -1,6 +1,7 @@
from binascii import hexlify, unhexlify
from time import ctime
import logging
from os.path import basename
from .tls import tls
@ -25,10 +26,10 @@ def default_fwext_name():
def upload_fwext(fw_path=None):
fwi=get_fw_info(2)
if fwi != None:
print('Detected firmware version %d.%d (%s))' % (fwi.major, fwi.minor, ctime(fwi.buildtime)))
logging.info('Detected firmware version %d.%d (%s))' % (fwi.major, fwi.minor, ctime(fwi.buildtime)))
return
else:
print('No firmware detected. Uploading...')
logging.info('No firmware detected. Uploading...')
# no idea what this is:
write_hw_reg32(0x8000205c, 7)
@ -36,7 +37,7 @@ def upload_fwext(fw_path=None):
raise Exception('Unexpected register value')
dev=identify_sensor()
print('Sensor: %s' % dev.name)
logging.debug('Sensor: %s' % dev.name)
# ^ TODO -- what is the real reason to detect HW at this stage?
# just a guess: perhaps it is used to construct fwext filename
@ -45,7 +46,7 @@ def upload_fwext(fw_path=None):
if not fw_path:
fw_path = default_name
elif basename(fw_path) != default_name:
print('WARNING: Your device fw is supposed to be called {}'.format(
logging.warning('WARNING: Your device fw is supposed to be called {}'.format(
default_name))
with open(fw_path, 'rb') as f:
@ -62,7 +63,7 @@ def upload_fwext(fw_path=None):
if fwi == None:
raise Exception('No firmware detected')
print('Loaded FWExt version %d.%d (%s), %d modules' % (fwi.major, fwi.minor, ctime(fwi.buildtime), len(fwi.modules)))
logging.info('Loaded FWExt version %d.%d (%s), %d modules' % (fwi.major, fwi.minor, ctime(fwi.buildtime), len(fwi.modules)))
# Reboot
reboot()

9
validitysensor/usb.py
View file

@ -1,4 +1,5 @@
import logging
import usb.core as ucore
from binascii import *
from .util import assert_status, unhex
@ -80,7 +81,7 @@ class Usb():
(err,), rsp = unpack('<H', rsp[:2]), rsp[2:]
if err != 0:
# fwext is not loaded
print('Clean slate')
logging.info('Clean slate')
self.cmd(init_hardcoded_clean_slate)
def cmd(self, out):
@ -113,12 +114,12 @@ class Usb():
self.trace('<int< %s' % hexlify(resp).decode())
cb=self.interrupt_cb
if cb is None:
print('Ignoring spurious interrupt: %s' % hexlify(resp).decode())
logging.warning('Ignoring spurious interrupt: %s' % hexlify(resp).decode())
else:
try:
cb(resp)
except Exception as e:
print('Exception on the interrupt thread: %s' % e)
logging.warning('Exception on the interrupt thread: %s' % e)
except USBError as e:
self.trace('<int< Exception on interrupt thread: %s' % repr(e))
if self.quit != None:
@ -129,6 +130,6 @@ class Usb():
def trace(self, s):
if self.trace_enabled:
print(s)
logging.debug(s)
usb=Usb()