First commit

.gitignore

@@ -0,0 +1 @@
+__pycache__

db97.py

@@ -0,0 +1,202 @@
+
+from usb97 import unhex
+from util import assert_status
+from struct import pack, unpack
+from binascii import hexlify, unhexlify
+from sid import *
+
+class UserStorage():
+    def __init__(self, dbid, name):
+        self.dbid=dbid
+        self.name=name
+        self.users=[]
+
+    def __repr__(self):
+        return '<UserStorage: dbid=%04x name=%s users=%s>' % (self.dbid, repr(self.name), repr(self.users))
+
+class User():
+    def __init__(self, dbid, identity):
+        self.dbid=dbid
+        self.identity=identity
+        self.fingers=[]
+
+    def __repr__(self):
+        return '<User: dbid=%04x identity=%s fingers=%s>' % (self.dbid, repr(self.identity), repr(self.fingers))
+
+def parse_user_storage(rsp):
+    assert_status(rsp[:2])
+    rsp=rsp[2:]
+
+    hdr, rsp = rsp[:8], rsp[8:]
+    recid, usercnt, namesz, unknwn = unpack('<HHHH', hdr)
+    usrtab, rsp = rsp[:4*usercnt], rsp[4*usercnt:]
+    name, rsp = rsp[:namesz], rsp[namesz:]
+
+    if len(rsp) > 0:
+        raise Exception('Junk at the end of the storage info response: %s' % unhexlify(rsp).decode())
+
+    storage=UserStorage(recid, name)
+
+    while len(usrtab) > 0:
+        rec, usrtab = usrtab[:4], usrtab[4:]
+        urid, valsz = unpack('<HH', rec)
+        storage.users += [ { 'dbid': urid, 'valueSize': valsz } ]
+
+    return storage
+
+def parse_identity(b):
+    t, b = b[:4], b[4:]
+    t, = unpack('<L', t)
+
+    if t == 3:
+        l, b = b[:4], b[4:]
+        l, = unpack('<L', l)
+        return sid_from_bytes(b[:l])
+
+    raise Exception('Don''t know how to handle identity type %d' % t)
+
+def parse_user(rsp):
+    assert_status(rsp[:2])
+    rsp=rsp[2:]
+
+    hdr, rsp = rsp[:8], rsp[8:]
+    recid, fingercnt, unknwn, identitysz = unpack('<HHHH', hdr)
+    fingertab, rsp = rsp[:8*fingercnt], rsp[8*fingercnt:]
+    identity, rsp = rsp[:identitysz], rsp[identitysz:]
+
+    if len(rsp) > 0:
+        raise Exception('Junk at the end of the user info response: %s' % unhexlify(rsp).decode())
+
+    identity = parse_identity(identity)
+    user=User(recid, identity)
+
+    while len(fingertab) > 0:
+        rec, fingertab = fingertab[:8], fingertab[8:]
+        frid, subtype, stgid, valsz = unpack('<HHHH', rec)
+        user.fingers += [ { 'dbid': frid, 'subtype': subtype, 'storage': stgid, 'valueSize': valsz } ]
+
+    return user
+
+def identity_to_bytes(identity):
+    if isinstance(identity, SidIdentity):
+        b=identity.to_bytes()
+        b = pack('<LL', 3, len(b)) + b
+
+        # May not be neccessary, but windows union has a minimum size of 0x4c bytes 
+        # and search by identity treats same SIDs with different sizes as different keys
+        while len(b) < 0x4c:
+            b += b'\0'
+
+        return b
+    else:
+        raise Exception('Don''t know how to handle identity %s' % repr(identity))
+    
+wtf_hardcoded=unhex('''
+0602000001648445315dac4d9c91edfb3c327ec18aee87a1fd5e0993fe74b56ee9a70f0eb4d65bfc877060728ad26e695be98899ef76b85
+2162cc5e6ccded51e5a26e7e345c4a8e83aafa2e10fa4cc78f0f34575b0565089b4c78ce1ac7fe3744f4721aa73209fa4a8699e7161bfd7
+ee58dee33f1af15ba3b9ce9a469c5c82eda2b2a2f01658ee14118304e865bc2ad415bec9f06757bb3f2263bd0349a3c089380b72a30edf8
+db01db75586c60ddc9e7bedc8cf9006fcc6825b0ec9dc8782b95f53325bfb6a6945644f2803a4c7fd35574d8b7b089102fb072dfa883049
+f9484a789ff763a029f78adfceac92db195573c65986e38aec2135b5da3c0797939154135be5227515602c0148dbcb8c8a3760dad71f877
+471244cf1053a3cc4bc73c437589a1a47e08c260ea40d3259138e754a556a23e09bf926ad90ff36096c27c460ebdd28f20f31e9f42a1925
+26d79b94e7f2c36b66db69946d93edf8fbd516e3ae59a0b81400c39f6bc1feb209924bc04703fd36ab53dd5658b2dd91633823917c37e94
+4e740409c5fd66974ed42b73d5d7db519c7037b1ed32e471750da95299463cb070c80cea4a03bf20351716ee14423f84670c53ef7b998c1
+79b84ae91cff6ca6b33a372e9947eebc873c6a98c6bcda68e5278b6f799a468bd1a33ae07bf717fbfc20f4d019f0fb25a5d6cb75a97fbd9
+b8490e6204d272d93f62ad393307cad3e24d292040e61fce6f1104c42dc4260de15f42971bbab18064b5dd6ceb12a32d9d25614d9326048
+052bc178b5a0aed3f350cbbec021c4b1ac6ed383f7f4eca0e475cf56caefc3edd3cde6526401936cc095b2df8e0008b2d9bd78c82af0a1e
+c0a71f047dd06b4ce6e91886607c55536f0d8712eae4ccb890008ccc27f2d6f6cb9b9bb88ec205019ec722160ffb1c65ca3177cff09c4bd
+ae8c40be5bf91488a600ed199d0aad40510afe580802c56c0241a8b2a836800e8a0d07ae37176934252f92fc0e501aee29e6c3567e37b3e
+1992ff00b561fd1b3ae37caecc3a1fea7720b0806d3b39cf8a3fb46087b43775b9e473e65cb18af1c9d87a5140a5edf8b7843524910d631
+79c436b6a806d100eb066db889768f7da603631d75a6d79e3d0eb1fe9ba55401b828d4a367f695f3922b982be10fdb1f139a0b1438abc91
+1db5da00ed464d80e3086d368dd131985a21610cd485740599afa772b5dc44d1906ea4c13b8e10b8db896fb81b706038a3328cd07aeb049
+e6ec92ecd30044c54eb171e9cf7a0034cd24614616048491f576826af5ca6afe4bc01f86dc01c2ea25efb32bc9b1c8d56cc7d4d86adf8f1
+f3bb6a649f98bae686ef93524721c618fad85a8125a8ac7d72bfeb7de9f590e8f43535e59259a7a1d13a12a1af91c8ba0dd45ddad02e5b8
+6224fc00dcbc143b0e1d2da638d6798fcc340bda8f1e3abf21cadb2161c4c4be7001371dc56f0d112b42b113288effea6cab8c4b610f730
+37643de21f433774d9fb28cfd213a62dcafab8b662b57e77239dc2fb69f026fcb773f09129c8db04b82dce68734516ffcb938d43deeeeda
+a02cc08d7d06f34ec55ba4b224d3270265001df114b946d4238d51cfbfb84cf715b745ec918259c1165ee12ab82b4373eaa2b0c7e68da3e
+b869ff033d5953800012bb006d1cf17dcbbe2fb2e129614351c000cab8ccc7c389df33dbe504a0c1df37868b839ef84a6c04072d69fc866
+33055b2569c9c04f1994758987fc7063b97853c727809185a25cd69bfebf6a4367760fd21e7942b997bb2b7098dbec715769d13413c1c93
+185409e54a2c9a97daf63114c71348f19549fb3a8551afbbb7476842f7045a0ec8e6d97fb4bb6de76090687ab4e9991a31e7625ba7088a3
+263797be4787db02573b72100b3ba117e8897b939548715bcfa11d98861b60fa2a4759630b47cbbf329fbc675c4cfc2f963cc9ba72cbe67
+13aff1851869917ba0b761f6532d1dee95331c4c621bb0513e993f7e6dda6da1e3076829b24093c8338c125aa2f5a0730f9e5631f7a9594
+c13025fde9bb4f2431540a1c743c7fc55fc8909d85847fe71af8a68d2981b0eee84a308e0160fc4226aa679ca9b35d0aacae8240aabd5b2
+17c5e66ac4795596f8742acb9869c3f52a96b1d946ff72f403ba491c1a0e4be778efab9d61a250562d705c2b459fc3a420143daf9355226
+4bb8c7caa64143697961fab1b6af9e852d43a94736e1e75e4b9a7c2bee73598360c7e4c648e47ba9400dbe22b3eea6d9ab87fe677b36242
+57f81d26b87536a642a7840bdccd8fce05c23891a4fa4ce2156fba2856313653207bf225b8114005432824d16a93473deb81721c77196af
+d80cf723e8cdea81aaed19f55a09a8986f2950c49f4eebfb142d5e4815632d93ccb0106c156b47efca25dc3d7f31eb9b2dee0d2d83a51cb
+e5d1270222920f2c87e5acc884c11b0df7a7e1bb49b06fa4f936964d3f2782fa46eec3e430815f82ab0fb5eb82377277f8fe3ae5b33732d
+7d76ba3de284c66fba372f04f9ad6a39fa669a00583313d51c8e6b1eb960bf1e82ae8e0504b2d3035b371d5ecf638c1cc0c1dd67eff286f
+c9ee0950da53f339a85fbbcbacf61fcdadee8cdba0950d23ec71e2239ae7ab5be1bf56364897af990428822d23deed81fbbe622420234de
+d92b8fc3073bfa0a9677cd9e24b9f81278c50584bdc880fa2cd5c5c4b03bd78be872a656f3ef74b30d4e66f1ff091a343f75f096e64ab25
+b64d426862ed3f21dc54c6b6a8cb90b065c7b33b03e97acde7d58a2fa35664ec03ef1f82a30441a021ec008bee9444bbe50ec8ff457d32d
+06a63b9dc84fd82fa110a30dd849bf9f8cb4d3599b9aa09a76ad1b7c52bb972e5b16670670994ef7471f2d738ba9430664b7a1c259745ba
+8b2b9d98e695bad0b59e1fd989475a61570b2b677734c3f1887f9555b760811e20caba91eff59e9fb328561e12749c32f603bd50da5c449
+7393ccd560e88f367a664c5f2f9fe73107e2664a07a0f015877ce085f469dd24be1a20ce9544692e23094650a991e5bd7ab7d3b575d5ef7
+1549fa442fe75a79a6c9f598eb1396e9332878c73f8af9e8f9f0cc11a4b42245acbb421646d6181c8f7f602298042a37d8b57520a1d68d6
+e3d473d5f1897e05db677c7ff6e260048255fdbe35fae42c791ea57ac0f2334e6ab5c650f59f3b27b1af30347314510a29fe19e20ac6e8e
+eb1e84deb687d2dc3d5cf4b03dbe35d165e
+''')
+
+class Db():
+    def __init__(self, tls):
+        self.tls = tls
+
+    def get_user_storage(self, dbid=0, name=''):
+        name=name.encode()
+
+        if len(name) > 0:
+            name += b'\0'
+
+        return parse_user_storage(self.tls.app(pack('<BHH', 0x4b, dbid, len(name)) + name))
+
+    def get_user(self, dbid):
+        return parse_user(self.tls.app(pack('<BHHH', 0x4a, dbid, 0, 0)))
+
+    def lookup_user(self, identity):
+        stg = self.get_user_storage(name='StgWindsor')
+        data = identity_to_bytes(identity)
+        
+        rsp = self.tls.app(pack('<BHHH', 0x4a, 0, stg.dbid, len(data)) + data)
+        rc, = unpack('<H', rsp[:2])
+
+        if rc == 0x04b3:
+            return None
+        else:
+            return parse_user(rsp)
+
+    def del_record(self, dbid):
+        assert_status(self.tls.app(pack('<BH', 0x48, dbid)))
+
+    def new_record(self, parent, typ, storage, data):
+        assert_status(self.tls.app(b'\x45'))
+        assert_status(self.tls.app(wtf_hardcoded))
+        rsp = self.tls.app(pack('<BHHHH', 0x47, parent, typ, storage, len(data)) + data)
+        assert_status(rsp)
+        recid, = unpack('<H', rsp[2:])
+        self.flush_changes()
+        return recid
+
+    def new_user(self, identity):
+        data = identity_to_bytes(identity)
+
+        stg = self.get_user_storage(name='StgWindsor')
+        rec = self.new_record(stg.dbid, 5, stg.dbid, data)
+        return rec
+
+    def new_finger(self, userid, template):
+        stg = self.get_user_storage(name='StgWindsor')
+        # We ask to create an object of type 0xb, 
+        # but because of the magical `wtf_hardcoded` in the new_record() it ends up being 0x6
+        rec = self.new_record(userid, 0xb, stg.dbid, template)
+        return rec
+
+    def flush_changes(self):
+        assert_status(self.tls.app(b'\x1a'))
+
+    def dump_all(self):
+        stg = self.get_user_storage(name='StgWindsor')
+        usrs = [self.get_user(u['dbid']) for u in stg.users]
+        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 (WINBIO_FINGER_UNSPECIFIED_POS_%02d)' % (f['dbid'], f['subtype'], f['subtype'] - 0xf5 + 1))
+

