Coverage for tests / test_derivepassphrase_ssh_agent / test_000_basic.py: 100.000%

426 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"""Tests for `derivepassphrase.ssh_agent`: basic tests.""" 

6 

7from __future__ import annotations 

8 

9import base64 

10import contextlib 

11import importlib.metadata 

12import re 

13import sys 

14from typing import TYPE_CHECKING, Final, cast 

15 

16import click 

17import hypothesis 

18import pytest 

19from hypothesis import strategies 

20from typing_extensions import TypeAlias 

21 

22from derivepassphrase import _types, ssh_agent, vault 

23from derivepassphrase._internals import cli_helpers 

24from derivepassphrase.ssh_agent import socketprovider 

25from tests import data, machinery 

26from tests.data import callables 

27from tests.machinery import hypothesis as hypothesis_machinery 

28from tests.machinery import pytest as pytest_machinery 

29 

30if TYPE_CHECKING: 

31 from collections.abc import ( 

32 Callable, 

33 Generator, 

34 Mapping, 

35 Sequence, 

36 ) 

37 

38 from typing_extensions import Any, Literal 

39 

40if sys.version_info < (3, 11): 

41 from exceptiongroup import ExceptionGroup 

42 

43 

44class Parametrize(pytest_machinery.Parametrize): 

45 BAD_ENTRY_POINTS = pytest.mark.parametrize( 

46 "additional_entry_points", 

47 [ 

48 pytest.param( 

49 [ 

50 importlib.metadata.EntryPoint( 

51 name=data.faulty_entry_callable.key, 

52 group=socketprovider.SocketProvider.ENTRY_POINT_GROUP_NAME, 

53 value="tests.data: faulty_entry_callable", 

54 ), 

55 ], 

56 id="not-callable", 

57 ), 

58 pytest.param( 

59 [ 

60 importlib.metadata.EntryPoint( 

61 name=data.faulty_entry_name_exists.key, 

62 group=socketprovider.SocketProvider.ENTRY_POINT_GROUP_NAME, 

63 value="tests.data: faulty_entry_name_exists", 

64 ), 

65 ], 

66 id="name-already-exists", 

67 ), 

68 pytest.param( 

69 [ 

70 importlib.metadata.EntryPoint( 

71 name=data.faulty_entry_alias_exists.key, 

72 group=socketprovider.SocketProvider.ENTRY_POINT_GROUP_NAME, 

73 value="tests.data: faulty_entry_alias_exists", 

74 ), 

75 ], 

76 id="alias-already-exists", 

77 ), 

78 ], 

79 ) 

80 GOOD_ENTRY_POINTS = pytest.mark.parametrize( 

81 "additional_entry_points", 

82 [ 

83 pytest.param( 

84 [ 

85 importlib.metadata.EntryPoint( 

86 name=data.ssh_auth_sock_on_posix_entry.key, 

87 group=socketprovider.SocketProvider.ENTRY_POINT_GROUP_NAME, 

88 value="tests.data: ssh_auth_sock_on_posix_entry", 

89 ), 

90 importlib.metadata.EntryPoint( 

91 name=data.ssh_auth_sock_on_windows_entry.key, 

92 group=socketprovider.SocketProvider.ENTRY_POINT_GROUP_NAME, 

93 value="tests.data: ssh_auth_sock_on_windows_entry", 

94 ), 

95 ], 

96 id="existing-entries", 

97 ), 

98 pytest.param( 

99 [ 

100 importlib.metadata.EntryPoint( 

101 name=callables.provider_entry1.key, 

102 group=socketprovider.SocketProvider.ENTRY_POINT_GROUP_NAME, 

103 value="tests.data.callables: provider_entry1", 

104 ), 

105 importlib.metadata.EntryPoint( 

106 name=callables.provider_entry2.key, 

107 group=socketprovider.SocketProvider.ENTRY_POINT_GROUP_NAME, 

108 value="tests.data.callables: provider_entry2", 

109 ), 

110 ], 

111 id="new-entries", 

112 ), 

113 ], 

114 ) 

115 EXISTING_REGISTRY_ENTRIES = pytest.mark.parametrize( 

116 "existing", 

117 [ 

118 _types.BuiltinSSHAgentSocketProvider.SSH_AUTH_SOCK_ON_POSIX, 

119 _types.BuiltinSSHAgentSocketProvider.SSH_AUTH_SOCK_ON_WINDOWS, 

120 ], 

121 ids=str, 

122 ) 

123 SSH_STRING_EXCEPTIONS = hypothesis_machinery.explicit_examples( 

124 ["input", "exc_type", "exc_pattern"], 

125 [ 

126 pytest.param( 

127 "some string", TypeError, "invalid payload type", id="str" 

128 ), 

129 ], 

130 ) 

131 UINT32_EXCEPTIONS = hypothesis_machinery.explicit_examples( 

132 ["input", "exc_type", "exc_pattern"], 

133 [ 

134 pytest.param( 

135 10000000000000000, 

136 OverflowError, 

137 "int too big to convert", 

138 id="10000000000000000", 

139 ), 

140 pytest.param( 

141 -1, 

142 OverflowError, 

143 "can't convert negative int to unsigned", 

144 id="-1", 

145 ), 

146 ], 

147 ) 

148 SSH_UNSTRING_EXCEPTIONS = hypothesis_machinery.explicit_examples( 

149 ["input", "exc_type", "exc_pattern", "parts"], 

150 [ 

151 pytest.param( 

152 b"ssh", 

153 ValueError, 

154 "malformed SSH byte string", 

155 None, 

156 id="unencoded", 

157 ), 

158 pytest.param( 

159 b"\x00\x00\x00\x08ssh-rsa", 

160 ValueError, 

161 "malformed SSH byte string", 

162 None, 

163 id="truncated", 

164 ), 

165 pytest.param( 

166 b"\x00\x00\x00\x04XXX trailing text", 

167 ValueError, 

168 "malformed SSH byte string", 

169 (b"XXX ", b"trailing text"), 

170 id="trailing-data", 

171 ), 

172 ], 

173 ) 

174 SSH_STRING_INPUT = hypothesis_machinery.explicit_examples( 

175 ["input", "expected"], 

176 [ 

177 pytest.param( 

178 b"ssh-rsa", 

179 b"\x00\x00\x00\x07ssh-rsa", 

180 id="ssh-rsa", 

181 ), 

182 pytest.param( 

183 b"ssh-ed25519", 

184 b"\x00\x00\x00\x0bssh-ed25519", 

185 id="ssh-ed25519", 

186 ), 

187 pytest.param( 

188 ssh_agent.SSHAgentClient.string(b"ssh-ed25519"), 

189 b"\x00\x00\x00\x0f\x00\x00\x00\x0bssh-ed25519", 

190 id="string(ssh-ed25519)", 

191 ), 

192 ], 

193 ) 

194 SSH_UNSTRING_INPUT = hypothesis_machinery.explicit_examples( 

195 ["input", "expected"], 

196 [ 

197 pytest.param( 

198 b"\x00\x00\x00\x07ssh-rsa", 

199 b"ssh-rsa", 

200 id="ssh-rsa", 

201 ), 

202 pytest.param( 

203 ssh_agent.SSHAgentClient.string(b"ssh-ed25519"), 

204 b"ssh-ed25519", 

205 id="ssh-ed25519", 

206 ), 

207 ], 

208 ) 

209 UINT32_INPUT = hypothesis_machinery.explicit_examples( 

210 ["input", "expected"], 

211 [ 

212 pytest.param(16777216, b"\x01\x00\x00\x00", id="16777216"), 

213 ], 

214 ) 

215 SIGN_ERROR_RESPONSES = hypothesis_machinery.explicit_examples( 

216 [ 

217 "key", 

218 "check", 

219 "response_code", 

220 "response", 

221 "exc_type", 

222 "exc_pattern", 

223 ], 

224 [ 

225 pytest.param( 

226 b"invalid-key", 

227 True, 

228 _types.SSH_AGENT.FAILURE, 

229 b"", 

230 KeyError, 

231 "target SSH key not loaded into agent", 

232 id="key-not-loaded", 

233 ), 

234 pytest.param( 

235 data.SUPPORTED_KEYS["ed25519"].public_key_data, 

236 True, 

237 _types.SSH_AGENT.FAILURE, 

238 b"", 

239 ssh_agent.SSHAgentFailedError, 

240 "failed to complete the request", 

241 id="failed-to-complete", 

242 ), 

243 ], 

244 ) 

