A few things.
- automatically locate supported usb device - prototype will call reboot() on exit - enroll() & identify() are now async - dbus service rewriten to use standard dbus.service - use own simplified DBus interface for backend comms
This commit is contained in:
Viktor Dragomiretskyy
parent
e158e9e969
commit
7f14da61a8
8 changed files with 517 additions and 359 deletions
197
dbus-service.py
197
dbus-service.py
|
|
@ -1,69 +1,77 @@
|
|||
|
||||
from threading import Thread
|
||||
from gi.repository import GLib
|
||||
from pydbus import SystemBus
|
||||
from pydbus.generic import signal
|
||||
import pkg_resources
|
||||
from time import sleep
|
||||
from prototype import *
|
||||
from proto9x.db import subtype_to_string
|
||||
from proto9x.sensor import cancel_capture
|
||||
import dbus
|
||||
import dbus.mainloop.glib
|
||||
import dbus.service
|
||||
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
|
||||
from gi.repository import GObject, GLib
|
||||
import logging
|
||||
|
||||
from proto9x import init
|
||||
from proto9x.tls import tls
|
||||
from proto9x.usb import usb
|
||||
from proto9x.sid import sid_from_string
|
||||
from proto9x.db import subtype_to_string, db
|
||||
from proto9x.sensor import sensor, reboot
|
||||
import pwd
|
||||
import atexit
|
||||
import signal
|
||||
|
||||
print("Starting up")
|
||||
GLib.threads_init()
|
||||
|
||||
loop = GLib.MainLoop()
|
||||
INTERFACE_NAME='io.github.uunicorn.Fprint.Device'
|
||||
|
||||
logging.basicConfig(level=logging.DEBUG)
|
||||
|
||||
|
||||
bus = dbus.SystemBus()
|
||||
bus_name = dbus.service.BusName('io.github.uunicorn.Fprint', bus)
|
||||
|
||||
loop = GObject.MainLoop()
|
||||
|
||||
usb.quit = lambda e: loop.quit()
|
||||
|
||||
bus = SystemBus()
|
||||
|
||||
def uid2identity(uid):
|
||||
sidstr='S-1-5-21-111111111-1111111111-1111111111-%d' % uid
|
||||
return sid_from_string(sidstr)
|
||||
|
||||
class AlreadyInUse(Exception):
|
||||
__name__ = 'net.reactivated.Fprint.Error.AlreadyInUse'
|
||||
|
||||
class Device():
|
||||
name = 'Validity sensor'
|
||||
|
||||
class Device(dbus.service.Object):
|
||||
def __init__(self):
|
||||
setattr(self, 'num-enroll-stages', 7)
|
||||
setattr(self, 'scan-type', 'press')
|
||||
self.capturing = False
|
||||
dbus.service.Object.__init__(self, bus_name, '/io/github/uunicorn/Fprint/Device')
|
||||
|
||||
def Claim(self, usr):
|
||||
print('In Claim %s' % usr)
|
||||
|
||||
def Release(self):
|
||||
print('In Release')
|
||||
self.caimed = False
|
||||
|
||||
def ListEnrolledFingers(self, usr, dbus_context):
|
||||
@dbus.service.method(dbus_interface=INTERFACE_NAME,
|
||||
in_signature="s",
|
||||
out_signature="as")
|
||||
def ListEnrolledFingers(self, usr):
|
||||
try:
|
||||
print('In ListEnrolledFingers %s' % usr)
|
||||
|
||||
if len(usr) > 0:
|
||||
pw=pwd.getpwnam(usr)
|
||||
uid=pw.pw_uid
|
||||
else:
|
||||
sender=dbus_context.sender
|
||||
uid=bus.dbus.GetConnectionUnixUser(dbus_context.sender)
|
||||
logging.debug('In ListEnrolledFingers %s' % usr)
|
||||
|
||||
pw=pwd.getpwnam(usr)
|
||||
uid=pw.pw_uid
|
||||
usr=db.lookup_user(uid2identity(uid))
|
||||
|
||||
if usr == None:
|
||||
print('User not found on this device')
|
||||
return []
|
||||
|
||||
rc = [subtype_to_string(f['subtype']) for f in usr.fingers]
|
||||
print(repr(rc))
|
||||
return rc
|
||||
except Exception as e:
|
||||
loop.quit()
|
||||
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)
|
||||
|
||||
def do_scan(self):
|
||||
if self.capturing:
|
||||
return
|
||||
|
|
@ -71,34 +79,37 @@ class Device():
|
|||
try:
|
||||
self.capturing = True
|
||||
z=identify()
|
||||
self.VerifyStatus('verify-match', True)
|
||||
except Exception as e:
|
||||
self.VerifyStatus('verify-no-match', True)
|
||||
#loop.quit();
|
||||
raise e
|
||||
finally:
|
||||
self.capturing = False
|
||||
|
||||
def VerifyStart(self, finger_name):
|
||||
print('In VerifyStart %s' % finger_name)
|
||||
Thread(target=lambda: self.do_scan()).start()
|
||||
#self.do_scan()
|
||||
@dbus.service.method(dbus_interface=INTERFACE_NAME,
|
||||
in_signature='ss',
|
||||
out_signature='')
|
||||
def VerifyStart(self, user, finger):
|
||||
logging.debug('In VerifyStart %s' % finger)
|
||||
|
||||
def VerifyStop(self):
|
||||
print('In VerifyStop')
|
||||
if self.capturing:
|
||||
cancel_capture()
|
||||
def complete_cb(rsp, e):
|
||||
if e is not None:
|
||||
self.VerifyStatus('verify-no-match', True)
|
||||
else:
|
||||
self.VerifyStatus('verify-match', True)
|
||||
usrid, subtype, hsh = rsp
|
||||
# TODO pass down the user DB id
|
||||
# check that a correct finger was identified
|
||||
|
||||
def DeleteEnrolledFingers(self, user):
|
||||
print('In DeleteEnrolledFingers %s' % user)
|
||||
pw=pwd.getpwnam(user)
|
||||
usr=db.lookup_user(uid2identity(pw.pw_uid))
|
||||
def update_cb(e):
|
||||
self.VerifyStatus('verify-retry-scan', False)
|
||||
|
||||
if usr == None:
|
||||
print('User not found on this device')
|
||||
return
|
||||
sensor.identify(update_cb, complete_cb)
|
||||
|
||||
db.del_record(usr.dbid)
|
||||
@dbus.service.method(dbus_interface=INTERFACE_NAME,
|
||||
in_signature='',
|
||||
out_signature='')
|
||||
def Cancel(self):
|
||||
sensor.cancel()
|
||||
|
||||
def do_enroll(self, finger_name, uid):
|
||||
# it is pointless to try and remember username passed in claim as Gnome does not seem to be passing anything useful anyway
|
||||
|
|
@ -112,56 +123,46 @@ class Device():
|
|||
#loop.quit();
|
||||
raise e
|
||||
|
||||
def EnrollStart(self, finger_name, dbus_context):
|
||||
sender=dbus_context.sender
|
||||
uid=bus.dbus.GetConnectionUnixUser(dbus_context.sender)
|
||||
print('In EnrollStart %s for %d' % (finger_name, uid))
|
||||
Thread(target=lambda: self.do_enroll(finger_name, uid)).start()
|
||||
@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 EnrollStop(self):
|
||||
print('In EnrollStop')
|
||||
def complete_cb(rsp, e):
|
||||
if e is not None:
|
||||
self.EnrollStatus('enroll-failed', True)
|
||||
else:
|
||||
self.EnrollStatus('enroll-completed', True)
|
||||
|
||||
VerifyFingerSelected = signal()
|
||||
VerifyStatus = signal()
|
||||
EnrollStatus = signal()
|
||||
sensor.enroll(uid2identity(uid), 0xf5, update_cb, complete_cb) # TODO parse the finger name
|
||||
|
||||
class Manager():
|
||||
def GetDevices(self):
|
||||
print('In GetDevices')
|
||||
return ['/net/reactivated/Fprint/Device/0']
|
||||
@dbus.service.signal(dbus_interface=INTERFACE_NAME, signature='sb')
|
||||
def VerifyStatus(self, result, done):
|
||||
logging.debug('VerifyStatus')
|
||||
|
||||
def GetDefaultDevice(self):
|
||||
print('In GetDefaultDevice')
|
||||
return '/net/reactivated/Fprint/Device/0'
|
||||
@dbus.service.signal(dbus_interface=INTERFACE_NAME, signature='sb')
|
||||
def EnrollStatus(self, result, done):
|
||||
logging.debug('EnrollStatus')
|
||||
|
||||
|
||||
def readif(fn):
|
||||
with open('/usr/share/dbus-1/interfaces/' + fn, 'rb') as f:
|
||||
# for some reason Gio can't seem to handle XML entities declared inline
|
||||
return f.read().decode('utf-8') \
|
||||
.replace('&ERROR_CLAIM_DEVICE;', 'net.reactivated.Fprint.Error.ClaimDevice') \
|
||||
.replace('&ERROR_ALREADY_IN_USE;', 'net.reactivated.Fprint.Error.AlreadyInUse') \
|
||||
.replace('&ERROR_INTERNAL;', 'net.reactivated.Fprint.Error.Internal') \
|
||||
.replace('&ERROR_PERMISSION_DENIED;', 'net.reactivated.Fprint.Error.PermissionDenied') \
|
||||
.replace('&ERROR_NO_ENROLLED_PRINTS;', 'net.reactivated.Fprint.Error.NoEnrolledPrints') \
|
||||
.replace('&ERROR_NO_ACTION_IN_PROGRESS;', 'net.reactivated.Fprint.Error.NoActionInProgress') \
|
||||
.replace('&ERROR_INVALID_FINGERNAME;', 'net.reactivated.Fprint.Error.InvalidFingername') \
|
||||
.replace('&ERROR_NO_SUCH_DEVICE;', 'net.reactivated.Fprint.Error.NoSuchDevice')
|
||||
|
||||
Device.dbus=[readif('net.reactivated.Fprint.Device.xml')]
|
||||
Manager.dbus=[readif('net.reactivated.Fprint.Manager.xml')]
|
||||
|
||||
|
||||
bus.publish('net.reactivated.Fprint',
|
||||
('/net/reactivated/Fprint/Manager', Manager()),
|
||||
('/net/reactivated/Fprint/Device/0', Device())
|
||||
)
|
||||
|
||||
open97()
|
||||
init.open()
|
||||
|
||||
usb.trace_enabled = True
|
||||
tls.trace_enabled = True
|
||||
|
||||
loop.run()
|
||||
svc = Device()
|
||||
try:
|
||||
loop.run()
|
||||
finally:
|
||||
reboot()
|
||||
raise
|
||||
|
||||
print("Normal exit")
|
||||
|
|
|
|||
4
pair.py
4
pair.py
|
|
@ -3,7 +3,7 @@ from time import sleep
|
|||
|
||||
from proto9x.usb import usb
|
||||
from proto9x.tls import tls
|
||||
from proto9x.flash import read_flash
|
||||
from proto9x.flash import read_tls_flash
|
||||
from proto9x.init_flash import init_flash
|
||||
from proto9x.upload_fwext import upload_fwext
|
||||
from proto9x.init_db import init_db
|
||||
|
|
@ -18,7 +18,7 @@ def restart():
|
|||
tls.reset()
|
||||
usb.open()
|
||||
usb.send_init()
|
||||
tls.parseTlsFlash(read_flash(1, 0, 0x1000))
|
||||
tls.parseTlsFlash(read_tls_flash())
|
||||
tls.open()
|
||||
|
||||
usb.open()
|
||||
|
|
|
|||
|
|
@ -130,3 +130,5 @@ def write_fw_signature(partition, signature):
|
|||
rsp=tls.cmd(pack('<BBxH', 0x42, partition, len(signature)) + signature)
|
||||
assert_status(rsp)
|
||||
|
||||
def read_tls_flash():
|
||||
return read_flash(1, 0, 0x1000)
|
||||
|
|
|
|||
16
proto9x/init.py
Normal file
16
proto9x/init.py
Normal file
|
|
@ -0,0 +1,16 @@
|
|||
|
||||
from proto9x.usb import usb
|
||||
from proto9x.tls import tls
|
||||
from proto9x.sensor import sensor
|
||||
from proto9x.flash import read_tls_flash
|
||||
from usb import core as usb_core
|
||||
|
||||
|
||||
def open():
|
||||
usb.open()
|
||||
usb.send_init()
|
||||
|
||||
tls.parseTlsFlash(read_tls_flash())
|
||||
tls.open()
|
||||
|
||||
sensor.open()
|
||||
|
|
@ -21,31 +21,15 @@ debug=False
|
|||
|
||||
line_update_type1_devices = [ 0xB5, 0x885, 0xB3, 0x143B, 0x1055, 0xE1, 0x8B1, 0xEA, 0xE4, 0xED, 0x1825, 0x1FF5, 0x199 ]
|
||||
|
||||
# TODO use more sophisticated glow patters in different cases
|
||||
def glow_start_scan():
|
||||
cmd=unhexlify('3920bf0200ffff0000019900200000000099990000000000000000000000000020000000000000000000000000ffff000000990020000000000000000000000000000000000000002000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000')
|
||||
assert_status(tls.app(cmd))
|
||||
|
||||
def glow_end_enroll():
|
||||
def glow_end_scan():
|
||||
cmd=unhexlify('39f4010000f401000001ff002000000000ffff0000000000000000000000000020000000000000000000000000f401000000ff0020000000000000000000000000000000000000002000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000')
|
||||
assert_status(tls.app(cmd))
|
||||
|
||||
def cancel_capture():
|
||||
usb.queue.put(b'')
|
||||
#sleep(0.2)
|
||||
#rsp=tls.app(b'\x04')
|
||||
#assert_status(rsp)
|
||||
usb.read_82()
|
||||
|
||||
def wait_for_finger():
|
||||
while True:
|
||||
b=usb.wait_int()
|
||||
|
||||
if len(b) == 0:
|
||||
raise Exception('Cancelled')
|
||||
|
||||
if b[0] == 2:
|
||||
break
|
||||
|
||||
def get_prg_status():
|
||||
return tls.app(unhexlify('5100000000'))
|
||||
|
||||
|
|
@ -58,7 +42,7 @@ def wait_till_finished():
|
|||
|
||||
sleep(0.2)
|
||||
|
||||
def stop_prg():
|
||||
def get_prg_status2():
|
||||
return tls.app(unhexlify('5100200000'))
|
||||
|
||||
|
||||
|
|
@ -223,10 +207,14 @@ class CaptureMode(Enum):
|
|||
IDENTIFY=2
|
||||
ENROLL=3
|
||||
|
||||
class CancelledException(Exception):
|
||||
pass
|
||||
|
||||
class Sensor():
|
||||
calib_data=b''
|
||||
|
||||
def open(self, load_calib_data=True):
|
||||
self.interrupt_cb = None
|
||||
self.device_info = identify_sensor()
|
||||
|
||||
print('Opening sensor: %s' % self.device_info.name)
|
||||
|
|
@ -624,244 +612,323 @@ class Sensor():
|
|||
self.persist_clean_slate(clean_slate)
|
||||
self.save()
|
||||
|
||||
sensor = Sensor()
|
||||
def cancel(self):
|
||||
cb=usb.interrupt_cb
|
||||
if cb is not None:
|
||||
cb(None)
|
||||
|
||||
def capture(mode):
|
||||
usb.purge_int_queue()
|
||||
def capture(self, mode, complete_cb):
|
||||
if usb.interrupt_cb is not None:
|
||||
raise Exception('Wrong state for capture')
|
||||
|
||||
assert_status(tls.app(sensor.build_cmd_02(mode)))
|
||||
def next(cb):
|
||||
if cb is None:
|
||||
usb.interrupt_cb = None
|
||||
return
|
||||
|
||||
b=usb.wait_int()
|
||||
if b[0] != 0:
|
||||
raise Exception('Unexpected interrupt type %s' % hexlify(b).decode())
|
||||
def run(b):
|
||||
try:
|
||||
if b is None:
|
||||
raise CancelledException()
|
||||
|
||||
try:
|
||||
wait_for_finger()
|
||||
#wait_till_finished()
|
||||
|
||||
while True:
|
||||
b=usb.wait_int()
|
||||
cb(b)
|
||||
except Exception as e:
|
||||
usb.interrupt_cb = None
|
||||
tls.app(unhexlify('04')) # capture stop if still running, cleanup
|
||||
complete_cb(None, e)
|
||||
|
||||
usb.interrupt_cb = run
|
||||
|
||||
|
||||
def wait_start(b):
|
||||
if b[0] != 0:
|
||||
raise Exception('wait_start: Unexpected interrupt type %s' % hexlify(b).decode())
|
||||
|
||||
next(wait_finger)
|
||||
|
||||
def wait_finger(b):
|
||||
if b[0] == 2:
|
||||
next(wait_capture_complete)
|
||||
|
||||
# TODO: report status?
|
||||
|
||||
def wait_capture_complete(b):
|
||||
if b[0] != 3:
|
||||
raise Exception('Unexpected interrupt type %s' % hexlify(b).decode())
|
||||
raise Exception('wait_finger: Unexpected interrupt type %s' % hexlify(b).decode())
|
||||
|
||||
if b[2] & 4:
|
||||
break
|
||||
finally:
|
||||
res=stop_prg()
|
||||
|
||||
assert_status(res)
|
||||
res = res[2:]
|
||||
|
||||
l, res = res[:4], res[4:]
|
||||
l, = unpack('<L', l)
|
||||
|
||||
if l != len(res):
|
||||
raise Exception('Response size does not match %d != %d', l, len(res))
|
||||
|
||||
x, y, w1, w2, error = unpack('<HHHHL', res)
|
||||
|
||||
return error
|
||||
|
||||
def enrollment_update_start(key=0):
|
||||
rsp=tls.app(pack('<BLL', 0x68, key, 0))
|
||||
assert_status(rsp)
|
||||
new_key, = unpack('<L', rsp[2:])
|
||||
|
||||
usb.wait_int()
|
||||
|
||||
return new_key
|
||||
|
||||
def enrollment_update_end():
|
||||
assert_status(tls.app(pack('<BL', 0x69, 0)))
|
||||
|
||||
def enrollment_update(prev):
|
||||
write_enable()
|
||||
rsp=tls.app(b'\x6b' + prev)
|
||||
assert_status(rsp)
|
||||
flush_changes()
|
||||
|
||||
return rsp[2:]
|
||||
|
||||
def append_new_image(key=0, prev=b''):
|
||||
enrollment_update(prev)
|
||||
|
||||
usb.wait_int()
|
||||
|
||||
res = enrollment_update(prev)
|
||||
|
||||
l, res = res[:2], res[2:]
|
||||
l, = unpack('<H', l)
|
||||
if l != len(res):
|
||||
raise Exception('Response size does not match %d != %d', l, len(res))
|
||||
|
||||
magic_len = 0x38 # hardcoded in the DLL
|
||||
template = header = tid = None
|
||||
|
||||
while len(res) > 0:
|
||||
tag, l = unpack('<HH', res[:4])
|
||||
|
||||
if tag == 0:
|
||||
template = res[:magic_len+l]
|
||||
elif tag == 1:
|
||||
header = res[magic_len:magic_len+l]
|
||||
elif tag == 3:
|
||||
tid = res[magic_len:magic_len+l]
|
||||
else:
|
||||
print('Ignoring unknown tag %x' % tag)
|
||||
capture_complete()
|
||||
return
|
||||
|
||||
res=res[magic_len+l:]
|
||||
# TODO: report status?
|
||||
|
||||
return (header, template, tid)
|
||||
def capture_complete():
|
||||
next(None)
|
||||
|
||||
def make_finger_data(subtype, template, tid):
|
||||
template = pack('<HH', 1, len(template)) + template
|
||||
tid = pack('<HH', 2, len(tid)) + tid
|
||||
res = get_prg_status2()
|
||||
|
||||
tinfo = template + tid
|
||||
assert_status(res)
|
||||
res = res[2:]
|
||||
|
||||
l, res = res[:4], res[4:]
|
||||
l, = unpack('<L', l)
|
||||
|
||||
tinfo = pack('<HHHH', subtype, 3, len(tinfo), 0x20) + tinfo
|
||||
tinfo += b'\0' * 0x20
|
||||
if l != len(res):
|
||||
raise Exception('Response size does not match %d != %d', l, len(res))
|
||||
|
||||
return tinfo
|
||||
x, y, w1, w2, error = unpack('<HHHHL', res)
|
||||
|
||||
def enroll(identity, subtype):
|
||||
key=0
|
||||
template=b''
|
||||
if error != 0:
|
||||
raise Exception('Scanning problem: %04x' % error)
|
||||
|
||||
print('Waiting for a finger...')
|
||||
complete_cb((x, y, w1, w2), None)
|
||||
|
||||
while True:
|
||||
glow_start_scan()
|
||||
next(wait_start)
|
||||
|
||||
try:
|
||||
err = capture(CaptureMode.ENROLL)
|
||||
if err != 0:
|
||||
print('Error %08x, try again' % err)
|
||||
continue
|
||||
except Exception as e:
|
||||
print('Capture failed (%s), try again' % repr(e))
|
||||
sleep(1)
|
||||
continue
|
||||
assert_status(tls.app(self.build_cmd_02(mode)))
|
||||
|
||||
def enrollment_update_start(self, key, complete_cb):
|
||||
if usb.interrupt_cb is not None:
|
||||
raise Exception('Wrong state, sensor\'s busy')
|
||||
|
||||
def wait(b):
|
||||
if b is None:
|
||||
raise Exception('Cancelling while enrollment is being updated is not a good idea.')
|
||||
|
||||
usb.interrupt_cb = None
|
||||
complete_cb(new_key, b)
|
||||
|
||||
usb.interrupt_cb = wait
|
||||
|
||||
rsp=tls.app(pack('<BLL', 0x68, key, 0))
|
||||
assert_status(rsp)
|
||||
new_key, = unpack('<L', rsp[2:])
|
||||
|
||||
def enrollment_update_end(self):
|
||||
assert_status(tls.app(pack('<BL', 0x69, 0)))
|
||||
|
||||
# Generates interrupt
|
||||
def enrollment_update(self, prev):
|
||||
write_enable()
|
||||
rsp=tls.app(b'\x6b' + prev)
|
||||
assert_status(rsp)
|
||||
flush_changes()
|
||||
|
||||
return rsp[2:]
|
||||
|
||||
def append_new_image(self, prev, complete_cb):
|
||||
if usb.interrupt_cb is not None:
|
||||
raise Exception('Wrong state, sensor\'s busy')
|
||||
|
||||
def finished(b):
|
||||
res = self.enrollment_update(prev)
|
||||
|
||||
l, res = res[:2], res[2:]
|
||||
l, = unpack('<H', l)
|
||||
if l != len(res):
|
||||
raise Exception('Response size does not match %d != %d', l, len(res))
|
||||
|
||||
magic_len = 0x38 # hardcoded in the DLL
|
||||
template = header = tid = None
|
||||
|
||||
while len(res) > 0:
|
||||
tag, l = unpack('<HH', res[:4])
|
||||
|
||||
if tag == 0:
|
||||
template = res[:magic_len+l]
|
||||
elif tag == 1:
|
||||
header = res[magic_len:magic_len+l]
|
||||
elif tag == 3:
|
||||
tid = res[magic_len:magic_len+l]
|
||||
else:
|
||||
print('Ignoring unknown tag %x' % tag)
|
||||
|
||||
res=res[magic_len+l:]
|
||||
|
||||
return (header, template, tid)
|
||||
|
||||
def wait(b):
|
||||
if b is None:
|
||||
raise Exception('Cancelling while enrollment is being updated is not a good idea.')
|
||||
|
||||
usb.interrupt_cb = None
|
||||
|
||||
try:
|
||||
complete_cb(finished(b), None)
|
||||
except Exception as e:
|
||||
complete_cb(None, e)
|
||||
|
||||
usb.interrupt_cb = wait
|
||||
|
||||
# Start the work. Interrupt will be generated when it is finished.
|
||||
self.enrollment_update(prev)
|
||||
|
||||
key = enrollment_update_start(key)
|
||||
header, template, tid = append_new_image(key, template)
|
||||
enrollment_update_end()
|
||||
|
||||
print(hexlify(header))
|
||||
def make_finger_data(self, subtype, template, tid):
|
||||
template = pack('<HH', 1, len(template)) + template
|
||||
tid = pack('<HH', 2, len(tid)) + tid
|
||||
|
||||
if tid:
|
||||
break
|
||||
tinfo = template + tid
|
||||
|
||||
# TODO check for duplicates
|
||||
tinfo = pack('<HHHH', subtype, 3, len(tinfo), 0x20) + tinfo
|
||||
tinfo += b'\0' * 0x20
|
||||
|
||||
tinfo = make_finger_data(subtype, template, tid)
|
||||
return tinfo
|
||||
|
||||
usr=db.lookup_user(identity)
|
||||
if usr == None:
|
||||
usr = db.new_user(identity)
|
||||
else:
|
||||
usr = usr.dbid
|
||||
|
||||
recid = db.new_finger(usr, tinfo)
|
||||
def enroll(self, identity, subtype, update_cb, complete_cb):
|
||||
def do_create_finger(final_template, tid):
|
||||
try:
|
||||
tinfo = self.make_finger_data(subtype, final_template, tid)
|
||||
|
||||
glow_end_enroll()
|
||||
usr=db.lookup_user(identity)
|
||||
if usr == None:
|
||||
usr = db.new_user(identity)
|
||||
else:
|
||||
usr = usr.dbid
|
||||
|
||||
recid = db.new_finger(usr, tinfo)
|
||||
|
||||
print('All done')
|
||||
glow_end_scan()
|
||||
|
||||
return recid
|
||||
complete_cb(recid, None)
|
||||
except Exception as e:
|
||||
complete_cb(None, e)
|
||||
|
||||
def parse_dict(x):
|
||||
rc={}
|
||||
|
||||
while len(x) > 0:
|
||||
(t, l), x = unpack('<HH', x[:4]), x[4:]
|
||||
rc[t], x = x[:l], x[l:]
|
||||
def start_iteration(key=0, template=b''):
|
||||
def wrap_cb(f):
|
||||
def r(res, ex):
|
||||
try:
|
||||
f(res, ex)
|
||||
except CancelledException as e:
|
||||
glow_end_scan()
|
||||
complete_cb(None, e)
|
||||
except Exception as e:
|
||||
update_cb(None, e)
|
||||
# sleep, so we don't end up in a busy loop spaming the sensor with requests in case of unrecoverable error
|
||||
sleep(1)
|
||||
self.enrollment_update_end()
|
||||
start_iteration(key, template)
|
||||
|
||||
return rc
|
||||
return r
|
||||
|
||||
|
||||
def identify():
|
||||
glow_start_scan()
|
||||
try:
|
||||
err = capture(CaptureMode.IDENTIFY)
|
||||
if err != 0:
|
||||
raise Exception('Capture failed: %08x' % err)
|
||||
def capture_cb(res, e):
|
||||
if e is not None: raise e
|
||||
def enrollment_update_start_cb(new_key, b):
|
||||
def append_new_image_cb(rsp, e):
|
||||
if e is not None: raise e
|
||||
|
||||
self.enrollment_update_end()
|
||||
|
||||
header, new_template, tid = rsp
|
||||
update_cb(header, None)
|
||||
|
||||
if tid:
|
||||
do_create_finger(new_template, tid)
|
||||
else:
|
||||
start_iteration(new_key, new_template)
|
||||
## end of append_new_image_cb
|
||||
|
||||
self.append_new_image(template, wrap_cb(append_new_image_cb))
|
||||
## end of enrollment_update_start_cb
|
||||
|
||||
self.enrollment_update_start(key, wrap_cb(enrollment_update_start_cb))
|
||||
## end of capture_cb
|
||||
|
||||
glow_start_scan()
|
||||
self.capture(CaptureMode.ENROLL, wrap_cb(capture_cb))
|
||||
|
||||
start_iteration()
|
||||
|
||||
def parse_dict(self, x):
|
||||
rc={}
|
||||
|
||||
while len(x) > 0:
|
||||
(t, l), x = unpack('<HH', x[:4]), x[4:]
|
||||
rc[t], x = x[:l], x[l:]
|
||||
|
||||
return rc
|
||||
|
||||
def match_finger(self, complete_cb):
|
||||
if usb.interrupt_cb is not None:
|
||||
raise Exception('Wrong state for capture')
|
||||
|
||||
def wait(b):
|
||||
if b is None:
|
||||
raise Exception('Cancelling while finger match is running is not a good idea.')
|
||||
|
||||
try:
|
||||
if b[0] != 3:
|
||||
raise Exception('Finger not recognized: %s' % hexlify(b).decode())
|
||||
|
||||
# get results
|
||||
rsp = tls.app(unhexlify('6000000000'))
|
||||
assert_status(rsp)
|
||||
rsp = rsp[2:]
|
||||
|
||||
(l,), rsp = unpack('<H', rsp[:2]), rsp[2:]
|
||||
if l != len(rsp):
|
||||
raise Exception('Response size does not match')
|
||||
|
||||
rsp=self.parse_dict(rsp)
|
||||
|
||||
usrid, subtype, hsh = rsp[1], rsp[3], rsp[4]
|
||||
usrid, = unpack('<L', usrid)
|
||||
subtype, = unpack('<H', subtype)
|
||||
|
||||
complete_cb((usrid, subtype, hsh), None)
|
||||
except Exception as e:
|
||||
complete_cb(None, e)
|
||||
finally:
|
||||
usb.interrupt_cb = None
|
||||
# cleanup, ignore any errors
|
||||
tls.app(unhexlify('6200000000'))
|
||||
|
||||
usb.interrupt_cb = wait
|
||||
|
||||
# which finger?
|
||||
stg_id=0 # match against any storage
|
||||
usr_id=0 # match against any user
|
||||
cmd=pack('<BBBHHHHH', 0x5e, 2, 0xff, stg_id, usr_id, 1, 0,0)
|
||||
rsp=tls.app(cmd)
|
||||
assert_status(rsp)
|
||||
|
||||
b = usb.wait_int()
|
||||
if b[0] != 3:
|
||||
raise Exception('Identification failed: %s' % hexlify(b).decode())
|
||||
|
||||
rsp = tls.app(unhexlify('6000000000'))
|
||||
assert_status(rsp)
|
||||
rsp = rsp[2:]
|
||||
|
||||
finally:
|
||||
# finish
|
||||
assert_status(tls.app(unhexlify('6200000000')))
|
||||
|
||||
(l,), rsp = unpack('<H', rsp[:2]), rsp[2:]
|
||||
if l != len(rsp):
|
||||
raise Exception('Response size does not match')
|
||||
|
||||
rsp=parse_dict(rsp)
|
||||
|
||||
|
||||
#for k in rsp:
|
||||
# print('%04x: %s (%d)' % (k, hexlify(rsp[k]).decode(), len(rsp[k])))
|
||||
|
||||
#on 0097
|
||||
#0001: 09000000 (4)
|
||||
#0003: f500 (2)
|
||||
#0004: 8dee792532d3432d41c872fd4d6d590fbc855ad449cf2753cd919eb9c94675c6 (32)
|
||||
#0005: 0000000000000000000000000000000000000000000000000000000000000000 (32)
|
||||
#0008: 0a00 (2) <-- finger record db id
|
||||
#0002: 010b0000 (4)
|
||||
#0006: 00000000000000000000000000000000000000000000000000000000000000000000000000000000 (40)
|
||||
|
||||
#on 009a (no finger record db id)
|
||||
#0000 8a00
|
||||
# 0100 0400 05000000
|
||||
# 0300 0200 f500
|
||||
# 0400 2000 b147dd1eda8da322fb7a2a51d0eab6fe94bef46c05204fbefb1fd16360903791
|
||||
# 0500 2000 0000000000000000000000000000000000000000000000000000000000000000
|
||||
# 0200 0400 650a0000
|
||||
# 0600 2800 00000000000000000000000000000000000000000000000000000000000000000000000000000000
|
||||
|
||||
usrid, subtype, hsh = rsp[1], rsp[3], rsp[4]
|
||||
usrid, = unpack('<L', usrid)
|
||||
subtype, = unpack('<H', subtype)
|
||||
|
||||
usr = db.get_user(usrid)
|
||||
fingerids = [f['dbid'] for f in usr.fingers if f['subtype'] == subtype]
|
||||
if len(fingerids) != 1:
|
||||
raise Exception('Unexpected matching finger count')
|
||||
finger_record = db.get_record_children(fingerids[0])
|
||||
|
||||
# Device won't let you add more than one data blob
|
||||
if len(finger_record.children) > 1:
|
||||
raise Exception('Expected only one child record for finger')
|
||||
|
||||
print('Recognised finger %02x (%s) from user %s' % (subtype, subtype_to_string(subtype), repr(usr.identity)))
|
||||
print('Template hash: %s' % hexlify(hsh).decode())
|
||||
|
||||
if len(finger_record.children) > 0:
|
||||
if finger_record.children[0]['type'] != 8:
|
||||
raise Exception('Expected data blob as a finger child')
|
||||
|
||||
blob_id = finger_record.children[0]['dbid']
|
||||
blob = db.get_record_value(blob_id).value
|
||||
def identify(self, update_cb, complete_cb):
|
||||
def start():
|
||||
try:
|
||||
glow_start_scan()
|
||||
self.capture(CaptureMode.IDENTIFY, capture_cb)
|
||||
except Exception as e:
|
||||
# failed to start the capture, pointless to retry
|
||||
glow_end_scan()
|
||||
complete_cb(None, e)
|
||||
|
||||
tag, sz = unpack('<HH', blob[:4])
|
||||
val = blob[4:4+sz]
|
||||
def capture_cb(_, e):
|
||||
try:
|
||||
if e is not None: raise e
|
||||
self.match_finger(complete_cb)
|
||||
except CancelledException as e:
|
||||
glow_end_scan()
|
||||
complete_cb(None, e)
|
||||
except Exception as e:
|
||||
# Capture failed, retry
|
||||
update_cb(e)
|
||||
sleep(1)
|
||||
start()
|
||||
|
||||
print('Data blob associated with the finger: %04x: %s' % (tag, hexlify(val).decode()))
|
||||
start()
|
||||
|
||||
def get_finger_blobs(self, usrid, subtype):
|
||||
usr = db.get_user(usrid)
|
||||
fingerids = [f['dbid'] for f in usr.fingers if f['subtype'] == subtype]
|
||||
|
||||
if len(fingerids) != 1:
|
||||
raise Exception('Unexpected matching finger count')
|
||||
|
||||
return rsp
|
||||
finger_record = db.get_record_children(fingerids[0])
|
||||
|
||||
ids=[r['dbid'] for r in finger_record.children if r['type'] == 8]
|
||||
return [db.get_record_value(id).value for id in ids]
|
||||
|
||||
sensor = Sensor()
|
||||
|
||||
|
|
|
|||
|
|
@ -14,11 +14,12 @@ def default_fwext_name():
|
|||
if usb.usb_dev().idVendor == 0x138a:
|
||||
if usb.usb_dev().idProduct == 0x0090:
|
||||
return '6_07f_Lenovo.xpfwext'
|
||||
elif usb.usb_dev().idProduct == 0x0097:
|
||||
return '6_07f_lenovo_mis.xpfwext'
|
||||
elif usb.usb_dev().idVendor == 0x06cb:
|
||||
if usb.usb_dev().idProduct == 0x009a:
|
||||
return '6_07f_lenovo_mis_qm.xpfwext'
|
||||
|
||||
# It looks like the firmware file must match DLL, not the hardware.
|
||||
# Both DLL and xpfwext are universal.
|
||||
# The device dependant code seems to be loaded dynamically (via encrypted blobs).
|
||||
# So, it is important that xpfwext file is matching the blobs contents.
|
||||
return '6_07f_lenovo_mis_qm.xpfwext'
|
||||
|
||||
def upload_fwext(fw_path=None):
|
||||
# no idea what this is:
|
||||
|
|
|
|||
|
|
@ -4,27 +4,36 @@ from binascii import *
|
|||
from .util import assert_status, unhex
|
||||
from struct import unpack
|
||||
from usb.util import claim_interface, release_interface
|
||||
from queue import Queue
|
||||
from threading import Thread
|
||||
from usb.core import USBError
|
||||
from .blobs import init_hardcoded, init_hardcoded_clean_slate
|
||||
|
||||
supported_devices=[
|
||||
(0x138a, 0x0097),
|
||||
(0x06cb, 0x009a),
|
||||
]
|
||||
|
||||
|
||||
def custom_match(d):
|
||||
return (d.idVendor, d.idProduct) in supported_devices
|
||||
|
||||
class Usb():
|
||||
def __init__(self):
|
||||
self.interrupt_cb = None
|
||||
self.trace_enabled = False
|
||||
self.queue = Queue(maxsize=10)
|
||||
self.quit = None
|
||||
self.dev = None
|
||||
|
||||
def purge_int_queue(self):
|
||||
try:
|
||||
while True:
|
||||
self.queue.get_nowait()
|
||||
except:
|
||||
pass
|
||||
def open(self, vendor=None, product=None):
|
||||
if vendor is not None and product is not None:
|
||||
dev = ucore.find(idVendor=vendor, idProduct=product)
|
||||
else:
|
||||
dev = ucore.find(custom_match=custom_match)
|
||||
|
||||
def open(self, vendor=0x06cb, product=0x009a):
|
||||
self.dev = ucore.find(idVendor=vendor, idProduct=product)
|
||||
self.open_dev(dev)
|
||||
|
||||
def open_dev(self, dev):
|
||||
self.dev = dev
|
||||
self.dev.default_timeout = 15000
|
||||
self.thread = Thread(target=lambda: self.int_thread())
|
||||
self.thread.daemon = True
|
||||
|
|
@ -80,7 +89,14 @@ class Usb():
|
|||
resp = self.dev.read(131, 1024, timeout=0)
|
||||
resp = bytes(resp)
|
||||
self.trace('<int< %s' % hexlify(resp).decode())
|
||||
self.queue.put(resp)
|
||||
cb=self.interrupt_cb
|
||||
if cb is None:
|
||||
print('Ignoring spurious interrupt: %s' % hexlify(resp).decode())
|
||||
else:
|
||||
try:
|
||||
cb(resp)
|
||||
except Exception as e:
|
||||
print('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:
|
||||
|
|
@ -89,10 +105,6 @@ class Usb():
|
|||
self.trace('<int< Interrupt thread is dead')
|
||||
|
||||
|
||||
def wait_int(self):
|
||||
resp = self.queue.get()
|
||||
return resp
|
||||
|
||||
def trace(self, s):
|
||||
if self.trace_enabled:
|
||||
print(s)
|
||||
|
|
|
|||
101
prototype.py
101
prototype.py
|
|
@ -5,33 +5,92 @@ from proto9x.db import db
|
|||
from proto9x.flash import read_flash
|
||||
from proto9x.sensor import *
|
||||
from proto9x.sid import *
|
||||
from proto9x.init import open as open9x
|
||||
from threading import Condition
|
||||
from time import sleep
|
||||
import code
|
||||
|
||||
def open_common():
|
||||
usb.send_init()
|
||||
#usb.trace_enabled = True
|
||||
#tls.trace_enabled = True
|
||||
|
||||
# try to init TLS session from the flash
|
||||
tls.parseTlsFlash(read_flash(1, 0, 0x1000))
|
||||
def identify():
|
||||
cv=Condition()
|
||||
result=[]
|
||||
|
||||
tls.open()
|
||||
tls.save()
|
||||
sensor.open()
|
||||
#usb.trace_enabled = True
|
||||
#tls.trace_enabled = True
|
||||
def update_cb(e):
|
||||
print('Capture error: %s, try again' % repr(e))
|
||||
|
||||
def open97():
|
||||
usb.open(vendor=0x138a, product=0x0097)
|
||||
open_common()
|
||||
def complete_cb(rsp, e):
|
||||
cv.acquire()
|
||||
try:
|
||||
result.append((rsp, e))
|
||||
cv.notify()
|
||||
finally:
|
||||
cv.release()
|
||||
|
||||
def open9a():
|
||||
usb.open(vendor=0x06cb, product=0x009a)
|
||||
open_common()
|
||||
sensor.identify(update_cb, complete_cb)
|
||||
|
||||
def load97():
|
||||
#usb.trace_enabled = True
|
||||
#tls.trace_enabled = True
|
||||
usb.open()
|
||||
tls.load()
|
||||
sensor.open()
|
||||
cv.acquire()
|
||||
try:
|
||||
cv.wait()
|
||||
rsp, e = result[0]
|
||||
if e is not None: raise e
|
||||
|
||||
except:
|
||||
sensor.cancel()
|
||||
raise
|
||||
finally:
|
||||
cv.release()
|
||||
|
||||
usrid, subtype, hsh = rsp
|
||||
|
||||
print('Got finger %x for user recordid %d. Hash: %s' % (subtype, usrid, hexlify(hsh).decode()))
|
||||
|
||||
def enroll(sid, finger):
|
||||
cv=Condition()
|
||||
result=[]
|
||||
|
||||
def update_cb(x, e):
|
||||
if e is not None:
|
||||
print('Enroll error: %s, try again' % repr(e))
|
||||
else:
|
||||
print('Enroll progress: %s' % hexlify(x).decode())
|
||||
|
||||
def complete_cb(rsp, e):
|
||||
cv.acquire()
|
||||
try:
|
||||
result.append((rsp, e))
|
||||
cv.notify()
|
||||
finally:
|
||||
cv.release()
|
||||
|
||||
sensor.enroll(sid, finger, update_cb, complete_cb)
|
||||
|
||||
cv.acquire()
|
||||
try:
|
||||
cv.wait()
|
||||
recid, e = result[0]
|
||||
if e is not None: raise e
|
||||
|
||||
except:
|
||||
sensor.cancel()
|
||||
raise
|
||||
finally:
|
||||
cv.release()
|
||||
|
||||
print('Created a finger record with dbid %d' % recid)
|
||||
|
||||
# can't use atexit as it conflicts with atexit installed by libusb
|
||||
class Blah:
|
||||
def __init__(self):
|
||||
self.tls=tls
|
||||
|
||||
def __del__(self):
|
||||
if usb.dev is not None:
|
||||
print('Rebooting device...')
|
||||
try:
|
||||
reboot()
|
||||
except:
|
||||
pass
|
||||
|
||||
blah=Blah()
|
||||
|
|
|
|||
Loading…
Reference in a new issue