diff --git a/.style.yapf b/.style.yapf new file mode 100644 index 0000000..e4b89ea --- /dev/null +++ b/.style.yapf @@ -0,0 +1,394 @@ +[style] +# Align closing bracket with visual indentation. +align_closing_bracket_with_visual_indent=True + +# Allow dictionary keys to exist on multiple lines. For example: +# +# x = { +# ('this is the first element of a tuple', +# 'this is the second element of a tuple'): +# value, +# } +allow_multiline_dictionary_keys=False + +# Allow lambdas to be formatted on more than one line. +allow_multiline_lambdas=False + +# Allow splitting before a default / named assignment in an argument list. +allow_split_before_default_or_named_assigns=True + +# Allow splits before the dictionary value. +allow_split_before_dict_value=True + +# Let spacing indicate operator precedence. For example: +# +# a = 1 * 2 + 3 / 4 +# b = 1 / 2 - 3 * 4 +# c = (1 + 2) * (3 - 4) +# d = (1 - 2) / (3 + 4) +# e = 1 * 2 - 3 +# f = 1 + 2 + 3 + 4 +# +# will be formatted as follows to indicate precedence: +# +# a = 1*2 + 3/4 +# b = 1/2 - 3*4 +# c = (1+2) * (3-4) +# d = (1-2) / (3+4) +# e = 1*2 - 3 +# f = 1 + 2 + 3 + 4 +# +arithmetic_precedence_indication=False + +# Number of blank lines surrounding top-level function and class +# definitions. +blank_lines_around_top_level_definition=2 + +# Insert a blank line before a class-level docstring. +blank_line_before_class_docstring=False + +# Insert a blank line before a module docstring. +blank_line_before_module_docstring=False + +# Insert a blank line before a 'def' or 'class' immediately nested +# within another 'def' or 'class'. For example: +# +# class Foo: +# # <------ this blank line +# def method(): +# ... +blank_line_before_nested_class_or_def=False + +# Do not split consecutive brackets. Only relevant when +# dedent_closing_brackets is set. For example: +# +# call_func_that_takes_a_dict( +# { +# 'key1': 'value1', +# 'key2': 'value2', +# } +# ) +# +# would reformat to: +# +# call_func_that_takes_a_dict({ +# 'key1': 'value1', +# 'key2': 'value2', +# }) +coalesce_brackets=False + +# The column limit. +column_limit=100 + +# The style for continuation alignment. Possible values are: +# +# - SPACE: Use spaces for continuation alignment. This is default behavior. +# - FIXED: Use fixed number (CONTINUATION_INDENT_WIDTH) of columns +# (ie: CONTINUATION_INDENT_WIDTH/INDENT_WIDTH tabs or +# CONTINUATION_INDENT_WIDTH spaces) for continuation alignment. +# - VALIGN-RIGHT: Vertically align continuation lines to multiple of +# INDENT_WIDTH columns. Slightly right (one tab or a few spaces) if +# cannot vertically align continuation lines with indent characters. +continuation_align_style=SPACE + +# Indent width used for line continuations. +continuation_indent_width=4 + +# Put closing brackets on a separate line, dedented, if the bracketed +# expression can't fit in a single line. Applies to all kinds of brackets, +# including function definitions and calls. For example: +# +# config = { +# 'key1': 'value1', +# 'key2': 'value2', +# } # <--- this bracket is dedented and on a separate line +# +# time_series = self.remote_client.query_entity_counters( +# entity='dev3246.region1', +# key='dns.query_latency_tcp', +# transform=Transformation.AVERAGE(window=timedelta(seconds=60)), +# start_ts=now()-timedelta(days=3), +# end_ts=now(), +# ) # <--- this bracket is dedented and on a separate line +dedent_closing_brackets=False + +# Disable the heuristic which places each list element on a separate line +# if the list is comma-terminated. +disable_ending_comma_heuristic=False + +# Place each dictionary entry onto its own line. +each_dict_entry_on_separate_line=True + +# Require multiline dictionary even if it would normally fit on one line. +# For example: +# +# config = { +# 'key1': 'value1' +# } +force_multiline_dict=False + +# The regex for an i18n comment. The presence of this comment stops +# reformatting of that line, because the comments are required to be +# next to the string they translate. +i18n_comment= + +# The i18n function call names. The presence of this function stops +# reformattting on that line, because the string it has cannot be moved +# away from the i18n comment. +i18n_function_call= + +# Indent blank lines. +indent_blank_lines=False + +# Put closing brackets on a separate line, indented, if the bracketed +# expression can't fit in a single line. Applies to all kinds of brackets, +# including function definitions and calls. For example: +# +# config = { +# 'key1': 'value1', +# 'key2': 'value2', +# } # <--- this bracket is indented and on a separate line +# +# time_series = self.remote_client.query_entity_counters( +# entity='dev3246.region1', +# key='dns.query_latency_tcp', +# transform=Transformation.AVERAGE(window=timedelta(seconds=60)), +# start_ts=now()-timedelta(days=3), +# end_ts=now(), +# ) # <--- this bracket is indented and on a separate line +indent_closing_brackets=False + +# Indent the dictionary value if it cannot fit on the same line as the +# dictionary key. For example: +# +# config = { +# 'key1': +# 'value1', +# 'key2': value1 + +# value2, +# } +indent_dictionary_value=False + +# The number of columns to use for indentation. +indent_width=4 + +# Join short lines into one line. E.g., single line 'if' statements. +join_multiple_lines=True + +# Do not include spaces around selected binary operators. For example: +# +# 1 + 2 * 3 - 4 / 5 +# +# will be formatted as follows when configured with "*,/": +# +# 1 + 2*3 - 4/5 +no_spaces_around_selected_binary_operators= + +# Use spaces around default or named assigns. +spaces_around_default_or_named_assign=False + +# Adds a space after the opening '{' and before the ending '}' dict delimiters. +# +# {1: 2} +# +# will be formatted as: +# +# { 1: 2 } +spaces_around_dict_delimiters=False + +# Adds a space after the opening '[' and before the ending ']' list delimiters. +# +# [1, 2] +# +# will be formatted as: +# +# [ 1, 2 ] +spaces_around_list_delimiters=False + +# Use spaces around the power operator. +spaces_around_power_operator=False + +# Use spaces around the subscript / slice operator. For example: +# +# my_list[1 : 10 : 2] +spaces_around_subscript_colon=False + +# Adds a space after the opening '(' and before the ending ')' tuple delimiters. +# +# (1, 2, 3) +# +# will be formatted as: +# +# ( 1, 2, 3 ) +spaces_around_tuple_delimiters=False + +# The number of spaces required before a trailing comment. +# This can be a single value (representing the number of spaces +# before each trailing comment) or list of values (representing +# alignment column values; trailing comments within a block will +# be aligned to the first column value that is greater than the maximum +# line length within the block). For example: +# +# With spaces_before_comment=5: +# +# 1 + 1 # Adding values +# +# will be formatted as: +# +# 1 + 1 # Adding values <-- 5 spaces between the end of the statement and comment +# +# With spaces_before_comment=15, 20: +# +# 1 + 1 # Adding values +# two + two # More adding +# +# longer_statement # This is a longer statement +# short # This is a shorter statement +# +# a_very_long_statement_that_extends_beyond_the_final_column # Comment +# short # This is a shorter statement +# +# will be formatted as: +# +# 1 + 1 # Adding values <-- end of line comments in block aligned to col 15 +# two + two # More adding +# +# longer_statement # This is a longer statement <-- end of line comments in block aligned to col 20 +# short # This is a shorter statement +# +# a_very_long_statement_that_extends_beyond_the_final_column # Comment <-- the end of line comments are aligned based on the line length +# short # This is a shorter statement +# +spaces_before_comment=2 + +# Insert a space between the ending comma and closing bracket of a list, +# etc. +space_between_ending_comma_and_closing_bracket=True + +# Use spaces inside brackets, braces, and parentheses. For example: +# +# method_call( 1 ) +# my_dict[ 3 ][ 1 ][ get_index( *args, **kwargs ) ] +# my_set = { 1, 2, 3 } +space_inside_brackets=False + +# Split before arguments +split_all_comma_separated_values=False + +# Split before arguments, but do not split all subexpressions recursively +# (unless needed). +split_all_top_level_comma_separated_values=False + +# Split before arguments if the argument list is terminated by a +# comma. +split_arguments_when_comma_terminated=False + +# Set to True to prefer splitting before '+', '-', '*', '/', '//', or '@' +# rather than after. +split_before_arithmetic_operator=False + +# Set to True to prefer splitting before '&', '|' or '^' rather than +# after. +split_before_bitwise_operator=True + +# Split before the closing bracket if a list or dict literal doesn't fit on +# a single line. +split_before_closing_bracket=True + +# Split before a dictionary or set generator (comp_for). For example, note +# the split before the 'for': +# +# foo = { +# variable: 'Hello world, have a nice day!' +# for variable in bar if variable != 42 +# } +split_before_dict_set_generator=True + +# Split before the '.' if we need to split a longer expression: +# +# foo = ('This is a really long string: {}, {}, {}, {}'.format(a, b, c, d)) +# +# would reformat to something like: +# +# foo = ('This is a really long string: {}, {}, {}, {}' +# .format(a, b, c, d)) +split_before_dot=False + +# Split after the opening paren which surrounds an expression if it doesn't +# fit on a single line. +split_before_expression_after_opening_paren=False + +# If an argument / parameter list is going to be split, then split before +# the first argument. +split_before_first_argument=False + +# Set to True to prefer splitting before 'and' or 'or' rather than +# after. +split_before_logical_operator=True + +# Split named assignments onto individual lines. +split_before_named_assigns=True + +# Set to True to split list comprehensions and generators that have +# non-trivial expressions and multiple clauses before each of these +# clauses. For example: +# +# result = [ +# a_long_var + 100 for a_long_var in xrange(1000) +# if a_long_var % 10] +# +# would reformat to something like: +# +# result = [ +# a_long_var + 100 +# for a_long_var in xrange(1000) +# if a_long_var % 10] +split_complex_comprehension=False + +# The penalty for splitting right after the opening bracket. +split_penalty_after_opening_bracket=300 + +# The penalty for splitting the line after a unary operator. +split_penalty_after_unary_operator=10000 + +# The penalty of splitting the line around the '+', '-', '*', '/', '//', +# ``%``, and '@' operators. +split_penalty_arithmetic_operator=300 + +# The penalty for splitting right before an if expression. +split_penalty_before_if_expr=0 + +# The penalty of splitting the line around the '&', '|', and '^' +# operators. +split_penalty_bitwise_operator=300 + +# The penalty for splitting a list comprehension or generator +# expression. +split_penalty_comprehension=80 + +# The penalty for characters over the column limit. +split_penalty_excess_character=7000 + +# The penalty incurred by adding a line split to the unwrapped line. The +# more line splits added the higher the penalty. +split_penalty_for_added_line_split=30 + +# The penalty of splitting a list of "import as" names. For example: +# +# from a_very_long_or_indented_module_name_yada_yad import (long_argument_1, +# long_argument_2, +# long_argument_3) +# +# would reformat to something like: +# +# from a_very_long_or_indented_module_name_yada_yad import ( +# long_argument_1, long_argument_2, long_argument_3) +split_penalty_import_names=0 + +# The penalty of splitting the line around the 'and' and 'or' +# operators. +split_penalty_logical_operator=300 + +# Use the Tab character for indentation. +use_tabs=False + diff --git a/bin/validity-led-dance b/bin/validity-led-dance index 73d1947..7a56f30 100755 --- a/bin/validity-led-dance +++ b/bin/validity-led-dance @@ -1,15 +1,13 @@ #!/usr/bin/env python3 import os - from binascii import unhexlify -from enum import Enum -from validitysensor.flash import read_flash -from validitysensor.sensor import * -from validitysensor import init from time import sleep -from usb import core as usb_core +from validitysensor import init +from validitysensor.sensor import glow_start_scan, glow_end_scan +from validitysensor.tls import tls +from validitysensor.util import assert_status if __name__ == "__main__": if os.geteuid() != 0: @@ -23,10 +21,9 @@ if __name__ == "__main__": glow_end_scan() sleep(0.05) - led_script = unhexlify( - '39ff100000ff03000001ff002000000000ffff0000ffff0000ff03000001ff00' \ - '200000000000000000ffff0000ff03000001ff002000000000ffff0000000000' \ - '0000000000000000000000000000000000000000000000000000000000000000' \ - '0000000000000000000000000000000000000000000000000000000000') + led_script = unhexlify('39ff100000ff03000001ff002000000000ffff0000ffff0000ff03000001ff00' + '200000000000000000ffff0000ff03000001ff002000000000ffff0000000000' + '0000000000000000000000000000000000000000000000000000000000000000' + '0000000000000000000000000000000000000000000000000000000000') assert_status(tls.app(led_script)) diff --git a/bin/validity-sensors-firmware b/bin/validity-sensors-firmware index 998017f..d1c77b4 100755 --- a/bin/validity-sensors-firmware +++ b/bin/validity-sensors-firmware @@ -20,17 +20,17 @@ import argparse import os +import shutil import subprocess import sys import tempfile import urllib.request -import shutil +from enum import Enum -from enum import Enum, auto -from time import sleep from usb import core as usb_core -python_validity_data='/usr/share/python-validity/' +python_validity_data = '/usr/share/python-validity/' + # FIXME: supported usb ids are duplicated in usb.py class VFS(Enum): @@ -38,6 +38,7 @@ class VFS(Enum): DEV_97 = (0x138a, 0x0097) DEV_9a = (0x06cb, 0x009a) + DEFAULT_URIS = { VFS.DEV_90: { 'driver': 'https://download.lenovo.com/pccbbs/mobiles/n1cgn08w.exe', @@ -76,15 +77,12 @@ def download_and_extract_fw(dev_type, fwdir, fwuri=None): with open(fwarchive, 'wb') as out_file: out_file.write(response.read()) - subprocess.check_call(['innoextract', - '--output-dir', fwdir, - '--include', fwname, - '--collisions', 'overwrite', + subprocess.check_call([ + 'innoextract', '--output-dir', fwdir, '--include', fwname, '--collisions', 'overwrite', fwarchive ]) - fwpath = subprocess.check_output([ - 'find', fwdir, '-name', fwname]).decode('utf-8').strip() + fwpath = subprocess.check_output(['find', fwdir, '-name', fwname]).decode('utf-8').strip() print('Found firmware at {}'.format(fwpath)) if not fwpath: @@ -92,6 +90,7 @@ def download_and_extract_fw(dev_type, fwdir, fwuri=None): return fwpath + if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument('--driver-uri') @@ -111,8 +110,7 @@ if __name__ == "__main__": raise Exception('No supported validity device found') try: - subprocess.check_call(['innoextract', '--version'], - stdout=subprocess.DEVNULL) + subprocess.check_call(['innoextract', '--version'], stdout=subprocess.DEVNULL) except Exception as e: print('Impossible to run innoextract: {}'.format(e)) sys.exit(1) diff --git a/dbus_service/dbus-service b/dbus_service/dbus-service index c19a6f7..82abb4f 100755 --- a/dbus_service/dbus-service +++ b/dbus_service/dbus-service @@ -1,6 +1,8 @@ #!/usr/bin/env python3 import argparse +import logging +import logging.handlers import os import pwd import re @@ -16,21 +18,18 @@ import dbus import dbus.mainloop.glib import dbus.service import yaml +from gi.repository import GLib from usb import core as usb_core -dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) -from gi.repository import GLib -import logging -import logging.handlers - 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, SidIdentity, User from validitysensor.sensor import sensor, RebootException +from validitysensor.sid import sid_from_string +from validitysensor.tls import tls +from validitysensor.usb import usb from validitysensor.winbio_constants import finger_ids +dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) GLib.threads_init() INTERFACE_NAME = 'io.github.uunicorn.Fprint.Device' @@ -64,9 +63,7 @@ class Device(dbus.service.Object): """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") + @dbus.service.method(dbus_interface=INTERFACE_NAME, in_signature="s", out_signature="as") def ListEnrolledFingers(self, user): try: logging.debug('In ListEnrolledFingers %s' % user) @@ -82,9 +79,7 @@ class Device(dbus.service.Object): except Exception as e: raise e - @dbus.service.method(dbus_interface=INTERFACE_NAME, - in_signature='s', - out_signature='') + @dbus.service.method(dbus_interface=INTERFACE_NAME, in_signature='s', out_signature='') def DeleteEnrolledFingers(self, user): logging.debug('In DeleteEnrolledFingers %s' % user) usr = self.user2record(user) @@ -94,9 +89,7 @@ class Device(dbus.service.Object): db.del_record(usr.dbid) - @dbus.service.method(dbus_interface=INTERFACE_NAME, - in_signature='ss', - out_signature='') + @dbus.service.method(dbus_interface=INTERFACE_NAME, in_signature='ss', out_signature='') def VerifyStart(self, user, finger): logging.debug('In VerifyStart for %s, %s' % (user, finger)) @@ -130,15 +123,11 @@ class Device(dbus.service.Object): thread.daemon = True thread.start() - @dbus.service.method(dbus_interface=INTERFACE_NAME, - in_signature='', - out_signature='') + @dbus.service.method(dbus_interface=INTERFACE_NAME, in_signature='', out_signature='') def Cancel(self): sensor.cancel() - @dbus.service.method(dbus_interface=INTERFACE_NAME, - in_signature='ss', - out_signature='') + @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)) @@ -169,11 +158,9 @@ class Device(dbus.service.Object): logging.exception(e) self.EnrollStatus('enroll-failed', True) - if index == None: - logging.exception( - 'Unknown finger name passed to enroll? ' + - finger_name + ' (' + winbio_name + ')' - ) + if index is None: + logging.exception('Unknown finger name passed to enroll? ' + finger_name + ' (' + + winbio_name + ')') self.EnrollStatus('enroll-failed', True) else: thread = Thread(target=run) @@ -192,9 +179,7 @@ class Device(dbus.service.Object): def EnrollStatus(self, result, done): logging.debug('EnrollStatus') - @dbus.service.method(dbus_interface=INTERFACE_NAME, - in_signature='s', - out_signature='s') + @dbus.service.method(dbus_interface=INTERFACE_NAME, in_signature='s', out_signature='s') def RunCmd(self, cmd): logging.debug('RunCmd') return hexlify(tls.app(unhexlify(cmd))).decode() @@ -228,7 +213,10 @@ 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--
') - parser.add_argument('--configpath', default='/etc/python-validity', type=Path, help='Path to config files') + parser.add_argument('--configpath', + default='/etc/python-validity', + type=Path, + help='Path to config files') args = parser.parse_args() if args.debug: diff --git a/scripts/dbus-cmd.py b/scripts/dbus-cmd.py index d6658db..3774e00 100755 --- a/scripts/dbus-cmd.py +++ b/scripts/dbus-cmd.py @@ -1,13 +1,11 @@ #!/usr/bin/env python3 -import argparse -import re -import dbus -import dbus.mainloop.glib -dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) -from gi.repository import GObject, GLib import sys +import dbus.mainloop.glib + +dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) + bus = dbus.SystemBus() o = bus.get_object('net.reactivated.Fprint', '/net/reactivated/Fprint/Manager', introspect=False) o = o.GetDefaultDevice() diff --git a/scripts/factory-reset.py b/scripts/factory-reset.py index a997d50..377513f 100644 --- a/scripts/factory-reset.py +++ b/scripts/factory-reset.py @@ -1,6 +1,5 @@ - -from validitysensor.usb import usb from validitysensor.sensor import factory_reset, RebootException +from validitysensor.usb import usb try: usb.open() diff --git a/scripts/lsdbus.py b/scripts/lsdbus.py index 637e255..fc66776 100755 --- a/scripts/lsdbus.py +++ b/scripts/lsdbus.py @@ -1,13 +1,8 @@ #!/usr/bin/env python3 -import argparse -import re -import dbus import dbus.mainloop.glib -dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) -from gi.repository import GObject, GLib -import pwd +dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) bus = dbus.SystemBus() o = bus.get_object('org.freedesktop.DBus', '/org/freedesktop/DBus') @@ -16,10 +11,10 @@ ls = o.ListNames() for n in ls: if n[0] != ':': continue - + pid = o.GetConnectionUnixProcessID(n) with open('/proc/%d/cmdline' % pid) as f: - s=f.read() - s=s.split('\0') + s = f.read() + s = s.split('\0') print('%-10s %-5d %s' % (n, pid, ' '.join(s))) diff --git a/scripts/prototype.py b/scripts/prototype.py index b015798..83853b9 100644 --- a/scripts/prototype.py +++ b/scripts/prototype.py @@ -1,3 +1,7 @@ +import code +import logging +from binascii import hexlify + from validitysensor.tls import tls from validitysensor.usb import usb from validitysensor.db import db @@ -7,11 +11,10 @@ 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 -#tls.trace_enabled = True +# usb.trace_enabled = True +# tls.trace_enabled = True + def identify(): def update_cb(e): @@ -21,6 +24,7 @@ def identify(): print('Got finger %x for user recordid %d. Hash: %s' % (subtype, usrid, hexlify(hsh).decode())) + def enroll(sid, finger): def update_cb(x, e): if e is not None: @@ -31,4 +35,3 @@ def enroll(sid, finger): recid = sensor.enroll(sid, finger, update_cb) print('Created a finger record with dbid %d' % recid) - diff --git a/setup.py b/setup.py index b39fef7..ec19199 100755 --- a/setup.py +++ b/setup.py @@ -3,26 +3,19 @@ from setuptools import setup setup(name='python-validity', - version='0.9', - py_modules = [], - packages=['validitysensor'], - scripts=[ - 'bin/validity-led-dance', - 'bin/validity-sensors-firmware', - ], - install_requires=[ - 'cryptography >= 2.1.4', - 'pyusb >= 1.0.0', - 'pyyaml >= 3.12' - ], - data_files=[ - ('share/dbus-1/system.d/', ['dbus_service/io.github.uunicorn.Fprint.conf']), - ('lib/python-validity/', ['dbus_service/dbus-service']), - ('share/python-validity/playground/', [ - 'scripts/dbus-cmd.py', - 'scripts/lsdbus.py', - 'scripts/factory-reset.py', - 'scripts/prototype.py' - ]), - ] -) + version='0.9', + py_modules=[], + packages=['validitysensor'], + scripts=[ + 'bin/validity-led-dance', + 'bin/validity-sensors-firmware', + ], + install_requires=['cryptography >= 2.1.4', 'pyusb >= 1.0.0', 'pyyaml >= 3.12'], + data_files=[ + ('share/dbus-1/system.d/', ['dbus_service/io.github.uunicorn.Fprint.conf']), + ('lib/python-validity/', ['dbus_service/dbus-service']), + ('share/python-validity/playground/', [ + 'scripts/dbus-cmd.py', 'scripts/lsdbus.py', 'scripts/factory-reset.py', + 'scripts/prototype.py' + ]), + ]) diff --git a/validitysensor/blobs.py b/validitysensor/blobs.py index 5834c49..edde185 100644 --- a/validitysensor/blobs.py +++ b/validitysensor/blobs.py @@ -1,13 +1,4 @@ -from enum import Enum, auto - -class Blobs(Enum): - init_hardcoded = auto() - init_hardcoded_clean_slate = auto() - reset_blob = auto() - db_write_enable = auto() - - -def __load_blob(blob): +def __load_blob(blob: str) -> bytes: from .usb import usb if usb.usb_dev().idVendor == 0x138a: @@ -22,6 +13,8 @@ def __load_blob(blob): globals()[blob] = getattr(blobs, blob) return globals()[blob] -for p in dir(Blobs): - if isinstance(getattr(Blobs, p), Blobs): - globals()[p] = lambda bname=p: __load_blob(bname) + +init_hardcoded = lambda: __load_blob('init_hardcoded') +init_hardcoded_clean_slate = lambda: __load_blob('init_hardcoded_clean_slate') +reset_blob = lambda: __load_blob('reset_blob') +db_write_enable = lambda: __load_blob('db_write_enable') diff --git a/validitysensor/blobs_90.py b/validitysensor/blobs_90.py index 222fedf..228d34d 100644 --- a/validitysensor/blobs_90.py +++ b/validitysensor/blobs_90.py @@ -1,7 +1,6 @@ - from .util import unhex -init_hardcoded=unhex(''' +init_hardcoded = unhex(''' 06020000013917b3dda91383b5bcac64fa4ad35dce96570a9d2d974b80926a431f9cd46248980a263c6fcef6a82839a90b59ac590848859 afac817b7d53bf51cd3205c1b8f43048be8253c3bd247937c837aca8b18d3cc8ee8c8971ac4f688813cf3d8550d71496985b7ec07ff2dc7 896d330fdab263a0ee433a5c4bc910439d1c6161853feb03f5502209502e7308beb7919473cfe69f422c30502d226a4d0a34d96c8c77956 @@ -13,9 +12,9 @@ e024ca6b9a48332c1a69a5a3fdd24b964cf7e7c55229bb0b48a6e339eb2c42d07ec850a4ee780660 b3d5cad8d5bd7cea37e58a31307a6d50e6ae379a53f1366678c0741a3d872b8dcfefa7f63128dc8245 ''') -init_hardcoded_clean_slate=unhex('') +init_hardcoded_clean_slate = unhex('') -reset_blob=unhex(''' +reset_blob = unhex(''' 0602000001d6123241376f110935a7ae38b614ced9952eaa38e2984842d0552d984b69ab5c1a70175513319970225d9ca25be063a7d70a2 9dff803240c099ba5386910b69c3b7ca8d0b26ec9a5684fe92af9c6d8068ace0954eb8582adc6208aa6c116b1778ff531dd2b6817a10d66 7e0385d18ac7ccb47eb67f1f54b5b94c84d65529c6e9fe9039218ae9cde123f38ffa52a953ed589d0d8bb48680aeea6544b7ceedbea3216 @@ -226,7 +225,7 @@ f72aea07e183577f591a081845ea340797b79c6c8742e8826a6d7df773ec8564025fbb570bfbc001 a5d4483f1 ''') -db_write_enable=unhex(''' +db_write_enable = unhex(''' 06020000015ddf5d76a8e3bab8b10d9f82f0b31652d5b208dff1d20745b7c5442e09db880c80ca103d5b2071d8bdcaab452b900d1785be2 de478669e72a9d082273f0f6176b700ffb65a026af052e76de5f9e247e994bed7c79e34815b465a8070a76da7d365525d7285dfb152d24c 732a471f031d22ebb67d6e27bfff74b3ab1c4780461a460c0ea5710fbd32ab59a6ca9ca3e3e24264640b15245a6c48b6bd20c6848c2307e @@ -262,7 +261,7 @@ a2da95eb8445a6381f3941917bed7fcf01ab199228eadd1968847e4b9a40022d2b0ec68318abf97c ''') # FIXME this must be very specific to a particular device and constructed on the fly from multiple hardcoded tables: -identify_prg=unhex(''' +identify_prg = unhex(''' 02980000002300000020000800002000800000010032007000000000802020050024200000502077362820010030200100082170000c210 000482102004c210000582000005c20000060200000682005006c20012970200121742001887820018084202000942001809c200902a020 0b19b4200000b8203b04bc201400c0200200c4200100c82002003300100000000080cc200000f503d0200000a1013200440000000080dc2 @@ -295,7 +294,7 @@ c1e1e1c221f201d21201e1c1f34242221201f20221f201e241e241d2020221e2420231d221e211e1 ''') # The following blob has only 1 byte difference from the above -enroll_prg=unhex(''' +enroll_prg = unhex(''' 02980000002300000020000800002000800000010032007000000000802020050024200000502077362820010030200100082170000c210 000482102004c210000582000005c20000060200000682005006c20012970200121742001887820018084202000942001809c200902a020 0b19b4200000b8203b04bc201400c0200200c4200100c82002003300100000000080cc200000f503d0200000a1013200440000000080dc2 @@ -327,7 +326,7 @@ c1e1e1c221f201d21201e1c1f34242221201f20221f201e241e241d2020221e2420231d221e211e1 81808180818081808180818081808180818081808080808080808080807f807f807f807f7f7e7e ''') -calibrate_prg=unhex(''' +calibrate_prg = unhex(''' 0298006103230000002000080000200080000001003200700000000080202005002420000050207736282 0010030200100082170000c210000482102004c210000582000005c20000060200000682005006c200129 70200121742001887820018084202000942001809c200902a0200b19b4200000b8203b04bc201400c0200 @@ -365,4 +364,3 @@ c2d182e1e30182e1c321d341d341e321c301e1e241e201f201d1c321a301e1c211e21341f1e20202 8080808180818081808080818081808180818081808180808081808180818081808180818081808180818 0818081808180818081808080808080808080807f807f807f807f7f7e7e ''') - diff --git a/validitysensor/blobs_97.py b/validitysensor/blobs_97.py index 0112667..e9e364c 100644 --- a/validitysensor/blobs_97.py +++ b/validitysensor/blobs_97.py @@ -1,7 +1,6 @@ - from .util import unhex -init_hardcoded=unhex(''' +init_hardcoded = unhex(''' 06020000014a231406e5542fc6dc3b1aedebe68f55596ad3ca13f6e019994c6f71672fff756fbde0511d09d45978b12ba415b3694a0e763 48c8cfe9dbb9abf86813fc0c67c1005519a6f87360c2fb3e12bd0a9e012b06d9f5c9b44ccc6645b0fbd47afe45c8c874fcb88fbfd18fb7a 9b3241351f256acce689f9586a52b01f8fdcb66cdf3b340b1f9f386d58ca24fdfcdfbcebefb5f3a3c2a08357721040235a20ce1ee2f4f78 @@ -15,7 +14,7 @@ f3135c5e78820e36653fa3db535f57c71897242939d7da50f81070ce9ab81c61af6ac29a6c6c4a5d 2999d1c0e7acf67e598696cd58cc4bdb1b7c037ee9a085f784c4 ''') -init_hardcoded_clean_slate=unhex(''' +init_hardcoded_clean_slate = unhex(''' 06020000012c40c9d271378bc0912ef5dced69bd81b7fc16972c7b46e621af54a00e2cc6baca6eb83ea30222dfc6c925262006ae93412ea cf482f2034ee7b13297474b7e1e91f279cac2ccb7195443e4dd3328cfd292ade073fcc2eaa8f07b77231130ba997f921b9be7b4fb6cc691 0d2976b3e050913b27dbe73afd6e964260b9435ebab5117e71f7cb68464d4b6f8afc7e1a421f671f5854a1d0c8ab93ed3b88b2bc1a42875 @@ -32,7 +31,7 @@ b4c74c39900830abc6906a1004bef1b5b7dbbbeb5ec1b22604ac86429b9f56511b746a7124c449b8 aa41249faeef4d838e280cb5d6fc19cfe86c75f ''') -reset_blob=unhex(''' +reset_blob = unhex(''' 06020000018772bd56dd58d64023e1745f7c253a49b32dd6a02bc32347195b6763bfcc23c9e0beb0c5809e06a5628629f28c4048530a5cd df6f48391ea0c2cb5a7ffe93ef04c8b4dad584117e65aac085c25062a0f12a8ee432c7ecbb6613c28b743e4a75e382afc6b8037e342d466 7b66a73691edc6b25698c15e78d9d67f7cc56274e99e6b7bb5fba32dd42d74dfa672f414c4a29302b30a202d00a2571d2a884169e82106c @@ -252,7 +251,7 @@ f20332657d2ca6728b9865a6bfdc9a9055a6ab86cb782307c2c0255525884bf8060f8aed7ec034f3 20f8bb8d94784228934870046f60ac937af4957c9414bcc33895fcc183062be4e4bff56f9c89838ead8e433dc604536938 ''') -db_write_enable=unhex(''' +db_write_enable = unhex(''' 0602000001f48001074892b6c57deb7889b5ebf86bc3040f6d91ff1f68765f046591184be08cf36c154b7ec5368139d0f9532382214379a ff3ffbfe4659e2f274e864bd0ad660f99e21da2bab677dbfa907a66ce110c180d2ddc5dfe40b8ed975cbedffc11631f12f8bd646a0ee82d 44d2a6c1ec9cfbd40f485cb3d9124376b97b4a3349b0a730adda626d8ac28ec20e886aab1b8851deee3431c4d89c8bb3e787eaa9c0323df diff --git a/validitysensor/blobs_9a.py b/validitysensor/blobs_9a.py index 0112667..e9e364c 100644 --- a/validitysensor/blobs_9a.py +++ b/validitysensor/blobs_9a.py @@ -1,7 +1,6 @@ - from .util import unhex -init_hardcoded=unhex(''' +init_hardcoded = unhex(''' 06020000014a231406e5542fc6dc3b1aedebe68f55596ad3ca13f6e019994c6f71672fff756fbde0511d09d45978b12ba415b3694a0e763 48c8cfe9dbb9abf86813fc0c67c1005519a6f87360c2fb3e12bd0a9e012b06d9f5c9b44ccc6645b0fbd47afe45c8c874fcb88fbfd18fb7a 9b3241351f256acce689f9586a52b01f8fdcb66cdf3b340b1f9f386d58ca24fdfcdfbcebefb5f3a3c2a08357721040235a20ce1ee2f4f78 @@ -15,7 +14,7 @@ f3135c5e78820e36653fa3db535f57c71897242939d7da50f81070ce9ab81c61af6ac29a6c6c4a5d 2999d1c0e7acf67e598696cd58cc4bdb1b7c037ee9a085f784c4 ''') -init_hardcoded_clean_slate=unhex(''' +init_hardcoded_clean_slate = unhex(''' 06020000012c40c9d271378bc0912ef5dced69bd81b7fc16972c7b46e621af54a00e2cc6baca6eb83ea30222dfc6c925262006ae93412ea cf482f2034ee7b13297474b7e1e91f279cac2ccb7195443e4dd3328cfd292ade073fcc2eaa8f07b77231130ba997f921b9be7b4fb6cc691 0d2976b3e050913b27dbe73afd6e964260b9435ebab5117e71f7cb68464d4b6f8afc7e1a421f671f5854a1d0c8ab93ed3b88b2bc1a42875 @@ -32,7 +31,7 @@ b4c74c39900830abc6906a1004bef1b5b7dbbbeb5ec1b22604ac86429b9f56511b746a7124c449b8 aa41249faeef4d838e280cb5d6fc19cfe86c75f ''') -reset_blob=unhex(''' +reset_blob = unhex(''' 06020000018772bd56dd58d64023e1745f7c253a49b32dd6a02bc32347195b6763bfcc23c9e0beb0c5809e06a5628629f28c4048530a5cd df6f48391ea0c2cb5a7ffe93ef04c8b4dad584117e65aac085c25062a0f12a8ee432c7ecbb6613c28b743e4a75e382afc6b8037e342d466 7b66a73691edc6b25698c15e78d9d67f7cc56274e99e6b7bb5fba32dd42d74dfa672f414c4a29302b30a202d00a2571d2a884169e82106c @@ -252,7 +251,7 @@ f20332657d2ca6728b9865a6bfdc9a9055a6ab86cb782307c2c0255525884bf8060f8aed7ec034f3 20f8bb8d94784228934870046f60ac937af4957c9414bcc33895fcc183062be4e4bff56f9c89838ead8e433dc604536938 ''') -db_write_enable=unhex(''' +db_write_enable = unhex(''' 0602000001f48001074892b6c57deb7889b5ebf86bc3040f6d91ff1f68765f046591184be08cf36c154b7ec5368139d0f9532382214379a ff3ffbfe4659e2f274e864bd0ad660f99e21da2bab677dbfa907a66ce110c180d2ddc5dfe40b8ed975cbedffc11631f12f8bd646a0ee82d 44d2a6c1ec9cfbd40f485cb3d9124376b97b4a3349b0a730adda626d8ac28ec20e886aab1b8851deee3431c4d89c8bb3e787eaa9c0323df diff --git a/validitysensor/db.py b/validitysensor/db.py index fa8c4f3..bb21ff1 100644 --- a/validitysensor/db.py +++ b/validitysensor/db.py @@ -1,62 +1,70 @@ import typing - -from .tls import tls -from .util import assert_status from binascii import hexlify +from struct import pack, unpack + from .blobs import db_write_enable from .flash import call_cleanups -from .sid import * +from .sid import SidIdentity, sid_from_bytes +from .tls import tls +from .util import assert_status from .winbio_constants import finger_names -class UserStorage(): - def __init__(self, dbid, name): - self.dbid=dbid - self.name=name - self.users=[] + +class UserStorage: + def __init__(self, dbid: int, name: str): + self.dbid = dbid + self.name = name + self.users = [] def __repr__(self): - return '' % (self.dbid, repr(self.name), repr(self.users)) + return '' % (self.dbid, repr( + self.name), repr(self.users)) -class User(): - def __init__(self, dbid, identity): - self.dbid=dbid - self.identity=identity - self.fingers=[] + +class User: + def __init__(self, dbid: int, identity: str): + self.dbid = dbid + self.identity = identity + self.fingers: typing.List[typing.Mapping[str, int]] = [] def __repr__(self): - return '' % (self.dbid, repr(self.identity), repr(self.fingers)) + return '' % (self.dbid, repr( + self.identity), repr(self.fingers)) + def subtype_to_string(s: int): finger_name = finger_names.get(s, None) return finger_name or 'Unknown' -def parse_user_storage(rsp): + +def parse_user_storage(rsp: bytes): rc, = unpack(' 0: raise Exception('Junk at the end of the storage info response: %s' % rsp.hex()) - storage=UserStorage(recid, name) + storage = UserStorage(recid, name) while len(usrtab) > 0: rec, usrtab = usrtab[:4], usrtab[4:] urid, valsz = unpack(' 0: raise Exception('Junk at the end of the user info response: %s' % rsp.hex()) identity = parse_identity(identity) - user=User(recid, identity) + user = User(recid, identity) while len(fingertab) > 0: rec, fingertab = fingertab[:8], fingertab[8:] frid, subtype, stgid, valsz = unpack('' % ( - self.dbid, - self.type, - self.storage, - repr(self.value), - repr(self.children) - ) - + self.dbid, self.type, self.storage, repr(self.value), repr(self.children)) -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 + +class Db: + class Info: + def __init__(self, total: int, used: int, free: int, records: int, 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)) + self.total, self.used, self.free, self.records, repr(self.roots)) def get_user_storage(self, dbid=0, name=''): - name=name.encode() + name = name.encode() if len(name) > 0: name += b'\0' @@ -148,15 +154,15 @@ class Db(): def get_storage_data(self): stg = self.get_user_storage(name='StgWindsor') rc = self.get_record_children(stg.dbid).children - return [i['dbid'] for i in rc if i['type'] == 8] # 8 == "data" type + return [i['dbid'] for i in rc if i['type'] == 8] # 8 == "data" type - def get_user(self, dbid): + def get_user(self, dbid: int): return parse_user(tls.cmd(pack(' typing.Optional[User]: stg = self.get_user_storage(name='StgWindsor') data = identity_to_bytes(identity) - + rsp = tls.cmd(pack(' 80: val = val[:80] + '...' - print('%s%d (type %d) %s' % (' '*depth, rec.dbid, rec.type, val)) + 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) + self.dump_raw(c['dbid'], depth + 1) def dump_all(self): stg = self.get_user_storage(name='StgWindsor') @@ -252,7 +258,8 @@ class Db(): for u in usrs: print('%2d: User %s with %d fingers:' % (u.dbid, repr(u.identity), len(u.fingers))) for f in u.fingers: - print(' %2d: %02x (%s)' % (f['dbid'], f['subtype'], subtype_to_string(f['subtype']))) + print(' %2d: %02x (%s)' % + (f['dbid'], f['subtype'], subtype_to_string(f['subtype']))) + db = Db() - diff --git a/validitysensor/flash.py b/validitysensor/flash.py index b14a773..54c5240 100644 --- a/validitysensor/flash.py +++ b/validitysensor/flash.py @@ -1,50 +1,66 @@ -from .tls import tls +import typing from struct import pack, unpack -from .util import assert_status, unhex + from .blobs import db_write_enable -from .hw_tables import flash_ic_table_lookup +from .hw_tables import flash_ic_table_lookup, FlashIcInfo +from .tls import tls +from .util import assert_status, unhex -class FlashInfo(): - def __init__(self, ic, blocks, unknown0, blocksize, unknown1, partitions): - self.ic, self.blocks, self.unknown0, self.blocksize, self.unknown1, self.partitions = ic, blocks, unknown0, blocksize, unknown1, partitions - - def __repr__(self): - return 'FlashInfo(%s, 0x%x, 0x%x, 0x%x, 0x%x, %s)' % (repr(self.ic), self.blocks, self.unknown0, self.blocksize, self.unknown1, repr(self.partitions)) # type 01 is for the firmware - written blocks are decrypted on the fly using fw key # access lvl: # 2 -- write only # 7 -- can read/write even without TLS -class PartitionInfo(): - def __init__(self, id, type, access_lvl, offset, size): +class PartitionInfo: + def __init__(self, id: int, type: int, access_lvl: int, offset: int, size: int): self.id, self.type, self.access_lvl, self.offset, self.size = id, type, access_lvl, offset, size def __repr__(self): - return 'PartitionInfo(0x%02x, 0x%02x, 0x%04x, 0x%08x, 0x%08x)' % (self.id, self.type, self.access_lvl, self.offset, self.size) + return 'PartitionInfo(0x%02x, 0x%02x, 0x%04x, 0x%08x, 0x%08x)' % ( + self.id, self.type, self.access_lvl, self.offset, self.size) + + +class FlashInfo: + def __init__(self, ic: FlashIcInfo, blocks: int, unknown0: int, blocksize: int, unknown1: int, + partitions: typing.Sequence[PartitionInfo]): + self.ic = ic + self.blocks = blocks + self.unknown0 = unknown0 + self.blocksize = blocksize + self.unknown1 = unknown1 + self.partitions = partitions + + def __repr__(self): + return 'FlashInfo(%s, 0x%x, 0x%x, 0x%x, 0x%x, %s)' % (repr( + self.ic), self.blocks, self.unknown0, self.blocksize, self.unknown1, + repr(self.partitions)) + def get_flash_info(): - rsp=tls.cmd(unhex('3e')) + rsp = tls.cmd(unhex('3e')) assert_status(rsp) - rsp=rsp[2:] - hdr=rsp[:0xe] - rsp=rsp[0xe:] + rsp = rsp[2:] + hdr = rsp[:0xe] + rsp = rsp[0xe:] jid0, jid1, blocks, unknown0, blocksize, unknown1, pcnt = unpack('>> 4302 -- get partition header (get fwext info) # b004 -- no fw detected # 0000 -# 0100 (major) 0100 (minor) 0800 (modules) c28c745a (buildtime) +# 0100 (major) 0100 (minor) 0800 (modules) c28c745a (buildtime) # type subtype major minor size # 0100 3446 0200 0700 d03e0000 # 0100 8408 0100 0700 00040000 @@ -54,64 +70,74 @@ def get_flash_info(): # 0200 2377 0000 0100 802f0000 # 0200 6637 0100 0c00 f0220200 # 0100 2556 0000 0100 60040000 -class ModuleInfo(): - def __init__(self, type, subtype, major, minor, size): +class ModuleInfo: + def __init__(self, type: int, subtype: int, major: int, minor: int, size: int): self.type, self.subtype, self.major, self.minor, self.size = type, subtype, major, minor, size def __repr__(self): - return 'ModuleInfo(0x%04x, 0x%04x, %d, %d, %d)' % (self.type, self.subtype, self.major, self.minor, self.size) + return 'ModuleInfo(0x%04x, 0x%04x, %d, %d, %d)' % (self.type, self.subtype, self.major, + self.minor, self.size) -class FirmwareInfo(): - def __init__(self, major, minor, buildtime, modules): + +class FirmwareInfo: + def __init__(self, major: int, minor: int, buildtime: int, + modules: typing.Sequence[ModuleInfo]): self.major, self.minor, self.buildtime, self.modules = major, minor, buildtime, modules def __repr__(self): - return 'FirmwareInfo(%d, %d, %d, %s)' % (self.major, self.minor, self.buildtime, repr(self.modules)) + return 'FirmwareInfo(%d, %d, %d, %s)' % (self.major, self.minor, self.buildtime, + repr(self.modules)) -def get_fw_info(partition): - rsp=tls.cmd(pack(' bytes: cmd = pack(' 0: - chunk, buf = buf[:bs], buf[bs:] + chunk, buf = buf[:bs], buf[bs:] write_flash(partition, ptr, chunk) ptr += len(chunk) -def read_flash_all(partition, start, size): + +def read_flash_all(partition: int, start: int, size: int): bs = 0x1000 - blocks = [read_flash(partition, addr, bs) for addr in range(start, start+size, bs)] + blocks = [read_flash(partition, addr, bs) for addr in range(start, start + size, bs)] return b''.join(blocks)[:size] -def write_fw_signature(partition, signature): - rsp=tls.cmd(pack(' 0: - x>>=1 - u+=1 + x >>= 1 + u += 1 # convert to array of binary strings with each element exactly u characters long - b=[bin(i-m+0x100)[-u:] for i in b] + b = [bin(i - m + 0x100)[-u:] for i in b] # combine chunks into one long text number with u*l binary digits and parse it as integer - b=int(''.join(b[::-1]), 2) + b = int(''.join(b[::-1]), 2) # convert back to bytes - b=b.to_bytes((u*l+7)//8, 'little') + b = b.to_bytes((u * l + 7) // 8, 'little') - return (u, m, b) + return u, m, b -class Line(): - mask=None - flags=None - data=None - v0=0 - v1=0 - v2=0 -def clip(x): +class Line: + def __init__(self): + self.mask: typing.Optional[int] = None + self.flags: typing.Optional[int] = None + self.data: typing.Optional[bytes] = None + self.v0 = 0 + self.v1 = 0 + self.v2 = 0 + + +def clip(x: int): if x < -128: - x=-128 + x = -128 if x > 127: - x=127 + x = 127 return x & 0xff -def scale(x): + +def scale(x: int): x -= 0x80 - x = int(x*10/0x22) # TODO: scaling factor depends on a device + x = int(x * 10 / 0x22) # TODO: scaling factor depends on a device return clip(x) -def add(l, r): +def add(l: int, r: int): # Make signed l, r = unpack('bb', pack('BB', l, r)) - return clip(l+r) + return clip(l + r) + + +def chunks(b: bytes, l: int): + return [b[i:i + l] for i in range(0, len(b), l)] -def chunks(b, l): - return [b[i:i+l] for i in range(0, len(b), l)] class CaptureMode(Enum): - CALIBRATE=1 - IDENTIFY=2 - ENROLL=3 + CALIBRATE = 1 + IDENTIFY = 2 + ENROLL = 3 -class Sensor(): - calib_data=b'' + +class Sensor: + calib_data = b'' def open(self): self.device_info = identify_sensor() 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: - self.key_calibration_line = 0x38 # (lines_per_calibration_data/2), but hardcoded for sensor type 0x199 - self.calibration_frames = 3 # TODO: workout where it's really comming from - self.calibration_iterations = 3 # hardcoded for type + self.key_calibration_line = 0x38 # (lines_per_calibration_data/2), but hardcoded for sensor type 0x199 + self.calibration_frames = 3 # TODO: workout where it's really comming from + self.calibration_iterations = 3 # hardcoded for type elif self.device_info.type == 0xdb: - self.key_calibration_line = 0x48 # TODO 48 is just a guess -- find it - self.calibration_frames = 6 # TODO: workout where it's really comming from + self.key_calibration_line = 0x48 # TODO 48 is just a guess -- find it + self.calibration_frames = 6 # TODO: workout where it's really comming from self.calibration_iterations = 0 else: - raise Exception('Device %s is not supported (sensor type 0x%x)' % (self.device_info.name, self.device_info.type)) - + raise Exception('Device %s is not supported (sensor type 0x%x)' % + (self.device_info.name, self.device_info.type)) self.rom_info = RomInfo.get() - self.hardcoded_prog = SensorCaptureProg.get(self.rom_info, self.device_info.type, 0x18, 0x19) # TODO: find where 0x18, 0x19 coming from + self.hardcoded_prog = SensorCaptureProg.get(self.rom_info, self.device_info.type, 0x18, + 0x19) # TODO: find where 0x18, 0x19 coming from if self.hardcoded_prog is None: - raise Exception('Can\'t find initial capture program for rom %s and sensor type %x' % (repr(self.rom_info), self.device_info.type)) + raise Exception('Can\'t find initial capture program for rom %s and sensor type %x' % + (repr(self.rom_info), self.device_info.type)) # Look for a "2D" chunk. It must have a 32 bit integer which represent the number of lines per frame - lines_2d = [unpack(' 1: - b[i+2] *= mult + if b[i + 2] > 1: + b[i + 2] *= mult if inc_address: - b[i+1] += 1 - i+=3 + b[i + 1] += 1 + i += 3 continue if b[i] == 0: - i+=1 + i += 1 continue if b[i] == 7: - i+=2 + i += 2 continue break return bytes(b) - def patch_timeslot_again(self, b): - b=bytearray(b) + def patch_timeslot_again(self, b: bytes): + b = bytearray(b) pc = 0 - match=None + match = None # Look for the last Call in the script while pc < len(b): opcode, l, *operands = prg.decode_insn(b[pc:]) @@ -276,7 +306,7 @@ class Sensor(): # Call if opcode == 11: - match = operands[1] # destination address + match = operands[1] # destination address pc += l @@ -303,88 +333,93 @@ class Sensor(): return bytes(b) # Hack the value to be taken from the factory calibration table right in the middle of a sensor - b[match+1] = self.factory_calibration_values[self.key_calibration_line] + b[match + 1] = self.factory_calibration_values[self.key_calibration_line] return bytes(b) - def average(self, raw_calib_data): + def average(self, raw_calib_data: bytes): frame_size = self.lines_per_frame * self.bytes_per_line - interleave_lines = self.lines_per_frame // self.type_info.lines_per_calibration_data # 2, TODO: algo is quite different when it is 1 + interleave_lines = self.lines_per_frame // self.type_info.lines_per_calibration_data # 2, TODO: algo is quite different when it is 1 input_frames = self.calibration_frames - + if interleave_lines > 1: if input_frames > 1: # skip the first frame input_frames -= 1 base_address = frame_size - frame=raw_calib_data[base_address:base_address+frame_size] + frame = raw_calib_data[base_address:base_address + frame_size] # split into groups of lines - frame=chunks(frame, interleave_lines*self.bytes_per_line) + frame = chunks(frame, interleave_lines * self.bytes_per_line) # split group of lines into lines - frame=[chunks(f, self.bytes_per_line) for f in frame] - + frame = [chunks(f, self.bytes_per_line) for f in frame] + # calculate averages across interleaved lines - frame=[bytes([sum(i)//len(f) for i in zip(*f)]) for f in frame] - frame=b''.join(frame) + frame = [bytes([sum(i) // len(f) for i in zip(*f)]) for f in frame] + frame = b''.join(frame) else: if input_frames > 1: # skip the first frame input_frames -= 2 - base_address = frame_size*2 + base_address = frame_size * 2 - frames=raw_calib_data[base_address:base_address+frame_size*input_frames] - frames=chunks(frames, frame_size) - frame=[int(sum(i)/input_frames) for i in zip(*frames)] - frame=bytes(frame) + frames = raw_calib_data[base_address:base_address + frame_size * input_frames] + frames = chunks(frames, frame_size) + frame = [int(sum(i) / input_frames) for i in zip(*frames)] + frame = bytes(frame) return frame - def process_calibration_results(self, cooked_data): - frame=chunks(cooked_data, self.bytes_per_line) + def process_calibration_results(self, cooked_data: bytes): + frame = chunks(cooked_data, self.bytes_per_line) # apply scaling factors - frame=[f[:8] + bytes(map(scale, f[8:])) for f in frame] - frame=b''.join(frame) + frame = [f[:8] + bytes(map(scale, f[8:])) for f in frame] + frame = b''.join(frame) if len(self.calib_data) > 0: # Not the first calibration run. Combine results # split previous calibration info into lines - lll=chunks(self.calib_data, self.bytes_per_line) + lll = chunks(self.calib_data, self.bytes_per_line) # split next calibration info into lines - rrr=chunks(frame, self.bytes_per_line) - + rrr = chunks(frame, self.bytes_per_line) + # Don't touch the first 8 bytes of each line, add everything else as signed characters, clipping the values - combined=[ll[:8] + bytes([add(l, r) for l, r in zip(ll[8:],rr[8:])]) for ll, rr in zip(lll, rrr)] + combined = [ + ll[:8] + bytes([add(l, r) for l, r in zip(ll[8:], rr[8:])]) + for ll, rr in zip(lll, rrr) + ] self.calib_data = bytes(b''.join(combined)) else: self.calib_data = frame def get_key_line(self): if len(self.calib_data) > 0: - bytes_per_calibration_line=len(self.calib_data) // self.type_info.lines_per_calibration_data - key_line_offset=8+bytes_per_calibration_line*self.key_calibration_line - key_line=self.calib_data[key_line_offset:key_line_offset+self.type_info.line_width] - key_line=bytes([i-1 if i == 5 else i for i in key_line]) + bytes_per_calibration_line = len( + self.calib_data) // self.type_info.lines_per_calibration_data + key_line_offset = 8 + bytes_per_calibration_line * self.key_calibration_line + key_line = self.calib_data[key_line_offset:key_line_offset + self.type_info.line_width] + key_line = bytes([i - 1 if i == 5 else i for i in key_line]) else: - key_line=b'\0'*self.type_info.line_width + key_line = b'\0' * self.type_info.line_width return key_line - def line_update_type_1(self, mode, chunks): + def line_update_type_1(self, mode: CaptureMode, + chunks: typing.List[typing.List[typing.Union[int, bytes]]]): for c in chunks: # Timeslot Table 2D if c[0] == 0x34: # TODO: figure out when to use address increment tst = self.patch_timeslot_table(c[1], True, self.type_info.repeat_multiplier) if mode != CaptureMode.CALIBRATE: - tst=self.patch_timeslot_again(tst) + tst = self.patch_timeslot_again(tst) c[1] = self.get_key_line() + tst[self.type_info.line_width:] - #---------------- Reply Configuration --------------- + # ---------------- Reply Configuration --------------- chunks += [[0x17, b'']] if mode == CaptureMode.IDENTIFY: @@ -392,53 +427,68 @@ class Sensor(): # It seems to be only used for identification and it looks almost identical to Finger Detect (0x26) # Seems to be the same all the time for a given sensor and mostly hardcoded # TODO: analyse construct_wtf_4e @0000000180090BF0 - chunks += [[0x4e, unhexlify('fbb20f0000000f00300000008700020067000a00018000000a0200000b1900008813b80b01091000')]] + chunks += [[ + 0x4e, + unhexlify( + 'fbb20f0000000f00300000008700020067000a00018000000a0200000b1900008813b80b01091000' + ) + ]] # Image Reconstruction. # TODO: analyse add_image_reconstruction_cmd_02_buff_list_item @000000018008EA70 - chunks += [[0x2e, unhexlify('0200180002000000700070004d010000a0008c003c32321e3c0a0202')]] + chunks += [[ + 0x2e, unhexlify('0200180002000000700070004d010000a0008c003c32321e3c0a0202') + ]] elif mode == CaptureMode.ENROLL: - chunks += [[0x26, unhexlify('fbb20f0000000f00300000008700020067000a00018000000a0200000b19000050c360ea01091000')]] + chunks += [[ + 0x26, + unhexlify( + 'fbb20f0000000f00300000008700020067000a00018000000a0200000b19000050c360ea01091000' + ) + ]] # Image Reconstruction. There is only one byte difference with the "identify" version. (same is true for 0097) - chunks += [[0x2e, unhexlify('0200180023000000700070004d010000a0008c003c32321e3c0a0202')]] + chunks += [[ + 0x2e, unhexlify('0200180023000000700070004d010000a0008c003c32321e3c0a0202') + ]] - #---------------- Interleave --------------- + # ---------------- Interleave --------------- chunks += [[0x44, pack(' 0: - bytes_per_calibration_line=len(self.calib_data) // self.type_info.lines_per_calibration_data + bytes_per_calibration_line = len( + self.calib_data) // self.type_info.lines_per_calibration_data for i in range(0, 112, 4): - l=Line() + l = Line() lines += [l] - l.mask=0xffffffff - l.flags=i | (0x85 << 24) - l.data=b'' + l.mask = 0xffffffff + l.flags = i | (0x85 << 24) + l.data = b'' for j in range(0, 112): - p=8+j*bytes_per_calibration_line+i - l.data += self.calib_data[p:p+4] + p = 8 + j * bytes_per_calibration_line + i + l.data += self.calib_data[p:p + 4] # Align to dwords, as the sensor demands it for l in lines: @@ -446,25 +496,28 @@ class Sensor(): if pad > 0: l.data += b'\0' * (4 - pad) - - #---------------- Line Update --------------- + # ---------------- Line Update --------------- line_update = pack('> 0x14) <= 1]) chunks += [[0x30, line_update]] - #---------------- Line Update Transform --------------- - update_transform = b''.join([pack('> 0x14) > 1]) + # ---------------- Line Update Transform --------------- + update_transform = b''.join([ + pack('> 0x14) > 1 + ]) chunks += [[0x43, update_transform]] return chunks - def line_update_type_2(self, mode, chunks): + def line_update_type_2(self, mode: CaptureMode, + chunks: typing.List[typing.List[typing.Union[int, bytes]]]): for c in chunks: - # patch the 2D params. + # patch the 2D params. # The following is only needed on some rom versions below 6.5 as reported by cmd_01 - #if c[0] == 0x2f: + # if c[0] == 0x2f: # c[1] = pack(' 0: l.data += b'\0' * (4 - pad) - - #---------------- Line Update --------------- + # ---------------- Line Update --------------- line_update = pack(' typing.Tuple[int, int, int, int]: try: assert_status(tls.app(self.build_cmd_02(mode))) @@ -638,8 +704,8 @@ class Sensor(): while True: b = usb.wait_int() if b[0] == 2: - break; - + break + # wait capture complete while True: b = usb.wait_int() @@ -648,12 +714,12 @@ class Sensor(): if b[2] & 4: break - + res = get_prg_status2() assert_status(res) res = res[2:] - + l, res = res[:4], res[4:] l, = unpack(' int: + rsp = tls.app(pack(' 0: tag, l = unpack(' 0: (t, l), x = unpack(' typing.Tuple[int, int, bytes]: try: - stg_id=0 # match against any storage - usr_id=0 # match against any user - cmd=pack('BBHL', self.revision, len(self.subauth), self.auth >> 32, self.auth & 0xffffffff) + b = pack('>BBHL', self.revision, len(self.subauth), self.auth >> 32, self.auth & 0xffffffff) for i in self.subauth: b += pack(' typing.Optional["SensorTypeInfo"]: + # noinspection PyUnresolvedReferences from . import generated_tables for i in cls.table: if i.sensor_type == sensor_type: return i - def __init__(self, sensor_type, bytes_per_line, repeat_multiplier, lines_per_calibration_data, line_width, calibration_blob): - self.sensor_type=sensor_type - self.repeat_multiplier=repeat_multiplier - self.lines_per_calibration_data=lines_per_calibration_data - self.line_width=line_width - self.bytes_per_line=bytes_per_line - self.calibration_blob=unhexlify(calibration_blob) + def __init__(self, sensor_type: int, bytes_per_line: int, repeat_multiplier: int, + lines_per_calibration_data: int, line_width: int, calibration_blob: str): + self.sensor_type = sensor_type + self.repeat_multiplier = repeat_multiplier + self.lines_per_calibration_data = lines_per_calibration_data + self.line_width = line_width + self.bytes_per_line = bytes_per_line + self.calibration_blob = unhexlify(calibration_blob) def __repr__(self): - calibration_blob=hexlify(self.calibration_blob).decode() + calibration_blob = hexlify(self.calibration_blob).decode() return 'SensorTypeInfo(sensor_type=0x%04x, bytes_per_line=0x%x, repeat_multiplier=%d, lines_per_calibration_data=%d, line_width=%d, calibration_blob=%s)' % ( - self.sensor_type, self.bytes_per_line, self.repeat_multiplier, self.lines_per_calibration_data, self.line_width, repr(calibration_blob)) + self.sensor_type, self.bytes_per_line, self.repeat_multiplier, + self.lines_per_calibration_data, self.line_width, repr(calibration_blob)) + def fuzzy(expected, actual): if expected == actual: @@ -32,6 +37,7 @@ def fuzzy(expected, actual): else: return 0 + def metric(i, rominfo): metric = 0 metric |= fuzzy(i.major, rominfo.major) @@ -42,13 +48,15 @@ def metric(i, rominfo): metric <<= 2 metric |= fuzzy(i.u1, rominfo.u1) - return metric + return metric + class SensorCaptureProg: - table=[] + table: typing.List["SensorCaptureProg"] = [] @classmethod - def get(cls, rominfo, sensor_type, a0, a1): + def get(cls, rominfo, sensor_type: int, a0: int, a1: int): + # noinspection PyUnresolvedReferences from . import generated_tables maximum = 0 @@ -74,20 +82,20 @@ class SensorCaptureProg: if found is not None: return b''.join(found.blobs) - - def __init__(self, major, minor, build, u1, dev_type, a0, a1, blobs): - self.major=major - self.minor=minor - self.build=build - self.u1=u1 - self.dev_type=dev_type - self.a0=a0 - self.a1=a1 - blobs=[unhexlify(b) for b in blobs] - self.blobs=blobs + + def __init__(self, major: int, minor: int, build: int, u1: int, dev_type: int, a0: int, a1: int, + blobs: typing.Sequence[str]): + self.major = major + self.minor = minor + self.build = build + self.u1 = u1 + self.dev_type = dev_type + self.a0 = a0 + self.a1 = a1 + self.blobs = [unhexlify(b) for b in blobs] def __repr__(self): - blobs=[hexlify(b).decode() for b in self.blobs] + blobs = [hexlify(b).decode() for b in self.blobs] return 'SensorCaptureProg(major=0x%x, minor=0x%x, build=0x%x, u1=0x%x, dev_type=0x%x, a0=0x%x, a1=0x%x, blobs=%s)' % ( self.major, self.minor, self.build, self.u1, self.dev_type, self.a0, self.a1, diff --git a/validitysensor/timeslot.py b/validitysensor/timeslot.py index 200d0da..b434ddd 100644 --- a/validitysensor/timeslot.py +++ b/validitysensor/timeslot.py @@ -1,151 +1,155 @@ - +import typing from binascii import hexlify from struct import unpack, pack -codes={} -codes[0x0] = "No Operation" -codes[0x1] = "Swipe" -codes[0x2] = "Timeslot Configuration" -codes[0x3] = "Register" -codes[0x4] = "Register Set 32" -codes[0x5] = "Register Operation 32" -codes[0x6] = "Security" -codes[0x7] = "WOE" -codes[0x8] = "Motion 1" -codes[0xa] = "CPUCLK" -codes[0xb] = "Motion 2" -codes[0xc] = "Calibration Block" -codes[0xd] = "Sweep" -codes[0xe] = "Zone Configuration" -codes[0xf] = "Zones Per Sweep" -codes[0x10] = "Lines Per Sweep Iteration" -codes[0x11] = "Lines Per Sweep" -codes[0x12] = "Total Zones" -codes[0x13] = "CAL WOE Ctrl" -codes[0x14] = "Cal WOE Mask" -codes[0x15] = "BW Reduciton" -codes[0x16] = "AGC" -codes[0x17] = "Reply Configuration" -codes[0x18] = "Motion 3" -codes[0x19] = "WOVAR" -codes[0x1a] = "Block MOde" -codes[0x1b] = "Bit Reduction" -codes[0x1c] = "Motion 4" -codes[0x1d] = "Calibration WOENF" -codes[0x1e] = "Calibration" -codes[0x1f] = "Zone Configuration A" -codes[0x20] = "Set Register 32" -codes[0x21] = "Register Operation 32A" -codes[0x22] = "Fingerprint Buffering" -codes[0x23] = "Reply Config + Timeslot Table" -codes[0x24] = "Baseline" -codes[0x25] = "SO Alternate" -codes[0x26] = "Finger Detect" -codes[0x27] = "Finger Detect Sample Register" -codes[0x28] = "Finger Detect Scan Registers" -codes[0x29] = "Timeslot Table Offset" -codes[0x2a] = "ACM Config" -codes[0x2b] = "ACM Control" -codes[0x2c] = "CEM Config" -codes[0x2d] = "CEM Control" -codes[0x2e] = "Image Reconstruction" -codes[0x2f] = "2D" -codes[0x30] = "Line Update" -codes[0x31] = "FDetect Timeslot Table" -codes[0x32] = "Register List 16" -codes[0x33] = "Register list 32" -codes[0x34] = "Timeslot Table 2D" -codes[0x35] = "Timeslot Table Offset for Finger Detect" -codes[0x36] = "Security Aligned" -codes[0x37] = "WOF2" -codes[0x38] = "WOE WOF" -codes[0x39] = "Navigation" -codes[0x3a] = "WOE WOF2 Version2" -codes[0x3b] = "Cal WOE WOF2" -codes[0x3c] = "Event Signal" -codes[0x3d] = "IFS Frame Stats" -codes[0x3e] = "SNR Method 4" -codes[0x3f] = "WOE WOF2 Version 3" -codes[0x40] = "Calibrate WOE WOF2 Version 3" -codes[0x41] = "Finger Detect Ratchet" -codes[0x42] = "Data Encoder" -codes[0x43] = "Line Update Transform" -codes[0x44] = "Line Update InterLeave" -codes[0x45] = "SO Table Values for Macros" -codes[0x46] = "Timeslot Macro Definitions" -codes[0x47] = "Enable ASP Feature" -codes[0x48] = "Baseline Frame" -codes[0x49] = "Rx Select" -codes[0x4e] = "WTF" -codes[0xffff] = "Unknown" +codes = {} +codes[0x0] = "No Operation" +codes[0x1] = "Swipe" +codes[0x2] = "Timeslot Configuration" +codes[0x3] = "Register" +codes[0x4] = "Register Set 32" +codes[0x5] = "Register Operation 32" +codes[0x6] = "Security" +codes[0x7] = "WOE" +codes[0x8] = "Motion 1" +codes[0xa] = "CPUCLK" +codes[0xb] = "Motion 2" +codes[0xc] = "Calibration Block" +codes[0xd] = "Sweep" +codes[0xe] = "Zone Configuration" +codes[0xf] = "Zones Per Sweep" +codes[0x10] = "Lines Per Sweep Iteration" +codes[0x11] = "Lines Per Sweep" +codes[0x12] = "Total Zones" +codes[0x13] = "CAL WOE Ctrl" +codes[0x14] = "Cal WOE Mask" +codes[0x15] = "BW Reduciton" +codes[0x16] = "AGC" +codes[0x17] = "Reply Configuration" +codes[0x18] = "Motion 3" +codes[0x19] = "WOVAR" +codes[0x1a] = "Block MOde" +codes[0x1b] = "Bit Reduction" +codes[0x1c] = "Motion 4" +codes[0x1d] = "Calibration WOENF" +codes[0x1e] = "Calibration" +codes[0x1f] = "Zone Configuration A" +codes[0x20] = "Set Register 32" +codes[0x21] = "Register Operation 32A" +codes[0x22] = "Fingerprint Buffering" +codes[0x23] = "Reply Config + Timeslot Table" +codes[0x24] = "Baseline" +codes[0x25] = "SO Alternate" +codes[0x26] = "Finger Detect" +codes[0x27] = "Finger Detect Sample Register" +codes[0x28] = "Finger Detect Scan Registers" +codes[0x29] = "Timeslot Table Offset" +codes[0x2a] = "ACM Config" +codes[0x2b] = "ACM Control" +codes[0x2c] = "CEM Config" +codes[0x2d] = "CEM Control" +codes[0x2e] = "Image Reconstruction" +codes[0x2f] = "2D" +codes[0x30] = "Line Update" +codes[0x31] = "FDetect Timeslot Table" +codes[0x32] = "Register List 16" +codes[0x33] = "Register list 32" +codes[0x34] = "Timeslot Table 2D" +codes[0x35] = "Timeslot Table Offset for Finger Detect" +codes[0x36] = "Security Aligned" +codes[0x37] = "WOF2" +codes[0x38] = "WOE WOF" +codes[0x39] = "Navigation" +codes[0x3a] = "WOE WOF2 Version2" +codes[0x3b] = "Cal WOE WOF2" +codes[0x3c] = "Event Signal" +codes[0x3d] = "IFS Frame Stats" +codes[0x3e] = "SNR Method 4" +codes[0x3f] = "WOE WOF2 Version 3" +codes[0x40] = "Calibrate WOE WOF2 Version 3" +codes[0x41] = "Finger Detect Ratchet" +codes[0x42] = "Data Encoder" +codes[0x43] = "Line Update Transform" +codes[0x44] = "Line Update InterLeave" +codes[0x45] = "SO Table Values for Macros" +codes[0x46] = "Timeslot Macro Definitions" +codes[0x47] = "Enable ASP Feature" +codes[0x48] = "Baseline Frame" +codes[0x49] = "Rx Select" +codes[0x4e] = "WTF" +codes[0xffff] = "Unknown" -insn_to_string=[ - 'NOOP', # 0 - 'End of Table', # 1 - 'Return', # 2 - 'Clear SO', # 3 - 'End of Data', # 4 - 'Marco %02x', # 5 - 'Enable Rx 0x%02x', # 6 - 'Idle Rx 0x%03x', # 7 - 'Enable SO 0x%03x', # 8 - 'Disable SO 0x%03x', # 9 - 'Interrupt %x', # 10 - 'Call rx_inc=%d, addr=%x, repeat=%d', # 11 - 'Features %02x', # 12 - 'Register Write *0x%08x = 0x%04x', # 13 - 'Sample %x, %x', # 14 - 'Sample Repeat %x, %x, repeat=%x', # 15 +insn_to_string = [ + 'NOOP', # 0 + 'End of Table', # 1 + 'Return', # 2 + 'Clear SO', # 3 + 'End of Data', # 4 + 'Marco %02x', # 5 + 'Enable Rx 0x%02x', # 6 + 'Idle Rx 0x%03x', # 7 + 'Enable SO 0x%03x', # 8 + 'Disable SO 0x%03x', # 9 + 'Interrupt %x', # 10 + 'Call rx_inc=%d, addr=%x, repeat=%d', # 11 + 'Features %02x', # 12 + 'Register Write *0x%08x = 0x%04x', # 13 + 'Sample %x, %x', # 14 + 'Sample Repeat %x, %x, repeat=%x', # 15 ] -def decode_insn(b): + +def decode_insn(b: bytes): if b[0] == 0: - return (0, 1) + return 0, 1 elif b[0] == 1: - return (1, 1) + return 1, 1 elif b[0] == 2: - return (2, 1) + return 2, 1 elif b[0] == 3: - return (3, 1) + return 3, 1 elif b[0] == 4: - return (4, 1) + return 4, 1 elif b[0] == 5: - return (5, 2, b[1]) + return 5, 2, b[1] elif b[0] == 6: - return (6, 2, b[1]) + return 6, 2, b[1] elif b[0] == 7: - return (7, 2, 0x100 if b[1] == 0 else b[1]) + return 7, 2, 0x100 if b[1] == 0 else b[1] elif b[0] & 0xfe == 8: - return (8, 2, (b[0] & 1) << 8 | b[1]) + return 8, 2, (b[0] & 1) << 8 | b[1] elif b[0] & 0xfe == 0xa: - return (9, 2, (b[0] & 1) << 8 | b[1]) + return 9, 2, (b[0] & 1) << 8 | b[1] elif b[0] & 0xfc == 0xc: - return (10, 1, b[0] & 3) + return 10, 1, b[0] & 3 elif b[0] & 0xf8 == 0x10: - return (11, 3, b[0] & 7, b[1] << 2, 0x100 if b[2] == 0 else b[2]) + return 11, 3, b[0] & 7, b[1] << 2, 0x100 if b[2] == 0 else b[2] elif b[0] & 0xe0 == 0x20: - return (12, 1, b[0] & 0x1f) # TODO check how features are converted to op args + return 12, 1, b[0] & 0x1f # TODO check how features are converted to op args elif b[0] & 0xc0 == 0x40: - return (13, 3, (b[0] & 0x3f)*4+0x80002000, b[1] | (b[2] << 8)) + return 13, 3, (b[0] & 0x3f) * 4 + 0x80002000, b[1] | (b[2] << 8) elif b[0] & 0xc0 == 0x80: - return (14, 1, (b[0] & 0x38) >> 3, b[0] & 7) + return 14, 1, (b[0] & 0x38) >> 3, b[0] & 7 elif b[0] & 0xc0 == 0xc0: - return (15, 2, (b[0] & 0x38) >> 3, b[0] & 7, 0x100 if b[1] == 0 else b[1]) + return 15, 2, (b[0] & 0x38) >> 3, b[0] & 7, 0x100 if b[1] == 0 else b[1] else: raise Exception('Unhandled instruction %02x' % b) -def disassm_timeslot_table(b, off): - pc=off + +def disassm_timeslot_table(b: bytes, off: int): + pc = off while len(b) > 0: op, sz, *operands = decode_insn(b) if sz > len(b): raise Exception('Truncated instruction') - print(' %04x: %-6s %s' % (pc, hexlify(b[:sz]).decode(), insn_to_string[op] % tuple(operands))) + print(' %04x: %-6s %s' % + (pc, hexlify(b[:sz]).decode(), insn_to_string[op] % tuple(operands))) b = b[sz:] pc += sz -def find_nth_insn(b, opcode, n): - pc=0 + +def find_nth_insn(b: bytes, opcode: int, n: int): + pc = 0 while len(b) > 0: op, sz, *_ = decode_insn(b) @@ -153,15 +157,16 @@ def find_nth_insn(b, opcode, n): raise Exception('Truncated instruction') if op == opcode: - n-=1 + n -= 1 if n == 0: - return (pc, b[:sz]) + return pc, b[:sz] b = b[sz:] pc += sz -def find_nth_regwrite(b, reg_addr, n): - pc=0 + +def find_nth_regwrite(b: bytes, reg_addr: int, n: int): + pc = 0 while len(b) > 0: op, sz, *operands = decode_insn(b) @@ -171,24 +176,28 @@ def find_nth_regwrite(b, reg_addr, n): if op == 13: addr, value = operands if addr == reg_addr: - n-=1 + n -= 1 if n == 0: - return (pc, b[:sz]) + return pc, b[:sz] b = b[sz:] pc += sz -def split_chunks(b): +def split_chunks(b: bytes) -> typing.Generator[typing.List[typing.Union[int, bytes]], None, None]: while len(b) > 0: (typ, sz), b = unpack(' 0: (typ, sz), b = unpack(' 0: (off, val), p = unpack(' 0: (off, val), p = unpack(' 0: - res += hmac.new(secret, a+seed, sha256).digest() + res += hmac.new(secret, a + seed, sha256).digest() a = hmac.new(secret, a, sha256).digest() n -= 1 return res[:length] + def hs_key(): - key=password_hardcoded[:0x10] - seed=password_hardcoded[0x10:] + b'\xaa'*2 - hs_key=prf(key, b'HS_KEY_PAIR_GEN' + seed, 0x20) + key = password_hardcoded[:0x10] + seed = password_hardcoded[0x10:] + b'\xaa' * 2 + hs_key = prf(key, b'HS_KEY_PAIR_GEN' + seed, 0x20) return int(hs_key[::-1].hex(), 16) -def with_2bytes_size(chunk): + +def with_2bytes_size(chunk: bytes): return pack('>H', len(chunk)) + chunk -def with_3bytes_size(chunk): + +def with_3bytes_size(chunk: bytes): return pack('>BH', len(chunk) >> 16, len(chunk)) + chunk -def with_1byte_size(chunk): + +def with_1byte_size(chunk: bytes): return pack('>B', len(chunk)) + chunk + def to_bytes(n): - b=b'' + b = b'' while n: b += (n & 0xff).to_bytes(1, 'big') n >>= 8 return b -def pad(b): - l = 16 - (len(b) % 16) - return b + bytes([l-1])*l -def unpad(b): - return b[:-1-b[-1]] +def pad(b: bytes): + l = 16 - (len(b) % 16) + return b + bytes([l - 1]) * l + + +def unpad(b: bytes): + return b[:-1 - b[-1]] # TODO assert the right state transitions -class Tls(): - - def __init__(self, usb): +class Tls: + def __init__(self, usb: Usb): self.usb = usb self.reset() try: @@ -101,20 +108,20 @@ class Tls(): self.secure_tx = False # Info about the host computer - def set_hwkey(self, product_name, serial_number): + def set_hwkey(self, product_name: str, serial_number: str): hw_key = bytes(product_name, 'ascii') + b'\0' + \ bytes(serial_number, 'ascii') + b'\0' # pre-TLS keys self.psk_encryption_key = prf(password_hardcoded, b'GWK' + hw_key, 0x20) - self.psk_validation_key = prf(self.psk_encryption_key, - b'GWK_SIGN' + gwk_sign_hardcoded, 0x20) + self.psk_validation_key = prf(self.psk_encryption_key, b'GWK_SIGN' + gwk_sign_hardcoded, + 0x20) - def cmd(self, cmd): + def cmd(self, cmd: typing.Union[bytes, typing.Callable[[], bytes]]): if self.secure_rx and self.secure_tx: - rsp=self.app(cmd) + rsp = self.app(cmd) else: - rsp=self.usb.cmd(cmd) + rsp = self.usb.cmd(cmd) return rsp @@ -124,31 +131,27 @@ class Tls(): self.handshake_hash = sha256() - rsp=self.usb.cmd(unhexlify('44000000') + self.make_handshake(self.make_client_hello())) + rsp = self.usb.cmd(unhexlify('44000000') + self.make_handshake(self.make_client_hello())) self.parse_tls_response(rsp) self.make_keys() - rsp=self.usb.cmd( - unhexlify('44000000') + - self.make_handshake( - self.make_certs() + - self.make_client_kex() + - self.make_cert_verify()) + - self.make_change_cipher_spec() + - self.make_handshake(self.make_finish())) + rsp = self.usb.cmd( + unhexlify('44000000') + self.make_handshake(self.make_certs() + self.make_client_kex() + + self.make_cert_verify()) + + self.make_change_cipher_spec() + self.make_handshake(self.make_finish())) self.parse_tls_response(rsp) - def trace(self, s): + def trace(self, s: str): if self.trace_enabled: logging.debug(s) - def app(self, b): + def app(self, b: typing.Union[bytes, typing.Callable[[], bytes]]): b = b() if callable(b) else b return self.parse_tls_response(self.usb.cmd(self.make_app_data(b))) - def update_neg(self, b): + def update_neg(self, b: bytes): self.handshake_hash.update(b) def make_keys(self): @@ -156,27 +159,28 @@ class Tls(): self.session_public = skey.private_numbers().public_numbers pre_master_secret = skey.exchange(ec.ECDH(), self.ecdh_q) seed = self.client_random + self.server_random - self.master_secret = prf(pre_master_secret, b'master secret'+seed, 0x30) - key_block = prf(self.master_secret, b'key expansion'+seed, 0x120) + self.master_secret = prf(pre_master_secret, b'master secret' + seed, 0x30) + key_block = prf(self.master_secret, b'key expansion' + seed, 0x120) self.sign_key = key_block[0x00:0x20] - self.validation_key = key_block[0x20:0x20+0x20] - self.encryption_key = key_block[0x40:0x40+0x20] - self.decryption_key = key_block[0x60:0x60+0x20] + self.validation_key = key_block[0x20:0x20 + 0x20] + self.encryption_key = key_block[0x40:0x40 + 0x20] + self.decryption_key = key_block[0x60:0x60 + 0x20] def save(self): with open('tls.dict', 'wb') as f: - pickle.dump({ - 'sign_key': self.sign_key, - 'validation_key': self.validation_key, - 'encryption_key': self.encryption_key, - 'decryption_key': self.decryption_key, - 'secure_rx': self.secure_rx, - 'secure_tx': self.secure_tx - }, f) + pickle.dump( + { + 'sign_key': self.sign_key, + 'validation_key': self.validation_key, + 'encryption_key': self.encryption_key, + 'decryption_key': self.decryption_key, + 'secure_rx': self.secure_rx, + 'secure_tx': self.secure_tx + }, f) def load(self): with open('tls.dict', 'rb') as f: - d=pickle.load(f) + d = pickle.load(f) self.sign_key = d['sign_key'] self.validation_key = d['validation_key'] self.encryption_key = d['encryption_key'] @@ -184,46 +188,46 @@ class Tls(): self.secure_rx = d['secure_rx'] self.secure_tx = d['secure_tx'] - def decrypt(self, c): + def decrypt(self, c: bytes): iv, c = c[:0x10], c[0x10:] cipher = Cipher(algorithms.AES(self.decryption_key), modes.CBC(iv), backend=crypto_backend) decryptor = cipher.decryptor() m = decryptor.update(c) + decryptor.finalize() - m=unpad(m) + m = unpad(m) return m - def encrypt(self, b): - #iv = unhexlify('454849acdd075174d6b9e713a957c2e7') + def encrypt(self, b: bytes): + # iv = unhexlify('454849acdd075174d6b9e713a957c2e7') iv = os.urandom(0x10) cipher = Cipher(algorithms.AES(self.encryption_key), modes.CBC(iv), backend=crypto_backend) encryptor = cipher.encryptor() - b=pad(b) - c=encryptor.update(b) + encryptor.finalize() + b = pad(b) + c = encryptor.update(b) + encryptor.finalize() return iv + c - def validate(self, t, b): + def validate(self, t: int, b: bytes): b, hs = b[:-0x20], b[-0x20:] hdr = pack('>BBBH', t, 3, 3, len(b)) - sig=hmac.new(self.validation_key, hdr+b, sha256).digest() + sig = hmac.new(self.validation_key, hdr + b, sha256).digest() if sig != hs: raise Exception('Packet signature validation check failed') self.trace('tls> %02x: %s' % (t, hexlify(b).decode())) hdr = pack('>BBBH', t, 3, 3, len(b)) - sig=hmac.new(self.sign_key, hdr+b, sha256).digest() + sig = hmac.new(self.sign_key, hdr + b, sha256).digest() return b + sig def make_finish(self): self.secure_tx = True hs_hash = self.handshake_hash.copy().digest() - verify_data = prf(self.master_secret, b'client finished'+hs_hash, 0xc) + verify_data = prf(self.master_secret, b'client finished' + hs_hash, 0xc) return b'\x14' + with_3bytes_size(verify_data) def make_change_cipher_spec(self): @@ -231,12 +235,13 @@ class Tls(): def make_certs(self): cert = self.tls_cert - cert = unhexlify('ac16') + cert # what's this? - cert = pack('>BH', 0, len(self.tls_cert)) + cert # this seems to violate the standard (should be len(cert)) - cert = pack('>BH', 0, len(self.tls_cert)) + cert # same + cert = unhexlify('ac16') + cert # what's this? + cert = pack('>BH', 0, len( + self.tls_cert)) + cert # this seems to violate the standard (should be len(cert)) + cert = pack('>BH', 0, len(self.tls_cert)) + cert # same return self.with_neg_hdr(0x0b, cert) - def with_neg_hdr(self, t, b): + def with_neg_hdr(self, t: int, b: bytes): b = pack('>B', t) + with_3bytes_size(b) self.update_neg(b) return b @@ -246,11 +251,11 @@ class Tls(): return self.with_neg_hdr(0x10, b) def make_cert_verify(self): - buf=self.handshake_hash.copy().digest() - b=self.priv_key.sign(buf, ec.ECDSA(Prehashed(hashes.SHA256()))) + buf = self.handshake_hash.copy().digest() + b = self.priv_key.sign(buf, ec.ECDSA(Prehashed(hashes.SHA256()))) return self.with_neg_hdr(0x0f, b) - def handle_server_hello(self, p): + def handle_server_hello(self, p: bytes): if p[:2] != unhexlify('0303'): raise Exception('unexpected TLS version %s' % hexlify(p[:2]).decode()) @@ -258,50 +263,54 @@ class Tls(): self.server_random, p = p[:0x20], p[0x20:] l = p[0] - self.server_sessid, p = p[1:1+l], p[1+l:] + self.server_sessid, p = p[1:1 + l], p[1 + l:] - (suite,), p = unpack('>H', p[:2]), p[2:] + (suite, ), p = unpack('>H', p[:2]), p[2:] if suite != 0xc005: raise Exception('Server accepted unsupported cipher suite %04x' % suite) if p[0] != 0: - raise Exception('Server selected to enable compression, which we don''t support %02x' % p[0]) + raise Exception('Server selected to enable compression, which we don' + 't support %02x' % p[0]) p = p[1:] if p != b'': raise Exception('Not expecting any more data') - def handle_cert_req(self, p): - (sign_and_hash_algo,), p = unpack('>H', p[:2]), p[2:] + def handle_cert_req(self, p: bytes): + (sign_and_hash_algo, ), p = unpack('>H', p[:2]), p[2:] if sign_and_hash_algo != 0x140: - raise Exception('Server requested a cert with an unsupported sign and hash algo combination %04x' % sign_and_hash_algo) + raise Exception( + 'Server requested a cert with an unsupported sign and hash algo combination %04x' % + sign_and_hash_algo) - (l,), p = unpack('>H', p[:2]), p[2:] + (l, ), p = unpack('>H', p[:2]), p[2:] if l != 0: raise Exception('Server requested a cert with non-empty list of CAs') if p != b'': raise Exception('Not expecting any more data') - def handle_server_hello_done(self, p): + def handle_server_hello_done(self, p: bytes): if p != b'': - raise Exception('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): + def handle_finish(self, b: bytes): hs_hash = self.handshake_hash.copy().digest() - verify_data = prf(self.master_secret, b'server finished'+hs_hash, 0xc) + verify_data = prf(self.master_secret, b'server finished' + hs_hash, 0xc) if verify_data != b: raise Exception('Final handshake check failed') - def handle_app_data(self, b): + def handle_app_data(self, b: bytes): if not self.secure_rx: raise Exception('App payload before secure connection established') return self.validate(0x17, self.decrypt(b)) - def handle_handshake(self, handshake): + def handle_handshake(self, handshake: bytes) -> None: if self.secure_rx: handshake = self.validate(0x16, self.decrypt(handshake)) @@ -325,10 +334,10 @@ class Tls(): else: raise Exception('Unknown handshake packet %02x' % t) - self.update_neg(hdr+p) + self.update_neg(hdr + p) - def parse_tls_response(self, rsp): - app_data=b'' + def parse_tls_response(self, rsp: bytes): + app_data = b'' while len(rsp) > 0: while len(rsp) < 5: @@ -347,7 +356,7 @@ class Tls(): elif t == 0x14: if pkt != unhexlify('01'): raise Exception('Unexpected ChangeCipherSpec payload') - + self.secure_rx = True elif t == 0x17: @@ -358,47 +367,48 @@ class Tls(): return app_data - def make_app_data(self, b): + def make_app_data(self, b: bytes): if not self.secure_tx: raise Exception('App payload before secure connection established') - b=self.encrypt(self.sign(0x17, b)) + b = self.encrypt(self.sign(0x17, b)) return unhexlify('170303') + with_2bytes_size(b) - def make_handshake(self, b): + def make_handshake(self, b: bytes): if self.secure_tx: - b=self.encrypt(self.sign(0x16, b)) + b = self.encrypt(self.sign(0x16, b)) return unhexlify('160303') + with_2bytes_size(b) def make_client_hello(self): - h = unhexlify('0303') # TLS 1.2 - #self.client_random = unhexlify('bc349559ac16c8f8362191395b4d04a435d870315f519eed8777488bc2b9600c') + h = unhexlify('0303') # TLS 1.2 + # self.client_random = unhexlify('bc349559ac16c8f8362191395b4d04a435d870315f519eed8777488bc2b9600c') self.client_random = os.urandom(0x20) - h += self.client_random # client's random - h += with_1byte_size(unhexlify('00000000000000')) # session ID + h += self.client_random # client's random + h += with_1byte_size(unhexlify('00000000000000')) # session ID suits = b'' - suits += pack('>H', 0xc005) # TLS_ECDH_ECDSA_WITH_AES_256_CBC_SHA - suits += pack('>H', 0x003d) # TLS_RSA_WITH_AES_256_CBC_SHA256 - suits += pack('>H', 0x008d) # TLS_RSA_WITH_AES_256_CBC_SHA256 + suits += pack('>H', 0xc005) # TLS_ECDH_ECDSA_WITH_AES_256_CBC_SHA + suits += pack('>H', 0x003d) # TLS_RSA_WITH_AES_256_CBC_SHA256 + suits += pack('>H', 0x008d) # TLS_RSA_WITH_AES_256_CBC_SHA256 h += with_2bytes_size(suits) - h += with_1byte_size(b'') # no compression options + h += with_1byte_size(b'') # no compression options exts = b'' - exts += self.make_ext(0x004, pack('>H', 0x0017)) # truncated_hmac = 0x17 - exts += self.make_ext(0x00b, with_1byte_size(unhexlify('00'))) # EC points format = uncompressed + exts += self.make_ext(0x004, pack('>H', 0x0017)) # truncated_hmac = 0x17 + exts += self.make_ext(0x00b, + with_1byte_size(unhexlify('00'))) # EC points format = uncompressed # h += with_2bytes_size(exts) - h += pack('>H', len(exts)-2) + exts # -2? WHY?!... + h += pack('>H', len(exts) - 2) + exts # -2? WHY?!... return self.with_neg_hdr(0x01, h) - def make_ext(self, id, b): + def make_ext(self, id: int, b: bytes): return pack('>H', id) + with_2bytes_size(b) - def parseTlsFlash(self, reply): + def parse_tls_flash(self, reply: bytes): while len(reply) > 0: hdr, reply = reply[:4], reply[4:] hs, reply = reply[:0x20], reply[0x20:] @@ -410,7 +420,7 @@ class Tls(): self.trace('block id %04x (%d bytes)' % (id, sz)) - m=sha256() + m = sha256() m.update(body) if m.digest() != hs: @@ -431,37 +441,37 @@ class Tls(): else: self.trace('unhandled block id %04x (%d bytes): %s' % (id, sz, hexlify(body))) - def makeTlsFlashBlock(self, id, body): - m=sha256() + def make_tls_flash_block(self, id: int, body: bytes): + m = sha256() m.update(body) hdr = pack('cmd> %s' % hexlify(out).decode()) self.dev.write(1, out) - resp = self.dev.read(129, 100*1024) + resp = self.dev.read(129, 100 * 1024) resp = bytes(resp) self.trace('