245 SSH_KEY_SELECTION = pytest.mark.parametrize( 

246 ["key", "single"], 

247 [ 

248 (value.public_key_data, False) 

249 for value in data.SUPPORTED_KEYS.values() 

250 ] 

251 + [(callables.list_keys_singleton()[0].key, True)], 

252 ids=[*data.SUPPORTED_KEYS.keys(), "singleton"], 

253 ) 

254 SH_EXPORT_LINES = hypothesis_machinery.explicit_examples( 

255 ["line", "env_name", "value"], 

256 [ 

257 pytest.param( 

258 "SSH_AUTH_SOCK=/tmp/pageant.user/pageant.27170; export SSH_AUTH_SOCK;", 

259 "SSH_AUTH_SOCK", 

260 "/tmp/pageant.user/pageant.27170", 

261 id="value-export-semicolon-pageant", 

262 ), 

263 pytest.param( 

264 "SSH_AUTH_SOCK=/tmp/ssh-3CSTC1W5M22A/agent.27270; export SSH_AUTH_SOCK;", 

265 "SSH_AUTH_SOCK", 

266 "/tmp/ssh-3CSTC1W5M22A/agent.27270", 

267 id="value-export-semicolon-openssh", 

268 ), 

269 pytest.param( 

270 "SSH_AUTH_SOCK=/tmp/pageant.user/pageant.27170; export SSH_AUTH_SOCK", 

271 "SSH_AUTH_SOCK", 

272 "/tmp/pageant.user/pageant.27170", 

273 id="value-export-pageant", 

274 ), 

275 pytest.param( 

276 "export SSH_AUTH_SOCK=/tmp/ssh-3CSTC1W5M22A/agent.27270;", 

277 "SSH_AUTH_SOCK", 

278 "/tmp/ssh-3CSTC1W5M22A/agent.27270", 

279 id="export-value-semicolon-openssh", 

280 ), 

281 pytest.param( 

282 "export SSH_AUTH_SOCK=/tmp/pageant.user/pageant.27170", 

283 "SSH_AUTH_SOCK", 

284 "/tmp/pageant.user/pageant.27170", 

285 id="export-value-pageant", 

286 ), 

287 pytest.param( 

288 "SSH_AGENT_PID=27170; export SSH_AGENT_PID;", 

289 "SSH_AGENT_PID", 

290 "27170", 

291 id="pid-export-semicolon", 

292 ), 

293 pytest.param( 

294 "SSH_AGENT_PID=27170; export SSH_AGENT_PID", 

295 "SSH_AGENT_PID", 

296 "27170", 

297 id="pid-export", 

298 ), 

299 pytest.param( 

300 "export SSH_AGENT_PID=27170;", 

301 "SSH_AGENT_PID", 

302 "27170", 

303 id="export-pid-semicolon", 

304 ), 

305 pytest.param( 

306 "export SSH_AGENT_PID=27170", 

307 "SSH_AGENT_PID", 

308 "27170", 

309 id="export-pid", 

310 ), 

311 pytest.param( 

312 "export VARIABLE=value; export OTHER_VARIABLE=other_value;", 

313 "VARIABLE", 

314 None, 

315 id="export-too-much", 

316 ), 

317 pytest.param( 

318 "VARIABLE=value", 

319 "VARIABLE", 

320 None, 

321 id="no-export", 

322 ), 

323 ], 

324 ) 

325 PUBLIC_KEY_DATA = pytest.mark.parametrize( 

326 "public_key_struct", 

327 list(data.SUPPORTED_KEYS.values()), 

328 ids=list(data.SUPPORTED_KEYS.keys()), 

329 ) 

330 REQUEST_ERROR_RESPONSES = hypothesis_machinery.explicit_examples( 

331 ["request_code", "response_code", "exc_type", "exc_pattern"], 

332 [ 

333 pytest.param( 

334 _types.SSH_AGENTC.REQUEST_IDENTITIES, 

335 _types.SSH_AGENT.SUCCESS, 

336 ssh_agent.SSHAgentFailedError, 

337 re.escape( 

338 "[Code {:d}]".format( # noqa: UP032 

339 int(_types.SSH_AGENT.IDENTITIES_ANSWER) 

340 ) 

341 ), 

342 id="REQUEST_IDENTITIES-expect-SUCCESS", 

343 ), 

344 ], 

345 ) 

346 TRUNCATED_AGENT_RESPONSES = hypothesis_machinery.explicit_examples( 

347 "response", 

348 [ 

349 b"\x00\x00", 

350 b"\x00\x00\x00\x1f some bytes missing", 

351 ], 

352 ids=["in-header", "in-body"], 

353 ) 

354 LIST_KEYS_ERROR_RESPONSES = hypothesis_machinery.explicit_examples( 

355 ["response_code", "response", "exc_type", "exc_pattern"], 

356 [ 

357 pytest.param( 

358 _types.SSH_AGENT.FAILURE, 

359 b"", 

360 ssh_agent.SSHAgentFailedError, 

361 "failed to complete the request", 

362 id="failed-to-complete", 

363 ), 

364 pytest.param( 

365 _types.SSH_AGENT.IDENTITIES_ANSWER, 

366 b"\x00\x00\x00\x01", 

367 EOFError, 

368 "truncated response", 

369 id="truncated-response", 

370 ), 

371 pytest.param( 

372 _types.SSH_AGENT.IDENTITIES_ANSWER, 

373 b"\x00\x00\x00\x00abc", 

374 ssh_agent.TrailingDataError, 

375 "Overlong response", 

376 id="overlong-response", 

377 ), 

378 ], 

379 ) 

380 QUERY_EXTENSIONS_MALFORMED_RESPONSES = ( 

381 hypothesis_machinery.explicit_examples( 

382 "response_data", 

383 [ 

384 pytest.param(b"\xde\xad\xbe\xef", id="truncated"), 

385 pytest.param( 

386 b"\x00\x00\x00\x0fwrong extension", id="wrong-extension" 

387 ), 

388 pytest.param( 

389 b"\x00\x00\x00\x05query\xde\xad\xbe\xef", id="with-trailer" 

390 ), 

391 pytest.param( 

392 b"\x00\x00\x00\x05query\x00\x00\x00\x04ext1\x00\x00", 

393 id="with-extra-fields", 

394 ), 

395 ], 

396 ) 

397 ) 

398 ALL_SSH_TEST_KEYS = pytest.mark.parametrize( 

399 ["ssh_test_key_type", "ssh_test_key"], 

400 list(data.ALL_KEYS.items()), 

401 ids=data.ALL_KEYS.keys(), 

402 ) 

403 RESOLVE_CHAINS = hypothesis_machinery.explicit_examples( 

404 ["terminal", "chain"], 

405 [ 

406 pytest.param("callable", ["a"], id="callable-1"), 

407 pytest.param("callable", ["a", "b", "c", "d"], id="callable-4"), 

408 pytest.param("alias", ["e"], id="alias-1"), 

409 pytest.param("alias", ["e", "f", "g", "h", "i"], id="alias-5"), 

410 pytest.param("unimplemented", ["j"], id="unimplemented-1"), 

411 pytest.param("unimplemented", ["j", "k"], id="unimplemented-2"), 

412 ], 

413 ) 

414 

415 

416class Strategies: 

417 """Common hypothesis data generation strategies.""" 

418 

419 @staticmethod 

420 def select_invalid_pipe_names(candidate: str) -> bool: 

421 """Select candidate strings that are invalid named pipe names.""" 

422 candidate = candidate.lower().replace("/", "\\") 1vn

423 return not candidate.startswith(socketprovider.PIPE_PREFIX) 1vn

424 

425 @classmethod 

426 def bad_named_pipe_names(cls) -> strategies.SearchStrategy[str]: 

427 r"""Generate invalid names for Windows named pipes. 

428 

429 Invalid named pipe names do not lie below the `\\.\pipe\` path. 

430 They also may or may not contain characters that are invalid 

431 in paths on The Annoying OS. 

432 

433 """ 

434 # This is an environment string on The Annoying OS, so it must 

435 # at least be a non-empty C string, i.e., without embedded NULs. 

436 return strategies.text( 

437 alphabet=strategies.characters(min_codepoint=1), min_size=1 

438 ).filter(cls.select_invalid_pipe_names) 

439 

440 @staticmethod 

441 def io_operations_on_handles() -> strategies.SearchStrategy[ 

442 Callable[[_types.SSHAgentSocket], Any] 

443 ]: 

444 """Return an I/O operation on an SSH agent socket.""" 

