Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Switch to tlv8 python lib #180

Merged
merged 4 commits into from
Mar 21, 2020
Merged

Switch to tlv8 python lib #180

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
17beff1
first working version without internal tlv code
jlusiardi Mar 17, 2020
811ec8d
Remove old TLV implementation
jlusiardi Mar 20, 2020
c262a99
Analyse test error
jlusiardi Mar 21, 2020
4dca1c8
Fix test for list_pairings
jlusiardi Mar 21, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
248 changes: 120 additions & 128 deletions homekit/accessoryserver.py
View file
Open in desktop

Large diffs are not rendered by default.

144 changes: 77 additions & 67 deletions homekit/controller/ble_impl/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,8 @@
#

__all__ = [
'Device', 'BlePairing', 'BleSession', 'find_characteristic_by_uuid', 'create_ble_pair_setup_write'
'Device', 'BlePairing', 'BleSession', 'find_characteristic_by_uuid', 'create_ble_pair_setup_write',
'AdditionalParameterTypes'
]

import time
Expand All @@ -25,9 +26,10 @@
import uuid
import struct
from distutils.util import strtobool
import tlv8

from homekit.controller.tools import AbstractPairing
from homekit.protocol.tlv import TLV
from homekit.protocol import Methods, States, TlvTypes
from homekit.model.characteristics import CharacteristicsTypes
from homekit.protocol import get_session_keys
from homekit.protocol.opcodes import HapBleOpCodes
Expand All @@ -37,6 +39,7 @@
from homekit.model.characteristics.characteristic_formats import BleCharacteristicFormats, CharacteristicFormats
from homekit.model.characteristics.characteristic_units import BleCharacteristicUnits
from homekit.exceptions import FormatError, RequestRejected, AccessoryDisconnectedError
from homekit.controller.ble_impl.additional_parameter_types import AdditionalParameterTypes

from homekit.tools import BLE_TRANSPORT_SUPPORTED

