1#!/usr/bin/env python3
2#
3#   Copyright 2019 - The Android Open Source Project
4#
5#   Licensed under the Apache License, Version 2.0 (the "License");
6#   you may not use this file except in compliance with the License.
7#   You may obtain a copy of the License at
8#
9#       http://www.apache.org/licenses/LICENSE-2.0
10#
11#   Unless required by applicable law or agreed to in writing, software
12#   distributed under the License is distributed on an "AS IS" BASIS,
13#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14#   See the License for the specific language governing permissions and
15#   limitations under the License.
16
17from blueberry.tests.gd.cert.event_stream import EventStream
18from blueberry.tests.gd.cert.truth import assertThat
19from blueberry.tests.gd.cert import gd_base_test
20from google.protobuf import empty_pb2 as empty_proto
21from blueberry.facade.hci import hci_facade_pb2 as hci_facade
22from blueberry.facade.hci import le_advertising_manager_facade_pb2 as le_advertising_facade
23from blueberry.facade.hci import le_initiator_address_facade_pb2 as le_initiator_address_facade
24from blueberry.facade.hci import le_scanning_manager_facade_pb2 as le_scanning_facade
25from blueberry.facade import common_pb2 as common
26from mobly import test_runner
27import hci_packets as hci
28
29
30class LeScanningWithSecurityTest(gd_base_test.GdBaseTestClass):
31
32    def setup_class(self):
33        gd_base_test.GdBaseTestClass.setup_class(self, dut_module='SECURITY', cert_module='HCI_INTERFACES')
34
35    def register_for_event(self, event_code):
36        msg = hci_facade.EventCodeMsg(code=int(event_code))
37        self.cert.hci.RegisterEventHandler(msg)
38
39    def register_for_le_event(self, event_code):
40        msg = hci_facade.LeSubeventCodeMsg(code=int(event_code))
41        self.cert.hci.RegisterLeEventHandler(msg)
42
43    def enqueue_hci_command(self, command, expect_complete):
44        cmd_bytes = bytes(command.serialize())
45        cmd = common.Data(command=cmd_bytes)
46        if (expect_complete):
47            self.cert.hci.EnqueueCommandWithComplete(cmd)
48        else:
49            self.cert.hci.EnqueueCommandWithStatus(cmd)
50
51    def test_le_ad_scan_dut_scans(self):
52        """
53            Verify that the IUT address policy is correctly initiated by SecurityManager, and we can start a scan.
54        """
55        cert_privacy_policy = le_initiator_address_facade.PrivacyPolicy(
56            address_policy=le_initiator_address_facade.AddressPolicy.USE_STATIC_ADDRESS,
57            address_with_type=common.BluetoothAddressWithType(
58                address=common.BluetoothAddress(address=bytes(b'C0:05:04:03:02:01')),
59                type=common.RANDOM_DEVICE_ADDRESS),
60            rotation_irk=b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00',
61            minimum_rotation_time=0,
62            maximum_rotation_time=0)
63        self.cert.hci_le_initiator_address.SetPrivacyPolicyForInitiatorAddress(cert_privacy_policy)
64        with EventStream(
65                # DUT Scans
66                self.dut.hci_le_scanning_manager.FetchAdvertisingReports(empty_proto.Empty()
67                                                                        )) as advertising_event_stream:
68
69            # CERT Advertises
70            gap_name = hci.GapData(data_type=hci.GapDataType.COMPLETE_LOCAL_NAME, data=list(bytes(b'Im_The_CERT!')))
71            gap_data = le_advertising_facade.GapDataMsg(data=gap_name.serialize())
72            gap_scan_name = hci.GapData(data_type=hci.GapDataType.SHORTENED_LOCAL_NAME, data=list(bytes(b'CERT!')))
73            gap_scan_data = le_advertising_facade.GapDataMsg(data=gap_scan_name.serialize())
74            config = le_advertising_facade.AdvertisingConfig(
75                advertisement=[gap_data],
76                scan_response=[gap_scan_data],
77                interval_min=512,
78                interval_max=768,
79                advertising_type=le_advertising_facade.AdvertisingEventType.ADV_IND,
80                own_address_type=common.USE_RANDOM_DEVICE_ADDRESS,
81                channel_map=7,
82                filter_policy=le_advertising_facade.AdvertisingFilterPolicy.ALL_DEVICES)
83            request = le_advertising_facade.CreateAdvertiserRequest(config=config)
84
85            create_response = self.cert.hci_le_advertising_manager.CreateAdvertiser(request)
86
87            scan_request = le_scanning_facade.ScanRequest(start=True)
88            self.dut.hci_le_scanning_manager.Scan(scan_request)
89
90            assertThat(advertising_event_stream).emits(lambda packet: b'Im_The_CERT' in packet.event)
91
92            remove_request = le_advertising_facade.RemoveAdvertiserRequest(advertiser_id=create_response.advertiser_id)
93            self.cert.hci_le_advertising_manager.RemoveAdvertiser(remove_request)
94
95
96if __name__ == '__main__':
97    test_runner.main()
98