Merge pull request #20 from VorpalBlade/feature/sid_mapping

Config file for SID mapping.

.gitignore

@@ -1,16 +1,17 @@
-debian/tmp/
-debian/debhelper-build-stamp
-debian/python3-validity/
-debian/python3-validity.substvars
-debian/files
 debian/*.debhelper
 debian/.debhelper/
+debian/debhelper-build-stamp
+debian/files
+debian/python3-validity.substvars
+debian/python3-validity/
+debian/tmp/
 
-__pycache__
+*.bin
+*.log
+.idea/*
 .pybuild/
+__pycache__
 build
 dist
 python_validity.egg-info/
 tls.dict
-*.bin
-*.log

README.md

@@ -57,6 +57,24 @@ auth  [success=2 default=ignore]  pam_fprintd.so max_tries=1 timeout=10 # debug
 auth  [success=1 default=ignore]  pam_unix.so nullok_secure try_first_pass
 ```
 
+## Windows interoperability
+
+Note: This section is likely only relevant if you will be dual booting.
+
+To be able to use the same set of fingerprints for Windows and Linux, you first
+need to extract the Windows user IDs (known as SIDs). To do this, start Windows,
+start `cmd.exe` and run `wmic useraccount get name,sid`. This will provide a
+list of all users and the corresponding SIDs.
+
+You can then create a mapping from the Linux user names (as written in the
+first `:`-separated field of `/etc/passwd`). This mapping is defined in
+`/etc/python-validity/dbus-service.yaml`. For example:
+```yaml
+user_to_sid:
+    "myusername": "S-1-5-21-1234567890-1234567890-1234567890-1001"
+    "someotheruser": "S-1-5-21-1234567890-1234567890-1234567890-1003"
+```
+
 ## Playground
 
 This package contains a set of scripts you can use to do a low-level debugging of the sensor protocol.

dbus_service/dbus-service

@@ -1,18 +1,23 @@
 #!/usr/bin/env python3
 
-import signal
 import argparse
-import re
 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 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
@@ -22,43 +27,54 @@ 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
+from validitysensor.db import subtype_to_string, db, SidIdentity, User
+from validitysensor.sensor import sensor, RebootException
 
 GLib.threads_init()
 
-INTERFACE_NAME='io.github.uunicorn.Fprint.Device'
+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):
+    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, usr):
+                         in_signature="s",
+                         out_signature="as")
+    def ListEnrolledFingers(self, user):
         try:
-            logging.debug('In ListEnrolledFingers %s' % usr)
+            logging.debug('In ListEnrolledFingers %s' % user)
 
-            pw=pwd.getpwnam(usr)
-            uid=pw.pw_uid
-            usr=db.lookup_user(uid2identity(uid))
+            usr = self.user2record(user)
 
-            if usr == None:
+            if usr is None:
                 return []
-            
+
             rc = [subtype_to_string(f['subtype']) for f in usr.fingers]
             logging.debug(repr(rc))
             return rc
@@ -66,14 +82,13 @@ class Device(dbus.service.Object):
             raise e
 
     @dbus.service.method(dbus_interface=INTERFACE_NAME,
-                     in_signature='s',
-                     out_signature='')
+                         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))
+        usr = self.user2record(user)
 
-        if usr == None:
+        if usr is None:
             return
 
         db.del_record(usr.dbid)
@@ -84,10 +99,9 @@ class Device(dbus.service.Object):
     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))
+        usr = self.user2record(user)
 
-        if usr == None:
+        if usr is None:
             raise NoEnrolledPrints()
 
         self.VerifyFingerSelected('any')
@@ -114,7 +128,6 @@ class Device(dbus.service.Object):
         thread = Thread(target=run)
         thread.daemon = True
         thread.start()
-        
 
     @dbus.service.method(dbus_interface=INTERFACE_NAME,
                          in_signature='',
@@ -127,8 +140,9 @@ class Device(dbus.service.Object):
                          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
+
+        usr = self.user2identity(user)
+
         def update_cb(rsp, e):
             if e is not None:
                 self.EnrollStatus('enroll-retry-scan', False)
@@ -137,7 +151,7 @@ class Device(dbus.service.Object):
 
         def run():
             try:
-                sensor.enroll(uid2identity(uid), 0xf5, update_cb) # TODO parse the finger name
+                sensor.enroll(usr, 0xf5, update_cb)  # TODO parse the finger name
                 self.EnrollStatus('enroll-completed', True)
             except usb_core.USBError as e:
                 logging.exception(e)
@@ -150,7 +164,6 @@ class Device(dbus.service.Object):
         thread = Thread(target=run)
         thread.daemon = True
         thread.start()
-        
 
     @dbus.service.signal(dbus_interface=INTERFACE_NAME, signature='sb')
     def VerifyStatus(self, result, done):
@@ -171,7 +184,9 @@ class Device(dbus.service.Object):
         logging.debug('RunCmd')
         return hexlify(tls.app(unhexlify(cmd))).decode()
 
-backoff_file='/usr/share/python-validity/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!
 def backoff():
@@ -179,10 +194,10 @@ def backoff():
         with open(backoff_file, 'r') as f:
             lines = list(map(float, f.readlines()))
     else:
-        lines=[]
+        lines = []
 
     while len(lines) > 0 and lines[0] + 60 < time.time():
-        lines=lines[1:]
+        lines = lines[1:]
 
     lines += [time.time()]
 
@@ -193,10 +208,12 @@ def backoff():
     if len(lines) > 10:
         raise Exception('Refusing to start more than 10 times per minute')
 
-if __name__ == '__main__':
+
+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:
@@ -206,16 +223,34 @@ if __name__ == '__main__':
     else:
         level = logging.INFO
 
-    handler = logging.handlers.SysLogHandler(address = '/dev/log')
+    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('^usb-(\d+)-(\d+)$', args.devpath)
+            z = re.match(r'^usb-(\d+)-(\d+)$', args.devpath)
             if not z:
                 parser.error('Option --devpath should look like this: usb-<busnum>-<address>')
 
@@ -226,9 +261,8 @@ if __name__ == '__main__':
 
     bus = dbus.SystemBus()
 
-    svc = Device(bus)
+    svc = Device(bus, config)
 
-    watcher = None
     def watch_cb(name):
         if name == '':
             logging.debug('Manager is offline')
@@ -237,13 +271,14 @@ if __name__ == '__main__':
             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)
+        logging.info('Caught signal %d. Stopping...' % x)
         loop.quit()
 
     GLib.unix_signal_add(GLib.PRIORITY_DEFAULT, signal.SIGINT, die, signal.SIGINT)
@@ -254,3 +289,6 @@ if __name__ == '__main__':
         logging.exception(e)
         raise e
 
+
+if __name__ == '__main__':
+    main()

debian/control

@@ -15,6 +15,7 @@ Depends: ${python3:Depends},
  ${misc:Depends},
  python3-dbus,
  python3-usb,
+ python3-yaml,
  dbus,
  open-fprintd (>= 0.2~),
  innoextract (>= 1.6~)

debian/install

@@ -4,3 +4,4 @@ usr/lib/python*/*-packages/validitysensor/*.py
 usr/lib/python-validity
 usr/share/dbus-1
 usr/share/python-validity
+etc/python-validity

debian/rules

@@ -12,4 +12,3 @@ override_dh_installsystemd:
 
 override_dh_auto_install:
 	python3 ./setup.py install --root=$(CURDIR)/debian/tmp --prefix=/usr --install-layout=deb
-

etc/python-validity/dbus-service.yaml

@@ -0,0 +1,8 @@
+# Mapping for users
+# Add mappings from a user to a SID to use the same fingerprints as enrolled by
+# Windows. By default, mappings will be generated from the numerical UID if not
+# found here. This is fine if you are not dual booting and want to share
+# mappings with Windows.
+user_to_sid:
+    # Example:
+    #"username": "S-1-5-21-1234567890-1234567890-1234567890-1234"

setup.py

@@ -12,7 +12,8 @@ setup(name='python-validity',
     ],
     install_requires=[
         'cryptography >= 2.1.4',
-        'pyusb >= 1.0.2'
+        'pyusb >= 1.0.0',
+        'pyyaml >= 3.12'
     ],
     data_files=[
         ('share/dbus-1/system.d/', ['dbus_service/io.github.uunicorn.Fprint.conf']),

validitysensor/db.py

@@ -1,10 +1,8 @@
+import typing
 
-import logging
-from .util import unhex
 from .tls import tls
 from .util import assert_status
-from struct import pack, unpack
-from binascii import hexlify, unhexlify
+from binascii import hexlify
 from .blobs import db_write_enable
 from .flash import call_cleanups
 from .sid import *
@@ -27,7 +25,7 @@ class User():
     def __repr__(self):
         return '<User: dbid=%04x identity=%s fingers=%s>' % (self.dbid, repr(self.identity), repr(self.fingers))
 
-def subtype_to_string(s):
+def subtype_to_string(s: int):
     if s < 0xf5 or s > 0xfe:
         return 'Unknown'
 
@@ -70,7 +68,7 @@ def parse_identity(b):
 
     raise Exception('Don''t know how to handle identity type %d' % t)
 
-def parse_user(rsp):
+def parse_user(rsp: bytes):
     assert_status(rsp[:2])
     rsp=rsp[2:]
 
@@ -92,7 +90,7 @@ def parse_user(rsp):
 
     return user
 
-def identity_to_bytes(identity):
+def identity_to_bytes(identity: str):
     if isinstance(identity, SidIdentity):
         b=identity.to_bytes()
         b = pack('<LL', 3, len(b)) + b
@@ -156,7 +154,7 @@ class Db():
     def get_user(self, dbid):
         return parse_user(tls.cmd(pack('<BHHH', 0x4a, dbid, 0, 0)))
 
-    def lookup_user(self, identity):
+    def lookup_user(self, identity: str) -> typing.Optional[User]:
         stg = self.get_user_storage(name='StgWindsor')
         data = identity_to_bytes(identity)
         

validitysensor/flash.py

@@ -1,7 +1,5 @@
-
 from .tls import tls
 from struct import pack, unpack
-from binascii import hexlify, unhexlify
 from .util import assert_status, unhex
 from .blobs import db_write_enable
 from .hw_tables import flash_ic_table_lookup

validitysensor/init.py

@@ -1,6 +1,5 @@
 
 import atexit
-import signal
 import logging
 
 from validitysensor.usb import usb

validitysensor/init_db.py

@@ -3,9 +3,6 @@ from struct import pack, unpack
 import logging
 
 from .db import db
-from .usb import usb
-from .tls import tls
-from .flash import read_flash
 
 def machine_id_rec_value(b):
     b = b.encode('utf-16le')

validitysensor/init_flash.py

@@ -1,7 +1,7 @@
 
 import os
 from struct import pack, unpack
-from binascii import hexlify, unhexlify
+from binascii import unhexlify
 import logging
 
 from hashlib import sha256

validitysensor/sensor.py

@@ -1,4 +1,4 @@
-
+import typing
 from enum import Enum
 from hashlib import sha256
 import os.path
@@ -6,7 +6,7 @@ import logging
 from usb import core as usb_core
 from .tls import tls
 from .usb import usb, CancelledException
-from .db import db, subtype_to_string
+from .db import db, SidIdentity
 from .flash import write_enable, call_cleanups, read_flash, erase_flash, write_flash_all, read_flash_all
 from time import sleep
 from struct import pack, unpack
@@ -70,11 +70,12 @@ def factory_reset():
     reboot()
 
 class RomInfo():
-    def get():
+    @classmethod
+    def get(cls):
         rsp=tls.cmd(b'\x01')
         assert_status(rsp)
         rsp=rsp[2:]
-        return RomInfo(*unpack('<LLBBxBxxxB', rsp[0:0x10]))
+        return cls(*unpack('<LLBBxBxxxB', rsp[0:0x10]))
 
     def __init__(self, timestamp, build, major, minor, product, u1):
         self.timestamp, self.build, self.major, self.minor, self.product, self.u1 = timestamp, build, major, minor, product, u1
@@ -739,7 +740,9 @@ class Sensor():
 
         return tinfo
 
-    def enroll(self, identity, subtype, update_cb):
+    # TODO: Better typing information needed.
+    def enroll(self, identity: SidIdentity, subtype: int,
+               update_cb: typing.Callable[[typing.Any, typing.Optional[Exception]], None]):
         def do_create_finger(final_template, tid):
             tinfo = self.make_finger_data(subtype, final_template, tid)
 
@@ -794,7 +797,7 @@ class Sensor():
 
         return rc
 
-    def match_finger(self):
+    def match_finger(self) -> typing.Tuple[int, int, bytes]:
         try:
             stg_id=0 # match against any storage
             usr_id=0 # match against any user

validitysensor/sid.py

@@ -1,8 +1,9 @@
 
 from struct import unpack, pack
+import typing
 
 class SidIdentity():
-    def __init__(self, revision, auth, subauth):
+    def __init__(self, revision: int, auth: int, subauth: typing.Sequence[int]):
         self.revision = revision
         self.auth = auth
         self.subauth = subauth
@@ -17,7 +18,7 @@ class SidIdentity():
     def __repr__(self):
         return 'S-%d-%d-%s' % (self.revision, self.auth, '-'.join(map(str, self.subauth)))
 
-def sid_from_bytes(b):
+def sid_from_bytes(b: bytes):
     revision = b[0]
     subcnt = b[1]
     auth = 0
@@ -30,7 +31,7 @@ def sid_from_bytes(b):
 
     return SidIdentity(revision, auth, subauth)
 
-def sid_from_string(s):
+def sid_from_string(s: str):
     parts = s.split('-')
 
     if parts[0] != 'S':

validitysensor/table_types.py

@@ -4,9 +4,10 @@ from binascii import hexlify, unhexlify
 class SensorTypeInfo:
     table=[]
 
-    def get_by_type(sensor_type):
+    @classmethod
+    def get_by_type(cls, sensor_type):
         from . import generated_tables
-        for i in SensorTypeInfo.table:
+        for i in cls.table:
             if i.sensor_type == sensor_type:
                 return i
 
@@ -46,7 +47,8 @@ def metric(i, rominfo):
 class SensorCaptureProg:
     table=[]
 
-    def get(rominfo, sensor_type, a0, a1):
+    @classmethod
+    def get(cls, rominfo, sensor_type, a0, a1):
         from . import generated_tables
 
         maximum = 0

validitysensor/tls.py

@@ -1,6 +1,4 @@
-import re
 import hmac
-import sys
 import os
 import pickle
 import logging
@@ -14,8 +12,8 @@ from cryptography.hazmat.primitives.asymmetric.utils import Prehashed
 from cryptography.hazmat.primitives import hashes
 from hashlib import sha256
 
-from .usb import unhex, usb
-from .util import assert_status
+from .util import unhex
+from .usb import usb
 
 
 password_hardcoded=unhexlify('717cd72d0962bc4a2846138dbb2c24192512a76407065f383846139d4bec2033')
@@ -289,7 +287,7 @@ class Tls():
 
     def handle_server_hello_done(self, p):
         if p != b'':
-            raise Exeception('Not expecting any body for "server hello done" pkt: %s' % hexlify(p).decode())
+            raise Exception('Not expecting any body for "server hello done" pkt: %s' % hexlify(p).decode())
 
     def handle_finish(self, b):
         hs_hash = self.handshake_hash.copy().digest()

validitysensor/upload_fwext.py

@@ -1,14 +1,10 @@
-
-from binascii import hexlify, unhexlify
 from time import ctime
 import logging
 from os.path import basename
 
-from .tls import tls
 from .usb import usb
 from .sensor import reboot, write_hw_reg32, read_hw_reg32, identify_sensor
-from .flash import read_flash, erase_flash, write_flash_all, write_fw_signature, get_fw_info
-from .util import assert_status
+from .flash import write_flash_all, write_fw_signature, get_fw_info
 
 firmware_home='/usr/share/python-validity'
 

validitysensor/usb.py

@@ -3,10 +3,8 @@ import logging
 import errno
 import usb.core as ucore
 from binascii import *
-from .util import assert_status, unhex
+from .util import assert_status
 from struct import unpack
-from usb.util import claim_interface, release_interface
-from threading import Thread
 from usb.core import USBError
 from .blobs import init_hardcoded, init_hardcoded_clean_slate