Usb timeout fixes - better scan failure handling - prevent dbus clash with fprintd service - "--quiet" dbus option

dbus_service/dbus-service

@@ -23,7 +23,6 @@ from usb import core as usb_core
 
 from validitysensor import init
 from validitysensor.db import subtype_to_string, db, SidIdentity, User
-from validitysensor.init_data_dir import PYTHON_VALIDITY_DATA_DIR, init_data_dir
 from validitysensor.sensor import sensor, RebootException
 from validitysensor.sid import sid_from_string
 from validitysensor.tls import tls
@@ -36,7 +35,6 @@ INTERFACE_NAME = 'io.github.uunicorn.Fprint.Device'
 
 loop = GLib.MainLoop()
 
-init_data_dir()
 
 class NoEnrolledPrints(dbus.DBusException):
     _dbus_error_name = 'net.reactivated.Fprint.Error.NoEnrolledPrints'
@@ -193,7 +191,7 @@ class Device(dbus.service.Object):
         return hexlify(tls.app(unhexlify(cmd))).decode()
 
 
-backoff_file = PYTHON_VALIDITY_DATA_DIR + 'backoff'
+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!
@@ -219,7 +217,9 @@ def backoff():
 
 def main():
     parser = argparse.ArgumentParser('Open fprintd DBus service')
