diff --git a/debian/python3-validity.service b/debian/python3-validity.service index 93c8c9f..e21fd2b 100644 --- a/debian/python3-validity.service +++ b/debian/python3-validity.service @@ -4,7 +4,7 @@ After=open-fprintd.service [Service] Type=simple -ExecStart=/usr/lib/python-validity/dbus-service --debug +ExecStart=/usr/lib/python-validity/dbus-service Restart=no [Install] diff --git a/debian/python3-validity.udev b/debian/python3-validity.udev index d9ca628..e9078d1 100644 --- a/debian/python3-validity.udev +++ b/debian/python3-validity.udev @@ -4,6 +4,7 @@ ENV{DEVTYPE}!="usb_device", GOTO="python_validity_end" ATTRS{idVendor}=="138a", ATTRS{idProduct}=="0090", GOTO="python_validity_match" ATTRS{idVendor}=="138a", ATTRS{idProduct}=="0097", GOTO="python_validity_match" ATTRS{idVendor}=="06cb", ATTRS{idProduct}=="009a", GOTO="python_validity_match" +ATTRS{idVendor}=="047d", ATTRS{idProduct}=="8054", GOTO="python_validity_match" GOTO="python_validity_end" diff --git a/validitysensor/init.py b/validitysensor/init.py index 1153cf7..1219b08 100644 --- a/validitysensor/init.py +++ b/validitysensor/init.py @@ -9,6 +9,7 @@ from validitysensor.tls import tls from validitysensor.upload_fwext import upload_fwext from validitysensor.usb import usb +from validitysensor import tudor def close(): @@ -27,6 +28,10 @@ def close(): def open_common(): + if tudor.present(): + tudor.open_session() + return + init_data_dir() init_flash() usb.send_init() @@ -46,6 +51,10 @@ def open_common(): def open(): + if tudor.present(): + tudor.open_session() + return + usb.open() open_common() diff --git a/validitysensor/tls.py b/validitysensor/tls.py index ec634c3..b0ea9e0 100644 --- a/validitysensor/tls.py +++ b/validitysensor/tls.py @@ -4,7 +4,7 @@ import pickle import typing from binascii import hexlify, unhexlify -from hashlib import sha256 +from hashlib import sha256, sha384 from struct import pack, unpack from cryptography.hazmat.backends import default_backend @@ -12,6 +12,7 @@ from cryptography.hazmat.primitives.asymmetric import ec from cryptography.hazmat.primitives.asymmetric.utils import Prehashed from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes +from cryptography.hazmat.primitives.ciphers.aead import AESGCM from .usb import usb, Usb from .util import unhex @@ -35,15 +36,16 @@ crypto_backend = default_backend() -def prf(secret: bytes, seed: bytes, length: int): - n = (length + 0x20 - 1) // 0x20 +def prf(secret: bytes, seed: bytes, length: int, hashf=sha256): + bs = hashf().digest_size + n = (length + bs - 1) // bs res = b'' - a = hmac.new(secret, seed, sha256).digest() + a = hmac.new(secret, seed, hashf).digest() while n > 0: - res += hmac.new(secret, a + seed, sha256).digest() - a = hmac.new(secret, a, sha256).digest() + res += hmac.new(secret, a + seed, hashf).digest() + a = hmac.new(secret, a, hashf).digest() n -= 1 return res[:length] @@ -153,18 +155,33 @@ def app(self, b: typing.Union[bytes, typing.Callable[[], bytes]]): def update_neg(self, b: bytes): self.handshake_hash.update(b) + if hasattr(self, + 'cv_hash'): # parallel hash for CertificateVerify (may differ from PRF hash) + self.cv_hash.update(b) def make_keys(self): + hf = getattr(self, 'prf_hash', sha256) skey = ec.generate_private_key(ec.SECP256R1(), crypto_backend) 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, hf) + if getattr(self, 'gcm', False): + # AES-256-GCM (RFC 5288): key_block = client_key(32) server_key(32) + # client_salt(4) server_salt(4); no MAC keys. GCM IV = salt || explicit_nonce(8). + self.tx_seq = 0 + self.rx_seq = 0 + key_block = prf(self.master_secret, b'key expansion' + seed, 2 * 0x20 + 2 * 4, hf) + self.encryption_key = key_block[0:0x20] + self.decryption_key = key_block[0x20:0x40] + self.client_salt = key_block[0x40:0x44] + self.server_salt = key_block[0x44:0x48] + return + key_block = prf(self.master_secret, b'key expansion' + seed, 0x120, hf) 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:0x40] + self.encryption_key = key_block[0x40:0x60] + self.decryption_key = key_block[0x60:0x80] def save(self): with open('tls.dict', 'wb') as f: @@ -197,7 +214,6 @@ def decrypt(self, c: bytes): return m def encrypt(self, b: bytes): - # iv = unhexlify('454849acdd075174d6b9e713a957c2e7') iv = os.urandom(0x10) cipher = Cipher(algorithms.AES(self.encryption_key), modes.CBC(iv), backend=crypto_backend) encryptor = cipher.encryptor() @@ -205,6 +221,24 @@ def encrypt(self, b: bytes): c = encryptor.update(b) + encryptor.finalize() return iv + c + def gcm_seal(self, t, plaintext): + # TLS 1.2 AES-GCM record body: explicit_nonce(8) || ciphertext || tag(16). + # GCM IV = client_salt(4) || explicit_nonce(8); AAD = seq(8) || type || ver || len. + seq = self.tx_seq + self.tx_seq = seq + 1 + explicit_nonce = os.urandom(8) + iv = self.client_salt + explicit_nonce + aad = pack('>Q', seq) + pack('>BBBH', t, 3, 3, len(plaintext)) + return explicit_nonce + AESGCM(self.encryption_key).encrypt(iv, plaintext, aad) + + def gcm_open(self, t, body): + explicit_nonce, ct = body[:8], body[8:] + seq = self.rx_seq + self.rx_seq = seq + 1 + iv = self.server_salt + explicit_nonce + aad = pack('>Q', seq) + pack('>BBBH', t, 3, 3, len(ct) - 16) + return AESGCM(self.decryption_key).decrypt(iv, ct, aad) + def validate(self, t: int, b: bytes): b, hs = b[:-0x20], b[-0x20:] @@ -226,8 +260,13 @@ def sign(self, t: int, b: bytes): 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) + # Some devices hash the handshake for both Finished messages only through + # CertVerify (the client Finished is not folded in); snapshot it here so + # handle_finish can check the server's verify_data against the same hash. + fh = getattr(self, 'fin_hash', None) or self.handshake_hash + self._fin_hash = fh.copy() + pf = getattr(self, 'fin_prf', None) or getattr(self, 'prf_hash', sha256) + verify_data = prf(self.master_secret, b'client finished' + self._fin_hash.digest(), 0xc, pf) return b'\x14' + with_3bytes_size(verify_data) def make_change_cipher_spec(self): @@ -247,16 +286,20 @@ def with_neg_hdr(self, t: int, b: bytes): return b def make_client_kex(self): - b = b'\x04' + to_bytes(self.session_public.x)[::-1] + to_bytes(self.session_public.y)[::-1] + # Uncompressed point 04 || X || Y, big-endian, zero-padded to 32 bytes each. + b = b'\x04' + self.session_public.x.to_bytes(32, 'big') + self.session_public.y.to_bytes( + 32, 'big') 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()))) + h = getattr(self, 'cv_hash', self.handshake_hash) + prehash = getattr(self, 'cv_prehash', hashes.SHA256()) + b = self.priv_key.sign(h.copy().digest(), ec.ECDSA(Prehashed(prehash))) return self.with_neg_hdr(0x0f, b) def handle_server_hello(self, p: bytes): - if p[:2] != unhexlify('0303'): + # Some devices report a non-standard version (0x0383); accept 0303 or 0383. + if p[:2] not in (unhexlify('0303'), unhexlify('0383')): raise Exception('unexpected TLS version %s' % hexlify(p[:2]).decode()) p = p[2:] @@ -267,7 +310,7 @@ def handle_server_hello(self, p: bytes): (suite, ), p = unpack('>H', p[:2]), p[2:] - if suite != 0xc005: + if suite not in (0xc005, 0xc02e): raise Exception('Server accepted unsupported cipher suite %04x' % suite) if p[0] != 0: @@ -299,8 +342,9 @@ def handle_server_hello_done(self, p: bytes): hexlify(p).decode()) def handle_finish(self, b: bytes): - hs_hash = self.handshake_hash.copy().digest() - verify_data = prf(self.master_secret, b'server finished' + hs_hash, 0xc) + fh = getattr(self, '_fin_hash', None) or self.handshake_hash + pf = getattr(self, 'fin_prf', None) or getattr(self, 'prf_hash', sha256) + verify_data = prf(self.master_secret, b'server finished' + fh.copy().digest(), 0xc, pf) if verify_data != b: raise Exception('Final handshake check failed') @@ -308,11 +352,14 @@ def handle_app_data(self, b: bytes): if not self.secure_rx: raise Exception('App payload before secure connection established') + if getattr(self, 'gcm', False): + return self.gcm_open(0x17, b) return self.validate(0x17, self.decrypt(b)) def handle_handshake(self, handshake: bytes) -> None: if self.secure_rx: - handshake = self.validate(0x16, self.decrypt(handshake)) + handshake = self.gcm_open(0x16, handshake) if getattr(self, 'gcm', False) \ + else self.validate(0x16, self.decrypt(handshake)) while len(handshake) > 0: while len(handshake) < 4: @@ -371,13 +418,15 @@ def make_app_data(self, b: bytes): if not self.secure_tx: raise Exception('App payload before secure connection established') - b = self.encrypt(self.sign(0x17, b)) + b = self.gcm_seal(0x17, b) if getattr(self, 'gcm', False) else self.encrypt( + self.sign(0x17, b)) return unhexlify('170303') + with_2bytes_size(b) def make_handshake(self, b: bytes): if self.secure_tx: - b = self.encrypt(self.sign(0x16, b)) + b = self.gcm_seal(0x16, b) if getattr(self, 'gcm', False) else self.encrypt( + self.sign(0x16, b)) return unhexlify('160303') + with_2bytes_size(b) @@ -390,8 +439,10 @@ def make_client_hello(self): suits = b'' suits += pack('>H', 0xc005) # TLS_ECDH_ECDSA_WITH_AES_256_CBC_SHA + suits += pack('>H', 0xc02e) # TLS_ECDH_ECDSA_WITH_AES_256_CBC_SHA384 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', 0x008d) # TLS_PSK_WITH_AES_256_CBC_SHA + suits += pack('>H', 0x00a8) # TLS_PSK_WITH_AES_256_CBC_SHA384 h += with_2bytes_size(suits) h += with_1byte_size(b'') # no compression options diff --git a/validitysensor/tudor.py b/validitysensor/tudor.py new file mode 100644 index 0000000..31d1020 --- /dev/null +++ b/validitysensor/tudor.py @@ -0,0 +1,565 @@ +import json +import logging +import os +import time +from binascii import hexlify, unhexlify +from hashlib import sha256, sha384 +from struct import pack, pack_into + +import usb.core as ucore +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import ec +from usb.core import USBError + +from .db import Db, User, db +from .sensor import Sensor, sensor +from .tls import tls, hs_key +from .usb import usb, CancelledException + +# Kensington VeriMark IT (Synaptics "Tudor"). Unlike the Prometheus sensors this +# reader takes ownership through a 0x93 command carrying the host key and then +# speaks AES-256-GCM TLS, so it needs its own bring-up instead of the flash based +# one in init.open_common(). +VERIMARK_IT = (0x047d, 0x8054) + +# The host key and the enrolled-finger map must survive reboots (the reader is +# paired to this key and re-pairing a fresh one wears its one-time-programmable +# memory), so they live under /var/lib rather than the runtime data dir. +data_dir = '/var/lib/python-validity/' +host_key_path = data_dir + 'host_key.pem' + +# Constant part of the 0x93 request captured from the Windows driver. Only the +# host public key gets written into it (X at 0x04, Y at 0x48, little-endian). +take_ownership_body = unhexlify( + '3f5f1700f034785215a99b8288ed111ee33722491aa2d9933d155679612eece967c02e2b000000000000000000000000000000000000000000000000000000000000000000000000911027e62eee3d4c10712a1a95de59c6acfd208d6806ea97cd4015a97dbf57030000000000000000000000000000000000000000000000000000000000000000000000000000' +) + + +def _find(): + return ucore.find(idVendor=VERIMARK_IT[0], idProduct=VERIMARK_IT[1]) + + +def present(): + return _find() is not None + + +def _read_reply(dev, timeout=8000): + try: + reply = bytes(dev.read(0x81, 1 << 20, timeout=timeout)) + except USBError: + return b'' + while True: + try: + chunk = bytes(dev.read(0x81, 1 << 20, timeout=150)) + except USBError: + break + if not chunk: + break + reply += chunk + return reply + + +def _reopen(): + dev = _find() + try: + dev.reset() + except USBError: + pass + for _ in range(20): + time.sleep(0.4) + if present(): + break + time.sleep(0.8) + dev = _find() + dev.set_configuration() + return dev + + +def _host_key(): + try: + with open(host_key_path, 'rb') as f: + return serialization.load_pem_private_key(f.read(), password=None) + except (FileNotFoundError, ValueError): + key = ec.generate_private_key(ec.SECP256R1()) + with open(host_key_path, 'wb') as f: + f.write( + key.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, + serialization.NoEncryption())) + os.chmod(host_key_path, 0o600) + return key + + +def _build_take_ownership(priv): + nums = priv.public_key().public_numbers() + body = bytearray(take_ownership_body) + body[0x04:0x24] = nums.x.to_bytes(0x20, 'little') + body[0x48:0x68] = nums.y.to_bytes(0x20, 'little') + signer = ec.derive_private_key(hs_key(), ec.SECP256R1()) # hardcoded Synaptics key + sig = signer.sign(bytes(body), ec.ECDSA(hashes.SHA256())) + payload = bytearray(401) + payload[0] = 0x93 + payload[1:1 + len(body)] = body + pack_into('BH', 0, len(o93body)) * 2 + b'\x44\x87' + o93body + return tls.with_neg_hdr(0x0b, inner) + + +def _handshake(o93body): + rsp = usb.cmd(unhexlify('44000000') + tls.make_handshake(tls.make_client_hello())) + tls.parse_tls_response(rsp) + tls.make_keys() + + cert = _client_certificate(o93body) + kex = tls.make_client_kex() + verify = tls.make_cert_verify() + msg = (unhexlify('44000000') + tls.make_handshake(cert + kex + verify) + + tls.make_change_cipher_spec() + tls.make_handshake(tls.make_finish())) + tls.parse_tls_response(usb.cmd(msg)) + + if not (tls.secure_rx and tls.secure_tx): + raise Exception('TLS handshake did not establish a secure channel') + + +def open_session(): + os.makedirs(data_dir, exist_ok=True) + priv = _host_key() + dev = _reopen() + usb.dev = dev + usb.dev.default_timeout = 10000 + + o93body, device_x, device_y = _pair(dev, priv) + logging.info('VeriMark IT paired') + + _configure_tls(priv, device_x, device_y) + _handshake(o93body) + logging.info('VeriMark IT secure channel established') + _install_backend() + + +# --------------------------------------------------------------------------- +# On-chip enroll / identify / delete +# --------------------------------------------------------------------------- + +enroll_max_touches = 40 +event_timeout = 8000 + +# Capture-program / config frames (0x39), event-arm payloads (0x86) and +# capture-start bodies (0x80), captured verbatim from the Windows enroll. +capture_prog = unhexlify('00710200ffff000005050020000000000505000000000000' + 'ffff000005050020000000000505000000000000' + 'ffff000005050020000000000505000000000000') +capture_prog += b'\0' * (124 - len(capture_prog)) +cfg_c = unhexlify( + '00000000000000000000002000000000000000000000000000000000000000200000000000000000000000000000000000000020000000000000000000000000000000000000002000000000000000000000000000000000000000200000000000000000000000000000000000000020000000000000000000000000' +) +cfg_a = unhexlify( + 'e80300004b000000070100200101000000000000000000004b000000010000200000000000000000000000004b000000010100200000000000000000000000004b0000000100002000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000' +) +cfg_final = unhexlify( + 'f4010000f4010000070500200000000005050000000000000000000000000020000000000000000000000000f401000000050020000000000000000000000000000000000000002000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000' +) +arm_06 = unhexlify('060000000000000000000000000000000600000000000000000000000000000000000000') +arm_04 = unhexlify('040000000000000000000000000000000400000000000000000000000000000000000000') +arm_00 = unhexlify('000000000000000000000000000000000000000000000000000000000000000004000000') +arm_01 = unhexlify('000000010000000000000000000000000000000100000000000000000000000000000000') +cap_start = unhexlify('0c000000010000000100000801010100') +cap_start_setup = unhexlify('14000000010000000100000801010100') +cap_start_id = unhexlify('14100000010000000100000801010100') +identify_cmd = unhexlify('99010000000000000000000000') +progress_marker = unhexlify('3c000000') + +# On-chip store record from the Windows enroll. The firmware keeps it verbatim; +# only the template id at [17:33] names the finger (the SID and trailing DPAPI +# blob are opaque). We swap in the freshly-enrolled id and track which identity +# and finger it belongs to host-side in TudorDb. +enroll_record = unhexlify( + '00000000000000bb010000000010000000929ddec6a39d4a66f07d98d6c2edcbdc01004c000000030000001c0000000105000000000005150000006ff57d1e90b75cb01feceafce803000000000000000000000000000000000000000000000000000000000000000000000000000000000000020001000000f503004601000001000000d08c9ddf0115d1118c7a00c04fc297eb0100000008ca566ecdff3e4ea771db3c50daaac000000000020000000000106600000001000020000000feacf00c6ff07754dafecacd39796ff9321175e1348af6fa2483a9a6de3f692d000000000e80000000020000200000007d0eda6947522cab327848357db444b907f363450f6102df69d0926841fcdf6b70000000f1506641ffbf55598bdc839db87d9b871c81d1ac34d338dac8a4558a9d442a69661052c398e3bac277f9128fb4ee90cc6c83cd020daaa1c98112a43413bcae37fb355a45a3f63a87cd5730aa2a65bc90085b6f37d4b602583b4d2393f769627aff9af79094871dc7a99f1ca3f4e3e83340000000dde605f36aa30d3ce039fc7c8bcd11576b489debe6bc30de89870ae71b367a8ccade5074d97df7984b52ad042396a8e0a65921fa4ad0d938d49bd4de8a066df0' +) + +tudor_db_path = data_dir + 'tudor-fingers.json' + + +class TudorDb(Db): + """Host-side identity store mapping a user identity and finger subtype to the + on-chip template id. The reader only matches biometrics and hands back a + template id, so which user and finger each print belongs to is kept here. + + Instances come from the class swap in _install_backend, so the fields are set + up there rather than in __init__.""" + def _ensure_loaded(self): + if self._loaded: + return + try: + with open(tudor_db_path) as f: + self.users = json.load(f) + except (FileNotFoundError, ValueError): + self.users = [] + self._loaded = True + + def _save(self): + with open(tudor_db_path, 'w') as f: + json.dump(self.users, f) + + def _next_id(self): + used = [u['dbid'] for u in self.users] + used += [f['dbid'] for u in self.users for f in u['fingers']] + return max(used) + 1 if used else 1 + + def _as_user(self, u): + user = User(u['dbid'], u['identity']) + user.fingers = [{ + 'dbid': f['dbid'], + 'subtype': f['subtype'], + 'storage': 0, + 'valueSize': 0 + } for f in u['fingers']] + return user + + def lookup_user(self, identity): + self._ensure_loaded() + key = hexlify(identity.to_bytes()).decode() + return next((self._as_user(u) for u in self.users if u['identity'] == key), None) + + def get_user(self, dbid): + self._ensure_loaded() + return next((self._as_user(u) for u in self.users if u['dbid'] == dbid), None) + + def new_user(self, identity): + self._ensure_loaded() + dbid = self._next_id() + self.users.append({ + 'dbid': dbid, + 'identity': hexlify(identity.to_bytes()).decode(), + 'fingers': [] + }) + self._save() + return dbid + + def new_finger(self, userid, subtype, tuid): + self._ensure_loaded() + dbid = self._next_id() + user = next(u for u in self.users if u['dbid'] == userid) + user['fingers'].append({'dbid': dbid, 'subtype': subtype, 'tuid': hexlify(tuid).decode()}) + self._save() + return dbid + + def find_by_tuid(self, tuid): + self._ensure_loaded() + key = hexlify(tuid).decode() + for u in self.users: + for f in u['fingers']: + if f['tuid'] == key: + return u['dbid'], f['subtype'] + return None + + def del_record(self, dbid): + self._ensure_loaded() + user = next((u for u in self.users if u['dbid'] == dbid), None) + if user is None: + return + for f in user['fingers']: + sensor.delete_template(unhexlify(f['tuid'])) + self.users.remove(user) + self._save() + + +class TudorSensor(Sensor): + ectr = 0 + + def open(self): + usb.dev.default_timeout = event_timeout + + def cancel(self): + usb.cancel = True + + def _cmd(self, payload, check=True, label=None): + rsp = tls.app(payload) + status = int.from_bytes(rsp[:2], 'little') + if check and status != 0: + raise Exception('%s failed: status 0x%04x' % (label or '0x%02x' % payload[0], status)) + return rsp + + def _config(self, frame): + return self._cmd(b'\x39' + frame, label='config') + + def _db_list(self): + rsp = self._cmd(unhexlify('9f02000000') + b'\xff' * 16, label='db_list') + n = int.from_bytes(rsp[2:4], 'little') + return [rsp[4 + i * 16:4 + i * 16 + 16] for i in range(n)] + + def _db_children(self, tuid): + # 0x9f03 lists a record's children; the Windows driver issues it per + # template while arming, so we replay it to match the capture. + return self._cmd(b'\x9f\x03\x00\x00\x00' + tuid, check=False, label='db_children') + + def _arm(self, payload): + rsp = self._cmd(b'\x86' + payload, label='arm') + self.ectr = int.from_bytes(rsp[-2:], 'little') + return rsp + + def _read_event(self, seq=None): + if seq is None: + seq = self.ectr + rsp = self._cmd(b'\x87' + pack('= 8 else '----' + return kind, rsp + + def _drain_int(self): + while True: + try: + usb.dev.read(0x83, 8, timeout=120) + except USBError: + return + + def _wait_finger(self, timeout_s=8): + deadline = time.time() + timeout_s + while time.time() < deadline: + if usb.cancel: + raise CancelledException() + try: + event = bytes(usb.dev.read(0x83, 8, timeout=600)) + except USBError: + continue + if event and event[0] == 2: + return event + return None + + @staticmethod + def _parse_progress(rsp): + i = rsp.find(progress_marker) + if i < 0: + return None, None, None + bitmap = rsp[i + 4] if i + 4 < len(rsp) else None + coverage = rsp[i + 6] if i + 6 < len(rsp) else None + tuid = rsp[i - 16:i] if i >= 16 else b'' + return bitmap, coverage, tuid + + def _poll_event(self, seq=None, want=None, timeout_s=2.5): + deadline = time.time() + timeout_s + while time.time() < deadline: + kind, rsp = self._read_event(seq=seq) + if kind != '----' and (want is None or kind == want): + return kind, rsp + time.sleep(0.03) + return '--', b'' + + def _read_progress(self, timeout_s=2.5): + rsp = b'' + deadline = time.time() + timeout_s + while time.time() < deadline: + rsp = self._cmd(unhexlify('9602000000'), check=False, label='progress') + if rsp.find(progress_marker) >= 0: + return rsp + time.sleep(0.04) + return rsp + + def _enroll_setup(self): + self._cmd(unhexlify('820000000000000207'), label='opinfo') + self._cmd(b'\x9e\x01', label='db_info') + self._cmd(b'\x9e\x01', label='db_info') + for tuid in self._db_list(): + self._db_children(tuid) + self._config(cfg_c) + self._cmd(b'\x19', label='startinfo') + self._config(cfg_c) + self._cmd(b'\x19', label='startinfo') + # finger-less calibration cycle + self._arm(arm_06) + self._read_event() + self._arm(arm_00) + self._config(capture_prog) + self._arm(arm_01) + self._cmd(b'\x80' + cap_start_setup, label='capture_start') + self._read_event(seq=0) + self._config(cfg_c) + self._arm(arm_00) + self._cmd(b'\x81', check=False, label='trigger') + self._cmd(identify_cmd, check=False, label='identify') + self._config(cfg_a) + self._cmd(b'\x96' + unhexlify('010000000000000000000000'), label='enroll_begin') + + def _capture_enroll(self, finger_timeout=8): + self._drain_int() + self._arm(arm_06) + if self._wait_finger(finger_timeout) is None: + return None + # the sensor reports finger-down / image / capture-done asynchronously + self._poll_event(want='0100') + self._arm(arm_00) + self._arm(arm_04) + self._poll_event(want='0200') + self._arm(arm_00) + self._config(capture_prog) + self._arm(arm_01) + self._cmd(b'\x80' + cap_start, label='capture_start') + self._poll_event(seq=0, want='1800', timeout_s=3) + self._config(cfg_c) + self._arm(arm_00) + self._cmd(b'\x81', check=False, label='trigger') + return self._parse_progress(self._read_progress()) + + def _store_template(self, tuid): + self._config(cfg_final) + self._cmd(identify_cmd, check=False, label='identify') + record = bytearray(enroll_record) + record[17:33] = tuid + return self._cmd(b'\x96\x03' + bytes(record), check=False, label='store') + + def _enroll_end(self): + try: + self._cmd(unhexlify('9604000000'), check=False, label='enroll_end') + self._config(cfg_c) + except USBError: + pass + + def enroll(self, identity, subtype, update_cb): + self.open() + usb.cancel = False + self._enroll_setup() + tuid = None + misses = 0 + try: + for _ in range(enroll_max_touches): + if usb.cancel: + raise CancelledException() + result = self._capture_enroll() + if result is None: + misses += 1 + if misses >= 5: + break + update_cb(None, Exception('no finger detected')) + continue + misses = 0 + bitmap, coverage, captured = result + logging.debug('enroll progress 0x%02x coverage %s%%', bitmap or 0, coverage) + update_cb(bitmap, None) + if bitmap == 0x7f: + tuid = captured + self._store_template(tuid) + break + finally: + self._enroll_end() + + if tuid is None: + raise Exception('Enrollment did not complete') + + user = db.lookup_user(identity) + userid = user.dbid if user is not None else db.new_user(identity) + return db.new_finger(userid, subtype, tuid) + + def _identify_setup(self): + self._cmd(unhexlify('820000000000000207'), label='opinfo') + self._cmd(b'\x9e\x01', label='db_info') + for tuid in self._db_list(): + self._db_children(tuid) + self._config(cfg_c) + self._cmd(b'\x19', label='startinfo') + # warm-up capture to arm interrupt-EP finger detection (stays in verify mode) + self._arm(arm_06) + self._read_event() + self._arm(arm_00) + self._config(capture_prog) + self._arm(arm_01) + self._cmd(b'\x80' + cap_start_setup, label='capture_start') + self._read_event(seq=0) + self._config(cfg_c) + self._arm(arm_00) + self._cmd(b'\x81', check=False, label='trigger') + + def _capture_verify(self, finger_timeout=8): + self._drain_int() + self._arm(arm_06) + if self._wait_finger(finger_timeout) is None: + return None + self._poll_event(want='0200') + self._arm(arm_00) + self._config(capture_prog) + self._arm(arm_01) + self._cmd(b'\x80' + cap_start_id, label='capture_start') + self._poll_event(seq=0, want='1800', timeout_s=3) + self._config(cfg_c) + self._arm(arm_00) + self._cmd(b'\x81', check=False, label='trigger') + rsp = self._cmd(identify_cmd, check=False, label='verdict') + if rsp[:2] == b'\x00\x00' and len(rsp) >= 18 and rsp[2:18] != b'\x00' * 16: + return rsp[2:18] + return b'' + + def identify(self, update_cb): + self.open() + usb.cancel = False + self._identify_setup() + while True: + if usb.cancel: + raise CancelledException() + match = self._capture_verify() + if match is None: + update_cb(Exception('no finger detected')) + continue + break + + if not match: + raise Exception('Finger not recognized') + found = db.find_by_tuid(match) + if found is None: + raise Exception('Matched an unenrolled template %s' % hexlify(match).decode()) + userid, subtype = found + return userid, subtype, match + + def delete_template(self, tuid): + # a4 80 begin txn -> a0 02 resolve tuid to record locator (rsp[20:36]) + # -> a3 01 delete locator -> a4 81 commit txn + self._cmd(unhexlify('a480'), check=False, label='txn_begin') + info = self._cmd(unhexlify('a002000000') + tuid, check=False, label='object_info') + locator = info[20:36] + self._cmd(unhexlify('a301000000') + locator, check=False, label='delete') + self._cmd(unhexlify('a481'), check=False, label='txn_commit') + + +def _install_backend(): + # dbus-service binds the sensor/db singletons at import time, before the + # device is known, so once the VeriMark IT is up we specialise them in place + # (it is a different device family) rather than swapping the bound globals. + sensor.__class__ = TudorSensor + db.__class__ = TudorDb + db.users = [] + db._loaded = False + sensor.open() diff --git a/validitysensor/usb.py b/validitysensor/usb.py index 464b092..2a481a8 100644 --- a/validitysensor/usb.py +++ b/validitysensor/usb.py @@ -18,6 +18,7 @@ class SupportedDevices(Enum): DEV_97 = (0x138a, 0x0097) DEV_9d = (0x138a, 0x009d) DEV_9a = (0x06cb, 0x009a) + DEV_8054 = (0x047d, 0x8054) # Kensington VeriMark IT @classmethod def from_usbid(cls, vendorid, productid):