prototype.py

@@ -0,0 +1,208 @@
+
+from util import assert_status
+from time import sleep
+from struct import pack, unpack
+from binascii import hexlify, unhexlify
+from tls97 import Tls
+from usb97 import Usb, unhex
+from db97 import *
+from sid import *
+
+usb = Usb()
+tls = Tls(usb)
+db = Db(tls)
+
+def open97():
+    usb.open()
+    tls.open()
+    #usb.trace_enabled = True
+    #tls.trace_enabled = True
+
+
+def glow_start_scan():
+    cmd=unhexlify('3920bf0200ffff0000019900200000000099990000000000000000000000000020000000000000000000000000ffff000000990020000000000000000000000000000000000000002000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000')
+    assert_status(tls.app(cmd))
+
+def glow_end_enroll():
+    cmd=unhexlify('39f4010000f401000001ff002000000000ffff0000000000000000000000000020000000000000000000000000f401000000ff0020000000000000000000000000000000000000002000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000')
+    assert_status(tls.app(cmd))
+
+# FIXME this must be very specific to a particular device:
+def start_enroll_prg():
+    cmd=unhex('''
+02980000002300000020000800002000800000010032007000000000802020050024200000502077362820010030200100082170000c210
+000482102004c210000582000005c20000060200000682005006c20012970200121742001887820018084202000942001809c200902a020
+0b19b4200000b8203b04bc201400c0200200c4200100c82002003300100000000080cc200000f503d0200000a1013200440000000080dc2
+0e803e0206401e420d002e8200001f0200500f8200500fc200000b8203b0000080400140800000808000008080000140830000808000014
+0831001c081a0032000c0000000080501101004c1126003400080310071d10071d10071d10071d10071c01065810080101000007c8078c0
+6100000204f80007f000003070107010c07032c08fc80095a800afc08fb800b5a095b800afb08fa800b5b095c800afa08f9800b5c095d80
+0af908f8800b5d095e800af808f7800b5e095f800af708f6800b5f0960800af608f5800b600961800af508f4800b610962800af408f3800
+b620963800af308f2800b630964800af208f1800b640965800af108f0800b650966800af008ef800b660967800aef08ee800b670968800a
+ee08ed800b68096c800aed08ec800b6c096d800aec08eb800b6d096e800aeb08ea800b6e096f800aea08e9800b6f0970800ae908e8800b7
+00971800ae808e7800b710972800ae708e6800b720973800ae608e5800b730974800ae508e4800b740975800ae408e3800b750976800ae3
+08e2800b760977800ae208e1800b770978800ae108e0800b780979800ae008df800b79097a800adf08de800b7a097b800ade08dd800b7b0
+97c800add08dc800b7c097d800adc08db800b7d097e800adb08da800b7e097f800ada08d9800b7f0980800ad908d8800b800981800ad808
+d7800b810982800ad708d6800b820983800ad608d5800b830984800ad508d4800b840985800ad408d3800b850986800ad308d2800b86098
+7800ad208d1800b870988800ad108d0800b880989800ad008cf800b89098a800acf08ce800b8a098b800ace08cd800b8b098c800acd08cc
+800b8c098d800acc08cb800b8d098e800acb08ca800b8e098f800aca08c9800b8f0990800ac908c8800b900991800ac808c7800b9109928
+00ac708c6800b920993800ac608c5800b930994800ac508c4800b940995800ac408c3800b950996800ac308c2800b960997800ac208c180
+0b970998800ac108c0800b980999800ac008bf800b99099a800abf08be800b9a099b800abe08bd800b9b099c800abd08bc800b9c099d800
+abc08bb800b9d099e800abb08ba800b9e099f800aba08b9800b9f09a0800ab908b8800ba00801800ab808b7800a010802800ab708b6800a
+020803800ab608b5800a030804802003070404020000000000002f000400900000002900040000000000350004001000000017000000260
+02800fbb20f00f2220f00300000006001020040010a00018000000a0200000b19000050c360ea010910002e001c00020018002300000090
+0090004d01000090017c013c323232640a02013000cc0103000000ff0000001d000003ff00000025000003ff00000022000003101112131
+415161718191a1b1c1d1e1f202122232425262728292a2b2c2d2e2f303132333435363738393a3b3c3d3e3f404142434445464748494a4b
+4c4d4e4f505152535455565758595a5b5c5d5e5f606162636465666768696a6b6c6d6e6f707172737475767778797a7b7c7d7e7f8081828
+38485868788898a8b8c8d8e8f909192939495969798999a9b9c9d9e9f2b23203c2d182e1e30182e1c321d341d341e321c301e1e241e201f
+201d1c321a301e1c211e21341f1e202024201f1e20201f212221221d221e23341e1d1e1d20341f1d193b341c1d1e35201e201c20221f341
+c1e1e1c221f201d21201e1c1f34242221201f20221f201e241e241d2020221e2420231d221e211e1f1e1e341c321e3220301d2d302f2d2c
+2b23223a211c7e7f807f8080808080808080808080808080808080808180818081808180808080808180818080808180818081808180818
+08180818081808180808081808180808081807f808081808080818081808180808081808180818081808180818080808180818081808180
+81808180818081808180818081808180818081808080808080808080807f807f807f807f7f7e7e
+''')
+    assert_status(tls.app(cmd))
+
+
+def wait_for_finger():
+    while True:
+        b=usb.wait_int()
+        if b[0] == 2:
+            break
+
+def get_prg_status():
+    return tls.app(unhexlify('5100000000'))
+
+def wait_till_finished():
+    while True:
+        status = get_prg_status()
+        
+        if status[0] in [0, 7]:
+            break
+
+        sleep(0.2)
+
+def stop_prg():
+    return tls.app(unhexlify('5100200000'))
+
+def capture_enroll():
+    start_enroll_prg()
+
+    b=usb.wait_int()
+    if b[0] != 0:
+        raise Exception('Unexpected interrupt type' % hexlify(b).decode())
+
+    wait_for_finger()
+    wait_till_finished()
+
+    res=stop_prg()
+    
+    b=usb.wait_int()
+    if b[0] != 3:
+        raise Exception('Unexpected interrupt type %s' % hexlify(b).decode())
+
+    assert_status(res)
+    res = res[2:]
+    
+    l, res = res[:4], res[4:]
+    l, = unpack('<L', l)
+
+    if l != len(res):
+        raise Exception('Response size does not match %d != %d', l, len(res))
+
+    x, y, w1, w2, error = unpack('<HHHHL', res)
+
+    return error
+
+def append_new_image(key=0, prev=b''):
+    rsp=tls.app(pack('<BLL', 0x68, key, 0))
+    assert_status(rsp)
+    new_key, = unpack('<L', rsp[2:])
+
+    usb.wait_int()
+
+    rsp=tls.app(b'\x6b' + prev)
+    assert_status(rsp)
+    
+    usb.wait_int()
+
+    rsp=tls.app(b'\x6b' + prev)
+    assert_status(rsp)
+    res=rsp[2:]
+
+    rsp=tls.app(unhexlify('6900000000'))
+    assert_status(rsp)
+
+    l, res = res[:2], res[2:]
+    l, = unpack('<H', l)
+    if l != len(res):
+        raise Exception('Response size does not match %d != %d', l, len(res))
+
+    # FIXME check how it's done rather than using a hardcoded offsets
+    res, new = res[:0x6c], res[0x6c:]
+
+    return (new_key, res, new)
+
+def parse_template(subtype, template):
+    # This number is very odd. It seems to be always a multiple of 0x10,
+    # which probably means that this is not a size of the plain text.
+    # On the other hand it is exactly 0x30 bytes less than the size of the payload.
+    # If we assume algos are all the same, 0x30 could be 0x10 for AES IV and
+    # 0x20 for SHA256 MAC
+    ciphertext_size, = unpack('<H', template[2:4])
+    template_size = 8+ciphertext_size+0x30
+    template, rest = template[:template_size], template[template_size:]
+
+    template = pack('<H', len(template)) + template
+
+    # FIXME this is most likely wrong, need to do a proper code dive to figure out
+    # how enrollment update response is handled in the dll
+    hs, rest = rest[-0x20:], rest[:-0x20]
+    hs = pack('<H', len(hs)) + hs
+
+    tinfo = pack('<H', 1) + template
+    tinfo += pack('<H', 2) + hs
+
+    tinfo = pack('<HHHH', subtype, 3, len(tinfo), 0x20) + tinfo
+    tinfo += b'\0' * 0x20
+
+    return tinfo
+
+def enroll(identity, subtype):
+    key=0
+    template=b''
+
+    print('Waiting for a finger...')
+
+    while True:
+        glow_start_scan()
+
+        err = capture_enroll()
+        if err != 0:
+            print('Error %08x, try again' % err)
+            continue
+        
+        key, rsp, template = append_new_image(key, template)
+
+        print('Progress: %d %% done' % rsp[0x3c])
+        
+        if rsp[0x3c] == 100:
+            break
+
+    # FIXME check for duplicates
+
+    tinfo = parse_template(subtype, template)
+
+    usr=db.lookup_user(identity)
+    if usr == None:
+        usr = db.new_user(identity)
+    else:
+        usr = usr.dbid
+    
+    recid = db.new_finger(usr, tinfo)
+
+    glow_end_enroll()
+
+    print('All done')
+
+    return recid
+

