From 9dc0c14aedba3ed31eb8dd42380b1b76f754cfe9 Mon Sep 17 00:00:00 2001 From: Arvid Norlander Date: Wed, 5 Aug 2020 18:17:31 +0200 Subject: [PATCH] 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('