Expand Down Expand Up @@ -83,13 +86,13 @@ def list_accessories_and_characteristics(self):
def list_pairings(self):
if not self.session:
self.session = BleSession(self.pairing_data, self.adapter)
request_tlv = TLV.encode_list([
(TLV.kTLVType_State, TLV.M1),
(TLV.kTLVType_Method, TLV.ListPairings)
request_tlv = tlv8.encode([
tlv8.Entry(TlvTypes.State, States.M1),
tlv8.Entry(TlvTypes.Method, Methods.ListPairings)
])
request_tlv = TLV.encode_list([
(TLV.kTLVHAPParamParamReturnResponse, bytearray(b'\x01')),
(TLV.kTLVHAPParamValue, request_tlv)
request_tlv = tlv8.encode([
tlv8.Entry(AdditionalParameterTypes.ParamReturnResponse, bytearray(b'\x01')),
tlv8.Entry(AdditionalParameterTypes.Value, request_tlv)
])
body = len(request_tlv).to_bytes(length=2, byteorder='little') + request_tlv

Expand All @@ -101,22 +104,23 @@ def list_pairings(self):
cid = c['iid']
fc, _ = self.session.find_characteristic_by_iid(cid)
response = self.session.request(fc, cid, HapBleOpCodes.CHAR_WRITE, body)
response = TLV.decode_bytes(response[1])
response = tlv8.decode(response.first_by_id(AdditionalParameterTypes.Value).data)
tmp = []
r = {}
for d in response[1:]:
if d[0] == TLV.kTLVType_Identifier:
if d.type_id == TlvTypes.Identifier:
r = {}
tmp.append(r)
r['pairingId'] = d[1].decode()
if d[0] == TLV.kTLVType_PublicKey:
r['publicKey'] = d[1].hex()
if d[0] == TLV.kTLVType_Permissions:
r['pairingId'] = d.data.decode()
if d.type_id == TlvTypes.PublicKey:
r['publicKey'] = d.data.hex()
if d.type_id == TlvTypes.Permissions:
controller_type = 'regular'
if d[1] == b'\x01':
if d.data == b'\x01':
controller_type = 'admin'
r['permissions'] = int.from_bytes(d[1], byteorder='little')
r['permissions'] = int.from_bytes(d.data, byteorder='little')
r['controllerType'] = controller_type
tmp.sort(key=lambda x: x['pairingId'])
return tmp

def get_events(self, characteristics, callback_fun, max_events=-1, max_seconds=-1):
Expand Down Expand Up @@ -184,7 +188,10 @@ def get_characteristics(self, characteristics, include_meta=False, include_perms
self.session = None
raise e

value = self._convert_to_python(aid, cid, response[1]) if 1 in response else None
if response.first_by_id(AdditionalParameterTypes.Value):
value = self._convert_to_python(aid, cid, response.first_by_id(AdditionalParameterTypes.Value).data)
else:
value = None

results[(aid, cid)] = {
'value': value
Expand Down Expand Up @@ -357,13 +364,15 @@ def put_characteristics(self, characteristics, do_conversion=False):
}
continue

value = TLV.encode_list([(1, self._convert_from_python(aid, cid, value))])
value = tlv8.encode([
tlv8.Entry(AdditionalParameterTypes.Value, self._convert_from_python(aid, cid, value))
])
body = len(value).to_bytes(length=2, byteorder='little') + value

try:
fc, fc_info = self.session.find_characteristic_by_iid(cid)
response = self.session.request(fc, cid, HapBleOpCodes.CHAR_WRITE, body)
logger.debug('response %s', response)
logger.debug('response %s', tlv8.format_string(response))
# TODO does the response contain useful information here?
except RequestRejected as e:
results[(aid, cid)] = {
Expand All @@ -381,23 +390,23 @@ def add_pairing(self, additional_controller_pairing_identifier, ios_device_ltpk,
if not self.session:
self.session = BleSession(self.pairing_data, self.adapter)
if permissions == 'User':
permissions = TLV.kTLVType_Permission_RegularUser
permissions = TlvTypes.Permission_RegularUser
elif permissions == 'Admin':
permissions = TLV.kTLVType_Permission_AdminUser
permissions = TlvTypes.Permission_AdminUser
else:
print('UNKNOWN')

request_tlv = TLV.encode_list([
(TLV.kTLVType_State, TLV.M1),
(TLV.kTLVType_Method, TLV.AddPairing),
(TLV.kTLVType_Identifier, additional_controller_pairing_identifier.encode()),
(TLV.kTLVType_PublicKey, bytes.fromhex(ios_device_ltpk)),
(TLV.kTLVType_Permissions, permissions)
request_tlv = tlv8.encode([
tlv8.Entry(TlvTypes.State, States.M1),
tlv8.Entry(TlvTypes.Method, Methods.AddPairing),
tlv8.Entry(TlvTypes.Identifier, additional_controller_pairing_identifier.encode()),
tlv8.Entry(TlvTypes.PublicKey, bytes.fromhex(ios_device_ltpk)),
tlv8.Entry(TlvTypes.Permissions, permissions)
])

request_tlv = TLV.encode_list([
(TLV.kTLVHAPParamParamReturnResponse, bytearray(b'\x01')),
(TLV.kTLVHAPParamValue, request_tlv)
request_tlv = tlv8.encode([
tlv8.Entry(AdditionalParameterTypes.ParamReturnResponse, bytearray(b'\x01')),
tlv8.Entry(AdditionalParameterTypes.Value, request_tlv)
])
body = len(request_tlv).to_bytes(length=2, byteorder='little') + request_tlv

Expand Down Expand Up @@ -563,9 +572,9 @@ def request(self, feature_char, feature_char_id, op, body=None):
logger.debug('expected body length %d (got %d)', length, len(data[5:]))

# parse tlvs and analyse information
tlv = TLV.decode_bytes(data[5:])
logger.debug('received TLV: %s', TLV.to_string(tlv))
return dict(tlv)
tlv = tlv8.decode(data[5:])
logger.debug('received TLV: %s', tlv8.format_string(tlv))
return tlv


def find_characteristic_by_uuid(device, service_uuid, char_uuid):
Expand Down Expand Up @@ -613,11 +622,11 @@ def find_characteristic_by_uuid(device, service_uuid, char_uuid):
def create_ble_pair_setup_write(characteristic, characteristic_id):
def write(request, expected):
# TODO document me
body = TLV.encode_list(request)
logger.debug('entering write function %s', TLV.to_string(TLV.decode_bytes(body)))
request_tlv = TLV.encode_list([
(TLV.kTLVHAPParamParamReturnResponse, bytearray(b'\x01')),
(TLV.kTLVHAPParamValue, body)
body = tlv8.encode(request)
logger.debug('entering write function %s', tlv8.format_string(tlv8.decode(body)))
request_tlv = tlv8.encode([
tlv8.Entry(AdditionalParameterTypes.ParamReturnResponse, bytearray(b'\x01')),
tlv8.Entry(AdditionalParameterTypes.Value, body)
])
transaction_id = random.randrange(0, 255)
data = bytearray([0x00, HapBleOpCodes.CHAR_WRITE, transaction_id])
Expand Down Expand Up @@ -647,9 +656,10 @@ def write(request, expected):

logger.debug('received %s', bytes(resp_data).hex())
logger.debug('decode %s', bytes(resp_data[5:]).hex())
resp_tlv = TLV.decode_bytes(bytes([int(a) for a in resp_data[5:]]), expected=[TLV.kTLVHAPParamValue])
result = TLV.decode_bytes(resp_tlv[0][1], expected)
logger.debug('leaving write function %s', TLV.to_string(result))
resp_tlv = tlv8.decode(bytes([int(a) for a in resp_data[5:]]),
expected={AdditionalParameterTypes.Value: tlv8.DataType.BYTES})
result = tlv8.decode(resp_tlv.first_by_id(AdditionalParameterTypes.Value).data, expected)
logger.debug('leaving write function %s', tlv8.format_string(result))
return result

return write
Expand Down Expand Up @@ -766,54 +776,54 @@ def parse_sig_read_response(data, expected_tid):
logger.debug('expected body length %d (got %d)', length, len(data[5:]))

# parse tlvs and analyse information
tlv = TLV.decode_bytes(data[5:])
tlv = tlv8.decode(data[5:])

description = ''
characteristic_format = ''
characteristic_range = None
characteristic_step = None
for t in tlv:
if t[0] == TLV.kTLVHAPParamCharacteristicType:
chr_type = [int(a) for a in t[1]]
if t.type_id == AdditionalParameterTypes.CharacteristicType:
chr_type = [int(a) for a in t.data]
chr_type.reverse()
chr_type = str(uuid.UUID(''.join('%02x' % b for b in chr_type)))
if t[0] == TLV.kTLVHAPParamServiceInstanceId:
svc_id = int.from_bytes(t[1], byteorder='little')
if t[0] == TLV.kTLVHAPParamServiceType:
svc_type = [int(a) for a in t[1]]
if t.type_id == AdditionalParameterTypes.ServiceInstanceId:
svc_id = int.from_bytes(t.data, byteorder='little')
if t.type_id == AdditionalParameterTypes.ServiceType:
svc_type = [int(a) for a in t.data]
svc_type.reverse()
svc_type = str(uuid.UUID(''.join('%02x' % b for b in svc_type)))
if t[0] == TLV.kTLVHAPParamHAPCharacteristicPropertiesDescriptor:
chr_prop_int = int.from_bytes(t[1], byteorder='little')
if t[0] == TLV.kTLVHAPParamGATTUserDescriptionDescriptor:
description = t[1].decode()
if t[0] == TLV.kTLVHAPParamHAPValidValuesDescriptor:
print('valid values', t[1])
if t[0] == TLV.kTLVHAPParamHAPValidValuesRangeDescriptor:
print('valid values range', t[1])
if t[0] == TLV.kTLVHAPParamGATTPresentationFormatDescriptor:
unit_bytes = t[1][2:4]
if t.type_id == AdditionalParameterTypes.HAPCharacteristicPropertiesDescriptor:
chr_prop_int = int.from_bytes(t.data, byteorder='little')
if t.type_id == AdditionalParameterTypes.GATTUserDescriptionDescriptor:
description = t.data.decode()
if t.type_id == AdditionalParameterTypes.HAPValidValuesDescriptor:
print('valid values', t.data)
if t.type_id == AdditionalParameterTypes.HAPValidValuesRangeDescriptor:
print('valid values range', t.data)
if t.type_id == AdditionalParameterTypes.GATTPresentationFormatDescriptor:
unit_bytes = bytearray(t.data[2:4])
unit_bytes.reverse()
characteristic_format = BleCharacteristicFormats.get(int(t[1][0]), 'unknown')
characteristic_format = BleCharacteristicFormats.get(int(t.data[0]), 'unknown')
unit = BleCharacteristicUnits.get(int.from_bytes(unit_bytes, byteorder='big'), 'unknown')
if t[0] == TLV.kTLVHAPParamGATTValidRange:
logger.debug('range: %s', t[1].hex())
if t.type_id == AdditionalParameterTypes.GATTValidRange:
logger.debug('range: %s', t.data.hex())
lower = None
upper = None
if characteristic_format == 'int32' or characteristic_format == 'int':
(lower, upper) = struct.unpack('ii', t[1])
(lower, upper) = struct.unpack('ii', t.data)
if characteristic_format == 'uint8':
(lower, upper) = struct.unpack('BB', t[1])
(lower, upper) = struct.unpack('BB', t.data)
if characteristic_format == 'float':
(lower, upper) = struct.unpack('ff', t[1])
(lower, upper) = struct.unpack('ff', t.data)
# TODO include all formats!
characteristic_range = (lower, upper)
if t[0] == TLV.kTLVHAPParamHAPStepValueDescriptor:
if t.type_id == AdditionalParameterTypes.HAPStepValueDescriptor:
characteristic_step = None
if characteristic_format == 'int32':
characteristic_step = struct.unpack('i', t[1])[0]
characteristic_step = struct.unpack('i', t.data)[0]
if characteristic_format == 'uint8':
characteristic_step = struct.unpack('B', t[1])[0]
characteristic_step = struct.unpack('B', t.data)[0]
# TODO include all formats!

# parse permissions
Expand Down
37 changes: 37 additions & 0 deletions homekit/controller/ble_impl/additional_parameter_types.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,37 @@
# Copyright 2020 Joachim Lusiardi
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
from enum import IntEnum


class AdditionalParameterTypes(IntEnum):
# Additional Parameter Types for BLE (Table 6-9 page 98)
Value = 0x01
AdditionalAuthorizationData = 0x02
Origin = 0x03
CharacteristicType = 0x04
CharacteristicInstanceId = 0x05
ServiceType = 0x06
ServiceInstanceId = 0x07
TTL = 0x08
ParamReturnResponse = 0x09
HAPCharacteristicPropertiesDescriptor = 0x0a
GATTUserDescriptionDescriptor = 0x0b
GATTPresentationFormatDescriptor = 0x0c
GATTValidRange = 0x0d
HAPStepValueDescriptor = 0x0e
HAPServiceProperties = 0x0f
HAPLinkedServices = 0x10
HAPValidValuesDescriptor = 0x11
HAPValidValuesRangeDescriptor = 0x12
Loading