diff --git a/src/ttlock_ble/__init__.py b/src/ttlock_ble/__init__.py index 51963f0..3685e5a 100644 --- a/src/ttlock_ble/__init__.py +++ b/src/ttlock_ble/__init__.py @@ -22,11 +22,23 @@ from .cloud import TTLockCloud from .constants import AutoLockOperate, KeyboardPwdType, LockState, LogOperate, PwdOperateType from .exceptions import CloudError, TTLockError -from .models import LockAdvertisement, LockEvent, LockVersion, LogEntry, SiteInfo, VirtualKey +from .models import ( + InventoryEntry, + InventoryResult, + LockAdvertisement, + LockEvent, + LockVersion, + LogEntry, + SiteInfo, + UnverifiedCardLayoutError, + VirtualKey, +) __all__ = [ "AutoLockOperate", "CloudError", + "InventoryEntry", + "InventoryResult", "KeyboardPwdType", "LockAdvertisement", "LockEvent", @@ -39,5 +51,6 @@ "TTLockClient", "TTLockCloud", "TTLockError", + "UnverifiedCardLayoutError", "VirtualKey", ] diff --git a/src/ttlock_ble/client.py b/src/ttlock_ble/client.py index f0d19c4..e5e6ed5 100644 --- a/src/ttlock_ble/client.py +++ b/src/ttlock_ble/client.py @@ -16,7 +16,7 @@ from .constants import KeyboardPwdType, LockState from .crypto import aes_decrypt, hex_key_to_bytes from .exceptions import TTLockError -from .models import LockEvent, LogEntry +from .models import InventoryEntry, InventoryResult, LockEvent, LogEntry, UnverifiedCardLayoutError from .protocol import Frame, FrameReassembler if TYPE_CHECKING: @@ -53,6 +53,15 @@ _POST_NOTIFY_SETTLE = 0.5 _DEFAULT_KEEP_ALIVE_SECONDS = 25.0 _KEEP_ALIVE_INTERVAL = 2.0 +_MAX_INVENTORY_ROUND_TRIPS = 300 + +_PIN_COVERAGE_LIMIT = ( + "CMD 0x07 does not show unused batch/offline PINs; returned items are not a complete " + "list of active keypad credentials" +) +_CYCLIC_COVERAGE_LIMIT = ( + "Cyclic day/time restrictions use a separate command and are not visible to this query" +) class _OwnedBleakClient(BleakClientWithServiceCache): @@ -448,6 +457,181 @@ async def clear_passcodes(self) -> None: "clear_passcodes", ) + async def _inventory_admin_handshake(self) -> None: + """Authorize read-only inventory before sending any list opcode.""" + if not self.key.adminPs or not self.key.adminPs.isdecimal(): + raise TTLockError("Credential inventory requires an admin password") + try: + await self._inventory_admin_challenge() + except Exception: # noqa: BLE001 + # Builders can also raise with credential input in their message. + raise TTLockError("Credential inventory admin authorization failed") from None + + async def _inventory_admin_challenge(self) -> None: + """Perform one challenge under the inventory error-redaction boundary.""" + payload = cmd.payload_check_admin(self.key.uid, self.key.adminPs, self.key.lockFlagPos) + plain = await self._inventory_exchange(cmd.CMD_CHECK_ADMIN, payload, "inventory_admin") + echo, status, data = self._parse_response_envelope(plain, "inventory_admin") + if echo != cmd.CMD_CHECK_ADMIN or status != cmd.RESPONSE_SUCCESS or len(data) < 4: + raise TTLockError("Credential inventory admin authorization was rejected") + token = int.from_bytes(data[:4], "big") + random_payload = cmd.payload_check_random(token, self.key.unlockKey) + plain = await self._inventory_exchange( + cmd.CMD_CHECK_RANDOM, random_payload, "inventory_random" + ) + echo, status, _data = self._parse_response_envelope(plain, "inventory_random") + if echo != cmd.CMD_CHECK_RANDOM or status != cmd.RESPONSE_SUCCESS: + raise TTLockError("Credential inventory admin challenge was rejected") + + async def _inventory_exchange(self, opcode: int, payload: bytes, label: str) -> bytes: + """Suppress potentially sensitive transport/decrypt exception chains.""" + try: + frame = Frame.for_lock(self.key.lockVersion, opcode, payload).encrypt_data( + self._aes_key + ) + return self._decrypt_response(await self._exchange(frame), label) + except Exception: # noqa: BLE001 + raise TTLockError("Credential inventory exchange failed") from None + + async def get_passcodes(self) -> InventoryResult: + """Read only discoverable PIN metadata; never return a PIN value. + + CMD 0x07 omits active batch/offline PINs until first physical use. + This result therefore always carries an explicit coverage limitation. + """ + async with self._command_lock: + await self._inventory_admin_handshake() + entries: list[InventoryEntry] = [] + skipped = 0 + cursor = 0 + seen: set[int] = set() + for _ in range(_MAX_INVENTORY_ROUND_TRIPS): + if cursor in seen: + return InventoryResult( + tuple(entries), + complete=False, + limitation="PIN cursor repeated", + skipped_unknown=skipped, + ) + seen.add(cursor) + try: + plain = await self._inventory_exchange( + cmd.CMD_GET_PASSCODES, cmd.payload_passcode_list(cursor), "get_passcodes" + ) + except Exception: # noqa: BLE001 + # Preserve only credential-free partial evidence. + if entries: + return InventoryResult( + tuple(entries), + complete=False, + limitation="PIN read interrupted", + skipped_unknown=skipped, + transport_interrupted=True, + ) + raise TTLockError("PIN inventory read failed") from None + try: + item, next_cursor, unknown = cmd.parse_passcode_list_response(plain) + except (ValueError, RuntimeError) as exc: + if entries: + return InventoryResult( + tuple(entries), + complete=False, + limitation="PIN response unrecognized", + skipped_unknown=skipped, + ) + raise TTLockError( + "PIN inventory response was rejected or unrecognized" + ) from exc + if item is not None: + entries.append(item) + skipped += int(unknown) + if next_cursor == 0: + return InventoryResult( + tuple(entries), + complete=True, + limitation=_PIN_COVERAGE_LIMIT, + skipped_unknown=skipped, + ) + cursor = next_cursor + return InventoryResult( + tuple(entries), + complete=False, + limitation="PIN inventory safety cap reached", + skipped_unknown=skipped, + ) + + async def _get_indexed_inventory( + self, + *, + opcode: int, + operation: int, + label: str, + parser: Callable[[bytes], InventoryEntry | None], + ) -> InventoryResult: + async with self._command_lock: + await self._inventory_admin_handshake() + entries: list[InventoryEntry] = [] + for index in range(_MAX_INVENTORY_ROUND_TRIPS): + payload = bytes([operation]) + index.to_bytes(2, "big") + try: + plain = await self._inventory_exchange(opcode, payload, label) + except Exception: # noqa: BLE001 + # Never publish an incomplete list as complete. + if entries: + return InventoryResult( + tuple(entries), + complete=False, + limitation="inventory read interrupted", + transport_interrupted=True, + ) + raise TTLockError("Credential inventory read failed") from None + try: + item = parser(plain) + except UnverifiedCardLayoutError: + return InventoryResult( + tuple(entries), + complete=False, + limitation="eight-byte IC-card layout unverified", + ) + except (ValueError, RuntimeError) as exc: + if entries: + return InventoryResult( + tuple(entries), + complete=False, + limitation="inventory response unrecognized", + ) + raise TTLockError(f"{label} response was rejected or unrecognized") from exc + if item is None: + return InventoryResult( + tuple(entries), complete=True, limitation=_CYCLIC_COVERAGE_LIMIT + ) + if int.from_bytes(plain[4:6], "big") != index + 1: + return InventoryResult( + tuple(entries), complete=False, limitation="inventory position repeated" + ) + entries.append(item) + return InventoryResult( + tuple(entries), complete=False, limitation="inventory safety cap reached" + ) + + async def get_fingerprints(self) -> InventoryResult: + """Read indexed fingerprint validity without returning biometric IDs or slots.""" + return await self._get_indexed_inventory( + opcode=cmd.CMD_MANAGE_FINGERPRINT, + operation=6, + label="get_fingerprints", + parser=cmd.parse_fingerprint_list_response, + ) + + async def get_cards(self) -> InventoryResult: + """Read four-byte IC-card records; explicitly reject the unverified layout.""" + return await self._get_indexed_inventory( + opcode=cmd.CMD_MANAGE_IC_CARD, + operation=1, + label="get_cards", + parser=cmd.parse_card_list_response, + ) + async def get_operation_log( self, *, diff --git a/src/ttlock_ble/commands/__init__.py b/src/ttlock_ble/commands/__init__.py index e5e38ba..eb04b81 100644 --- a/src/ttlock_ble/commands/__init__.py +++ b/src/ttlock_ble/commands/__init__.py @@ -11,6 +11,14 @@ from __future__ import annotations +from .access_inventory import ( + parse_card_list_response, + parse_fingerprint_list_response, + parse_passcode_list_response, + payload_card_list, + payload_fingerprint_list, + payload_passcode_list, +) from .auto_lock import parse_auto_lock_response, payload_auto_lock_search, payload_auto_lock_set from .clock import parse_get_lock_time_response, payload_get_lock_time, payload_time_calibrate from .control import ( @@ -34,8 +42,11 @@ CMD_GET_AES_KEY, CMD_GET_LOCK_TIME, CMD_GET_OPERATE_LOG, + CMD_GET_PASSCODES, CMD_INIT_PASSWORDS, CMD_LOCK, + CMD_MANAGE_FINGERPRINT, + CMD_MANAGE_IC_CARD, CMD_MANAGE_KEYBOARD_PASSWORD, CMD_QUERY_STATE, CMD_RESPONSE, @@ -67,8 +78,11 @@ "CMD_GET_AES_KEY", "CMD_GET_LOCK_TIME", "CMD_GET_OPERATE_LOG", + "CMD_GET_PASSCODES", "CMD_INIT_PASSWORDS", "CMD_LOCK", + "CMD_MANAGE_FINGERPRINT", + "CMD_MANAGE_IC_CARD", "CMD_MANAGE_KEYBOARD_PASSWORD", "CMD_QUERY_STATE", "CMD_RESPONSE", @@ -82,23 +96,29 @@ "UNLOCKED", "VENDOR", "parse_auto_lock_response", + "parse_card_list_response", "parse_check_user_time_response", + "parse_fingerprint_list_response", "parse_get_lock_time_response", "parse_lock_status", "parse_operate_log_response", + "parse_passcode_list_response", "parse_response_status", "parse_state_battery", "payload_auto_lock_search", "payload_auto_lock_set", + "payload_card_list", "payload_check_admin", "payload_check_random", "payload_check_user_time", + "payload_fingerprint_list", "payload_get_aes_key", "payload_get_lock_time", "payload_operate_log_request", "payload_passcode_add", "payload_passcode_clear", "payload_passcode_delete", + "payload_passcode_list", "payload_query_state", "payload_time_calibrate", "payload_unlock", diff --git a/src/ttlock_ble/commands/access_inventory.py b/src/ttlock_ble/commands/access_inventory.py new file mode 100644 index 0000000..5b173c7 --- /dev/null +++ b/src/ttlock_ble/commands/access_inventory.py @@ -0,0 +1,97 @@ +"""Read-only credential inventory layouts, adapted from upstream 0.1.11. + +Provenance: roquerodrigo/ttlock-ble passcode/fingerprint inventory and +PR #96's hardware-tested four-byte IC-card layout. No plaintext fields are +returned or included in diagnostics. The eight-byte card layout is rejected. +""" + +from __future__ import annotations + +from ..models.access_inventory import InventoryEntry, UnverifiedCardLayoutError +from .envelope import RESPONSE_SUCCESS, parse_response_status + +_END_SENTINEL = bytes([99, 1, 1, 0, 0]) +_PIN_TRAILER_LENGTHS = {1: 5, 3: 10, 4: 7} + + +def payload_passcode_list(sequence: int) -> bytes: + """Request one CMD 0x07 page by UInt16 big-endian sequence cursor.""" + return sequence.to_bytes(2, "big") + + +def payload_fingerprint_list(index: int) -> bytes: + """Request indexed CMD 0x06/0x06 fingerprint inventory.""" + return bytes([6]) + index.to_bytes(2, "big") + + +def payload_card_list(index: int) -> bytes: + """Request indexed CMD 0x05/0x01 IC-card inventory.""" + return bytes([1]) + index.to_bytes(2, "big") + + +def _data(plaintext: bytes, opcode: int) -> bytes: + echo, status, data = parse_response_status(plaintext) + if echo != opcode or status != RESPONSE_SUCCESS: + raise ValueError(f"inventory response rejected (opcode={opcode:#x}, status={status})") + return data + + +def parse_passcode_list_response(plaintext: bytes) -> tuple[InventoryEntry | None, int, bool]: + """Return credential-free item, next cursor and unknown-layout flag.""" + if plaintext == b"\x00\x00": # captured bare end-of-list marker + return None, 0, False + data = _data(plaintext, 7) + if len(data) == 0: # captured bare [echo][success] terminator + return None, 0, False + if len(data) < 4: + raise ValueError("passcode inventory has a truncated cursor") + cursor = int.from_bytes(data[2:4], "big") + item = data[4:] + if not item: + return None, cursor, False + if len(item) < 4: + raise ValueError("passcode inventory has a truncated item") + pwd_type = item[1] + trailer_length = _PIN_TRAILER_LENGTHS.get(pwd_type) + if trailer_length is None: + return None, cursor, True + new_length = item[2] + old_length_offset = 3 + new_length + if old_length_offset >= len(item): + raise ValueError("passcode inventory has a truncated value") + old_length = item[old_length_offset] + trailer_offset = old_length_offset + 1 + old_length + if len(item) - trailer_offset < trailer_length: + raise ValueError("passcode inventory has a truncated trailer") + return InventoryEntry({1: "permanent", 3: "time_window", 4: "cyclic"}[pwd_type]), cursor, False + + +def _indexed_entry(plaintext: bytes, *, opcode: int, operation: int) -> bytes | None: + data = _data(plaintext, opcode) + if len(data) == 4 and data[1:] == bytes([operation, 0xFF, 0xFF]): + return None + if len(data) < 4 or data[1] != operation: + raise ValueError("indexed inventory has an unexpected operation or length") + return data + + +def parse_fingerprint_list_response(plaintext: bytes) -> InventoryEntry | None: + """Discard fingerprint ID and slot from the confirmed 20-byte layout.""" + data = _indexed_entry(plaintext, opcode=6, operation=6) + if data is None: + return None + if len(data) != 20: + raise ValueError("fingerprint inventory has an unrecognized layout") + return InventoryEntry("permanent" if data[15:20] == _END_SENTINEL else "time_window") + + +def parse_card_list_response(plaintext: bytes) -> InventoryEntry | None: + """Accept only PR #96's hardware-validated four-byte card layout.""" + data = _indexed_entry(plaintext, opcode=5, operation=1) + if data is None: + return None + if len(data) == 22: + raise UnverifiedCardLayoutError("eight-byte IC-card inventory layout is unverified") + if len(data) != 18: + raise ValueError("IC-card inventory has an unrecognized layout") + return InventoryEntry("permanent" if data[13:18] == _END_SENTINEL else "time_window") diff --git a/src/ttlock_ble/commands/opcodes.py b/src/ttlock_ble/commands/opcodes.py index 8bddbae..d5b1a63 100644 --- a/src/ttlock_ble/commands/opcodes.py +++ b/src/ttlock_ble/commands/opcodes.py @@ -21,6 +21,9 @@ CMD_RESPONSE = 0x54 CMD_AUTO_LOCK_MANAGE = 0x36 CMD_MANAGE_KEYBOARD_PASSWORD = 0x03 +CMD_MANAGE_IC_CARD = 0x05 +CMD_MANAGE_FINGERPRINT = 0x06 +CMD_GET_PASSCODES = 0x07 CMD_GET_OPERATE_LOG = 0x25 APICMD_UNLOCK_BY_USER = 4 diff --git a/src/ttlock_ble/commands/passcode.py b/src/ttlock_ble/commands/passcode.py index 10a1b0a..33e6271 100644 --- a/src/ttlock_ble/commands/passcode.py +++ b/src/ttlock_ble/commands/passcode.py @@ -8,7 +8,7 @@ def _check_passcode(code: str) -> None: if not (4 <= len(code) <= 9) or not code.isdigit(): - raise ValueError(f"keyboard passcode must be 4-9 digits, got {code!r}") + raise ValueError("keyboard passcode must be 4-9 digits") def payload_passcode_add( diff --git a/src/ttlock_ble/models/__init__.py b/src/ttlock_ble/models/__init__.py index 21d0e68..d43fe45 100644 --- a/src/ttlock_ble/models/__init__.py +++ b/src/ttlock_ble/models/__init__.py @@ -2,6 +2,7 @@ from __future__ import annotations +from .access_inventory import InventoryEntry, InventoryResult, UnverifiedCardLayoutError from .cloud_credentials import CloudCredentials from .lock_advertisement import LockAdvertisement from .lock_event import LockEvent @@ -12,10 +13,13 @@ __all__ = [ "CloudCredentials", + "InventoryEntry", + "InventoryResult", "LockAdvertisement", "LockEvent", "LockVersion", "LogEntry", "SiteInfo", + "UnverifiedCardLayoutError", "VirtualKey", ] diff --git a/src/ttlock_ble/models/access_inventory.py b/src/ttlock_ble/models/access_inventory.py new file mode 100644 index 0000000..381f81e --- /dev/null +++ b/src/ttlock_ble/models/access_inventory.py @@ -0,0 +1,27 @@ +"""Credential-free summaries of read-only, admin-gated BLE inventory queries.""" + +from __future__ import annotations + +from dataclasses import dataclass + + +@dataclass(frozen=True, slots=True) +class InventoryEntry: + """One observed item without PIN, card UID, fingerprint ID or slot.""" + + validity: str + + +@dataclass(frozen=True, slots=True) +class InventoryResult: + """Enumeration result; `complete` means the cursor ended, not full coverage.""" + + entries: tuple[InventoryEntry, ...] + complete: bool + limitation: str + skipped_unknown: int = 0 + transport_interrupted: bool = False + + +class UnverifiedCardLayoutError(ValueError): + """An 8-byte card response was seen, but that layout lacks live validation.""" diff --git a/tests/test_access_inventory.py b/tests/test_access_inventory.py new file mode 100644 index 0000000..c842e04 --- /dev/null +++ b/tests/test_access_inventory.py @@ -0,0 +1,225 @@ +"""Read-only inventory, including capture-shaped layouts and secret handling.""" + +from __future__ import annotations + +import asyncio +import traceback +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from tests.conftest import make_virtual_key +from tests.test_client_ble import FakeBleakClient, _resp_frame +from ttlock_ble import TTLockClient +from ttlock_ble import commands as cmd +from ttlock_ble.exceptions import TTLockError + +pytest_plugins = ("tests.test_client_ble",) + + +def _response(opcode: int, data: bytes = b"") -> bytes: + return bytes([opcode, cmd.RESPONSE_SUCCESS]) + data + + +def _pin(code: bytes, *, cursor: int, validity: int = 1) -> bytes: + trailer = bytes([0, 1, 1, 0, 0]) if validity == 1 else bytes(10) + item = bytes([0, validity, len(code)]) + code + bytes([len(code)]) + code + trailer + return _response(7, bytes([0, 0]) + cursor.to_bytes(2, "big") + item) + + +def _fingerprint(identifier: bytes, *, end: bytes) -> bytes: + return _response(6, bytes([100, 6, 0, 1]) + identifier + bytes([0, 3]) + bytes(5) + end) + + +def _card(identifier: bytes, *, end: bytes) -> bytes: + return _response(5, bytes([85, 1, 0, 1]) + identifier + bytes(5) + end) + + +def test_pin_parser_discards_values_and_tracks_cursor() -> None: + code = b"8675309" + item, cursor, unknown = cmd.parse_passcode_list_response(_pin(code, cursor=4)) + assert item is not None and item.validity == "permanent" + assert cursor == 4 and not unknown + assert code.decode() not in repr(item) + assert cmd.parse_passcode_list_response(b"\x00\x00") == (None, 0, False) + + +def test_unknown_pin_type_is_skipped_without_dropping_cursor() -> None: + item, cursor, unknown = cmd.parse_passcode_list_response(_pin(b"1456", cursor=2, validity=2)) + assert item is None and cursor == 2 and unknown + + +def test_confirmed_indexed_layouts_discard_identifiers() -> None: + fp = bytes.fromhex("cafebabe") + uid = bytes.fromhex("deadbeef") + assert cmd.parse_fingerprint_list_response( + _fingerprint(fp, end=bytes([99, 1, 1, 0, 0])) + ) == cmd.parse_card_list_response(_card(uid, end=bytes([99, 1, 1, 0, 0]))) + assert fp.hex() not in repr(cmd.parse_fingerprint_list_response(_fingerprint(fp, end=bytes(5)))) + assert uid.hex() not in repr(cmd.parse_card_list_response(_card(uid, end=bytes(5)))) + assert cmd.parse_fingerprint_list_response(_response(6, bytes([100, 6, 255, 255]))) is None + assert cmd.parse_card_list_response(_response(5, bytes([85, 1, 255, 255]))) is None + + +def test_unverified_card_and_error_are_credential_free() -> None: + uid = bytes.fromhex("deadbeefcafebabe") + with pytest.raises(ValueError) as long_error: + cmd.parse_card_list_response(_card(uid, end=bytes(5))) + with pytest.raises(ValueError) as failed_error: + cmd.parse_card_list_response(bytes([5, 0]) + uid) + assert uid.hex() not in str(long_error.value) + str(failed_error.value) + + +@pytest.mark.parametrize("method", ["get_passcodes", "get_fingerprints", "get_cards"]) +async def test_admin_password_missing_prevents_any_ble_write(method: str) -> None: + key = make_virtual_key() + key.adminPs = "" + client = TTLockClient(key, device=MagicMock()) + client._exchange = AsyncMock() # type: ignore[method-assign] + with pytest.raises(TTLockError, match="admin password"): + await getattr(client, method)() + client._exchange.assert_not_awaited() + + +async def test_page_navigation_partial_coverage_and_no_secret() -> None: + key = make_virtual_key() + client = TTLockClient(key, device=MagicMock()) + client._inventory_admin_handshake = AsyncMock() # type: ignore[method-assign] + replies = [_pin(b"8675309", cursor=4), _pin(b"1456", cursor=0)] + client._inventory_exchange = AsyncMock(side_effect=replies) # type: ignore[method-assign] + result = await client.get_passcodes() + assert result.complete and len(result.entries) == 2 + assert "not a complete" in result.limitation + assert b"\x00\x04" in client._inventory_exchange.await_args_list[1].args[1] + assert "8675309" not in repr(result) + + +async def test_repeated_cursor_and_midscan_rejection_are_partial() -> None: + client = TTLockClient(make_virtual_key(), device=MagicMock()) + client._inventory_admin_handshake = AsyncMock() # type: ignore[method-assign] + client._inventory_exchange = AsyncMock(return_value=_pin(b"1456", cursor=2)) # type: ignore[method-assign] + result = await client.get_passcodes() + assert not result.complete and len(result.entries) == 2 + assert "cursor repeated" in result.limitation + client._inventory_exchange = AsyncMock(side_effect=[_pin(b"1456", cursor=1), bytes([7, 0])]) # type: ignore[method-assign] + result = await client.get_passcodes() + assert not result.complete and len(result.entries) == 1 + + +async def test_indexed_card_unverified_layout_returns_partial() -> None: + client = TTLockClient(make_virtual_key(), device=MagicMock()) + client._inventory_admin_handshake = AsyncMock() # type: ignore[method-assign] + client._inventory_exchange = AsyncMock( + side_effect=[_card(bytes.fromhex("deadbeef"), end=bytes(5)), _card(bytes(8), end=bytes(5))] + ) # type: ignore[method-assign] + result = await client.get_cards() + assert len(result.entries) == 1 and not result.complete + assert "unverified" in result.limitation + assert "deadbeef" not in repr(result) + + +async def test_inventory_cancellation_propagates_without_partial_result() -> None: + client = TTLockClient(make_virtual_key(), device=MagicMock()) + client._inventory_admin_handshake = AsyncMock() # type: ignore[method-assign] + client._inventory_exchange = AsyncMock(side_effect=asyncio.CancelledError) # type: ignore[method-assign] + with pytest.raises(asyncio.CancelledError): + await client.get_fingerprints() + + +async def test_transport_error_is_sanitized_and_later_failure_is_partial() -> None: + client = TTLockClient(make_virtual_key(), device=MagicMock()) + client._inventory_admin_handshake = AsyncMock() # type: ignore[method-assign] + client._inventory_exchange = AsyncMock(side_effect=OSError("private wire detail")) # type: ignore[method-assign] + with pytest.raises(TTLockError, match="inventory read failed") as error: + await client.get_cards() + assert "private wire detail" not in str(error.value) + client._inventory_exchange = AsyncMock( # type: ignore[method-assign] + side_effect=[_card(bytes.fromhex("deadbeef"), end=bytes(5)), OSError("private wire detail")] + ) + result = await client.get_cards() + assert not result.complete and len(result.entries) == 1 + assert "private wire detail" not in repr(result) + assert result.transport_interrupted + + +@pytest.mark.parametrize("phase", ["admin", "random", "page"]) +async def test_admin_transport_failure_traceback_is_credential_free(phase: str) -> None: + key = make_virtual_key() + client = TTLockClient(key, device=MagicMock()) + replies = [] + if phase != "admin": + replies.append( + _resp_frame(key, cmd.CMD_CHECK_ADMIN, _response(cmd.CMD_CHECK_ADMIN, bytes(4))) + ) + if phase == "page": + replies.append(_resp_frame(key, cmd.CMD_CHECK_RANDOM, _response(cmd.CMD_CHECK_RANDOM))) + replies.append(OSError("synthetic-secret-frame")) + client._exchange = AsyncMock(side_effect=replies) # type: ignore[method-assign] + with pytest.raises(TTLockError) as error: + await client.get_cards() + assert "synthetic-secret-frame" not in "".join(traceback.format_exception(error.value)) + + +async def test_pin_transport_interruption_keeps_only_partial_metadata() -> None: + client = TTLockClient(make_virtual_key(), device=MagicMock()) + client._inventory_admin_handshake = AsyncMock() # type: ignore[method-assign] + client._inventory_exchange = AsyncMock( # type: ignore[method-assign] + side_effect=[_pin(b"8675309", cursor=1), OSError("synthetic-secret-frame")] + ) + result = await client.get_passcodes() + assert result.transport_interrupted + assert not result.complete and len(result.entries) == 1 + assert "8675309" not in repr(result) + assert "synthetic-secret-frame" not in repr(result) + assert not client._command_lock.locked() + + +async def test_malformed_admin_challenge_key_does_not_escape_in_traceback() -> None: + key = make_virtual_key() + key.unlockKey = "synthetic-private-key" + client = TTLockClient(key, device=MagicMock()) + client._exchange = AsyncMock( # type: ignore[method-assign] + return_value=_resp_frame(key, cmd.CMD_CHECK_ADMIN, _response(cmd.CMD_CHECK_ADMIN, bytes(4))) + ) + with pytest.raises(TTLockError) as error: + await client.get_cards() + assert "synthetic-private-key" not in "".join(traceback.format_exception(error.value)) + + +async def test_real_encrypted_ble_handshake_and_indexed_read(patched_connect) -> None: + key = make_virtual_key() + client = TTLockClient(key, device=MagicMock(), keep_alive_after_command=0) + fake = FakeBleakClient(key) + patched_connect(fake) + await client.connect() + fake.reply_for_next = [ + _resp_frame(key, cmd.CMD_CHECK_ADMIN, _response(cmd.CMD_CHECK_ADMIN, bytes([1, 2, 3, 4]))), + _resp_frame(key, cmd.CMD_CHECK_RANDOM, _response(cmd.CMD_CHECK_RANDOM)), + _resp_frame( + key, + cmd.CMD_MANAGE_IC_CARD, + _card(bytes.fromhex("deadbeef"), end=bytes([99, 1, 1, 0, 0])), + ), + _resp_frame(key, cmd.CMD_MANAGE_IC_CARD, _response(5, bytes([85, 1, 255, 255]))), + ] + result = await client.get_cards() + assert result.complete and len(result.entries) == 1 + assert result.entries[0].validity == "permanent" + assert sum(chunk.startswith(b"\x7fZ") for chunk in fake.written) == 4 + assert "deadbeef" not in repr(result) + await client.disconnect() + + +async def test_admin_rejection_prevents_inventory_read(patched_connect) -> None: + key = make_virtual_key() + client = TTLockClient(key, device=MagicMock(), keep_alive_after_command=0) + fake = FakeBleakClient(key) + patched_connect(fake) + await client.connect() + fake.reply_for_next = [ + _resp_frame(key, cmd.CMD_CHECK_ADMIN, bytes([cmd.CMD_CHECK_ADMIN, cmd.RESPONSE_FAILED])), + ] + with pytest.raises(TTLockError, match="admin authorization"): + await client.get_fingerprints() + assert sum(chunk.startswith(b"\x7fZ") for chunk in fake.written) == 1 + await client.disconnect() diff --git a/tests/test_commands.py b/tests/test_commands.py index 326d67e..40ed4c7 100644 --- a/tests/test_commands.py +++ b/tests/test_commands.py @@ -63,6 +63,11 @@ def test_passcode_validation_rejects_short(self) -> None: with pytest.raises(ValueError, match="4-9 digits"): cmd.payload_passcode_add(int(KeyboardPwdType.PERMANENT), "12") + def test_passcode_validation_does_not_echo_credential(self) -> None: + with pytest.raises(ValueError) as error: + cmd.payload_passcode_delete(int(KeyboardPwdType.PERMANENT), "12xSECRET") + assert "12xSECRET" not in str(error.value) + def test_passcode_delete_and_clear(self) -> None: assert cmd.payload_passcode_delete(int(KeyboardPwdType.PERMANENT), "1234") assert len(cmd.payload_passcode_clear()) == 1