445 return strategies.one_of( 

446 strategies.integers(min_value=0).map( 

447 lambda i: lambda handle: handle.recv(i) 

448 ), 

449 strategies.binary().map( 

450 lambda bs: lambda handle: handle.sendall(bs) 

451 ), 

452 ) 

453 

454 

455class TestStaticFunctionality: 

456 """Test the static functionality of the `ssh_agent` module.""" 

457 

458 # TODO(the-13th-letter): Put this functionality into the 

459 # tests.data.callables module. It is nice to have for other tests as 

460 # well: `vault` tests, stubbed SSH agent tests, agent protocol 

461 # response queue tests, etc. Also add an unstring and a u32 

462 # function. 

463 @staticmethod 

464 def as_ssh_string(bytestring: bytes) -> bytes: 

465 """Return an encoded SSH string from a bytestring. 

466 

467 This is a helper function for hypothesis data generation. 

468 

469 """ 

470 return int.to_bytes(len(bytestring), 4, "big") + bytestring 1el

471 

472 @staticmethod 

473 def canonicalize1(data: bytes) -> bytes: 

474 """Return an encoded SSH string from a bytestring. 

475 

476 This is a helper function for hypothesis testing. 

477 

478 References: 

479 

480 * [David R. MacIver: Another invariant to test for 

481 encoders][DECODE_ENCODE] 

482 

483 [DECODE_ENCODE]: https://hypothesis.works/articles/canonical-serialization/ 

484 

485 """ 

486 return ssh_agent.SSHAgentClient.string( 1l

487 ssh_agent.SSHAgentClient.unstring(data) 

488 ) 

489 

490 @staticmethod 

491 def canonicalize2(data: bytes) -> bytes: 

492 """Return an encoded SSH string from a bytestring. 

493 

494 This is a helper function for hypothesis testing. 

495 

496 References: 

497 

498 * [David R. MacIver: Another invariant to test for 

499 encoders][DECODE_ENCODE] 

500 

501 [DECODE_ENCODE]: https://hypothesis.works/articles/canonical-serialization/ 

502 

503 """ 

504 unstringed, trailer = ssh_agent.SSHAgentClient.unstring_prefix(data) 1l

505 assert not trailer 1l

506 return ssh_agent.SSHAgentClient.string(unstringed) 1l

507 

508 

509class TestShellExportScriptParsing: 

510 """Test the shell export script parsing utility function.""" 

511 

512 @Parametrize.SH_EXPORT_LINES 

513 def test_sh_export_line_parsing( 1ax

514 self, line: str, env_name: str, value: str | None 

515 ) -> None: 

516 """[`tests.data.callables.parse_sh_export_line`][] works.""" 

517 if value is not None: 1x

518 assert ( 1x

519 callables.parse_sh_export_line(line, env_name=env_name) 

520 == value 

521 ) 

522 else: 

523 with pytest.raises(ValueError, match="Cannot parse sh line:"): 1x

524 callables.parse_sh_export_line(line, env_name=env_name) 1x

525 

526 

527class TestSSHProtocolDatatypes(TestStaticFunctionality): 

528 """Tests for the utility functions for the SSH protocol datatypes.""" 

529 

530 @Parametrize.UINT32_INPUT 

531 def test_uint32(self, input: int, expected: bytes | bytearray) -> None: 1aH

532 """`uint32` encoding works.""" 

533 uint32 = ssh_agent.SSHAgentClient.uint32 1H

534 assert uint32(input) == expected 1H

535 

536 @hypothesis.given( 

537 num=strategies.integers(min_value=0, max_value=0xFFFFFFFF) 

538 ) 

539 @hypothesis.example(num=0xDEADBEEF) 

540 def test_uint32_from_number(self, num: int) -> None: 

541 """`uint32` encoding works, starting from numbers.""" 

542 uint32 = ssh_agent.SSHAgentClient.uint32 1K

543 assert int.from_bytes(uint32(num), "big", signed=False) == num 1K

544 

545 @hypothesis.given(bytestring=strategies.binary(min_size=4, max_size=4)) 

546 @hypothesis.example(bytestring=b"\xde\xad\xbe\xef") 1aI

547 def test_uint32_from_bytestring(self, bytestring: bytes) -> None: 

548 """`uint32` encoding works, starting from length four byte strings.""" 

549 uint32 = ssh_agent.SSHAgentClient.uint32 1I

550 assert ( 1I

551 uint32(int.from_bytes(bytestring, "big", signed=False)) 

552 == bytestring 

553 ) 

554 

555 @Parametrize.SSH_STRING_INPUT 

556 def test_string( 1aJ

557 self, input: bytes | bytearray, expected: bytes | bytearray 

558 ) -> None: 

559 """SSH string encoding works.""" 

560 string = ssh_agent.SSHAgentClient.string 1J

561 assert bytes(string(input)) == expected 1J

562 

563 @hypothesis.given(bytestring=strategies.binary(max_size=0x0001FFFF)) 

564 # example with highest order bit set 

565 @hypothesis.example(bytestring=b"DEADBEEF" * 10000) 

566 def test_string_from_bytestring(self, bytestring: bytes) -> None: 

567 """SSH string encoding works, starting from a byte string.""" 

568 res = ssh_agent.SSHAgentClient.string(bytestring) 1D

569 assert res.startswith((b"\x00\x00", b"\x00\x01")) 1D

570 assert int.from_bytes(res[:4], "big", signed=False) == len(bytestring) 1D

571 assert res[4:] == bytestring 1D

572 

573 @Parametrize.SSH_UNSTRING_INPUT 

574 def test_unstring( 1ay

575 self, input: bytes | bytearray, expected: bytes | bytearray 

576 ) -> None: 

577 """SSH string decoding works.""" 

578 unstring = ssh_agent.SSHAgentClient.unstring 1y

579 unstring_prefix = ssh_agent.SSHAgentClient.unstring_prefix 1y

580 assert bytes(unstring(input)) == expected 1y

581 assert tuple(bytes(x) for x in unstring_prefix(input)) == ( 1y

582 expected, 

583 b"", 

584 ) 

585 

586 @hypothesis.given(bytestring=strategies.binary(max_size=0x00FFFFFF)) 

587 # should not double-decode 

588 @hypothesis.example(bytestring=b"\x00\x00\x00\x07ssh-rsa") 

589 # should not choke on ill-formed SSH string payloads 

590 @hypothesis.example(bytestring=b"\x00\x00\x00\x01") 

591 def test_unstring_of_string_of_data(self, bytestring: bytes) -> None: 

592 """SSH string decoding of encoded SSH strings works. 

593 

594 References: 

595 

596 * [David R. MacIver: The Encode/Decode invariant][ENCODE_DECODE] 

597 

598 [ENCODE_DECODE]: https://hypothesis.works/articles/encode-decode-invariant/ 

599 

600 """ 

601 string = ssh_agent.SSHAgentClient.string 1o

602 unstring = ssh_agent.SSHAgentClient.unstring 1o

603 unstring_prefix = ssh_agent.SSHAgentClient.unstring_prefix 1o

604 encoded = string(bytestring) 1o

605 assert unstring(encoded) == bytestring 1o

606 assert unstring_prefix(encoded) == (bytestring, b"") 1o

607 trailing_data = b" trailing data" 1o

608 encoded2 = string(bytestring) + trailing_data 1o

609 assert unstring_prefix(encoded2) == (bytestring, trailing_data) 1o

610 

611 @hypothesis.given( 

612 encoded=strategies.binary(max_size=0x00FFFFFF).map( 

613 # Scoping issues, and the fact that staticmethod objects 

614 # (before class finalization) are not callable, necessitate 

615 # wrapping this staticmethod call in a lambda. 

616 lambda x: TestStaticFunctionality.as_ssh_string(x) # noqa: PLW0108 

617 ), 

618 ) 

619 def test_string_of_unstring_of_data(self, encoded: bytes) -> None: 

620 """SSH string decoding of encoded SSH strings works. 

621 

622 References: 

623 

624 * [David R. MacIver: Another invariant to test for 

625 encoders][DECODE_ENCODE] 

626 

627 [DECODE_ENCODE]: https://hypothesis.works/articles/canonical-serialization/ 

628 

629 """ 

630 # Both canonicalization functions are built with slightly 

631 # different primitives. However, because SSH string encoding is 

632 # a bijective encoding, their result should be the same. 

633 canonical_functions = [self.canonicalize1, self.canonicalize2] 1l

634 # Test encoding equivalence. 

635 for canon1 in canonical_functions: 1l

636 for canon2 in canonical_functions: 1l

637 assert canon1(encoded) == canon2(encoded) 1l

