1# Copyright 2024 Google LLC
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7#     https://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14
15import logging
16
17import grpc
18from bumble.device import Connection as BumbleConnection
19from bumble.device import Device
20from bumble.pairing import PairingConfig
21from bumble.pairing import PairingDelegate as BasePairingDelegate
22from bumble.pandora import Config, utils
23from bumble.pandora.security import PairingDelegate, SecurityService
24from google.protobuf.empty_pb2 import Empty
25from pandora.host_pb2 import PUBLIC
26from pandora_experimental.bumble_config_grpc_aio import BumbleConfigServicer
27from pandora_experimental.bumble_config_pb2 import (KeyDistribution, OverrideRequest)
28
29
30class BumbleConfigService(BumbleConfigServicer):
31    device: Device
32
33    def __init__(self, device: Device, server_config: Config) -> None:
34        self.log = utils.BumbleServerLoggerAdapter(logging.getLogger(), {
35            "service_name": "BumbleConfig",
36            "device": device,
37        })
38        self.device = device
39        self.server_config = server_config
40
41    @utils.rpc
42    async def Override(self, request: OverrideRequest, context: grpc.ServicerContext) -> Empty:
43
44        def parseProtoKeyDistribution(key: KeyDistribution,) -> BasePairingDelegate.KeyDistribution:
45            return [
46                BasePairingDelegate.KeyDistribution.DISTRIBUTE_ENCRYPTION_KEY,
47                BasePairingDelegate.KeyDistribution.DISTRIBUTE_IDENTITY_KEY,
48                BasePairingDelegate.KeyDistribution.DISTRIBUTE_SIGNING_KEY,
49                BasePairingDelegate.KeyDistribution.DISTRIBUTE_LINK_KEY,
50            ][key]  # type: ignore
51
52        def pairing_config_factory(connection: BumbleConnection) -> PairingConfig:
53            pairing_delegate = PairingDelegate(
54                connection=connection,
55                service=SecurityService(self.device, self.server_config),
56                io_capability=BasePairingDelegate.IoCapability(request.io_capability),
57                local_initiator_key_distribution=parseProtoKeyDistribution(request.initiator_key_distribution),
58                local_responder_key_distribution=parseProtoKeyDistribution(request.responder_key_distribution),
59            )
60
61            pc_req = request.pairing_config
62            pairing_config = PairingConfig(
63                sc=pc_req.sc,
64                mitm=pc_req.mitm,
65                bonding=pc_req.bonding,
66                identity_address_type=PairingConfig.AddressType.PUBLIC
67                if pc_req.identity_address_type == PUBLIC else PairingConfig.AddressType.RANDOM,
68                delegate=pairing_delegate,
69            )
70            self.log.debug(f"Override: {pairing_config}")
71
72            return pairing_config
73
74        self.device.pairing_config_factory = pairing_config_factory
75
76        return Empty()
77