From 165f609756a9fc363eb68d555459299cc0065e95 Mon Sep 17 00:00:00 2001 From: Robert Schulze Date: Tue, 15 Feb 2022 09:07:08 +0100 Subject: [PATCH 1/3] Support remove_pairing if accessory is not reachable This is the same handling as supported by the Apple Home App. --- homekit/controller/controller.py | 91 ++++++++++++++++++-------------- homekit/exceptions.py | 5 ++ 2 files changed, 55 insertions(+), 41 deletions(-) diff --git a/homekit/controller/controller.py b/homekit/controller/controller.py index a4c4328..d837265 100644 --- a/homekit/controller/controller.py +++ b/homekit/controller/controller.py @@ -651,47 +651,56 @@ def remove_pairing(self, alias, pairingId=None): tlv8.Entry(TlvTypes.Identifier, pairingIdToDelete.encode()) ]) - if connection_type == 'IP': - if not IP_TRANSPORT_SUPPORTED: - raise TransportNotSupportedError('IP') - session = IpSession(pairing_data) - response = session.post('/pairings', request_tlv, content_type='application/pairing+tlv8') - session.close() - data = response.read() - data = tlv8.decode(data, { - TlvTypes.State: tlv8.DataType.INTEGER, - TlvTypes.Error: tlv8.DataType.INTEGER - - }) - elif connection_type == 'BLE': - if not BLE_TRANSPORT_SUPPORTED: - raise TransportNotSupportedError('BLE') - inner = tlv8.encode([ - tlv8.Entry(AdditionalParameterTypes.ParamReturnResponse, bytearray(b'\x01')), - tlv8.Entry(AdditionalParameterTypes.Value, request_tlv) - ]) - - body = len(inner).to_bytes(length=2, byteorder='little') + inner - - from .ble_impl.device import DeviceManager - manager = DeviceManager(self.ble_adapter) - device = manager.make_device(pairing_data['AccessoryMAC']) - device.connect() - - logging.debug('resolved %d services', len(device.services)) - pair_remove_char, pair_remove_char_id = find_characteristic_by_uuid(device, ServicesTypes.PAIRING_SERVICE, - CharacteristicsTypes.PAIRING_PAIRINGS) - logging.debug('setup char: %s %s', pair_remove_char, pair_remove_char.service.device) - - session = BleSession(pairing_data, self.ble_adapter) - response = session.request(pair_remove_char, pair_remove_char_id, HapBleOpCodes.CHAR_WRITE, body) - data = tlv8.decode(response.first_by_id(AdditionalParameterTypes.Value).data, { - TlvTypes.State: tlv8.DataType.INTEGER, - TlvTypes.Error: tlv8.DataType.INTEGER - - }) - else: - raise Exception('not implemented (neither IP nor BLE)') + try: + if connection_type == 'IP': + if not IP_TRANSPORT_SUPPORTED: + raise TransportNotSupportedError('IP') + session = IpSession(pairing_data) + response = session.post('/pairings', request_tlv, content_type='application/pairing+tlv8') + session.close() + data = response.read() + data = tlv8.decode(data, { + TlvTypes.State: tlv8.DataType.INTEGER, + TlvTypes.Error: tlv8.DataType.INTEGER + + }) + elif connection_type == 'BLE': + if not BLE_TRANSPORT_SUPPORTED: + raise TransportNotSupportedError('BLE') + inner = tlv8.encode([ + tlv8.Entry(AdditionalParameterTypes.ParamReturnResponse, bytearray(b'\x01')), + tlv8.Entry(AdditionalParameterTypes.Value, request_tlv) + ]) + + body = len(inner).to_bytes(length=2, byteorder='little') + inner + + from .ble_impl.device import DeviceManager + manager = DeviceManager(self.ble_adapter) + device = manager.make_device(pairing_data['AccessoryMAC']) + device.connect() + + logging.debug('resolved %d services', len(device.services)) + pair_remove_char, pair_remove_char_id = find_characteristic_by_uuid(device, + ServicesTypes.PAIRING_SERVICE, + CharacteristicsTypes.PAIRING_PAIRINGS) + logging.debug('setup char: %s %s', pair_remove_char, pair_remove_char.service.device) + + session = BleSession(pairing_data, self.ble_adapter) + response = session.request(pair_remove_char, pair_remove_char_id, HapBleOpCodes.CHAR_WRITE, body) + data = tlv8.decode(response.first_by_id(AdditionalParameterTypes.Value).data, { + TlvTypes.State: tlv8.DataType.INTEGER, + TlvTypes.Error: tlv8.DataType.INTEGER + }) + else: + raise NotImplementedError('not implemented (neither IP nor BLE)') + except NotImplementedError: + raise + except Exception: + logging.debug('error sending unpair request, remove without accessory notification') + data = tlv8.EntryList([ + tlv8.Entry(TlvTypes.State, States.M2), + tlv8.Entry(TlvTypes.Error, 0) + ]) # act upon the response (the same is returned for IP and BLE accessories) # handle the result, spec says, if it has only one entry with state == M2 we unpaired, else its an error. diff --git a/homekit/exceptions.py b/homekit/exceptions.py index 89b230a..4e28c67 100644 --- a/homekit/exceptions.py +++ b/homekit/exceptions.py @@ -266,3 +266,8 @@ def __init__(self, transport): class DisconnectedControllerError(HomeKitException): def __init__(self): Exception.__init__(self, 'Controller has passed away') + + +class NotImplementedError(HomeKitException): + def __init__(self, message): + Exception.__init__(self, message) From 0aad6533cb66a333a6b6ad027d10a0a89cdd641e Mon Sep 17 00:00:00 2001 From: Robert Schulze Date: Tue, 15 Feb 2022 10:28:16 +0100 Subject: [PATCH 2/3] Fix flake induced workflow fails --- homekit/controller/controller.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/homekit/controller/controller.py b/homekit/controller/controller.py index d837265..66c23cc 100644 --- a/homekit/controller/controller.py +++ b/homekit/controller/controller.py @@ -680,9 +680,10 @@ def remove_pairing(self, alias, pairingId=None): device.connect() logging.debug('resolved %d services', len(device.services)) - pair_remove_char, pair_remove_char_id = find_characteristic_by_uuid(device, - ServicesTypes.PAIRING_SERVICE, - CharacteristicsTypes.PAIRING_PAIRINGS) + pair_remove_char, pair_remove_char_id = \ + find_characteristic_by_uuid(device, + ServicesTypes.PAIRING_SERVICE, + CharacteristicsTypes.PAIRING_PAIRINGS) logging.debug('setup char: %s %s', pair_remove_char, pair_remove_char.service.device) session = BleSession(pairing_data, self.ble_adapter) From e5105455b8b5431daac01563d1eb4e6531eb7f88 Mon Sep 17 00:00:00 2001 From: Robert Schulze Date: Thu, 24 Feb 2022 08:15:27 +0100 Subject: [PATCH 3/3] Make removing without reachable accessory a "default off"-option --- homekit/controller/controller.py | 6 +++++- homekit/remove_pairing.py | 4 +++- 2 files changed, 8 insertions(+), 2 deletions(-) diff --git a/homekit/controller/controller.py b/homekit/controller/controller.py index 66c23cc..4a69df9 100644 --- a/homekit/controller/controller.py +++ b/homekit/controller/controller.py @@ -619,7 +619,7 @@ def finish_pairing(pin): return finish_pairing - def remove_pairing(self, alias, pairingId=None): + def remove_pairing(self, alias, pairingId=None, force=False): """ Remove a pairing between the controller and the accessory. The pairing data is delete on both ends, on the accessory and the controller. @@ -632,6 +632,7 @@ def remove_pairing(self, alias, pairingId=None): :param alias: the controller's alias for the accessory :param pairingId: the pairing id to be removed + :param force: remove the pairing even if the accessory cannot be reached :raises AuthenticationError: if the controller isn't authenticated to the accessory. :raises AccessoryNotFoundError: if the device can not be found via zeroconf :raises UnknownError: on unknown errors @@ -697,6 +698,9 @@ def remove_pairing(self, alias, pairingId=None): except NotImplementedError: raise except Exception: + if not force: + raise + logging.debug('error sending unpair request, remove without accessory notification') data = tlv8.EntryList([ tlv8.Entry(TlvTypes.State, States.M2), diff --git a/homekit/remove_pairing.py b/homekit/remove_pairing.py index d661aaf..c2a96fc 100755 --- a/homekit/remove_pairing.py +++ b/homekit/remove_pairing.py @@ -31,6 +31,8 @@ def setup_args_parser(): help='this pairing ID identifies the controller who should be removed from accessory') parser.add_argument('--adapter', action='store', dest='adapter', default='hci0', help='the bluetooth adapter to be used (defaults to hci0)') + parser.add_argument('--force', action='store_true', dest='force', + help='force removing of pairing in pairing data even if accessory is not reachable') add_log_arguments(parser) return parser.parse_args() @@ -46,6 +48,6 @@ def setup_args_parser(): print('"{a}" is no known alias'.format(a=args.alias)) exit(-1) - controller.remove_pairing(args.alias, args.controllerPairingId) + controller.remove_pairing(args.alias, args.controllerPairingId, args.force) controller.save_data(args.file) print('Pairing for "{a}" was removed.'.format(a=args.alias))