-    parser.add_argument('--debug', help='Enable tracing', action='store_true')
+    group = parser.add_mutually_exclusive_group()
+    group.add_argument('--debug', help='Enable debug output and tracing', action='store_true')
+    group.add_argument('--quiet', help='Suppress all output below INFO level', action='store_true')
     parser.add_argument('--devpath', help='USB device path: usb-<busnum>-<address>')
     parser.add_argument('--configpath',
                         default='/etc/python-validity',
@@ -231,11 +231,15 @@ def main():
         level = logging.DEBUG
         usb.trace_enabled = True
         tls.trace_enabled = True
+    elif args.quiet:
+        level = logging.WARNING
     else:
         level = logging.INFO
 
     handler = logging.handlers.SysLogHandler(address='/dev/log')
-    logging.basicConfig(level=level, handlers=[handler])
+    formatter = logging.Formatter('%(message)s')
+    handler.setFormatter(formatter)
+    logging.basicConfig(level=level, handlers=[handler], format='%(message)s')
 
     # Load and perform basic validation of config file.
     try:

debian/changelog

@@ -1,3 +1,12 @@
+python-validity (0.15~ppa2) noble; urgency=medium
+
+  * Change all write paths to /var/run/python-validity
+  * Use a different partition table + signature for 0090 devices
+  * Docs fixes
+  * switch to noble
+
+ -- unicorn <[email protected]>  Mon, 09 Jun 2025 20:19:55 +1200
+
 python-validity (0.14~ppa1) bionic; urgency=medium
 
   * Retry establishing TLS on Resume in case if it was already established. 

debian/control

@@ -19,7 +19,6 @@ Depends: ${python3:Depends},
  dbus,
  open-fprintd (>= 0.6~),
  innoextract (>= 1.6~)
-XB-Python-Version: ${python3:Versions}
 Description: Validity Fingerprint Sensor DBus Driver
  This package adds support to some Validity sensors.
  .

debian/python3-validity.postinst

@@ -4,9 +4,20 @@ set -e
 #DEBHELPER#
 
 if [ "$1" = "configure" ]; then
+    # Mask fprintd.service to prevent conflicts with open-fprintd
+    if systemctl -q is-enabled fprintd.service 2>/dev/null; then
+        systemctl mask fprintd.service
+    fi
+
+    # Install firmware and reload systemd/udev
     validity-sensors-firmware || true
     systemctl daemon-reload || true
     udevadm control --reload-rules || true
     udevadm trigger || true
+
+    # Enable our service
+    if [ -d /run/systemd/system ]; then
+        systemctl enable python3-validity.service >/dev/null 2>&1 || :
+    fi
 fi
 

debian/python3-validity.prerm

@@ -2,10 +2,23 @@
 
 set -e
 
-#if [ -x "/usr/bin/deb-systemd-invoke" ] && [ "$1" = remove ]; then
-#    deb-systemd-invoke stop 'python3-validity.service' || true
-#fi
-
 #DEBHELPER#
 
+if [ "$1" = "remove" ] || [ "$1" = "deconfigure" ]; then
+    # Stop our service
+    if [ -d /run/systemd/system ] && systemctl is-active --quiet python3-validity.service; then
+        systemctl stop python3-validity.service >/dev/null 2>&1 || :
+    fi
+
+    # Unmask fprintd.service if it was previously masked by us
+    if systemctl is-enabled fprintd.service 2>/dev/null | grep -q masked; then
+        systemctl unmask fprintd.service >/dev/null 2>&1 || :
+    fi
+
+    # Try to start fprintd if it's not running
+    if ! systemctl is-active --quiet fprintd.service 2>/dev/null; then
+        systemctl start fprintd.service >/dev/null 2>&1 || :
+    fi
+fi
+
 exit 0

debian/python3-validity.service

@@ -4,7 +4,7 @@ After=open-fprintd.service
 
 [Service]
 Type=simple
-ExecStart=/usr/lib/python-validity/dbus-service --debug
+ExecStart=/usr/lib/python-validity/dbus-service --quiet
 Restart=no
 
 [Install]

debian/rules

@@ -8,6 +8,9 @@
 override_dh_installsystemd:
 	dh_installsystemd --name=python3-validity
 
-
 override_dh_auto_install:
 	python3 ./setup.py install --root=$(CURDIR)/debian/tmp --prefix=/usr --install-layout=deb
+
+override_dh_auto_clean:
+	python3 ./setup.py clean
+

setup.py

@@ -3,7 +3,7 @@
 from setuptools import setup
 
 setup(name='python-validity',
-      version='0.14',
+      version='0.15',
       py_modules=[],
       packages=['validitysensor'],
       scripts=[

validitysensor/init.py

@@ -1,7 +1,6 @@
 import atexit
 import logging
 
-from validitysensor.init_data_dir import init_data_dir
 from validitysensor.flash import read_tls_flash
 from validitysensor.init_db import init_db
 from validitysensor.init_flash import init_flash
@@ -27,7 +26,6 @@ def close():
 
 
 def open_common():
-    init_data_dir()
     init_flash()
     usb.send_init()
     tls.parse_tls_flash(read_tls_flash())

validitysensor/init_flash.py

@@ -36,24 +36,15 @@
 dbd0df42d534904de00b6389f68867646e9d7c3d0b1dffd74070b2d0f2049b9f1dc7b0c9651c59be3ea891674725e1f2f7a484a941615b80211105978369cf71
 ''')
 
-flash_layout_hardcoded_0090 = [
-    #             id  type  access  offset       size
-    #                       lvl
-    PartitionInfo(1, 4, 7, 0x00001000, 0x00001000),  # cert store
-    PartitionInfo(2, 1, 2, 0x00002000, 0x0003e000),  # xpfwext
-    PartitionInfo(5, 5, 3, 0x00040000, 0x00008000),  # ???
-    PartitionInfo(6, 6, 3, 0x00048000, 0x00008000),  # calibration data
-    PartitionInfo(4, 3, 5, 0x00050000, 0x00030000),  # template database
-]
+crypto_backend = default_backend()
 
-partition_signature_0090 = unhex('''
-e44f7a80d6137794d330b5d026c328a73c907f3f653d411255b7c2f8b425d870a8a53c6630ca864b84590e3c6786f0d69be4bbab5736388f8527237a0a86bbce
-7ced9450c4964709e89ac535aa00787158e0a8d9b1fb75f0f7ae53d4bd11abfcf5ee67a5a71e248a426b3aff4567048fa93de65939ccfbe3f31149a82c64fbfd
-6a2a6cf748e1d9bd8562cf39b1a4b307b37be223317b1b817e364f2877d29d123731314aa627cbf234e0ea69a406a4735a03a45495023ef706bdb542c949d243
-ac2c08c00abf43faa5528a0a8e49b02c507b01b6f1c9abffc669d8c84d7e4a714da32aade7928eca9698b82bee6b72c642c9add80bbd7ccc4121b80220d52b8a
-''')
 
-crypto_backend = default_backend()
+def get_partition_signature():
+    if usb.usb_dev().idVendor == 0x138a:
+        if usb.usb_dev().idProduct == 0x0090:
+            return b''
+
+    return partition_signature
 
 
 def with_hdr(id: int, buf: bytes):
@@ -99,13 +90,13 @@ def serialize_partition(p: PartitionInfo):
     return b
 
 
-def partition_flash(info: FlashInfo, layout: typing.List[PartitionInfo], signature, client_public):
+def partition_flash(info: FlashInfo, layout: typing.List[PartitionInfo], client_public):
     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))
     cmd += with_hdr(1,
-                    b''.join([serialize_partition(p) for p in layout]) + signature)
+                    b''.join([serialize_partition(p) for p in layout]) + get_partition_signature())
     cmd += with_hdr(5, make_cert(client_public))
     cmd += with_hdr(3, crt_hardcoded)
     rsp = tls.cmd(cmd)
@@ -134,15 +125,7 @@ def init_flash():
     client_private = snums.private_value
     client_public = snums.public_numbers
 
-    layout = flash_layout_hardcoded
-    signature = partition_signature
-
-    if usb.usb_dev().idVendor == 0x138a:
-        if usb.usb_dev().idProduct == 0x0090:
-            layout = flash_layout_hardcoded_0090
-            signature = partition_signature_0090
-
-    partition_flash(info, layout, signature, client_public)
+    partition_flash(info, flash_layout_hardcoded, client_public)
 
     RomInfo.get()
     # ^ TODO: use the firmware version which to lookup pubkey for server cert validation

validitysensor/sensor.py

@@ -1,5 +1,6 @@
 import logging
 import os.path
+import time
 import typing
 from binascii import hexlify, unhexlify
 from enum import Enum
@@ -14,14 +15,13 @@
 from .db import db, SidIdentity
 from .flash import write_enable, call_cleanups, read_flash, erase_flash, write_flash_all, read_flash_all
 from .hw_tables import dev_info_lookup
-from .init_data_dir import PYTHON_VALIDITY_DATA_DIR
 from .table_types import SensorTypeInfo, SensorCaptureProg
 from .tls import tls
 from .usb import usb, CancelledException
 from .util import assert_status, unhex
 
 # TODO: this should be specific to an individual device (system may have more than one sensor)
-calib_data_path = PYTHON_VALIDITY_DATA_DIR + 'calib-data.bin'
+calib_data_path = '/usr/share/python-validity/calib-data.bin'
 
 line_update_type1_devices = [
     0xB5, 0x885, 0xB3, 0x143B, 0x1055, 0xE1, 0x8B1, 0xEA, 0xE4, 0xED, 0x1825, 0x1FF5, 0x199
@@ -862,7 +862,7 @@ def parse_dict(self, x: bytes):
 
         return rc
 
-    def match_finger(self) -> typing.Tuple[int, int, bytes]:
+    def match_finger(self) -> typing.Optional[typing.Tuple[int, int, bytes]]:
         try:
             stg_id = 0  # match against any storage
             usr_id = 0  # match against any user
@@ -872,7 +872,8 @@ def match_finger(self) -> typing.Tuple[int, int, bytes]:
 
             b = usb.wait_int()
             if b[0] != 3:
-                raise Exception('Finger not recognized: %s' % hexlify(b).decode())
+                logging.debug('Finger not recognized: %s' % hexlify(b).decode())
+                return None
 
             # get results
             rsp = tls.app(unhexlify('6000000000'))
@@ -881,36 +882,93 @@ def match_finger(self) -> typing.Tuple[int, int, bytes]:
 
             (l, ), rsp = unpack('<H', rsp[:2]), rsp[2:]
             if l != len(rsp):
-                raise Exception('Response size does not match')
+                logging.debug('Response size does not match')
+                return None
 
             rsp = self.parse_dict(rsp)
 
             usrid, subtype, hsh = rsp[1], rsp[3], rsp[4]
             usrid, = unpack('<L', usrid)
             subtype, = unpack('<H', subtype)
 
+
             return usrid, subtype, hsh
+        except Exception as e:
+            logging.debug('Error in match_finger: %s', str(e))
+            return None
         finally:
             # cleanup, ignore any errors
-            tls.app(unhexlify('6200000000'))
+            try:
+                tls.app(unhexlify('6200000000'))
+            except:
+                pass
 
     def identify(self, update_cb: typing.Callable[[Exception], None]):
-        while True:
+        last_error_time = 0
+        error_cooldown = 5  # seconds between error notifications
+
+        try:
+            while True:
+                try:
+                    glow_start_scan()
+                    try:
+                        self.capture(CaptureMode.IDENTIFY)
+                        result = self.match_finger()
+                        if result is not None:
+                            try:
+                                return result
+                            finally:
+                                # Ensure cleanup happens even if return raises an exception
+                                glow_end_scan()
+
+                        # If we get here, the finger wasn't recognized
+                        current_time = time.time()
+                        if current_time - last_error_time > error_cooldown:
+                            update_cb(Exception('Finger not recognized, please try again'))
+                            last_error_time = current_time
+
+                    except usb_core.USBTimeoutError as e:
+                        # Ignore timeouts, just continue scanning
+                        logging.debug('USB timeout during capture, continuing...')
+                        continue
+                    except Exception as e:
+                        # Log other errors but continue scanning
+                        logging.debug(f'Error during capture: {str(e)}')
+                        current_time = time.time()
+                        if current_time - last_error_time > error_cooldown:
+                            update_cb(Exception('Scan error, please try again'))
+                            last_error_time = current_time
+                    finally:
+                        # Always clean up after capture attempt
+                        try:
+                            glow_end_scan()
+                        except Exception as e:
+                            logging.debug(f'Error during scan cleanup: {str(e)}')
+
+                    # Small delay to prevent busy waiting
+                    sleep(0.1)
+
+                except usb_core.USBError as e:
+                    logging.error(f'USB error: {str(e)}')
+                    raise
+                except CancelledException as e:
+                    logging.debug('Scan cancelled by user')
+                    try:
+                        glow_end_scan()
+                    except Exception as e:
+                        logging.debug(f'Error during scan cleanup: {str(e)}')
+                    raise
+                except Exception as e:
+                    logging.error(f'Unexpected error during identification: {str(e)}')
+                    update_cb(e)
+                    sleep(1)
+        except Exception as e:
+            # Final cleanup in case of any unhandled exceptions
             try:
-                glow_start_scan()
-                self.capture(CaptureMode.IDENTIFY)
-                break
-            except usb_core.USBError as e:
-                raise e
-            except CancelledException as e:
                 glow_end_scan()
-                raise e
-            except Exception as e:
-                # Capture failed, retry
-                update_cb(e)
-                sleep(1)
-
-        return self.match_finger()
+            except:
+                pass
+            raise
 
     def get_finger_blobs(self, usrid: int, subtype: int):
         usr = db.get_user(usrid)