You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
293 lines
14 KiB
293 lines
14 KiB
#!/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.gd_base_test import GdBaseTestClass
|
|
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 LeAclManagerTest(GdBaseTestClass):
|
|
|
|
def setup_class(self):
|
|
super().setup_class(dut_module='HCI_INTERFACES', cert_module='HCI')
|
|
|
|
def setup_test(self):
|
|
super().setup_test()
|
|
self.dut_le_acl_manager = PyLeAclManager(self.dut)
|
|
self.cert_hci_le_event_stream = EventStream(self.cert.hci.StreamLeSubevents(empty_proto.Empty()))
|
|
self.cert_acl_data_stream = EventStream(self.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)
|
|
super().teardown_test()
|
|
|
|
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\x13\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.GetPeerResolvablePrivateAddress()
|
|
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.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\x13\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)
|