638 # Test idempotence. 

639 for canon1 in canonical_functions: 1l

640 for canon2 in canonical_functions: 1l

641 assert canon1(canon2(encoded)) == canon1(encoded) 1l

642 

643 # TODO(the-13th-letter): Write a hypothesis strategy. 

644 @Parametrize.UINT32_EXCEPTIONS 

645 def test_uint32_exceptions( 1aE

646 self, input: int, exc_type: type[Exception], exc_pattern: str 

647 ) -> None: 

648 """`uint32` encoding fails for out-of-bound values.""" 

649 uint32 = ssh_agent.SSHAgentClient.uint32 1E

650 with pytest.raises(exc_type, match=exc_pattern): 1E

651 uint32(input) 1E

652 

653 # TODO(the-13th-letter): Write a hypothesis strategy. 

654 @Parametrize.SSH_STRING_EXCEPTIONS 

655 def test_string_exceptions( 1aF

656 self, input: Any, exc_type: type[Exception], exc_pattern: str 

657 ) -> None: 

658 """SSH string encoding fails for non-strings.""" 

659 string = ssh_agent.SSHAgentClient.string 1F

660 with pytest.raises(exc_type, match=exc_pattern): 1F

661 string(input) 1F

662 

663 # TODO(the-13th-letter): Write a hypothesis strategy. 

664 @Parametrize.SSH_UNSTRING_EXCEPTIONS 

665 def test_unstring_exceptions( 1ap

666 self, 

667 input: bytes, 

668 exc_type: type[Exception], 

669 exc_pattern: str, 

670 parts: tuple[bytes, bytes] | None, 

671 ) -> None: 

672 """SSH string decoding fails for invalid values.""" 

673 unstring = ssh_agent.SSHAgentClient.unstring 1p

674 unstring_prefix = ssh_agent.SSHAgentClient.unstring_prefix 1p

675 with pytest.raises(exc_type, match=exc_pattern): 1p

676 unstring(input) 1p

677 if parts is not None: 1p

678 assert unstring_prefix(input) == parts 1p

679 else: 

680 with pytest.raises(exc_type, match=exc_pattern): 1p

681 unstring_prefix(input) 1p

682 

683 

684# TODO(the-13th-letter): Introduce a registry analyzer that decomposes 

685# the registry into chains. On this decomposed representation, provide 

686# strategies to draw chains, existing entries, non-existing entries, 

687# leaves, aliases, etc. 

688class TestSSHAgentSocketProviderRegistry: 

689 """Tests for the SSH agent socket provider registry.""" 

690 

691 def test_resolve( 

692 self, 

693 ) -> None: 

694 """Resolving entries in the socket provider registry works.""" 

695 registry = socketprovider.SocketProvider.registry 1m

696 resolve = socketprovider.SocketProvider.resolve 1m

697 lookup = socketprovider.SocketProvider.lookup 1m

698 with pytest.MonkeyPatch.context() as monkeypatch: 1m

699 monkeypatch.setitem( 1m

700 registry, _types.BuiltinSSHAgentSocketProvider.STUB_AGENT, None 

701 ) 

702 assert callable( 1m

703 lookup(_types.BuiltinSSHAgentSocketProvider.NATIVE) 

704 ) 

705 assert callable( 1m

706 resolve(_types.BuiltinSSHAgentSocketProvider.NATIVE) 

707 ) 

708 assert ( 1m

709 lookup(_types.BuiltinSSHAgentSocketProvider.STUB_AGENT) is None 

710 ) 

711 with pytest.raises(NotImplementedError): 1m

712 resolve(_types.BuiltinSSHAgentSocketProvider.STUB_AGENT) 1m

713 

714 @Parametrize.RESOLVE_CHAINS 

715 @hypothesis.given( 1ag

716 terminal=strategies.sampled_from([ 

717 "unimplemented", 

718 "alias", 

719 "callable", 

720 ]), 

721 chain=strategies.lists( 

722 strategies.sampled_from([ 

723 "c1", 

724 "c2", 

725 "c3", 

726 "c4", 

727 "c5", 

728 "c6", 

729 "c7", 

730 "c8", 

731 "c9", 

732 "c10", 

733 ]), 

734 min_size=1, 

735 unique=True, 

736 ), 

737 ) 

738 def test_resolve_chains( 

739 self, 

740 terminal: Literal["unimplemented", "alias", "callable"], 

741 chain: list[str], 

742 ) -> None: 

743 """Resolving a chain of providers works.""" 

744 registry = socketprovider.SocketProvider.registry 1g

745 resolve = socketprovider.SocketProvider.resolve 1g

746 lookup = socketprovider.SocketProvider.lookup 1g

747 try: 1g

748 implementation = resolve( 1g

749 _types.BuiltinSSHAgentSocketProvider.NATIVE 

750 ) 

751 except NotImplementedError: # pragma: no cover 

752 hypothesis.note(f"{registry = }") 

753 pytest.fail("Native SSH agent socket provider is unavailable?!") 

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

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

756 target: _types.SSHAgentSocketProvider | str | None = ( 1g

757 None 

758 if terminal == "unimplemented" 

759 else _types.BuiltinSSHAgentSocketProvider.NATIVE 

760 if terminal == "alias" 

761 else implementation 

762 ) 

763 with pytest.MonkeyPatch.context() as monkeypatch: 1g

764 for link in chain: 1g

765 monkeypatch.setitem(registry, link, target) 1g

766 target = link 1g

767 for link in chain: 1g

768 assert lookup(link) == ( 1g

769 implementation if terminal != "unimplemented" else None 

770 ) 

771 if terminal == "unimplemented": 1g

772 with pytest.raises(NotImplementedError): 1g

773 resolve(link) 1g

774 else: 

775 assert resolve(link) == implementation 1g

776 

777 # TODO(the-13th-letter): Reevaluate whether this makes sense as a 

778 # property-based test. 

779 @Parametrize.GOOD_ENTRY_POINTS 

780 def test_find_all_socket_providers( 

781 self, 

782 additional_entry_points: list[importlib.metadata.EntryPoint], 

783 ) -> None: 

784 """Finding all SSH agent socket providers works.""" 

785 resolve = socketprovider.SocketProvider.resolve 1t

786 old_registry = socketprovider.SocketProvider.registry 1t

787 with pytest_machinery.faked_entry_point_list( 1t

788 additional_entry_points, remove_conflicting_entries=False 

789 ) as names: 

790 socketprovider.SocketProvider._find_all_ssh_agent_socket_providers() 1t

791 for name in names: 1t

792 assert name in socketprovider.SocketProvider.registry 1t

793 assert resolve(name) in { 1t

794 callables.provider_entry_provider, 

795 *old_registry.values(), 

796 } 

797 

798 # TODO(the-13th-letter): Reevaluate whether this makes sense as a 

799 # property-based test. 

800 @Parametrize.BAD_ENTRY_POINTS 

801 def test_find_all_socket_providers_errors( 

802 self, 

803 additional_entry_points: list[importlib.metadata.EntryPoint], 

804 ) -> None: 

805 """Finding faulty SSH agent socket providers raises errors.""" 

806 with contextlib.ExitStack() as stack: 1G

807 stack.enter_context( 1G

808 pytest_machinery.faked_entry_point_list( 

809 additional_entry_points, remove_conflicting_entries=False 

810 ) 

811 ) 

812 stack.enter_context(pytest.raises(AssertionError)) 1G

813 socketprovider.SocketProvider._find_all_ssh_agent_socket_providers() 1G

814 

815 # TODO(the-13th-letter): Convert this to a property-based test once 

816 # the registry analyzer is ready. 

817 def test_overwriting_existing_entries( 

818 self, 

819 ) -> None: 

820 """The registry forbids incompatibly overwriting entries.""" 

821 registry = socketprovider.SocketProvider.registry.copy() 1b

822 resolve = socketprovider.SocketProvider.resolve 1b

823 register = socketprovider.SocketProvider.register 1b

824 

825 def ancestry_chain(name: str) -> Generator[str, None, None]: 1b

826 current: _types.SSHAgentSocketProvider | str | None = name 1b

827 seen: set[_types.SSHAgentSocketProvider | str | None] = set() 1b

828 while isinstance(current, str) and current not in seen: 1b

829 yield current 1b

830 seen.add(current) 1b

831 current = registry[current] 1b

832 

833 # Entries with at least one level of alias resolving, and where each 

834 # entry ultimately resolves to a different provider. They needn't 

835 # technically be leaves of the socket provider forest, but "leaves" 

