Coverage for tests / test_000_testing_machinery.py: 100.000%
254 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
5"""Tests for the test suite's data and machinery.
7Currently, this entails testing [the SSH test keys][data.ALL_KEYS]
8for internal consistency, and testing the functionality of the [stubbed
9SSH agent][tests.machinery.StubbedSSHAgentSocket], in all variations.
11"""
13from __future__ import annotations
15import base64
16import contextlib
17import errno
18import math
19import os
20import pathlib
21import re
22from typing import TYPE_CHECKING
24import hypothesis
25import pytest
26from hypothesis import strategies
28from derivepassphrase import _types, ssh_agent, vault
29from tests import data, machinery
31if TYPE_CHECKING:
32 from collections.abc import Generator
34 from typing_extensions import Buffer
36OPENSSH_MAGIC = b"openssh-key-v1\x00"
37OPENSSH_HEADER = (
38 OPENSSH_MAGIC # magic
39 + b"\x00\x00\x00\x04none" # ciphername
40 + b"\x00\x00\x00\x04none" # kdfname
41 + b"\x00\x00\x00\x00" # kdfoptions
42 + b"\x00\x00\x00\x01" # number of keys
43)
44OPENSSH_NONE_CIPHER_BLOCKSIZE = 8
47def as_openssh_keyfile_payload(
48 public_key: bytes, private_key: bytes, checkint: int
49) -> bytes:
50 """Format an SSH private key in OpenSSH format.
52 Args:
53 public_key:
54 The unframed public key, in SSH wire format.
55 private_key:
56 The unframed private key, in SSH wire format, including the
57 comment.
58 checkint:
59 The "check" integer to use.
61 Returns:
62 The payload for a formatted OpenSSH private key, as a byte
63 string, without the base64 encoding and the framing lines.
65 """
66 # The OpenSSH private key file format is described in PROTOCOL.key
67 # in their git repository; see below for links to OpenSSH 10.0p2.
68 # The block size of the "none" cipher is 8 bytes; see line 108 of
69 # cipher.c, with definitions from line 67 onwards. Padding is not
70 # used if the payload already is a multiple of 8 bytes long; see
71 # line 2935 onwards of sshkey.c
72 #
73 # https://github.com/openssh/openssh-portable/raw/2593769fb291fe6c542173927698c69e9f9a08b9/PROTOCOL.key
74 # https://github.com/openssh/openssh-portable/raw/2593769fb291fe6c542173927698c69e9f9a08b9/cipher.c
75 # https://github.com/openssh/openssh-portable/raw/2593769fb291fe6c542173927698c69e9f9a08b9/sshkey.c
76 string = ssh_agent.SSHAgentClient.string 1b
77 uint32 = ssh_agent.SSHAgentClient.uint32 1b
78 payload = bytearray(OPENSSH_HEADER) 1b
79 payload.extend(string(public_key)) 1b
80 secret = bytearray() 1b
81 secret.extend(uint32(checkint)) # checkint 1b
82 secret.extend(uint32(checkint)) # checkint 1b
83 secret.extend(private_key) # privatekey1 and comment1 1b
84 i = 1 1b
85 while len(secret) % OPENSSH_NONE_CIPHER_BLOCKSIZE != 0: 1b
86 secret.append(i) 1b
87 i += 1 1b
88 payload.extend(string(secret)) # encrypted, padded list of private keys 1b
89 return bytes(payload) 1b
92def minimize_openssh_keyfile_padding(
93 decoded_openssh_private_key: bytes,
94) -> bytes:
95 """Minimize the padding used in an OpenSSH private key file.
97 Args:
98 decoded_openssh_private_key:
99 The non-base64-encoded, unframed, formatted OpenSSH private
100 key.
102 Returns:
103 The same non-base64-encoded, unframed, formatted OpenSSH private
104 key, but with minimal padding applied.
106 """
107 string = ssh_agent.SSHAgentClient.string 1b
108 unstring_prefix = ssh_agent.SSHAgentClient.unstring_prefix 1b
110 _public_key, framed_private_block = unstring_prefix( 1b
111 decoded_openssh_private_key.removeprefix(OPENSSH_HEADER)
112 )
113 result = bytearray(decoded_openssh_private_key).removesuffix( 1b
114 framed_private_block
115 )
116 private_block, trailer = unstring_prefix(framed_private_block) 1b
117 assert not trailer 1b
119 # Skip two checkint values.
120 key_type, remainder = unstring_prefix(private_block[8:]) 1b
121 # We need to semi-generically skip private key payloads. Currently,
122 # all supported (test) key types exclusively store multi-precision
123 # integers or strings as their private key payload (which are both
124 # parsed the same way, but interpreted differently). We can
125 # therefore generically parse `k` strings/mpints (for different
126 # values of `k`, depending on key type) to correctly skip the
127 # private key payload, and don't have to deal with having to parse
128 # and skip other types of data such as uint32s.
129 #
130 # (This scheme needs updating if ever a different data type needs to
131 # be parsed.)
132 num_mpints = { 1b
133 b"ssh-ed25519": 2,
134 b"ssh-ed448": 2,
135 b"ssh-rsa": 6,
136 b"ssh-dss": 5,
137 b"ecdsa-sha2-nistp256": 3,
138 b"ecdsa-sha2-nistp384": 3,
139 b"ecdsa-sha2-nistp521": 3,
140 }
141 for _ in range(num_mpints[key_type]): 1b
142 _, remainder = unstring_prefix(remainder) 1b
143 # Skip comment.
144 _comment, remainder = unstring_prefix(remainder) 1b
145 new_private_block = bytearray(private_block).removesuffix(remainder) 1b
146 padding = bytearray(remainder) 1b
148 expected_padding = bytearray() 1b
149 for i in range(1, len(padding) + 1): 1b
150 expected_padding.append(i & 0xFF) 1b
151 assert padding == expected_padding 1b
152 while len(padding) >= OPENSSH_NONE_CIPHER_BLOCKSIZE: 1b
153 padding[-OPENSSH_NONE_CIPHER_BLOCKSIZE:] = b"" 1b
155 new_private_block.extend(padding) 1b
156 result.extend(string(new_private_block)) 1b
157 return bytes(result) 1b
160class Parametrize:
161 """Common test parametrizations."""
163 LIST_IDENTITIES = pytest.mark.parametrize(
164 ["extended_agent", "query_request"],
165 [
166 pytest.param(
167 None,
168 (
169 # SSH string header
170 b"\x00\x00\x00\x01"
171 # request code: SSH_AGENTC_REQUEST_IDENTITIES
172 b"\x0b"
173 ),
174 id="base",
175 ),
176 pytest.param(
177 True,
178 (
179 # SSH string header
180 b"\x00\x00\x00\x2e"
181 # request code: SSH_AGENTC_REQUEST_IDENTITIES
182 b"\x1b"
183 # extension type: list-extended@putty.projects.tartarus.org
184 b"\x00\x00\x00\x29list-extended@putty.projects.tartarus.org"
185 # (no payload)
186 ),
187 id="extended",
188 ),
189 ],
190 )
191 QUERY_EXTENSION = pytest.mark.parametrize(
192 ["extended_agent", "query_response"],
193 [
194 pytest.param(
195 None,
196 (
197 # SSH string header
198 b"\x00\x00\x00\x01"
199 # response code: SSH_AGENT_FAILURE
200 b"\x05"
201 ),
202 id="base",
203 ),
204 pytest.param(
205 True,
206 (
207 # SSH string header
208 b"\x00\x00\x00\x40"
209 # response code: SSH_AGENT_EXTENSION_RESPONSE
210 b"\x1d"
211 # extension response: extension type ("query")
212 b"\x00\x00\x00\x05query"
213 # supported extension #1: query
214 b"\x00\x00\x00\x05query"
215 # supported extension #2:
216 # list-extended@putty.projects.tartarus.org
217 b"\x00\x00\x00\x29list-extended@putty.projects.tartarus.org"
218 ),
219 id="extended",
220 ),
221 ],
222 )
223 TEST_KEYS = pytest.mark.parametrize(
224 ["keyname", "key"],
225 data.ALL_KEYS.items(),
226 ids=data.ALL_KEYS.keys(),
227 )
228 SUPPORTED_SSH_TEST_KEYS = pytest.mark.parametrize(
229 ["ssh_test_key_type", "ssh_test_key"],
230 list(data.SUPPORTED_KEYS.items()),
231 ids=data.SUPPORTED_KEYS.keys(),
232 )
235class Strategies:
236 """Common hypothesis data generation strategies."""
238 @strategies.composite
239 @staticmethod
240 def proper_bytestring_partition(
241 draw: strategies.DrawFn, bs: bytes | bytearray, /
242 ) -> list[bytes]:
243 """Partition a non-empty string into non-empty parts.
245 A string of length `n` is partitioned [`math.isqrt(n
246 - 1)`][math.isqrt] many times, such that each substring is
247 non-empty.
249 """
250 n = len(bs) 1c
251 hypothesis.assume(n > 0) 1c
252 num_divisions = math.isqrt(n - 1) 1c
253 divisions = draw( 1c
254 strategies.lists(
255 strategies.integers(1, n - 1),
256 min_size=num_divisions,
257 max_size=num_divisions,
258 unique=True,
259 ).map(sorted),
260 label="divisions",
261 )
262 result: list[bytes] = [] 1c
263 i = 0 1c
264 for div in divisions: 1c
265 result.append(bytes(bs[i:div])) 1c
266 i = div 1c
267 result.append(bytes(bs[i:n])) 1c
268 return result 1c
270 @strategies.composite
271 @staticmethod
272 def truncated_agent_request(
273 draw: strategies.DrawFn, /, min_size: int = 0, max_size: int = 100
274 ) -> bytes:
275 """Generate a private-use SSH agent request with truncated payload.
277 The message will adhere to specified size bounds before
278 truncation. The request message code will be from the private
279 use area.
281 Args:
282 min_size:
283 The minimum size of the request payload, after truncation.
284 max_size:
285 The maximum size of the request payload, before truncation.
287 """
288 request_code = draw( 1e
289 strategies.integers(240, 255), label="request_code"
290 )
291 full_message_size = draw( 1e
292 strategies.integers(min_size + 1, max_size),
293 label="full_message_size",
294 )
295 truncated_message = draw( 1e
296 strategies.binary(
297 min_size=min_size, max_size=full_message_size - 1
298 )
299 )
300 payload = bytearray( 1e
301 int.to_bytes(1 + full_message_size, 4, "big", signed=False)
302 )
303 payload.extend(int.to_bytes(request_code, 1, "big", signed=False)) 1e
304 payload.extend(truncated_message) 1e
305 return bytes(payload) 1e
307 @staticmethod
308 def invalid_ssh_agent_messages() -> strategies.SearchStrategy[bytes]:
309 """Generate invalid agent request messages."""
310 string = ssh_agent.SSHAgentClient.string
311 empty_message = [b""]
312 invalid_extension_name = [b"\x1b", b"\x00\x00\x00\x01\xff"]
313 sign_with_trailing_data = [
314 b"\x0d",
315 b"\x00\x00\x00\x00",
316 b"\x00\x00\x00\x00",
317 b"\x00\x00\x00\x00",
318 b"\x00\x00\x00\x00",
319 ]
320 sign_without_payload = [b"\x0d", b"\x00\x00\x00\x00"]
321 return (
322 strategies
323 .sampled_from([
324 empty_message,
325 invalid_extension_name,
326 sign_with_trailing_data,
327 sign_without_payload,
328 ])
329 .map(b"".join)
330 .map(string)
331 )
333 @staticmethod
334 def unsupported_ssh_agent_messages() -> strategies.SearchStrategy[bytes]:
335 """Generate agent request messages unsupported by the stubbed agent."""
336 string = ssh_agent.SSHAgentClient.string
337 sign_with_flags = [
338 b"\x0d",
339 string(data.ALL_KEYS["rsa"].public_key_data),
340 string(vault.Vault.UUID),
341 b"\x00\x00\x00\x02",
342 ]
343 sign_with_nonstandard_passphrase = [
344 b"\x0d",
345 string(data.ALL_KEYS["ed25519"].public_key_data),
346 b"\x00\x00\x00\x08\x00\x01\x02\x03\x04\x05\x06\x07",
347 b"\x00\x00\x00\x00",
348 ]
349 # NOTE: only unsupported when stubbed agent has RFC 6979 support
350 # disabled
351 sign_key_no_expected_signature = [
352 b"\x0d",
353 string(data.ALL_KEYS["dsa1024"].public_key_data),
354 string(vault.Vault.UUID),
355 b"\x00\x00\x00\x00",
356 ]
357 sign_key_unregistered_test_key = [
358 b"\x0d",
359 b"\x00\x00\x00\x00",
360 string(vault.Vault.UUID),
361 b"\x00\x00\x00\x00",
362 ]
363 return (
364 strategies
365 .sampled_from([
366 sign_with_flags,
367 sign_with_nonstandard_passphrase,
368 sign_key_no_expected_signature,
369 sign_key_unregistered_test_key,
370 ])
371 .map(b"".join)
372 .map(string)
373 )
375 @staticmethod
376 def stubbed_agent_addresses() -> strategies.SearchStrategy[
377 tuple[str, type[OSError], str]
378 ]:
379 """Generate agent addresses for the stubbed agents.
381 These are all error calls. Non-error calls are supplied by the
382 explicit examples.
384 """
385 invalid_url = (
386 str(pathlib.Path("~").expanduser()),
387 FileNotFoundError,
388 os.strerror(errno.ENOENT),
389 )
390 protocol_not_supported = (
391 "stub-ssh-agent:EPROTONOSUPPORT",
392 OSError,
393 os.strerror(errno.EPROTONOSUPPORT),
394 )
395 invalid_error_code = (
396 "stub-ssh-agent:ABCDEFGHIJKLMNOPQRSTUVWXYZ",
397 OSError,
398 os.strerror(errno.EINVAL),
399 )
400 return strategies.sampled_from([
401 invalid_url,
402 protocol_not_supported,
403 invalid_error_code,
404 ])
407class TestTestKeys:
408 """Tests testing the test keys."""
410 @Parametrize.TEST_KEYS
411 def test_public_keys_are_internally_consistent(
412 self,
413 keyname: str,
414 key: data.SSHTestKey,
415 ) -> None:
416 """The public key data structures are internally consistent."""
417 del keyname 1n
418 string = ssh_agent.SSHAgentClient.string 1n
419 public_key_lines = key.public_key.splitlines(keepends=False) 1n
420 assert len(public_key_lines) == 1 1n
421 line_parts = public_key_lines[0].strip(b"\r\n").split(None, 2) 1n
422 key_type_name, public_key_b64 = line_parts[:2] 1n
423 assert base64.standard_b64encode(key.public_key_data) == public_key_b64 1n
424 assert key.public_key_data.startswith(string(key_type_name)) 1n
426 # TODO(the-13th-letter): Put RSA key mangling into helper method.
427 @Parametrize.TEST_KEYS
428 def test_private_keys_are_consistent_with_public_keys(
429 self,
430 keyname: str,
431 key: data.SSHTestKey,
432 ) -> None:
433 """The private key data are consistent with their public parts."""
434 del keyname 1h
435 string = ssh_agent.SSHAgentClient.string 1h
437 if key.public_key_data.startswith(string(b"ssh-rsa")): 1h
438 # RSA public keys are *not* prefixes of the corresponding
439 # private key in OpenSSH format! RSA public keys consist of
440 # an exponent e and a modulus n, which in the public key are
441 # in the order (e, n), but in the order (n, e) in the
442 # OpenSSH private key. We thus need to parse and rearrange
443 # the components of the public key into a new "mangled"
444 # public key that then *is* a prefix of the respective
445 # private key.
446 unstring_prefix = ssh_agent.SSHAgentClient.unstring_prefix 1h
447 key_type, numbers = unstring_prefix(key.public_key_data) 1h
448 e, encoded_n = unstring_prefix(numbers) 1h
449 n, trailer = unstring_prefix(encoded_n) 1h
450 assert not trailer 1h
451 mangled_public_key_data = string(key_type) + string(n) + string(e) 1h
452 assert ( 1h
453 key.private_key_blob[: len(mangled_public_key_data)]
454 == mangled_public_key_data
455 )
456 else:
457 assert ( 1h
458 key.private_key_blob[: len(key.public_key_data)]
459 == key.public_key_data
460 )
462 @Parametrize.TEST_KEYS
463 def test_private_keys_are_internally_consistent(
464 self,
465 keyname: str,
466 key: data.SSHTestKey,
467 ) -> None:
468 """The private key data structures are internally consistent."""
469 del keyname 1b
470 string = ssh_agent.SSHAgentClient.string 1b
472 private_key_lines = [ 1b
473 line
474 for line in key.private_key.splitlines(keepends=False)
475 if line and not line.startswith((b"-----BEGIN", b"-----END"))
476 ]
477 private_key_from_openssh = base64.standard_b64decode( 1b
478 b"".join(private_key_lines)
479 )
480 wrapped_public_key = string(key.public_key_data) 1b
481 assert ( 1b
482 private_key_from_openssh[
483 len(OPENSSH_HEADER) : len(OPENSSH_HEADER)
484 + len(wrapped_public_key)
485 ]
486 == wrapped_public_key
487 )
489 # Offset skips the header, the wrapped public key, and the
490 # framing of the private keys section.
491 offset = len(OPENSSH_HEADER) + len(wrapped_public_key) + 4 1b
492 checkint = int.from_bytes( 1b
493 private_key_from_openssh[offset : offset + 4], "big"
494 )
495 assert minimize_openssh_keyfile_padding( 1b
496 private_key_from_openssh
497 ) == minimize_openssh_keyfile_padding(
498 as_openssh_keyfile_payload(
499 public_key=key.public_key_data,
500 private_key=key.private_key_blob,
501 checkint=checkint,
502 )
503 )
506class TestStubbedSSHAgentSocket:
507 """Test the stubbed SSH agent socket: common machinery."""
509 @contextlib.contextmanager
510 def _get_agent(
511 self, *, extended_agent: bool | None = False
512 ) -> Generator[machinery.StubbedSSHAgentSocket, None, None]:
513 agent_class: type[machinery.StubbedSSHAgentSocket] = ( 1iflgdcej
514 machinery.StubbedSSHAgentSocketWithAddressAndDeterministicDSA
515 if extended_agent
516 else machinery.StubbedSSHAgentSocketWithAddress
517 if extended_agent is not None
518 else machinery.StubbedSSHAgentSocket
519 )
520 with contextlib.ExitStack() as stack: 1iflgdcej
521 monkeypatch = stack.enter_context(pytest.MonkeyPatch.context()) 1iflgdcej
522 if issubclass( 1iflgdcej
523 agent_class, machinery.StubbedSSHAgentSocketWithAddress
524 ):
525 monkeypatch.setenv("SSH_AUTH_SOCK", agent_class.ADDRESS) 1gd
526 else:
527 monkeypatch.delenv("SSH_AUTH_SOCK", raising=False) 1iflgdcej
528 yield stack.enter_context(agent_class()) 1iflgdcej
531class TestStubbedSSHAgentSocketRequests(TestStubbedSSHAgentSocket):
532 """Test the stubbed SSH agent socket: normal requests."""
534 @Parametrize.QUERY_EXTENSION
535 def test_query_extensions(
536 self, extended_agent: bool, query_response: bytes
537 ) -> None:
538 """The agent implements a known list of extensions.
540 The list is empty for the base agent, and non-empty for the
541 extended agent.
543 """
544 query_request = ( 1g
545 # SSH string header
546 b"\x00\x00\x00\x0a"
547 # request code: SSH_AGENTC_EXTENSION
548 b"\x1b"
549 # payload: SSH string "query"
550 b"\x00\x00\x00\x05query"
551 )
552 with self._get_agent(extended_agent=extended_agent) as agent: 1g
553 assert ("query" in agent.enabled_extensions) == bool( 1g
554 extended_agent
555 )
556 agent.sendall(query_request) 1g
557 assert agent.recv(1000) == query_response 1g
559 @Parametrize.LIST_IDENTITIES
560 def test_request_identities(
561 self, extended_agent: bool | None, query_request: bytes
562 ) -> None:
563 """The agent implements a known list of identities.
565 The extended agent implements PuTTY's `list-extended` extension.
567 """
568 unstring_prefix = ssh_agent.SSHAgentClient.unstring_prefix 1d
569 with self._get_agent(extended_agent=extended_agent) as agent: 1d
570 agent.sendall(query_request) 1d
571 message_length = int.from_bytes(agent.recv(4), "big") 1d
572 orig_message: bytes | bytearray = bytearray( 1d
573 agent.recv(message_length)
574 )
575 assert ( 1d
576 orig_message[0] == _types.SSH_AGENT.SUCCESS
577 if extended_agent
578 else _types.SSH_AGENT.IDENTITIES_ANSWER
579 )
580 identity_count = int.from_bytes(orig_message[1:5], "big") 1d
581 message = bytes(orig_message[5:]) 1d
582 for _ in range(identity_count): 1d
583 key, message = unstring_prefix(message) 1d
584 _comment, message = unstring_prefix(message) 1d
585 flags, message = ( 1d
586 unstring_prefix(message)
587 if extended_agent
588 else (b"", message)
589 )
590 assert not extended_agent or flags == b"\x00\x00\x00\x00" 1d
591 assert key 1d
592 assert key in { 1d
593 k.public_key_data for k in data.ALL_KEYS.values()
594 }
595 assert not message 1d
597 @Parametrize.SUPPORTED_SSH_TEST_KEYS
598 def test_sign(
599 self,
600 ssh_test_key_type: str,
601 ssh_test_key: data.SSHTestKey,
602 ) -> None:
603 """The agent signs known key/message pairs."""
604 del ssh_test_key_type 1m
605 spec = data.SSHTestKeyDeterministicSignatureClass.SPEC 1m
606 assert ssh_test_key.expected_signatures[spec].signature is not None 1m
607 string = ssh_agent.SSHAgentClient.string 1m
608 query_request = string( 1m
609 # request code: SSH_AGENTC_SIGN_REQUEST
610 b"\x0d"
611 # key: SSH string of the public key
612 + string(ssh_test_key.public_key_data)
613 # payload: SSH string of the vault UUID
614 + string(vault.Vault.UUID)
615 # signing flags (uint32, empty)
616 + b"\x00\x00\x00\x00"
617 )
618 query_response = string( 1m
619 # response code: SSH_AGENT_SIGN_RESPONSE
620 b"\x0e"
621 # expected payload: the binary signature as recorded in the test key data structure
622 + string(ssh_test_key.expected_signatures[spec].signature)
623 )
624 with machinery.StubbedSSHAgentSocket() as agent: 1m
625 agent.sendall(query_request) 1m
626 assert agent.recv(1000) == query_response 1m
629class TestStubbedSSHAgentSocketProperOperations(TestStubbedSSHAgentSocket):
630 """Test the stubbed SSH agent socket: proper use and misuse."""
632 def test_close_multiple(
633 self,
634 ) -> None:
635 """The agent can be closed repeatedly."""
636 with self._get_agent(extended_agent=None) as agent: 1i
637 pass 1i
638 with agent: 1i
639 pass 1i
640 del agent 1i
642 def test_closed_agents_cannot_be_interacted_with(
643 self,
644 ) -> None:
645 """The agent cannot be usefully used after close."""
646 with self._get_agent(extended_agent=None) as agent: 1f
647 pass 1f
648 query_request = ( 1f
649 # SSH string header
650 b"\x00\x00\x00\x0a"
651 # request code: SSH_AGENTC_EXTENSION
652 b"\x1b"
653 # payload: SSH string "query"
654 b"\x00\x00\x00\x05query"
655 )
656 with pytest.raises(OSError, match=re.escape(os.strerror(errno.EBADF))): 1f
657 agent.sendall(query_request) 1f
658 with pytest.raises(OSError, match=re.escape(os.strerror(errno.EBADF))): 1f
659 agent.recv(100) 1f
661 def test_no_recv_without_sendall(
662 self,
663 ) -> None:
664 """The agent requires a message before sending a response."""
665 with self._get_agent(extended_agent=None) as agent: # noqa: SIM117 1l
666 with pytest.raises( 1l
667 AssertionError,
668 match=re.escape(
669 machinery.StubbedSSHAgentSocket._PROTOCOL_VIOLATION
670 ),
671 ):
672 agent.recv(100) 1l
674 @hypothesis.given(
675 query_request_parts=Strategies.proper_bytestring_partition(
676 b"\x00\x00\x00\x0a\x1b\x00\x00\x00\x05query"
677 )
678 )
679 @hypothesis.example(
680 query_request_parts=[
681 b"",
682 b"\x00\x00\x00",
683 b"\x0a",
684 b"\x1b",
685 b"\x00\x00\x00\x05",
686 b"query",
687 ]
688 )
689 def test_piecemeal_sendall(
690 self,
691 query_request_parts: list[bytes],
692 ) -> None:
693 """The agent supports receiving messages incrementally."""
694 with self._get_agent(extended_agent=None) as agent: 1c
695 agent.enabled_extensions = frozenset({"query"}) 1c
696 for part in query_request_parts: 1c
697 agent.sendall(part) 1c
698 header = agent.recv(agent.HEADER_SIZE) 1c
699 count = int.from_bytes(header, "big", signed=False) 1c
700 payload = agent.recv(count) 1c
701 assert len(payload) == count 1c
702 assert bytes.startswith( 1c
703 payload,
704 (
705 bytes(_types.SSH_AGENT.SUCCESS),
706 bytes(_types.SSH_AGENT.EXTENSION_RESPONSE),
707 ),
708 )
709 code = payload[: agent.CODE_SIZE] 1c
710 assert code == bytes(_types.SSH_AGENT.SUCCESS) or code == bytes( 1c
711 _types.SSH_AGENT.EXTENSION_RESPONSE
712 )
714 @hypothesis.given(message=Strategies.invalid_ssh_agent_messages())
715 def test_invalid_ssh_agent_messages( 1ao
716 self,
717 message: Buffer,
718 ) -> None:
719 """The agent responds with errors on invalid messages."""
720 query_response = ( 1o
721 # SSH string header
722 b"\x00\x00\x00\x01"
723 # response code: SSH_AGENT_FAILURE
724 b"\x05"
725 )
726 with machinery.StubbedSSHAgentSocket() as agent: 1o
727 agent.sendall(message) 1o
728 assert agent.recv(100) == query_response 1o
730 @hypothesis.given(message=Strategies.truncated_agent_request())
731 @hypothesis.example(message=b"\x00\x00\x00\x0f\xff") 1ae
732 def test_truncated_ssh_agent_message(
733 self,
734 message: bytes,
735 ) -> None:
736 """The agent diagnoses a final truncated request message."""
737 with pytest.raises( # noqa: SIM117 1e
738 AssertionError,
739 match=re.escape(
740 machinery.StubbedSSHAgentSocket._INCOMPLETE_REQUEST
741 ),
742 ):
743 with self._get_agent(extended_agent=None) as agent: 1e
744 agent.sendall(message) 1e
747class TestStubbedSSHAgentSocketSupportedAndUnsupportedFeatures(
748 TestStubbedSSHAgentSocket
749):
750 """Test the stubbed SSH agent socket: supported/unsupported features."""
752 @hypothesis.given(message=Strategies.unsupported_ssh_agent_messages())
753 def test_unsupported_ssh_agent_messages( 1aj
754 self,
755 message: Buffer,
756 ) -> None:
757 """The agent responds with errors on unsupported messages."""
758 query_response = ( 1j
759 # SSH string header
760 b"\x00\x00\x00\x01"
761 # response code: SSH_AGENT_FAILURE
762 b"\x05"
763 )
764 with self._get_agent(extended_agent=None) as agent: 1j
765 agent.sendall(message) 1j
766 assert agent.recv(100) == query_response 1j
768 @hypothesis.given(args_tuple=Strategies.stubbed_agent_addresses())
769 @hypothesis.example(args_tuple=(None, KeyError, "SSH_AUTH_SOCK")) 1ak
770 @hypothesis.example(args_tuple=("stub-ssh-agent:", None, ""))
771 def test_addresses(
772 self,
773 args_tuple: tuple[str | None, type[Exception] | None, str],
774 ) -> None:
775 """The agent accepts addresses."""
776 address, exception, match = args_tuple 1k
777 with contextlib.ExitStack() as stack: 1k
778 monkeypatch = stack.enter_context(pytest.MonkeyPatch.context()) 1k
779 if address: 1k
780 monkeypatch.setenv("SSH_AUTH_SOCK", address) 1k
781 else:
782 monkeypatch.delenv("SSH_AUTH_SOCK", raising=False) 1k
783 if exception: 1k
784 stack.enter_context( 1k
785 pytest.raises(exception, match=re.escape(match))
786 )
787 machinery.StubbedSSHAgentSocketWithAddress() 1k