From 9dc0c14aedba3ed31eb8dd42380b1b76f754cfe9 Mon Sep 17 00:00:00 2001 From: Arvid Norlander Date: Wed, 5 Aug 2020 18:17:31 +0200 Subject: [PATCH 1/9] Automated formatting with YAPF. PEP8 settings are used except the low 80 column limit has be increased to a more reasonable 100 columns. --- .style.yapf | 394 ++++++++++++++ bin/validity-led-dance | 1 - bin/validity-sensors-firmware | 17 +- dbus_service/dbus-service | 29 +- scripts/factory-reset.py | 1 - scripts/lsdbus.py | 6 +- scripts/prototype.py | 3 +- setup.py | 39 +- validitysensor/blobs.py | 2 + validitysensor/blobs_90.py | 16 +- validitysensor/blobs_97.py | 9 +- validitysensor/blobs_9a.py | 9 +- validitysensor/db.py | 102 ++-- validitysensor/flash.py | 77 ++- validitysensor/generated_tables.py | 846 ++++++++++++++++++++++++++--- validitysensor/hw_tables.py | 21 +- validitysensor/init.py | 7 +- validitysensor/init_db.py | 11 +- validitysensor/init_flash.py | 64 ++- validitysensor/sensor.py | 392 +++++++------ validitysensor/sid.py | 10 +- validitysensor/table_types.py | 53 +- validitysensor/timeslot.py | 217 ++++---- validitysensor/tls.py | 207 +++---- validitysensor/upload_fwext.py | 26 +- validitysensor/usb.py | 23 +- validitysensor/util.py | 5 +- 27 files changed, 1890 insertions(+), 697 deletions(-) create mode 100644 .style.yapf 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..48e12cc 100755 --- a/bin/validity-led-dance +++ b/bin/validity-led-dance @@ -10,7 +10,6 @@ from validitysensor import init from time import sleep from usb import core as usb_core - if __name__ == "__main__": if os.geteuid() != 0: raise Exception('This script needs to be executed as root') diff --git a/bin/validity-sensors-firmware b/bin/validity-sensors-firmware index 998017f..5457947 100755 --- a/bin/validity-sensors-firmware +++ b/bin/validity-sensors-firmware @@ -30,7 +30,8 @@ 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 +39,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 +78,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 +91,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 +111,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..8c40b25 100755 --- a/dbus_service/dbus-service +++ b/dbus_service/dbus-service @@ -64,9 +64,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 +80,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 +90,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 +124,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)) @@ -192,9 +182,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 +216,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/factory-reset.py b/scripts/factory-reset.py index a997d50..9ef5ced 100644 --- a/scripts/factory-reset.py +++ b/scripts/factory-reset.py @@ -1,4 +1,3 @@ - from validitysensor.usb import usb from validitysensor.sensor import factory_reset, RebootException diff --git a/scripts/lsdbus.py b/scripts/lsdbus.py index 637e255..74fec3f 100755 --- a/scripts/lsdbus.py +++ b/scripts/lsdbus.py @@ -16,10 +16,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..f39bfcb 100644 --- a/scripts/prototype.py +++ b/scripts/prototype.py @@ -13,6 +13,7 @@ import code #usb.trace_enabled = True #tls.trace_enabled = True + def identify(): def update_cb(e): print('Capture error: %s, try again' % repr(e)) @@ -21,6 +22,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 +33,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..16550e1 100644 --- a/validitysensor/blobs.py +++ b/validitysensor/blobs.py @@ -1,5 +1,6 @@ from enum import Enum, auto + class Blobs(Enum): init_hardcoded = auto() init_hardcoded_clean_slate = auto() @@ -22,6 +23,7 @@ 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) 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..bafd875 100644 --- a/validitysensor/db.py +++ b/validitysensor/db.py @@ -8,28 +8,34 @@ from .flash import call_cleanups from .sid import * from .winbio_constants import finger_names + class UserStorage(): def __init__(self, dbid, name): - self.dbid=dbid - self.name=name - self.users=[] + 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=[] + self.dbid = dbid + self.identity = identity + self.fingers = [] 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): 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 + 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,7 +153,7 @@ 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): 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 +257,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..0c4a60f 100644 --- a/validitysensor/flash.py +++ b/validitysensor/flash.py @@ -4,12 +4,16 @@ from .util import assert_status, unhex from .blobs import db_write_enable from .hw_tables import flash_ic_table_lookup + 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)) + 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: @@ -20,31 +24,35 @@ class PartitionInfo(): 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) + 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 @@ -59,43 +67,50 @@ class ModuleInfo(): 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): 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(' 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): 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) + class Line(): - mask=None - flags=None - data=None - v0=0 - v1=0 - v2=0 + mask = None + flags = None + data = None + v0 = 0 + v1 = 0 + v2 = 0 + def clip(x): if x < -128: - x=-128 + x = -128 if x > 127: - x=127 + x = 127 return x & 0xff + def scale(x): 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): # Make signed l, r = unpack('bb', pack('BB', l, r)) - return clip(l+r) + return clip(l + r) + def chunks(b, l): - return [b[i:i+l] for i in range(0, len(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'' + 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 @@ -262,10 +289,10 @@ class Sensor(): return bytes(b) def patch_timeslot_again(self, b): - b=bytearray(b) + 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 +303,7 @@ class Sensor(): # Call if opcode == 11: - match = operands[1] # destination address + match = operands[1] # destination address pc += l @@ -303,74 +330,78 @@ 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): 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) + 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 @@ -381,7 +412,7 @@ class Sensor(): # 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 --------------- @@ -392,53 +423,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 --------------- 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,7 +492,6 @@ class Sensor(): if pad > 0: l.data += b'\0' * (4 - pad) - #---------------- Line Update --------------- line_update = pack('> 0x14) > 1]) + update_transform = b''.join([ + pack('> 0x14) > 1 + ]) chunks += [[0x43, update_transform]] return chunks def line_update_type_2(self, mode, chunks): 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: # c[1] = pack(' 0: l.data += b'\0' * (4 - pad) - #---------------- Line Update --------------- line_update = 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('> 3, b[0] & 7) elif b[0] & 0xc0 == 0xc0: @@ -134,18 +134,21 @@ def decode_insn(b): else: raise Exception('Unhandled instruction %02x' % b) + def disassm_timeslot_table(b, off): - pc=off + 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 + pc = 0 while len(b) > 0: op, sz, *_ = decode_insn(b) @@ -153,15 +156,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]) b = b[sz:] pc += sz + def find_nth_regwrite(b, reg_addr, n): - pc=0 + pc = 0 while len(b) > 0: op, sz, *operands = decode_insn(b) @@ -171,7 +175,7 @@ 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]) @@ -185,9 +189,11 @@ def split_chunks(b): p, b = b[:sz], b[sz:] yield [typ, p] + def merge_chunks(cs): return b''.join([pack(' 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): return pack('>H', len(chunk)) + chunk + def with_3bytes_size(chunk): return pack('>BH', len(chunk) >> 16, len(chunk)) + chunk + def with_1byte_size(chunk): 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 + return b + bytes([l - 1]) * l + def unpad(b): - return b[:-1-b[-1]] + return b[:-1 - b[-1]] # TODO assert the right state transitions class Tls(): - def __init__(self, usb): self.usb = usb self.reset() @@ -107,14 +113,14 @@ class Tls(): # 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): 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,19 +130,15 @@ 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) @@ -156,27 +158,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'] @@ -189,7 +192,7 @@ class Tls(): 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): @@ -197,33 +200,33 @@ class Tls(): 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): 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,9 +234,10 @@ 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): @@ -246,8 +250,8 @@ 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): @@ -258,15 +262,16 @@ 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:] @@ -274,11 +279,13 @@ class Tls(): raise Exception('Not expecting any more data') def handle_cert_req(self, p): - (sign_and_hash_algo,), p = unpack('>H', p[:2]), p[2:] + (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') @@ -287,11 +294,12 @@ class Tls(): def handle_server_hello_done(self, p): 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): 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') @@ -325,10 +333,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'' + app_data = b'' while len(rsp) > 0: while len(rsp) < 5: @@ -347,7 +355,7 @@ class Tls(): elif t == 0x14: if pkt != unhexlify('01'): raise Exception('Unexpected ChangeCipherSpec payload') - + self.secure_rx = True elif t == 0x17: @@ -362,36 +370,37 @@ class Tls(): 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): 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 + 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) @@ -410,7 +419,7 @@ class Tls(): self.trace('block id %04x (%d bytes)' % (id, sz)) - m=sha256() + m = sha256() m.update(body) if m.digest() != hs: @@ -432,20 +441,20 @@ class Tls(): self.trace('unhandled block id %04x (%d bytes): %s' % (id, sz, hexlify(body))) def makeTlsFlashBlock(self, id, body): - m=sha256() + 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(' Date: Wed, 5 Aug 2020 18:25:51 +0200 Subject: [PATCH 2/9] Get rid of star imports. These are dangerous as you will import not only the functions defined in the other module but also all of it's imports! This makes automated cleanups brittle. --- bin/validity-led-dance | 4 +++- validitysensor/tls.py | 2 +- validitysensor/usb.py | 2 +- 3 files changed, 5 insertions(+), 3 deletions(-) diff --git a/bin/validity-led-dance b/bin/validity-led-dance index 48e12cc..2e67a2b 100755 --- a/bin/validity-led-dance +++ b/bin/validity-led-dance @@ -5,7 +5,9 @@ import os from binascii import unhexlify from enum import Enum from validitysensor.flash import read_flash -from validitysensor.sensor import * +from validitysensor.sensor import glow_start_scan, glow_end_scan +from validitysensor.util import assert_status +from validitysensor.tls import tls from validitysensor import init from time import sleep from usb import core as usb_core diff --git a/validitysensor/tls.py b/validitysensor/tls.py index dc9cdcf..7be4ff5 100644 --- a/validitysensor/tls.py +++ b/validitysensor/tls.py @@ -3,7 +3,7 @@ import os import pickle import logging from struct import pack, unpack -from binascii import * +from binascii import hexlify, unhexlify from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes from cryptography.hazmat.backends import default_backend diff --git a/validitysensor/usb.py b/validitysensor/usb.py index fd7c2f2..6b80bc9 100644 --- a/validitysensor/usb.py +++ b/validitysensor/usb.py @@ -1,7 +1,7 @@ import logging import errno import usb.core as ucore -from binascii import * +from binascii import hexlify, unhexlify from .util import assert_status from struct import unpack from usb.core import USBError From a17ad521022d2920795955b31f99fc144220fc9b Mon Sep 17 00:00:00 2001 From: Arvid Norlander Date: Wed, 5 Aug 2020 20:27:04 +0200 Subject: [PATCH 3/9] Use optimise imports in PyCharm for cleanup. This sorts imports into sections: * From standard library * From other libraries * From the current project Each section is sorted alphabetically. --- bin/validity-led-dance | 13 +++++-------- bin/validity-sensors-firmware | 5 ++--- dbus_service/dbus-service | 15 +++++++-------- scripts/dbus-cmd.py | 6 +----- scripts/lsdbus.py | 6 ------ scripts/prototype.py | 15 ++++----------- validitysensor/db.py | 7 +++---- validitysensor/flash.py | 5 +++-- validitysensor/init.py | 10 +++++----- validitysensor/init_db.py | 2 +- validitysensor/init_flash.py | 20 ++++++++++---------- validitysensor/sensor.py | 24 +++++++++++++----------- validitysensor/sid.py | 2 +- validitysensor/tls.py | 12 ++++++------ validitysensor/upload_fwext.py | 6 +++--- validitysensor/usb.py | 8 +++++--- validitysensor/util.py | 2 +- 17 files changed, 70 insertions(+), 88 deletions(-) diff --git a/bin/validity-led-dance b/bin/validity-led-dance index 2e67a2b..4c5fc2c 100755 --- a/bin/validity-led-dance +++ b/bin/validity-led-dance @@ -1,16 +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 glow_start_scan, glow_end_scan -from validitysensor.util import assert_status -from validitysensor.tls import tls -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: diff --git a/bin/validity-sensors-firmware b/bin/validity-sensors-firmware index 5457947..d1c77b4 100755 --- a/bin/validity-sensors-firmware +++ b/bin/validity-sensors-firmware @@ -20,14 +20,13 @@ 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/' diff --git a/dbus_service/dbus-service b/dbus_service/dbus-service index 8c40b25..031a089 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' diff --git a/scripts/dbus-cmd.py b/scripts/dbus-cmd.py index d6658db..93b44bc 100755 --- a/scripts/dbus-cmd.py +++ b/scripts/dbus-cmd.py @@ -1,11 +1,7 @@ #!/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 bus = dbus.SystemBus() @@ -13,4 +9,4 @@ o = bus.get_object('net.reactivated.Fprint', '/net/reactivated/Fprint/Manager', o = o.GetDefaultDevice() o = bus.get_object('net.reactivated.Fprint', o, introspect=False) o = dbus.Interface(o, 'net.reactivated.Fprint.Device') -print(o.RunCmd(sys.argv[1])) +print(o.RunCmd(sys.argv[1])) \ No newline at end of file diff --git a/scripts/lsdbus.py b/scripts/lsdbus.py index 74fec3f..9ad1770 100755 --- a/scripts/lsdbus.py +++ b/scripts/lsdbus.py @@ -1,13 +1,7 @@ #!/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 bus = dbus.SystemBus() o = bus.get_object('org.freedesktop.DBus', '/org/freedesktop/DBus') diff --git a/scripts/prototype.py b/scripts/prototype.py index f39bfcb..9ffabe5 100644 --- a/scripts/prototype.py +++ b/scripts/prototype.py @@ -1,14 +1,7 @@ -from validitysensor.tls import tls -from validitysensor.usb import usb -from validitysensor.db import db -from validitysensor.flash import read_flash -from validitysensor.sensor import * -from validitysensor.sid import * -from validitysensor.init import open as open9x -from threading import Condition -from time import sleep -import logging -import code +from binascii import hexlify + +from validitysensor.sensor import sensor + #usb.trace_enabled = True #tls.trace_enabled = True diff --git a/validitysensor/db.py b/validitysensor/db.py index bafd875..3431fe0 100644 --- a/validitysensor/db.py +++ b/validitysensor/db.py @@ -1,11 +1,10 @@ -import typing - -from .tls import tls -from .util import assert_status from binascii import hexlify + from .blobs import db_write_enable from .flash import call_cleanups from .sid import * +from .tls import tls +from .util import assert_status from .winbio_constants import finger_names diff --git a/validitysensor/flash.py b/validitysensor/flash.py index 0c4a60f..2db41a3 100644 --- a/validitysensor/flash.py +++ b/validitysensor/flash.py @@ -1,8 +1,9 @@ -from .tls import tls 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 .tls import tls +from .util import assert_status, unhex class FlashInfo(): diff --git a/validitysensor/init.py b/validitysensor/init.py index 97fb8a4..909c3b8 100644 --- a/validitysensor/init.py +++ b/validitysensor/init.py @@ -1,13 +1,13 @@ import atexit import logging -from validitysensor.usb import usb -from validitysensor.tls import tls -from validitysensor.sensor import sensor, reboot, RebootException from validitysensor.flash import read_tls_flash -from validitysensor.init_flash import init_flash -from validitysensor.upload_fwext import upload_fwext from validitysensor.init_db import init_db +from validitysensor.init_flash import init_flash +from validitysensor.sensor import sensor, reboot, RebootException +from validitysensor.tls import tls +from validitysensor.upload_fwext import upload_fwext +from validitysensor.usb import usb def close(): diff --git a/validitysensor/init_db.py b/validitysensor/init_db.py index d2e37f8..d9b9211 100644 --- a/validitysensor/init_db.py +++ b/validitysensor/init_db.py @@ -1,5 +1,5 @@ -from struct import pack, unpack import logging +from struct import pack, unpack from .db import db diff --git a/validitysensor/init_flash.py b/validitysensor/init_flash.py index 3f45a4a..e427206 100644 --- a/validitysensor/init_flash.py +++ b/validitysensor/init_flash.py @@ -1,21 +1,21 @@ -import os -from struct import pack, unpack -from binascii import unhexlify -import logging - -from hashlib import sha256 import hmac +import logging +import os +from binascii import unhexlify +from hashlib import sha256 +from struct import pack, unpack + +from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.asymmetric import ec from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes -from cryptography.hazmat.backends import default_backend -from .tls import tls, hs_key, crt_hardcoded -from .usb import usb +from .blobs import reset_blob from .flash import write_flash, erase_flash, call_cleanups, PartitionInfo, get_flash_info from .sensor import reboot, RomInfo +from .tls import tls, hs_key, crt_hardcoded +from .usb import usb from .util import assert_status, unhex -from .blobs import reset_blob flash_layout_hardcoded = [ # id type access offset size diff --git a/validitysensor/sensor.py b/validitysensor/sensor.py index 0398ae0..4c220ca 100644 --- a/validitysensor/sensor.py +++ b/validitysensor/sensor.py @@ -1,21 +1,23 @@ +import logging +import os.path import typing +from binascii import hexlify, unhexlify from enum import Enum from hashlib import sha256 -import os.path -import logging +from struct import pack, unpack +from time import sleep + from usb import core as usb_core -from .tls import tls -from .usb import usb, CancelledException + +from . import timeslot as prg +from .blobs import reset_blob from .db import db, SidIdentity from .flash import write_enable, call_cleanups, read_flash, erase_flash, write_flash_all, read_flash_all -from time import sleep -from struct import pack, unpack -from .table_types import SensorTypeInfo, SensorCaptureProg -from binascii import hexlify, unhexlify -from .util import assert_status, unhex from .hw_tables import dev_info_lookup -from .blobs import reset_blob -from . import timeslot as prg +from .table_types import SensorTypeInfo, SensorCaptureProg +from .tls import tls +from .usb import usb, CancelledException +from .util import assert_status, unhex # TODO: this should be specific to an individual device (system may have more than one sensor) calib_data_path = '/usr/share/python-validity/calib-data.bin' diff --git a/validitysensor/sid.py b/validitysensor/sid.py index 1a68a63..47af524 100644 --- a/validitysensor/sid.py +++ b/validitysensor/sid.py @@ -1,5 +1,5 @@ -from struct import unpack, pack import typing +from struct import unpack, pack class SidIdentity(): diff --git a/validitysensor/tls.py b/validitysensor/tls.py index 7be4ff5..e9f9827 100644 --- a/validitysensor/tls.py +++ b/validitysensor/tls.py @@ -1,19 +1,19 @@ import hmac +import logging import os import pickle -import logging -from struct import pack, unpack from binascii import hexlify, unhexlify +from hashlib import sha256 +from struct import pack, unpack -from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes from cryptography.hazmat.backends import default_backend +from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.asymmetric import ec from cryptography.hazmat.primitives.asymmetric.utils import Prehashed -from cryptography.hazmat.primitives import hashes -from hashlib import sha256 +from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes -from .util import unhex from .usb import usb +from .util import unhex password_hardcoded = unhexlify('717cd72d0962bc4a2846138dbb2c24192512a76407065f383846139d4bec2033') gwk_sign_hardcoded = unhexlify('3a4c76b76a97981d1274247e166610e77f4d9c9d07d3c728e532916bdd28b454') diff --git a/validitysensor/upload_fwext.py b/validitysensor/upload_fwext.py index 26509de..dbf012c 100644 --- a/validitysensor/upload_fwext.py +++ b/validitysensor/upload_fwext.py @@ -1,10 +1,10 @@ -from time import ctime import logging from os.path import basename +from time import ctime -from .usb import usb -from .sensor import reboot, write_hw_reg32, read_hw_reg32, identify_sensor from .flash import write_flash_all, write_fw_signature, get_fw_info +from .sensor import reboot, write_hw_reg32, read_hw_reg32, identify_sensor +from .usb import usb firmware_home = '/usr/share/python-validity' diff --git a/validitysensor/usb.py b/validitysensor/usb.py index 6b80bc9..f6cf468 100644 --- a/validitysensor/usb.py +++ b/validitysensor/usb.py @@ -1,11 +1,13 @@ -import logging import errno -import usb.core as ucore +import logging from binascii import hexlify, unhexlify -from .util import assert_status from struct import unpack + +import usb.core as ucore from usb.core import USBError + from .blobs import init_hardcoded, init_hardcoded_clean_slate +from .util import assert_status supported_devices = [ (0x138a, 0x0090), diff --git a/validitysensor/util.py b/validitysensor/util.py index 9a35a8d..1966248 100644 --- a/validitysensor/util.py +++ b/validitysensor/util.py @@ -1,6 +1,6 @@ import re -from struct import unpack from binascii import unhexlify +from struct import unpack def assert_status(b): From 6a5a32d29fb6fbb9f884d48b35b0a4c44b401d01 Mon Sep 17 00:00:00 2001 From: Arvid Norlander Date: Wed, 5 Aug 2020 18:42:34 +0200 Subject: [PATCH 4/9] Various small and easy cleanup found by inspection in PyCharm --- bin/validity-led-dance | 6 +++--- scripts/dbus-cmd.py | 8 +++++--- scripts/prototype.py | 4 ++-- validitysensor/db.py | 10 +++++----- validitysensor/flash.py | 13 ++++++------ validitysensor/hw_tables.py | 4 ++-- validitysensor/init.py | 2 +- validitysensor/init_db.py | 4 ++-- validitysensor/init_flash.py | 2 +- validitysensor/sensor.py | 26 ++++++++++++------------ validitysensor/sid.py | 2 +- validitysensor/timeslot.py | 36 +++++++++++++++++----------------- validitysensor/tls.py | 22 ++++++++++----------- validitysensor/upload_fwext.py | 4 ++-- validitysensor/usb.py | 4 ++-- validitysensor/util.py | 2 +- 16 files changed, 76 insertions(+), 73 deletions(-) diff --git a/bin/validity-led-dance b/bin/validity-led-dance index 4c5fc2c..4612938 100755 --- a/bin/validity-led-dance +++ b/bin/validity-led-dance @@ -22,9 +22,9 @@ if __name__ == "__main__": sleep(0.05) led_script = unhexlify( - '39ff100000ff03000001ff002000000000ffff0000ffff0000ff03000001ff00' \ - '200000000000000000ffff0000ff03000001ff002000000000ffff0000000000' \ - '0000000000000000000000000000000000000000000000000000000000000000' \ + '39ff100000ff03000001ff002000000000ffff0000ffff0000ff03000001ff00' + '200000000000000000ffff0000ff03000001ff002000000000ffff0000000000' + '0000000000000000000000000000000000000000000000000000000000000000' '0000000000000000000000000000000000000000000000000000000000') assert_status(tls.app(led_script)) diff --git a/scripts/dbus-cmd.py b/scripts/dbus-cmd.py index 93b44bc..3774e00 100755 --- a/scripts/dbus-cmd.py +++ b/scripts/dbus-cmd.py @@ -1,12 +1,14 @@ #!/usr/bin/env python3 -import dbus.mainloop.glib -dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) 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() o = bus.get_object('net.reactivated.Fprint', o, introspect=False) o = dbus.Interface(o, 'net.reactivated.Fprint.Device') -print(o.RunCmd(sys.argv[1])) \ No newline at end of file +print(o.RunCmd(sys.argv[1])) diff --git a/scripts/prototype.py b/scripts/prototype.py index 9ffabe5..3a64e40 100644 --- a/scripts/prototype.py +++ b/scripts/prototype.py @@ -3,8 +3,8 @@ from binascii import hexlify from validitysensor.sensor import sensor -#usb.trace_enabled = True -#tls.trace_enabled = True +# usb.trace_enabled = True +# tls.trace_enabled = True def identify(): diff --git a/validitysensor/db.py b/validitysensor/db.py index 3431fe0..6853e1c 100644 --- a/validitysensor/db.py +++ b/validitysensor/db.py @@ -8,7 +8,7 @@ from .util import assert_status from .winbio_constants import finger_names -class UserStorage(): +class UserStorage: def __init__(self, dbid, name): self.dbid = dbid self.name = name @@ -19,7 +19,7 @@ class UserStorage(): self.name), repr(self.users)) -class User(): +class User: def __init__(self, dbid, identity): self.dbid = dbid self.identity = identity @@ -112,7 +112,7 @@ def identity_to_bytes(identity: str): raise Exception('Don' 't know how to handle identity %s' % repr(identity)) -class DbRecord(): +class DbRecord: def __init__(self): self.dbid = 0 self.type = 0 @@ -125,8 +125,8 @@ class DbRecord(): self.dbid, self.type, self.storage, repr(self.value), repr(self.children)) -class Db(): - class Info(): +class Db: + class Info: def __init__(self, total, used, free, records, roots): self.total = total # partition size self.used = used # used (not deleted) diff --git a/validitysensor/flash.py b/validitysensor/flash.py index 2db41a3..de0befb 100644 --- a/validitysensor/flash.py +++ b/validitysensor/flash.py @@ -6,9 +6,10 @@ from .tls import tls from .util import assert_status, unhex -class FlashInfo(): +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 + 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( @@ -20,7 +21,7 @@ class FlashInfo(): # access lvl: # 2 -- write only # 7 -- can read/write even without TLS -class PartitionInfo(): +class PartitionInfo: def __init__(self, id, type, access_lvl, offset, size): self.id, self.type, self.access_lvl, self.offset, self.size = id, type, access_lvl, offset, size @@ -39,7 +40,7 @@ def get_flash_info(): ic = flash_ic_table_lookup(jid0, jid1, blocks * blocksize) - if ic == None: + if ic is None: raise Exception('Unknown flash IC. JEDEC id=%x:%x, size=%dx%d' % (jid0, jid1, blocks, blocksize)) @@ -63,7 +64,7 @@ def get_flash_info(): # 0200 2377 0000 0100 802f0000 # 0200 6637 0100 0c00 f0220200 # 0100 2556 0000 0100 60040000 -class ModuleInfo(): +class ModuleInfo: def __init__(self, type, subtype, major, minor, size): self.type, self.subtype, self.major, self.minor, self.size = type, subtype, major, minor, size @@ -72,7 +73,7 @@ class ModuleInfo(): self.minor, self.size) -class FirmwareInfo(): +class FirmwareInfo: def __init__(self, major, minor, buildtime, modules): self.major, self.minor, self.buildtime, self.modules = major, minor, buildtime, modules diff --git a/validitysensor/hw_tables.py b/validitysensor/hw_tables.py index 868d2ed..7c657a9 100644 --- a/validitysensor/hw_tables.py +++ b/validitysensor/hw_tables.py @@ -1,4 +1,4 @@ -class DeviceInfo(): +class DeviceInfo: def __init__(self, major, type, version, version_mask, name): self.major, self.type, self.version, self.version_mask, self.name = major, type, version, version_mask, name @@ -445,7 +445,7 @@ def dev_info_lookup(major, ver): return fuzzy_match -class FlashIcInfo(): +class FlashIcInfo: def __init__(self, name, size, f18, jid0, jid1, f1b, f1c, f1e, secror_size, sector_erase_cmd, f25, f26): self.name, self.size, self.f18, self.jid0, self.jid1, self.f1b, self.f1c, self.f1e, self.secror_size, self.sector_erase_cmd, self.f25, self.f26 = name, size, f18, jid0, jid1, f1b, f1c, f1e, secror_size, sector_erase_cmd, f25, f26 diff --git a/validitysensor/init.py b/validitysensor/init.py index 909c3b8..a990993 100644 --- a/validitysensor/init.py +++ b/validitysensor/init.py @@ -28,7 +28,7 @@ def close(): def open_common(): init_flash() usb.send_init() - tls.parseTlsFlash(read_tls_flash()) + tls.parse_tls_flash(read_tls_flash()) tls.open() upload_fwext() sensor.open() diff --git a/validitysensor/init_db.py b/validitysensor/init_db.py index d9b9211..4ae1de2 100644 --- a/validitysensor/init_db.py +++ b/validitysensor/init_db.py @@ -32,8 +32,8 @@ def init_machine_guid(machine_guid='e7260876-58db-4d27-8c40-8d13110d6a71'): def init_db(): stg = db.get_user_storage(name='StgWindsor') - if stg == None: + if stg is None: logging.info('Creating a new user storage object') db.new_user_storate() - #init_machine_guid() + # init_machine_guid() diff --git a/validitysensor/init_flash.py b/validitysensor/init_flash.py index e427206..ee46909 100644 --- a/validitysensor/init_flash.py +++ b/validitysensor/init_flash.py @@ -157,7 +157,7 @@ def init_flash(): erase_flash(4) # Persist certs and keys on cert partition. - write_flash(1, 0, tls.makeTlsFlash()) + write_flash(1, 0, tls.make_tls_flash()) # Reboot. # The device will disconnect and our service will be started by udev as soon as it is connected again. diff --git a/validitysensor/sensor.py b/validitysensor/sensor.py index 4c220ca..2e8dec7 100644 --- a/validitysensor/sensor.py +++ b/validitysensor/sensor.py @@ -87,7 +87,7 @@ def factory_reset(): reboot() -class RomInfo(): +class RomInfo: @classmethod def get(cls): rsp = tls.cmd(b'\x01') @@ -173,10 +173,10 @@ def bitpack(b): # convert back to bytes b = b.to_bytes((u * l + 7) // 8, 'little') - return (u, m, b) + return u, m, b -class Line(): +class Line: mask = None flags = None data = None @@ -217,7 +217,7 @@ class CaptureMode(Enum): ENROLL = 3 -class Sensor(): +class Sensor: calib_data = b'' def open(self): @@ -417,7 +417,7 @@ class Sensor(): 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: @@ -448,7 +448,7 @@ class Sensor(): 0x2e, unhexlify('0200180023000000700070004d010000a0008c003c32321e3c0a0202') ]] - #---------------- Interleave --------------- + # ---------------- Interleave --------------- chunks += [[0x44, pack(' 0: l.data += b'\0' * (4 - pad) - #---------------- Line Update --------------- + # ---------------- Line Update --------------- line_update = pack('> 0x14) <= 1]) chunks += [[0x30, line_update]] - #---------------- Line Update Transform --------------- + # ---------------- Line Update Transform --------------- update_transform = b''.join([ pack('> 0x14) > 1 @@ -525,7 +525,7 @@ class Sensor(): tst = self.patch_timeslot_again(tst) c[1] = tst - #---------------- Reply Configuration --------------- + # ---------------- Reply Configuration --------------- chunks += [[0x17, b'']] if mode == CaptureMode.IDENTIFY: @@ -728,7 +728,7 @@ class Sensor(): if error != 0: raise Exception('Scanning problem: %04x' % error) - return (x, y, w1, w2) + return x, y, w1, w2 finally: tls.app(unhexlify('04')) # capture stop if still running, cleanup @@ -788,7 +788,7 @@ class Sensor(): res = res[magic_len + l:] - return (header, template, tid) + return header, template, tid def make_finger_data(self, subtype, template, tid): template = pack('> 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) @@ -158,7 +158,7 @@ def find_nth_insn(b, opcode, n): if op == opcode: n -= 1 if n == 0: - return (pc, b[:sz]) + return pc, b[:sz] b = b[sz:] pc += sz @@ -177,7 +177,7 @@ def find_nth_regwrite(b, reg_addr, n): if addr == reg_addr: n -= 1 if n == 0: - return (pc, b[:sz]) + return pc, b[:sz] b = b[sz:] pc += sz diff --git a/validitysensor/tls.py b/validitysensor/tls.py index e9f9827..388f1ca 100644 --- a/validitysensor/tls.py +++ b/validitysensor/tls.py @@ -86,7 +86,7 @@ def unpad(b): # TODO assert the right state transitions -class Tls(): +class Tls: def __init__(self, usb): self.usb = usb self.reset() @@ -407,7 +407,7 @@ class Tls(): def make_ext(self, id, b): return pack('>H', id) + with_2bytes_size(b) - def parseTlsFlash(self, reply): + def parse_tls_flash(self, reply): while len(reply) > 0: hdr, reply = reply[:4], reply[4:] hs, reply = reply[:0x20], reply[0x20:] @@ -440,20 +440,20 @@ class Tls(): else: self.trace('unhandled block id %04x (%d bytes): %s' % (id, sz, hexlify(body))) - def makeTlsFlashBlock(self, id, body): + def make_tls_flash_block(self, id, body): m = sha256() m.update(body) hdr = pack(' Date: Thu, 6 Aug 2020 10:41:45 +0200 Subject: [PATCH 5/9] Fix potential use of variable before initialisation. --- validitysensor/timeslot.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/validitysensor/timeslot.py b/validitysensor/timeslot.py index 47c811d..265d312 100644 --- a/validitysensor/timeslot.py +++ b/validitysensor/timeslot.py @@ -194,7 +194,9 @@ def merge_chunks(cs): return b''.join([pack(' 0: (typ, sz), b = unpack(' Date: Thu, 6 Aug 2020 11:35:38 +0200 Subject: [PATCH 6/9] Add typing information --- bin/validity-led-dance | 9 ++--- validitysensor/db.py | 32 ++++++++-------- validitysensor/flash.py | 50 +++++++++++++----------- validitysensor/hw_tables.py | 10 ++--- validitysensor/init.py | 2 +- validitysensor/init_db.py | 2 +- validitysensor/init_flash.py | 12 +++--- validitysensor/sensor.py | 65 ++++++++++++++++---------------- validitysensor/table_types.py | 21 ++++++----- validitysensor/timeslot.py | 17 +++++---- validitysensor/tls.py | 69 +++++++++++++++++----------------- validitysensor/upload_fwext.py | 3 +- validitysensor/usb.py | 12 +++--- validitysensor/util.py | 4 +- 14 files changed, 162 insertions(+), 146 deletions(-) diff --git a/bin/validity-led-dance b/bin/validity-led-dance index 4612938..7a56f30 100755 --- a/bin/validity-led-dance +++ b/bin/validity-led-dance @@ -21,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/validitysensor/db.py b/validitysensor/db.py index 6853e1c..e55cbd9 100644 --- a/validitysensor/db.py +++ b/validitysensor/db.py @@ -1,15 +1,17 @@ from binascii import hexlify +from struct import pack, unpack +import typing 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): + def __init__(self, dbid: int, name: str): self.dbid = dbid self.name = name self.users = [] @@ -20,10 +22,10 @@ class UserStorage: class User: - def __init__(self, dbid, identity): + def __init__(self, dbid: int, identity: str): self.dbid = dbid self.identity = identity - self.fingers = [] + self.fingers: typing.List[typing.Mapping[str, int]] = [] def __repr__(self): return '' % (self.dbid, repr( @@ -35,7 +37,7 @@ def subtype_to_string(s: int): return finger_name or 'Unknown' -def parse_user_storage(rsp): +def parse_user_storage(rsp: bytes): rc, = unpack(' typing.Optional[User]: @@ -169,7 +171,7 @@ class Db: else: return parse_user(rsp) - def get_record_value(self, dbid): + def get_record_value(self, dbid: int): rsp = tls.cmd(pack(' bytes: cmd = pack(' 0: chunk, buf = buf[:bs], buf[bs:] @@ -148,13 +154,13 @@ def write_flash_all(partition, ptr, buf): 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)] return b''.join(blocks)[:size] -def write_fw_signature(partition, signature): +def write_fw_signature(partition: int, signature: bytes): rsp = tls.cmd(pack(' typing.Tuple[int, int, int, int]: try: assert_status(tls.app(self.build_cmd_02(mode))) @@ -733,7 +734,7 @@ class Sensor: finally: tls.app(unhexlify('04')) # capture stop if still running, cleanup - def enrollment_update_start(self, key): + def enrollment_update_start(self, key: int) -> int: rsp = tls.app(pack(' 0: @@ -890,7 +891,7 @@ class Sensor: # cleanup, ignore any errors tls.app(unhexlify('6200000000')) - def identify(self, update_cb): + def identify(self, update_cb: typing.Callable[[Exception], None]): while True: try: glow_start_scan() @@ -908,7 +909,7 @@ class Sensor: return self.match_finger() - def get_finger_blobs(self, usrid, subtype): + def get_finger_blobs(self, usrid: int, subtype: int): usr = db.get_user(usrid) fingerids = [f['dbid'] for f in usr.fingers if f['subtype'] == subtype] diff --git a/validitysensor/table_types.py b/validitysensor/table_types.py index 1c1fdde..9c0907d 100644 --- a/validitysensor/table_types.py +++ b/validitysensor/table_types.py @@ -1,18 +1,20 @@ +import typing from binascii import hexlify, unhexlify class SensorTypeInfo: - table = [] + table: typing.List["SensorTypeInfo"] = [] @classmethod - def get_by_type(cls, sensor_type): + def get_by_type(cls, sensor_type: int) -> 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): + 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 @@ -50,10 +52,11 @@ def metric(i, rominfo): 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 @@ -80,7 +83,8 @@ class SensorCaptureProg: if found is not None: return b''.join(found.blobs) - def __init__(self, major, minor, build, u1, dev_type, a0, a1, 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 @@ -88,8 +92,7 @@ class SensorCaptureProg: self.dev_type = dev_type self.a0 = a0 self.a1 = a1 - blobs = [unhexlify(b) for b in blobs] - self.blobs = blobs + self.blobs = [unhexlify(b) for b in blobs] def __repr__(self): blobs = [hexlify(b).decode() for b in self.blobs] diff --git a/validitysensor/timeslot.py b/validitysensor/timeslot.py index 265d312..46d50c2 100644 --- a/validitysensor/timeslot.py +++ b/validitysensor/timeslot.py @@ -1,5 +1,6 @@ from binascii import hexlify from struct import unpack, pack +import typing codes = {} codes[0x0] = "No Operation" @@ -98,7 +99,7 @@ insn_to_string = [ ] -def decode_insn(b): +def decode_insn(b: bytes): if b[0] == 0: return 0, 1 elif b[0] == 1: @@ -135,7 +136,7 @@ def decode_insn(b): raise Exception('Unhandled instruction %02x' % b) -def disassm_timeslot_table(b, off): +def disassm_timeslot_table(b: bytes, off: int): pc = off while len(b) > 0: op, sz, *operands = decode_insn(b) @@ -147,7 +148,7 @@ def disassm_timeslot_table(b, off): pc += sz -def find_nth_insn(b, opcode, n): +def find_nth_insn(b: bytes, opcode: int, n: int): pc = 0 while len(b) > 0: op, sz, *_ = decode_insn(b) @@ -164,7 +165,7 @@ def find_nth_insn(b, opcode, n): pc += sz -def find_nth_regwrite(b, reg_addr, n): +def find_nth_regwrite(b: bytes, reg_addr: int, n: int): pc = 0 while len(b) > 0: op, sz, *operands = decode_insn(b) @@ -183,20 +184,20 @@ def find_nth_regwrite(b, reg_addr, n): 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('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 @@ -76,18 +77,18 @@ def to_bytes(n): return b -def pad(b): +def pad(b: bytes): l = 16 - (len(b) % 16) return b + bytes([l - 1]) * l -def unpad(b): +def unpad(b: bytes): return b[:-1 - b[-1]] # TODO assert the right state transitions class Tls: - def __init__(self, usb): + def __init__(self, usb: Usb): self.usb = usb self.reset() try: @@ -107,7 +108,7 @@ 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' @@ -116,7 +117,7 @@ class Tls: 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) else: @@ -142,15 +143,15 @@ class Tls: 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): @@ -187,7 +188,7 @@ 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() @@ -195,7 +196,7 @@ class Tls: m = unpad(m) return m - def encrypt(self, b): + 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) @@ -204,7 +205,7 @@ class Tls: 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)) @@ -216,7 +217,7 @@ class Tls: self.trace('tls> %02x: %s' % (t, hexlify(b).decode())) hdr = pack('>BBBH', t, 3, 3, len(b)) @@ -240,7 +241,7 @@ class Tls: 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 @@ -254,7 +255,7 @@ class Tls: 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()) @@ -278,7 +279,7 @@ class Tls: if p != b'': raise Exception('Not expecting any more data') - def handle_cert_req(self, p): + 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( @@ -292,24 +293,24 @@ class Tls: 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()) - 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) 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)) @@ -335,7 +336,7 @@ class Tls: self.update_neg(hdr + p) - def parse_tls_response(self, rsp): + def parse_tls_response(self, rsp: bytes): app_data = b'' while len(rsp) > 0: @@ -366,7 +367,7 @@ 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') @@ -374,7 +375,7 @@ class Tls: 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)) @@ -404,10 +405,10 @@ class Tls: 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 parse_tls_flash(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:] @@ -440,7 +441,7 @@ class Tls: else: self.trace('unhandled block id %04x (%d bytes): %s' % (id, sz, hexlify(body))) - def make_tls_flash_block(self, id, body): + def make_tls_flash_block(self, id: int, body: bytes): m = sha256() m.update(body) hdr = pack(' Date: Thu, 6 Aug 2020 11:35:54 +0200 Subject: [PATCH 7/9] Simplify blob design --- validitysensor/blobs.py | 19 +++++-------------- 1 file changed, 5 insertions(+), 14 deletions(-) diff --git a/validitysensor/blobs.py b/validitysensor/blobs.py index 16550e1..edde185 100644 --- a/validitysensor/blobs.py +++ b/validitysensor/blobs.py @@ -1,14 +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: @@ -24,6 +14,7 @@ def __load_blob(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') From 4417e49612bb503ebefede4318bd7383a928e583 Mon Sep 17 00:00:00 2001 From: Arvid Norlander Date: Thu, 6 Aug 2020 11:53:03 +0200 Subject: [PATCH 8/9] New reformatting of code after rebase and changes. --- dbus_service/dbus-service | 8 ++--- scripts/factory-reset.py | 2 +- scripts/lsdbus.py | 1 + scripts/prototype.py | 1 - validitysensor/db.py | 2 +- validitysensor/flash.py | 3 +- validitysensor/sensor.py | 10 +++--- validitysensor/table_types.py | 4 +-- validitysensor/timeslot.py | 2 +- validitysensor/tls.py | 8 ++--- validitysensor/upload_fwext.py | 4 +-- validitysensor/usb.py | 2 +- validitysensor/winbio_constants.py | 56 ++++++++++++++---------------- 13 files changed, 51 insertions(+), 52 deletions(-) diff --git a/dbus_service/dbus-service b/dbus_service/dbus-service index 031a089..82abb4f 100755 --- a/dbus_service/dbus-service +++ b/dbus_service/dbus-service @@ -158,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) diff --git a/scripts/factory-reset.py b/scripts/factory-reset.py index 9ef5ced..377513f 100644 --- a/scripts/factory-reset.py +++ b/scripts/factory-reset.py @@ -1,5 +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 9ad1770..fc66776 100755 --- a/scripts/lsdbus.py +++ b/scripts/lsdbus.py @@ -1,6 +1,7 @@ #!/usr/bin/env python3 import dbus.mainloop.glib + dbus.mainloop.glib.DBusGMainLoop(set_as_default=True) bus = dbus.SystemBus() diff --git a/scripts/prototype.py b/scripts/prototype.py index 3a64e40..7addf46 100644 --- a/scripts/prototype.py +++ b/scripts/prototype.py @@ -2,7 +2,6 @@ from binascii import hexlify from validitysensor.sensor import sensor - # usb.trace_enabled = True # tls.trace_enabled = True diff --git a/validitysensor/db.py b/validitysensor/db.py index e55cbd9..bb21ff1 100644 --- a/validitysensor/db.py +++ b/validitysensor/db.py @@ -1,6 +1,6 @@ +import typing from binascii import hexlify from struct import pack, unpack -import typing from .blobs import db_write_enable from .flash import call_cleanups diff --git a/validitysensor/flash.py b/validitysensor/flash.py index 5030018..54c5240 100644 --- a/validitysensor/flash.py +++ b/validitysensor/flash.py @@ -80,7 +80,8 @@ class ModuleInfo: class FirmwareInfo: - def __init__(self, major: int, minor: int, buildtime: int, modules: typing.Sequence[ModuleInfo]): + 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): diff --git a/validitysensor/sensor.py b/validitysensor/sensor.py index 3a4b91b..a135641 100644 --- a/validitysensor/sensor.py +++ b/validitysensor/sensor.py @@ -408,7 +408,8 @@ class Sensor: return key_line - def line_update_type_1(self, mode: CaptureMode, chunks: typing.List[typing.List[typing.Union[int, bytes]]]): + 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: @@ -511,11 +512,12 @@ class Sensor: return chunks - def line_update_type_2(self, mode: CaptureMode, chunks: typing.List[typing.List[typing.Union[int, bytes]]]): + 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. # 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(' Date: Thu, 6 Aug 2020 12:38:14 +0200 Subject: [PATCH 9/9] Restore imports in interactive environment prototype. --- scripts/prototype.py | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/scripts/prototype.py b/scripts/prototype.py index 7addf46..83853b9 100644 --- a/scripts/prototype.py +++ b/scripts/prototype.py @@ -1,6 +1,16 @@ +import code +import logging from binascii import hexlify -from validitysensor.sensor import sensor +from validitysensor.tls import tls +from validitysensor.usb import usb +from validitysensor.db import db +from validitysensor.flash import read_flash +from validitysensor.sensor import * +from validitysensor.sid import * +from validitysensor.init import open as open9x +from threading import Condition +from time import sleep # usb.trace_enabled = True # tls.trace_enabled = True