Coverage for src / derivepassphrase / _internals / cli_helpers.py: 100.000%

1# SPDX-FileCopyrightText: 2026 Marco Ricci <software@the13thletter.info> 

2# 

3# SPDX-License-Identifier: Zlib 

4 

5 

6"""Helper functions for the derivepassphrase command-line. 

7 

8Warning: 

9 Non-public module (implementation detail), provided for didactical and 

10 educational purposes only. Subject to change without notice, including 

11 removal. 

12 

13""" 

14 

15from __future__ import annotations (empty)

16 

17import base64 (empty)

18import copy (empty)

19import enum (empty)

20import hashlib (empty)

21import json (empty)

22import logging (empty)

23import os (empty)

24import pathlib (empty)

25import shlex (empty)

26import sys (empty)

27import threading (empty)

28import unicodedata (empty)

29from collections.abc import Sequence (empty)

30from typing import TYPE_CHECKING, cast (empty)

31 

32import click (empty)

33import click.shell_completion (empty)

34import exceptiongroup (empty)

35from typing_extensions import Any (empty)

36 

37from derivepassphrase import _types, ssh_agent, vault (empty)

38from derivepassphrase._internals import cli_messages as _msg (empty)

39from derivepassphrase.ssh_agent import socketprovider (empty)

40 

41if sys.version_info >= (3, 11): (empty)

42 import tomllib (empty)

43else: 

44 import tomli as tomllib (empty)

45 from exceptiongroup import BaseExceptionGroup (empty)

46 

47if TYPE_CHECKING: (empty)

48 import types 

49 from collections.abc import ( 

50 Generator, 

51 Mapping, 

52 ) 

53 from contextlib import AbstractContextManager 

54 from typing import ( 

55 BinaryIO, 

56 Callable, 

57 Literal, 

58 NoReturn, 

59 TextIO, 

60 ) 

61 

62 from typing_extensions import Buffer, Self 

63 

64PROG_NAME = _msg.PROG_NAME (empty)

65KEY_DISPLAY_LENGTH = 50 (empty)

66 

67# Error messages 

68INVALID_VAULT_CONFIG = "Invalid vault config" (empty)

69AGENT_COMMUNICATION_ERROR = "Error communicating with the SSH agent" (empty)

70NO_SUITABLE_KEYS = "No suitable SSH keys were found" (empty)

71EMPTY_SELECTION = "Empty selection" (empty)

72 

73 

74# Shell completion 

75# ================ 

76 

77# Use naive filename completion for the `path` argument of 

78# `derivepassphrase vault`'s `--import` and `--export` options, as well 

79# as the `path` argument of `derivepassphrase export vault`. The latter 

80# treats the pseudo-filename `VAULT_PATH` specially, but this is awkward 

81# to combine with standard filename completion, particularly in bash, so 

82# we would probably have to implement *all* completion (`VAULT_PATH` and 

83# filename completion) ourselves, lacking some niceties of bash's 

84# built-in completion (e.g., adding spaces or slashes depending on 

85# whether the completion is a directory or a complete filename). 

86 

87 

88def shell_complete_path( (empty)

89 ctx: click.Context, 

90 parameter: click.Parameter, 

91 value: str, 

92) -> list[str | click.shell_completion.CompletionItem]: 

93 """Request standard path completion for the `path` argument.""" # noqa: DOC201 

94 del ctx, parameter, value 2 ctx2zbZ

95 return [click.shell_completion.CompletionItem("", type="file")] 2 ctx2zbZ

96 

97 

98# The standard `click` shell completion scripts serialize the completion 

99# items as newline-separated one-line entries, which get silently 

100# corrupted if the value contains newlines. Each shell imposes 

101# additional restrictions: Fish uses newlines in all internal completion 

102# helper scripts, so it is difficult, if not impossible, to register 

103# completion entries containing newlines if completion comes from within 

104# a Fish completion function (instead of a Fish builtin). Zsh's 

105# completion system supports descriptions for each completion item, and 

106# the completion helper functions parse every entry as a colon-separated 

107# 2-tuple of item and description, meaning any colon in the item value 

108# must be escaped. Finally, Bash requires the result array to be 

109# populated at the completion function's top-level scope, but for/while 

110# loops within pipelines do not run at top-level scope, and Bash *also* 

111# strips NUL characters from command substitution output, making it 

112# difficult to read in external data into an array in a cross-platform 

113# manner from entirely within Bash. 

114# 

115# We capitulate in front of these problems---most egregiously because of 

116# Fish---and ensure that completion items (in this case: service names) 

117# never contain ASCII control characters by refusing to offer such 

118# items as valid completions. On the other side, `derivepassphrase` 

119# will warn the user when configuring or importing a service with such 

120# a name that it will not be available for shell completion. 

121 

122 

123def is_completable_item(obj: object) -> bool: (empty)

124 """Return whether the item is completable on the command-line. 

125 

126 The item is completable if and only if it contains no ASCII control 

127 characters (U+0000 through U+001F, and U+007F). 

128 

129 """ 

130 obj = str(obj) 28 ctx2a z t n A 1 q ybZ T d e g B u r v w I G s x y b J D E F

131 forbidden = frozenset(chr(i) for i in range(32)) | {"\x7f"} 28 ctx2a z t n A 1 q ybZ T d e g B u r v w I G s x y b J D E F

132 return not any(f in obj for f in forbidden) 28 ctx2a z t n A 1 q ybZ T d e g B u r v w I G s x y b J D E F

133 

134 

135def shell_complete_service( (empty)

136 ctx: click.Context, 

137 parameter: click.Parameter, 

138 value: str, 

139) -> list[str]: 

140 """Return known vault service names as completion items. 

141 

142 Service names are looked up in the vault configuration file. All 

143 errors will be suppressed. Additionally, any service names deemed 

144 not completable as per [`is_completable_item`][] will be silently 

145 skipped. 

146 

147 """ 

148 del ctx, parameter 6 ctx10(1qZT

149 try: 6 ctx10(1qZT

150 config = load_config() 6 ctx10(1qZT

151 return sorted( 3 ctx11qZ

152 sv 

153 for sv in config["services"] 

154 if sv.startswith(value) and is_completable_item(sv) 

155 ) 

156 except FileNotFoundError: 3 ctx10(T

157 try: 2 ctx10T

158 config, _exc = migrate_and_load_old_config() 2 ctx10T

159 return sorted( 1 ctx1T

160 sv 

161 for sv in config["services"] 

162 if sv.startswith(value) and is_completable_item(sv) 

163 ) 

164 except FileNotFoundError: 1 ctx10

165 return [] 1 ctx10

166 except Exception: # noqa: BLE001 1 ctx1(

167 return [] 1 ctx1(

168 

169 

170# Vault 

171# ===== 

172 

173config_filename_table = { (empty)

174 None: ".", 

175 "write lock": "", 

176 "vault": "vault.json", 

177 "user configuration": "config.toml", 

178 # TODO(the-13th-letter): Remove the old settings.json file. 

179 # https://the13thletter.info/derivepassphrase/latest/upgrade-notes.html#v1.0-old-settings-file 

180 "old settings.json": "settings.json", 

181 "notes backup": "old-notes.txt", 

182} 

183 

184LOCK_SIZE = 4096 (empty)

185""" (empty)

186The size of the record to lock at the beginning of the file, for locking 

187implementations that lock byte ranges instead of whole files. 

188 

189While POSIX specifies that [`fcntl`][] locks shall support a size of zero to 

190denote "any conceivable file size", the locking system available in 

191[`msvcrt`][] does not support this, and requires an explicit size. 

192""" 

193 

194 

195class ConfigurationMutex: (empty)

196 """A mutual exclusion context manager for configuration edits. 

197 

198 See [`configuration_mutex`][]. 

199 

200 """ 

201 

202 lock: Callable[[], None] (empty)

203 """A function to lock the mutex exclusively. (empty)

204 

205 This implementation uses a file descriptor of a well-known file, 

206 which is opened before locking and closed after unlocking (and on 

207 error when locking). On Windows, we use [`msvcrt.locking`][], on 

208 other systems, we use [`fcntl.flock`][]. 

209 

210 Note: 

211 This is a normal Python function, not a method. 

212 

213 Warning: 

214 You really should not have to change this. *If you absolutely 

215 must*, then it is *your responsibility* to ensure that 

216 [`lock`][] and [`unlock`][] are still compatible. 

217 

218 """ 

219 unlock: Callable[[], None] (empty)

220 """A function to unlock the mutex. (empty)

221 

222 This implementation uses a file descriptor of a well-known file, 

223 which is opened before locking and closed after unlocking (and on 

224 error when locking). It will fail if the file descriptor is 

225 unavailable. On Windows, we use [`msvcrt.locking`][], on other 

226 systems, we use [`fcntl.flock`][]. 

227 

228 Note: 

229 This is a normal Python function, not a method. 

230 

231 Warning: 

232 You really should not have to change this. *If you absolutely 

233 must*, then it is *your responsibility* to ensure that 

234 [`lock`][] and [`unlock`][] are still compatible. 

235 

236 """ 

237 write_lock_file: pathlib.Path (empty)

238 """The filename to lock.""" (empty)

239 write_lock_fileobj: BinaryIO | None (empty)

240 """The file object, if currently locked by this context manager.""" (empty)

241 write_lock_condition: threading.Condition (empty)

242 """The lock protecting access to the file object.""" (empty)

243 

244 def __init__(self) -> None: (empty)

245 """Initialize self.""" 

246 if sys.platform == "win32": # pragma: unless the-annoying-os no cover 38 ctx1aztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

247 import msvcrt # noqa: PLC0415 38 ctx1aztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

248 

249 locking = msvcrt.locking 38 ctx1aztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

250 LK_LOCK = msvcrt.LK_LOCK # noqa: N806 38 ctx1aztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

251 LK_UNLCK = msvcrt.LK_UNLCK # noqa: N806 38 ctx1aztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

252 

253 def lock_fd(fd: int, /) -> None: 38 ctx1aztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

254 locking(fd, LK_LOCK, LOCK_SIZE) 37 ctx1ztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

255 

256 def unlock_fd(fd: int, /) -> None: 38 ctx1aztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

257 locking(fd, LK_UNLCK, LOCK_SIZE) 37 ctx1ztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

258 

259 else: # pragma: unless posix no cover 

260 import fcntl # noqa: PLC0415 38 ctx1aztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

261 

262 flock = fcntl.flock 38 ctx1aztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

263 LOCK_EX = fcntl.LOCK_EX # noqa: N806 38 ctx1aztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

264 LOCK_UN = fcntl.LOCK_UN # noqa: N806 38 ctx1aztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

265 

266 def lock_fd(fd: int, /) -> None: 38 ctx1aztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

267 flock(fd, LOCK_EX) 37 ctx1ztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

268 

269 def unlock_fd(fd: int, /) -> None: 38 ctx1aztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

270 flock(fd, LOCK_UN) 37 ctx1ztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

271 

272 def lock_func() -> None: 38 ctx1aztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

273 with self.write_lock_condition: 37 ctx1ztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

274 self.write_lock_condition.wait_for( 37 ctx1ztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

275 lambda: self.write_lock_fileobj is None 

276 ) 

277 self.write_lock_condition.notify() 37 ctx1ztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

278 self.write_lock_file.touch() 37 ctx1ztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

279 self.write_lock_fileobj = self.write_lock_file.open("wb") 37 ctx1ztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

280 lock_fd(self.write_lock_fileobj.fileno()) 37 ctx1ztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

281 

282 def unlock_func() -> None: 38 ctx1aztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

283 with self.write_lock_condition: 37 ctx1ztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

284 assert self.write_lock_fileobj is not None, ( 37 ctx1ztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

285 "We lost track of the configuration write lock " 

286 "file object, so we cannot unlock it anymore!" 

287 ) 

288 unlock_fd(self.write_lock_fileobj.fileno()) 37 ctx1ztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

289 self.write_lock_fileobj.close() 37 ctx1ztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

290 self.write_lock_fileobj = None 37 ctx1ztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

291 

292 self.lock = lock_func 38 ctx1aztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

293 self.unlock = unlock_func 38 ctx1aztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

294 self.write_lock_fileobj = None 38 ctx1aztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

295 self.write_lock_file = config_filename("write lock") 38 ctx1aztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

296 self.write_lock_condition = threading.Condition(threading.Lock()) 38 ctx1aztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

297 

298 def __enter__(self) -> Self: (empty)

299 """Enter the context, locking the configuration file.""" # noqa: DOC201 

300 self.lock() 38 ctx1aztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

301 return self 38 ctx1aztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

302 

303 def __exit__( (empty)

304 self, 

305 exc_type: type[BaseException] | None, 

306 exc_value: BaseException | None, 

307 exc_tb: types.TracebackType | None, 

308 /, 

309 ) -> Literal[False]: 

310 """Exit the context, releasing the lock on the configuration file.""" # noqa: DOC201 

311 self.unlock() 38 ctx1aztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

312 return False 38 ctx1aztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

313 

314 

315def configuration_mutex() -> AbstractContextManager[AbstractContextManager]: (empty)

316 """Enter a mutually exclusive context for configuration writes. 

317 

318 Within this context, no other cooperating instance of 

319 `derivepassphrase` will attempt to write to its configuration 

320 directory. We achieve this by locking a specific temporary file 

321 (whose name depends on the location of the configuration directory) 

322 for the duration of the context. 

323 

324 Returns: 

325 A reusable but not reentrant context manager, ensuring mutual 

326 exclusion (while within its context) with all other 

327 `derivepassphrase` instances using the same configuration 

328 directory. 

329 

330 Upon entering the context, the context manager returns itself. 

331 

332 Note: Locking specifics 

333 The directory for the lock file is determined via 

334 [`get_tempdir`][]. The lock filename is 

335 `derivepassphrase-lock-<hash>.txt`, where `<hash>` is computed 

336 as follows. First, canonicalize the path to the configuration 

337 directory with [`pathlib.Path.resolve`][]. Then encode the 

338 result as per the filesystem encoding ([`os.fsencode`][]), and 

339 hash it with SHA256. Finally, convert the result to standard 

340 base32 and use the first twelve characters, in lowercase, as 

341 `<hash>`. 

342 

343 We use [`msvcrt.locking`][] on Windows platforms (`sys.platform 

344 == "win32"`) and [`fcntl.flock`][] on all others. All locks are 

345 exclusive locks. If the locking system requires a byte range, 

346 we lock the first [`LOCK_SIZE`][] bytes. For maximum 

347 portability between locking implementations, we first open the 

348 lock file for writing, which is sometimes necessary to lock 

349 a file exclusively. Thus locking will fail if we lack 

350 permission to write to an already-existing lockfile. 

351 

352 """ 

353 return ConfigurationMutex() 37 ctx1ztnlAqHdhegBurLMNvwmofIGiKcjksxybJDEF

354 

355 

356def get_tempdir() -> pathlib.Path: (empty)

357 """Return a suitable temporary directory. 

358 

359 We implement the same algorithm as [`tempfile.gettempdir`][], except 

360 that we default to the `derivepassphrase` configuration directory 

361 instead of the current directory if no other choice is suitable, and 

362 that we return [`pathlib.Path`][] objects directly. 

363 

364 """ 

365 paths_to_try: list[pathlib.PurePath] = [] 108 ctx2a z t n l . / A @ [ ] ^ _ ` 0 ( 1 q Z { bbV 8 4 3 ) T 6 7 5 * : H ; d h e g | W X } B ~ ab9 ! u C Q R O p S r U # + = ? , 2 - L M N v w m o f I G i K c j k s x y b J $ D E % ' F cbdbebfbgbhbibjbkblbmbnbobpbqbrbsbtbub

366 env_paths_to_try = [ 108 ctx2a z t n l . / A @ [ ] ^ _ ` 0 ( 1 q Z { bbV 8 4 3 ) T 6 7 5 * : H ; d h e g | W X } B ~ ab9 ! u C Q R O p S r U # + = ? , 2 - L M N v w m o f I G i K c j k s x y b J $ D E % ' F cbdbebfbgbhbibjbkblbmbnbobpbqbrbsbtbub

367 os.getenv("TMPDIR"), 

368 os.getenv("TEMP"), 

369 os.getenv("TMP"), 

370 ] 

371 paths_to_try.extend( 108 ctx2a z t n l . / A @ [ ] ^ _ ` 0 ( 1 q Z { bbV 8 4 3 ) T 6 7 5 * : H ; d h e g | W X } B ~ ab9 ! u C Q R O p S r U # + = ? , 2 - L M N v w m o f I G i K c j k s x y b J $ D E % ' F cbdbebfbgbhbibjbkblbmbnbobpbqbrbsbtbub

372 pathlib.PurePath(p) for p in env_paths_to_try if p is not None 

373 ) 

374 posix_paths_to_try = [ 108 ctx2a z t n l . / A @ [ ] ^ _ ` 0 ( 1 q Z { bbV 8 4 3 ) T 6 7 5 * : H ; d h e g | W X } B ~ ab9 ! u C Q R O p S r U # + = ? , 2 - L M N v w m o f I G i K c j k s x y b J $ D E % ' F cbdbebfbgbhbibjbkblbmbnbobpbqbrbsbtbub

375 pathlib.PurePosixPath("/tmp"), # noqa: S108 

376 pathlib.PurePosixPath("/var/tmp"), # noqa: S108 

377 pathlib.PurePosixPath("/usr/tmp"), 

378 ] 

379 windows_paths_to_try = [ 108 ctx2a z t n l . / A @ [ ] ^ _ ` 0 ( 1 q Z { bbV 8 4 3 ) T 6 7 5 * : H ; d h e g | W X } B ~ ab9 ! u C Q R O p S r U # + = ? , 2 - L M N v w m o f I G i K c j k s x y b J $ D E % ' F cbdbebfbgbhbibjbkblbmbnbobpbqbrbsbtbub

380 pathlib.PureWindowsPath(r"~\AppData\Local\Temp"), 

381 pathlib.PureWindowsPath(os.path.expandvars(r"%SYSTEMROOT%\Temp")), 

382 pathlib.PureWindowsPath(r"C:\TEMP"), 

383 pathlib.PureWindowsPath(r"C:\TMP"), 

384 pathlib.PureWindowsPath(r"\TEMP"), 

385 pathlib.PureWindowsPath(r"\TMP"), 

386 ] 

387 paths_to_try.extend( 108 ctx2a z t n l . / A @ [ ] ^ _ ` 0 ( 1 q Z { bbV 8 4 3 ) T 6 7 5 * : H ; d h e g | W X } B ~ ab9 ! u C Q R O p S r U # + = ? , 2 - L M N v w m o f I G i K c j k s x y b J $ D E % ' F cbdbebfbgbhbibjbkblbmbnbobpbqbrbsbtbub

388 windows_paths_to_try if sys.platform == "win32" else posix_paths_to_try 

389 ) 

390 for p in paths_to_try: 108 ctx2a z t n l . / A @ [ ] ^ _ ` 0 ( 1 q Z { bbV 8 4 3 ) T 6 7 5 * : H ; d h e g | W X } B ~ ab9 ! u C Q R O p S r U # + = ? , 2 - L M N v w m o f I G i K c j k s x y b J $ D E % ' F cbdbebfbgbhbibjbkblbmbnbobpbqbrbsbtbub

391 path = pathlib.Path(p).expanduser() 108 ctx2a z t n l . / A @ [ ] ^ _ ` 0 ( 1 q Z { bbV 8 4 3 ) T 6 7 5 * : H ; d h e g | W X } B ~ ab9 ! u C Q R O p S r U # + = ? , 2 - L M N v w m o f I G i K c j k s x y b J $ D E % ' F cbdbebfbgbhbibjbkblbmbnbobpbqbrbsbtbub

392 try: 108 ctx2a z t n l . / A @ [ ] ^ _ ` 0 ( 1 q Z { bbV 8 4 3 ) T 6 7 5 * : H ; d h e g | W X } B ~ ab9 ! u C Q R O p S r U # + = ? , 2 - L M N v w m o f I G i K c j k s x y b J $ D E % ' F cbdbebfbgbhbibjbkblbmbnbobpbqbrbsbtbub

393 points_to_dir = path.is_dir() 108 ctx2a z t n l . / A @ [ ] ^ _ ` 0 ( 1 q Z { bbV 8 4 3 ) T 6 7 5 * : H ; d h e g | W X } B ~ ab9 ! u C Q R O p S r U # + = ? , 2 - L M N v w m o f I G i K c j k s x y b J $ D E % ' F cbdbebfbgbhbibjbkblbmbnbobpbqbrbsbtbub

394 except OSError: 1 ctx1;

395 continue 1 ctx1;

396 else: 

397 if points_to_dir: 108 ctx2a z t n l . / A @ [ ] ^ _ ` 0 ( 1 q Z { bbV 8 4 3 ) T 6 7 5 * : H ; d h e g | W X } B ~ ab9 ! u C Q R O p S r U # + = ? , 2 - L M N v w m o f I G i K c j k s x y b J $ D E % ' F cbdbebfbgbhbibjbkblbmbnbobpbqbrbsbtbub

398 return path.resolve(strict=True) 107 ctx2a z t n l . / A @ [ ] ^ _ ` 0 ( 1 q Z { bbV 8 4 3 ) T 6 7 5 * : H d h e g | W X } B ~ ab9 ! u C Q R O p S r U # + = ? , 2 - L M N v w m o f I G i K c j k s x y b J $ D E % ' F cbdbebfbgbhbibjbkblbmbnbobpbqbrbsbtbub

399 return config_filename(subsystem=None) 1 ctx1;

400 

401 

402def config_filename( (empty)

403 subsystem: str | None = "old settings.json", 

404) -> pathlib.Path: 

405 """Return the filename of the configuration file for the subsystem. 

406 

407 The (implicit default) file is currently named `settings.json`, 

408 located within the configuration directory as determined by the 

409 `DERIVEPASSPHRASE_PATH` environment variable, or by 

410 [`click.get_app_dir`][] in POSIX mode. Depending on the requested 

411 subsystem, this will usually be a different file within that 

412 directory. 

413 

414 Args: 

415 subsystem: 

416 Name of the configuration subsystem whose configuration 

417 filename to return. If not given, return the old filename 

418 from before the subcommand migration. If `None`, return the 

419 configuration directory instead. 

420 

421 Raises: 

422 AssertionError: 

423 An unknown subsystem was passed. 

424 

425 Deprecated: 

426 Since v0.2.0: The implicit default subsystem and the old 

427 configuration filename are deprecated, and will be removed in v1.0. 

428 The subsystem will be mandatory to specify. 

429 

430 """ 

431 path = pathlib.Path( 108 ctx2a z t n l . / A @ [ ] ^ _ ` 0 ( 1 q Z { bbV 8 4 3 ) T 6 7 5 * : H ; d h e g | W X } B ~ ab9 ! u C Q R O p S r U # + = ? , 2 - L M N v w m o f I G i K c j k s x y b J $ D E % ' F cbdbebfbgbhbibjbkblbmbnbobpbqbrbsbtbub

432 os.getenv(PROG_NAME.upper() + "_PATH") 

433 or click.get_app_dir(PROG_NAME, force_posix=True) 

434 ) 

435 if subsystem == "write lock": 108 ctx2a z t n l . / A @ [ ] ^ _ ` 0 ( 1 q Z { bbV 8 4 3 ) T 6 7 5 * : H ; d h e g | W X } B ~ ab9 ! u C Q R O p S r U # + = ? , 2 - L M N v w m o f I G i K c j k s x y b J $ D E % ' F cbdbebfbgbhbibjbkblbmbnbobpbqbrbsbtbub

436 path_hash = base64.b32encode( 108 ctx2a z t n l . / A @ [ ] ^ _ ` 0 ( 1 q Z { bbV 8 4 3 ) T 6 7 5 * : H ; d h e g | W X } B ~ ab9 ! u C Q R O p S r U # + = ? , 2 - L M N v w m o f I G i K c j k s x y b J $ D E % ' F cbdbebfbgbhbibjbkblbmbnbobpbqbrbsbtbub

437 hashlib.sha256(os.fsencode(path.resolve())).digest() 

438 ) 

439 path_hash_text = path_hash[:12].lower().decode("ASCII") 108 ctx2a z t n l . / A @ [ ] ^ _ ` 0 ( 1 q Z { bbV 8 4 3 ) T 6 7 5 * : H ; d h e g | W X } B ~ ab9 ! u C Q R O p S r U # + = ? , 2 - L M N v w m o f I G i K c j k s x y b J $ D E % ' F cbdbebfbgbhbibjbkblbmbnbobpbqbrbsbtbub

440 temp_path = get_tempdir() 108 ctx2a z t n l . / A @ [ ] ^ _ ` 0 ( 1 q Z { bbV 8 4 3 ) T 6 7 5 * : H ; d h e g | W X } B ~ ab9 ! u C Q R O p S r U # + = ? , 2 - L M N v w m o f I G i K c j k s x y b J $ D E % ' F cbdbebfbgbhbibjbkblbmbnbobpbqbrbsbtbub

441 filename_ = f"derivepassphrase-lock-{path_hash_text}.txt" 108 ctx2a z t n l . / A @ [ ] ^ _ ` 0 ( 1 q Z { bbV 8 4 3 ) T 6 7 5 * : H ; d h e g | W X } B ~ ab9 ! u C Q R O p S r U # + = ? , 2 - L M N v w m o f I G i K c j k s x y b J $ D E % ' F cbdbebfbgbhbibjbkblbmbnbobpbqbrbsbtbub

442 return temp_path / filename_ 108 ctx2a z t n l . / A @ [ ] ^ _ ` 0 ( 1 q Z { bbV 8 4 3 ) T 6 7 5 * : H ; d h e g | W X } B ~ ab9 ! u C Q R O p S r U # + = ? , 2 - L M N v w m o f I G i K c j k s x y b J $ D E % ' F cbdbebfbgbhbibjbkblbmbnbobpbqbrbsbtbub

443 try: 88 ctx2a z t n l . / A @ [ ] ^ _ ` 0 ( 1 q Z { V 8 4 3 ) T 6 7 5 * : H ; d h e g | W X } B ~ ab9 ! u C Q R O p S r U # + = ? , 2 - L M N v w m o f I G i K c j k s x y b J $ D E % ' F

444 filename = config_filename_table[subsystem] 88 ctx2a z t n l . / A @ [ ] ^ _ ` 0 ( 1 q Z { V 8 4 3 ) T 6 7 5 * : H ; d h e g | W X } B ~ ab9 ! u C Q R O p S r U # + = ? , 2 - L M N v w m o f I G i K c j k s x y b J $ D E % ' F

445 except (KeyError, TypeError): # pragma: no cover [failsafe] 

446 msg = f"Unknown configuration subsystem: {subsystem!r}" 

447 raise AssertionError(msg) from None 

448 return path / filename 88 ctx2a z t n l . / A @ [ ] ^ _ ` 0 ( 1 q Z { V 8 4 3 ) T 6 7 5 * : H ; d h e g | W X } B ~ ab9 ! u C Q R O p S r U # + = ? , 2 - L M N v w m o f I G i K c j k s x y b J $ D E % ' F

449 

450 

451def load_config() -> _types.VaultConfig: (empty)

452 """Load a vault(1)-compatible config from the application directory. 

453 

454 The filename is obtained via [`config_filename`][]. This must be 

455 an unencrypted JSON file. 

456 

457 Returns: 

458 The vault settings. See [`_types.VaultConfig`][] for details. 

459 

460 Raises: 

461 OSError: 

462 There was an OS error accessing the file. 

463 ValueError: 

464 The data loaded from the file is not a vault(1)-compatible 

465 config. 

466 

467 """ 

468 filename = config_filename(subsystem="vault") 65 ctx1aztnlA01qZV843T*HdhegWXB9!uCQROpSrU#+=?,2-vwmofIGiKcjksxybJ$DE%'F

469 with filename.open("rb") as fileobj: 65 ctx1aztnlA01qZV843T*HdhegWXB9!uCQROpSrU#+=?,2-vwmofIGiKcjksxybJ$DE%'F

470 data = json.load(fileobj) 52 ctx1aztnl1qZ*HdhegB9!uCQROpSrU#+,-vwmofIGiKcjkxybJ$DE%'F

471 if not _types.is_vault_config(data): 52 ctx1aztnl1qZ*HdhegB9!uCQROpSrU#+,-vwmofIGiKcjkxybJ$DE%'F

472 raise ValueError(INVALID_VAULT_CONFIG) 1 ctx1+

473 return data 51 ctx1aztnl1qZ*HdhegB9!uCQROpSrU#,-vwmofIGiKcjkxybJ$DE%'F

474 

475 

476# TODO(the-13th-letter): Remove this function. 

477# https://the13thletter.info/derivepassphrase/latest/upgrade-notes.html#v1.0-old-settings-file 

478def migrate_and_load_old_config() -> tuple[_types.VaultConfig, OSError | None]: (empty)

479 """Load and migrate a vault(1)-compatible config. 

480 

481 The (old) filename is obtained via [`config_filename`][]. This 

482 must be an unencrypted JSON file. After loading, the file is 

483 migrated to the new standard filename. 

484 

485 Returns: 

486 The vault settings, and an optional exception encountered during 

487 migration. See [`_types.VaultConfig`][] for details on the 

488 former. 

489 

490 Raises: 

491 OSError: 

492 There was an OS error accessing the old file. 

493 ValueError: 

494 The data loaded from the file is not a vault(1)-compatible 

495 config. 

496 

497 """ 

498 new_filename = config_filename(subsystem="vault") 15 ctx1A0V843)T675WX2s

499 old_filename = config_filename(subsystem="old settings.json") 15 ctx1A0V843)T675WX2s

500 with old_filename.open("rb") as fileobj: 15 ctx1A0V843)T675WX2s

501 data = json.load(fileobj) 7 ctx143)T675

502 if not _types.is_vault_config(data): 7 ctx143)T675

503 raise ValueError(INVALID_VAULT_CONFIG) 1 ctx1)

504 try: 6 ctx143T675

505 old_filename.rename(new_filename) 6 ctx143T675

506 except OSError as exc: 2 ctx135

507 return data, exc 2 ctx135

508 else: 

509 return data, None 4 ctx14T67

510 

511 

512def save_config(config: _types.VaultConfig, /) -> None: (empty)

513 """Save a vault(1)-compatible config to the application directory. 

514 

515 The filename is obtained via [`config_filename`][]. The config 

516 will be stored as an unencrypted JSON file. 

517 

518 Args: 

519 config: 

520 vault configuration to save. 

521 

522 Raises: 

523 OSError: 

524 There was an OS error accessing or writing the file. 

525 ValueError: 

526 The data cannot be stored as a vault(1)-compatible config. 

527 

528 """ 

529 if not _types.is_vault_config(config): 24 ctx1aznAq:HdhegBurvwGsxybDEF

530 raise ValueError(INVALID_VAULT_CONFIG) 1 ctx1:

531 filename = config_filename(subsystem="vault") 23 ctx1aznAqHdhegBurvwGsxybDEF

532 filedir = filename.resolve().parent 23 ctx1aznAqHdhegBurvwGsxybDEF

533 filedir.mkdir(parents=True, exist_ok=True) 23 ctx1aznAqHdhegBurvwGsxybDEF

534 with filename.open("w", encoding="UTF-8") as fileobj: 22 ctx1aznAqHdhegBurvGsxybDEF

535 json.dump( 21 ctx1aznAqHdhegBurvsxybDEF

536 config, fileobj, ensure_ascii=False, indent=2, sort_keys=True 

537 ) 

538 

539 

540def load_user_config() -> dict[str, Any]: (empty)

541 """Load the user config from the application directory. 

542 

543 The filename is obtained via [`config_filename`][]. 

544 

545 Returns: 

546 The user configuration, as a nested `dict`. 

547 

548 Raises: 

549 OSError: 

550 There was an OS error accessing the file. 

551 ValueError: 

552 The data loaded from the file is not a valid configuration 

553 file. 

554 

555 """ 

556 filename = config_filename(subsystem="user configuration") 56 ctx1aztnl./AqV8dhegWXB9!uCQROpSrU#LMNvwmofIGiKcjksxybJ$DE%'F

557 with filename.open("rb") as fileobj: 56 ctx1aztnl./AqV8dhegWXB9!uCQROpSrU#LMNvwmofIGiKcjksxybJ$DE%'F

558 return tomllib.load(fileobj) 5 ctx1tn.RO

559 

560 

561def get_suitable_ssh_keys( (empty)

562 conn: ssh_agent.SSHAgentClient 

563 | _types.SSHAgentSocket 

564 | Sequence[str] 

565 | None = None, 

566 /, 

567) -> Generator[_types.SSHKeyCommentPair, None, None]: 

