Merge "Start CertL2cap, an abstraction to make L2cap testing simpler" am: 20f228891f (38d72aa1) · Commits · e / os / android_packages_modules_Bluetooth

system/gd/cert/py_l2cap.py

+14 −2

Original line number	Diff line number	Diff line
		@@ -18,16 +18,28 @@ from l2cap.classic import facade_pb2 as l2cap_facade_pb2
		from bluetooth_packets_python3 import l2cap_packets


		class PyL2capChannel(object):

		def __init__(self, device, psm):
		self._device = device
		self._psm = psm

		def send(self, payload):
		self._device.l2cap.SendDynamicChannelPacket(
		l2cap_facade_pb2.DynamicChannelPacket(psm=0x33, payload=payload))


		class PyL2cap(object):

		def __init__(self, device):
		self.device = device
		self._device = device

		def open_channel(self,
		psm=0x33,
		mode=l2cap_facade_pb2.RetransmissionFlowControlMode.BASIC):

		# todo, I don't understand what SetDynamicChannel means?
		self.device.l2cap.SetDynamicChannel(
		self._device.l2cap.SetDynamicChannel(
		l2cap_facade_pb2.SetEnableDynamicChannelRequest(
		psm=psm, retransmission_mode=mode))
		return PyL2capChannel(self._device, psm)

system/gd/l2cap/classic/cert/cert_l2cap.py

0 → 100644

+348 −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.

		from cert.closable import Closable
		from cert.closable import safeClose
		from cert.py_acl_manager import PyAclManager
		from cert.truth import assertThat
		import bluetooth_packets_python3 as bt_packets
		from bluetooth_packets_python3 import l2cap_packets
		from bluetooth_packets_python3.l2cap_packets import CommandCode
		from cert.event_stream import FilteringEventStream
		from cert.event_stream import IEventStream
		from cert.matchers import L2capMatchers


		class CertL2capChannel(IEventStream):

		def __init__(self, device, scid, acl_stream):
		self._device = device
		self._scid = scid
		self._our_acl_view = acl_stream

		def get_event_queue(self):
		return self._our_acl_view.get_event_queue()


		class CertL2cap(Closable):

		def __init__(self, device):
		self._device = device
		self._acl_manager = PyAclManager(device)
		self._acl = None

		self.control_table = {
		CommandCode.CONNECTION_REQUEST:
		self._on_connection_request_default,
		CommandCode.CONNECTION_RESPONSE:
		self._on_connection_response_default,
		CommandCode.CONFIGURATION_REQUEST:
		self._on_configuration_request_default,
		CommandCode.CONFIGURATION_RESPONSE:
		self._on_configuration_response_default,
		CommandCode.DISCONNECTION_REQUEST:
		self._on_disconnection_request_default,
		CommandCode.DISCONNECTION_RESPONSE:
		self._on_disconnection_response_default,
		CommandCode.INFORMATION_REQUEST:
		self._on_information_request_default,
		CommandCode.INFORMATION_RESPONSE:
		self._on_information_response_default
		}

		self.scid_to_dcid = {}
		self.ertm_tx_window_size = 10
		self.ertm_max_transmit = 20

		def close(self):
		self._acl_manager.close()
		safeClose(self._acl)

		def connect_acl(self, remote_addr):
		self._acl = self._acl_manager.initiate_connection(remote_addr)
		self._acl.wait_for_connection_complete()
		self.get_acl_stream().register_callback(self._handle_control_packet)

		def open_channel(self, signal_id, psm, scid):
		# what is the 1 here for?
		open_channel = l2cap_packets.BasicFrameBuilder(
		1, l2cap_packets.ConnectionRequestBuilder(signal_id, psm, scid))
		self.send_acl(open_channel)

		assertThat(self._acl).emits(L2capMatchers.ConnectionResponse(scid))
		return CertL2capChannel(self._device, scid, self.get_acl_stream())

		# prefer to use channel abstraction instead, if at all possible
		def send_acl(self, packet):
		self._acl.send(packet.Serialize())

		# temporary until clients migrated
		def get_acl_stream(self):
		return self._acl_manager.get_acl_stream()

		# temporary until clients migrated
		def get_acl(self):
		return self._acl

		# temporary until clients migrated
		def get_dcid(self, scid):
		return self.scid_to_dcid[scid]

		# more of a hack for the moment
		def turn_on_ertm(self, tx_window_size=10, max_transmit=20):
		self.ertm_tx_window_size = tx_window_size
		self.ertm_max_transmit = max_transmit
		self.control_table[
		CommandCode.
		CONNECTION_RESPONSE] = self._on_connection_response_use_ertm

		# more of a hack for the moment
		def turn_on_ertm_and_fcs(self):
		self.control_table[
		CommandCode.
		CONNECTION_RESPONSE] = self._on_connection_response_use_ertm_and_fcs

		# more of a hack for the moment
		def ignore_config_and_connections(self):
		self.control_table[CommandCode.CONFIGURATION_REQUEST] = lambda _: True
		self.control_table[CommandCode.CONNECTION_RESPONSE] = lambda _: True

		# more of a hack for the moment
		def reply_with_unacceptable_parameters(self):
		self.control_table[
		CommandCode.
		CONFIGURATION_REQUEST] = self._on_configuration_request_unacceptable_parameters

		# more of a hack for the moment
		def reply_with_unknown_options_and_hint(self):
		self.control_table[
		CommandCode.
		CONNECTION_RESPONSE] = self._on_connection_response_configuration_request_with_unknown_options_and_hint

		def _on_connection_request_default(self, l2cap_control_view):
		connection_request_view = l2cap_packets.ConnectionRequestView(
		l2cap_control_view)
		sid = connection_request_view.GetIdentifier()
		cid = connection_request_view.GetSourceCid()

		self.scid_to_dcid[cid] = cid

		connection_response = l2cap_packets.ConnectionResponseBuilder(
		sid, cid, cid, l2cap_packets.ConnectionResponseResult.SUCCESS,
		l2cap_packets.ConnectionResponseStatus.
		NO_FURTHER_INFORMATION_AVAILABLE)
		connection_response_l2cap = l2cap_packets.BasicFrameBuilder(
		1, connection_response)
		self.send_acl(connection_response_l2cap)
		return True

		def _on_connection_response_default(self, l2cap_control_view):
		connection_response_view = l2cap_packets.ConnectionResponseView(
		l2cap_control_view)
		sid = connection_response_view.GetIdentifier()
		scid = connection_response_view.GetSourceCid()
		dcid = connection_response_view.GetDestinationCid()
		self.scid_to_dcid[scid] = dcid

		config_request = l2cap_packets.ConfigurationRequestBuilder(
		sid + 1, dcid, l2cap_packets.Continuation.END, [])
		config_request_l2cap = l2cap_packets.BasicFrameBuilder(
		1, config_request)
		self.send_acl(config_request_l2cap)
		return True

		def _on_connection_response_use_ertm(self, l2cap_control_view):
		connection_response_view = l2cap_packets.ConnectionResponseView(
		l2cap_control_view)
		sid = connection_response_view.GetIdentifier()
		scid = connection_response_view.GetSourceCid()
		dcid = connection_response_view.GetDestinationCid()
		self.scid_to_dcid[scid] = dcid

		# FIXME: This doesn't work!
		ertm_option = l2cap_packets.RetransmissionAndFlowControlConfigurationOption(
		)
		ertm_option.mode = l2cap_packets.RetransmissionAndFlowControlModeOption.L2CAP_BASIC
		ertm_option.tx_window_size = self.ertm_tx_window_size
		ertm_option.max_transmit = self.ertm_max_transmit
		ertm_option.retransmission_time_out = 2000
		ertm_option.monitor_time_out = 12000
		ertm_option.maximum_pdu_size = 1010

		options = [ertm_option]

		config_request = l2cap_packets.ConfigurationRequestBuilder(
		sid + 1, dcid, l2cap_packets.Continuation.END, options)

		config_request_l2cap = l2cap_packets.BasicFrameBuilder(
		1, config_request)

		self.send_acl(config_request_l2cap)
		return True

		def _on_connection_response_use_ertm_and_fcs(self, l2cap_control_view):
		connection_response_view = l2cap_packets.ConnectionResponseView(
		l2cap_control_view)
		sid = connection_response_view.GetIdentifier()
		scid = connection_response_view.GetSourceCid()
		dcid = connection_response_view.GetDestinationCid()
		self.scid_to_dcid[scid] = dcid

		# FIXME: This doesn't work!
		ertm_option = l2cap_packets.RetransmissionAndFlowControlConfigurationOption(
		)
		ertm_option.mode = l2cap_packets.RetransmissionAndFlowControlModeOption.L2CAP_BASIC
		ertm_option.tx_window_size = self.ertm_tx_window_size
		ertm_option.max_transmit = self.ertm_max_transmit
		ertm_option.retransmission_time_out = 2000
		ertm_option.monitor_time_out = 12000
		ertm_option.maximum_pdu_size = 1010

		fcs_option = l2cap_packets.FrameCheckSequenceOption()
		fcs_option.fcs_type = l2cap_packets.FcsType.DEFAULT

		options = [ertm_option, fcs_option]

		config_request = l2cap_packets.ConfigurationRequestBuilder(
		sid + 1, dcid, l2cap_packets.Continuation.END, options)

		config_request_l2cap = l2cap_packets.BasicFrameBuilder(
		1, config_request)

		self.send_acl(config_request_l2cap)
		return True

		def _on_connection_response_configuration_request_with_unknown_options_and_hint(
		self, l2cap_control_view):
		connection_response_view = l2cap_packets.ConnectionResponseView(
		l2cap_control_view)
		sid = connection_response_view.GetIdentifier()
		scid = connection_response_view.GetSourceCid()
		dcid = connection_response_view.GetDestinationCid()
		self.scid_to_dcid[scid] = dcid

		mtu_opt = l2cap_packets.MtuConfigurationOption()
		mtu_opt.mtu = 0x1234
		mtu_opt.is_hint = l2cap_packets.ConfigurationOptionIsHint.OPTION_IS_A_HINT

		options = [mtu_opt]
		config_request = l2cap_packets.ConfigurationRequestBuilder(
		sid + 1, dcid, l2cap_packets.Continuation.END, options)
		config_request_l2cap = l2cap_packets.BasicFrameBuilder(
		1, config_request)

		byte_array = bytearray(config_request_l2cap.Serialize())
		## Modify configuration option type to be a unknown
		byte_array[12] \|= 0x7f
		self._acl.send(bytes(byte_array))
		return True

		def _on_configuration_request_default(self, l2cap_control_view):
		configuration_request = l2cap_packets.ConfigurationRequestView(
		l2cap_control_view)
		sid = configuration_request.GetIdentifier()
		dcid = configuration_request.GetDestinationCid()
		config_response = l2cap_packets.ConfigurationResponseBuilder(
		sid, self.scid_to_dcid.get(dcid, 0), l2cap_packets.Continuation.END,
		l2cap_packets.ConfigurationResponseResult.SUCCESS, [])
		config_response_l2cap = l2cap_packets.BasicFrameBuilder(
		1, config_response)
		self.send_acl(config_response_l2cap)

		def _on_configuration_request_unacceptable_parameters(
		self, l2cap_control_view):
		configuration_request = l2cap_packets.ConfigurationRequestView(
		l2cap_control_view)
		sid = configuration_request.GetIdentifier()
		dcid = configuration_request.GetDestinationCid()

		mtu_opt = l2cap_packets.MtuConfigurationOption()
		mtu_opt.mtu = 123
		fcs_opt = l2cap_packets.FrameCheckSequenceOption()
		fcs_opt.fcs_type = l2cap_packets.FcsType.DEFAULT
		rfc_opt = l2cap_packets.RetransmissionAndFlowControlConfigurationOption(
		)
		rfc_opt.mode = l2cap_packets.RetransmissionAndFlowControlModeOption.L2CAP_BASIC

		config_response = l2cap_packets.ConfigurationResponseBuilder(
		sid, self.scid_to_dcid.get(dcid, 0), l2cap_packets.Continuation.END,
		l2cap_packets.ConfigurationResponseResult.UNACCEPTABLE_PARAMETERS,
		[mtu_opt, fcs_opt, rfc_opt])
		config_response_l2cap = l2cap_packets.BasicFrameBuilder(
		1, config_response)
		self.send_acl(config_response_l2cap)

		def _on_configuration_response_default(self, l2cap_control_view):
		configuration_response = l2cap_packets.ConfigurationResponseView(
		l2cap_control_view)
		sid = configuration_response.GetIdentifier()

		def _on_disconnection_request_default(self, l2cap_control_view):
		disconnection_request = l2cap_packets.DisconnectionRequestView(
		l2cap_control_view)
		sid = disconnection_request.GetIdentifier()
		scid = disconnection_request.GetSourceCid()
		dcid = disconnection_request.GetDestinationCid()
		disconnection_response = l2cap_packets.DisconnectionResponseBuilder(
		sid, dcid, scid)
		disconnection_response_l2cap = l2cap_packets.BasicFrameBuilder(
		1, disconnection_response)
		self.send_acl(disconnection_response_l2cap)

		def _on_disconnection_response_default(self, l2cap_control_view):
		disconnection_response = l2cap_packets.DisconnectionResponseView(
		l2cap_control_view)

		def _on_information_request_default(self, l2cap_control_view):
		information_request = l2cap_packets.InformationRequestView(
		l2cap_control_view)
		sid = information_request.GetIdentifier()
		information_type = information_request.GetInfoType()
		if information_type == l2cap_packets.InformationRequestInfoType.CONNECTIONLESS_MTU:
		response = l2cap_packets.InformationResponseConnectionlessMtuBuilder(
		sid, l2cap_packets.InformationRequestResult.SUCCESS, 100)
		response_l2cap = l2cap_packets.BasicFrameBuilder(1, response)
		self.send_acl(response_l2cap)
		return
		if information_type == l2cap_packets.InformationRequestInfoType.EXTENDED_FEATURES_SUPPORTED:
		response = l2cap_packets.InformationResponseExtendedFeaturesBuilder(
		sid, l2cap_packets.InformationRequestResult.SUCCESS, 0, 0, 0, 1,
		0, 1, 0, 0, 0, 0)
		response_l2cap = l2cap_packets.BasicFrameBuilder(1, response)
		self.send_acl(response_l2cap)
		return
		if information_type == l2cap_packets.InformationRequestInfoType.FIXED_CHANNELS_SUPPORTED:
		response = l2cap_packets.InformationResponseFixedChannelsBuilder(
		sid, l2cap_packets.InformationRequestResult.SUCCESS, 2)
		response_l2cap = l2cap_packets.BasicFrameBuilder(1, response)
		self.send_acl(response_l2cap)
		return

		def _on_information_response_default(self, l2cap_control_view):
		information_response = l2cap_packets.InformationResponseView(
		l2cap_control_view)

		def _handle_control_packet(self, l2cap_packet):
		packet_bytes = l2cap_packet.payload
		l2cap_view = l2cap_packets.BasicFrameView(
		bt_packets.PacketViewLittleEndian(list(packet_bytes)))
		if l2cap_view.GetChannelId() != 1:
		return
		l2cap_control_view = l2cap_packets.ControlView(l2cap_view.GetPayload())
		fn = self.control_table.get(l2cap_control_view.GetCode())
		if fn is not None:
		fn(l2cap_control_view)
		return

system/gd/l2cap/classic/cert/l2cap_test.py

+69 −367

File changed.

Preview size limit exceeded, changes collapsed.