836 # emphasizes the quality we are looking for. 

837 leaves = [ 1b

838 _types.BuiltinSSHAgentSocketProvider.POSIX, 

839 _types.BuiltinSSHAgentSocketProvider.WINDOWS, 

840 ] 

841 assert all([isinstance(registry.get(leaf), str) for leaf in leaves]), ( 1b

842 "Registry is shaped incompatibly; cannot determine base entries" 

843 ) 

844 bases = [resolve(leaf) for leaf in leaves] 1b

845 names = [list(ancestry_chain(leaf)) for leaf in leaves] 1b

846 

847 with pytest.MonkeyPatch.context() as monkeypatch: 1b

848 monkeypatch.setattr( 1b

849 socketprovider.SocketProvider, "registry", registry 

850 ) 

851 # Registering names with different values is forbidden... 

852 with pytest.raises(ValueError, match="already registered"): 1b

853 register(names[0][0])(bases[1]) 1b

854 with pytest.raises(ValueError, match="already registered"): 1b

855 register(names[1][0])(bases[0]) 1b

856 # ...even if other names of the same registration are valid. 

857 with pytest.raises(ValueError, match="already registered"): 1b

858 register(names[0][0], names[1][1])(bases[0]) 1b

859 with pytest.raises(ValueError, match="already registered"): 1b

860 register(names[1][0], names[0][1])(bases[1]) 1b

861 

862 # TODO(the-13th-letter): Convert this to a property-based test once 

863 # the registry analyzer is ready. 

864 def test_resolve_non_existant_entries( 

865 self, 

866 ) -> None: 

867 """Resolving a non-existant entry fails.""" 

868 new_registry = { 1z

869 _types.BuiltinSSHAgentSocketProvider.POSIX: socketprovider.SocketProvider.resolve( 

870 _types.BuiltinSSHAgentSocketProvider.POSIX 

871 ), 

872 _types.BuiltinSSHAgentSocketProvider.WINDOWS: socketprovider.SocketProvider.resolve( 

873 _types.BuiltinSSHAgentSocketProvider.WINDOWS 

874 ), 

875 } 

876 with pytest.MonkeyPatch.context() as monkeypatch: 1z

877 monkeypatch.setattr( 1z

878 socketprovider.SocketProvider, "registry", new_registry 

879 ) 

880 with pytest.raises(socketprovider.NoSuchProviderError): 1z

881 socketprovider.SocketProvider.resolve( 1z

882 _types.BuiltinSSHAgentSocketProvider.NATIVE 

883 ) 

884 

885 # TODO(the-13th-letter): Convert this to a property-based test once 

886 # the registry analyzer is ready. 

887 def test_register_new_entry( 

888 self, 

889 ) -> None: 

890 """Registering new entries works.""" 

891 

892 def socket_provider() -> _types.SSHAgentSocket: 1q

893 raise AssertionError 

894 

895 names = ["spam", "ham", "eggs", "parrot"] 1q

896 new_registry = { 1q

897 _types.BuiltinSSHAgentSocketProvider.POSIX: socketprovider.SocketProvider.resolve( 

898 _types.BuiltinSSHAgentSocketProvider.POSIX 

899 ), 

900 _types.BuiltinSSHAgentSocketProvider.WINDOWS: socketprovider.SocketProvider.resolve( 

901 _types.BuiltinSSHAgentSocketProvider.WINDOWS 

902 ), 

903 } 

904 with pytest.MonkeyPatch.context() as monkeypatch: 1q

905 monkeypatch.setattr( 1q

906 socketprovider.SocketProvider, "registry", new_registry 

907 ) 

908 assert not any( 1q

909 map(socketprovider.SocketProvider.registry.__contains__, names) 

910 ) 

911 assert ( 1q

912 socketprovider.SocketProvider.register(*names)(socket_provider) 

913 is socket_provider 

914 ) 

915 assert all( 1q

916 map(socketprovider.SocketProvider.registry.__contains__, names) 

917 ) 

918 assert all([ 1q

919 socketprovider.SocketProvider.resolve(n) is socket_provider 

920 for n in names 

921 ]) 

922 

923 # TODO(the-13th-letter): Convert this to a property-based test once 

924 # the registry analyzer is ready. 

925 @Parametrize.EXISTING_REGISTRY_ENTRIES 

926 def test_register_old_entry( 

927 self, 

928 existing: str, 

929 ) -> None: 

930 """Registering old entries works.""" 

931 

932 provider = socketprovider.SocketProvider.resolve(existing) 1r

933 new_registry = { 1r

934 _types.BuiltinSSHAgentSocketProvider.SSH_AUTH_SOCK_ON_POSIX: socketprovider.SocketProvider.resolve( 

935 _types.BuiltinSSHAgentSocketProvider.SSH_AUTH_SOCK_ON_POSIX 

936 ), 

937 _types.BuiltinSSHAgentSocketProvider.SSH_AUTH_SOCK_ON_WINDOWS: socketprovider.SocketProvider.resolve( 

938 _types.BuiltinSSHAgentSocketProvider.SSH_AUTH_SOCK_ON_WINDOWS 

939 ), 

940 _types.BuiltinSSHAgentSocketProvider.UNIX_DOMAIN: _types.BuiltinSSHAgentSocketProvider.SSH_AUTH_SOCK_ON_POSIX, 

941 _types.BuiltinSSHAgentSocketProvider.WINDOWS_NAMED_PIPE: _types.BuiltinSSHAgentSocketProvider.SSH_AUTH_SOCK_ON_WINDOWS, 

942 } 

943 names = [ 1r

944 k 

945 for k, v in socketprovider.SocketProvider.registry.items() 

946 if v == existing 

947 ] 

948 with pytest.MonkeyPatch.context() as monkeypatch: 1r

949 monkeypatch.setattr( 1r

950 socketprovider.SocketProvider, "registry", new_registry 

951 ) 

952 assert names, ( 1r

953 f"Existing SSH agent socket provider entry " 

954 f"{existing!r} is not existing?!" 

955 ) 

956 # TODO(the-13th-letter): Do we require, forbid, or not care 

957 # if `names` discovers aliases not present in the mocked 

958 # registry? Currently, the "ssh_auth_sock" alias for 

959 # "ssh_auth_sock_on_posix" would be discovered, but 

960 # "ssh_auth_sock_on_windows" has no undiscovered aliases. 

961 # 

962 # assert not 

963 # all(map(socketprovider.SocketProvider.registry.__contains__, 

964 # names)) 

965 assert ( 1r

966 socketprovider.SocketProvider.register(existing, *names)( 

967 provider 

968 ) 

969 is provider 

970 ) 

971 assert all( 1r

972 map(socketprovider.SocketProvider.registry.__contains__, names) 

973 ) 

974 assert all([ 1r

975 socketprovider.SocketProvider.resolve(n) is provider 

976 for n in [existing, *names] 

977 ]) 

978 

979 

980class TestAgentSigning: 

981 """Test actually talking to the SSH agent: signing data.""" 

982 

983 @Parametrize.ALL_SSH_TEST_KEYS 

984 def test_sign_data_via_agent( 

985 self, 

986 ssh_agent_client_with_test_keys_loaded: ssh_agent.SSHAgentClient, 

987 ssh_test_key_type: str, 

988 ssh_test_key: data.SSHTestKey, 

989 ) -> None: 

990 """Signing data with specific SSH keys works iff the key is suitable. 

991 

992 Single tests may abort early (skip) if the indicated key is not 

993 loaded in the agent. Presumably this means the key type is 

994 unsupported. 

995 

996 """ 

997 client = ssh_agent_client_with_test_keys_loaded 1c

998 key_comment_pairs = {bytes(k): bytes(c) for k, c in client.list_keys()} 1c

999 public_key_data = ssh_test_key.public_key_data 1c

1000 if public_key_data not in key_comment_pairs: # pragma: no cover 1c

1001 pytest.skip(f"prerequisite {ssh_test_key_type} SSH key not loaded") 1c

1002 if ssh_test_key.is_suitable(): 1c

1003 assert ( 1c

1004 data.SSHTestKeyDeterministicSignatureClass.SPEC 

1005 in ssh_test_key.expected_signatures 

1006 ) 

1007 sig = ssh_test_key.expected_signatures[ 1c

1008 data.SSHTestKeyDeterministicSignatureClass.SPEC 

1009 ] 

1010 expected_signature = sig.signature 1c

1011 derived_passphrase = sig.derived_passphrase 1c

1012 signature = bytes( 1c

1013 client.sign(payload=vault.Vault.UUID, key=public_key_data) 

1014 ) 

