| #!/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. |
| |
| import logging |
| |
| from bluetooth_packets_python3 import hci_packets |
| from cert.closable import Closable |
| from cert.closable import safeClose |
| from cert.event_stream import EventStream |
| from cert.truth import assertThat |
| from facade import common_pb2 as common |
| from google.protobuf import empty_pb2 as empty_proto |
| |
| from security.facade_pb2 import AuthenticationRequirements |
| from security.facade_pb2 import AuthenticationRequirementsMessage |
| from security.facade_pb2 import BondMsgType |
| from security.facade_pb2 import SecurityPolicyMessage |
| from security.facade_pb2 import IoCapabilities |
| from security.facade_pb2 import IoCapabilityMessage |
| from security.facade_pb2 import OobDataBondMessage |
| from security.facade_pb2 import OobDataMessage |
| from security.facade_pb2 import OobDataPresentMessage |
| from security.facade_pb2 import UiMsgType |
| from security.facade_pb2 import UiCallbackMsg |
| from security.facade_pb2 import UiCallbackType |
| |
| |
| class PySecurity(Closable): |
| """ |
| Abstraction for security tasks and GRPC calls |
| """ |
| |
| _io_capabilities_name_lookup = { |
| IoCapabilities.DISPLAY_ONLY: "DISPLAY_ONLY", |
| IoCapabilities.DISPLAY_YES_NO_IO_CAP: "DISPLAY_YES_NO_IO_CAP", |
| IoCapabilities.KEYBOARD_ONLY: "KEYBOARD_ONLY", |
| IoCapabilities.NO_INPUT_NO_OUTPUT: "NO_INPUT_NO_OUTPUT", |
| } |
| |
| _auth_reqs_name_lookup = { |
| AuthenticationRequirements.NO_BONDING: "NO_BONDING", |
| AuthenticationRequirements.NO_BONDING_MITM_PROTECTION: "NO_BONDING_MITM_PROTECTION", |
| AuthenticationRequirements.DEDICATED_BONDING: "DEDICATED_BONDING", |
| AuthenticationRequirements.DEDICATED_BONDING_MITM_PROTECTION: "DEDICATED_BONDING_MITM_PROTECTION", |
| AuthenticationRequirements.GENERAL_BONDING: "GENERAL_BONDING", |
| AuthenticationRequirements.GENERAL_BONDING_MITM_PROTECTION: "GENERAL_BONDING_MITM_PROTECTION", |
| } |
| |
| _ui_event_stream = None |
| _bond_event_stream = None |
| _oob_data_event_stream = None |
| |
| def __init__(self, device): |
| logging.info("DUT: Init") |
| self._device = device |
| self._device.wait_channel_ready() |
| self._ui_event_stream = EventStream(self._device.security.FetchUiEvents(empty_proto.Empty())) |
| self._bond_event_stream = EventStream(self._device.security.FetchBondEvents(empty_proto.Empty())) |
| self._enforce_security_policy_stream = EventStream( |
| self._device.security.FetchEnforceSecurityPolicyEvents(empty_proto.Empty())) |
| self._disconnect_event_stream = EventStream(self._device.security.FetchDisconnectEvents(empty_proto.Empty())) |
| self._oob_data_event_stream = EventStream( |
| self._device.security.FetchGetOutOfBandDataEvents(empty_proto.Empty())) |
| |
| def create_bond(self, address, type): |
| """ |
| Triggers stack under test to create bond |
| """ |
| logging.info("DUT: Creating bond to '%s' from '%s'" % (str(address), str(self._device.address))) |
| self._device.security.CreateBond( |
| common.BluetoothAddressWithType(address=common.BluetoothAddress(address=address), type=type)) |
| |
| def create_bond_out_of_band(self, address, type, p192_oob_data, p256_oob_data): |
| """ |
| Triggers stack under test to create bond using Out of Band method |
| """ |
| |
| logging.info("DUT: Creating OOB bond to '%s' from '%s'" % (str(address), str(self._device.address))) |
| |
| self._device.security.CreateBondOutOfBand( |
| OobDataBondMessage( |
| address=common.BluetoothAddressWithType(address=common.BluetoothAddress(address=address), type=type), |
| p192_data=OobDataMessage( |
| address=common.BluetoothAddressWithType( |
| address=common.BluetoothAddress(address=address), type=type), |
| confirmation_value=bytes(bytearray(p192_oob_data[0])), |
| random_value=bytes(bytearray(p192_oob_data[1]))), |
| p256_data=OobDataMessage( |
| address=common.BluetoothAddressWithType( |
| address=common.BluetoothAddress(address=address), type=type), |
| confirmation_value=bytes(bytearray(p256_oob_data[0])), |
| random_value=bytes(bytearray(p256_oob_data[1]))))) |
| |
| def remove_bond(self, address, type): |
| """ |
| Removes bond from stack under test |
| """ |
| self._device.security.RemoveBond( |
| common.BluetoothAddressWithType(address=common.BluetoothAddress(address=address), type=type)) |
| |
| def set_io_capabilities(self, io_capabilities): |
| """ |
| Set the IO Capabilities used for the DUT |
| """ |
| logging.info("DUT: setting IO Capabilities data to '%s'" % self._io_capabilities_name_lookup.get( |
| io_capabilities, "ERROR")) |
| self._device.security.SetIoCapability(IoCapabilityMessage(capability=io_capabilities)) |
| |
| def set_authentication_requirements(self, auth_reqs): |
| """ |
| Establish authentication requirements for the stack |
| """ |
| logging.info("DUT: setting Authentication Requirements data to '%s'" % self._auth_reqs_name_lookup.get( |
| auth_reqs, "ERROR")) |
| self._device.security.SetAuthenticationRequirements(AuthenticationRequirementsMessage(requirement=auth_reqs)) |
| |
| def __send_ui_callback(self, address, callback_type, b, uid, pin): |
| """ |
| Send a callback from the UI as if the user pressed a button on the dialog |
| """ |
| logging.info("DUT: Sending user input response uid: %d; response: %s" % (uid, b)) |
| self._device.security.SendUiCallback( |
| UiCallbackMsg( |
| message_type=callback_type, |
| boolean=b, |
| unique_id=uid, |
| pin=bytes(pin), |
| address=common.BluetoothAddressWithType( |
| address=common.BluetoothAddress(address=address), |
| type=common.BluetoothAddressTypeEnum.PUBLIC_DEVICE_ADDRESS))) |
| |
| def enable_secure_simple_pairing(self): |
| """ |
| This is called when you want to enable SSP for testing |
| Since the stack under test already enables it by default |
| we do not need to do anything here for the time being |
| """ |
| pass |
| |
| def enable_secure_connections(self): |
| pass |
| |
| def accept_pairing(self, cert_address, reply_boolean): |
| """ |
| Here we pass, but in cert we perform pairing flow tasks. |
| This was added here in order to be more dynamic, but the stack |
| under test will handle the pairing flow. |
| """ |
| pass |
| |
| def accept_oob_pairing(self, cert_address, reply_boolean): |
| """ |
| Here we pass, but in cert we perform pairing flow tasks. |
| This was added here in order to be more dynamic, but the stack |
| under test will handle the pairing flow. |
| """ |
| pass |
| |
| def wait_for_passkey(self, cert_address): |
| """ |
| Respond to the UI event |
| """ |
| passkey = -1 |
| |
| def get_passkey(event): |
| if event.message_type == UiMsgType.DISPLAY_PASSKEY: |
| nonlocal passkey |
| passkey = event.numeric_value |
| return True |
| return False |
| |
| logging.info("DUT: Waiting for expected UI event") |
| assertThat(self._ui_event_stream).emits(get_passkey) |
| return passkey |
| |
| def input_pin(self, cert_address, pin): |
| """ |
| Respond to the UI event |
| """ |
| logging.info("DUT: Inputting pin code: %s" % str(pin)) |
| self.on_user_input( |
| cert_address=cert_address, reply_boolean=True, expected_ui_event=UiMsgType.DISPLAY_PIN_ENTRY, pin=pin) |
| |
| def on_user_input(self, cert_address, reply_boolean, expected_ui_event, pin=[]): |
| """ |
| Respond to the UI event |
| """ |
| if expected_ui_event is None: |
| return |
| |
| ui_id = -1 |
| |
| def get_unique_id(event): |
| if event.message_type == expected_ui_event: |
| nonlocal ui_id |
| ui_id = event.unique_id |
| return True |
| return False |
| |
| logging.info("DUT: Waiting for expected UI event") |
| assertThat(self._ui_event_stream).emits(get_unique_id) |
| callback_type = UiCallbackType.YES_NO if len(pin) == 0 else UiCallbackType.PIN |
| self.__send_ui_callback(cert_address, callback_type, reply_boolean, ui_id, pin) |
| |
| def get_address(self): |
| return self._device.address |
| |
| def wait_for_bond_event(self, expected_bond_event): |
| """ |
| A bond event will be triggered once the bond process |
| is complete. For the DUT we need to wait for it, |
| for Cert it isn't needed. |
| """ |
| logging.info("DUT: Waiting for Bond Event: %s " % expected_bond_event) |
| assertThat(self._bond_event_stream).emits( |
| lambda event: event.message_type == expected_bond_event or logging.info("DUT: Actual Bond Event: %s" % event.message_type) |
| ) |
| |
| def wait_for_enforce_security_event(self, expected_enforce_security_event): |
| """ |
| We expect a 'True' or 'False' from the enforce security call |
| |
| This interface will allow the caller to wait for a callback |
| result from enforcing security policy over the facade. |
| """ |
| logging.info("DUT: Waiting for enforce security event") |
| assertThat(self._enforce_security_policy_stream).emits( |
| lambda event: event.result == expected_enforce_security_event or logging.info(event.result)) |
| |
| def wait_for_disconnect_event(self): |
| """ |
| The Address is expected to be returned |
| """ |
| logging.info("DUT: Waiting for Disconnect Event") |
| assertThat(self._disconnect_event_stream).emits(lambda event: logging.info("event: %s" % event.address) or True) |
| |
| def enforce_security_policy(self, address, type, policy): |
| """ |
| Call to enforce classic security policy |
| """ |
| self._device.security.EnforceSecurityPolicy( |
| SecurityPolicyMessage( |
| address=common.BluetoothAddressWithType(address=common.BluetoothAddress(address=address), type=type), |
| policy=policy)) |
| |
| def get_oob_data_from_controller(self, oob_data_present): |
| self._device.security.GetOutOfBandData(empty_proto.Empty()) |
| oob_data = [] |
| |
| def get_oob_data(event): |
| nonlocal oob_data |
| oob_data = [ |
| list(event.p192_data.confirmation_value), |
| list(event.p192_data.random_value), [0 for i in range(0, 16)], [0 for i in range(0, 16)] |
| ] |
| return True |
| |
| assertThat(self._oob_data_event_stream).emits(get_oob_data) |
| return oob_data |
| |
| def close(self): |
| safeClose(self._ui_event_stream) |
| safeClose(self._bond_event_stream) |
| safeClose(self._enforce_security_policy_stream) |
| safeClose(self._disconnect_event_stream) |
| safeClose(self._oob_data_event_stream) |
