floss: Support more pairing methods in OnPairing (59eb7b15) · Commits · e / os / android_packages_modules_Bluetooth

floss/pandora/floss/adapter_client.py

+73 −1

Original line number	Diff line number	Diff line
		@@ -68,6 +68,30 @@ class BluetoothCallbacks:
		"""
		pass

		def on_pin_request(self, remote_device, cod, min_16_digit):
		"""When there is a pin request to display the event to client.

		Args:
		remote_device:
		Remote device that is being paired.
		cod:
		Class of device as described in HCI spec.
		min_16_digit:
		True if the pin is 16 digit, False otherwise.
		"""
		pass

		def on_pin_display(self, remote_device, pincode):
		"""When there is a auto-gen pin to display the event to client.

		Args:
		remote_device:
		Remote device that is being paired.
		pincode:
		PIN code to display.
		"""
		pass

		def on_bond_state_changed(self, status, address, state):
		"""Bonding/Pairing state has changed for a device.

		@@ -155,6 +179,15 @@ class FlossAdapterClient(BluetoothCallbacks, BluetoothConnectionCallbacks):
		<arg type="u" name="variant" direction="in" />
		<arg type="u" name="passkey" direction="in" />
		</method>
		<method name="OnPinRequest">
		<arg type="a{sv}" name="remote_device_dbus" direction="in" />
		<arg type="u" name="cod" direction="in" />
		<arg type="b" name="min_16_digit" direction="in" />
		</method>
		<method name="OnPinDisplay">
		<arg type="a{sv}" name="remote_device_dbus" direction="in" />
		<arg type="s" name="pincode" direction="in" />
		</method>
		<method name="OnBondStateChanged">
		<arg type="u" name="status" direction="in" />
		<arg type="s" name="address" direction="in" />
		@@ -192,12 +225,32 @@ class FlossAdapterClient(BluetoothCallbacks, BluetoothConnectionCallbacks):
		"""Handle pairing/bonding request to agent."""
		parsed, remote_device = FlossAdapterClient.parse_dbus_device(remote_device_dbus)
		if not parsed:
		logging.debug('OnSspRequest parse error: {}'.format(remote_device_dbus))
		logging.error('OnSspRequest parse error: {}'.format(remote_device_dbus))
		return

		for observer in self.observers.values():
		observer.on_ssp_request(remote_device, class_of_device, variant, passkey)

		def OnPinRequest(self, remote_device_dbus, cod, min_16_digit):
		"""Handle PIN request callback."""
		parsed, remote_device = FlossAdapterClient.parse_dbus_device(remote_device_dbus)
		if not parsed:
		logging.error('OnPinRequest parse error: {}'.format(remote_device_dbus))
		return

		for observer in self.observers.values():
		observer.on_pin_request(remote_device, cod, min_16_digit)

		def OnPinDisplay(self, remote_device_dbus, pincode):
		"""Handle PIN display callback."""
		parsed, remote_device = FlossAdapterClient.parse_dbus_device(remote_device_dbus)
		if not parsed:
		logging.error('OnPinDisplay parse error: {}'.format(remote_device_dbus))
		return

		for observer in self.observers.values():
		observer.on_pin_display(remote_device, pincode)

		def OnBondStateChanged(self, status, address, state):
		"""Handle bond state changed callbacks."""
		for observer in self.observers.values():
		@@ -725,6 +778,25 @@ class FlossAdapterClient(BluetoothCallbacks, BluetoothConnectionCallbacks):

		return True

		@utils.glib_call(False)
		def set_pin(self, address, accept, pin_code):
		"""Set pin on bonding device.

		Args:
		address: Device address to reply.
		accept: True to accept the pin request, False to reject the pin request.
		pin_code: PIN code to reply. The PIN code is a list of up to 16
		integers.
		"""
		if address not in self.known_devices:
		logging.debug('[%s] Unknown device in set_pin', address)
		return False

		device = self.known_devices[address]
		remote_device = self._make_dbus_device(address, device['name'])

		return self.proxy().SetPin(remote_device, accept, pin_code)

		@utils.glib_call(False)
		def set_pairing_confirmation(self, address, accept):
		"""Confirm that a pairing should be completed on a bonding device."""

floss/pandora/floss/floss_enums.py

+9 −2

Original line number	Diff line number	Diff line
		@@ -147,15 +147,22 @@ class Transport(enum.IntEnum):
		AUTO = 0
		BREDR = 1
		LE = 2
		DUAL = 3


		class SspVariant(enum.IntEnum):
		"""Bluetooth SSP variant type."""
		class PairingVariant(enum.IntEnum):
		"""Bluetooth pairing variant type."""
		# SSP variants.
		PASSKEY_CONFIRMATION = 0
		PASSKEY_ENTRY = 1
		CONSENT = 2
		PASSKEY_NOTIFICATION = 3

		# Legacy pairing variants.
		PIN_ENTRY = 4
		PIN_16_DIGITS_ENTRY = 5
		PIN_NOTIFICATION = 6


		class BleAddressType(enum.IntEnum):
		BLE_ADDR_PUBLIC = 0x00