1015 assert signature == expected_signature, ( 1c

1016 f"SSH signature mismatch ({ssh_test_key_type})" 

1017 ) 

1018 signature2 = bytes( 1c

1019 client.sign(payload=vault.Vault.UUID, key=public_key_data) 

1020 ) 

1021 assert signature2 == expected_signature, ( 1c

1022 f"SSH signature mismatch ({ssh_test_key_type})" 

1023 ) 

1024 assert ( 1c

1025 vault.Vault.phrase_from_key(public_key_data, conn=client) 

1026 == derived_passphrase 

1027 ), f"SSH signature mismatch ({ssh_test_key_type})" 

1028 else: 

1029 assert ( 1c

1030 data.SSHTestKeyDeterministicSignatureClass.SPEC 

1031 not in ssh_test_key.expected_signatures 

1032 ) 

1033 assert not vault.Vault.is_suitable_ssh_key( 1c

1034 public_key_data, client=None 

1035 ), f"Expected {ssh_test_key_type} key to be unsuitable in general" 

1036 if vault.Vault.is_suitable_ssh_key( 1c

1037 public_key_data, client=client 

1038 ): # pragma: no cover [external] 

1039 potential_signatures = { 1c

1040 sig_class: sig.signature 

1041 for sig_class, sig in ssh_test_key.expected_signatures.items() 

1042 if sig_class 

1043 != data.SSHTestKeyDeterministicSignatureClass.SPEC 

1044 } 

1045 signature = bytes( 1c

1046 client.sign(payload=vault.Vault.UUID, key=public_key_data) 

1047 ) 

1048 assert signature in potential_signatures.values() 1c

1049 else: # pragma: no cover [external] 

1050 with pytest.raises(ValueError, match="unsuitable SSH key"): 1c

1051 vault.Vault.phrase_from_key(public_key_data, conn=client) 1c

1052 

1053 

1054class TestSuitableKeys: 

1055 """Test actually talking to the SSH agent: determining suitable keys.""" 

1056 

1057 @Parametrize.SSH_KEY_SELECTION 

1058 def test_ssh_key_selector( 

1059 self, 

1060 monkeypatch: pytest.MonkeyPatch, 

1061 ssh_agent_client_with_test_keys_loaded: ssh_agent.SSHAgentClient, 

1062 key: bytes, 

1063 single: bool, 

1064 ) -> None: 

1065 """The key selector presents exactly the suitable keys. 

1066 

1067 "Suitable" here means suitability for this SSH agent 

1068 specifically. 

1069 

1070 """ 

1071 client = ssh_agent_client_with_test_keys_loaded 1d

1072 

1073 monkeypatch.setattr( 1d

1074 ssh_agent.SSHAgentClient, 

1075 "list_keys", 

1076 callables.list_keys_singleton if single else callables.list_keys, 

1077 ) 

1078 raw_keys_list = ( 1d

1079 callables.list_keys_singleton() 

1080 if single 

1081 else callables.list_keys() 

1082 ) 

1083 keys = [ 1d

1084 pair.key 

1085 for pair in raw_keys_list 

1086 if vault.Vault.is_suitable_ssh_key(pair.key, client=client) 

1087 ] 

1088 n = len(keys) 1d

1089 if single or n == 1: 1d

1090 input_text = "yes\n" 1d

1091 text = "Use this key? yes\n" 1d

1092 else: 

1093 input_text = f"{1 + keys.index(key)}\n" 1d

1094 text = ( 1d

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

1096 ) 

1097 b64_key = base64.standard_b64encode(key).decode("ASCII") 1d

1098 

1099 @click.command() 1d

1100 def driver() -> None: 1d

1101 """Call [`cli_helpers.select_ssh_key`][] directly, as a command.""" 

1102 key = cli_helpers.select_ssh_key( 1d

1103 client, 

1104 error_callback=lambda s, *_args: pytest.fail(str(s)), 

1105 warning_callback=lambda s, *_args: pytest.fail(str(s)), 

1106 ) 

1107 click.echo(base64.standard_b64encode(key).decode("ASCII")) 1d

1108 

1109 runner = machinery.CliRunner(mix_stderr=True) 1d

1110 result = runner.invoke( 1d

1111 driver, [], input=input_text, catch_exceptions=True 

1112 ) 

1113 for snippet in ("Suitable SSH keys:\n", text, f"\n{b64_key}\n"): 1d

1114 assert result.clean_exit(output=snippet), "expected clean exit" 1d

1115 

1116 

1117class TestConstructorFailures: 

1118 """Test actually talking to the SSH agent: constructor failures.""" 

1119 

1120 def test_constructor_bad_running_agent( 

1121 self, 

1122 running_ssh_agent: data.RunningSSHAgentInfo, 

1123 ) -> None: 

1124 """Fail if the agent address is invalid.""" 

1125 with pytest.MonkeyPatch.context() as monkeypatch: 1A

1126 new_socket_name = ( 1A

1127 running_ssh_agent.socket + "~" 

1128 if isinstance(running_ssh_agent.socket, str) 

1129 else "<invalid//address>" 

1130 ) 

1131 monkeypatch.setenv("SSH_AUTH_SOCK", new_socket_name) 1A

1132 with pytest.raises(OSError): # noqa: PT011 1A

1133 ssh_agent.SSHAgentClient() 1A

1134 

1135 @Parametrize.MISSING_AGENT_SUPPORT 

1136 def test_constructor_no_support( 

1137 self, 

1138 provider: _types.BuiltinSSHAgentSocketProvider, 

1139 sys_action: data.SystemSupportAction, 

1140 exception: type[NotImplementedError], 

1141 message: str, 

1142 ) -> None: 

1143 """Fail without support for the communication technology.""" 

1144 del message 1u

1145 assert provider in socketprovider.SocketProvider.registry 1u

1146 with pytest.MonkeyPatch.context() as monkeypatch: 1u

1147 monkeypatch.setenv( 1u

1148 "SSH_AUTH_SOCK", "the value doesn't even matter" 

1149 ) 

1150 sys_action(monkeypatch) 1u

1151 with pytest.raises(exception): 1u

1152 ssh_agent.SSHAgentClient(socket=provider) 1u

1153 

1154 def test_no_ssh_agent_socket_provider_available( 

1155 self, 

1156 ) -> None: 

1157 """Fail if no SSH agent socket provider is available.""" 

1158 with pytest.MonkeyPatch.context() as monkeypatch: 1B

1159 monkeypatch.setitem( 1B

1160 socketprovider.SocketProvider.registry, "stub_agent", None 

1161 ) 

1162 with pytest.raises(ExceptionGroup) as excinfo: 1B

1163 ssh_agent.SSHAgentClient( 1B

1164 socket=["stub_agent", "stub_agent", "stub_agent"] 

1165 ) 

1166 assert all([ 1B

1167 isinstance(e, NotImplementedError) 

1168 for e in excinfo.value.exceptions 

1169 ]) 

1170 

1171 

1172class TestOtherConstructorFeatures: 

1173 """Test actually talking to the SSH agent: other constructor features.""" 

1174 

1175 def test_explicit_socket( 

1176 self, 

1177 spawn_ssh_agent: data.SpawnedSSHAgentInfo, 

1178 ) -> None: 

1179 """The constructor accepts existing sockets.""" 

1180 conn = spawn_ssh_agent.client._connection 1L

1181 ssh_agent.SSHAgentClient(socket=conn) 1L

1182 

1183 

1184@pytest.mark.usefixtures("use_stub_agent") 

1185class TestAgentErrorResponses: 

1186 """Test actually talking to the SSH agent: errors from the SSH agent.""" 

1187 

1188 @contextlib.contextmanager 

1189 def _setup_agent_and_request_handler( 

1190 self, 

1191 request_responses_map: Mapping[ 

1192 data.AgentProtocolRequest, Sequence[data.AgentProtocolResponse] 

1193 ], 

1194 /, 

1195 ) -> Generator[ 

1196 tuple[pytest.MonkeyPatch, ssh_agent.SSHAgentClient], None, None 

1197 ]: 

1198 # TODO(the-13th-letter): Rewrite using parenthesized 

1199 # with-statements. 

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

1201 with contextlib.ExitStack() as stack: 1ejkf

1202 monkeypatch = stack.enter_context(pytest.MonkeyPatch.context()) 1ejkf

1203 client = stack.enter_context( 1ejkf

1204 ssh_agent.SSHAgentClient.ensure_agent_subcontext() 

1205 ) 

