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

* Roll back async changes for enroll & identify.

Browse source 
* Add a couple of tweaks to stop the sensor from crashing.
* A few more comments and TODOs
This commit is contained in:
Viktor Dragomiretskyy 2020-07-16 23:13:46 +12:00
parent 229fffe14c
commit 75c6ddb5b3
9 changed files with 263 additions and 332 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

50
dbus_service/dbus-service
View file

@ -10,6 +10,7 @@ import dbus
import dbus.mainloop.glib
import dbus.service
from binascii import hexlify, unhexlify
from threading import Thread
dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
from gi.repository import GLib
import logging
@ -92,19 +93,21 @@ class Device(dbus.service.Object):
def VerifyStart(self, user, finger):
logging.debug('In VerifyStart %s' % finger)
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 update_cb(e):
self.VerifyStatus('verify-retry-scan', False)
sensor.identify(update_cb, complete_cb)
def run():
try:
usrid, subtype, hsh = sensor.identify(update_cb)
self.VerifyStatus('verify-match', True)
except Exception as e:
logging.exception('Oops', 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='',
@ -112,18 +115,6 @@ class Device(dbus.service.Object):
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
try:
# TODO hardcode the username and finger for now
z=enroll(uid2identity(uid), 0xf5)
logging.debug('Enroll was successfull')
self.EnrollStatus('enroll-completed', True)
except Exception as e:
self.EnrollStatus('enroll-failed', True)
#loop.quit();
raise e
@dbus.service.method(dbus_interface=INTERFACE_NAME,
in_signature='ss',
out_signature='')
@ -137,13 +128,18 @@ class Device(dbus.service.Object):
else:
self.EnrollStatus('enroll-stage-passed', False)
def complete_cb(rsp, e):
if e is not None:
self.EnrollStatus('enroll-failed', True)
else:
def run():
try:
sensor.enroll(uid2identity(uid), 0xf5, update_cb) # TODO parse the finger name
self.EnrollStatus('enroll-completed', True)
except Exception as e:
logging.exception('Oops', e)
self.EnrollStatus('enroll-failed', True)
sensor.enroll(uid2identity(uid), 0xf5, update_cb, complete_cb) # TODO parse the finger name
thread = Thread(target=run)
thread.daemon = True
thread.start()
@dbus.service.signal(dbus_interface=INTERFACE_NAME, signature='sb')
def VerifyStatus(self, result, done):

52
scripts/prototype.py
View file

@ -14,69 +14,21 @@ import code
#tls.trace_enabled = True
def identify():
cv=Condition()
result=[]
def update_cb(e):
print('Capture error: %s, try again' % repr(e))
def complete_cb(rsp, e):
cv.acquire()
try:
result.append((rsp, e))
cv.notify()
finally:
cv.release()
sensor.identify(update_cb, complete_cb)
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
usrid, subtype, hsh = sensor.identify(update_cb)
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()
recid = sensor.enroll(sid, finger, update_cb)
print('Created a finger record with dbid %d' % recid)

50
validitysensor/db.py
View file

@ -6,7 +6,7 @@ from .util import assert_status
from struct import pack, unpack
from binascii import hexlify, unhexlify
from .blobs import db_write_enable
from .flash import flush_changes
from .flash import call_cleanups
from .sid import *
class UserStorage():
@ -125,6 +125,18 @@ class DbRecord():
class Db():
class Info():
def __init__(self, total, used, free, records, roots):
self.total = total # partition size
self.used = used # used (not deleted)
self.free = free # unallocated space
self.records = records # total number, including deleted
self.roots = roots
def __repr__(self):
return 'Db.Info(total=%d, used=%d, free=%d, records=%d, roots=%s)' % (
self.total, self.used, self.free, self.records, repr(self.roots))
def get_user_storage(self, dbid=0, name=''):
name=name.encode()
@ -175,7 +187,7 @@ class Db():
rsp = rsp[14:]
rec.children=[]
for i in range(0, cnt):
dbid, typ = unpack('<HH', rsp[i:i+4])
dbid, typ = unpack('<HH', rsp[i*4:i*4+4])
rec.children += [{ 'dbid': dbid, 'type': typ }]
return rec
@ -183,14 +195,27 @@ class Db():
def del_record(self, dbid):
assert_status(tls.cmd(pack('<BH', 0x48, dbid)))
def db_info(self):
rsp = tls.cmd(b'\x45')
assert_status(rsp)
rsp = rsp[2:]
unknown1, unknown0, total, used, free, records, nroots = unpack('<LLLLLHH', rsp[:0x18])
# Seems to always be unknown1 == 1, unknown0 == 0
rsp = rsp[0x18:]
roots = [unpack('<H', rsp[i*2:i*2+2])[0] for i in range(0, nroots)]
return Db.Info(total, used, free, records, roots)
def new_record(self, parent, typ, storage, data):
assert_status(tls.cmd(b'\x45'))
self.db_info() # TODO check free space, compact the partition when out of storage
assert_status(tls.cmd(db_write_enable))
rsp = tls.cmd(pack('<BHHHH', 0x47, parent, typ, storage, len(data)) + data)
assert_status(rsp)
recid, = unpack('<H', rsp[2:])
flush_changes()
try:
rsp = tls.cmd(pack('<BHHHH', 0x47, parent, typ, storage, len(data)) + data)
assert_status(rsp)
recid, = unpack('<H', rsp[2:])
finally:
call_cleanups()
return recid
def new_user(self, identity):
@ -213,6 +238,17 @@ class Db():
rec = self.new_record(parent, 0x8, stg.dbid, data)
return rec
def dump_raw(self, root=3, depth=0):
rec = self.get_record_value(root)
val = hexlify(rec.value).decode()
if len(val) > 80:
val = val[:80] + '...'
print('%s%d (type %d) %s' % (' '*depth, rec.dbid, rec.type, val))
rec = self.get_record_children(root)
for c in rec.children:
self.dump_raw(c['dbid'], depth+1)
def dump_all(self):
stg = self.get_user_storage(name='StgWindsor')
usrs = [self.get_user(u['dbid']) for u in stg.users]

22
validitysensor/flash.py
View file

@ -91,13 +91,19 @@ def get_fw_info(partition):
def write_enable():
assert_status(tls.cmd(db_write_enable))
def flush_changes():
assert_status(tls.cmd(b'\x1a'))
def call_cleanups():
rsp = tls.cmd(b'\x1a')
err = unpack('<H', rsp[:2])[0]
if err == 0x0491: # Nothing to commit
return # don't throw an exception, just ignore
assert_status(rsp)
def erase_flash(partition):
assert_status(tls.cmd(db_write_enable))
assert_status(tls.cmd(pack('<BB', 0x3f, partition)))
flush_changes()
try:
assert_status(tls.cmd(pack('<BB', 0x3f, partition)))
finally:
call_cleanups()
def read_flash(partition, addr, size):
cmd = pack('<BBBHLL', 0x40, partition, 1, 0, addr, size)
@ -110,9 +116,11 @@ def read_flash(partition, addr, size):
def write_flash(partition, addr, buf):
tls.cmd(db_write_enable)
cmd = pack('<BBBHLL', 0x41, partition, 1, 0, addr, len(buf)) + buf
rsp = tls.cmd(cmd)
assert_status(rsp)
flush_changes()
try:
rsp = tls.cmd(cmd)
assert_status(rsp)
finally:
call_cleanups()
def write_flash_all(partition, ptr, buf):
bs = 0x1000

17
validitysensor/init_flash.py
View file

@ -13,8 +13,8 @@ from cryptography.hazmat.backends import default_backend
from .tls import tls, hs_key, crt_hardcoded
from .usb import usb
from .flash import write_flash, erase_flash, flush_changes, PartitionInfo, get_flash_info
from .sensor import reboot
from .flash import write_flash, erase_flash, call_cleanups, PartitionInfo, get_flash_info
from .sensor import reboot, RomInfo
from .util import assert_status, unhex
from .blobs import reset_blob
@ -121,13 +121,14 @@ def init_flash():
partition_flash(info, flash_layout_hardcoded, client_public)
rsp=usb.cmd(unhex('01'))
assert_status(rsp)
# ^ get device info, contains firmware version which is needed to lookup pubkey for server cert validation
RomInfo.get()
# ^ TODO: use the firmware version which to lookup pubkey for server cert validation
rsp=usb.cmd(unhex('50'))
assert_status(rsp)
flush_changes()
try:
rsp=usb.cmd(unhex('50'))
assert_status(rsp)
finally:
call_cleanups()
rsp=rsp[2:]
l,=unpack('<L', rsp[:4])

363
validitysensor/sensor.py
View file

@ -3,10 +3,11 @@ from enum import Enum
from hashlib import sha256
import os.path
import logging
from usb import core as usb_core
from .tls import tls
from .usb import usb
from .usb import usb, CancelledException
from .db import db, subtype_to_string
from .flash import write_enable, flush_changes, read_flash, erase_flash, write_flash_all, read_flash_all
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
from .table_types import SensorTypeInfo, SensorCaptureProg
@ -190,14 +191,10 @@ class CaptureMode(Enum):
IDENTIFY=2
ENROLL=3
class CancelledException(Exception):
pass
class Sensor():
calib_data=b''
def open(self):
self.interrupt_cb = None
self.device_info = identify_sensor()
logging.info('Opening sensor: %s' % self.device_info.name)
@ -625,58 +622,32 @@ class Sensor():
self.save()
def cancel(self):
cb=usb.interrupt_cb
if cb is not None:
cb(None)
usb.cancel = True
def capture(self, mode, complete_cb):
if usb.interrupt_cb is not None:
raise Exception('Wrong state for capture')
def capture(self, mode):
try:
assert_status(tls.app(self.build_cmd_02(mode)))
def next(cb):
if cb is None:
usb.interrupt_cb = None
return
def run(b):
try:
if b is None:
raise CancelledException()
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):
# start
b = usb.wait_int()
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('wait_finger: Unexpected interrupt type %s' % hexlify(b).decode())
if b[2] & 4:
capture_complete()
return
# wait for finger
while True:
b = usb.wait_int()
if b[0] == 2:
break;
# TODO: report status?
def capture_complete():
next(None)
# wait capture complete
while True:
b = usb.wait_int()
if b[0] != 3:
raise Exception('Unexpected interrupt type %s' % hexlify(b).decode())
if b[2] & 4:
break
res = get_prg_status2()
assert_status(res)
@ -693,88 +664,69 @@ class Sensor():
if error != 0:
raise Exception('Scanning problem: %04x' % error)
complete_cb((x, y, w1, w2), None)
return (x, y, w1, w2)
next(wait_start)
finally:
tls.app(unhexlify('04')) # capture stop if still running, cleanup
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
def enrollment_update_start(self, key):
rsp=tls.app(pack('<BLL', 0x68, key, 0))
assert_status(rsp)
new_key, = unpack('<L', rsp[2:])
usb.wait_int()
return new_key
def create_enrollment(self):
assert_status(tls.app(pack('<BL', 0x69, 1)))
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()
try:
rsp=tls.app(b'\x6b' + prev)
assert_status(rsp)
finally:
call_cleanups()
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:
logging.warning('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.
def append_new_image(self, prev):
self.enrollment_update(prev)
usb.wait_int()
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:
logging.warning('Ignoring unknown tag %x' % tag)
res=res[magic_len+l:]
return (header, template, tid)
def make_finger_data(self, subtype, template, tid):
template = pack('<HH', 1, len(template)) + template
@ -787,70 +739,51 @@ class Sensor():
return tinfo
def enroll(self, identity, subtype, update_cb, complete_cb):
def enroll(self, identity, subtype, update_cb):
def do_create_finger(final_template, tid):
tinfo = self.make_finger_data(subtype, final_template, tid)
usr=db.lookup_user(identity)
if usr == None:
usr = db.new_user(identity)
else:
usr = usr.dbid
recid = db.new_finger(usr, tinfo)
usb.wait_int()
glow_end_scan()
return recid
key=0
template=b''
self.create_enrollment()
while True:
try:
tinfo = self.make_finger_data(subtype, final_template, tid)
usr=db.lookup_user(identity)
if usr == None:
usr = db.new_user(identity)
else:
usr = usr.dbid
recid = db.new_finger(usr, tinfo)
glow_start_scan()
self.capture(CaptureMode.ENROLL)
key = self.enrollment_update_start(key)
rsp = self.append_new_image(template)
header, template, tid = rsp
update_cb(header, None)
if tid:
break
except usb_core.USBError as e:
raise e
except CancelledException as e:
glow_end_scan()
complete_cb(recid, None)
raise e
except Exception as e:
complete_cb(None, e)
print(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
finally:
self.enrollment_update_end()
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 r
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()
self.enrollment_update_end() # done twice for some reason
return do_create_finger(template, tid)
def parse_dict(self, x):
rc={}
@ -861,74 +794,56 @@ class Sensor():
return rc
def match_finger(self, complete_cb):
if usb.interrupt_cb is not None:
raise Exception('Wrong state for capture')
def match_finger(self):
try:
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)
def wait(b):
if b is None:
raise Exception('Cancelling while finger match is running is not a good idea.')
b = usb.wait_int()
if b[0] != 3:
raise Exception('Finger not recognized: %s' % hexlify(b).decode())
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:]
# 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')
(l,), rsp = unpack('<H', rsp[:2]), rsp[2:]
if l != len(rsp):
raise Exception('Response size does not match')
rsp=self.parse_dict(rsp)
rsp=self.parse_dict(rsp)
usrid, subtype, hsh = rsp[1], rsp[3], rsp[4]
usrid, = unpack('<L', usrid)
subtype, = unpack('<H', subtype)
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
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)
return (usrid, subtype, hsh)
finally:
# cleanup, ignore any errors
tls.app(unhexlify('6200000000'))
def identify(self, update_cb, complete_cb):
def start():
def identify(self, update_cb):
while True:
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)
def capture_cb(_, e):
try:
if e is not None: raise e
self.match_finger(complete_cb)
self.capture(CaptureMode.IDENTIFY)
break
except usb_core.USBError as e:
raise e
except CancelledException as e:
glow_end_scan()
complete_cb(None, e)
raise e
except Exception as e:
# Capture failed, retry
update_cb(e)
sleep(1)
start()
start()
return self.match_finger()
def get_finger_blobs(self, usrid, subtype):
usr = db.get_user(usrid)