sid.py

@@ -0,0 +1,42 @@
+
+from struct import unpack, pack
+
+class SidIdentity():
+    def __init__(self, revision, auth, subauth):
+        self.revision = revision
+        self.auth = auth
+        self.subauth = subauth
+
+    def to_bytes(self):
+        b=pack('>BBHL', self.revision, len(self.subauth), self.auth >> 32, self.auth & 0xffffffff)
+        for i in self.subauth:
+            b += pack('<L', i)
+
+        return b
+
+    def __repr__(self):
+        return 'S-%d-%d-%s' % (self.revision, self.auth, '-'.join(map(str, self.subauth)))
+
+def sid_from_bytes(b):
+    revision = b[0]
+    subcnt = b[1]
+    auth = 0
+
+    for i in b[2:8]:
+        auth <<= 8
+        auth |= i
+    
+    subauth=unpack('<%dL' % subcnt, b[8:])
+
+    return SidIdentity(revision, auth, subauth)
+
+def sid_from_string(s):
+    parts = s.split('-')
+
+    if parts[0] != 'S':
+        raise Exception('Expected "S" as a first part')
+
+    parts = list(map(int, parts[1:]))
+    revision, auth, subauth = parts[0], parts[1], parts[2:]
+
+    return SidIdentity(revision, auth, subauth)

