Coverage for src / derivepassphrase / ssh_agent / __init__.py: 100.000%
199 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"""A bare-bones SSH agent client supporting signing and key listing."""
7from __future__ import annotations
9import collections
10import contextlib
11import sys
12from collections.abc import Sequence
13from typing import TYPE_CHECKING, ClassVar, overload
15from typing_extensions import Never, Self, assert_type
17from derivepassphrase import _types
19if sys.version_info < (3, 11):
20 from exceptiongroup import ExceptionGroup
22if TYPE_CHECKING:
23 from collections.abc import Generator, Iterable
24 from collections.abc import Set as AbstractSet
25 from types import TracebackType
27 from typing_extensions import Buffer
29__all__ = ("SSHAgentClient",)
31# In SSH bytestrings, the "length" of the byte string is stored as
32# a 4-byte/32-bit unsigned integer at the beginning.
33HEAD_LEN = 4
35# Needed for handling OpenSSH bug 3967. See
36# _check_for_extra_success_response_after_query_extension_request.
37_OPENSSH_AGENT_10_3_QUERY_EXTENSION_RESPONSE_PAYLOAD = bytes.fromhex("""
3800 00 00 05 71 75 65 72 79
3900 00 00 18 73 65 73 73 69 6f 6e 2d
4062 69 6e 64 40 6f 70 65 6e 73 73 68 2e 63 6f 6d
41""")
44def _check_for_extra_empty_success_response_after_query_extension_request(
45 client: SSHAgentClient, response_data: bytes | bytearray
46) -> None:
47 """Check for extra `SUCCESS` responses after a "query" extension request.
49 OpenSSH's `ssh-agent` version 10.3 erroneously answers a "query"
50 extension request with two response messages: the expected
51 [`EXTENSION_RESPONSE`][_types.SSH_AGENT.EXTENSION_RESPONSE] or
52 [`SUCCESS`][_types.SSH_AGENT.SUCCESS] message, and the unexpected
53 extra `SUCCESS` message. This causes a misalignment of the request
54 and response messages if further requests are later issued to the
55 agent.
57 ([OpenSSH acknowledges this as Bug 3967][OPENSSH_BZ3967], and
58 intends to fix this in version 10.4.)
60 This callback is intended to run immediately after issuing a "query"
61 extension request to the agent, and after reading its response. If
62 we detect the aforementioned OpenSSH behavior, then we re-align the
63 request and response messages queue. Otherwise, we do nothing. In
64 particular, it is safe to call us (at the aforementioned time) even
65 on different SSH agents, or on versions of OpenSSH's agent that do
66 not exhibit this bug.
68 [OPENSSH_BZ3967]: https://bugzilla.mindrot.org/show_bug.cgi?id=3967
70 Args:
71 client:
72 The client connected to the SSH agent that may or may not
73 exhibit this OpenSSH behavior. We assume that the client
74 has just issued a "query" extension request, that the client
75 has read and parsed the (first) response of the agent, that
76 the response message was an `EXTENSION_RESPONSE` or
77 a `SUCCESS` message, and that the response message (except
78 for the response code) is exactly the contents of the
79 `response_data` argument.
80 response_data:
81 The payload of the response message to a "query" extension
82 request issued to the SSH agent immediately prior.
84 """
85 if response_data == _OPENSSH_AGENT_10_3_QUERY_EXTENSION_RESPONSE_PAYLOAD: 1dfbig
86 # We cannot blindly read from the agent socket to check
87 # for a possible extra SUCCESS response: that would
88 # deadlock if the extra response is missing. Instead,
89 # we issue a second "query" request, and depending on
90 # whether we next see the extra empty SUCCESS response or not,
91 # we know whether to discard an additional two response
92 # messages or not.
93 #
94 # (Principally any request that does *not* ellicit
95 # a SUCCESS response and which does not modify the
96 # agent's state would do for this.)
97 response_code, response_data2 = client.request( 1fbi
98 _types.SSH_AGENTC.REQUEST_IDENTITIES, b"", response_code=None
99 )
100 if response_code == _types.SSH_AGENT.SUCCESS and not response_data2: 1fbi
101 response_code, response_data2 = client._get_response_pair() # noqa: SLF001 1fbi
102 assert response_code == _types.SSH_AGENT.IDENTITIES_ANSWER, ( 1fbi
103 "failed to re-synchronize with agent via REQUEST_IDENTITIES "
104 'after "query" EXTENSION_REQUEST: want IDENTITIES_ANSWER, '
105 f"got {response_code!r}, {response_data2!r}"
106 )
109class TrailingDataError(RuntimeError):
110 """The result contained trailing data."""
112 def __init__(
113 self, raw: bytes | bytearray = b"", trailing: bytes | bytearray = b""
114 ) -> None:
115 super().__init__("Overlong response from SSH agent") 1coh
116 self.raw = bytes(raw) 1coh
117 self.trailing = bytes(trailing) 1coh
119 def __str__(self) -> str: # pragma: no cover [debug]
120 if self.raw or self.trailing: 1oh
121 return ( 1h
122 f"Overlong response from SSH agent: " 1h
123 f"raw = {self.raw!r}, trailing = {self.trailing!r}" 1h
124 )
125 return super().__str__() 1o
127 def __repr__(self) -> str: # pragma: no cover [debug]
128 """""" # noqa: D419
129 return (
130 f"{self.__class__.__name__}"
131 f"(raw={self.raw!r}, trailing={self.trailing!r})"
132 )
135class SSHAgentFailedError(RuntimeError):
136 """The SSH agent failed to complete the requested operation."""
138 def __str__(self) -> str:
139 # TODO(the-13th-letter): Rewrite using structural pattern matching.
140 # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9
141 if self.args == ( # pragma: no branch 1rhkj
142 _types.SSH_AGENT.FAILURE,
143 b"",
144 ):
145 return "The SSH agent failed to complete the request" 1rhj
146 elif self.args[1]: # noqa: RET505 # pragma: no cover [failsafe] 1k
147 code = self.args[0] 1k
148 msg = self.args[1] 1k
149 return f"[Code {code:d}] {msg!r}" 1k
150 else: # pragma: no cover [failsafe]
151 return repr(self)
153 def __repr__(self) -> str: # pragma: no cover [debug]
154 """""" # noqa: D419
155 return f"{self.__class__.__name__}{self.args!r}"
158class SSHAgentClient:
159 """A bare-bones SSH agent client supporting signing and key listing.
161 The main use case is requesting the agent sign some data, after
162 checking that the necessary key is already loaded.
164 The main fleshed out methods are [`list_keys`][] and [`sign`][],
165 which implement the [`REQUEST_IDENTITIES`]
166 [_types.SSH_AGENTC.REQUEST_IDENTITIES] and [`SIGN_REQUEST`]
167 [_types.SSH_AGENTC.SIGN_REQUEST] requests. If you *really* wanted
168 to, there is enough infrastructure in place to issue other requests
169 as defined in the protocol---it's merely the wrapper functions and
170 the protocol numbers table that are missing.
172 """
174 _connection: _types.SSHAgentSocket
175 SOCKET_PROVIDERS: ClassVar = ("native",)
176 """
177 The default list of SSH agent socket providers.
178 """
180 def __init__( # noqa: C901, PLR0912
181 self,
182 /,
183 *,
184 socket: _types.SSHAgentSocket | Sequence[str] | None = None,
185 ) -> None:
186 """Initialize the client.
188 Args:
189 socket:
190 An optional socket-like object, already connected to the
191 SSH agent, or a list of names of socket providers to try.
192 If not given, we query platform-specific default
193 addresses, if possible.
195 Raises:
196 derivepassphrase.ssh_agent.socketprovider.NoSuchProviderError:
197 The list of socket provider names contained an invalid
198 entry.
199 KeyError:
200 An expected configuration entry or an expected environment
201 variable is missing.
202 NotImplementedError:
203 The named socket provider is not functional or not
204 applicable to this `derivepassphrase` installation, e.g.
205 because it is generally not implemented yet, or it requires
206 a specific operating system, or specific system
207 functionality that is not provided by all Python versions,
208 or external packages that are unavailable on this
209 installation.
211 This error may be raised multiple times, as an exception
212 group.
213 OSError:
214 There was an error setting up a socket connection to the
215 agent.
217 """ # noqa: DOC501
218 import socket as _socket # noqa: PLC0415 1asqpdcntvCemwxurofAUBLhgkjl
220 from derivepassphrase.ssh_agent import socketprovider # noqa: PLC0415 1asqpdcntvCemwxurofAUBLhgkjl
222 # On POSIX, the predominant type of communication channel is the
223 # UNIX domain socket, which in Python is represented as
224 # a `socket.socket` object. This is (as of January 2026)
225 # unsupported on The Annoying OS: Only in Windows 10 (on some
226 # later builds at least) did Microsoft ship a partial
227 # implementation of named UNIX domain sockets in stream mode,
228 # but because of the limited scope Python does not provide
229 # bindings to this functionality. (The Annoying OS has other
230 # predominant communication channels, but these are not
231 # represented as `socket.socket` objects.)
232 #
233 # Thus, for practical purposes, using a `socket.socket` object
234 # to interface a running SSH agent is not a valid code path on
235 # The Annoying OS. The code path should therefore be marked as
236 # such.
237 #
238 # https://devblogs.microsoft.com/commandline/af_unix-comes-to-windows/
239 # https://github.com/microsoft/WSL/issues/4240
240 # https://github.com/python/cpython/issues/77589
241 if isinstance( 1asqpdcntvCemwxurofAUBLhgkjl
242 socket, _socket.socket
243 ): # pragma: unless posix no cover [external]
244 self._connection = socket 1dL
245 # Test whether the socket is connected.
246 self._connection.getpeername() 1dL
247 elif isinstance(socket, str): 1asqpdcntvCemwxurofAUBLhgkjl
248 self._connection = socketprovider.SocketProvider.resolve(socket)() 1U
249 elif socket is None or isinstance(socket, Sequence): 1asqpdcntvCemwxurofABLhgkjl
250 if not socket: 1asqpdcntvCemwxurofABhgkjl
251 socket = self.SOCKET_PROVIDERS 1aspdcntemwxurofAhgkjl
252 assert isinstance(socket, Sequence) # for the type checker 1asqpdcntvCemwxurofABhgkjl
253 excs: list[NotImplementedError] = [] 1asqpdcntvCemwxurofABhgkjl
254 providers: list[_types.SSHAgentSocketProvider] = [] 1asqpdcntvCemwxurofABhgkjl
255 for candidate in socket: 1asqpdcntvCemwxurofABhgkjl
256 try: 1asqpdcntvCemwxurofABhgkjl
257 provider = socketprovider.SocketProvider.resolve(candidate) 1asqpdcntvCemwxurofABhgkjl
258 except NotImplementedError as exc: 1CB
259 excs.append(exc) 1B
260 continue 1B
261 else:
262 providers.append(provider) 1asqpdcntvemwxurofAhgkjl
263 for provider in providers: 1asqpdcntvemwxurofABhgkjl
264 try: 1asqpdcntvemwxurofAhgkjl
265 self._connection = provider() 1asqpdcntvemwxurofAhgkjl
266 except NotImplementedError as exc: 1asqpcwxA
267 excs.append(exc) 1sqpc
268 continue 1sqpc
269 else:
270 break 1aqpdcntvemurofhgkjl
271 else:
272 msg = "No supported SSH agent socket provider found." 1sqpcB
273 raise ( 1sqpcB
274 ExceptionGroup(msg, excs)
275 if excs
276 else NotImplementedError(msg)
277 )
278 elif isinstance(socket, _types.SSHAgentSocket): 1aL
279 self._connection = socket 1aL
280 else: # pragma: no cover [failsafe]
281 assert_type(socket, Never)
282 msg = f"invalid socket object: {socket!r}"
283 raise TypeError(msg)
285 def __enter__(self) -> Self:
286 """Close socket connection upon context manager completion.
288 Returns:
289 Self.
291 """
292 self._connection.__enter__() 1aqpdcntvemurofhgkj
293 return self 1aqpdcntvemurofhgkj
295 def __exit__(
296 self,
297 exc_type: type[BaseException] | None,
298 exc_val: BaseException | None,
299 exc_tb: TracebackType | None,
300 ) -> bool:
301 """Close socket connection upon context manager completion.
303 Args:
304 exc_type: An optional exception type.
305 exc_val: An optional exception value.
306 exc_tb: An optional exception traceback.
308 Returns:
309 True if the exception was handled, false if it should
310 propagate.
312 """
313 return bool( 1aqpdcntvemurofhgkj
314 self._connection.__exit__(exc_type, exc_val, exc_tb) # type: ignore[func-returns-value]
315 )
317 @staticmethod
318 def uint32(num: int, /) -> bytes:
319 r"""Format the number as a `uint32`, as per the agent protocol.
321 Args:
322 num: A number.
324 Returns:
325 The number in SSH agent wire protocol format, i.e. as
326 a 32-bit big endian number.
328 Raises:
329 OverflowError:
330 As per [`int.to_bytes`][].
332 Examples:
333 >>> SSHAgentClient.uint32(16777216)
334 b'\x01\x00\x00\x00'
336 """
337 return int.to_bytes(num, 4, "big", signed=False) 1aMEFGHNDOIdcefbigkjPQylWXYZz
339 @classmethod
340 def string(cls, payload: Buffer, /) -> bytes:
341 r"""Format the payload as an SSH string, as per the agent protocol.
343 Args:
344 payload: A bytes-like object.
346 Returns:
347 The payload, framed in the SSH agent wire protocol format,
348 as a bytes object.
350 Examples:
351 >>> SSHAgentClient.string(b"ssh-rsa")
352 b'\x00\x00\x00\x07ssh-rsa'
354 """ # noqa: DOC501
355 try: 1aMEFGHNDOIdcefbigkjPVQylz
356 payload = memoryview(payload) 1aMEFGHNDOIdcefbigkjPVQylz
357 except TypeError as e: 1V
358 msg = "invalid payload type" 1V
359 raise TypeError(msg) from e 1V
360 ret = bytearray() 1aMEFGHNDOIdcefbigkjPQylz
361 ret.extend(cls.uint32(len(payload))) 1aMEFGHNDOIdcefbigkjPQylz
362 ret.extend(payload) 1aMEFGHNDOIdcefbigkjPQylz
363 return bytes(ret) 1aMEFGHNDOIdcefbigkjPQylz
365 @classmethod
366 def unstring(cls, bytestring: Buffer, /) -> bytes:
367 r"""Unpack an SSH string.
369 Args:
370 bytestring: A framed bytes-like object.
372 Returns:
373 The payload, as a bytes object.
375 Raises:
376 ValueError:
377 The byte string is not an SSH string.
379 Examples:
380 >>> SSHAgentClient.unstring(b"\x00\x00\x00\x07ssh-rsa")
381 b'ssh-rsa'
382 >>> SSHAgentClient.unstring(SSHAgentClient.string(b"ssh-ed25519"))
383 b'ssh-ed25519'
385 """
386 bytestring = memoryview(bytestring) 1aembyKJz
387 n = len(bytestring) 1aembyKJz
388 msg = "malformed SSH byte string" 1aembyKJz
389 if n < HEAD_LEN or n != HEAD_LEN + int.from_bytes( 1aembyKJz
390 bytestring[:HEAD_LEN], "big", signed=False
391 ):
392 raise ValueError(msg) 1J
393 return bytes(bytestring[HEAD_LEN:]) 1aembyKz
395 @classmethod
396 def unstring_prefix(cls, bytestring: Buffer, /) -> tuple[bytes, bytes]:
397 r"""Unpack an SSH string at the beginning of the byte string.
399 Args:
400 bytestring:
401 A bytes-like object, beginning with a framed/SSH byte
402 string.
404 Returns:
405 A 2-tuple `(a, b)`, where `a` is the unframed byte
406 string/payload at the beginning of input byte string, and
407 `b` is the remainder of the input byte string.
409 Raises:
410 ValueError:
411 The byte string does not begin with an SSH string.
413 Examples:
414 >>> SSHAgentClient.unstring_prefix(
415 ... b"\x00\x00\x00\x07ssh-rsa____trailing data"
416 ... )
417 (b'ssh-rsa', b'____trailing data')
418 >>> SSHAgentClient.unstring_prefix(
419 ... SSHAgentClient.string(b"ssh-ed25519")
420 ... )
421 (b'ssh-ed25519', b'')
423 """
424 bytestring = memoryview(bytestring).toreadonly() 1aEFGHDIdnemRSTfbigyKJz
425 n = len(bytestring) 1aEFGHDIdnemRSTfbigyKJz
426 msg = "malformed SSH byte string" 1aEFGHDIdnemRSTfbigyKJz
427 if n < HEAD_LEN: 1aEFGHDIdnemRSTfbigyKJz
428 raise ValueError(msg) 1DgJ
429 m = int.from_bytes(bytestring[:HEAD_LEN], "big", signed=False) 1aEFGHDIdnemRSTfbigyKJz
430 if m + HEAD_LEN > n: 1aEFGHDIdnemRSTfbigyKJz
431 raise ValueError(msg) 1gJ
432 return ( 1aEFGHDIdnemRSTfbigyKJz
433 bytes(bytestring[HEAD_LEN : m + HEAD_LEN]),
434 bytes(bytestring[m + HEAD_LEN :]),
435 )
437 @classmethod
438 @contextlib.contextmanager
439 def ensure_agent_subcontext(
440 cls,
441 conn: SSHAgentClient
442 | _types.SSHAgentSocket
443 | Sequence[str]
444 | None = None,
445 ) -> Generator[SSHAgentClient, None, None]:
446 """Return an SSH agent client subcontext.
448 If necessary, construct an SSH agent client first using the
449 connection hint.
451 Args:
452 conn:
453 If an existing SSH agent client, then enter a context
454 within this client's scope. After exiting the context,
455 the client persists, including its socket.
457 If a socket, then construct a client using this socket,
458 then enter a context within this client's scope. After
459 exiting the context, the client is destroyed and the
460 socket is closed.
462 If `None`, or a list of names of socket providers, then
463 construct a client according to those connection hints
464 ("auto-discovery"), and enter a context within this
465 client's scope. After exiting the context, both the
466 client and its socket are destroyed.
468 Yields:
469 When entering this context, return the SSH agent client.
471 Raises:
472 derivepassphrase.ssh_agent.socketprovider.NoSuchProviderError:
473 As per [`__init__`][SSHAgentClient].
474 KeyError:
475 As per [`__init__`][SSHAgentClient], including the
476 multiple raise as an exception group.
477 NotImplementedError:
478 As per [`__init__`][SSHAgentClient].
479 OSError:
480 As per [`__init__`][SSHAgentClient].
482 """ # noqa: DOC501
483 # TODO(the-13th-letter): Rewrite using structural pattern matching.
484 # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9
485 if isinstance(conn, SSHAgentClient): 1asqpdcntvCemwxurofbihgj
486 with contextlib.nullcontext(): 1dcembi
487 yield conn 1dcembi
488 elif ( 1asqpdcntvCemwxurofhgj
489 isinstance(conn, (_types.SSHAgentSocket, str, Sequence))
490 or conn is None
491 ):
492 with SSHAgentClient(socket=conn) as client: 1asqpdcntvCemwxurofhgj
493 yield client 1aqpdcntvemurofhgj
494 else: # pragma: no cover [failsafe]
495 assert_type(conn, Never)
496 msg = f"invalid connection hint: {conn!r}"
497 raise TypeError(msg)
499 def _agent_is_pageant(self) -> bool:
500 """Return True if we are connected to Pageant.
502 Warning:
503 This is a heuristic, not a verified query or computation.
505 """
506 return ( 1debi
507 b"list-extended@putty.projects.tartarus.org"
508 in self.query_extensions()
509 )
511 def has_deterministic_dsa_signatures(self) -> bool:
512 """Check whether the agent returns deterministic DSA signatures.
514 This includes ECDSA signatures.
516 Generally, this means that the SSH agent implements [RFC 6979][]
517 or a similar system.
519 [RFC 6979]: https://www.rfc-editor.org/rfc/rfc6979
521 Returns:
522 True if a known agent was detected where signatures are
523 deterministic for all DSA key types, false otherwise.
525 Note: Known agents with deterministic signatures
526 | agent | detected via |
527 |:----------------|:--------------------------------------------------------------|
528 | Pageant (PuTTY) | `list-extended@putty.projects.tartarus.org` extension request |
530 """ # noqa: E501
531 known_good_agents = { 1debi
532 "Pageant": self._agent_is_pageant,
533 }
534 return any( # pragma: no branch 1debi
535 v() for v in known_good_agents.values()
536 )
538 def _send_request_message(
539 self, code: int | _types.SSH_AGENTC, payload: Buffer, /
540 ) -> None:
541 """Prepare and send a generic request to the SSH agent.
543 We take care of packing, framing and sending the request.
545 Args:
546 code:
547 See [request][].
548 payload:
549 See [request][].
551 """
552 payload = memoryview(payload) 1adcebikl
553 request_message = bytearray([code]) 1adcebikl
554 request_message.extend(payload) 1adcebikl
555 self._connection.sendall(self.string(request_message)) 1adcebikl
557 def _get_response_pair(self) -> tuple[int, bytes]:
558 """Read the response for a generic request to the SSH agent.
560 Returns:
561 A 2-tuple consisting of the response code and the payload,
562 with all wire framing removed.
564 Raises:
565 EOFError:
566 The response from the SSH agent is truncated or missing.
567 OSError:
568 There was a communication error with the SSH agent.
570 """
571 chunk = self._connection.recv(HEAD_LEN) 1adcebikl
572 if len(chunk) < HEAD_LEN: 1adcebikl
573 msg = "cannot read response length" 1l
574 raise EOFError(msg) 1l
575 response_length = int.from_bytes(chunk, "big", signed=False) 1adcebikl
576 response = self._connection.recv(response_length) 1adcebikl
577 if len(response) < response_length: 1adcebikl
578 msg = "truncated response from SSH agent" 1l
579 raise EOFError(msg) 1l
580 return response[0], response[1:] 1adcebik
582 @overload
583 def request(
584 self,
585 code: int | _types.SSH_AGENTC,
586 payload: Buffer,
587 /,
588 *,
589 response_code: None = None,
590 ) -> tuple[int, bytes]: ...
592 @overload
593 def request(
594 self,
595 code: int | _types.SSH_AGENTC,
596 payload: Buffer,
597 /,
598 *,
599 response_code: Iterable[_types.SSH_AGENT | int] = frozenset({
600 _types.SSH_AGENT.SUCCESS
601 }),
602 ) -> bytes: ...
604 @overload
605 def request(
606 self,
607 code: int | _types.SSH_AGENTC,
608 payload: Buffer,
609 /,
610 *,
611 response_code: _types.SSH_AGENT | int = _types.SSH_AGENT.SUCCESS,
612 ) -> bytes: ...
614 def request(
615 self,
616 code: int | _types.SSH_AGENTC,
617 payload: Buffer,
618 /,
619 *,
620 response_code: (
621 Iterable[_types.SSH_AGENT | int] | _types.SSH_AGENT | int | None
622 ) = None,
623 ) -> tuple[int, bytes] | bytes:
624 """Issue a generic request to the SSH agent.
626 Args:
627 code:
628 The request code. See the SSH agent protocol for
629 protocol numbers to use here (and which protocol numbers
630 to expect in a response).
631 payload:
632 A bytes-like object containing the payload, or
633 "contents", of the request. Request-specific.
635 It is our responsibility to add any necessary wire
636 framing around the request code and the payload,
637 not the caller's.
638 response_code:
639 An optional response code, or a set of response codes,
640 that we expect. If given, and the actual response code
641 does not match, raise an error.
643 Returns:
644 A 2-tuple consisting of the response code and the payload,
645 with all wire framing removed.
647 If a response code was passed, then only return the payload.
649 Raises:
650 EOFError:
651 The response from the SSH agent is truncated or missing.
652 OSError:
653 There was a communication error with the SSH agent.
654 SSHAgentFailedError:
655 We expected specific response codes, but did not receive
656 any of them.
658 """
659 if isinstance(response_code, int): # pragma: no branch 1adcefbihgkjl
660 response_code = frozenset({response_code}) 1acbhkj
661 self._send_request_message(code, payload) 1adcefbihgkjl
662 actual_code, response = self._get_response_pair() 1adcefbihgkjl
663 if not response_code: # pragma: no cover [failsafe] 1adcefbihgkj
664 return actual_code, response 1fbi
665 if actual_code not in response_code: 1adcefbihgkj
666 raise SSHAgentFailedError(actual_code, response) 1aefihkj
667 return response 1adcfbihg
669 def list_keys(self) -> Sequence[_types.SSHKeyCommentPair]:
670 """Request a list of keys known to the SSH agent.
672 Returns:
673 A read-only sequence of key/comment pairs.
675 Raises:
676 EOFError:
677 The response from the SSH agent is truncated or missing.
678 OSError:
679 There was a communication error with the SSH agent.
680 TrailingDataError:
681 The response from the SSH agent is too long.
682 SSHAgentFailedError:
683 The agent failed to complete the request.
685 """
686 response = self.request( 1acbh
687 _types.SSH_AGENTC.REQUEST_IDENTITIES.value,
688 b"",
689 response_code=_types.SSH_AGENT.IDENTITIES_ANSWER,
690 )
691 response_stream = collections.deque(response) 1acbh
693 def shift(num: int) -> bytes: 1acbh
694 buf: collections.deque[int] = collections.deque() 1acbh
695 for _ in range(num): 1acbh
696 try: 1acbh
697 val = response_stream.popleft() 1acbh
698 except IndexError: 1h
699 response_stream.extendleft(reversed(buf)) 1h
700 msg = "truncated response from SSH agent" 1h
701 raise EOFError(msg) from None 1h
702 buf.append(val) 1acbh
703 return bytes(buf) 1acbh
705 key_count = int.from_bytes(shift(4), "big") 1acbh
706 keys: collections.deque[_types.SSHKeyCommentPair]
707 keys = collections.deque() 1acbh
708 for _ in range(key_count): 1acbh
709 key_size = int.from_bytes(shift(4), "big") 1acbh
710 key = shift(key_size) 1acb
711 comment_size = int.from_bytes(shift(4), "big") 1acb
712 comment = shift(comment_size) 1acb
713 # Both `key` and `comment` are not wrapped as SSH strings.
714 keys.append(_types.SSHKeyCommentPair(key, comment)) 1acb
715 if response_stream: 1acbh
716 raise TrailingDataError( 1h
717 raw=response, trailing=bytes(response_stream)
718 )
719 return keys 1acb
721 def sign(
722 self,
723 /,
724 key: Buffer,
725 payload: Buffer,
726 *,
727 flags: int = 0,
728 check_if_key_loaded: bool = False,
729 ) -> bytes:
730 """Request the SSH agent sign the payload with the key.
732 Args:
733 key:
734 The public SSH key to sign the payload with, in the same
735 format as returned by, e.g., the [`list_keys`][] method.
736 The corresponding private key must have previously been
737 loaded into the agent to successfully issue a signature.
738 payload:
739 A byte string of data to sign.
740 flags:
741 Optional flags for the signing request. Currently
742 passed on as-is to the agent. In real-world usage, this
743 could be used, e.g., to request more modern hash
744 algorithms when signing with RSA keys. (No such
745 real-world usage is currently implemented.)
746 check_if_key_loaded:
747 If true, check beforehand (via [`list_keys`][]) if the
748 corresponding key has been loaded into the agent.
750 Returns:
751 The binary signature of the payload under the given key.
753 Raises:
754 EOFError:
755 The response from the SSH agent is truncated or missing.
756 OSError:
757 There was a communication error with the SSH agent.
758 TrailingDataError:
759 The response from the SSH agent is too long.
760 SSHAgentFailedError:
761 The agent failed to complete the request.
762 KeyError:
763 `check_if_key_loaded` is true, and the `key` was not
764 loaded into the agent.
766 """
767 key = memoryview(key) 1bj
768 payload = memoryview(payload) 1bj
769 if check_if_key_loaded: 1bj
770 loaded_keys = frozenset({pair.key for pair in self.list_keys()}) 1j
771 if bytes(key) not in loaded_keys: 1j
772 msg = "target SSH key not loaded into agent" 1j
773 raise KeyError(msg) 1j
774 request_data = bytearray(self.string(key)) 1bj
775 request_data.extend(self.string(payload)) 1bj
776 request_data.extend(self.uint32(flags)) 1bj
777 return bytes( 1bj
778 self.unstring(
779 self.request(
780 _types.SSH_AGENTC.SIGN_REQUEST,
781 request_data,
782 response_code=_types.SSH_AGENT.SIGN_RESPONSE,
783 )
784 )
785 )
787 def query_extensions(self) -> AbstractSet[bytes]:
788 """Request a listing of extensions supported by the SSH agent.
790 Returns:
791 A read-only set of extension names the SSH agent says it
792 supports.
794 Raises:
795 EOFError:
796 The response from the SSH agent is truncated or missing.
797 OSError:
798 There was a communication error with the SSH agent.
799 RuntimeError:
800 The response from the SSH agent is malformed.
802 Note:
803 The set of supported extensions is queried via the `query`
804 extension request. If the agent does not support the query
805 extension request, or extension requests in general, then an
806 empty set is returned. This does not however imply that the
807 agent doesn't support *any* extension request... merely that
808 it doesn't support extension autodiscovery.
810 """
811 try: 1defbig
812 # The draft-miller-ssh-agent-14 (2024-04-15) version of the
813 # SSH agent protocol specification introduced the
814 # SSH_AGENT_EXTENSION_RESPONSE protocol message, and
815 # suggested the use of this message in the response for
816 # SSH_AGENTC_EXTENSION requests. Older versions of the spec
817 # used the SSH_AGENT_SUCCESS response, and even the newer
818 # spec versions contain wording that allows such usage
819 # across all SSH_AGENTC_EXTENSION requests. Both response
820 # codes are used in the wild (`ssh-agent`/OpenSSH 10.3,
821 # `pageant`/PuTTY 0.83), so both of them need to be
822 # supported for compatibility reasons.
823 response_data = self.request( 1defbig
824 _types.SSH_AGENTC.EXTENSION,
825 self.string(b"query"),
826 response_code={
827 _types.SSH_AGENT.EXTENSION_RESPONSE,
828 _types.SSH_AGENT.SUCCESS,
829 },
830 )
831 _check_for_extra_empty_success_response_after_query_extension_request( 1dfbig
832 self, response_data
833 )
834 except SSHAgentFailedError: 1efi
835 # Cannot query extension support. Assume no extensions.
836 # This isn't necessarily true, e.g. for OpenSSH's ssh-agent
837 # 10.2 and older.
838 return frozenset() 1efi
839 extensions: set[bytes] = set() 1dfbig
840 msg = "Malformed response from SSH agent" 1dfbig
841 msg2 = "Extension response message does not match request" 1dfbig
842 try: 1dfbig
843 query, response_data = self.unstring_prefix(response_data) 1dfbig
844 except ValueError as e: 1g
845 raise RuntimeError(msg) from e 1g
846 if bytes(query) != b"query": 1dfbig
847 raise RuntimeError(msg2) 1g
848 while response_data: 1dfbig
849 try: 1dfbig
850 extension, response_data = self.unstring_prefix(response_data) 1dfbig
851 except ValueError as e: 1g
852 raise RuntimeError(msg) from e 1g
853 else:
854 extensions.add(bytes(extension)) 1dfbig
855 return frozenset(extensions) 1dfbi