3
validitysensor/upload_fwext.py
View file

@ -7,7 +7,7 @@ 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 flush_changes, read_flash, erase_flash, write_flash_all, write_fw_signature, get_fw_info
from .flash import read_flash, erase_flash, write_flash_all, write_fw_signature, get_fw_info
from .util import assert_status
firmware_home='/usr/share/python-validity'
@ -55,7 +55,6 @@ def upload_fwext(fw_path=None):
fwext=fwext[fwext.index(b'\x1a')+1:]
fwext, signature = fwext[:-0x100], fwext[-0x100:]
#flush_changes()
write_flash_all(2, 0, fwext)
write_fw_signature(2, signature)

34
validitysensor/usb.py
View file

@ -1,5 +1,6 @@
import logging
import errno
import usb.core as ucore
from binascii import *
from .util import assert_status, unhex
@ -15,6 +16,8 @@ supported_devices=[
(0x06cb, 0x009a),
]
class CancelledException(Exception):
pass
class Usb():
def __init__(self):
@ -22,6 +25,7 @@ class Usb():
self.trace_enabled = False
self.quit = None
self.dev = None
self.cancel = False
def open(self, vendor=None, product=None):
if vendor is not None and product is not None:
@ -48,9 +52,6 @@ class Usb():
self.dev = dev
self.dev.default_timeout = 15000
self.thread = Thread(target=lambda: self.int_thread())
self.thread.daemon = True
self.thread.start()
def close(self):
if self.dev is not None:
@ -59,8 +60,6 @@ class Usb():
self.dev = None
except:
pass
finally:
self.thread.join()
def usb_dev(self):
return self.dev
@ -69,12 +68,12 @@ class Usb():
#self.dev.set_configuration()
# TODO analyse responses, detect hardware type
assert_status(self.cmd(unhexlify('01')))
assert_status(self.cmd(unhexlify('01'))) # RomInfo.get()
assert_status(self.cmd(unhexlify('19')))
# 43 -- get partition header(?) (02 -- fwext partition)
# c28c745a in response is a FwextBuildtime = 0x5A748CC2
rsp=self.cmd(unhexlify('4302'))
rsp=self.cmd(unhexlify('4302')) # get_fw_info()
assert_status(self.cmd(init_hardcoded))
@ -106,6 +105,27 @@ class Usb():
self.trace('<130< Error: %s' % repr(e))
return None
# FIXME There is a chance of a race condition here
def cancel(self):
self.cancel = True
def wait_int(self):
self.cancel = False
while True:
try:
resp = self.dev.read(131, 1024, timeout=100)
resp = bytes(resp)
self.trace('<int< %s' % hexlify(resp).decode())
return resp
except USBError as e:
if e.errno == errno.ETIMEDOUT:
if self.cancel:
raise CancelledException()
else:
raise e
def int_thread(self):
try:
while True:

4
validitysensor/util.py
View file

@ -6,8 +6,12 @@ from binascii import unhexlify
def assert_status(b):
s,=unpack('<H', b[:2])
if s != 0:
if s == 0x44f:
raise Exception('Signature validation failed: %04x' % s)
raise Exception('Failed: %04x' % s)
def unhex(x):
return unhexlify(re.sub('\W', '', x))