tls97.py

@@ -0,0 +1,435 @@
+import re
+import hmac
+import sys
+from hashlib import sha256, md5, sha1
+from binascii import *
+from struct import pack, unpack
+from Crypto.Cipher import AES
+from Crypto.Random import get_random_bytes
+from fastecdsa.curve import P256
+from fastecdsa.point import Point
+from fastecdsa.keys import gen_private_key, get_public_key
+from fastecdsa.ecdsa import sign
+from fastecdsa.encoding.der import DEREncoder
+
+
+# Info about the host computer
+product_name, serial_number='VirtualBox', '0'
+
+hw_key=b'GWK' + \
+    bytes(product_name, 'ascii') + b'\0' + \
+    bytes(serial_number, 'ascii') + b'\0'
+
+password_hardcoded=unhexlify('717cd72d0962bc4a2846138dbb2c24192512a76407065f383846139d4bec2033')
+gwk_sign_hardcoded=unhexlify('47574b5f5349474e3a4c76b76a97981d1274247e166610e77f4d9c9d07d3c728e532916bdd28b454')
+
+def hmac2(key, msg):
+    p1=hmac.new(key, msg, sha256).digest()
+    p2=hmac.new(key, p1+msg, sha256).digest()
+    return p2
+
+# pre-TLS keys
+psk_encryption_key=hmac2(password_hardcoded, hw_key)
+psk_validation_key=hmac2(psk_encryption_key, gwk_sign_hardcoded)
+
+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''
+    while n:
+        b += (n & 0xff).to_bytes(1, 'big')
+        n >>= 8
+
+    return b
+
+def pad(b):
+    l = 16 - (len(b) % 16)
+    return b + bytes([l-1])*l
+
+def unpad(b):
+    return b[:-1-b[-1]]
+
+def prf(secret, seed, length):
+    n = (length + 0x20 - 1) // 0x20
+    
+    res = b''
+    a = hmac.new(secret, seed, sha256).digest()
+
+    while n > 0:
+        res += hmac.new(secret, a+seed, sha256).digest()
+        a = hmac.new(secret, a, sha256).digest()
+        n -= 1
+
+    return res[:length]
+
+    
+# TODO assert the right state transitions
+class Tls():
+    
+    def __init__(self, usb):
+        self.usb = usb
+        self.trace_enabled = False
+
+    def open(self):
+        self.parse40(self.usb.cmd(unhexlify('40010100000000000000100000')))
+
+        self.secure_rx = False
+        self.secure_tx = False
+        self.handshake_hash = sha256()
+
+        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()))
+
+        self.parse_tls_response(rsp)
+
+    def trace(self, s):
+        if self.trace_enabled:
+            print(s)
+
+    def app(self, b):
+        return self.parse_tls_response(self.usb.cmd(self.make_app_data(b)))
+
+    def update_neg(self, b):
+        self.handshake_hash.update(b)
+
+    def make_keys(self):
+        #self.session_private=0x2E38AFE3D563398E5962D2CDEA7FE16D3CFEA36656A9DEC412C648EE3A232D21
+        self.session_private = gen_private_key(P256)
+        self.session_public = get_public_key(self.session_private, P256)
+
+        pre_master_secret = self.session_private*self.ecdh_q
+        pre_master_secret = pre_master_secret.x
+        pre_master_secret = to_bytes(pre_master_secret)[::-1]
+
+        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.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]
+
+    def decrypt(self, c):
+        iv, c = c[:0x10], c[0x10:]
+        aes=AES.new(self.decryption_key, AES.MODE_CBC, iv)
+        m=aes.decrypt(c)
+        m=unpad(m)
+        return m
+
+    def encrypt(self, b):
+        #iv = unhexlify('454849acdd075174d6b9e713a957c2e7')
+        iv = get_random_bytes(0x10)
+        aes=AES.new(self.encryption_key, AES.MODE_CBC, iv)
+        b=pad(b)
+        c=aes.encrypt(b)
+        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()
+
+        if sig != hs:
+            raise Exception('Packet signature validation check failed')
+
+        self.trace('<tls< %02x: %s' % (t, hexlify(b).decode()))
+        return b
+        
+    def sign(self, t, b):
+        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()
+        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)
+        return b'\x14' + with_3bytes_size(verify_data)
+
+    def make_change_cipher_spec(self):
+        return unhexlify('140303000101')
+
+    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
+        return self.with_neg_hdr(0x0b, cert)
+
+    def with_neg_hdr(self, t, b):
+        b = pack('>B', t) + with_3bytes_size(b)
+        self.update_neg(b)
+        return b
+
+    def make_client_kex(self):
+        b = b'\x04' + to_bytes(self.session_public.x)[::-1] + to_bytes(self.session_public.y)[::-1]
+        return self.with_neg_hdr(0x10, b)
+
+    def make_cert_verify(self):
+        buf=self.handshake_hash.copy().digest()
+        s=sign(hexlify(buf).decode(), self.priv_key, prehashed=True)
+        b=DEREncoder().encode_signature(s[0], s[1])
+        return self.with_neg_hdr(0x0f, b)
+
+    def handle_server_hello(self, p):
+        if p[:2] != unhexlify('0303'):
+            raise Exception('unexpected TLS version %s' % hexlify(p[:2]).decode())
+
+        p = p[2:]
+
+        self.server_random, p = p[:0x20], p[0x20:]
+        l = p[0]
+        self.server_sessid, p = p[1:1+l], p[1+l:]
+
+        (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])
+
+        p = p[1:]
+
+        if p != b'':
+            raise Exception('Not expecting any more data')
+
+    def handle_cert_req(self, p):
+        (sign_and_hash_algo,), p = unpack('>H', p[:2]), p[2:]
+        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)
+
+        (l,), p = unpack('>H', p[:2]), p[2:]
+        if l != 0:
+            raise Exception('Server requested a cert with non-empty list of CAs')
+
+        if p != b'':
+            raise Exception('Not expecting any more data')
+
+    def handle_server_hello_done(self, p):
+        if p != b'':
+            raise Exeception('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)
+        if verify_data != b:
+            raise Exception('Final handshake check failed')
+
+    def handle_app_data(self, b):
+        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):
+        if self.secure_rx:
+            handshake = self.validate(0x16, self.decrypt(handshake))
+
+        while len(handshake) > 0:
+            while len(handshake) < 4:
+                handshake += b'\0'
+
+            hdr, handshake = handshake[:4], handshake[4:]
+            t, l12, l3 = unpack('>BHB', hdr)
+            l = (l12 << 8) | l3
+            p, handshake = handshake[:l], handshake[l:]
+
+            if t == 2:
+                self.handle_server_hello(p)
+            elif t == 0xd:
+                self.handle_cert_req(p)
+            elif t == 0xe:
+                self.handle_server_hello_done(p)
+            elif t == 0x14:
+                self.handle_finish(p)
+            else:
+                raise Exception('Unknown handshake packet %02x' % t)
+
+            self.update_neg(hdr+p)
+
+    def parse_tls_response(self, rsp):
+        app_data=b''
+
+        while len(rsp) > 0:
+            while len(rsp) < 5:
+                rsp += b'\0'
+
+            hdr, rsp = rsp[:5], rsp[5:]
+            t, mj, mn, sz = unpack('>BBBH', hdr)
+            pkt, rsp = rsp[:sz], rsp[sz:]
+
+            if mj != 3 or mn != 3:
+                raise Exception('Unexpected TLS version %d %d' % (mj, mn))
+
+            if t == 0x16:
+                self.handle_handshake(pkt)
+
+            elif t == 0x14:
+                if pkt != unhexlify('01'):
+                    raise Exception('Unexpected ChangeCipherSpec payload')
+                
+                self.secure_rx = True
+
+            elif t == 0x17:
+                app_data += self.handle_app_data(pkt)
+
+            else:
+                raise Exception('Dont know how to handle message type %02x' % t)
+
+        return app_data
+
+    def make_app_data(self, b):
+        if not self.secure_tx:
+            raise Exception('App payload before secure connection established')
+
+        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))
+
+        return unhexlify('160303') + with_2bytes_size(b)
+
+    def make_client_hello(self):
+        h = unhexlify('0303') # TLS 1.2
+        #self.client_random = unhexlify('bc349559ac16c8f8362191395b4d04a435d870315f519eed8777488bc2b9600c')
+        self.client_random = get_random_bytes(0x20)
+        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
+        h += with_2bytes_size(suits)
+
+        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
+        # h += with_2bytes_size(exts)
+        h += pack('>H', len(exts)-2) + exts # -2? WHY?!...
+
+        return self.with_neg_hdr(0x01, h)
+
+    def make_ext(self, id, b):
+        return pack('>H', id) + with_2bytes_size(b)
+
+    def parse40(self, reply):
+        h, reply = reply[:8], reply[8:]
+
+        if h != unhexlify('0000001000000000'):
+            raise Exception('unexpected header: %s' % hexlify(h).decode())
+
+        while len(reply) > 0:
+            hdr, reply = reply[:4], reply[4:]
+            hs, reply = reply[:0x20], reply[0x20:]
+            id, sz = unpack('<HH', hdr)
+            body, reply = reply[:sz], reply[sz:]
+
+            if id == 0xffff:
+                break
+
+            if id == 4:
+                self.handle_priv(body)
+            elif id == 6:
+                self.handle_ecdh(body)
+            elif id == 3:
+                self.handle_cert(body)
+            elif id == 1:
+                self.handle_empty(body)
+            elif id == 2:
+                self.handle_empty(body)
+            else:
+                self.trace('unhandled block id %04x (%d bytes): %s' % (id, sz, hexlify(body)))
+
+            m=sha256()
+            m.update(body)
+
+            if m.digest() != hs:
+                raise Exception('hash mismatch')
+
+    def handle_empty(self, body):
+        if body != b'\0' * len(body):
+            raise Exception('Expected empty block')
+
+    def handle_cert(self, body):
+        # TODO validate cert, check if pub keys match
+        self.tls_cert = body
+        self.trace('TLS cert pub key: %s' % hexlify(self.tls_cert))
+
+    def handle_ecdh(self, body):
+        # TODO check the signature
+        x = body[0x8:0x8+0x20]
+        y = body[0x4c:0x4c+0x20]
+
+        x, y = [int(hexlify(i[::-1]), 0x10) for i in [x, y]]
+
+        if not P256.is_point_on_curve( (x, y) ):
+            raise Exception('Point is not on the curve')
+
+        self.ecdh_q = Point(x, y, P256)
+        self.trace('ECDH params:')
+        self.trace('x=0x%x' % x)
+        self.trace('y=0x%x' % y)
+        
+
+    def handle_priv(self, body):
+        prefix, body = body[0], body[1:]
+        if prefix != 2:
+            raise Exception('Unknown private key prefix %02x' % prefix)
+
+        c, hs = body[:-0x20], body[-0x20:]
+        sig=hmac.new(psk_validation_key, c, sha256).digest()
+        if hs != sig:
+            raise Exception('Signature verification failed. This device was probably paired with another computer.')
+        
+        iv, c = c[:AES.block_size], c[AES.block_size:]
+        aes=AES.new(psk_encryption_key, AES.MODE_CBC, iv)
+        m=aes.decrypt(c)
+
+        x, m = m[:0x20], m[0x20:]
+        y, m = m[:0x20], m[0x20:]
+        d, m = m[:0x20], m[0x20:]
+
+        x, y, d = [int(hexlify(i[::-1]), 0x10) for i in [x, y, d]]
+
+        if not P256.is_point_on_curve( (x, y) ):
+            raise Exception('Point is not on the curve')
+
+        # TODO check if the priv key belogs to this public key
+
+        self.trace('Private key:')
+        self.trace('x=0x%x' % x)
+        self.trace('y=0x%x' % y)
+        self.trace('d=0x%x' % d)
+
+        self.pub_key = Point(x, y, P256)
+        self.priv_key = d
+