1206 agent_io_queue = machinery.AgentProtocolResponseQueue( 1ejkf

1207 request_responses_map 

1208 ) 

1209 stack.enter_context(agent_io_queue) 1ejkf

1210 monkeypatch.setattr( 1ejkf

1211 client, 

1212 "_send_request_message", 

1213 agent_io_queue._send_request_message, 

1214 ) 

1215 monkeypatch.setattr( 1ejkf

1216 client, "_get_response_pair", agent_io_queue._get_response_pair 

1217 ) 

1218 yield monkeypatch, client 1ejkf

1219 

1220 @Parametrize.TRUNCATED_AGENT_RESPONSES 

1221 def test_truncated_server_response( 1as

1222 self, 

1223 response: bytes, 

1224 ) -> None: 

1225 """Fail on truncated responses from the SSH agent.""" 

1226 client = ssh_agent.SSHAgentClient() 1s

1227 agent = cast("machinery.StubbedSSHAgentSocket", client._connection) 1s

1228 with pytest.MonkeyPatch.context() as monkeypatch: 1s

1229 monkeypatch.setattr(agent, "send_to_client", bytearray(response)) 1s

1230 monkeypatch.setattr(agent, "sendall", lambda *_a, **_kw: None) 1s

1231 with pytest.raises(EOFError): 1s

1232 client.request(255, b"") 1s

1233 

1234 @Parametrize.LIST_KEYS_ERROR_RESPONSES 

1235 def test_list_keys_error_responses( 1aj

1236 self, 

1237 response_code: _types.SSH_AGENT, 

1238 response: bytes | bytearray, 

1239 exc_type: type[Exception], 

1240 exc_pattern: str, 

1241 ) -> None: 

1242 """Fail on problems during key listing. 

1243 

1244 Known problems: 

1245 

1246 - The agent refuses, or otherwise indicates the operation 

1247 failed. 

1248 - The agent response is truncated. 

1249 - The agent response is overlong. 

1250 

1251 """ 

1252 with self._setup_agent_and_request_handler({ 1j

1253 data.AgentProtocolRequest( 

1254 _types.SSH_AGENTC.REQUEST_IDENTITIES, b"" 

1255 ): [data.AgentProtocolResponse(response_code, bytes(response))], 

1256 }) as contexts: 

1257 _, client = contexts 1j

1258 with pytest.raises(exc_type, match=exc_pattern): 1j

1259 client.list_keys() 1j

1260 

1261 @Parametrize.SIGN_ERROR_RESPONSES 

1262 def test_sign_error_responses( 1af

1263 self, 

1264 key: bytes | bytearray, 

1265 check: bool, 

1266 response_code: _types.SSH_AGENT, 

1267 response: bytes | bytearray, 

1268 exc_type: type[Exception], 

1269 exc_pattern: str, 

1270 ) -> None: 

1271 """Fail on problems during signing. 

1272 

1273 Known problems: 

1274 

1275 - The key is not loaded into the agent. 

1276 - The agent refuses, or otherwise indicates the operation 

1277 failed. 

1278 

1279 """ 

1280 Pair: TypeAlias = _types.SSHKeyCommentPair 1f

1281 com = b"no comment" 1f

1282 loaded_keys = [ 1f

1283 Pair(v.public_key_data, com).toreadonly() 

1284 for v in data.SUPPORTED_KEYS.values() 

1285 ] 

1286 sign_payload = ( 1f

1287 ssh_agent.SSHAgentClient.string(key) 

1288 + ssh_agent.SSHAgentClient.string(b"abc") 

1289 + ssh_agent.SSHAgentClient.uint32(0) 

1290 ) 

1291 with self._setup_agent_and_request_handler({ 1f

1292 data.AgentProtocolRequest( 

1293 _types.SSH_AGENTC.SIGN_REQUEST, sign_payload 

1294 ): [data.AgentProtocolResponse(response_code, bytes(response))], 

1295 }) as contexts: 

1296 monkeypatch, client = contexts 1f

1297 monkeypatch.setattr(client, "list_keys", lambda: loaded_keys) 1f

1298 with pytest.raises(exc_type, match=exc_pattern): 1f

1299 client.sign(key, b"abc", check_if_key_loaded=check) 1f

1300 

1301 @Parametrize.REQUEST_ERROR_RESPONSES 

1302 def test_request_error_responses( 1aC

1303 self, 

1304 request_code: _types.SSH_AGENTC, 

1305 response_code: _types.SSH_AGENT, 

1306 exc_type: type[Exception], 

1307 exc_pattern: str, 

1308 ) -> None: 

1309 """Fail on problems during signing. 

1310 

1311 Known problems: 

1312 

1313 - The key is not loaded into the agent. 

1314 - The agent refuses, or otherwise indicates the operation 

1315 failed. 

1316 

1317 """ 

1318 # TODO(the-13th-letter): Rewrite using parenthesized 

1319 # with-statements. 

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

1321 with contextlib.ExitStack() as stack: 1C

1322 client = stack.enter_context(ssh_agent.SSHAgentClient()) 1C

1323 stack.enter_context(pytest.raises(exc_type, match=exc_pattern)) 1C

1324 client.request(request_code, b"", response_code=response_code) 1C

1325 

1326 @Parametrize.QUERY_EXTENSIONS_MALFORMED_RESPONSES 

1327 def test_query_extensions_malformed_responses( 1ak

1328 self, 

1329 response_data: bytes, 

1330 ) -> None: 

1331 """Fail on malformed responses while querying extensions.""" 

1332 with self._setup_agent_and_request_handler({ 1k

1333 data.AgentProtocolRequest( 

1334 _types.SSH_AGENTC.EXTENSION, 

1335 ssh_agent.SSHAgentClient.string(b"query"), 

1336 ): [ 

1337 data.AgentProtocolResponse( 

1338 _types.SSH_AGENT.EXTENSION_RESPONSE, response_data 

1339 ) 

1340 ], 

1341 }) as contexts: 

1342 _, client = contexts 1k

1343 with pytest.raises( 1k

1344 RuntimeError, 

1345 match=r"Malformed response|does not match request", 

1346 ): 

1347 client.query_extensions() 1k

1348 

1349 def _query_response( 

1350 self, 

1351 /, 

1352 *extensions: str, 

1353 code: Literal[ 

1354 _types.SSH_AGENT.EXTENSION_RESPONSE, _types.SSH_AGENT.SUCCESS 

1355 ] = _types.SSH_AGENT.EXTENSION_RESPONSE, 

1356 ) -> data.AgentProtocolResponse: 

1357 """Construct a response object from the given extensions. 

1358 

1359 Args: 

1360 extensions: 

1361 Names of extensions to advertise support for. 

1362 code: 

1363 The response code of the message. 

1364 

1365 Returns: 

1366 An appropriately constructed response object. 

1367 

1368 """ 

1369 payload = bytearray(b"\x00\x00\x00\x05query") 1e

1370 for ext in extensions: 1e

1371 payload.extend(TestStaticFunctionality.as_ssh_string(ext.encode())) 1e

1372 return data.AgentProtocolResponse(code, bytes(payload)) 1e

1373 

1374 IDENTITIES_REQUEST: Final = data.AgentProtocolRequest( 

1375 _types.SSH_AGENTC.REQUEST_IDENTITIES, b"" 

1376 ) 

1377 IDENTITIES_RESPONSE: Final = data.AgentProtocolResponse( 

1378 _types.SSH_AGENT.IDENTITIES_ANSWER, b"\x00\x00\x00\x00" 

1379 ) 

1380 QUERY_REQUEST: Final = data.AgentProtocolRequest( 

1381 _types.SSH_AGENTC.EXTENSION, b"\x00\x00\x00\x05query" 

1382 ) 

1383 EXTRA_RESPONSE: Final = data.AgentProtocolResponse( 

1384 _types.SSH_AGENT.SUCCESS, b"" 

1385 ) 

1386 

1387 # TODO(the-13th-letter): Reconsider whether to parametrize or 

1388 # hypothesize this test. 

