Cert: Add CertSecurity (5ac09664) · Commits · e / os / android_packages_modules_Bluetooth

system/gd/security/cert/cert_security.py

0 → 100644

+239 −0

Original line number	Diff line number	Diff line
		#!/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 safeClose
		from cert.event_stream import EventStream
		from cert.py_hci import PyHci
		from cert.py_security import PySecurity
		from datetime import datetime
		from google.protobuf import empty_pb2 as empty_proto
		from hci.facade import facade_pb2 as hci_facade
		from l2cap.classic import facade_pb2 as l2cap_facade
		from security.facade_pb2 import IoCapabilities
		from security.facade_pb2 import AuthenticationRequirements
		from security.facade_pb2 import OobDataPresent


		class CertSecurity(PySecurity):
		"""
		Contain all of the certification stack logic for sending and receiving
		HCI commands following the Classic Pairing flows.
		"""
		_io_cap_lookup = {
		IoCapabilities.DISPLAY_ONLY:
		hci_packets.IoCapability.DISPLAY_ONLY,
		IoCapabilities.DISPLAY_YES_NO_IO_CAP:
		hci_packets.IoCapability.DISPLAY_YES_NO,
		IoCapabilities.KEYBOARD_ONLY:
		hci_packets.IoCapability.KEYBOARD_ONLY,
		IoCapabilities.NO_INPUT_NO_OUTPUT:
		hci_packets.IoCapability.NO_INPUT_NO_OUTPUT,
		}

		_auth_req_lookup = {
		AuthenticationRequirements.NO_BONDING:
		hci_packets.AuthenticationRequirements.NO_BONDING,
		AuthenticationRequirements.NO_BONDING_MITM_PROTECTION:
		hci_packets.AuthenticationRequirements.NO_BONDING_MITM_PROTECTION,
		AuthenticationRequirements.DEDICATED_BONDING:
		hci_packets.AuthenticationRequirements.DEDICATED_BONDING,
		AuthenticationRequirements.DEDICATED_BONDING_MITM_PROTECTION:
		hci_packets.AuthenticationRequirements.
		DEDICATED_BONDING_MITM_PROTECTION,
		AuthenticationRequirements.GENERAL_BONDING:
		hci_packets.AuthenticationRequirements.GENERAL_BONDING,
		AuthenticationRequirements.GENERAL_BONDING_MITM_PROTECTION:
		hci_packets.AuthenticationRequirements.GENERAL_BONDING_MITM_PROTECTION,
		}

		_oob_present_lookup = {
		OobDataPresent.NOT_PRESENT:
		hci_packets.OobDataPresent.NOT_PRESENT,
		OobDataPresent.P192_PRESENT:
		hci_packets.OobDataPresent.P_192_PRESENT,
		OobDataPresent.P256_PRESENT:
		hci_packets.OobDataPresent.P_256_PRESENT,
		OobDataPresent.P192_AND_256_PRESENT:
		hci_packets.OobDataPresent.P_192_AND_256_PRESENT,
		}

		_hci_event_stream = None
		_io_caps = hci_packets.IoCapability.DISPLAY_ONLY
		_oob_data = hci_packets.OobDataPresent.NOT_PRESENT
		_auth_reqs = hci_packets.AuthenticationRequirements.DEDICATED_BONDING_MITM_PROTECTION

		_hci = None

		def _enqueue_hci_command(self, command, expect_complete):
		if (expect_complete):
		self._hci.send_command_with_complete(command)
		else:
		self._hci.send_command_with_status(command)

		def __init__(self, device):
		"""
		Don't call super b/c the gRPC stream setup will crash test
		"""
		logging.info("Cert: Init")
		self._device = device
		self._device.wait_channel_ready()
		self._hci = PyHci(device)
		self._hci.register_for_events(
		hci_packets.EventCode.LINK_KEY_REQUEST,
		hci_packets.EventCode.IO_CAPABILITY_REQUEST,
		hci_packets.EventCode.IO_CAPABILITY_RESPONSE,
		hci_packets.EventCode.USER_PASSKEY_NOTIFICATION,
		hci_packets.EventCode.USER_PASSKEY_REQUEST,
		hci_packets.EventCode.USER_CONFIRMATION_REQUEST,
		hci_packets.EventCode.REMOTE_HOST_SUPPORTED_FEATURES_NOTIFICATION,
		hci_packets.EventCode.LINK_KEY_NOTIFICATION,
		hci_packets.EventCode.SIMPLE_PAIRING_COMPLETE)
		self._hci_event_stream = self._hci.get_event_stream()

		def create_bond(self, address, type):
		"""
		Creates a bond from the cert perspective
		"""
		logging.info("Cert: Creating bond to '%s' from '%s'" %
		(str(address), str(self._device.address)))
		# TODO(optedoblivion): Trigger connection to Send AuthenticationRequested

		def remove_bond(self, address):
		"""
		We store the link key locally in the test and pretend
		So to remove_bond we need to Remove the "stored" data
		"""
		pass

		def set_io_capabilities(self, io_capabilities):
		"""
		Set the IO Capabilities used for the cert
		"""
		logging.info(
		"Cert: setting IO Capabilities data to '%s'" % io_capabilities)
		self._io_caps = self._io_cap_lookup.get(
		io_capabilities, hci_packets.IoCapability.DISPLAY_YES_NO)

		def set_authentication_requirements(self, auth_reqs):
		"""
		Establish authentication requirements for the stack
		"""
		logging.info("Cert: setting Authentication Requirements data to '%s'" %
		auth_reqs)
		self._auth_reqs = self._auth_req_lookup.get(
		auth_reqs, hci_packets.AuthenticationRequirements.GENERAL_BONDING)

		def set_oob_data(self, data):
		"""
		Set the Out-of-band data for SSP pairing
		"""
		logging.info("Cert: setting OOB data present to '%s'" % data)
		self._oob_data = self._oob_present_lookup.get(
		data, hci_packets.OobDataPresent.NOT_PRESENT)

		def send_ui_callback(self, address, callback_type, b, uid):
		"""
		Pretend to answer the pairing dailog as a user
		"""
		logging.info(
		"Cert: Send user input callback uid:%d; response: %s" % (uid, b))
		# TODO(optedoblivion): Make callback and set it to the module

		def enable_secure_simple_pairing(self):
		"""
		This is called when you want to enable SSP for testing
		"""
		logging.info("Cert: Sending WRITE_SIMPLE_PAIRING_MODE [True]")
		self._enqueue_hci_command(
		hci_packets.WriteSimplePairingModeBuilder(
		hci_packets.Enable.ENABLED), True)
		logging.info("Cert: Waiting for controller response")
		self._hci_event_stream.assert_event_occurs(
		lambda msg: b'\x0e\x04\x01\x56\x0c' in msg.event)

		def accept_pairing(self, dut_address, reply_boolean):
		"""
		Here we handle the pairing events at the HCI level
		"""
		logging.info("Cert: Waiting for LINK_KEY_REQUEST")
		self._hci_event_stream.assert_event_occurs(
		lambda event: logging.debug(event.event) or hci_packets.EventCode.LINK_KEY_REQUEST in event.event
		)
		logging.info("Cert: Sending LINK_KEY_REQUEST_NEGATIVE_REPLY")
		self._enqueue_hci_command(
		hci_packets.LinkKeyRequestNegativeReplyBuilder(
		dut_address.decode('utf8')), True)
		logging.info("Cert: Waiting for IO_CAPABILITY_REQUEST")
		self._hci_event_stream.assert_event_occurs(
		lambda event: logging.debug(event.event) or hci_packets.EventCode.IO_CAPABILITY_REQUEST in event.event
		)
		logging.info("Cert: Sending IO_CAPABILITY_REQUEST_REPLY")

		self._enqueue_hci_command(
		hci_packets.IoCapabilityRequestReplyBuilder(
		dut_address.decode('utf8'), self._io_caps, self._oob_data,
		self._auth_reqs), True)

		logging.info("Cert: Waiting for USER_CONFIRMATION_REQUEST")
		self._hci_event_stream.assert_event_occurs(
		lambda event: logging.debug(event.event) or hci_packets.EventCode.USER_CONFIRMATION_REQUEST in event.event
		)
		logging.info(
		"Cert: Sending Simulated User Response '%s'" % reply_boolean)
		if reply_boolean:
		logging.info("Cert: Sending USER_CONFIRMATION_REQUEST_REPLY")
		self._enqueue_hci_command(
		hci_packets.UserConfirmationRequestReplyBuilder(
		dut_address.decode('utf8')), True)
		logging.info("Cert: Waiting for LINK_KEY_NOTIFICATION")
		self._hci_event_stream.assert_event_occurs(
		lambda event: logging.debug(event.event) or hci_packets.EventCode.LINK_KEY_NOTIFICATION in event.event
		)
		logging.info("Cert: Waiting for SIMPLE_PAIRING_COMPLETE")
		self._hci_event_stream.assert_event_occurs(
		lambda event: logging.debug(event.event) or hci_packets.EventCode.SIMPLE_PAIRING_COMPLETE in event.event
		)
		else:
		logging.info(
		"Cert: Sending USER_CONFIRMATION_REQUEST_NEGATIVE_REPLY")
		self._enqueue_hci_command(
		hci_packets.UserConfirmationRequestNegativeReplyBuilder(
		dut_address.decode('utf8')), True)
		logging.info("Cert: Waiting for SIMPLE_PAIRING_COMPLETE")
		self._hci_event_stream.assert_event_occurs(
		lambda event: logging.debug(event.event) or hci_packets.EventCode.SIMPLE_PAIRING_COMPLETE in event.event
		)

		def on_user_input(self, dut_address, reply_boolean, expected_ui_event):
		"""
		Cert doesn't need the test to respond to the ui event
		Cert responds in accept pairing
		"""
		pass

		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.
		"""
		pass

		def close(self):
		safeClose(self._hci)