floss/pandora/server/bluetooth.py

+6 −4

Original line number	Diff line number	Diff line
		@@ -54,10 +54,6 @@ class Bluetooth(object):
		# self state
		self.is_clean = False

		# GRPC server state
		self.pairing_events: asyncio.Queue = None
		self.pairing_answers = None

		# DBUS clients
		self.manager_client = manager_client.FlossManagerClient(self.bus)
		self.adapter_client = adapter_client.FlossAdapterClient(self.bus, self.DEFAULT_ADAPTER)
		@@ -181,6 +177,9 @@ class Bluetooth(object):
		def get_address(self):
		return self.adapter_client.get_address()

		def get_remote_type(self):
		return self.adapter_client.get_remote_property('Type')

		def is_connected(self, address):
		return self.adapter_client.is_connected(address)

		@@ -205,6 +204,9 @@ class Bluetooth(object):
		def forget_device(self, address):
		return self.adapter_client.forget_device(address)

		def set_pin(self, address, accept, pin_code):
		return self.adapter_client.set_pin(address, accept, pin_code)

		def set_pairing_confirmation(self, address, accept):
		return self.adapter_client.set_pairing_confirmation(address, accept)

floss/pandora/server/host.py

+1 −1

Original line number	Diff line number	Diff line
		@@ -134,7 +134,7 @@ class HostService(host_grpc_aio.HostServicer):
		if address != self.task['address']:
		return

		if variant in (floss_enums.SspVariant.CONSENT, floss_enums.SspVariant.PASSKEY_CONFIRMATION):
		if variant in (floss_enums.PairingVariant.CONSENT, floss_enums.PairingVariant.PASSKEY_CONFIRMATION):
		self.client.set_pairing_confirmation(address,
		True,
		method_callback=self.on_set_pairing_confirmation)

floss/pandora/server/security.py

+81 −22

