Coverage for rdak/config.py: 92%

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

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

3 

4"""Configuration handling for the rdak client. 

5 

6The configuration is a TOML file describing one or more named *instances* 

7(e.g. the main archive and the security archive). Each instance fully 

8specifies how to reach a dak RPC server *and* which token to authenticate 

9with, so a token is never combined with a connection it was not bound to. 

10There are deliberately no command-line overrides for connection or token 

11settings. 

12""" 

13 

14import os 

15import subprocess 

16import tomllib 

17from pathlib import Path 

18 

19from pydantic import BaseModel, ConfigDict, ValidationError, model_validator 

20 

21from rdak.errors import CliError 

22 

23 

24def _expand(path: str) -> Path: 

25 return Path(path).expanduser() 

26 

27 

28class InstanceConfig(BaseModel): 

29 """Connection and authentication settings for one dak RPC server.""" 

30 

31 model_config = ConfigDict(extra="forbid") 

32 

33 # `unix:/path/to.sock` for a local socket, or `host:port` for TCP. 

34 address: str 

35 # Use TLS. Always on for TCP; for a unix socket it defaults off (local 

36 # channel credentials) and can be enabled to run TLS over the socket 

37 tls: bool = False 

38 # TLS server name override - the certificate name to verify. Required for 

39 # TLS over a unix socket (a socket path is not a host name) and useful when 

40 # reaching a TCP server through a forwarded `localhost` port. 

41 server_name: str | None = None 

42 # Path to a CA bundle for a private CA; omitted means the system store. 

43 ca_cert: str | None = None 

44 

45 # Exactly one token source must be set. 

46 token: str | None = None 

47 token_file: str | None = None 

48 token_command: str | None = None 

49 

50 @property 

51 def is_unix(self) -> bool: 

52 return self.address.startswith(("unix:", "unix-abstract:")) 

53 

54 @property 

55 def use_tls(self) -> bool: 

56 # TCP always uses TLS; a unix socket only when explicitly requested. 

57 return self.tls or not self.is_unix 

58 

59 @model_validator(mode="after") 

60 def _validate(self) -> "InstanceConfig": 

61 sources = [self.token, self.token_file, self.token_command] 

62 if sum(source is not None for source in sources) != 1: 

63 raise ValueError( 

64 "exactly one of token, token_file or token_command must be set" 

65 ) 

66 if not self.use_tls and (self.server_name or self.ca_cert): 

67 raise ValueError("server_name/ca_cert require TLS") 

68 if self.use_tls and self.is_unix and self.server_name is None: 

69 raise ValueError("server_name is required for TLS over a unix socket") 

70 return self 

71 

72 def ca_certificates(self) -> bytes | None: 

73 if self.ca_cert is None: 73 ↛ 75line 73 didn't jump to line 75 because the condition on line 73 was always true

74 return None 

75 return _expand(self.ca_cert).read_bytes() 

76 

77 def get_token(self) -> str: 

78 """Resolve the token from the configured source.""" 

79 if self.token is not None: 

80 return self.token.strip() 

81 if self.token_file is not None: 

82 try: 

83 return _expand(self.token_file).read_text().strip() 

84 except OSError as e: 

85 raise CliError(f"cannot read token_file: {e}") from e 

86 assert self.token_command is not None 

87 try: 

88 result = subprocess.run( 

89 self.token_command, 

90 shell=True, 

91 capture_output=True, 

92 text=True, 

93 check=True, 

94 ) 

95 except subprocess.CalledProcessError as e: 

96 detail = e.stderr.strip() or self.token_command 

97 raise CliError(f"token_command failed ({e.returncode}): {detail}") from e 

98 token = result.stdout.strip() 

99 if not token: 

100 raise CliError("token_command produced an empty token") 

101 return token 

102 

103 

104class RdakConfig(BaseModel): 

105 """Top-level rdak configuration""" 

106 

107 model_config = ConfigDict(extra="forbid") 

108 

109 default_instance: str | None = None 

110 instances: dict[str, InstanceConfig] = {} 

111 

112 def select(self, name: str | None) -> tuple[str, InstanceConfig]: 

113 """Resolve the instance to use. 

114 

115 Order: explicit `name` -> `$RDAK_INSTANCE` -> `default_instance`; 

116 a lone configured instance is used if nothing else selects one. 

117 """ 

118 chosen = name or os.environ.get("RDAK_INSTANCE") or self.default_instance 

119 if chosen is None: 

120 if len(self.instances) == 1: 

121 ((only_name, only_instance),) = self.instances.items() 

122 return only_name, only_instance 

123 raise CliError( 

124 "no instance selected: pass -i/--instance, set $RDAK_INSTANCE " 

125 "or define default_instance in the config", 

126 exit_code=2, 

127 ) 

128 try: 

129 return chosen, self.instances[chosen] 

130 except KeyError: 

131 known = ", ".join(sorted(self.instances)) or "(none)" 

132 raise CliError( 

133 f"unknown instance {chosen!r}; configured instances: {known}", 

134 exit_code=2, 

135 ) from None 

136 

137 

138def _candidate_paths(explicit: str | None) -> list[Path]: 

139 if explicit is not None: 139 ↛ 141line 139 didn't jump to line 141 because the condition on line 139 was always true

140 return [_expand(explicit)] 

141 env = os.environ.get("RDAK_CONFIG") 

142 if env: 

143 return [_expand(env)] 

144 return [_expand("~/.config/rdak.toml"), Path("/etc/dak/rdak.toml")] 

145 

146 

147def load_config(explicit: str | None = None) -> RdakConfig: 

148 """Load and validate the rdak configuration.""" 

149 candidates = _candidate_paths(explicit) 

150 for path in candidates: 

151 if not path.is_file(): 

152 continue 

153 try: 

154 with path.open("rb") as f: 

155 data = tomllib.load(f) 

156 except (OSError, tomllib.TOMLDecodeError) as e: 

157 raise CliError(f"cannot read config {path}: {e}", exit_code=2) from e 

158 try: 

159 return RdakConfig.model_validate(data) 

160 except ValidationError as e: 

161 raise CliError(f"invalid config {path}: {e}", exit_code=2) from e 

162 searched = ", ".join(str(p) for p in candidates) 

163 raise CliError(f"no rdak config found (looked in: {searched})", exit_code=2)