Coverage for rdak/connection.py: 100%

1# SPDX-License-Identifier: GPL-2.0-or-later 

2# © 2026, Ansgar 🙀 <ansgar@debian.org> 

3 

4""" 

5Build authenticated gRPC channels to a dak RPC server. 

6""" 

7 

8import grpc 

9 

10from rdak.config import InstanceConfig 

11 

12 

13def tls_options(instance: InstanceConfig) -> list[tuple[str, str]]: 

14 """Channel options for TLS, applying the server-name override if set.""" 

15 if instance.server_name is None: 

16 return [] 

17 return [ 

18 ("grpc.ssl_target_name_override", instance.server_name), 

19 ("grpc.default_authority", instance.server_name), 

20 ] 

21 

22 

23def channel_credentials(instance: InstanceConfig) -> grpc.ChannelCredentials: 

24 """Transport credentials: TLS when enabled, else local (unix socket).""" 

25 if instance.use_tls: 

26 return grpc.ssl_channel_credentials( 

27 root_certificates=instance.ca_certificates() 

28 ) 

29 return grpc.local_channel_credentials(grpc.LocalConnectionType.UDS) 

30 

31 

32def connect(instance: InstanceConfig) -> grpc.Channel: 

33 """Create a ready-to-use channel with the bearer token attached.""" 

34 credentials = grpc.composite_channel_credentials( 

35 channel_credentials(instance), 

36 grpc.access_token_call_credentials(instance.get_token()), 

37 ) 

38 return grpc.secure_channel( 

39 instance.address, credentials, options=tls_options(instance) 

40 )