Original line number	Diff line number	Diff line
		@@ -22,11 +22,9 @@ from floss.pandora.floss import adapter_client
		from floss.pandora.floss import floss_enums
		from floss.pandora.floss import utils
		from floss.pandora.server import bluetooth as bluetooth_module
		from google.protobuf import any_pb2
		from google.protobuf import empty_pb2
		from google.protobuf import wrappers_pb2
		import grpc
		from pandora import host_pb2
		from pandora import security_grpc_aio
		from pandora import security_pb2

		@@ -45,9 +43,13 @@ class SecurityService(security_grpc_aio.SecurityServicer):
		self.server = server
		self.bluetooth = bluetooth
		self.manually_confirm = False
		self.on_pairing_count = 0

		async def OnPairing(self, request: AsyncIterator[security_pb2.PairingEventAnswer],
		context: grpc.ServicerContext) -> AsyncGenerator[security_pb2.PairingEvent, None]:
		logging.info('OnPairing')
		on_pairing_id = self.on_pairing_count
		self.on_pairing_count = self.on_pairing_count + 1

		class PairingObserver(adapter_client.BluetoothCallbacks):
		"""Observer to observe all pairing events."""
		@@ -60,60 +62,117 @@ class SecurityService(security_grpc_aio.SecurityServicer):
		def on_ssp_request(self, remote_device, class_of_device, variant, passkey):
		address, name = remote_device

		result = (address, name, class_of_device, variant, passkey)
		result = (address, name, variant, passkey)
		asyncio.run_coroutine_threadsafe(self.task['pairing_events'].put(result), self.loop)

		@utils.glib_callback()
		def on_pin_request(self, remote_device, cod, min_16_digit):
		address, name = remote_device

		if min_16_digit:
		variant = floss_enums.PairingVariant.PIN_16_DIGITS_ENTRY
		else:
		variant = floss_enums.PairingVariant.PIN_ENTRY
		result = (address, name, variant, min_16_digit)
		asyncio.run_coroutine_threadsafe(self.task['pairing_events'].put(result), self.loop)

		@utils.glib_callback()
		def on_pin_display(self, remote_device, pincode):
		address, name = remote_device

		variant = floss_enums.PairingVariant.PIN_NOTIFICATION
		result = (address, name, variant, pincode)
		asyncio.run_coroutine_threadsafe(self.task['pairing_events'].put(result), self.loop)

		pairing_answers = request

		async def streaming_answers(self):
		while True:
		pairing_answer = await utils.anext(self.bluetooth.pairing_answers)
		answer = pairing_answer.WhichOneof('answer')
		address = utils.address_from(pairing_answer.event.connection.cookie.value)
		nonlocal pairing_answers
		nonlocal on_pairing_id

		logging.info('pairing_answer: %s address: %s', pairing_answer, address)
		logging.info('OnPairing[%s]: Wait for pairing answer...', on_pairing_id)
		pairing_answer = await utils.anext(pairing_answers)

		answer = pairing_answer.WhichOneof('answer')
		address = utils.address_from(pairing_answer.event.address)
		logging.info('OnPairing[%s]: Pairing answer: %s address: %s', on_pairing_id, answer, address)

		if answer == 'confirm':
		self.bluetooth.set_pairing_confirmation(address, True)
		elif answer == 'passkey':
		pass # TODO: b/289480188 - Supports this method.
		self.bluetooth.set_pin(address, True, list(str(answer.passkey).zfill(6).encode()))
		elif answer == 'pin':
		pass # TODO: b/289480188 - Supports this method.
		self.bluetooth.set_pin(address, True, list(answer.pin))

		observers = []
		try:
		self.manually_confirm = True

		self.bluetooth.pairing_events = asyncio.Queue()
		observer = PairingObserver(asyncio.get_running_loop(), {'pairing_events': self.bluetooth.pairing_events})
		pairing_events = asyncio.Queue()
		observer = PairingObserver(asyncio.get_running_loop(), {'pairing_events': pairing_events})
		name = utils.create_observer_name(observer)
		self.bluetooth.adapter_client.register_callback_observer(name, observer)
		observers.append((name, observer))

		self.bluetooth.pairing_answers = request
		streaming_answers_task = asyncio.create_task(streaming_answers(self))
		await streaming_answers_task

		while True:
		address, name, _, variant, passkey = await self.bluetooth.pairing_events.get()
		logging.info('OnPairing[%s]: Wait for pairing events...', on_pairing_id)
		address, name, variant, *variables = await pairing_events.get()
		logging.info('OnPairing[%s]: Pairing event: address: %s, name: %s, variant: %s, variables: %s',
		on_pairing_id, address, name, variant, variables)

		event = security_pb2.PairingEvent()
		event.connection.CopyFrom(host_pb2.Connection(cookie=any_pb2.Any(value=utils.address_to(address))))
		event.address = utils.address_to(address)

		if variant == floss_enums.SspVariant.PASSKEY_CONFIRMATION:
		# SSP
		if variant == floss_enums.PairingVariant.PASSKEY_CONFIRMATION:
		[passkey] = variables
		event.numeric_comparison = passkey
		elif variant == floss_enums.SspVariant.PASSKEY_ENTRY:
		elif variant == floss_enums.PairingVariant.PASSKEY_ENTRY:
		event.passkey_entry_request.CopyFrom(empty_pb2.Empty())
		elif variant == floss_enums.SspVariant.CONSENT:
		elif variant == floss_enums.PairingVariant.CONSENT:
		event.just_works.CopyFrom(empty_pb2.Empty())
		elif variant == floss_enums.SspVariant.PASSKEY_NOTIFICATION:
		event.passkey_entry_notification.CopyFrom(passkey)
		elif variant == floss_enums.PairingVariant.PASSKEY_NOTIFICATION:
		[passkey] = variables
		event.passkey_entry_notification = passkey
		# Legacy
		elif variant == floss_enums.PairingVariant.PIN_ENTRY:
		transport = self.bluetooth.get_remote_type(address)

		if transport == floss_enums.Transport.BREDR:
		event.pin_code_request.CopyFrom(empty_pb2.Empty())
		elif transport == floss_enums.Transport.LE:
		event.passkey_entry_request.CopyFrom(empty_pb2.Empty())
		else:
		logging.error('Cannot determine pairing variant from unknown transport.')
		continue
		elif variant == floss_enums.PairingVariant.PIN_16_DIGITS_ENTRY:
		event.pin_code_request.CopyFrom(empty_pb2.Empty())
		elif variant == floss_enums.PairingVarint.PIN_NOTIFICATION:
		transport = self.bluetooth.get_remote_type(address)
		[pincode] = variables

		if transport == floss_enums.Transport.BREDR:
		event.pin_code_notification = pincode.encode()
		elif transport == floss_enums.Transport.LE:
		event.passkey_entry_notification = int(pincode)
		else:
		logging.error('Cannot determine pairing variant from unknown transport.')
		continue
		else:
		logging.error('Unknown pairing variant: %s', variant)
		continue

		yield event
		finally:
		streaming_answers_task.cancel()
		for name, observer in observers:
		self.bluetooth.adapter_client.unregister_callback_observer(name, observer)

		self.bluetooth.pairing_events = None
		self.bluetooth.pairing_answers = None
		pairing_events = None
		pairing_answers = None

		async def Secure(self, request: security_pb2.SecureRequest,
		context: grpc.ServicerContext) -> security_pb2.SecureResponse: