Coverage for src / derivepassphrase / _types.py: 100.000%

339 statements  

« prev     ^ index     » next       coverage.py v7.13.1, created at 2026-07-06 21:34 +0200

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

2# 

3# SPDX-License-Identifier: Zlib 

4 

5"""Types used by derivepassphrase.""" 

6 

7from __future__ import annotations 

8 

9import enum 

10import json 

11import math 

12import string 

13import warnings 

14from typing import TYPE_CHECKING, Generic, Protocol, TypeVar, cast 

15 

16from typing_extensions import ( 

17 Buffer, 

18 NamedTuple, 

19 NotRequired, 

20 TypeAlias, 

21 TypedDict, 

22 deprecated, 

23 get_overloads, 

24 overload, 

25 runtime_checkable, 

26) 

27 

28if TYPE_CHECKING: 

29 from collections.abc import Callable, Generator, Sequence 

30 from typing import Literal 

31 

32 from typing_extensions import ( 

33 Any, 

34 Required, 

35 TypeIs, 

36 ) 

37 

38__all__ = ( 

39 "SSH_AGENT", 

40 "SSH_AGENTC", 

41 "SSHKeyCommentPair", 

42 "VaultConfig", 

43 "is_vault_config", 

44) 

45 

46 

47class _Omitted: 

48 def __bool__(self) -> bool: 

49 return False 

50 

51 def __repr__(self) -> str: 

52 return "..." 

53 

54 

55class VaultConstructorArgs(TypedDict, total=False): 

56 r"""Common, optional arguments for the vault constructor. 

57 

58 (Also doubles as a base class for the vault configuration global 

59 settings.) 

60 

61 Attributes: 

62 length: 

63 Desired passphrase length. 

64 repeat: 

65 The maximum number of immediate character repetitions 

66 allowed in the passphrase. Disabled if set to 0. 

67 lower: 

68 Optional constraint on ASCII lowercase characters. If 

69 positive, include this many lowercase characters 

70 somewhere in the passphrase. If 0, avoid lowercase 

71 characters altogether. 

72 upper: 

73 Same as `lower`, but for ASCII uppercase characters. 

74 number: 

75 Same as `lower`, but for ASCII digits. 

76 space: 

77 Same as `lower`, but for the space character. 

78 dash: 

79 Same as `lower`, but for the hyphen-minus and underscore 

80 characters. 

81 symbol: 

82 Same as `lower`, but for all other hitherto unlisted 

83 ASCII printable characters (except backquote). 

84 

85 """ 

86 

87 length: NotRequired[int] 

88 """""" 

89 repeat: NotRequired[int] 

90 """""" 

91 lower: NotRequired[int] 

92 """""" 

93 upper: NotRequired[int] 

94 """""" 

95 number: NotRequired[int] 

96 """""" 

97 space: NotRequired[int] 

98 """""" 

99 dash: NotRequired[int] 

100 """""" 

101 symbol: NotRequired[int] 

102 """""" 

103 

104 

105class VaultConfigGlobalSettings(VaultConstructorArgs, total=False): 

106 r"""Configuration for vault: global settings. 

107 

108 Attributes: 

109 key: 

110 The base64-encoded ssh public key to use, overriding the 

111 master passphrase. Optional. 

112 phrase: 

113 The master passphrase. Optional. 

114 unicode_normalization_form: 

115 The preferred Unicode normalization form; we warn the user 

116 if textual passphrases do not match their normalized forms. 

117 Optional, and a `derivepassphrase` extension. 

118 

119 """ 

120 

121 key: NotRequired[str] 

122 """""" 

123 phrase: NotRequired[str] 

124 """""" 

125 unicode_normalization_form: NotRequired[ 

126 Literal["NFC", "NFD", "NFKC", "NFKD"] 

127 ] 

128 """""" 

129 

130 

131class VaultConfigServicesSettings(VaultConfigGlobalSettings, total=False): 

132 r"""Configuration for vault: services settings. 

133 

134 Attributes: 

135 notes: 

136 Optional notes for this service, to display to the user when 

137 generating the passphrase. 

138 key: 

139 As per the global settings. 

140 phrase: 

141 As per the global settings. 

142 unicode_normalization_form: 

143 As per the global settings. 

144 length: 

145 As per the global settings. 

146 repeat: 

147 As per the global settings. 

148 lower: 

149 As per the global settings. 

150 upper: 

151 As per the global settings. 

152 number: 

153 As per the global settings. 

154 space: 

155 As per the global settings. 

156 dash: 

157 As per the global settings. 

158 symbol: 

159 As per the global settings. 

160 

161 """ 

162 

163 notes: NotRequired[str] 

164 """""" 

165 

166 

167_VaultConfig = TypedDict( 

168 "_VaultConfig", 

169 {"global": NotRequired[VaultConfigGlobalSettings]}, 

170 total=False, 

171) 

172 

173 

174class VaultConfig(TypedDict, _VaultConfig, total=False): 

175 r"""Configuration for vault. For typing purposes. 

176 

177 Usually stored as JSON. 

178 

179 Attributes: 

180 global (NotRequired[VaultConfigGlobalSettings]): 

181 Global settings. 

182 services (Required[dict[str, VaultConfigServicesSettings]]): 

183 Service-specific settings. 

184 

185 """ 

186 

187 services: Required[dict[str, VaultConfigServicesSettings]] 

188 

189 

190def json_path(path: Sequence[str | int], /) -> str: 

191 r"""Transform a series of keys and indices into a JSONPath selector. 

192 

193 The resulting JSONPath selector conforms to RFC 9535, is always 

194 rooted at the JSON root node (i.e., starts with `$`), and only 

195 contains name and index selectors (in shorthand dot notation, where 

196 possible). 

197 

198 Args: 

199 path: 

200 A sequence of object keys or array indices to navigate to 

201 the desired JSON value, starting from the root node. 

202 

203 Returns: 

204 A valid JSONPath selector (a string) identifying the desired 

205 JSON value. 

206 

207 Examples: 

208 >>> json_path(["global", "phrase"]) 

209 '$.global.phrase' 

210 >>> json_path(["services", "service name with spaces", "length"]) 

211 '$.services["service name with spaces"].length' 

212 >>> json_path(["services", "special\u000acharacters", "notes"]) 

213 '$.services["special\\ncharacters"].notes' 

214 >>> print(json_path(["services", "special\u000acharacters", "notes"])) 

215 $.services["special\ncharacters"].notes 

216 >>> json_path(["custom_array", 2, 0]) 

217 '$.custom_array[2][0]' 

218 

219 """ 

220 

221 def needs_longhand(x: str | int) -> bool: 1ampkqTrUvVW'XBCDEOlsjnoYZt0FGH1IJ2Khg3456LM789!#$deNPwQRxySzuAicfb