568 """Yield all SSH keys suitable for passphrase derivation. 

569 

570 Suitable SSH keys are queried from the running SSH agent (see 

571 [`ssh_agent.SSHAgentClient.list_keys`][]). 

572 

573 Args: 

574 conn: 

575 An optional connection hint to the SSH agent. See 

576 [`ssh_agent.SSHAgentClient.ensure_agent_subcontext`][]. 

577 

578 Yields: 

579 Every SSH key from the SSH agent that is suitable for passphrase 

580 derivation. 

581 

582 Raises: 

583 derivepassphrase.ssh_agent.socketprovider.NoSuchProviderError: 

584 As per 

585 [`ssh_agent.SSHAgentClient.__init__`][ssh_agent.SSHAgentClient]. 

586 Only applicable if agent auto-discovery is used. 

587 KeyError: 

588 As per 

589 [`ssh_agent.SSHAgentClient.__init__`][ssh_agent.SSHAgentClient]. 

590 Only applicable if agent auto-discovery is used. 

591 NotImplementedError: 

592 As per 

593 [`ssh_agent.SSHAgentClient.__init__`][ssh_agent.SSHAgentClient], 

594 including the mulitple raise as an exception group. Only 

595 applicable if agent auto-discovery is used. 

596 OSError: 

597 If the connection hint was a socket, then there was an error 

598 setting up the socket connection to the agent. 

599 

600 Otherwise, as per 

601 [`ssh_agent.SSHAgentClient.__init__`][ssh_agent.SSHAgentClient]. 

602 Only applicable if agent auto-discovery is used. 

603 LookupError: 

604 No keys usable for passphrase derivation are loaded into the 

605 SSH agent. 

606 RuntimeError: 

607 There was an error communicating with the SSH agent. 

608 ssh_agent.SSHAgentFailedError: 

609 The agent failed to supply a list of loaded keys. 

610 

611 """ 

612 with ssh_agent.SSHAgentClient.ensure_agent_subcontext(conn) as client: 9 ctx2l xbp m o i j k P

613 try: 6 ctx2xbp i j k P

614 all_key_comment_pairs = list(client.list_keys()) 6 ctx2xbp i j k P

615 except EOFError as exc: # pragma: no cover [failsafe] 2 ctx1jk

616 raise RuntimeError(AGENT_COMMUNICATION_ERROR) from exc 

617 suitable_keys = copy.copy(all_key_comment_pairs) 4 ctx2xbp i P

618 for pair in all_key_comment_pairs: 4 ctx2xbp i P

619 key, _comment = pair 3 ctx2xbp P

620 if vault.Vault.is_suitable_ssh_key(key, client=client): 3 ctx2xbp P

621 yield pair 5 ctx2xbp j k P

622 if not suitable_keys: 4 ctx2xbp i P

623 raise LookupError(NO_SUITABLE_KEYS) 1 ctx1i

624 

625 

626def prompt_for_selection( (empty)

627 items: Sequence[str | bytes], 

628 heading: str = "Possible choices:", 

629 single_choice_prompt: str = "Confirm this choice?", 

630 ctx: click.Context | None = None, 

631) -> int: 

632 """Prompt user for a choice among the given items. 

633 

634 Print the heading, if any, then present the items to the user. If 

635 there are multiple items, prompt the user for a selection, validate 

636 the choice, then return the list index of the selected item. If 

637 there is only a single item, request confirmation for that item 

638 instead, and return the correct index. 

639 

640 Args: 

641 items: 

642 The list of items to choose from. 

643 heading: 

644 A heading for the list of items, to print immediately 

645 before. Defaults to a reasonable standard heading. If 

646 explicitly empty, print no heading. 

647 single_choice_prompt: 

648 The confirmation prompt if there is only a single possible 

649 choice. Defaults to a reasonable standard prompt. 

650 ctx: 

651 An optional `click` context, from which output device 

652 properties and color preferences will be queried. 

653 

654 Returns: 

655 An index into the items sequence, indicating the user's 

656 selection. 

657 

658 Raises: 

659 IndexError: 

660 The user made an invalid or empty selection, or requested an 

661 abort. 

662 

663 """ 