1389 @pytest.mark.parametrize( 

1390 ["success_after_query", "extensions"], 

1391 [ 

1392 pytest.param( 

1393 False, 

1394 data.AgentProtocolResponse(_types.SSH_AGENT.FAILURE, b""), 

1395 id="openssh-10.2", 

1396 ), 

1397 pytest.param( 

1398 True, 

1399 data.AgentProtocolResponse(_types.SSH_AGENT.FAILURE, b""), 

1400 id="openssh-10.2-mocked", 

1401 marks=[ 

1402 pytest.mark.xfail( 

1403 reason="specific to OpenSSH 10.3 output", 

1404 raises=AssertionError, 

1405 strict=True, 

1406 ) 

1407 ], 

1408 ), 

1409 pytest.param( 

1410 False, ["session-bind@openssh.com"], id="openssh-fixed" 

1411 ), 

1412 pytest.param( 

1413 True, ["session-bind@openssh.com"], id="openssh-10.3" 

1414 ), 

1415 pytest.param( 

1416 False, 

1417 [ 

1418 "add-ppk@putty.projects.tartarus.org", 

1419 "reencrypt@putty.projects.tartarus.org", 

1420 "reencrypt-all@putty.projects.tartarus.org", 

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

1422 ], 

1423 id="pageant", 

1424 ), 

1425 pytest.param( 

1426 True, 

1427 [ 

1428 "add-ppk@putty.projects.tartarus.org", 

1429 "reencrypt@putty.projects.tartarus.org", 

1430 "reencrypt-all@putty.projects.tartarus.org", 

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

1432 ], 

1433 id="pageant-mocked", 

1434 marks=[ 

1435 pytest.mark.xfail( 

1436 reason="specific to OpenSSH 10.3 output", 

1437 raises=AssertionError, 

1438 strict=True, 

1439 ) 

1440 ], 

1441 ), 

1442 ], 

1443 ) 

1444 def test_tolerating_extra_success_after_extension_response( 

1445 self, 

1446 success_after_query: bool, 

1447 extensions: list[str] | data.AgentProtocolResponse, 

1448 ) -> None: 

1449 """Tolerate an extra empty SUCCESS message from a "query" EXTENSION. 

1450 

1451 Specifically, in a "query" 

1452 [`EXTENSION`][_types.SSH_AGENTC.EXTENSION] request, tolerate 

1453 getting potentially both an 

1454 [`EXTENSION_RESPONSE`][_types.SSH_AGENT.EXTENSION_RESPONSE] or 

1455 [`SUCCESS`][_types.SSH_AGENT.SUCCESS] message and an additional 

1456 empty [`SUCCESS`][_types.SSH_AGENT.SUCCESS] message afterwards, 

1457 instead of only the first message. This is spec-uncompliant 

1458 behavior observed in `ssh-agent` from OpenSSH 10.3. (And since 

1459 then reported and acknowledged upstream, to be fixed in OpenSSH 

1460 10.4: [OpenSSH Bug 3967][OPENSSH_BZ3967].) 

1461 

1462 [OPENSSH_BZ3967]: https://bugzilla.mindrot.org/show_bug.cgi?id=3967 

1463 

1464 """ 

1465 query_response = ( 1e

1466 extensions 

1467 if isinstance(extensions, data.AgentProtocolResponse) 

1468 else self._query_response(*extensions) 

1469 ) 

1470 request_responses_map = { 1e

1471 self.IDENTITIES_REQUEST: [self.IDENTITIES_RESPONSE], 

1472 self.QUERY_REQUEST: [query_response, self.EXTRA_RESPONSE] 

1473 if success_after_query 

1474 else [query_response], 

1475 } 

1476 with self._setup_agent_and_request_handler( 1e

1477 request_responses_map 

1478 ) as contexts: 

1479 _, client = contexts 1e

1480 client.query_extensions() 1e

1481 

1482 

1483class TestSSHAgentSocketProviderFailures: 

1484 """Test SSH agent socket provider-specific failures.""" 

1485 

1486 def test_posix_provider_no_ssh_auth_sock( 

1487 self, 

1488 skip_if_no_af_unix_support: None, 

1489 ) -> None: 

1490 """For `posix`, fail if the running agent cannot be located.""" 

1491 del skip_if_no_af_unix_support 1w

1492 provider = socketprovider.SocketProvider.resolve( 1w

1493 _types.BuiltinSSHAgentSocketProvider.POSIX 

1494 ) 

1495 with pytest.MonkeyPatch.context() as monkeypatch: 1w

1496 monkeypatch.delenv("SSH_AUTH_SOCK", raising=False) 1w

1497 with pytest.raises( 1w

1498 KeyError, match="SSH_AUTH_SOCK environment variable" 

1499 ): 

1500 provider() 1w

1501 

1502 @hypothesis.given(named_pipe_name=Strategies.bad_named_pipe_names()) 

1503 def test_ssh_auth_sock_on_windows_provider_bad_named_pipe_name( 1an

1504 self, 

1505 named_pipe_name: str, 

1506 skip_if_no_ctypes_windll_support: None, 

1507 ) -> None: 

1508 """For `ssh_auth_sock_on_windows`, fail for bad named pipe names.""" 

1509 del skip_if_no_ctypes_windll_support 1n

1510 provider = socketprovider.SocketProvider.resolve( 1n

1511 _types.BuiltinSSHAgentSocketProvider.SSH_AUTH_SOCK_ON_WINDOWS 

1512 ) 

1513 with pytest.MonkeyPatch.context() as monkeypatch: 1n

1514 monkeypatch.setenv("SSH_AUTH_SOCK", named_pipe_name) 1n

1515 with pytest.raises(ValueError, match="Invalid named pipe name"): 1n

1516 provider() 1n

1517 

1518 

1519class TestWindowsNamedPipeHandle: 

1520 """Test the WindowsNamedPipeHandle wrapper class. 

1521 

1522 Even works on other operating systems, because the relevant parts 

1523 are stubbed out anyway. 

1524 

1525 """ 

1526 

1527 DUMMY_ADDRESS = f"{socketprovider.PIPE_PREFIX}dummy-address" 

1528 IO_ON_CLOSED_HANDLE_PATTERN = re.escape( 

1529 socketprovider.WindowsNamedPipeHandle._IO_ON_CLOSED_FILE 

1530 ) 

1531 

1532 @hypothesis.given(named_pipe_name=Strategies.bad_named_pipe_names()) 

1533 def test_constructor_bad_named_pipe_name( 1av

1534 self, 

1535 named_pipe_name: str, 

1536 ) -> None: 

1537 """The constructor fails for bad named pipe names.""" 

1538 with pytest.raises(ValueError, match="Invalid named pipe name"): # noqa: SIM117 1v

1539 with socketprovider.WindowsNamedPipeHandle(named_pipe_name): 1v

1540 pass 1v

1541 

1542 @contextlib.contextmanager 

1543 def stub_createfile_and_closehandle( 

1544 self, 

1545 ) -> Generator[None, None, None]: 

1546 singleton = object() 1ih

1547 

1548 def create_file(*_args: Any, **_kwargs: Any) -> object: 1ih

1549 return singleton 1ih

1550 

1551 def close_handle(*_args: Any, **_kwargs: Any) -> bool: 1ih

1552 return True 1ih

1553 

1554 with pytest.MonkeyPatch.context() as monkeypatch: 1ih

1555 monkeypatch.setattr(socketprovider, "CreateFile", create_file) 1ih

1556 monkeypatch.setattr(socketprovider, "CloseHandle", close_handle) 1ih

1557 yield 1ih

1558 

1559 @hypothesis.given(io_operation=Strategies.io_operations_on_handles()) 

1560 def test_closed_handle_cannot_be_operated_on( 1ai

1561 self, 

1562 io_operation: Callable[[Any], Any], 

1563 ) -> None: 

1564 """Operating on a closed handle fails.""" 

1565 with self.stub_createfile_and_closehandle(): 1i

1566 handle = socketprovider.WindowsNamedPipeHandle(self.DUMMY_ADDRESS) 1i

1567 with handle: 1i

1568 pass # close the handle 1i

1569 with pytest.raises( 1i

1570 ValueError, match=self.IO_ON_CLOSED_HANDLE_PATTERN 

1571 ): 

1572 io_operation(handle) 1i

1573 

1574 @hypothesis.given(io_operation=Strategies.io_operations_on_handles()) 

1575 def test_handle_is_non_reentrant( 1ah

1576 self, 

1577 io_operation: Callable[[Any], Any], 

1578 ) -> None: 

1579 """The handler cannot be sensibly used in nested `with` blocks.""" 

1580 with self.stub_createfile_and_closehandle(): 1h

1581 handle = socketprovider.WindowsNamedPipeHandle(self.DUMMY_ADDRESS) 1h

1582 with handle: 1h

1583 with handle: 1h

1584 pass # close the handle 1h

1585 with pytest.raises( 1h

1586 ValueError, match=self.IO_ON_CLOSED_HANDLE_PATTERN 

1587 ): 

1588 io_operation(handle) 1h