222 initial = ( 1ampkqTrUvVW'XBCDEOlsjnoYZt0FGH1IJ2Khg3456LM789!#$deNPwQRxySzuAicfb

223 frozenset(string.ascii_lowercase) 

224 | frozenset(string.ascii_uppercase) 

225 | frozenset("_") 

226 ) 

227 chars = initial | frozenset(string.digits) 1ampkqTrUvVW'XBCDEOlsjnoYZt0FGH1IJ2Khg3456LM789!#$deNPwQRxySzuAicfb

228 return not ( 1ampkqTrUvVW'XBCDEOlsjnoYZt0FGH1IJ2Khg3456LM789!#$deNPwQRxySzuAicfb

229 isinstance(x, str) 

230 and x 

231 and set(x).issubset(chars) 

232 and x[:1] in initial 

233 ) 

234 

235 chunks = ["$"] 1ampkqTrUvVW'XBCDEOlsjnoYZt0FGH1IJ2Khg3456LM789!#$deNPwQRxySzuAicfb

236 chunks.extend( 1ampkqTrUvVW'XBCDEOlsjnoYZt0FGH1IJ2Khg3456LM789!#$deNPwQRxySzuAicfb

237 f"[{json.dumps(x)}]" if needs_longhand(x) else f".{x}" for x in path 

238 ) 

239 return "".join(chunks) 1ampkqTrUvVW'XBCDEOlsjnoYZt0FGH1IJ2Khg3456LM789!#$deNPwQRxySzuAicfb

240 

241 

242class _VaultConfigValidator: 

243 INVALID_CONFIG_ERROR = "vault config is invalid" 

244 

245 def __init__(self, maybe_config: Any) -> None: # noqa: ANN401 

246 self.maybe_config = maybe_config 1ampk*qTrUvVW'XBCDE,OlsjnoYZt0FGH1IJ2K-+h(g3456LM789!#$deNPwQRxyS.zuAicfb

247 

248 def traverse_path(self, path: tuple[str, ...]) -> Any: # noqa: ANN401 

249 obj = self.maybe_config 1ampkqrlsjnohgdeicfb

250 for key in path: 1ampkqrlsjnohgdeicfb

251 obj = obj[key] 1ampkqrlsjnohgdeicfb

252 return obj 1ampkqrlsjnohgdeicfb

253 

254 def walk_subconfigs( 

255 self, 

256 ) -> Generator[tuple[tuple[str] | tuple[str, str], str, Any], None, None]: 

257 obj = cast("dict[str, dict[str, Any]]", self.maybe_config) 1ampk*qTrUvVWXBCDEOlsjnoYZt0FGH1IJ2K+hg3456LM789!#$deNPwQRxySzuAicfb

258 if isinstance(obj.get("global", False), dict): 1ampk*qTrUvVWXBCDEOlsjnoYZt0FGH1IJ2K+hg3456LM789!#$deNPwQRxySzuAicfb

259 for k, v in list(obj["global"].items()): 1amkvBCDEOlsotFGH1IJKhg3456LM789!#$deNPwQRxySzuAicfb

260 yield ("global",), k, v 1amkvBCDEOlsotFGH1IJKhg3456LM789!#$deNPwQRxySzuAicfb

261 for sv_name, sv_obj in list(obj["services"].items()): 1ampk*qTrUvVWXBCDEOlsjnoYZt0FGH1IJ2K+hg3456LM789!#$deNPwQRxySzuAicfb

262 for k, v in list(sv_obj.items()): 1ampkqTrUvVWXBCDEOjnoYZt0FGH1IJ2KhgLMdeNPwQRxySzuAicfb

263 yield ("services", sv_name), k, v 1ampkqTrUvVWXBCDEOjnoYZt0FGH1IJ2KhgLMdeNPwQRxySzuAicfb

264 

265 def validate( # noqa: C901,PLR0912 

266 self, 

267 *, 

268 allow_unknown_settings: bool = False, 

269 ) -> None: 

270 err_obj_not_a_dict = "vault config is not a dict" 1ampk*qTrUvVW'XBCDE,OlsjnoYZt0FGH1IJ2K-+h(g3456LM789!#$deNPwQRxyS.zuAicfb

271 err_non_str_service_name = ( 1ampk*qTrUvVW'XBCDE,OlsjnoYZt0FGH1IJ2K-+h(g3456LM789!#$deNPwQRxyS.zuAicfb

272 "vault config contains non-string service name {sv_name!r}" 

273 ) 

274 err_not_a_dict = "vault config entry {json_path_str} is not a dict" 1ampk*qTrUvVW'XBCDE,OlsjnoYZt0FGH1IJ2K-+h(g3456LM789!#$deNPwQRxyS.zuAicfb

275 err_not_a_string = "vault config entry {json_path_str} is not a string" 1ampk*qTrUvVW'XBCDE,OlsjnoYZt0FGH1IJ2K-+h(g3456LM789!#$deNPwQRxyS.zuAicfb

276 err_not_an_int = "vault config entry {json_path_str} is not an integer" 1ampk*qTrUvVW'XBCDE,OlsjnoYZt0FGH1IJ2K-+h(g3456LM789!#$deNPwQRxyS.zuAicfb

277 err_unknown_setting = ( 1ampk*qTrUvVW'XBCDE,OlsjnoYZt0FGH1IJ2K-+h(g3456LM789!#$deNPwQRxyS.zuAicfb

278 "vault config entry {json_path_str} uses unknown setting {key!r}" 

279 ) 

280 err_bad_number0 = "vault config entry {json_path_str} is negative" 1ampk*qTrUvVW'XBCDE,OlsjnoYZt0FGH1IJ2K-+h(g3456LM789!#$deNPwQRxyS.zuAicfb

281 err_bad_number1 = "vault config entry {json_path_str} is not positive" 1ampk*qTrUvVW'XBCDE,OlsjnoYZt0FGH1IJ2K-+h(g3456LM789!#$deNPwQRxyS.zuAicfb

282 

283 kwargs: dict[str, Any] = { 1ampk*qTrUvVW'XBCDE,OlsjnoYZt0FGH1IJ2K-+h(g3456LM789!#$deNPwQRxyS.zuAicfb

284 "allow_unknown_settings": allow_unknown_settings, 

285 } 

286 if not isinstance(self.maybe_config, dict): 1ampk*qTrUvVW'XBCDE,OlsjnoYZt0FGH1IJ2K-+h(g3456LM789!#$deNPwQRxyS.zuAicfb

287 raise TypeError(err_obj_not_a_dict.format(**kwargs)) 1,-(.c

288 if "global" in self.maybe_config: 1ampk*qTrUvVW'XBCDEOlsjnoYZt0FGH1IJ2K+hg3456LM789!#$deNPwQRxySzuAicfb

289 o_global = self.maybe_config["global"] 1amkv'BCDEOlsotFGH1IJKhg3456LM789!#$deNPwQRxySzuAicfb

290 if not isinstance(o_global, dict): 1amkv'BCDEOlsotFGH1IJKhg3456LM789!#$deNPwQRxySzuAicfb

291 kwargs["json_path_str"] = json_path(["global"]) 1'c

292 raise TypeError(err_not_a_dict.format(**kwargs)) 1'c

293 if not isinstance(self.maybe_config.get("services"), dict): 1ampk*qTrUvVW'XBCDEOlsjnoYZt0FGH1IJ2K+hg3456LM789!#$deNPwQRxySzuAicfb

294 kwargs["json_path_str"] = json_path(["services"]) 1'c

295 raise TypeError(err_not_a_dict.format(**kwargs)) 1'c

296 for sv_name, service in self.maybe_config["services"].items(): 1ampk*qTrUvVWXBCDEOlsjnoYZt0FGH1IJ2K+hg3456LM789!#$deNPwQRxySzuAicfb

297 if not isinstance(sv_name, str): 1ampkqTrUvVWXBCDEOjnoYZt0FGH1IJ2KhgLMdeNPwQRxySzuAicfb

298 kwargs["sv_name"] = sv_name 1cb

299 raise TypeError(err_non_str_service_name.format(**kwargs)) 1cb

300 if not isinstance(service, dict): 1ampkqTrUvVWXBCDEOjnoYZt0FGH1IJ2KhgLMdeNPwQRxySzuAicfb

301 kwargs["json_path_str"] = json_path(["services", sv_name]) 1c

302 raise TypeError(err_not_a_dict.format(**kwargs)) 1c

303 for path, key, value in self.walk_subconfigs(): 1ampk*qTrUvVWXBCDEOlsjnoYZt0FGH1IJ2K+hg3456LM789!#$deNPwQRxySzuAicfb

304 kwargs["path"] = path 1ampkqTrUvVWXBCDEOlsjnoYZt0FGH1IJ2Khg3456LM789!#$deNPwQRxySzuAicfb

305 kwargs["key"] = key 1ampkqTrUvVWXBCDEOlsjnoYZt0FGH1IJ2Khg3456LM789!#$deNPwQRxySzuAicfb

306 kwargs["value"] = value 1ampkqTrUvVWXBCDEOlsjnoYZt0FGH1IJ2Khg3456LM789!#$deNPwQRxySzuAicfb

307 kwargs["json_path_str"] = json_path([*path, key]) 1ampkqTrUvVWXBCDEOlsjnoYZt0FGH1IJ2Khg3456LM789!#$deNPwQRxySzuAicfb

308 # TODO(the-13th-letter): Rewrite using structural pattern 

309 # matching. 

310 # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9 

311 if key in {"key", "phrase"}: 1ampkqTrUvVWXBCDEOlsjnoYZt0FGH1IJ2Khg3456LM789!#$deNPwQRxySzuAicfb

312 if not isinstance(value, str): 1pkvBCDEOlsjntFGH1IJKhg3456LM789!#$deNPwQRxySzuAicfb

313 raise TypeError(err_not_a_string.format(**kwargs)) 1cb

314 elif key == "unicode_normalization_form" and path == ("global",): 1ampkqTrUvVWXBCDEOljnoYZ0FGHIJ2KhgLMdeNPwQRxySzuAicfb

315 if not isinstance(value, str): 1hgdeicfb

316 raise TypeError(err_not_a_string.format(**kwargs)) 1cb

317 if not allow_unknown_settings: 1hgdeicfb

318 raise ValueError(err_unknown_setting.format(**kwargs)) 1cb

319 elif key == "notes" and path != ("global",): 1ampkqTrUvVWXBCDEOljnoYZ0FGHIJ2KhgLMdeNPwQRxySzuAicfb

320 if not isinstance(value, str): 1vOjnhgdePwQRxySzuAicfb

321 raise TypeError(err_not_a_string.format(**kwargs)) 1cb

322 elif key in { 1ampkqTrUvVWXBCDEljoYZ0FGHIJ2KhgLMdeNwxyzuAicfb

323 "length", 

324 "repeat", 

325 "lower", 

326 "upper", 

327 "number", 

328 "space", 

329 "dash", 

330 "symbol", 

331 }: 

332 if not isinstance(value, int): 1ampkqTrUvVWXBCDEljoYZ0FGHIJ2KhgLMdeNwxyzuAicfb

333 raise TypeError(err_not_an_int.format(**kwargs)) 1cb

334 if key == "length" and value < 1: 1ampkqTrUvVWXBCDEljoYZ0FGHIJ2KhgLMdeNwxyzuAicfb

335 raise ValueError(err_bad_number1.format(**kwargs)) 1cb

336 if key != "length" and value < 0: 1ampkqTrUvVWXBCDEljoYZ0FGHIJ2KhgLMdeNwxyzuAicfb

337 raise ValueError(err_bad_number0.format(**kwargs)) 1cb

338 elif not allow_unknown_settings: 1hgdeicfb

339 raise ValueError(err_unknown_setting.format(**kwargs)) 1cb

340 

341 def clean_up_falsy_values(self) -> Generator[CleanupStep, None, None]: # noqa: C901 

342 obj = self.maybe_config 1ampkqrlsjnoth(gdeicfb

343 if ( 1ampkqrlsjnoth(gdeicfb

344 not isinstance(obj, dict) 

345 or "services" not in obj 

346 or not isinstance(obj["services"], dict) 

347 ): 

348 raise ValueError( 1(c

349 self.INVALID_CONFIG_ERROR 

350 ) # pragma: no cover [failsafe] 

351 if "global" in obj and not isinstance(obj["global"], dict): 1ampkqrlsjnothgdeicfb

352 raise ValueError( 1c

353 self.INVALID_CONFIG_ERROR 

354 ) # pragma: no cover [failsafe] 

355 if not all( 1ampkqrlsjnothgdeicfb

356 isinstance(service_obj, dict) 

357 for service_obj in obj["services"].values() 

358 ): 

359 raise ValueError( 1c

360 self.INVALID_CONFIG_ERROR 

361 ) # pragma: no cover [failsafe] 

362 

363 def falsy(value: Any) -> bool: # noqa: ANN401 1ampkqrlsjnothgdeicfb

364 return not js_truthiness(value) 1ampkqrlsjnohgdeicfb

365 

366 def falsy_but_not_zero(value: Any) -> bool: # noqa: ANN401 1ampkqrlsjnothgdeicfb

367 return not js_truthiness(value) and not ( 1ampkljhgdeicfb

368 isinstance(value, int) and value == 0 

369 ) 

370 

371 def falsy_but_not_string(value: Any) -> bool: # noqa: ANN401 1ampkqrlsjnothgdeicfb

372 return not js_truthiness(value) and value != "" # noqa: PLC1901 1pklsjnhgdeicfb

373 

374 for path, key, value in self.walk_subconfigs(): 1ampkqrlsjnothgdeicfb

375 service_obj = self.traverse_path(path) 1ampkqrlsjnohgdeicfb

376 # TODO(the-13th-letter): Rewrite using structural pattern 

377 # matching. 

378 # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9 

379 if key == "phrase" and falsy_but_not_string(value): 1ampkqrlsjnohgdeicfb

380 yield CleanupStep( 1defb

381 (*path, key), service_obj[key], "replace", "" 

382 ) 

383 service_obj[key] = "" 1defb

384 elif key == "notes" and falsy(value): 1ampkqrlsjnohgdeicfb

385 yield CleanupStep( 1jndefb

386 (*path, key), service_obj[key], "remove", None 

387 ) 

388 service_obj.pop(key) 1jndefb

389 elif key == "key" and falsy(value): 1ampkqrlsjnohgdeicfb

390 if path == ("global",): 1lsjndefb

391 yield CleanupStep( 1lsdefb

392 (*path, key), service_obj[key], "remove", None 

393 ) 

394 service_obj.pop(key) 1lsdefb

395 else: 

396 yield CleanupStep( 1jndefb

397 (*path, key), service_obj[key], "replace", "" 

398 ) 

399 service_obj[key] = "" 1jndefb

400 elif key == "length" and falsy(value): 1ampkqrlsjnohgdeicfb

401 yield CleanupStep( 1defb

402 (*path, key), service_obj[key], "replace", 20 

403 ) 

404 service_obj[key] = 20 1defb

405 elif key == "repeat" and falsy_but_not_zero(value): 1ampkqrlsjnohgdeicfb

406 yield CleanupStep((*path, key), service_obj[key], "replace", 0) 1defb

407 service_obj[key] = 0 1defb

408 elif key in { 1ampkqrlsjnohgdeicfb

409 "lower", 

410 "upper", 

411 "number", 

412 "space", 

413 "dash", 

414 "symbol", 

415 } and falsy_but_not_zero(value): 

416 yield CleanupStep( 1defb

417 (*path, key), service_obj[key], "remove", None 

418 ) 

419 service_obj.pop(key) 1defb

420 

421 

422@overload 

423@deprecated( 

424 "allow_derivepassphrase_extensions argument is deprecated since v0.4.0, " 

425 "to be removed in v1.0: no extensions are defined" 

426) 

427def validate_vault_config( 

428 obj: Any, # noqa: ANN401 

429 /, 

430 *, 

431 allow_derivepassphrase_extensions: bool, 

432 allow_unknown_settings: bool = False, 

433) -> None: ... 

434 

435 

436@overload 

437def validate_vault_config( 

438 obj: Any, # noqa: ANN401 

439 /, 

440 *, 

441 allow_unknown_settings: bool = False, 

442) -> None: ... 

443 

444 

445def validate_vault_config( 

446 obj: Any, 

447 /, 

448 *, 

449 allow_unknown_settings: bool = False, 

450 allow_derivepassphrase_extensions: bool = _Omitted(), # type: ignore[assignment] 

451) -> None: 

452 """Check that `obj` is a valid vault config. 

453 

454 Args: 

455 obj: 

456 The object to test. 

457 allow_unknown_settings: 

458 If false, abort on unknown settings. 

459 allow_derivepassphrase_extensions: 

460 (Deprecated.) Ignored since v0.4.0. 

461 

462 Raises: 

463 TypeError: 

464 An entry in the vault config, or the vault config itself, 

465 has the wrong type. 

466 ValueError: 

467 An entry in the vault config is not allowed, or has a 

468 disallowed value. 

469 

470 Warning: Deprecated argument 

471 **v0.4.0**: 

472 The `allow_derivepassphrase_extensions` keyword argument is 

473 deprecated, and will be removed in v1.0. There are no 

474 specified `derivepassphrase` extensions. 

475 

476 """ 

477 # TODO(the-13th-letter): Remove this block in v1.0. 

478 # https://the13thletter.info/derivepassphrase/latest/upgrade-notes/#v1.0-allow-derivepassphrase-extensions 

479 # TODO(the-13th-letter): Add tests that trigger the deprecation warning, 

480 # then include this in coverage. 

481 if not isinstance( 1ampk*qTrUvVW'XBCDE,OlsjnoYZt0FGH1IJ2K-+h(g3456LM789!#$deNPwQRxyS.zuAicfb

482 allow_derivepassphrase_extensions, _Omitted 

483 ): # pragma: no cover [unused] 

484 warnings.warn( 

485 get_overloads(validate_vault_config)[0].__deprecated__, # type: ignore[attr-defined] 

486 DeprecationWarning, 

487 stacklevel=2, 

488 ) 

489 

490 return _VaultConfigValidator(obj).validate( 1ampk*qTrUvVW'XBCDE,OlsjnoYZt0FGH1IJ2K-+h(g3456LM789!#$deNPwQRxyS.zuAicfb

491 allow_unknown_settings=allow_unknown_settings 

492 ) 

493 

494 

495def is_vault_config(obj: Any) -> TypeIs[VaultConfig]: # noqa: ANN401 

496 """Check if `obj` is a valid vault config, according to typing. 

497 

498 Args: 

499 obj: The object to test. 

500 

501 Returns: 

502 True if this is a vault config, false otherwise. 

503 

504 """ # noqa: DOC501 

505 try: 1ampk*qTrUvVW'XBCDE,OlsjnoYZt0FGH1IJ2K-+h(g3456LM789!#$deNPwQRxyS.zuAif

506 validate_vault_config( 1ampk*qTrUvVW'XBCDE,OlsjnoYZt0FGH1IJ2K-+h(g3456LM789!#$deNPwQRxyS.zuAif

507 obj, 

508 allow_unknown_settings=True, 

509 ) 

510 except (TypeError, ValueError) as exc: 1',-(.

511 if "vault config " not in str(exc): # pragma: no cover [failsafe] 1',-(.

512 raise 

513 return False 1',-(.

514 return True 1ampk*qTrUvVWXBCDEOlsjnoYZt0FGH1IJ2K+hg3456LM789!#$deNPwQRxySzuAif

515 

516 

517def js_truthiness(value: Any, /) -> bool: # noqa: ANN401 

518 """Return the truthiness of the value, according to JavaScript/ECMAScript. 

519 

520 Like Python, ECMAScript considers certain values to be false in 

521 a boolean context, and every other value to be true. These 

522 considerations do not agree: ECMAScript considers [`math.nan`][] to 

523 be false too, and empty arrays and objects/dicts to be true, 

524 contrary to Python. Because of these discrepancies, we cannot defer 

525 to [`bool`][] for ECMAScript truthiness checking, and need 

526 a separate, explicit predicate. 

527 

528 (Some falsy values in ECMAScript aren't defined in Python: 

529 `undefined`, and `document.all`. We do not implement support for 

530 those.) 

531 

532 !!! note 

533 

534 We cannot use a simple `value not in falsy_values` check, 

535 because [`math.nan`][] behaves in annoying and obstructive ways. 

536 In general, `float('NaN') == float('NaN')` is false, and 

537 `float('NaN') != math.nan` and `math.nan != math.nan` are true. 

538 CPython says `float('NaN') in [math.nan]` is false, PyPy3 says 

539 it is true. Seemingly the only reliable and portable way to 

540 check for [`math.nan`][] is to use [`math.isnan`][] directly. 

541 

542 Args: 

543 value: The value to test. 

544 

545 """ # noqa: RUF002 

546 try: 1a:mpkqrlsjnothgdeicfb;

547 if value in {None, False, 0, 0.0, ""}: # noqa: B033 1a:mpkqrlsjnothgdeicfb;

548 return False 1a:mkqrlsjnotgdefb

549 except TypeError: 1:cb

550 # All falsy values are hashable, so this can't be falsy. 

551 return True 1:cb

552 return not (isinstance(value, float) and math.isnan(value)) 1a:mpkqrlsjnohgdeicfb;

553 

554 

555class CleanupStep(NamedTuple): 

556 """A single executed step during vault config cleanup. 

557 

558 Attributes: 

559 path: 

560 A sequence of object keys or array indices to navigate to 

561 the JSON value that was cleaned up. 

562 old_value: 

563 The old value. 

564 action: 

565 Either `'replace'` if `old_value` was replaced with 

566 `new_value`, or `'remove'` if `old_value` was removed. 

567 new_value: 

568 The new value. 

569 

570 """ 

571 

572 path: Sequence[str | int] 

573 """""" 

574 old_value: Any 

575 """""" 

576 action: Literal["replace", "remove"] 

577 """""" 

578 new_value: Any 

579 """""" 

580 

581 

582def clean_up_falsy_vault_config_values( 

583 obj: Any, # noqa: ANN401 

584) -> Sequence[CleanupStep] | None: 

585 """Convert falsy values in a vault config to correct types, in-place. 

586 

587 Needed for compatibility with vault(1), which sometimes uses only 

588 truthiness checks. 

589 

590 If vault(1) considered `obj` to be valid, then after clean up, 

591 `obj` will be valid as per [`validate_vault_config`][]. 

592 

593 Args: 

594 obj: 

595 A presumed valid vault configuration save for using falsy 

596 values of the wrong type. 

597 

598 Returns: 

599 A list of 4-tuples `(key_tup, old_value, action, new_value)`, 

600 indicating the cleanup actions performed. `key_tup` is 

601 a sequence of object keys and/or array indices indicating the 

602 JSON path to the leaf value that was cleaned up, `old_value` is 

603 the old value, `new_value` is the new value, and `action` is 

604 either `replace` (`old_value` was replaced with `new_value`) or 

605 `remove` (`old_value` was removed, and `new_value` is 

606 meaningless). 

607 

608 If cleanup was never attempted because of an obviously invalid 

609 vault configuration, then `None` is returned, directly. 

610 

611 """ 

612 try: 1ampkqrlsjnoth(gdeicfb

613 return list(_VaultConfigValidator(obj).clean_up_falsy_values()) 1ampkqrlsjnoth(gdeicfb

614 except ValueError: 1(c

615 return None 1(c

616 

617 

618# TODO(the-13th-letter): Use type variables local to each class. 

619# https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.11 

620T_Buffer = TypeVar("T_Buffer", bound=Buffer) 

621""" 

622A [`TypeVar`][] for classes implementing the [`Buffer`][] interface. 

623 

624Warning: 

625 Non-public attribute, provided for didactical and educational 

626 purposes only. Subject to change without notice, including 

627 removal. 

628 

629""" 

630 

631 

632class SSHKeyCommentPair(NamedTuple, Generic[T_Buffer]): 

633 """SSH key plus comment pair. For typing purposes. 

634 

635 Attributes: 

636 key: SSH key. 

637 comment: SSH key comment. 

638 

639 """ 

640 

641 key: T_Buffer 

642 """""" 

643 comment: T_Buffer 

644 """""" 

645 

646 def toreadonly(self) -> SSHKeyCommentPair[bytes]: 

647 """Return a copy with read-only entries.""" 

648 return SSHKeyCommentPair( 1^

649 key=bytes(self.key), 

650 comment=bytes(self.comment), 

651 ) 

652 

653 

654class SSH_AGENTC(int, enum.Enum): # noqa: N801 

655 """SSH agent protocol numbers: client requests. 

656 

657 Attributes: 

658 REQUEST_IDENTITIES (int): 

659 List identities. Expecting 

660 [`SSH_AGENT.IDENTITIES_ANSWER`][]. 

661 SIGN_REQUEST (int): 

662 Sign data. Expecting [`SSH_AGENT.SIGN_RESPONSE`][]. 

663 ADD_IDENTITY (int): 

664 Add an (SSH2) identity. 

665 REMOVE_IDENTITY (int): 

666 Remove an (SSH2) identity. 

667 ADD_ID_CONSTRAINED (int): 

668 Add an (SSH2) identity, including key constraints. 

669 EXTENSION (int): 

670 Issue a named request that isn't part of the core agent 

671 protocol. Expecting [`SSH_AGENT.EXTENSION_RESPONSE`][] or 

672 [`SSH_AGENT.EXTENSION_FAILURE`][] if the named request is 

673 supported, [`SSH_AGENT.FAILURE`][] otherwise. 

674 

675 """ 

676 

677 REQUEST_IDENTITIES = 11 

678 """""" 

679 SIGN_REQUEST = 13 

680 """""" 

681 ADD_IDENTITY = 17 

682 """""" 

683 REMOVE_IDENTITY = 18 

684 """""" 

685 ADD_ID_CONSTRAINED = 25 

686 """""" 

687 EXTENSION = 27 

688 """""" 

689 

690 def __bytes__(self) -> bytes: # pragma: no cover [unused] 

691 """Return the corresponding request code.""" 

692 return int.to_bytes(self, 1, "big", signed=False) 

693 

694 

695class SSH_AGENT(int, enum.Enum): # noqa: N801 

696 """SSH agent protocol numbers: server replies. 

697 

698 Attributes: 

699 FAILURE (int): 

700 Generic failure code. 

701 SUCCESS (int): 

702 Generic success code. 

703 IDENTITIES_ANSWER (int): 

704 Successful answer to [`SSH_AGENTC.REQUEST_IDENTITIES`][]. 

705 SIGN_RESPONSE (int): 

706 Successful answer to [`SSH_AGENTC.SIGN_REQUEST`][]. 

707 EXTENSION_FAILURE (int): 

708 Unsuccessful answer to [`SSH_AGENTC.EXTENSION`][]. 

709 EXTENSION_RESPONSE (int): 

710 Successful answer to [`SSH_AGENTC.EXTENSION`][]. 

711 

712 """ 

713 

714 FAILURE = 5 

715 """""" 

716 SUCCESS = 6 

717 """""" 

718 IDENTITIES_ANSWER = 12 

719 """""" 

720 SIGN_RESPONSE = 14 

721 """""" 

722 EXTENSION_FAILURE = 28 

723 """""" 

724 EXTENSION_RESPONSE = 29 

725 """""" 

726 

727 def __bytes__(self) -> bytes: # pragma: no cover [unused] 

728 """Return the corresponding response code.""" 

729 return int.to_bytes(self, 1, "big", signed=False) 1a_`{|1}

730 

731 

732class StoreroomKeyPair(NamedTuple, Generic[T_Buffer]): 

733 """A pair of AES256 keys, one for encryption and one for signing. 

734 

735 Attributes: 

736 encryption_key: 

737 AES256 key, used for encryption with AES256-CBC (with PKCS#7 

738 padding). 

739 signing_key: 

740 AES256 key, used for signing with HMAC-SHA256. 

741 

742 """ 

743 

744 encryption_key: T_Buffer 

745 """""" 

746 signing_key: T_Buffer 

747 """""" 

748 

749 def toreadonly(self) -> StoreroomKeyPair[bytes]: 

750 """Return a copy with read-only entries.""" 

751 return StoreroomKeyPair( 1u=?@[

752 encryption_key=bytes(self.encryption_key), 

753 signing_key=bytes(self.signing_key), 

754 ) 

755 

756 

757class StoreroomMasterKeys(NamedTuple, Generic[T_Buffer]): 

758 """A triple of AES256 keys, for encryption, signing and hashing. 

759 

760 Attributes: 

761 hashing_key: 

762 AES256 key, used for hashing with HMAC-SHA256 to derive 

763 a hash table slot for an item. 

764 encryption_key: 

765 AES256 key, used for encryption with AES256-CBC (with PKCS#7 

766 padding). 

767 signing_key: 

768 AES256 key, used for signing with HMAC-SHA256. 

769 

770 """ 

771 

772 hashing_key: T_Buffer 

773 """""" 

774 encryption_key: T_Buffer 

775 """""" 

776 signing_key: T_Buffer 

777 """""" 

778 

779 def toreadonly(self) -> StoreroomMasterKeys[bytes]: 

780 """Return a copy with read-only entries.""" 

781 return StoreroomMasterKeys( 2u ~ ab= ? @ [

782 hashing_key=bytes(self.hashing_key), 

783 encryption_key=bytes(self.encryption_key), 

784 signing_key=bytes(self.signing_key), 

785 ) 

786 

787 

788class FeatureTestEnum(Protocol): 

789 """An [`enum.Enum`][] subclass supporting feature tests. 

790 

791 Each value of the enum supports the `test` method, which tests 

792 whether the feature is enabled or not. (The specific test function 

793 may in general require arguments to actually execute the test, 

794 though no function currently does.) 

795 

796 """ 

797 

798 def test(self, *args: Any, **kwargs: Any) -> bool: ... # noqa: ANN401 

799 

800 

801def _feature_test_function(f: Callable[..., bool]) -> Callable[..., bool]: 

802 """Mark a function as a feature test function. 

803 

804 This decorator exists purely to hold some shared commentary. 

805 

806 Test functions may accept arbitrary arguments, dependent on which 

807 feature they test for. They should make sure to raise no 

808 exceptions. They may use delayed imports to avoid the import cost 

809 until it is actually warranted. 

810 

811 Returns: 

812 The callable. 

813 

814 """ 

815 return f 

816 

817 

818class PEP508Extra(str, enum.Enum): 

819 """PEP 508 extras supported by `derivepassphrase`. 

820 

821 Attributes: 

822 EXPORT: 

823 The necessary dependencies to allow the `export` subcommand 

824 to handle as many foreign configuration formats as possible. 

825 

826 """ 

827 

828 EXPORT = "export" 

829 """""" 

830 

831 @_feature_test_function 

832 @staticmethod 

833 def _test_export() -> bool: 

834 """Return true if [`PEP508Extra.EXPORT`][] is currently supported.""" 

835 import importlib.util # noqa: PLC0415 1%/

836 

837 return importlib.util.find_spec("cryptography") is not None 1%/

838 

839 def test(self, *_args: Any, **_kwargs: Any) -> bool: # noqa: ANN401 

840 """Return true if this extra is enabled.""" 

841 if self == PEP508Extra.EXPORT: 1%/

842 return self._test_export() 1%/

843 return False # pragma: no cover [unused] 

844 

845 __str__ = str.__str__ 

846 __format__ = str.__format__ # type: ignore[assignment] 

847 

848 

849class Feature(str, enum.Enum): 

850 """Optional features supported by `derivepassphrase`. 

851 

852 Attributes: 

853 SSH_KEY: 

854 The `vault` subcommand supports using a master SSH key, 

855 instead of a master passphrase, if an SSH agent is running 

856 and the master SSH key is loaded into it. 

857 

858 This feature requires Python support for the SSH agent's 

859 chosen communication channel technology. 

860 

861 """ 

862 

863 SSH_KEY = "master SSH key" 

864 """""" 

865 

866 @_feature_test_function 

867 @staticmethod 

868 def _test_ssh_key() -> bool: # pragma: no cover [external] 

869 """Return true if [`_types.Feature.SSH_KEY`][] is currently supported. 

870 

871 We test this by attempting to construct an SSH agent client, 

872 reporting whether this can principally work, or not. 

873 

874 """ 

875 # Because BuiltinSSHAgentSocketProvider.test tests the same 

876 # thing, just specific to a certain socket provider, we delegate 

877 # to that test... not only out of laziness, but also out of 

878 # consistency, avoiding two reimplementations of the same logic. 

879 from derivepassphrase import ssh_agent # noqa: PLC0415 1%)

880 

881 for provider_name in ssh_agent.SSHAgentClient.SOCKET_PROVIDERS: 1%)

882 try: 1%)

883 provider = BuiltinSSHAgentSocketProvider(provider_name) 1%)

884 except ValueError: 

885 continue 

886 else: 

887 if provider.test(): 1%)

888 return True 1%)

889 return False 

890 

891 def test(self, *_args: Any, **_kwargs: Any) -> bool: # noqa: ANN401 

892 """Return true if this feature is enabled.""" 

893 if self == Feature.SSH_KEY: 1%)

894 return self._test_ssh_key() 1%)

895 return False # pragma: no cover [unused] 

896 

897 __str__ = str.__str__ 

898 __format__ = str.__format__ # type: ignore[assignment] 

899 

900 

901class DerivationScheme(str, enum.Enum): 

902 """Derivation schemes provided by `derivepassphrase`. 

903 

904 Attributes: 

905 VAULT: 

906 The derivation scheme used by James Coglan's `vault`. 

907 

908 """ 

909 

910 VAULT = "vault" 

911 """""" 

912 

913 def test(self, *_args: Any, **_kwargs: Any) -> bool: # noqa: ANN401 

914 """Return true if this derivation scheme is enabled.""" 

915 return self == DerivationScheme.VAULT 1%]

916 

917 __str__ = str.__str__ 

918 __format__ = str.__format__ # type: ignore[assignment] 

919 

920 

921class ForeignConfigurationFormat(str, enum.Enum): 

922 """Configuration formats supported by `derivepassphrase export`. 

923 

924 Attributes: 

925 VAULT_STOREROOM: 

926 The vault "storeroom" format for the `export vault` 

927 subcommand. 

928 VAULT_V02: 

929 The vault-native "v0.2" format for the `export vault` 

930 subcommand. 

931 VAULT_V03: 

932 The vault-native "v0.3" format for the `export vault` 

933 subcommand. 

934 

935 """ 

936 

937 VAULT_STOREROOM = "vault storeroom" 

938 """""" 

939 

940 @_feature_test_function 

941 @staticmethod 

942 def _test_vault_storeroom() -> bool: 

943 """Return true if [`ForeignConfigurationFormat.VAULT_STOREROOM`][] is currently supported.""" # noqa: E501 

944 from derivepassphrase.exporter import storeroom # noqa: PLC0415 1%/

945 

946 return not storeroom.STUBBED 1%/

947 

948 VAULT_V02 = "vault v0.2" 

949 """""" 

950 VAULT_V03 = "vault v0.3" 

951 """""" 

952 

953 @_feature_test_function 

954 @staticmethod 

955 def _test_vault_v02_v03() -> bool: 

956 """Return true if [`ForeignConfigurationFormat.VAULT_V02`][] and [`ForeignConfigurationFormat.VAULT_V03`][] is currently supported.""" # noqa: E501 

957 from derivepassphrase.exporter import vault_native # noqa: PLC0415 1%/

958 

959 return not vault_native.STUBBED 1%/

960 

961 def test(self, *_args: Any, **_kwargs: Any) -> bool: # noqa: ANN401 

962 """Return true if this foreign configuration format is enabled.""" 

963 if self == ForeignConfigurationFormat.VAULT_STOREROOM: 1%/

964 return self._test_vault_storeroom() 1%/

965 if self in { 1%/

966 ForeignConfigurationFormat.VAULT_V02, 

967 ForeignConfigurationFormat.VAULT_V03, 

968 }: 

969 return self._test_vault_v02_v03() 1%/

970 return False # pragma: no cover [unused] 

971 

972 __str__ = str.__str__ 

973 __format__ = str.__format__ # type: ignore[assignment] 

974 

975 

976class ExportSubcommand(str, enum.Enum): 

977 """Subcommands provided by `derivepassphrase export`. 

978 

979 Attributes: 

980 VAULT: 

981 The `export vault` subcommand. 

982 

983 """ 

984 

985 VAULT = "vault" 

986 """""" 

987 

988 def test(self, *_args: Any, **_kwargs: Any) -> bool: # noqa: ANN401 

989 """Return true if this `export` subcommand is enabled.""" 

990 return self == ExportSubcommand.VAULT 2% bb

991 

992 __str__ = str.__str__ 

993 __format__ = str.__format__ # type: ignore[assignment] 

994 

995 

996class Subcommand(str, enum.Enum): 

997 """Subcommands provided by `derivepassphrase`. 

998 

999 Attributes: 

1000 EXPORT: 

1001 The `export` subcommand. 

1002 VAULT: 

1003 The `vault` subcommand. 

1004 

1005 """ 

1006 

1007 EXPORT = "export" 

1008 """""" 

1009 VAULT = "vault" 

1010 """""" 

1011 

1012 def test(self, *_args: Any, **_kwargs: Any) -> bool: # noqa: ANN401 

1013 """Return true if this subcommand is enabled.""" 

1014 return self in {Subcommand.VAULT, Subcommand.EXPORT} 1%]

1015 

1016 __str__ = str.__str__ 

1017 __format__ = str.__format__ # type: ignore[assignment] 

1018 

1019 

1020class BuiltinSSHAgentSocketProvider(str, enum.Enum): 

1021 """SSH agent socket providers built into `derivepassphrase`. 

1022 

1023 Attributes: 

1024 SSH_AUTH_SOCK_ON_POSIX: 

1025 A socket provider (on POSIX) that queries the 

1026 `SSH_AUTH_SOCK` environment variable. 

1027 SSH_AUTH_SOCK_ON_WINDOWS: 

1028 A socket provider (on The Annoying OS, a.k.a. Microsoft 

1029 Windows) that queries the `SSH_AUTH_SOCK` environment 

1030 variable. 

1031 PAGEANT_ON_WINDOWS: 

1032 A socket provider (on The Annoying OS, a.k.a. Microsoft 

1033 Windows) that connects to Pageant's standard socket. The 

1034 socket address is computed by the socket provider. 

1035 OPENSSH_ON_WINDOWS: 

1036 A socket provider (on The Annoying OS, a.k.a. Microsoft 

1037 Windows) that connects to OpenSSH on Windows's standard 

1038 socket. The socket address is hardcoded by the socket 

1039 provider. 

1040 STUB_AGENT: 

1041 A basic fake agent's socket provider that only reacts to 

1042 known test keys. Used by the test suite. 

1043 STUB_AGENT_WITH_ADDRESS: 

1044 A more orchestratable fake agent's socket provider, compared 

1045 to [`STUB_AGENT`][], that only reacts to known test keys. 

1046 Used by the test suite. 

1047 STUB_AGENT_WITH_ADDRESS_AND_DETERMINISTIC_DSA: 

1048 An elaborate fake agent's socket provider that only reacts 

1049 to known test keys. Used by the test suite. 

1050 SSH_AUTH_SOCK: 

1051 A registry alias for [`SSH_AUTH_SOCK_ON_POSIX`][]. 

1052 UNIX_DOMAIN: 

1053 A registry alias for [`SSH_AUTH_SOCK_ON_POSIX`][]. 

1054 POSIX: 

1055 A registry alias for [`UNIX_DOMAIN`][]. 

1056 WINDOWS_NAMED_PIPE: 

1057 A registry alias for [`SSH_AUTH_SOCK_ON_WINDOWS`][]. 

1058 WINDOWS: 

1059 A registry alias for [`WINDOWS_NAMED_PIPE`][]. 

1060 NATIVE: 

1061 A registry alias for [`WINDOWS`][] if on The Annoying OS, or 

1062 for [`POSIX`][] otherwise. 

1063 

1064 """ 

1065 

1066 SSH_AUTH_SOCK_ON_POSIX = "ssh_auth_sock_on_posix" 

1067 """""" 

1068 SSH_AUTH_SOCK_ON_WINDOWS = "ssh_auth_sock_on_windows" 

1069 """""" 

1070 PAGEANT_ON_WINDOWS = "pageant_on_windows" 

1071 """""" 

1072 OPENSSH_ON_WINDOWS = "openssh_on_windows" 

1073 """""" 

1074 

1075 STUB_AGENT = "stub_agent" 

1076 """""" 

1077 STUB_AGENT_WITH_ADDRESS = "stub_agent_with_address" 

1078 """""" 

1079 STUB_AGENT_WITH_ADDRESS_AND_DETERMINISTIC_DSA = ( 

1080 "stub_agent_with_address_and_deterministic_dsa" 

1081 ) 

1082 """""" 

1083 

1084 SSH_AUTH_SOCK = "ssh_auth_sock" 

1085 """""" 

1086 UNIX_DOMAIN = "unix_domain" 

1087 """""" 

1088 POSIX = "posix" 

1089 """""" 

1090 WINDOWS_NAMED_PIPE = "windows_named_pipe" 

1091 """""" 

1092 WINDOWS = "windows" 

1093 """""" 

1094 NATIVE = "native" 

1095 """""" 

1096 

1097 def test( 

1098 self, 

1099 *_args: Any, # noqa: ANN401 

1100 **_kwargs: Any, # noqa: ANN401 

1101 ) -> bool: # pragma: no cover [external] 

1102 """Return true if this SSH agent socket provider is available. 

1103 

1104 This works by actually attempting to connect to an agent via the 

1105 socket provider. 

1106 

1107 """ 

1108 import sys # noqa: PLC0415 1%)

1109 

1110 import exceptiongroup # noqa: PLC0415 1%)

1111 

1112 from derivepassphrase import ssh_agent # noqa: PLC0415 1%)

1113 

1114 if sys.version_info < (3, 11): 1%)

1115 from exceptiongroup import BaseExceptionGroup # noqa: PLC0415 1%)

1116 

1117 ret = True 1%)

1118 

1119 def handle_notimplementederror( 1%)

1120 _exc: BaseExceptionGroup, 1%)

1121 ) -> None: # pragma: no cover [unused] 1%)

1122 nonlocal ret 

1123 ret = False 1%)

1124 

1125 with exceptiongroup.catch({ # noqa: SIM117 1%)

1126 NotImplementedError: handle_notimplementederror, 1%)

1127 Exception: lambda _exc: None, 1%)

1128 }): 

1129 with ssh_agent.SSHAgentClient.ensure_agent_subcontext(conn=[self]): 1%)

1130 pass 1%)

1131 return ret 1%)

1132 

1133 def is_known_fake_agent(self, *_args: Any, **_kwargs: Any) -> bool: # noqa: ANN401 

1134 """Return true if this SSH agent is a known fake agent.""" 

1135 return self in { 1%)

1136 BuiltinSSHAgentSocketProvider.STUB_AGENT, 

1137 BuiltinSSHAgentSocketProvider.STUB_AGENT_WITH_ADDRESS, 

1138 BuiltinSSHAgentSocketProvider.STUB_AGENT_WITH_ADDRESS_AND_DETERMINISTIC_DSA, 

1139 } 

1140 

1141 __str__ = str.__str__ 

1142 __format__ = str.__format__ # type: ignore[assignment] 

1143 

1144 

1145@runtime_checkable 

1146class SSHAgentSocket(Protocol): 

1147 """An abstract networking socket connected to an SSH agent. 

1148 

1149 The abstract socket supports the [`sendall`][socket.socket.sendall] 

1150 and a [`recv`][socket.socket.recv] operation, with the same 

1151 signatures and semantics as for "real" sockets. The abstract socket 

1152 also supports use as a context manager, for automatically closing 

1153 the socket upon exiting the context. 

1154 

1155 """ 

1156 

1157 def __enter__(self) -> Any: ... # noqa: ANN401 

1158 

1159 # mypy/typeshed has a *very* lax annotation of 

1160 # socket.socket.__exit__, which we need to be compatible with. 

1161 # *sigh* 

1162 def __exit__( 

1163 self, 

1164 *args: object, 

1165 ) -> bool | None: ... 

1166 

1167 def sendall(self, data: Buffer, flags: int = 0, /) -> None: ... 

1168 

1169 def recv(self, bufsize: int, flags: int = 0, /) -> bytes: ... 

1170 

1171 

1172SSHAgentSocketProvider: TypeAlias = "Callable[[], SSHAgentSocket]" 

1173"""A callable that provides an SSH agent socket.""" 

1174 

1175 

1176class SSHAgentSocketProviderEntry(NamedTuple): 

1177 """Registry information for the table of SSH agent socket providers. 

1178 

1179 Third-party developers can register new socket providers for 

1180 auto-discovery by setting up an entry point named 

1181 [`derivepassphrase.ssh_agent_socket_providers`][derivepassphrase.ssh_agent.socketprovider.SocketProvider.ENTRY_POINT_GROUP_NAME], 

1182 referencing an instance of this class. Upon startup of the `vault` 

1183 subsystem, `derivepassphrase` will then add appropriate entries to 

1184 the registry. 

1185 

1186 Attributes: 

1187 provider: The callable that provides the socket. 

1188 key: The table key which this entry is registered as. 

1189 aliases: Other keys that shall point to this entry's key. 

1190 

1191 Note: 

1192 The socket provider registry table uses the key as the key, and 

1193 the provider as the value. It does not store this info object 

1194 directly. 

1195 

1196 """ 

1197 

1198 provider: SSHAgentSocketProvider 

1199 """""" 

1200 key: str 

1201 """""" 

1202 aliases: tuple[str, ...]