664 n = len(items) 7 ctx2vbwbC p c b P

665 color = ctx.color if ctx is not None else None 7 ctx2vbwbC p c b P

666 if heading: 7 ctx2vbwbC p c b P

667 click.echo(click.style(heading, bold=True), err=True, color=color) 6 ctx2vbC p c b P

668 for i, x in enumerate(items, start=1): 7 ctx2vbwbC p c b P

669 click.echo( 7 ctx2vbwbC p c b P

670 click.style(f"[{i}]", bold=True), nl=False, err=True, color=color 

671 ) 

672 click.echo(" ", nl=False, err=True, color=color) 7 ctx2vbwbC p c b P

673 click.echo(x, err=True, color=color) 7 ctx2vbwbC p c b P

674 if n > 1: 7 ctx2vbwbC p c b P

675 choices = click.Choice([""] + [str(i) for i in range(1, n + 1)]) 6 ctx2vbC p c b P

676 try: 6 ctx2vbC p c b P

677 choice = click.prompt( 6 ctx2vbC p c b P

678 f"Your selection? (1-{n}, leave empty to abort)", 

679 err=True, 

680 type=choices, 

681 show_choices=False, 

682 show_default=False, 

683 default="", 

684 ) 

685 except click.Abort: # pragma: no cover [external] 2 ctx2vbc

686 # This branch will not be triggered during testing on 

687 # `click` versions < 8.2.1, due to (non-monkeypatch-able) 

688 # deficiencies in `click.testing.CliRunner`. Therefore, as 

689 # an external source of nondeterminism, exclude it from 

690 # coverage. 

691 # 

692 # https://github.com/pallets/click/issues/2934 

693 choice = "" 2 ctx2vbc

694 if not choice: 6 ctx2vbC p c b P

695 raise IndexError(EMPTY_SELECTION) 2 ctx2vbc

696 return int(choice) - 1 5 ctx2vbC p b P

697 prompt_suffix = ( 2 ctx2wbP

698 " " if single_choice_prompt.endswith(tuple("?.!")) else ": " 

699 ) 

700 try: 2 ctx2wbP

701 click.confirm( 2 ctx2wbP

702 single_choice_prompt, 

703 prompt_suffix=prompt_suffix, 

704 err=True, 

705 abort=True, 

706 default=False, 

707 show_default=False, 

708 ) 

709 except click.Abort: 1 ctx2wb

710 raise IndexError(EMPTY_SELECTION) from None 1 ctx2wb

711 return 0 2 ctx2wbP

712 

713 

714def handle_nosuchprovidererror( (empty)

715 error_callback: Callable[..., NoReturn], 

716 warning_callback: Callable[..., None], 

717) -> Callable[[BaseExceptionGroup], NoReturn]: 

718 """Generate a handler for socketprovider.NoSuchProviderException in try-except*. 

719 

720 Returns a function emitting a standard user-facing message. 

721 

722 """ # noqa: DOC201, E501 

723 del warning_callback 17 ctx1lYCQROpSmoficjkbP

724 

725 def handle_nosuchprovidererror(excgroup: BaseExceptionGroup) -> NoReturn: 17 ctx1lYCQROpSmoficjkbP

726 subset = excgroup.subgroup(socketprovider.NoSuchProviderError) 1 ctx1O

727 provider_name = ( 1 ctx1O

728 ascii(subset.exceptions[0].args[0]) 

729 if subset is not None 

730 else "(null)" 

731 ) 

732 error_callback( 1 ctx1O

733 _msg.TranslatedString( 

734 _msg.ErrMsgTemplate.UNKNOWN_SSH_AGENT_SOCKET_PROVIDER, 

735 provider=provider_name, 

736 ) 

737 ) 

738 

739 return handle_nosuchprovidererror 17 ctx1lYCQROpSmoficjkbP

740 

741 

742def handle_keyerror( (empty)

743 error_callback: Callable[..., NoReturn], 

744 warning_callback: Callable[..., None], 

745) -> Callable[[BaseExceptionGroup], NoReturn]: 

746 """Generate a handler for KeyError in try-except*. 

747 

748 Returns a function emitting a standard user-facing message. 

749 

750 """ # noqa: DOC201 

751 del warning_callback 17 ctx1lYCQROpSmoficjkbP

752 

753 def handle_keyerror(_excgroup: BaseExceptionGroup) -> NoReturn: 17 ctx1lYCQROpSmoficjkbP

754 error_callback( 2 ctx1Yo

755 _msg.TranslatedString(_msg.ErrMsgTemplate.NO_SSH_AGENT_FOUND) 

756 ) 

757 

758 return handle_keyerror 17 ctx1lYCQROpSmoficjkbP

759 

760 

761def handle_notimplementederror( (empty)

762 error_callback: Callable[..., NoReturn], 

763 warning_callback: Callable[..., None], 

764) -> Callable[[BaseExceptionGroup], NoReturn]: 

765 """Generate a handler for NotImplementedError in try-except*. 

766 

767 Returns a function emitting a standard user-facing message. 

768 

769 """ # noqa: DOC201 

770 

771 def handle_notimplementederror(excgroup: BaseExceptionGroup) -> NoReturn: 17 ctx1lYCQROpSmoficjkbP

772 if excgroup.subgroup( 2 ctx1lY

773 socketprovider.UnixDomainSocketsNotAvailableError 

774 ): 

775 warning_callback( 2 ctx1lY

776 _msg.TranslatedString( 

777 _msg.WarnMsgTemplate.NO_UNIX_DOMAIN_SOCKETS 

778 ) 

779 ) 

780 if excgroup.subgroup( 2 ctx1lY

781 socketprovider.WindowsNamedPipesNotAvailableError 

782 ): 

783 warning_callback( 2 ctx1lY

784 _msg.TranslatedString( 

785 _msg.WarnMsgTemplate.NO_WINDOWS_NAMED_PIPES 

786 ) 

787 ) 

788 error_callback( 2 ctx1lY

789 _msg.TranslatedString(_msg.ErrMsgTemplate.NO_AGENT_SUPPORT) 

790 ) 

791 

792 return handle_notimplementederror 17 ctx1lYCQROpSmoficjkbP

793 

794 

795def handle_oserror( (empty)

796 error_callback: Callable[..., NoReturn], 

797 warning_callback: Callable[..., None], 

798) -> Callable[[BaseExceptionGroup], NoReturn]: 

799 """Generate a handler for OSError in try-except*. 

800 

801 Returns a function emitting a standard user-facing message. 

802 

803 """ # noqa: DOC201 

804 del warning_callback 17 ctx1lYCQROpSmoficjkbP

805 

806 def handle_oserror(excgroup: BaseExceptionGroup) -> NoReturn: 17 ctx1lYCQROpSmoficjkbP

807 for exc in excgroup.exceptions: 2 ctx1Ym

808 assert isinstance(exc, OSError) 2 ctx1Ym

809 error_callback( 2 ctx1Ym

810 _msg.TranslatedString( 

811 _msg.ErrMsgTemplate.CANNOT_CONNECT_TO_AGENT, 

812 error=exc.strerror, 

813 filename=exc.filename, 

814 ).maybe_without_filename() 

815 ) 

816 raise AssertionError() 

817 

818 return handle_oserror 17 ctx1lYCQROpSmoficjkbP

819 

820 

821def handle_runtimeerror( (empty)

822 error_callback: Callable[..., NoReturn], 

823 warning_callback: Callable[..., None], 

824) -> Callable[[BaseExceptionGroup], NoReturn]: 

825 """Generate a handler for RuntimeError in try-except*. 

826 

827 Returns a function emitting a standard user-facing message. 

828 

829 """ # noqa: DOC201 

830 del warning_callback 17 ctx1lYCQROpSmoficjkbP

831 

832 def handle_runtimeerror(excgroup: BaseExceptionGroup) -> NoReturn: 17 ctx1lYCQROpSmoficjkbP

833 for exc in excgroup.exceptions: 2 ctx1Yk

834 error_callback( 2 ctx1Yk

835 _msg.TranslatedString( 

836 _msg.ErrMsgTemplate.CANNOT_UNDERSTAND_AGENT 

837 ), 

838 exc_info=exc, 

839 ) 

840 raise AssertionError() 

841 

842 return handle_runtimeerror 17 ctx1lYCQROpSmoficjkbP

843 

844 

845def default_error_callback( (empty)

846 message: Any, # noqa: ANN401 (empty)

847 /, 

848 *_args: Any, # noqa: ANN401 (empty)

849 **_kwargs: Any, # noqa: ANN401 (empty)

850) -> NoReturn: # pragma: no cover [failsafe] (empty)

851 """Calls [`sys.exit`][] on its first argument, ignoring the rest.""" 

852 sys.exit(message) 

853 

854 

855def select_ssh_key( (empty)

856 conn: ssh_agent.SSHAgentClient 

857 | _types.SSHAgentSocket 

858 | Sequence[str] 

859 | None = None, 

860 /, 

861 *, 

862 ctx: click.Context | None = None, 

863 error_callback: Callable[..., NoReturn], 

864 warning_callback: Callable[..., None], 

865) -> bytes | bytearray: 

866 """Interactively select an SSH key for passphrase derivation. 

867 

868 Suitable SSH keys are queried from the running SSH agent (see 

869 [`ssh_agent.SSHAgentClient.list_keys`][]), then the user is prompted 

870 interactively (see [`click.prompt`][]) for a selection. If any 

871 error occurs, the user receives an appropriate error message. 

872 

873 Args: 

874 conn: 

875 An optional connection hint to the SSH agent. See 

876 [`ssh_agent.SSHAgentClient.ensure_agent_subcontext`][]. 

877 ctx: 

878 An `click` context, queried for output device properties and 

879 color preferences when issuing the prompt. 

880 error_callback: 

881 A callback function for an error message, if any. The 

882 callback function is responsible for aborting this function 

883 call after acknowledging, formatting and/or forwarding the 

884 error message; it would typically call [`sys.exit`][] or 

885 raise an exception of its own, based on the provided error 

886 message. 

887 warning_callback: 

888 A callback function for a warning message, if any. The 

889 callback function is responsible for formatting the warning 

890 message and dispatching it into the warning system, if so 

891 desired. 

892 

893 Returns: 

894 The selected SSH key. 

895 

896 """ 

897 

898 def handle_lookuperror(_excgroup: BaseExceptionGroup) -> NoReturn: 12 ctx1lCpmoficjkbP

899 error_callback( 1 ctx1i

900 _msg.TranslatedString( 

901 _msg.ErrMsgTemplate.NO_SUITABLE_SSH_KEYS, 

902 PROG_NAME=PROG_NAME, 

903 ) 

904 ) 

905 

906 def handle_sshagentfailederror(excgroup: BaseExceptionGroup) -> NoReturn: 12 ctx1lCpmoficjkbP

907 for exc in excgroup.exceptions: 1 ctx1j

908 error_callback( 1 ctx1j

909 _msg.TranslatedString( 

910 _msg.ErrMsgTemplate.AGENT_REFUSED_LIST_KEYS 

911 ), 

912 exc_info=exc, 

913 ) 

914 raise AssertionError() 

915 

916 with exceptiongroup.catch({ 12 ctx1lCpmoficjkbP

917 socketprovider.NoSuchProviderError: handle_nosuchprovidererror( 

918 error_callback, warning_callback 

919 ), 

920 KeyError: handle_keyerror(error_callback, warning_callback),