| commit | author | age | ||
| f83421 | 1 | import logging |
| JK | 2 | import math |
| 3 | import struct | |
| 4 | import usb.util | |
| 5 | ||
| 6 | def to_hex(val): | |
| 7 | return ' '.join([bytes([i]).hex() for i in val]) | |
| 8 | ||
| 9 | class ControlVaultCommunicator: | |
| 10 | def __init__(self, device, spi_master=0x01, spi_slave=0x00, spi_crc=0x00): | |
| 11 | self.logger = logging.getLogger(__name__) | |
| 12 | self.device = device | |
| 13 | self.bulk_in, self.bulk_out = self._find_endpoints() | |
| 14 | ||
| 15 | self.spi_master = spi_master | |
| 16 | self.spi_slave = spi_slave | |
| 17 | self.spi_crc = spi_crc | |
| 18 | self.spi_slave_prefix = struct.pack('>BB', self.spi_slave, self.spi_crc) | |
| 19 | ||
| 20 | def ctrl_transfer(self, *args, **kwargs): | |
| 21 | self.logger.debug('Control: {} {}'.format(args, kwargs)) | |
| 22 | return self.device.ctrl_transfer(*args, **kwargs) | |
| 23 | ||
| 24 | def write(self, *args, **kwargs): | |
| 25 | return self.bulk_out.write(*args, **kwargs) | |
| 26 | ||
| 27 | def read(self, *args, **kwargs): | |
| 28 | return self.bulk_in.read(*args, **kwargs) | |
| 29 | ||
| 30 | def send_packet(self, payload): | |
| 31 | length = len(payload) | |
| 32 | packet = struct.pack('>BBH', self.spi_master, self.spi_crc, length) + payload | |
| 33 | self.logger.debug('Put: {}'.format(to_hex(packet))) | |
| 34 | self.write(packet) | |
| 35 | ||
| 36 | def recv_packet(self): | |
| 37 | packet = self.read(64, timeout=5000).tobytes() | |
| 38 | tag = packet[0:2] | |
| 39 | if tag != self.spi_slave_prefix: | |
| 40 | raise Exception('Unknown tag: {}'.format(tag.hex())) | |
| 41 | length = packet[2:4] | |
| 42 | length = struct.unpack('>H', length)[0] | |
| 43 | ||
| 44 | for i in range(0, math.ceil(length/64) - 1): | |
| 45 | packet += self.read(64).tobytes() | |
| 46 | ||
| 47 | self.logger.debug('Got: {}'.format(to_hex(packet))) | |
| 48 | return packet[4:] | |
| 49 | ||
| 50 | def talk(self, exchange): | |
| 51 | for packet in exchange: | |
| 52 | self.send_packet(bytes.fromhex(packet)) | |
| 53 | data = self.recv_packet() | |
| 54 | ||
| 55 | if data[1] == 0x61: | |
| 56 | packet = self.recv_packet() | |
| 57 | ||
| 58 | def _find_endpoints(self): | |
| 59 | self.logger.debug('Enumerating interfaces...') | |
| 60 | configuration = self.device.get_active_configuration() | |
| 61 | bcm_interface = None | |
| 09f3f6 | 62 | has_contacted = False |
| JK | 63 | has_contactless = False |
| f83421 | 64 | for interface in configuration: |
| JK | 65 | if interface.bInterfaceClass == 0xff: |
| 66 | if bcm_interface is not None: | |
| 67 | raise Exception('More than one vendor-specific interface found!') | |
| 68 | bcm_interface = interface | |
| 09f3f6 | 69 | if interface.bInterfaceClass == 0xb: |
| JK | 70 | if interface.iInterface == 0x5: |
| 71 | has_contacted = True | |
| 72 | if interface.iInterface == 0x6: | |
| 73 | has_contactless = True | |
| 74 | if not has_contactless: | |
| 75 | raise Exception('No contactless reader on this device!') | |
| f83421 | 76 | if bcm_interface is None: |
| JK | 77 | raise Exception('Cannot find vendor-specific interface') |
| 78 | self.logger.debug('Interface found: {}'.format(bcm_interface._str())) | |
| 79 | ||
| 80 | self.logger.debug('Enumerating endpoints...') | |
| 81 | bulk_in = None | |
| 82 | bulk_out = None | |
| 83 | for endpoint in bcm_interface: | |
| 84 | if endpoint.bmAttributes & usb.util._ENDPOINT_TRANSFER_TYPE_MASK == usb.util.ENDPOINT_TYPE_BULK: | |
| 85 | if endpoint.bEndpointAddress & usb.util._ENDPOINT_DIR_MASK == usb.util.ENDPOINT_IN: | |
| 86 | if bulk_in is not None: | |
| 87 | raise Exception('More than one BULK IN endpoint found!') | |
| 88 | bulk_in = endpoint | |
| 89 | self.logger.debug('BULK IN found: {}'.format(bulk_in._str())) | |
| 90 | if endpoint.bEndpointAddress & usb.util._ENDPOINT_DIR_MASK == usb.util.ENDPOINT_OUT: | |
| 91 | if bulk_out is not None: | |
| 92 | raise Exception('More than one BULK OUT endpoint found!') | |
| 93 | bulk_out = endpoint | |
| 94 | self.logger.debug('BULK OUT found: {}'.format(bulk_out._str())) | |
| 95 | ||
| 96 | if bulk_in is None: | |
| 97 | raise Exception('BULK IN endpoint not found!') | |
| 98 | if bulk_out is None: | |
| 99 | raise Exception('BULK OUT endpoint not found!') | |
| 100 | ||
| 101 | self.logger.debug('Endpoint discovery successful.') | |
| 102 | return bulk_in, bulk_out | |