usb97.py

@@ -0,0 +1,56 @@
+
+import re
+import usb.core
+from binascii import *
+
+def unhex(x):
+    return unhexlify(re.sub('\W', '', x))
+
+init_hardcoded=unhex('''
+06020000015cb560afa595d0dcf4fca09ebb69301e2b9e24a5dfb6f2602833427d3bd65222c418ff15b54587ab2854c4fe9aea74fa55567
+2036fa3ec73c6912c58b1b49cbdcc7664165482cd70d5bb95d9d469626fdc2e012d74157c5792260968a2a4572b1ccccc26ec2e2ed0ddda
+1b0931d3d0566460915d43e0a654a0582706be9113ea88f0c15ba058507efb0dadecb4a4cc64444dddf050b6e8d4ebb34ebba33e8651c75
+e5fb28f85c83197df1de460d9e1cb82208753ceff0ef60a823dba75d05548f5b3a5a0e2972232f7403bd6869da90e5371a0ab8ad23972f1
+597630f5ff7c8b8272800563477288b5591bbb0341d3975efc1778225767fa35480ff7f8dd633e4034ac32e4af58b86ebc63552cb35b12b
+285255deaf3a32bf46cdc5ad3bc1c9ed1bcc112c72143f9aec568e2cacfa89ba0c7bb65590d8b93e6871a33c6c6983c0acd04e737ff55ee
+e024ca6b9a48332c1a69a5a3fdd24b964cf7e7c55229bb0b48a6e339eb2c42d07ec850a4ee780660ad6c77ffa302a63bd19426134c4533d
+69192ef2e16591df26394791a4ecb994a24f5a7f70f1eb2604e6bfb67a452cb74ead8b0d9808f890ac386750cbac06ee03a852145094053
+b2b074b9905de5cdbf2272b67e51f159164778d6d2ef7a1ccb81df9f896ddb38ce11e8140cf6cb9c82
+''')
+
+class Usb():
+    def __init__(self):
+        self.trace_enabled = False
+
+    def open(self):
+        self.dev = usb.core.find(idVendor=0x138a, idProduct=0x0097)
+        self.dev.default_timeout = 5000
+        self.dev.set_configuration()
+
+        # TODO analyse responses
+        self.cmd(unhexlify('01'))
+        self.cmd(unhexlify('19'))
+        self.cmd(unhexlify('4302'))
+        # TODO is this one always the same?
+        self.cmd(init_hardcoded)
+
+        self.cmd(unhexlify('3e'))
+
+    def cmd(self, out):
+        self.trace('>cmd> %s' % hexlify(out).decode())
+        self.dev.write(1, out)
+        resp = self.dev.read(129, 100*1024)
+        resp = bytes(resp)
+        self.trace('<cmd< %s' % hexlify(resp).decode())
+        return resp
+
+    def wait_int(self):
+        resp = self.dev.read(131, 1024, timeout=0)
+        resp = bytes(resp)
+        self.trace('<int< %s' % hexlify(resp).decode())
+        return resp
+
+    def trace(self, s):
+        if self.trace_enabled:
+            print(s)
+

util.py

@@ -0,0 +1,8 @@
+
+from struct import unpack
+
+def assert_status(b):
+    s,=unpack('<H', b[:2])
+    if s != 0:
+        raise Exception('Failed: %04x' % s)
+