| #!/usr/bin/env python3 |
| # |
| # Copyright 2020 - The Android Open Source Project |
| # |
| # 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 cert.closable import safeClose |
| from cert.event_stream import EventStream |
| from cert.truth import assertThat |
| from cert.py_le_acl_manager import PyLeAclManager |
| from google.protobuf import empty_pb2 as empty_proto |
| from facade import common_pb2 as common |
| from hci.facade import le_acl_manager_facade_pb2 as le_acl_manager_facade |
| from hci.facade import le_advertising_manager_facade_pb2 as le_advertising_facade |
| from hci.facade import le_initiator_address_facade_pb2 as le_initiator_address_facade |
| from hci.facade import hci_facade_pb2 as hci_facade |
| import bluetooth_packets_python3 as bt_packets |
| from bluetooth_packets_python3 import hci_packets |
| from bluetooth_packets_python3 import RawBuilder |
| |
| |
| class LeAclManagerTestBase(): |
| |
| def setup_test(self, dut, cert): |
| self.dut_le_acl_manager = PyLeAclManager(dut) |
| self.cert_hci_le_event_stream = EventStream(cert.hci.StreamLeSubevents(empty_proto.Empty())) |
| self.cert_acl_data_stream = EventStream(cert.hci.StreamAcl(empty_proto.Empty())) |
| |
| def teardown_test(self): |
| safeClose(self.cert_hci_le_event_stream) |
| safeClose(self.cert_acl_data_stream) |
| safeClose(self.dut_le_acl_manager) |
| |
| def set_privacy_policy_static(self): |
| self.dut_address = b'd0:05:04:03:02:01' |
| private_policy = le_initiator_address_facade.PrivacyPolicy( |
| address_policy=le_initiator_address_facade.AddressPolicy.USE_STATIC_ADDRESS, |
| address_with_type=common.BluetoothAddressWithType( |
| address=common.BluetoothAddress(address=bytes(self.dut_address)), type=common.RANDOM_DEVICE_ADDRESS)) |
| self.dut.hci_le_initiator_address.SetPrivacyPolicyForInitiatorAddress(private_policy) |
| |
| def register_for_event(self, event_code): |
| msg = hci_facade.EventRequest(code=int(event_code)) |
| self.cert.hci.RequestEvent(msg) |
| |
| def register_for_le_event(self, event_code): |
| msg = hci_facade.EventRequest(code=int(event_code)) |
| self.cert.hci.RequestLeSubevent(msg) |
| |
| def enqueue_hci_command(self, command): |
| cmd_bytes = bytes(command.Serialize()) |
| cmd = common.Data(payload=cmd_bytes) |
| self.cert.hci.SendCommand(cmd) |
| |
| def enqueue_acl_data(self, handle, pb_flag, b_flag, data): |
| acl = hci_packets.AclBuilder(handle, pb_flag, b_flag, RawBuilder(data)) |
| self.cert.hci.SendAcl(common.Data(payload=bytes(acl.Serialize()))) |
| |
| def dut_connects(self, check_address): |
| self.register_for_le_event(hci_packets.SubeventCode.CONNECTION_COMPLETE) |
| self.register_for_le_event(hci_packets.SubeventCode.ENHANCED_CONNECTION_COMPLETE) |
| |
| # Cert Advertises |
| advertising_handle = 0 |
| self.enqueue_hci_command( |
| hci_packets.LeSetExtendedAdvertisingLegacyParametersBuilder( |
| advertising_handle, |
| hci_packets.LegacyAdvertisingProperties.ADV_IND, |
| 400, |
| 450, |
| 7, |
| hci_packets.OwnAddressType.RANDOM_DEVICE_ADDRESS, |
| hci_packets.PeerAddressType.PUBLIC_DEVICE_OR_IDENTITY_ADDRESS, |
| '00:00:00:00:00:00', |
| hci_packets.AdvertisingFilterPolicy.ALL_DEVICES, |
| 0xF8, |
| 1, #SID |
| hci_packets.Enable.DISABLED # Scan request notification |
| )) |
| |
| self.enqueue_hci_command( |
| hci_packets.LeSetExtendedAdvertisingRandomAddressBuilder(advertising_handle, '0C:05:04:03:02:01')) |
| |
| gap_name = hci_packets.GapData() |
| gap_name.data_type = hci_packets.GapDataType.COMPLETE_LOCAL_NAME |
| gap_name.data = list(bytes(b'Im_A_Cert')) |
| |
| self.enqueue_hci_command( |
| hci_packets.LeSetExtendedAdvertisingDataBuilder( |
| advertising_handle, hci_packets.Operation.COMPLETE_ADVERTISEMENT, |
| hci_packets.FragmentPreference.CONTROLLER_SHOULD_NOT, [gap_name])) |
| |
| gap_short_name = hci_packets.GapData() |
| gap_short_name.data_type = hci_packets.GapDataType.SHORTENED_LOCAL_NAME |
| gap_short_name.data = list(bytes(b'Im_A_C')) |
| |
| self.enqueue_hci_command( |
| hci_packets.LeSetExtendedAdvertisingScanResponseBuilder( |
| advertising_handle, hci_packets.Operation.COMPLETE_ADVERTISEMENT, |
| hci_packets.FragmentPreference.CONTROLLER_SHOULD_NOT, [gap_short_name])) |
| |
| enabled_set = hci_packets.EnabledSet() |
| enabled_set.advertising_handle = advertising_handle |
| enabled_set.duration = 0 |
| enabled_set.max_extended_advertising_events = 0 |
| self.enqueue_hci_command( |
| hci_packets.LeSetExtendedAdvertisingEnableBuilder(hci_packets.Enable.ENABLED, [enabled_set])) |
| |
| self.dut_le_acl = self.dut_le_acl_manager.connect_to_remote( |
| remote_addr=common.BluetoothAddressWithType( |
| address=common.BluetoothAddress(address=bytes('0C:05:04:03:02:01', 'utf8')), |
| type=int(hci_packets.AddressType.RANDOM_DEVICE_ADDRESS))) |
| |
| # Cert gets ConnectionComplete with a handle and sends ACL data |
| handle = 0xfff |
| address = hci_packets.Address() |
| |
| def get_handle(packet): |
| packet_bytes = packet.payload |
| nonlocal handle |
| nonlocal address |
| if b'\x3e\x13\x01\x00' in packet_bytes: |
| cc_view = hci_packets.LeConnectionCompleteView( |
| hci_packets.LeMetaEventView( |
| hci_packets.EventView(bt_packets.PacketViewLittleEndian(list(packet_bytes))))) |
| handle = cc_view.GetConnectionHandle() |
| address = cc_view.GetPeerAddress() |
| return True |
| if b'\x3e\x1f\x0A\x00' in packet_bytes: |
| cc_view = hci_packets.LeEnhancedConnectionCompleteView( |
| hci_packets.LeMetaEventView( |
| hci_packets.EventView(bt_packets.PacketViewLittleEndian(list(packet_bytes))))) |
| handle = cc_view.GetConnectionHandle() |
| address = cc_view.GetPeerAddress() |
| return True |
| return False |
| |
| self.cert_hci_le_event_stream.assert_event_occurs(get_handle) |
| self.cert_handle = handle |
| dut_address_from_complete = address |
| if check_address: |
| assertThat(dut_address_from_complete).isEqualTo(self.dut_address.decode()) |
| |
| def send_receive_and_check(self): |
| self.enqueue_acl_data(self.cert_handle, hci_packets.PacketBoundaryFlag.FIRST_NON_AUTOMATICALLY_FLUSHABLE, |
| hci_packets.BroadcastFlag.POINT_TO_POINT, |
| bytes(b'\x19\x00\x07\x00SomeAclData from the Cert')) |
| |
| self.dut_le_acl.send(b'\x1C\x00\x07\x00SomeMoreAclData from the DUT') |
| self.cert_acl_data_stream.assert_event_occurs(lambda packet: b'SomeMoreAclData' in packet.payload) |
| assertThat(self.dut_le_acl).emits(lambda packet: b'SomeAclData' in packet.payload) |
| |
| def test_dut_connects(self): |
| self.set_privacy_policy_static() |
| self.dut_connects(check_address=True) |
| self.send_receive_and_check() |
| |
| def test_dut_connects_resolvable_address(self): |
| privacy_policy = le_initiator_address_facade.PrivacyPolicy( |
| address_policy=le_initiator_address_facade.AddressPolicy.USE_RESOLVABLE_ADDRESS, |
| rotation_irk=b'\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f', |
| minimum_rotation_time=7 * 60 * 1000, |
| maximum_rotation_time=15 * 60 * 1000) |
| self.dut.hci_le_initiator_address.SetPrivacyPolicyForInitiatorAddress(privacy_policy) |
| self.dut_connects(check_address=False) |
| self.send_receive_and_check() |
| |
| def test_dut_connects_non_resolvable_address(self): |
| privacy_policy = le_initiator_address_facade.PrivacyPolicy( |
| address_policy=le_initiator_address_facade.AddressPolicy.USE_NON_RESOLVABLE_ADDRESS, |
| rotation_irk=b'\x10\x11\x12\x13\x14\x15\x16\x17\x18\x19\x1a\x1b\x1c\x1d\x1e\x1f', |
| minimum_rotation_time=8 * 60 * 1000, |
| maximum_rotation_time=14 * 60 * 1000) |
| self.dut.hci_le_initiator_address.SetPrivacyPolicyForInitiatorAddress(privacy_policy) |
| self.dut_connects(check_address=False) |
| self.send_receive_and_check() |
| |
| def test_dut_connects_public_address(self): |
| self.dut.hci_le_initiator_address.SetPrivacyPolicyForInitiatorAddress( |
| le_initiator_address_facade.PrivacyPolicy( |
| address_policy=le_initiator_address_facade.AddressPolicy.USE_PUBLIC_ADDRESS)) |
| self.dut_connects(check_address=False) |
| self.send_receive_and_check() |
| |
| def test_dut_connects_public_address_cancelled(self): |
| self.dut.hci_le_initiator_address.SetPrivacyPolicyForInitiatorAddress( |
| le_initiator_address_facade.PrivacyPolicy( |
| address_policy=le_initiator_address_facade.AddressPolicy.USE_PUBLIC_ADDRESS)) |
| self.dut_connects(check_address=False) |
| self.send_receive_and_check() |
| |
| def test_cert_connects(self): |
| self.set_privacy_policy_static() |
| self.register_for_le_event(hci_packets.SubeventCode.CONNECTION_COMPLETE) |
| self.register_for_le_event(hci_packets.SubeventCode.ENHANCED_CONNECTION_COMPLETE) |
| |
| self.dut_le_acl_manager.listen_for_incoming_connections() |
| |
| # DUT Advertises |
| gap_name = hci_packets.GapData() |
| gap_name.data_type = hci_packets.GapDataType.COMPLETE_LOCAL_NAME |
| gap_name.data = list(bytes(b'Im_The_DUT')) |
| gap_data = le_advertising_facade.GapDataMsg(data=bytes(gap_name.Serialize())) |
| config = le_advertising_facade.AdvertisingConfig( |
| advertisement=[gap_data], |
| interval_min=512, |
| interval_max=768, |
| advertising_type=le_advertising_facade.AdvertisingEventType.ADV_IND, |
| own_address_type=common.USE_RANDOM_DEVICE_ADDRESS, |
| peer_address_type=common.PUBLIC_DEVICE_OR_IDENTITY_ADDRESS, |
| peer_address=common.BluetoothAddress(address=bytes(b'A6:A5:A4:A3:A2:A1')), |
| channel_map=7, |
| filter_policy=le_advertising_facade.AdvertisingFilterPolicy.ALL_DEVICES) |
| request = le_advertising_facade.CreateAdvertiserRequest(config=config) |
| |
| self.dut.hci_le_advertising_manager.CreateAdvertiser(request) |
| |
| # Cert Connects |
| self.enqueue_hci_command(hci_packets.LeSetRandomAddressBuilder('0C:05:04:03:02:01')) |
| phy_scan_params = hci_packets.LeCreateConnPhyScanParameters() |
| phy_scan_params.scan_interval = 0x60 |
| phy_scan_params.scan_window = 0x30 |
| phy_scan_params.conn_interval_min = 0x18 |
| phy_scan_params.conn_interval_max = 0x28 |
| phy_scan_params.conn_latency = 0 |
| phy_scan_params.supervision_timeout = 0x1f4 |
| phy_scan_params.min_ce_length = 0 |
| phy_scan_params.max_ce_length = 0 |
| self.enqueue_hci_command( |
| hci_packets.LeExtendedCreateConnectionBuilder(hci_packets.InitiatorFilterPolicy.USE_PEER_ADDRESS, |
| hci_packets.OwnAddressType.RANDOM_DEVICE_ADDRESS, |
| hci_packets.AddressType.RANDOM_DEVICE_ADDRESS, |
| self.dut_address.decode(), 1, [phy_scan_params])) |
| |
| # Cert gets ConnectionComplete with a handle and sends ACL data |
| handle = 0xfff |
| |
| def get_handle(packet): |
| packet_bytes = packet.payload |
| nonlocal handle |
| if b'\x3e\x13\x01\x00' in packet_bytes: |
| cc_view = hci_packets.LeConnectionCompleteView( |
| hci_packets.LeMetaEventView( |
| hci_packets.EventView(bt_packets.PacketViewLittleEndian(list(packet_bytes))))) |
| handle = cc_view.GetConnectionHandle() |
| return True |
| if b'\x3e\x1f\x0A\x00' in packet_bytes: |
| cc_view = hci_packets.LeEnhancedConnectionCompleteView( |
| hci_packets.LeMetaEventView( |
| hci_packets.EventView(bt_packets.PacketViewLittleEndian(list(packet_bytes))))) |
| handle = cc_view.GetConnectionHandle() |
| return True |
| return False |
| |
| self.cert_hci_le_event_stream.assert_event_occurs(get_handle) |
| self.cert_handle = handle |
| |
| self.enqueue_acl_data(self.cert_handle, hci_packets.PacketBoundaryFlag.FIRST_NON_AUTOMATICALLY_FLUSHABLE, |
| hci_packets.BroadcastFlag.POINT_TO_POINT, |
| bytes(b'\x19\x00\x07\x00SomeAclData from the Cert')) |
| |
| # DUT gets a connection complete event and sends and receives |
| handle = 0xfff |
| self.dut_le_acl = self.dut_le_acl_manager.complete_incoming_connection() |
| |
| self.send_receive_and_check() |
| |
| def test_recombination_l2cap_packet(self): |
| self.set_privacy_policy_static() |
| self.dut_connects(check_address=True) |
| |
| self.enqueue_acl_data(self.cert_handle, hci_packets.PacketBoundaryFlag.FIRST_NON_AUTOMATICALLY_FLUSHABLE, |
| hci_packets.BroadcastFlag.POINT_TO_POINT, bytes(b'\x06\x00\x07\x00Hello')) |
| self.enqueue_acl_data(self.cert_handle, hci_packets.PacketBoundaryFlag.CONTINUING_FRAGMENT, |
| hci_packets.BroadcastFlag.POINT_TO_POINT, bytes(b'!')) |
| |
| assertThat(self.dut_le_acl).emits(lambda packet: b'Hello!' in packet.payload) |
| |
| def test_background_connection(self): |
| self.register_for_le_event(hci_packets.SubeventCode.CONNECTION_COMPLETE) |
| self.register_for_le_event(hci_packets.SubeventCode.ENHANCED_CONNECTION_COMPLETE) |
| self.set_privacy_policy_static() |
| |
| # Start background and direct connection |
| token = self.dut_le_acl_manager.initiate_background_and_direct_connection( |
| remote_addr=common.BluetoothAddressWithType( |
| address=common.BluetoothAddress(address=bytes('0C:05:04:03:02:01', 'utf8')), |
| type=int(hci_packets.AddressType.RANDOM_DEVICE_ADDRESS))) |
| |
| # Wait for direct connection timeout |
| self.dut_le_acl_manager.wait_for_connection_fail(token) |
| |
| # Cert Advertises |
| advertising_handle = 0 |
| self.enqueue_hci_command( |
| hci_packets.LeSetExtendedAdvertisingLegacyParametersBuilder( |
| advertising_handle, |
| hci_packets.LegacyAdvertisingProperties.ADV_IND, |
| 400, |
| 450, |
| 7, |
| hci_packets.OwnAddressType.RANDOM_DEVICE_ADDRESS, |
| hci_packets.PeerAddressType.PUBLIC_DEVICE_OR_IDENTITY_ADDRESS, |
| '00:00:00:00:00:00', |
| hci_packets.AdvertisingFilterPolicy.ALL_DEVICES, |
| 0xF8, |
| 1, #SID |
| hci_packets.Enable.DISABLED # Scan request notification |
| )) |
| |
| self.enqueue_hci_command( |
| hci_packets.LeSetExtendedAdvertisingRandomAddressBuilder(advertising_handle, '0C:05:04:03:02:01')) |
| |
| gap_name = hci_packets.GapData() |
| gap_name.data_type = hci_packets.GapDataType.COMPLETE_LOCAL_NAME |
| gap_name.data = list(bytes(b'Im_A_Cert')) |
| |
| self.enqueue_hci_command( |
| hci_packets.LeSetExtendedAdvertisingDataBuilder( |
| advertising_handle, hci_packets.Operation.COMPLETE_ADVERTISEMENT, |
| hci_packets.FragmentPreference.CONTROLLER_SHOULD_NOT, [gap_name])) |
| |
| gap_short_name = hci_packets.GapData() |
| gap_short_name.data_type = hci_packets.GapDataType.SHORTENED_LOCAL_NAME |
| gap_short_name.data = list(bytes(b'Im_A_C')) |
| |
| self.enqueue_hci_command( |
| hci_packets.LeSetExtendedAdvertisingScanResponseBuilder( |
| advertising_handle, hci_packets.Operation.COMPLETE_ADVERTISEMENT, |
| hci_packets.FragmentPreference.CONTROLLER_SHOULD_NOT, [gap_short_name])) |
| |
| enabled_set = hci_packets.EnabledSet() |
| enabled_set.advertising_handle = advertising_handle |
| enabled_set.duration = 0 |
| enabled_set.max_extended_advertising_events = 0 |
| self.enqueue_hci_command( |
| hci_packets.LeSetExtendedAdvertisingEnableBuilder(hci_packets.Enable.ENABLED, [enabled_set])) |
| |
| # Check background connection complete |
| self.dut_le_acl_manager.complete_outgoing_connection(token) |
