typechecking fixes

This commit is contained in:
Philippe Teuwen
2023-10-24 23:44:56 +02:00
parent d77e166249
commit ece5e79060
5 changed files with 132 additions and 108 deletions
+3 -2
View File
@@ -10,7 +10,7 @@ import pathlib
import prompt_toolkit
from prompt_toolkit.formatted_text import ANSI
from prompt_toolkit.history import FileHistory
from chameleon_utils import CR, CG, CB, CC, CY, CM, C0
from chameleon_utils import CR, CG, CY, C0
ULTRA = r"""
╦ ╦╦ ╔╦╗╦═╗╔═╗
@@ -43,7 +43,8 @@ class ChameleonCLI:
def __init__(self):
self.completer = chameleon_utils.CustomNestedCompleter.from_clitree(chameleon_cli_unit.root)
self.session = prompt_toolkit.PromptSession(completer=self.completer,
history=FileHistory(pathlib.Path.home() / ".chameleon_history"))
history=FileHistory(str(pathlib.Path.home() /
".chameleon_history")))
# new a device communication instance(only communication)
self.device_com = chameleon_com.ChameleonCom()
+18 -18
View File
@@ -10,6 +10,7 @@ from datetime import datetime
import serial.tools.list_ports
import threading
import struct
from typing import Union
from pathlib import Path
from platform import uname
@@ -17,7 +18,7 @@ import chameleon_com
import chameleon_cmd
from chameleon_utils import ArgumentParserNoExit, ArgsParserError, UnexpectedResponseError
from chameleon_utils import CLITree
from chameleon_utils import CR, CG, CB, CC, CY, CM, C0
from chameleon_utils import CR, CG, CB, CC, CY, C0
from chameleon_enum import Command, Status, SlotNumber, TagSenseType, TagSpecificType
from chameleon_enum import MifareClassicWriteMode, MifareClassicPrngType, MifareClassicDarksideStatus, MfcKeyType
from chameleon_enum import AnimationMode, ButtonType, ButtonPressFunction
@@ -52,11 +53,12 @@ def check_tools():
class BaseCLIUnit:
def __init__(self):
# new a device command transfer and receiver instance(Send cmd and receive response)
self._device_com: chameleon_com.ChameleonCom | None = None
self._device_cmd: chameleon_cmd.ChameleonCMD = chameleon_cmd.ChameleonCMD(self._device_com)
self._device_com: Union[chameleon_com.ChameleonCom, None] = None
self._device_cmd: Union[chameleon_cmd.ChameleonCMD, None] = None
@property
def device_com(self) -> chameleon_com.ChameleonCom:
assert self._device_com is not None
return self._device_com
@device_com.setter
@@ -66,6 +68,7 @@ class BaseCLIUnit:
@property
def cmd(self) -> chameleon_cmd.ChameleonCMD:
assert self._device_cmd is not None
return self._device_cmd
def args_parser(self) -> ArgumentParserNoExit:
@@ -112,6 +115,7 @@ class BaseCLIUnit:
def thread_read_output(self):
while self._process.poll() is None:
assert self._process.stdout is not None
data = self._process.stdout.read(1024)
if len(data) > 0:
self.output += data.decode(encoding="utf-8")
@@ -635,7 +639,7 @@ class HF14AInfo(ReaderRequiredUnit):
def on_exec(self, args: argparse.Namespace):
scan = HF14AScan()
scan.device_com = self.device_com
scan.scan(deep=1)
scan.scan(deep=True)
@hf_mf.command('nested')
@@ -665,7 +669,7 @@ class HFMFNested(ReaderRequiredUnit):
if nt_level == 2:
return 'HardNested'
def recover_a_key(self, block_known, type_known, key_known, block_target, type_target) -> str or None:
def recover_a_key(self, block_known, type_known, key_known, block_target, type_target) -> Union[str, None]:
"""
recover a key from key known.
@@ -740,12 +744,11 @@ class HFMFNested(ReaderRequiredUnit):
block_known = args.blk
# default to A
type_known = MfcKeyType.B if args.b else MfcKeyType.A
key_known: str = args.key
if not re.match(r"^[a-fA-F0-9]{12}$", key_known):
print("key must include 12 HEX symbols")
return
key_known: bytearray = bytearray.fromhex(key_known)
key_known_bytes = bytes.fromhex(key_known)
block_target = args.tblk
# default to A
type_target = MfcKeyType.B if args.b else MfcKeyType.A
@@ -753,7 +756,7 @@ class HFMFNested(ReaderRequiredUnit):
print(f"{CR}Target key already known{C0}")
return
print(f" - {C0}Nested recover one key running...{C0}")
key = self.recover_a_key(block_known, type_known, key_known, block_target, type_target)
key = self.recover_a_key(block_known, type_known, key_known_bytes, block_target, type_target)
if key is None:
print(f"{CY}No key found, you can retry.{C0}")
else:
@@ -862,8 +865,8 @@ class HFMFWRBL(MF1AuthArgsUnit):
param = self.get_param(args)
if not re.match(r"^[a-fA-F0-9]{32}$", args.data):
raise ArgsParserError("Data must include 32 HEX symbols")
param.data = bytearray.fromhex(args.data)
resp = self.cmd.mf1_write_one_block(param.block, param.type, param.key, param.data)
data = bytearray.fromhex(args.data)
resp = self.cmd.mf1_write_one_block(param.block, param.type, param.key, data)
if resp:
print(f" - {CG}Write done.{C0}")
else:
@@ -1411,7 +1414,7 @@ class HWSlotList(DeviceRequiredUnit):
def get_slot_name(self, slot, sense):
try:
name = self.cmd.get_slot_tag_nick(slot, sense).decode(encoding="utf8")
name = self.cmd.get_slot_tag_nick(slot, sense)
return {'baselen': len(name), 'metalen': len(CC+C0), 'name': f'{CC}{name}{C0}'}
except UnexpectedResponseError:
return {'baselen': 0, 'metalen': 0, 'name': ''}
@@ -1666,10 +1669,7 @@ class HWSlotNick(SlotIndexArgsUnit, SenseTypeArgsUnit):
sense_type = TagSenseType.HF
if args.name is not None:
name: str = args.name
encoded_name = name.encode(encoding="utf8")
if len(encoded_name) > 32:
raise ValueError("Your tag nick name too long.")
self.cmd.set_slot_tag_nick(slot_num, sense_type, encoded_name)
self.cmd.set_slot_tag_nick(slot_num, sense_type, name)
print(f' - Set tag nick name for slot {slot_num} {sense_type.name}: {name}')
elif args.delete:
self.cmd.delete_slot_tag_nick(slot_num, sense_type)
@@ -1677,7 +1677,7 @@ class HWSlotNick(SlotIndexArgsUnit, SenseTypeArgsUnit):
else:
res = self.cmd.get_slot_tag_nick(slot_num, sense_type)
print(f' - Get tag nick name for slot {slot_num} {sense_type.name}'
f': {res.decode(encoding="utf8")}')
f': {res}')
@hw_slot.command('store')
@@ -1919,9 +1919,9 @@ class HWSettingsBLEKey(DeviceRequiredUnit):
return parser
def on_exec(self, args: argparse.Namespace):
resp = self.cmd.get_ble_pairing_key()
key = self.cmd.get_ble_pairing_key()
print(" - The current key of the device(ascii): "
f"{CG}{resp.decode(encoding='ascii')}{C0}")
f"{CG}{key}{C0}")
if args.key is not None:
if len(args.key) != 6:
+74 -56
View File
@@ -1,5 +1,6 @@
import struct
import ctypes
from typing import Union
import chameleon_com
from chameleon_utils import expect_response
@@ -28,7 +29,7 @@ class ChameleonCMD:
"""
resp = self.device.send_cmd_sync(Command.GET_APP_VERSION)
if resp.status == Status.SUCCESS:
resp.data = struct.unpack('!BB', resp.data)
resp.parsed = struct.unpack('!BB', resp.data)
# older protocol, must upgrade!
if resp.status == 0 and resp.data == b'\x00\x01':
print("Chameleon does not understand new protocol. Please update firmware")
@@ -43,7 +44,7 @@ class ChameleonCMD:
"""
resp = self.device.send_cmd_sync(Command.GET_DEVICE_CHIP_ID)
if resp.status == Status.SUCCESS:
resp.data = resp.data.hex()
resp.parsed = resp.data.hex()
return resp
@expect_response(Status.SUCCESS)
@@ -53,21 +54,21 @@ class ChameleonCMD:
"""
resp = self.device.send_cmd_sync(Command.GET_DEVICE_ADDRESS)
if resp.status == Status.SUCCESS:
resp.data = resp.data.hex()
resp.parsed = resp.data.hex()
return resp
@expect_response(Status.SUCCESS)
def get_git_version(self) -> str:
def get_git_version(self):
resp = self.device.send_cmd_sync(Command.GET_GIT_VERSION)
if resp.status == Status.SUCCESS:
resp.data = resp.data.decode('utf-8')
resp.parsed = resp.data.decode('utf-8')
return resp
@expect_response(Status.SUCCESS)
def get_device_mode(self):
resp = self.device.send_cmd_sync(Command.GET_DEVICE_MODE)
if resp.status == Status.SUCCESS:
resp.data, = struct.unpack('!?', resp.data)
resp.parsed, = struct.unpack('!?', resp.data)
return resp
def is_device_reader_mode(self) -> bool:
@@ -113,7 +114,7 @@ class ChameleonCMD:
ats, = struct.unpack_from(f'!{atslen}s', resp.data, offset)
offset += struct.calcsize(f'!{atslen}s')
data.append({'uid': uid, 'atqa': atqa, 'sak': sak, 'ats': ats})
resp.data = data
resp.parsed = data
return resp
def mf1_detect_support(self):
@@ -134,7 +135,7 @@ class ChameleonCMD:
"""
resp = self.device.send_cmd_sync(Command.MF1_DETECT_PRNG)
if resp.status == Status.HF_TAG_OK:
resp.data = resp.data[0]
resp.parsed = resp.data[0]
return resp
@expect_response(Status.HF_TAG_OK)
@@ -148,7 +149,7 @@ class ChameleonCMD:
resp = self.device.send_cmd_sync(Command.MF1_DETECT_NT_DIST, data)
if resp.status == Status.HF_TAG_OK:
uid, dist = struct.unpack('!II', resp.data)
resp.data = {'uid': uid, 'dist': dist}
resp.parsed = {'uid': uid, 'dist': dist}
return resp
@expect_response(Status.HF_TAG_OK)
@@ -160,12 +161,12 @@ class ChameleonCMD:
data = struct.pack('!BB6sBB', type_known, block_known, key_known, type_target, block_target)
resp = self.device.send_cmd_sync(Command.MF1_NESTED_ACQUIRE, data)
if resp.status == Status.HF_TAG_OK:
resp.data = [{'nt': nt, 'nt_enc': nt_enc, 'par': par}
for nt, nt_enc, par in struct.iter_unpack('!IIB', resp.data)]
resp.parsed = [{'nt': nt, 'nt_enc': nt_enc, 'par': par}
for nt, nt_enc, par in struct.iter_unpack('!IIB', resp.data)]
return resp
@expect_response(Status.HF_TAG_OK)
def mf1_darkside_acquire(self, block_target, type_target, first_recover: int or bool, sync_max):
def mf1_darkside_acquire(self, block_target, type_target, first_recover: Union[int, bool], sync_max):
"""
Collect the key parameters needed for Darkside decryption.
@@ -180,9 +181,9 @@ class ChameleonCMD:
if resp.status == Status.HF_TAG_OK:
if resp.data[0] == MifareClassicDarksideStatus.OK:
darkside_status, uid, nt1, par, ks1, nr, ar = struct.unpack('!BIIQQII', resp.data)
resp.data = (darkside_status, {'uid': uid, 'nt1': nt1, 'par': par, 'ks1': ks1, 'nr': nr, 'ar': ar})
resp.parsed = (darkside_status, {'uid': uid, 'nt1': nt1, 'par': par, 'ks1': ks1, 'nr': nr, 'ar': ar})
else:
resp.data = (resp.data[0],)
resp.parsed = (resp.data[0],)
return resp
@expect_response([Status.HF_TAG_OK, Status.MF_ERR_AUTH])
@@ -197,7 +198,7 @@ class ChameleonCMD:
"""
data = struct.pack('!BB6s', type_value, block, key)
resp = self.device.send_cmd_sync(Command.MF1_AUTH_ONE_KEY_BLOCK, data)
resp.data = resp.status == Status.HF_TAG_OK
resp.parsed = resp.status == Status.HF_TAG_OK
return resp
@expect_response(Status.HF_TAG_OK)
@@ -211,7 +212,9 @@ class ChameleonCMD:
:return:
"""
data = struct.pack('!BB6s', type_value, block, key)
return self.device.send_cmd_sync(Command.MF1_READ_ONE_BLOCK, data)
resp = self.device.send_cmd_sync(Command.MF1_READ_ONE_BLOCK, data)
resp.parsed = resp.data
return resp
@expect_response(Status.HF_TAG_OK)
def mf1_write_one_block(self, block, type_value, key, block_data):
@@ -226,7 +229,7 @@ class ChameleonCMD:
"""
data = struct.pack('!BB6s16s', type_value, block, key, block_data)
resp = self.device.send_cmd_sync(Command.MF1_WRITE_ONE_BLOCK, data)
resp.data = resp.status == Status.HF_TAG_OK
resp.parsed = resp.status == Status.HF_TAG_OK
return resp
@expect_response(Status.HF_TAG_OK)
@@ -270,7 +273,9 @@ class ChameleonCMD:
f'must be between {((len(data) - 1) * 8 )+1} and {len(data) * 8} included')
data = bytes(cs)+struct.pack(f'!HH{len(data)}s', resp_timeout_ms, bitlen, bytearray(data))
return self.device.send_cmd_sync(Command.HF14A_RAW, data, timeout=(resp_timeout_ms / 1000) + 1)
resp = self.device.send_cmd_sync(Command.HF14A_RAW, data, timeout=(resp_timeout_ms // 1000) + 1)
resp.parsed = resp.data
return resp
@expect_response(Status.HF_TAG_OK)
def mf1_static_nested_acquire(self, block_known, type_known, key_known, block_target, type_target):
@@ -281,7 +286,7 @@ class ChameleonCMD:
data = struct.pack('!BB6sBB', type_known, block_known, key_known, type_target, block_target)
resp = self.device.send_cmd_sync(Command.MF1_STATIC_NESTED_ACQUIRE, data)
if resp.status == Status.HF_TAG_OK:
resp.data = {
resp.parsed = {
'uid': struct.unpack('!I', resp.data[0:4])[0],
'nts': [
{
@@ -299,7 +304,9 @@ class ChameleonCMD:
:return:
"""
return self.device.send_cmd_sync(Command.EM410X_SCAN)
resp = self.device.send_cmd_sync(Command.EM410X_SCAN)
resp.parsed = resp.data
return resp
@expect_response(Status.LF_TAG_OK)
def em410x_write_to_t55xx(self, id_bytes: bytes):
@@ -325,8 +332,8 @@ class ChameleonCMD:
"""
resp = self.device.send_cmd_sync(Command.GET_SLOT_INFO)
if resp.status == Status.SUCCESS:
resp.data = [{'hf': hf, 'lf': lf}
for hf, lf in struct.iter_unpack('!HH', resp.data)]
resp.parsed = [{'hf': hf, 'lf': lf}
for hf, lf in struct.iter_unpack('!HH', resp.data)]
return resp
@expect_response(Status.SUCCESS)
@@ -338,7 +345,7 @@ class ChameleonCMD:
"""
resp = self.device.send_cmd_sync(Command.GET_ACTIVE_SLOT)
if resp.status == Status.SUCCESS:
resp.data = resp.data[0]
resp.parsed = resp.data[0]
return resp
@expect_response(Status.SUCCESS)
@@ -425,7 +432,9 @@ class ChameleonCMD:
"""
Get the simulated EM410x card id
"""
return self.device.send_cmd_sync(Command.EM410X_GET_EMU_ID)
resp = self.device.send_cmd_sync(Command.EM410X_GET_EMU_ID)
resp.parsed = resp.data
return resp
@expect_response(Status.SUCCESS)
def mf1_set_detection_enable(self, enabled: bool):
@@ -447,7 +456,7 @@ class ChameleonCMD:
"""
resp = self.device.send_cmd_sync(Command.MF1_GET_DETECTION_COUNT)
if resp.status == Status.SUCCESS:
resp.data, = struct.unpack('!I', resp.data)
resp.parsed, = struct.unpack('!I', resp.data)
return resp
@expect_response(Status.SUCCESS)
@@ -476,7 +485,7 @@ class ChameleonCMD:
'ar': ar.hex()
})
pos += struct.calcsize('!BB4s4s4s4s')
resp.data = result_list
resp.parsed = result_list
return resp
@expect_response(Status.SUCCESS)
@@ -498,7 +507,9 @@ class ChameleonCMD:
Gets data for selected block range
"""
data = struct.pack('!BB', block_start, block_count)
return self.device.send_cmd_sync(Command.MF1_READ_EMU_BLOCK_DATA, data)
resp = self.device.send_cmd_sync(Command.MF1_READ_EMU_BLOCK_DATA, data)
resp.parsed = resp.data
return resp
@expect_response(Status.SUCCESS)
def hf14a_set_anti_coll_data(self, uid: bytes, atqa: bytes, sak: bytes, ats: bytes = b''):
@@ -515,7 +526,7 @@ class ChameleonCMD:
return self.device.send_cmd_sync(Command.HF14A_SET_ANTI_COLL_DATA, data)
@expect_response(Status.SUCCESS)
def set_slot_tag_nick(self, slot: SlotNumber, sense_type: TagSenseType, name: bytes):
def set_slot_tag_nick(self, slot: SlotNumber, sense_type: TagSenseType, name: str):
"""
Set the nick name of the slot.
@@ -524,8 +535,11 @@ class ChameleonCMD:
:param name: Card slot nickname
:return:
"""
encoded_name = name.encode(encoding="utf8")
if len(encoded_name) > 32:
raise ValueError("Your tag nick name too long.")
# SlotNumber() will raise error for us if slot not in slot range
data = struct.pack(f'!BB{len(name)}s', SlotNumber.to_fw(slot), sense_type, name)
data = struct.pack(f'!BB{len(encoded_name)}s', SlotNumber.to_fw(slot), sense_type, encoded_name)
return self.device.send_cmd_sync(Command.SET_SLOT_TAG_NICK, data)
@expect_response(Status.SUCCESS)
@@ -539,7 +553,9 @@ class ChameleonCMD:
"""
# SlotNumber() will raise error for us if slot not in slot range
data = struct.pack('!BB', SlotNumber.to_fw(slot), sense_type)
return self.device.send_cmd_sync(Command.GET_SLOT_TAG_NICK, data)
resp = self.device.send_cmd_sync(Command.GET_SLOT_TAG_NICK, data)
resp.parsed = resp.data.decode(encoding="utf8")
return resp
@expect_response(Status.SUCCESS)
def delete_slot_tag_nick(self, slot: SlotNumber, sense_type: TagSenseType):
@@ -569,11 +585,11 @@ class ChameleonCMD:
resp = self.device.send_cmd_sync(Command.MF1_GET_EMULATOR_CONFIG)
if resp.status == Status.SUCCESS:
b1, b2, b3, b4, b5 = struct.unpack('!????B', resp.data)
resp.data = {'detection': b1,
'gen1a_mode': b2,
'gen2_mode': b3,
'block_anti_coll_mode': b4,
'write_mode': b5}
resp.parsed = {'detection': b1,
'gen1a_mode': b2,
'gen2_mode': b3,
'block_anti_coll_mode': b4,
'write_mode': b5}
return resp
@expect_response(Status.SUCCESS)
@@ -630,7 +646,7 @@ class ChameleonCMD:
"""
resp = self.device.send_cmd_sync(Command.GET_ANIMATION_MODE)
if resp.status == Status.SUCCESS:
resp.data = resp.data[0]
resp.parsed = resp.data[0]
return resp
@expect_response(Status.SUCCESS)
@@ -640,7 +656,7 @@ class ChameleonCMD:
"""
resp = self.device.send_cmd_sync(Command.GET_ENABLED_SLOTS)
if resp.status == Status.SUCCESS:
resp.data = [{'hf': hf, 'lf': lf} for hf, lf in struct.iter_unpack('!BB', resp.data)]
resp.parsed = [{'hf': hf, 'lf': lf} for hf, lf in struct.iter_unpack('!BB', resp.data)]
return resp
@expect_response(Status.SUCCESS)
@@ -657,7 +673,7 @@ class ChameleonCMD:
Reset settings stored in flash memory
"""
resp = self.device.send_cmd_sync(Command.RESET_SETTINGS)
resp.data = resp.status == Status.SUCCESS
resp.parsed = resp.status == Status.SUCCESS
return resp
@expect_response(Status.SUCCESS)
@@ -666,7 +682,7 @@ class ChameleonCMD:
Store settings to flash memory
"""
resp = self.device.send_cmd_sync(Command.SAVE_SETTINGS)
resp.data = resp.status == Status.SUCCESS
resp.parsed = resp.status == Status.SUCCESS
return resp
@expect_response(Status.SUCCESS)
@@ -675,7 +691,7 @@ class ChameleonCMD:
Reset to factory settings
"""
resp = self.device.send_cmd_sync(Command.WIPE_FDS)
resp.data = resp.status == Status.SUCCESS
resp.parsed = resp.status == Status.SUCCESS
self.device.close()
return resp
@@ -686,7 +702,7 @@ class ChameleonCMD:
"""
resp = self.device.send_cmd_sync(Command.GET_BATTERY_INFO)
if resp.status == Status.SUCCESS:
resp.data = struct.unpack('!HB', resp.data)
resp.parsed = struct.unpack('!HB', resp.data)
return resp
@expect_response(Status.SUCCESS)
@@ -697,7 +713,7 @@ class ChameleonCMD:
data = struct.pack('!B', button)
resp = self.device.send_cmd_sync(Command.GET_BUTTON_PRESS_CONFIG, data)
if resp.status == Status.SUCCESS:
resp.data = resp.data[0]
resp.parsed = resp.data[0]
return resp
@expect_response(Status.SUCCESS)
@@ -716,7 +732,7 @@ class ChameleonCMD:
data = struct.pack('!B', button)
resp = self.device.send_cmd_sync(Command.GET_LONG_BUTTON_PRESS_CONFIG, data)
if resp.status == Status.SUCCESS:
resp.data = resp.data[0]
resp.parsed = resp.data[0]
return resp
@expect_response(Status.SUCCESS)
@@ -746,7 +762,9 @@ class ChameleonCMD:
"""
Get config of ble connect key
"""
return self.device.send_cmd_sync(Command.GET_BLE_PAIRING_KEY)
resp = self.device.send_cmd_sync(Command.GET_BLE_PAIRING_KEY)
resp.parsed = resp.data.decode(encoding='ascii')
return resp
@expect_response(Status.SUCCESS)
def delete_all_ble_bonds(self):
@@ -768,7 +786,7 @@ class ChameleonCMD:
status=Status.NOT_IMPLEMENTED)
else:
if resp.status == Status.SUCCESS:
resp.data = [x[0] for x in struct.iter_unpack('!H', resp.data)]
resp.parsed = [x[0] for x in struct.iter_unpack('!H', resp.data)]
return resp
@expect_response(Status.SUCCESS)
@@ -781,7 +799,7 @@ class ChameleonCMD:
resp = self.device.send_cmd_sync(Command.GET_DEVICE_MODEL)
if resp.status == Status.SUCCESS:
resp.data = resp.data[0]
resp.parsed = resp.data[0]
return resp
@expect_response(Status.SUCCESS)
@@ -809,14 +827,14 @@ class ChameleonCMD:
settings_version, animation_mode, btn_press_A, btn_press_B, btn_long_press_A, \
btn_long_press_B, ble_pairing_enable, ble_pairing_key = \
struct.unpack('!BBBBBBB6s', resp.data)
resp.data = {'settings_version': settings_version,
'animation_mode': animation_mode,
'btn_press_A': btn_press_A,
'btn_press_B': btn_press_B,
'btn_long_press_A': btn_long_press_A,
'btn_long_press_B': btn_long_press_B,
'ble_pairing_enable': ble_pairing_enable,
'ble_pairing_key': ble_pairing_key}
resp.parsed = {'settings_version': settings_version,
'animation_mode': animation_mode,
'btn_press_A': btn_press_A,
'btn_press_B': btn_press_B,
'btn_long_press_A': btn_long_press_A,
'btn_long_press_B': btn_long_press_B,
'ble_pairing_enable': ble_pairing_enable,
'ble_pairing_key': ble_pairing_key}
return resp
@expect_response(Status.SUCCESS)
@@ -836,7 +854,7 @@ class ChameleonCMD:
offset += struct.calcsize(f'!{uidlen}s2s1sB')
ats, = struct.unpack_from(f'!{atslen}s', resp.data, offset)
offset += struct.calcsize(f'!{atslen}s')
resp.data = {'uid': uid, 'atqa': atqa, 'sak': sak, 'ats': ats}
resp.parsed = {'uid': uid, 'atqa': atqa, 'sak': sak, 'ats': ats}
return resp
@expect_response(Status.SUCCESS)
@@ -848,7 +866,7 @@ class ChameleonCMD:
"""
resp = self.device.send_cmd_sync(Command.GET_BLE_PAIRING_ENABLE)
if resp.status == Status.SUCCESS:
resp.data, = struct.unpack('!?', resp.data)
resp.parsed, = struct.unpack('!?', resp.data)
return resp
@expect_response(Status.SUCCESS)
+24 -21
View File
@@ -3,7 +3,8 @@ import struct
import threading
import time
import serial
from chameleon_utils import CR, CG, CB, CC, CY, CM, C0
from typing import Union
from chameleon_utils import CR, CG, CC, CY, C0
from chameleon_enum import Command, Status
# each thread is waiting for its data for 100 ms before looping again
@@ -36,10 +37,11 @@ class Response:
Chameleon Response Data
"""
def __init__(self, cmd, status, data=b''):
def __init__(self, cmd, status, data=b'', parsed=None):
self.cmd = cmd
self.status = status
self.data: bytearray = data
self.data: bytes = data
self.parsed = parsed
class ChameleonCom:
@@ -55,20 +57,20 @@ class ChameleonCom:
"""
Create a chameleon device instance
"""
self.serial_instance: serial.Serial | None = None
self.serial_instance: Union[serial.Serial, None] = None
self.send_data_queue = queue.Queue()
self.wait_response_map = {}
self.event_closing = threading.Event()
def isOpen(self):
def isOpen(self) -> bool:
"""
Chameleon is connected and init.
:return:
"""
return self.serial_instance is not None and self.serial_instance.isOpen()
return self.serial_instance is not None and self.serial_instance.is_open
def open(self, port):
def open(self, port) -> "ChameleonCom":
"""
Open chameleon port to communication
And init some variables
@@ -86,8 +88,9 @@ class ChameleonCom:
finally:
if error is not None:
raise OpenFailException(error)
assert self.serial_instance is not None
try:
self.serial_instance.dtr = 1 # must make dtr enable
self.serial_instance.dtr = True # must make dtr enable
except Exception:
# not all serial support dtr, e.g. virtual serial over BLE
pass
@@ -102,7 +105,7 @@ class ChameleonCom:
threading.Thread(target=self.thread_check_timeout).start()
return self
def check_open(self):
def check_open(self) -> None:
"""
:return:
@@ -111,7 +114,7 @@ class ChameleonCom:
raise NotOpenException("Please call open() function to start device.")
@staticmethod
def lrc_calc(array):
def lrc_calc(array: Union[bytearray, bytes]) -> int:
"""
Calc lrc and auto cut byte.
@@ -133,6 +136,7 @@ class ChameleonCom:
"""
self.event_closing.set()
try:
assert self.serial_instance is not None
self.serial_instance.close()
except Exception:
pass
@@ -156,6 +160,7 @@ class ChameleonCom:
while self.isOpen():
# receive
try:
assert self.serial_instance is not None
data_bytes = self.serial_instance.read()
except Exception as e:
if not self.event_closing.is_set():
@@ -197,8 +202,8 @@ class ChameleonCom:
# ok, lrc for data is correct.
# and we are receive completed
# print(f"Buffer data = {data_buffer.hex()}")
data_response = data_buffer[struct.calcsize('!BBHHHB'):
struct.calcsize(f'!BBHHHB{data_length}s')]
data_response = bytes(data_buffer[struct.calcsize('!BBHHHB'):
struct.calcsize(f'!BBHHHB{data_length}s')])
if DEBUG:
try:
command = Command(data_cmd)
@@ -263,6 +268,7 @@ class ChameleonCom:
self.wait_response_map[task_cmd]['end_time'] = start_time + task_timeout
self.wait_response_map[task_cmd]['is_timeout'] = False
try:
assert self.serial_instance is not None
# send to device
self.serial_instance.write(task['frame'])
except Exception as e:
@@ -292,7 +298,7 @@ class ChameleonCom:
self.wait_response_map[task_cmd]['is_timeout'] = True
time.sleep(THREAD_BLOCKING_TIMEOUT)
def make_data_frame_bytes(self, cmd: int, data: bytearray = None, status: int = 0) -> bytearray:
def make_data_frame_bytes(self, cmd: int, data: Union[bytes, None] = None, status: int = 0) -> bytes:
"""
Make data frame
@@ -308,9 +314,9 @@ class ChameleonCom:
frame[struct.calcsize('!BBHHH')] = self.lrc_calc(frame[:struct.calcsize('!BBHHH')])
# lrc3
frame[struct.calcsize(f'!BBHHHB{len(data)}s')] = self.lrc_calc(frame[:struct.calcsize(f'!BBHHHB{len(data)}s')])
return frame
return bytes(frame)
def send_cmd_auto(self, cmd: int, data: bytearray = None, status: int = 0, callback=None, timeout: int = 3,
def send_cmd_auto(self, cmd: int, data: Union[bytes, None] = None, status: int = 0, callback=None, timeout: int = 3,
close: bool = False):
"""
Send cmd to device
@@ -342,9 +348,8 @@ class ChameleonCom:
if callable(callback):
task['callback'] = callback
self.send_data_queue.put(task)
return self
def send_cmd_sync(self, cmd: int, data: bytearray or bytes or list or int = None, status: int = 0,
def send_cmd_sync(self, cmd: int, data: Union[bytes, None] = None, status: int = 0,
timeout: int = 3) -> Response:
"""
Send cmd to device, and block receive data.
@@ -355,13 +360,11 @@ class ChameleonCom:
:param timeout: wait response timeout
:return: response data
"""
if isinstance(data, int):
data = [data] # warp array.
if len(self.commands):
# check if chameleon can understand this command
if cmd not in self.commands:
raise CMDInvalidException(f"This device doesn't declare that it can support this command: {cmd}.\nMake "
f"sure firmware is up to date and matches client")
raise CMDInvalidException(f"This device doesn't declare that it can support this command: {cmd}.\n"
f"Make sure firmware is up to date and matches client")
# first to send cmd, no callback mode(sync)
self.send_cmd_auto(cmd, data, status, None, timeout)
# wait cmd start process
+13 -11
View File
@@ -1,7 +1,8 @@
import argparse
import colorama
from functools import wraps
from typing import Union
# once Python3.10 is mainstream, we can replace Union[str, None] by str | None
from typing import Union, Callable, Any
from prompt_toolkit.completion import Completer, NestedCompleter, WordCompleter
from prompt_toolkit.completion.base import Completion
from prompt_toolkit.document import Document
@@ -38,12 +39,12 @@ class ArgumentParserNoExit(argparse.ArgumentParser):
we must raise exception to stop parse
"""
def __init__(self, **args):
super().__init__(*args)
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.add_help = False
self.description = "Please enter correct parameters"
def exit(self, status: int = ..., message: str or None = ...):
def exit(self, status: int = 0, message: Union[str, None] = None):
if message:
raise ParserExitIntercept(message)
@@ -97,7 +98,7 @@ class ArgumentParserNoExit(argparse.ArgumentParser):
print('\n'.join(lines))
def expect_response(accepted_responses: Union[int, list[int]]):
def expect_response(accepted_responses: Union[int, list[int]]) -> Callable[..., Any]:
"""
Decorator for wrapping a Chameleon CMD function to check its response
for expected return codes and throwing an exception otherwise
@@ -116,7 +117,7 @@ def expect_response(accepted_responses: Union[int, list[int]]):
status_string = f"Unexpected response and unknown status {ret.status}"
raise UnexpectedResponseError(status_string)
return ret.data
return ret.parsed
return error_throwing_func
@@ -133,11 +134,12 @@ class CLITree:
:param cls: A BaseCLIUnit instance handling the command
"""
def __init__(self, name=None, help_text=None, fullname=None, children=None, cls=None, root=False) -> None:
self.name: str = name
self.help_text: str = help_text
self.fullname: str = fullname if fullname else name
self.children: list[CLITree] = children if children else list()
def __init__(self, name: str = "", help_text: Union[str, None] = None, fullname: Union[str, None] = None,
children: Union[list["CLITree"], None] = None, cls=None, root=False) -> None:
self.name = name
self.help_text = help_text
self.fullname = fullname if fullname else name
self.children = children if children else list()
self.cls = cls
self.root = root
if self.help_text is None and not root: