Coverage for tests / machinery / __init__.py: 100.000%

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

2# 

3# SPDX-License-Identifier: Zlib 

4 

5"""Testing machinery for `derivepassphrase`. 

6 

7This is all the standalone, non-test-system-specific machinery used to 

8test `derivepassphrase`; this includes global settings, external API 

9wrappers, stub systems, and the like. 

10 

11""" 

12 

13from __future__ import annotations (empty)

14 

15import collections (empty)

16import contextlib (empty)

17import errno (empty)

18import logging (empty)

19import os (empty)

20import re (empty)

21import sys (empty)

22from typing import TYPE_CHECKING, TypedDict (empty)

23 

24import click.testing (empty)

25from typing_extensions import NamedTuple (empty)

26 

27from derivepassphrase import _types, cli, ssh_agent, vault (empty)

28from derivepassphrase.ssh_agent import socketprovider (empty)

29from tests import data (empty)

30 

31__all__ = () (empty)

32 

33if TYPE_CHECKING: (empty)

34 import types 

35 from collections.abc import ( 

36 Callable, 

37 Generator, 

38 Iterable, 

39 Mapping, 

40 Sequence, 

41 ) 

42 from contextlib import AbstractContextManager 

43 from typing import IO, Literal, NotRequired 

44 

45 from typing_extensions import Any, Buffer, Self 

46 

47 

48# Test suite settings 

49# =================== 

50 

51MIN_CONCURRENCY = 4 (empty)

52""" (empty)

53The minimum amount of concurrent threads used for testing. 

54""" 

55 

56 

57def get_concurrency_limit() -> int: (empty)

58 """Return the imposed limit on the number of concurrent threads. 

59 

60 We use [`os.process_cpu_count`][] as the limit on Python 3.13 and 

61 higher, and [`os.cpu_count`][] on Python 3.12 and below. On 

62 Python 3.12 and below, we explicitly support the `PYTHON_CPU_COUNT` 

63 environment variable. We guarantee at least [`MIN_CONCURRENCY`][] 

64 many threads in any case. 

65 

66 """ # noqa: RUF002 

67 result: int | None = None 2 ctx1aF

68 if sys.version_info >= (3, 13): 2 ctx1aF

69 result = os.process_cpu_count() 2 ctx1aF

70 else: 

71 with contextlib.suppress(KeyError, ValueError): 2 ctx1aF

72 result = result or int(os.environ["PYTHON_CPU_COUNT"], 10) 2 ctx1aF

73 with contextlib.suppress(AttributeError): 2 ctx1aF

74 result = result or len(os.sched_getaffinity(os.getpid())) 2 ctx1aF

75 return max(result if result is not None else 0, MIN_CONCURRENCY) 2 ctx1aF

76 

77 

78# Log/Error message searching 

79# =========================== 

80 

81 

82def message_emitted_factory( (empty)

83 level: int, 

84 *, 

85 logger_name: str = cli.PROG_NAME, 

86) -> Callable[[str | re.Pattern[str], Sequence[tuple[str, int, str]]], bool]: 

87 """Return a function to test if a matching message was emitted. 

88 

89 Args: 

90 level: The level to match messages at. 

91 logger_name: The name of the logger to match against. 

92 

93 """ 

94 

95 def message_emitted( (empty)

96 text: str | re.Pattern[str], 

97 record_tuples: Sequence[tuple[str, int, str]], 

98 ) -> bool: 

99 """Return true if a matching message was emitted. 

100 

101 Args: 

102 text: Substring or pattern to match against. 

103 record_tuples: Items to match. 

104 

105 """ 

106 

107 def check_record(record: tuple[str, int, str]) -> bool: 25 ctx1OzPAQRBKLSTMNUVWXYZCGHDEI

108 if record[:2] != (logger_name, level): 23 ctx1OzPAQRBKLSTUVWXYZCGHDEI

109 return False 5 ctx1KLCDE

110 if isinstance(text, str): 23 ctx1OzPAQRBKLSTUVWXYZCGHDEI

111 return text in record[2] 23 ctx1OzPAQRBKLSTUVWXYZCGHDEI

112 return text.match(record[2]) is not None # pragma: no cover 

113 

114 return any(map(check_record, record_tuples)) 25 ctx1OzPAQRBKLSTMNUVWXYZCGHDEI

115 

116 return message_emitted (empty)

117 

118 

119# No need to assert debug messages as of yet. 

120info_emitted = message_emitted_factory(logging.INFO) (empty)

121warning_emitted = message_emitted_factory(logging.WARNING) (empty)

122deprecation_warning_emitted = message_emitted_factory( (empty)

123 logging.WARNING, logger_name=f"{cli.PROG_NAME}.deprecation" 

124) 

125deprecation_info_emitted = message_emitted_factory( (empty)

126 logging.INFO, logger_name=f"{cli.PROG_NAME}.deprecation" 

127) 

128error_emitted = message_emitted_factory(logging.ERROR) (empty)

129 

130 

131# click.testing.CliRunner handling 

132# ================================ 

133 

134 

135class ReadableResult(NamedTuple): (empty)

136 """Helper class for formatting and testing click.testing.Result objects.""" 

137 

138 exception: BaseException | None (empty)

139 exit_code: int (empty)

140 stdout: str (empty)

141 stderr: str (empty)

142 

143 def clean_exit( (empty)

144 self, *, output: str = "", empty_stderr: bool = False 

145 ) -> bool: 

146 """Return whether the invocation exited cleanly. 

147 

148 Args: 

149 output: 

150 An expected output string. 

151 

152 """ 

153 return ( 51 ctx2a F O ] ^ _ ` { | P Q R K L } 3 4 ~ abbbcbdbebfbS 0 gbhbs c n T ibjbkblbU mbV W nbX Y 5 @ obZ pbqbrbi

154 ( 

155 not self.exception 

156 or ( 

157 isinstance(self.exception, SystemExit) 

158 and self.exit_code == 0 

159 ) 

160 ) 

161 and (not output or output in self.stdout) 

162 and (not empty_stderr or not self.stderr) 

163 ) 

164 

165 def error_exit( (empty)

166 self, 

167 *, 

168 error: str | re.Pattern[str] | type[BaseException] = BaseException, 

169 record_tuples: Sequence[tuple[str, int, str]] = (), 

170 ) -> bool: 

171 """Return whether the invocation exited uncleanly. 

172 

173 Args: 

174 error: 

175 An expected error message, or an expected numeric error 

176 code, or an expected exception type. 

177 

178 """ 

179 

180 def error_match(error: str | re.Pattern[str], line: str) -> bool: 41 ctx16z78AB349!#$0%'(M)N*+,mo-./p:;qr=5CGHDE?I

181 return ( 32 ctx16z78AB9!#$0%'(M)N*+,mo-./p:;qr=?

182 error in line 

183 if isinstance(error, str) 

184 else error.match(line) is not None 

185 ) 

186 

187 # TODO(the-13th-letter): Rewrite using structural pattern matching. 

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

189 if isinstance(error, type): 41 ctx16z78AB349!#$0%'(M)N*+,mo-./p:;qr=5CGHDE?I

190 return isinstance(self.exception, error) 2 ctx134

191 else: # noqa: RET505 

192 assert isinstance(error, (str, re.Pattern)) 39 ctx16z78AB9!#$0%'(M)N*+,mo-./p:;qr=5CGHDE?I

193 return ( 39 ctx16z78AB9!#$0%'(M)N*+,mo-./p:;qr=5CGHDE?I

194 isinstance(self.exception, SystemExit) 

195 and self.exit_code > 0 

196 and ( 

197 not error 

198 or any( 

199 error_match(error, line) 

200 for line in self.stderr.splitlines(True) 

201 ) 

202 or error_emitted(error, record_tuples) 

203 ) 

204 ) 

205 

206 

207class CliRunner: (empty)

208 """An abstracted CLI runner class. 

209 

210 Intended to provide similar functionality and scope as the 

211 [`click.testing.CliRunner`][] class, though not necessarily 

212 `click`-specific. Also allows for seamless migration away from 

213 `click`, if/when we decide this. 

214 

215 """ 

216 

217 _SUPPORTS_MIX_STDERR_ATTRIBUTE = not hasattr(click.testing, "StreamMixer") (empty)

218 """ (empty)

219 True if and only if [`click.testing.CliRunner`][] supports the 

220 `mix_stderr` attribute. It was removed in 8.2.0 in favor of the 

221 `click.testing.StreamMixer` class. 

222 

223 See also 

224 [`pallets/click#2523`](https://github.com/pallets/click/pull/2523). 

225 """ 

226 

227 def __init__( (empty)

228 self, 

229 *, 

230 mix_stderr: bool = False, 

231 color: bool | None = None, 

232 ) -> None: 

233 self.color = color 111 ctx2a F 6 O z 7 8 sb] ^ _ ` { | tbubvbP wbA Q R B K L xbybzbAbBbCbDb} 3 4 Eb~ abbbcbFbdbebfbS 9 ! # $ 0 gbhbs % c n T ibjb' ( M ) kblbN * + U , m o - . / p : ; q r mbV W nb= X Y 5 @ obZ C G H D E pbqbrbGbHbIbJbKbLbMbNb? I Obi

234 self.mix_stderr = mix_stderr 111 ctx2a F 6 O z 7 8 sb] ^ _ ` { | tbubvbP wbA Q R B K L xbybzbAbBbCbDb} 3 4 Eb~ abbbcbFbdbebfbS 9 ! # $ 0 gbhbs % c n T ibjb' ( M ) kblbN * + U , m o - . / p : ; q r mbV W nb= X Y 5 @ obZ C G H D E pbqbrbGbHbIbJbKbLbMbNb? I Obi

235 

236 class MixStderrAttribute(TypedDict): 111 ctx2a F 6 O z 7 8 sb] ^ _ ` { | tbubvbP wbA Q R B K L xbybzbAbBbCbDb} 3 4 Eb~ abbbcbFbdbebfbS 9 ! # $ 0 gbhbs % c n T ibjb' ( M ) kblbN * + U , m o - . / p : ; q r mbV W nb= X Y 5 @ obZ C G H D E pbqbrbGbHbIbJbKbLbMbNb? I Obi

237 mix_stderr: NotRequired[bool] 111 ctx2a F 6 O z 7 8 sb] ^ _ ` { | tbubvbP wbA Q R B K L xbybzbAbBbCbDb} 3 4 Eb~ abbbcbFbdbebfbS 9 ! # $ 0 gbhbs % c n T ibjb' ( M ) kblbN * + U , m o - . / p : ; q r mbV W nb= X Y 5 @ obZ C G H D E pbqbrbGbHbIbJbKbLbMbNb? I Obi

238 

239 mix_stderr_args: MixStderrAttribute = ( 111 ctx2a F 6 O z 7 8 sb] ^ _ ` { | tbubvbP wbA Q R B K L xbybzbAbBbCbDb} 3 4 Eb~ abbbcbFbdbebfbS 9 ! # $ 0 gbhbs % c n T ibjb' ( M ) kblbN * + U , m o - . / p : ; q r mbV W nb= X Y 5 @ obZ C G H D E pbqbrbGbHbIbJbKbLbMbNb? I Obi

240 {"mix_stderr": mix_stderr} 

241 if self._SUPPORTS_MIX_STDERR_ATTRIBUTE 

242 else {} 

243 ) 

244 # The "mix_stderr" argument, mandatory for our use case, was 

245 # removed in click 8.2.0 without a transition period. Since we 

246 # cannot branch on the click version available to the type 

247 # checker, we disable static type checking for this call in 

248 # particular; our test suite will have to uncover any breakage 

249 # dynamically instead. 

250 self.click_testing_clirunner = click.testing.CliRunner( # type: ignore[misc] 111 ctx2a F 6 O z 7 8 sb] ^ _ ` { | tbubvbP wbA Q R B K L xbybzbAbBbCbDb} 3 4 Eb~ abbbcbFbdbebfbS 9 ! # $ 0 gbhbs % c n T ibjb' ( M ) kblbN * + U , m o - . / p : ; q r mbV W nb= X Y 5 @ obZ C G H D E pbqbrbGbHbIbJbKbLbMbNb? I Obi

251 **mix_stderr_args 

252 ) 

253 

254 def invoke( (empty)

255 self, 

256 # The click.BaseCommand abstract base class, previously the base 

257 # class of click.Command and click.Group, was removed in click 

258 # 8.2.0 without a transition period, and click.Command instated 

259 # as the common base class instead. To keep some degree of 

260 # compatibility with both old click and new click, we explicitly 

261 # list the (somewhat) concrete base classes we actually care 

262 # about here. 

263 cli: click.Command | click.Group, 

264 args: Sequence[str] | str | None = None, 

265 input: str | bytes | IO[Any] | None = None, 

266 env: Mapping[str, str | None] | None = None, 

267 catch_exceptions: bool = True, 

268 color: bool | None = None, 

269 **extra: Any, 

270 ) -> ReadableResult: 

271 if color is None: # pragma: no cover 89 ctx2a F 6 O z 7 8 sb] ^ _ ` { | P A Q R B K L } 3 4 ~ abbbcbdbebfbS 9 ! # $ 0 gbhbs % c n T ibjb' ( M ) kblbN * + U , m o - . / p : ; q r mbV W nb= X Y 5 @ obZ C G H D E pbqbrb? I i

272 color = self.color if self.color is not None else False 88 ctx2a F 6 O z 7 8 ] ^ _ ` { | P A Q R B K L } 3 4 ~ abbbcbdbebfbS 9 ! # $ 0 gbhbs % c n T ibjb' ( M ) kblbN * + U , m o - . / p : ; q r mbV W nb= X Y 5 @ obZ C G H D E pbqbrb? I i

273 raw_result = self.click_testing_clirunner.invoke( 89 ctx2a F 6 O z 7 8 sb] ^ _ ` { | P A Q R B K L } 3 4 ~ abbbcbdbebfbS 9 ! # $ 0 gbhbs % c n T ibjb' ( M ) kblbN * + U , m o - . / p : ; q r mbV W nb= X Y 5 @ obZ C G H D E pbqbrb? I i

274 cli, 

275 args=args, 

276 input=input, 

277 env=env, 

278 catch_exceptions=catch_exceptions, 

279 color=color, 

280 **extra, 

281 ) 

282 # In 8.2.0, r.stdout is no longer a property aliasing the 

283 # `output` attribute, but rather the raw stdout value. 

284 try: 89 ctx2a F 6 O z 7 8 sb] ^ _ ` { | P A Q R B K L } 3 4 ~ abbbcbdbebfbS 9 ! # $ 0 gbhbs % c n T ibjb' ( M ) kblbN * + U , m o - . / p : ; q r mbV W nb= X Y 5 @ obZ C G H D E pbqbrb? I i

285 stderr = raw_result.stderr 89 ctx2a F 6 O z 7 8 sb] ^ _ ` { | P A Q R B K L } 3 4 ~ abbbcbdbebfbS 9 ! # $ 0 gbhbs % c n T ibjb' ( M ) kblbN * + U , m o - . / p : ; q r mbV W nb= X Y 5 @ obZ C G H D E pbqbrb? I i

286 except ValueError: 4 ctx134@i

287 stderr = raw_result.stdout 4 ctx134@i

288 return ReadableResult( 89 ctx2a F 6 O z 7 8 sb] ^ _ ` { | P A Q R B K L } 3 4 ~ abbbcbdbebfbS 9 ! # $ 0 gbhbs % c n T ibjb' ( M ) kblbN * + U , m o - . / p : ; q r mbV W nb= X Y 5 @ obZ C G H D E pbqbrb? I i

289 raw_result.exception, 

290 raw_result.exit_code, 

291 (raw_result.stdout if not self.mix_stderr else raw_result.output) 

292 or "", 

293 stderr or "", 

294 ) 

295 

296 def isolated_filesystem( (empty)

297 self, 

298 temp_dir: str | os.PathLike[str] | None = None, 

299 ) -> AbstractContextManager[str]: 

300 return self.click_testing_clirunner.isolated_filesystem( 107 ctx2F 6 O z 7 8 sb] ^ _ ` { | tbubvbP wbA Q R B K L xbybzbAbBbCbDb} Eb~ abbbcbFbdbebfbS 9 ! # $ 0 gbhbs % c n T ibjb' ( M ) kblbN * + U , m o - . / p : ; q r mbV W nb= X Y 5 @ obZ C G H D E pbqbrbGbHbIbJbKbLbMbNb? I Ob

301 temp_dir=temp_dir 

302 ) 

303 

304 

305# Stubbed SSH agent request/response queue 

306# ======================================== 

307 

308 

309class AgentProtocolResponseQueue: (empty)

310 """A response queue for the SSH agent protocol. 

311 

312 This queue wraps a map of request messages 

313 ([`AgentProtocolRequest`][data.AgentProtocolRequest]) to multiple 

314 response messages 

315 ([`AgentProtocolResponse`][data.AgentProtocolResponse]). Whenever 

316 [`_send_request_message`][] is called with a payload that matches 

317 a [`AgentProtocolRequest`][data.AgentProtocolRequest] from the map, 

318 the corresponding 

319 [`AgentProtocolResponse`][data.AgentProtocolResponse] objects are 

320 added to the queue, to be read by subsequent 

321 [`_get_response_pair`][] calls. The [`_send_request_message`][] and 

322 [`_get_response_pair`][] methods are function- and 

323 interface-compatible with the internal methods of the same name in 

324 [`ssh_agent.SSHAgentClient`][]. 

325 

326 The queue is a context manager. When entered or exited, it 

327 verifies that the queue is empty. 

328 

329 Use this agent stub implementation if you are modelling well-formed 

330 requests and responses, without network errors or protocol framing 

331 errors. For modelling the latter, see [`StubbedSSHAgentSocket`][]. 

332 

333 """ 

334 

335 def __init__( (empty)

336 self, 

337 request_responses_mapping: Mapping[ 

338 data.AgentProtocolRequest, 

339 Sequence[data.AgentProtocolResponse], 

340 ], 

341 ) -> None: 

342 self.responses_map = request_responses_mapping 4 ctx1tuvw

343 self.queue: collections.deque[data.AgentProtocolResponse] = ( 4 ctx1tuvw

344 collections.deque() 

345 ) 

346 

347 def __bool__(self) -> bool: (empty)

348 return bool(self.queue) 4 ctx1tuvw

349 

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

351 assert not self, ( 4 ctx1tuvw

352 f"Agent I/O queue already has contents: {self.queue!r}" 

353 ) 

354 return self 4 ctx1tuvw

355 

356 def __exit__( (empty)

357 self, 

358 exc_type: type[BaseException] | None, 

359 exc_value: BaseException | None, 

360 exc_tb: types.TracebackType | None, 

361 ) -> Literal[False]: 

362 if (exc_type, exc_value, exc_tb) != ( 4 ctx1tuvw

363 None, 

364 None, 

365 None, 

366 ): # pragma: no cover [external] 

367 return False 

368 assert not self, f"Agent I/O queue still has contents: {self.queue!r}" 4 ctx1tuvw

369 return False 4 ctx1tuvw

370 

371 def _send_request_message( (empty)

372 self, 

373 request_code: int | _types.SSH_AGENTC, 

374 payload: bytes | bytearray, 

375 ) -> None: 

376 req = data.AgentProtocolRequest(request_code, bytes(payload)) 4 ctx1tuvw

377 assert req in self.responses_map, ( 4 ctx1tuvw

378 f"No prepared response for the request {req!r}" 

379 ) 

380 self.queue.extend(self.responses_map[req]) 4 ctx1tuvw

381 

382 def _get_response_pair( (empty)

383 self, 

384 ) -> tuple[int | _types.SSH_AGENT, bytes | bytearray]: 

385 assert self.queue, "No more responses left to replay" 4 ctx1tuvw

386 response = self.queue.popleft() 4 ctx1tuvw

387 return response.code, response.payload 4 ctx1tuvw

388 

389 

390# Stubbed SSH agent socket 

391# ======================== 

392 

393# Base variant 

394# ------------ 

395 

396 

397@socketprovider.SocketProvider.register( (empty)

398 _types.BuiltinSSHAgentSocketProvider.STUB_AGENT 

399) 

400class StubbedSSHAgentSocket: (empty)

401 """A stubbed SSH agent presenting an [`_types.SSHAgentSocket`][]. 

402 

403 On the network protocol side, the agent implements the full 

404 [`_types.SSHAgentSocket`][] interface, including pipelined and 

405 unaligned agent requests. However, on the application side, the 

406 agent is intrinsically tied to [the set of SSH test 

407 keys][data.ALL_KEYS], and only gives meaningful answers for 

408 operations on the test keys and for agent operations in use by 

409 [`ssh_agent.SSHAgentClient`][]. The agent does not actually 

410 implement any cryptography; all cryptography-related answers are 

411 derived from the recorded test key data. 

412 

413 It is not safe to further monkeypatch the agent's [`recv`][] or 

414 [`sendall`][] methods on their own: either monkeypatch both of them, 

415 or manipulate the [`send_to_client`][] bytes queue directly instead. 

416 Given an [`ssh_agent.SSHAgentClient`][] connected to 

417 a [`StubbedSSHAgentSocket`][], if the test ensures proper message 

418 serialization and protocol framing and if the monkeypatching can be 

419 expressed in terms of full request messages and full response 

420 messages, prefer using a [`AgentProtocolResponseQueue`][] to 

421 monkeypatch the client's high-level request/response-loop instead of 

422 monkeypatching the low-level socket communication in this agent 

423 socket. 

424 

425 """ 

426 

427 _NO_FLAG_SUPPORT = "This stubbed SSH agent socket does not support flags." (empty)

428 _PROTOCOL_VIOLATION = "SSH agent protocol violation." (empty)

429 _INVALID_REQUEST = "Invalid request." (empty)

430 _UNSUPPORTED_REQUEST = "Unsupported request." (empty)

431 _INCOMPLETE_REQUEST = "The last request was incomplete." (empty)

432 

433 HEADER_SIZE = 4 (empty)

434 CODE_SIZE = 1 (empty)

435 

436 KNOWN_EXTENSIONS = frozenset({ (empty)

437 "query", 

438 "list-extended@putty.projects.tartarus.org", 

439 }) 

440 """Known and implemented protocol extensions.""" (empty)

441 

442 def __init__(self, *extensions: str) -> None: (empty)

443 """Initialize the agent.""" 

444 self.send_to_client = bytearray() 27 ctx1a[J1dbjxhkygfscnmopqrtuvlw2

445 """ 27 ctx1a[J1dbjxhkygfscnmopqrtuvlw2

446 The buffered response to the client, read piecemeal by [`recv`][]. 

447 """ 

448 self.receive_from_client = bytearray() 27 ctx1a[J1dbjxhkygfscnmopqrtuvlw2

449 """The last request issued by the client.""" 27 ctx1a[J1dbjxhkygfscnmopqrtuvlw2

450 self.closed = False 27 ctx1a[J1dbjxhkygfscnmopqrtuvlw2

451 """True if the connection is closed, false otherwise.""" 27 ctx1a[J1dbjxhkygfscnmopqrtuvlw2

452 self.enabled_extensions = frozenset(extensions) & self.KNOWN_EXTENSIONS 27 ctx1a[J1dbjxhkygfscnmopqrtuvlw2

453 """ 27 ctx1a[J1dbjxhkygfscnmopqrtuvlw2

454 Extensions actually enabled in this particular stubbed SSH agent. 

455 """ 

456 self.try_rfc6979 = False 27 ctx1a[J1dbjxhkygfscnmopqrtuvlw2

457 """ 27 ctx1a[J1dbjxhkygfscnmopqrtuvlw2

458 Attempt to issue DSA and ECDSA signatures according to RFC 6979? 

459 """ 

460 self.try_pageant_068_080 = False 27 ctx1a[J1dbjxhkygfscnmopqrtuvlw2

461 """ 27 ctx1a[J1dbjxhkygfscnmopqrtuvlw2

462 Attempt to issue DSA and ECDSA signatures as per Pageant 0.68–0.80? 

463 """ # noqa: RUF001 

464 

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

466 """Return self.""" 

467 return self 23 ctx1a[J1dbjhkygfscnpqrtuvlw

468 

469 def __exit__(self, *args: object) -> None: (empty)

470 """Mark the agent's socket as closed. 

471 

472 Raises: 

473 AssertionError: 

474 The last request to the agent was incomplete. 

475 

476 """ 

477 self.closed = True 23 ctx1a[J1dbjhkygfscnpqrtuvlw

478 assert not self.receive_from_client, self._INCOMPLETE_REQUEST 23 ctx1a[J1dbjhkygfscnpqrtuvlw

479 

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

481 """Send data to the SSH agent. 

482 

483 The signature, and behavior, is identical to 

484 [`socket.socket.sendall`][]. Upon successful sending, this 

485 agent will parse the request, call the appropriate handler, and 

486 buffer the result such that it can be read via [`recv`][], in 

487 accordance with the SSH agent protocol. 

488 

489 Args: 

490 data: Binary data to send to the agent. 

491 flags: Reserved. Must be 0. 

492 

493 Returns: 

494 Nothing. The result should be requested via [`recv`][], and 

495 interpreted in accordance with the SSH agent protocol. 

496 

497 Raises: 

498 AssertionError: 

499 The flags argument, if specified, must be 0. 

500 OSError: 

501 The socket connection is already closed. 

502 

503 """ 

504 assert not flags, self._NO_FLAG_SUPPORT 14 ctx1aJdbjhkygfceil

505 self._check_for_io_on_closed_connection() 14 ctx1aJdbjhkygfceil

506 self.receive_from_client.extend(memoryview(data)) 13 ctx1adbjhkygfceil

507 while self.receive_from_client: 13 ctx1adbjhkygfceil

508 result: Buffer | Iterable[int] 

509 if len(self.receive_from_client) < self.HEADER_SIZE: 13 ctx1adbjhkygfceil

510 break 1 ctx1k

511 count = int.from_bytes( 13 ctx1adbjhkygfceil

512 self.receive_from_client[: self.HEADER_SIZE], 

513 "big", 

514 signed=False, 

515 ) 

516 if count: 13 ctx1adbjhkygfceil

517 code = int.from_bytes( 13 ctx1adbjhkygfceil

518 self.receive_from_client[ 

519 self.HEADER_SIZE : self.HEADER_SIZE + self.CODE_SIZE 

520 ], 

521 "big", 

522 signed=False, 

523 ) 

524 request = bytes( 13 ctx1adbjhkygfceil

525 self.receive_from_client[: self.HEADER_SIZE + count] 

526 ) 

527 if len(request) < self.HEADER_SIZE + count: 13 ctx1adbjhkygfceil

528 break 2 ctx1ky

529 request_payload = request[self.HEADER_SIZE + self.CODE_SIZE :] 12 ctx1adbjhkgfceil

530 

531 if code == _types.SSH_AGENTC.REQUEST_IDENTITIES: 12 ctx1adbjhkgfceil

532 result = self.request_identities(list_extended=False) 5 ctx1abfel

533 elif code == _types.SSH_AGENTC.SIGN_REQUEST: 10 ctx1adbjhkgcei

534 result = self.sign(request_payload) 4 ctx1jhge

535 elif self._check_for_extension(code, "query"): 8 ctx1adbhkcei

536 result = self.query_extensions() 4 ctx1dkei

537 elif self._check_for_extension( 6 ctx1adbhci

538 code, "list-extended@putty.projects.tartarus.org" 

539 ): 

540 result = self.request_identities(list_extended=True) 1 ctx1b

541 else: 

542 result = self._failure() 5 ctx1adhci

543 else: 

544 request = bytes(self.receive_from_client[: self.HEADER_SIZE]) 1 ctx1h

545 result = self._failure() 1 ctx1h

546 self.send_to_client.extend( 12 ctx1adbjhkgfceil

547 ssh_agent.SSHAgentClient.string(bytes(result)) 

548 ) 

549 self.receive_from_client[: len(request)] = b"" 12 ctx1adbjhkgfceil

550 

551 def recv(self, count: int, flags: int = 0, /) -> bytes: (empty)

552 """Read data from the SSH agent. 

553 

554 As per the SSH agent protocol, data is only available to be read 

555 immediately after a request via [`sendall`][] and if the socket 

556 connection is still open. Calls to [`recv`][] at other points 

557 in time that attempt to read data violate the protocol, and will 

558 fail. (A [`recv`][] of zero bytes does not read data.) Calls 

559 to [`recv`][] when the socket connection is closed always fail. 

560 

561 Args: 

562 count: 

563 Number of bytes to read from the agent. 

564 flags: 

565 Reserved. Must be 0. 

566 

567 Returns: 

568 (A chunk of) the SSH agent's response to the most recent 

569 request. If reading 0 bytes, the returned chunk is always 

570 an empty byte string. 

571 

572 Raises: 

573 AssertionError: 

574 The flags argument, if specified, must be 0. 

575 

576 Alternatively, `recv` was called when there was no 

577 response to be obtained, in violation of the SSH agent 

578 protocol. 

579 OSError: 

580 The socket connection is already closed. 

581 

582 """ 

583 assert not flags, self._NO_FLAG_SUPPORT 15 ctx1aJ1dbjhkgfceil2

584 self._check_for_io_on_closed_connection() 15 ctx1aJ1dbjhkgfceil2

585 assert not count or self.send_to_client, self._PROTOCOL_VIOLATION 14 ctx1a1dbjhkgfceil2

586 ret = bytes(self.send_to_client[:count]) 13 ctx1adbjhkgfceil2

587 del self.send_to_client[:count] 13 ctx1adbjhkgfceil2

588 return ret 13 ctx1adbjhkgfceil2

589 

590 def _failure(self) -> bytes: (empty)

591 return bytes(_types.SSH_AGENT.FAILURE) 6 ctx1adhgci

592 

593 def _check_for_io_on_closed_connection(self) -> None: (empty)

594 if self.closed: 16 ctx1aJ1dbjhkygfceil2

595 raise OSError(errno.EBADF, os.strerror(errno.EBADF)) 1 ctx1J

596 

597 def _check_for_extension(self, code: int, extension: str) -> bool: (empty)

598 if ( 8 ctx1adbhkcei

599 extension not in self.enabled_extensions 

600 or code != _types.SSH_AGENTC.EXTENSION 

601 ): 

602 return False 5 ctx1adhci

603 string = ssh_agent.SSHAgentClient.string 5 ctx1dbkei

604 extension_marker = b"\x1b" + string(extension.encode("ascii")) 5 ctx1dbkei

605 return self.receive_from_client.startswith(extension_marker, 4) 5 ctx1dbkei

606 

607 def query_extensions(self) -> Generator[int, None, None]: (empty)

608 """Answer an `SSH_AGENTC_EXTENSION` request. 

609 

610 Yields: 

611 The bytes payload of the response, without the protocol 

612 framing. The payload is yielded byte by byte, as an 

613 iterable of 8-bit integers. 

614 

615 """ 

616 yield _types.SSH_AGENT.EXTENSION_RESPONSE 4 ctx1dkei

617 yield from ssh_agent.SSHAgentClient.string(b"query") 4 ctx1dkei

618 extension_answers = [ 4 ctx1dkei

619 b"query", 

620 b"list-extended@putty.projects.tartarus.org", 

621 ] 

622 for a in extension_answers: 4 ctx1dkei

623 yield from ssh_agent.SSHAgentClient.string(a) 4 ctx1dkei

624 

625 def request_identities( (empty)

626 self, *, list_extended: bool = False 

627 ) -> Generator[int, None, None]: 

628 """Answer an `SSH_AGENTC_REQUEST_IDENTITIES` request. 

629 

630 Args: 

631 list_extended: 

632 If true, answer an `SSH_AGENTC_EXTENSION` request for 

633 the `list-extended@putty.projects.tartarus.org` 

634 extension. Otherwise, answer an 

635 `SSH_AGENTC_REQUEST_IDENTITIES` request. 

636 

637 Yields: 

638 The bytes payload of the response, without the protocol 

639 framing. The payload is yielded byte by byte, as an 

640 iterable of 8-bit integers. 

641 

642 """ 

643 if list_extended: 5 ctx1abfel

644 yield _types.SSH_AGENT.SUCCESS 1 ctx1b

645 else: 

646 yield _types.SSH_AGENT.IDENTITIES_ANSWER 5 ctx1abfel

647 signature_classes = [ 5 ctx1abfel

648 data.SSHTestKeyDeterministicSignatureClass.SPEC, 

649 ] 

650 if ( 5 ctx1abfel

651 "list-extended@putty.projects.tartarus.org" 

652 in self.enabled_extensions 

653 ): 

654 signature_classes.append( 3 ctx1abe

655 data.SSHTestKeyDeterministicSignatureClass.RFC_6979 

656 ) 

657 keys = [ 5 ctx1abfel

658 v 

659 for v in data.ALL_KEYS.values() 

660 if any(cls in v.expected_signatures for cls in signature_classes) 

661 ] 

662 yield from ssh_agent.SSHAgentClient.uint32(len(keys)) 5 ctx1abfel

663 for key in keys: 5 ctx1abfel

664 yield from ssh_agent.SSHAgentClient.string(key.public_key_data) 5 ctx1abfel

665 yield from ssh_agent.SSHAgentClient.string( 5 ctx1abfel

666 b"test key without passphrase" 

667 ) 

668 if list_extended: 5 ctx1abfel

669 yield from ssh_agent.SSHAgentClient.string( 1 ctx1b

670 ssh_agent.SSHAgentClient.uint32(0) 

671 ) 

672 

673 def sign(self, request_payload: bytes, /) -> bytes: (empty)

674 """Answer an `SSH_AGENTC_SIGN_REQUEST` request. 

675 

676 Args: 

677 request_payload: 

678 The data of the sign request, without the protocol 

679 framing or the request code. 

680 

681 Returns: 

682 The bytes payload of the response, without the protocol 

683 framing. 

684 

685 """ 

686 try_rfc6979 = ( 4 ctx1jhge

687 "list-extended@putty.projects.tartarus.org" 

688 in self.enabled_extensions 

689 ) 

690 spec = data.SSHTestKeyDeterministicSignatureClass.SPEC 4 ctx1jhge

691 rfc6979 = data.SSHTestKeyDeterministicSignatureClass.RFC_6979 4 ctx1jhge

692 try: 4 ctx1jhge

693 key_blob, rest = ssh_agent.SSHAgentClient.unstring_prefix( 4 ctx1jhge

694 request_payload 

695 ) 

696 sign_data, rest = ssh_agent.SSHAgentClient.unstring_prefix(rest) 4 ctx1jhge

697 except ValueError: 1 ctx1h

698 return self._failure() 1 ctx1h

699 if len(rest) != 4: 4 ctx1jhge

700 return self._failure() 1 ctx1h

701 flags = int.from_bytes(rest, "big") 3 ctx1jge

702 if flags: 3 ctx1jge

703 return self._failure() 1 ctx1g

704 if sign_data != vault.Vault.UUID: 3 ctx1jge

705 return self._failure() 1 ctx1g

706 for key in data.ALL_KEYS.values(): 3 ctx1jge

707 if key.public_key_data == key_blob: 3 ctx1jge

708 if spec in key.expected_signatures: 3 ctx1jge

709 return int.to_bytes( 2 ctx1je

710 _types.SSH_AGENT.SIGN_RESPONSE, 1, "big" 

711 ) + ssh_agent.SSHAgentClient.string( 

712 key.expected_signatures[spec].signature 

713 ) 

714 if try_rfc6979 and rfc6979 in key.expected_signatures: 2 ctx1ge

715 return int.to_bytes( 1 ctx1e

716 _types.SSH_AGENT.SIGN_RESPONSE, 1, "big" 

717 ) + ssh_agent.SSHAgentClient.string( 

718 key.expected_signatures[rfc6979].signature 

719 ) 

720 return self._failure() 1 ctx1g

721 return self._failure() 1 ctx1g

722 

723 

724# Standard variant 

725# ---------------- 

726 

727 

728@socketprovider.SocketProvider.register( (empty)

729 _types.BuiltinSSHAgentSocketProvider.STUB_AGENT_WITH_ADDRESS 

730) 

731class StubbedSSHAgentSocketWithAddress(StubbedSSHAgentSocket): (empty)

732 """A [`StubbedSSHAgentSocket`][] requiring a specific address.""" 

733 

734 ADDRESS = "stub-ssh-agent:" (empty)

735 """The correct address for connecting to this stubbed agent.""" (empty)

736 

737 def __init__(self, *extensions: str) -> None: (empty)

738 """Initialize the agent, based on `SSH_AUTH_SOCK`. 

739 

740 Socket addresses of the form `stub-ssh-agent:<errno_value>` will 

741 raise an [`OSError`][] (or the respective subclass) with the 

742 specified [`errno`][] value. For example, 

743 `stub-ssh-agent:EPERM` will raise a [`PermissionError`][]. 

744 

745 Raises: 

746 KeyError: 

747 The `SSH_AUTH_SOCK` environment variable is not set. 

748 OSError: 

749 The address in `SSH_AUTH_SOCK` is unsuited. 

750 

751 """ 

752 super().__init__(*extensions) 9 ctx1adbxfcnmo

753 try: 9 ctx1adbxfcnmo

754 orig_address = os.environ["SSH_AUTH_SOCK"] 9 ctx1adbxfcnmo

755 except KeyError as exc: 3 ctx1xfo

756 msg = "SSH_AUTH_SOCK environment variable" 3 ctx1xfo

757 raise KeyError(msg) from exc 3 ctx1xfo

758 address = orig_address 8 ctx1adbxfcnm

759 if not address.startswith(self.ADDRESS): 8 ctx1adbxfcnm

760 address = self.ADDRESS + "ENOENT" 2 ctx1xm

761 errcode = address.removeprefix(self.ADDRESS) 8 ctx1adbxfcnm

762 if errcode and not ( 8 ctx1adbxfcnm

763 errcode.startswith("E") and hasattr(errno, errcode) 

764 ): 

765 errcode = "EINVAL" 2 ctx1xf

766 if errcode: 8 ctx1adbxfcnm

767 errno_val = getattr(errno, errcode) 3 ctx1xfm

768 raise OSError(errno_val, os.strerror(errno_val), orig_address) 3 ctx1xfm

769 

770 

771# Deterministic variant 

772# --------------------- 

773 

774 

775@socketprovider.SocketProvider.register( (empty)

776 _types.BuiltinSSHAgentSocketProvider.STUB_AGENT_WITH_ADDRESS_AND_DETERMINISTIC_DSA 

777) 

778class StubbedSSHAgentSocketWithAddressAndDeterministicDSA( (empty)

779 StubbedSSHAgentSocketWithAddress 

780): 

781 """A [`StubbedSSHAgentSocketWithAddress`][] supporting deterministic DSA.""" 

782 

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

784 """Initialize the agent. 

785 

786 Set the supported extensions, and try issuing RFC 6979 and 

787 Pageant 0.68–0.80 DSA/ECDSA signatures, if possible. See the 

788 [superclass constructor][StubbedSSHAgentSocketWithAddress] for 

789 other details. 

790 

791 Raises: 

792 KeyError: See superclass. 

793 OSError: See superclass. 

794 

795 """ # noqa: RUF002 

796 super().__init__("query", "list-extended@putty.projects.tartarus.org") 3 ctx1adb

797 self.try_rfc6979 = True 3 ctx1adb

798 self.try_pageant_068_080 = True 3 ctx1adb