derivepassphrase.git

src/derivepassphrase/ssh_agent.py            1) # SPDX-FileCopyrightText: 2024 Marco Ricci <software@the13thletter.info>

src/ssh_agent_client/__init__.py             2) #
src/ssh_agent_client/__init__.py             3) # SPDX-License-Identifier: MIT
src/ssh_agent_client/__init__.py             4) 
src/ssh_agent_client/__init__.py             5) """A bare-bones SSH agent client supporting signing and key listing."""
src/ssh_agent_client/__init__.py             6) 
src/ssh_agent_client/__init__.py             7) from __future__ import annotations
src/ssh_agent_client/__init__.py             8) 
src/ssh_agent_client/__init__.py             9) import collections
src/ssh_agent_client/__init__.py            10) import errno
src/ssh_agent_client/__init__.py            11) import os
src/ssh_agent_client/__init__.py            12) import socket

src/derivepassphrase/ssh_agent.py           13) from typing import TYPE_CHECKING, overload

src/ssh_agent_client/__init__.py            14) 

src/ssh_agent_client/__init__.py            15) from typing_extensions import Self

src/ssh_agent_client/__init__.py            16) 

src/derivepassphrase/ssh_agent/__init__.py  17) from derivepassphrase import _types

src/ssh_agent_client/__init__.py            18) 
src/ssh_agent_client/__init__.py            19) if TYPE_CHECKING:

src/derivepassphrase/ssh_agent.py           20)     from collections.abc import Iterable, Sequence

src/ssh_agent_client/__init__.py            21)     from types import TracebackType

src/ssh_agent_client/__init__.py            22) 
src/ssh_agent_client/__init__.py            23) __all__ = ('SSHAgentClient',)

src/derivepassphrase/ssh_agent.py           24) __author__ = 'Marco Ricci <software@the13thletter.info>'

src/ssh_agent_client/__init__.py            25) 

src/ssh_agent_client/__init__.py            26) # In SSH bytestrings, the "length" of the byte string is stored as
src/ssh_agent_client/__init__.py            27) # a 4-byte/32-bit unsigned integer at the beginning.
src/ssh_agent_client/__init__.py            28) HEAD_LEN = 4
src/ssh_agent_client/__init__.py            29) 

src/ssh_agent_client/__init__.py            30) _socket = socket
src/ssh_agent_client/__init__.py            31) 

src/ssh_agent_client/__init__.py            32) 

src/ssh_agent_client/__init__.py            33) class TrailingDataError(RuntimeError):
src/ssh_agent_client/__init__.py            34)     """The result contained trailing data."""

src/ssh_agent_client/__init__.py            35) 

src/ssh_agent_client/__init__.py            36)     def __init__(self) -> None:

src/ssh_agent_client/__init__.py            37)         super().__init__('Overlong response from SSH agent')

src/ssh_agent_client/__init__.py            38) 

src/ssh_agent_client/__init__.py            39) 

src/derivepassphrase/ssh_agent.py           40) class SSHAgentFailedError(RuntimeError):
src/derivepassphrase/ssh_agent.py           41)     """The SSH agent failed to complete the requested operation."""
src/derivepassphrase/ssh_agent.py           42) 
src/derivepassphrase/ssh_agent.py           43)     def __str__(self) -> str:
src/derivepassphrase/ssh_agent.py           44)         match self.args:
src/derivepassphrase/ssh_agent.py           45)             case (_types.SSH_AGENT.FAILURE.value, b''):  # pragma: no branch
src/derivepassphrase/ssh_agent.py           46)                 return 'The SSH agent failed to complete the request'
src/derivepassphrase/ssh_agent.py           47)             case (_, _msg) if _msg:  # pragma: no cover
src/derivepassphrase/ssh_agent.py           48)                 code = self.args[0]
src/derivepassphrase/ssh_agent.py           49)                 msg = self.args[1].decode('utf-8', 'surrogateescape')
src/derivepassphrase/ssh_agent.py           50)                 return f'[Code {code:d}] {msg:s}'
src/derivepassphrase/ssh_agent.py           51)             case _:  # pragma: no cover
src/derivepassphrase/ssh_agent.py           52)                 return repr(self)
src/derivepassphrase/ssh_agent.py           53) 
src/derivepassphrase/ssh_agent.py           54)     def __repr__(self) -> str:  # pragma: no cover
src/derivepassphrase/ssh_agent.py           55)         return f'{self.__class__.__name__}{self.args!r}'
src/derivepassphrase/ssh_agent.py           56) 
src/derivepassphrase/ssh_agent.py           57) 

src/ssh_agent_client/__init__.py            58) class SSHAgentClient:
src/ssh_agent_client/__init__.py            59)     """A bare-bones SSH agent client supporting signing and key listing.
src/ssh_agent_client/__init__.py            60) 
src/ssh_agent_client/__init__.py            61)     The main use case is requesting the agent sign some data, after
src/ssh_agent_client/__init__.py            62)     checking that the necessary key is already loaded.
src/ssh_agent_client/__init__.py            63) 

src/derivepassphrase/ssh_agent.py           64)     The main fleshed out methods are [`list_keys`][] and [`sign`][],
src/derivepassphrase/ssh_agent.py           65)     which implement the [`REQUEST_IDENTITIES`]
src/derivepassphrase/ssh_agent.py           66)     [_types.SSH_AGENTC.REQUEST_IDENTITIES] and [`SIGN_REQUEST`]
src/derivepassphrase/ssh_agent.py           67)     [_types.SSH_AGENTC.SIGN_REQUEST] requests.  If you *really* wanted
src/derivepassphrase/ssh_agent.py           68)     to, there is enough infrastructure in place to issue other requests
src/derivepassphrase/ssh_agent.py           69)     as defined in the protocol---it's merely the wrapper functions and
src/derivepassphrase/ssh_agent.py           70)     the protocol numbers table that are missing.

src/ssh_agent_client/__init__.py            71) 
src/ssh_agent_client/__init__.py            72)     """

src/ssh_agent_client/__init__.py            73) 

src/ssh_agent_client/__init__.py            74)     _connection: socket.socket

src/ssh_agent_client/__init__.py            75) 

src/ssh_agent_client/__init__.py            76)     def __init__(
src/ssh_agent_client/__init__.py            77)         self, /, *, socket: socket.socket | None = None, timeout: int = 125
src/ssh_agent_client/__init__.py            78)     ) -> None:
src/ssh_agent_client/__init__.py            79)         """Initialize the client.
src/ssh_agent_client/__init__.py            80) 
src/ssh_agent_client/__init__.py            81)         Args:
src/ssh_agent_client/__init__.py            82)             socket:
src/ssh_agent_client/__init__.py            83)                 An optional socket, connected to the SSH agent.  If not
src/ssh_agent_client/__init__.py            84)                 given, we query the `SSH_AUTH_SOCK` environment
src/ssh_agent_client/__init__.py            85)                 variable to auto-discover the correct socket address.
src/ssh_agent_client/__init__.py            86)             timeout:
src/ssh_agent_client/__init__.py            87)                 A connection timeout for the SSH agent.  Only used if
src/ssh_agent_client/__init__.py            88)                 the socket is not yet connected.  The default value
src/ssh_agent_client/__init__.py            89)                 gives ample time for agent connections forwarded via
src/ssh_agent_client/__init__.py            90)                 SSH on high-latency networks (e.g. Tor).
src/ssh_agent_client/__init__.py            91) 

src/ssh_agent_client/__init__.py            92)         Raises:
src/ssh_agent_client/__init__.py            93)             KeyError:
src/ssh_agent_client/__init__.py            94)                 The `SSH_AUTH_SOCK` environment was not found.
src/ssh_agent_client/__init__.py            95)             OSError:
src/ssh_agent_client/__init__.py            96)                 There was an error setting up a socket connection to the
src/ssh_agent_client/__init__.py            97)                 agent.
src/ssh_agent_client/__init__.py            98) 

src/ssh_agent_client/__init__.py            99)         """
src/ssh_agent_client/__init__.py           100)         if socket is not None:

src/ssh_agent_client/__init__.py           101)             self._connection = socket

src/ssh_agent_client/__init__.py           102)         else:

src/ssh_agent_client/__init__.py           103)             self._connection = _socket.socket(family=_socket.AF_UNIX)

src/ssh_agent_client/__init__.py           104)         try:

src/ssh_agent_client/__init__.py           105)             # Test whether the socket is connected.
src/ssh_agent_client/__init__.py           106)             self._connection.getpeername()

src/ssh_agent_client/__init__.py           107)         except OSError as e:

src/ssh_agent_client/__init__.py           108)             # This condition is hard to test purposefully, so exclude
src/ssh_agent_client/__init__.py           109)             # from coverage.
src/ssh_agent_client/__init__.py           110)             if e.errno != errno.ENOTCONN:  # pragma: no cover

src/ssh_agent_client/__init__.py           111)                 raise

src/ssh_agent_client/__init__.py           112)             if 'SSH_AUTH_SOCK' not in os.environ:

src/ssh_agent_client/__init__.py           113)                 msg = 'SSH_AUTH_SOCK environment variable'
src/ssh_agent_client/__init__.py           114)                 raise KeyError(msg) from None

src/ssh_agent_client/__init__.py           115)             ssh_auth_sock = os.environ['SSH_AUTH_SOCK']

src/ssh_agent_client/__init__.py           116)             self._connection.settimeout(timeout)

src/ssh_agent_client/__init__.py           117)             self._connection.connect(ssh_auth_sock)

src/ssh_agent_client/__init__.py           118) 
src/ssh_agent_client/__init__.py           119)     def __enter__(self) -> Self:

src/derivepassphrase/ssh_agent.py          120)         """Close socket connection upon context manager completion.
src/derivepassphrase/ssh_agent.py          121) 
src/derivepassphrase/ssh_agent.py          122)         Returns:
src/derivepassphrase/ssh_agent.py          123)             Self.
src/derivepassphrase/ssh_agent.py          124) 
src/derivepassphrase/ssh_agent.py          125)         """

src/ssh_agent_client/__init__.py           126)         self._connection.__enter__()

src/ssh_agent_client/__init__.py           127)         return self
src/ssh_agent_client/__init__.py           128) 
src/ssh_agent_client/__init__.py           129)     def __exit__(

src/ssh_agent_client/__init__.py           130)         self,
src/ssh_agent_client/__init__.py           131)         exc_type: type[BaseException] | None,
src/ssh_agent_client/__init__.py           132)         exc_val: BaseException | None,

src/ssh_agent_client/__init__.py           133)         exc_tb: TracebackType | None,

src/ssh_agent_client/__init__.py           134)     ) -> bool:

src/derivepassphrase/ssh_agent.py          135)         """Close socket connection upon context manager completion.
src/derivepassphrase/ssh_agent.py          136) 
src/derivepassphrase/ssh_agent.py          137)         Args:
src/derivepassphrase/ssh_agent.py          138)             exc_type: An optional exception type.
src/derivepassphrase/ssh_agent.py          139)             exc_val: An optional exception value.
src/derivepassphrase/ssh_agent.py          140)             exc_tb: An optional exception traceback.
src/derivepassphrase/ssh_agent.py          141) 
src/derivepassphrase/ssh_agent.py          142)         Returns:
src/derivepassphrase/ssh_agent.py          143)             True if the exception was handled, false if it should
src/derivepassphrase/ssh_agent.py          144)             propagate.
src/derivepassphrase/ssh_agent.py          145) 
src/derivepassphrase/ssh_agent.py          146)         """

src/ssh_agent_client/__init__.py           147)         return bool(

src/ssh_agent_client/__init__.py           148)             self._connection.__exit__(exc_type, exc_val, exc_tb)  # type: ignore[func-returns-value]

src/ssh_agent_client/__init__.py           149)         )
src/ssh_agent_client/__init__.py           150) 
src/ssh_agent_client/__init__.py           151)     @staticmethod

src/ssh_agent_client/__init__.py           152)     def uint32(num: int, /) -> bytes:
src/ssh_agent_client/__init__.py           153)         r"""Format the number as a `uint32`, as per the agent protocol.
src/ssh_agent_client/__init__.py           154) 
src/ssh_agent_client/__init__.py           155)         Args:
src/ssh_agent_client/__init__.py           156)             num: A number.
src/ssh_agent_client/__init__.py           157) 
src/ssh_agent_client/__init__.py           158)         Returns:
src/ssh_agent_client/__init__.py           159)             The number in SSH agent wire protocol format, i.e. as
src/ssh_agent_client/__init__.py           160)             a 32-bit big endian number.
src/ssh_agent_client/__init__.py           161) 
src/ssh_agent_client/__init__.py           162)         Raises:
src/ssh_agent_client/__init__.py           163)             OverflowError:
src/ssh_agent_client/__init__.py           164)                 As per [`int.to_bytes`][].
src/ssh_agent_client/__init__.py           165) 
src/ssh_agent_client/__init__.py           166)         Examples:
src/ssh_agent_client/__init__.py           167)             >>> SSHAgentClient.uint32(16777216)
src/ssh_agent_client/__init__.py           168)             b'\x01\x00\x00\x00'
src/ssh_agent_client/__init__.py           169) 
src/ssh_agent_client/__init__.py           170)         """

src/ssh_agent_client/__init__.py           171)         return int.to_bytes(num, 4, 'big', signed=False)
src/ssh_agent_client/__init__.py           172) 
src/ssh_agent_client/__init__.py           173)     @classmethod
src/ssh_agent_client/__init__.py           174)     def string(cls, payload: bytes | bytearray, /) -> bytes | bytearray:

src/ssh_agent_client/__init__.py           175)         r"""Format the payload as an SSH string, as per the agent protocol.
src/ssh_agent_client/__init__.py           176) 
src/ssh_agent_client/__init__.py           177)         Args:
src/ssh_agent_client/__init__.py           178)             payload: A byte string.
src/ssh_agent_client/__init__.py           179) 
src/ssh_agent_client/__init__.py           180)         Returns:
src/ssh_agent_client/__init__.py           181)             The payload, framed in the SSH agent wire protocol format.
src/ssh_agent_client/__init__.py           182) 
src/ssh_agent_client/__init__.py           183)         Examples:
src/ssh_agent_client/__init__.py           184)             >>> bytes(SSHAgentClient.string(b'ssh-rsa'))
src/ssh_agent_client/__init__.py           185)             b'\x00\x00\x00\x07ssh-rsa'
src/ssh_agent_client/__init__.py           186) 
src/ssh_agent_client/__init__.py           187)         """

src/ssh_agent_client/__init__.py           188)         try:
src/ssh_agent_client/__init__.py           189)             ret = bytearray()
src/ssh_agent_client/__init__.py           190)             ret.extend(cls.uint32(len(payload)))
src/ssh_agent_client/__init__.py           191)             ret.extend(payload)
src/ssh_agent_client/__init__.py           192)         except Exception as e:

src/ssh_agent_client/__init__.py           193)             msg = 'invalid payload type'

src/derivepassphrase/ssh_agent.py          194)             raise TypeError(msg) from e  # noqa: DOC501

src/ssh_agent_client/__init__.py           195)         return ret

src/ssh_agent_client/__init__.py           196) 
src/ssh_agent_client/__init__.py           197)     @classmethod
src/ssh_agent_client/__init__.py           198)     def unstring(cls, bytestring: bytes | bytearray, /) -> bytes | bytearray:

src/ssh_agent_client/__init__.py           199)         r"""Unpack an SSH string.
src/ssh_agent_client/__init__.py           200) 
src/ssh_agent_client/__init__.py           201)         Args:
src/ssh_agent_client/__init__.py           202)             bytestring: A framed byte string.
src/ssh_agent_client/__init__.py           203) 
src/ssh_agent_client/__init__.py           204)         Returns:
src/ssh_agent_client/__init__.py           205)             The unframed byte string, i.e., the payload.
src/ssh_agent_client/__init__.py           206) 
src/ssh_agent_client/__init__.py           207)         Raises:
src/ssh_agent_client/__init__.py           208)             ValueError:

src/ssh_agent_client/__init__.py           209)                 The byte string is not an SSH string.

src/ssh_agent_client/__init__.py           210) 
src/ssh_agent_client/__init__.py           211)         Examples:
src/ssh_agent_client/__init__.py           212)             >>> bytes(SSHAgentClient.unstring(b'\x00\x00\x00\x07ssh-rsa'))
src/ssh_agent_client/__init__.py           213)             b'ssh-rsa'

src/ssh_agent_client/__init__.py           214)             >>> bytes(
src/ssh_agent_client/__init__.py           215)             ...     SSHAgentClient.unstring(SSHAgentClient.string(b'ssh-ed25519'))
src/ssh_agent_client/__init__.py           216)             ... )

src/ssh_agent_client/__init__.py           217)             b'ssh-ed25519'
src/ssh_agent_client/__init__.py           218) 

src/ssh_agent_client/__init__.py           219)         """  # noqa: E501

src/ssh_agent_client/__init__.py           220)         n = len(bytestring)

src/ssh_agent_client/__init__.py           221)         msg = 'malformed SSH byte string'

src/ssh_agent_client/__init__.py           222)         if n < HEAD_LEN or n != HEAD_LEN + int.from_bytes(
src/ssh_agent_client/__init__.py           223)             bytestring[:HEAD_LEN], 'big', signed=False

src/ssh_agent_client/__init__.py           224)         ):
src/ssh_agent_client/__init__.py           225)             raise ValueError(msg)
src/ssh_agent_client/__init__.py           226)         return bytestring[HEAD_LEN:]

src/ssh_agent_client/__init__.py           227) 

src/ssh_agent_client/__init__.py           228)     @classmethod
src/ssh_agent_client/__init__.py           229)     def unstring_prefix(
src/ssh_agent_client/__init__.py           230)         cls, bytestring: bytes | bytearray, /
src/ssh_agent_client/__init__.py           231)     ) -> tuple[bytes | bytearray, bytes | bytearray]:
src/ssh_agent_client/__init__.py           232)         r"""Unpack an SSH string at the beginning of the byte string.
src/ssh_agent_client/__init__.py           233) 
src/ssh_agent_client/__init__.py           234)         Args:
src/ssh_agent_client/__init__.py           235)             bytestring:
src/ssh_agent_client/__init__.py           236)                 A (general) byte string, beginning with a framed/SSH
src/ssh_agent_client/__init__.py           237)                 byte string.
src/ssh_agent_client/__init__.py           238) 
src/ssh_agent_client/__init__.py           239)         Returns:
src/ssh_agent_client/__init__.py           240)             A 2-tuple `(a, b)`, where `a` is the unframed byte
src/ssh_agent_client/__init__.py           241)             string/payload at the beginning of input byte string, and
src/ssh_agent_client/__init__.py           242)             `b` is the remainder of the input byte string.
src/ssh_agent_client/__init__.py           243) 
src/ssh_agent_client/__init__.py           244)         Raises:
src/ssh_agent_client/__init__.py           245)             ValueError:
src/ssh_agent_client/__init__.py           246)                 The byte string does not begin with an SSH string.
src/ssh_agent_client/__init__.py           247) 
src/ssh_agent_client/__init__.py           248)         Examples:
src/ssh_agent_client/__init__.py           249)             >>> a, b = SSHAgentClient.unstring_prefix(

src/ssh_agent_client/__init__.py           250)             ...     b'\x00\x00\x00\x07ssh-rsa____trailing data'
src/ssh_agent_client/__init__.py           251)             ... )

src/ssh_agent_client/__init__.py           252)             >>> (bytes(a), bytes(b))
src/ssh_agent_client/__init__.py           253)             (b'ssh-rsa', b'____trailing data')
src/ssh_agent_client/__init__.py           254)             >>> a, b = SSHAgentClient.unstring_prefix(

src/ssh_agent_client/__init__.py           255)             ...     SSHAgentClient.string(b'ssh-ed25519')
src/ssh_agent_client/__init__.py           256)             ... )

src/ssh_agent_client/__init__.py           257)             >>> (bytes(a), bytes(b))
src/ssh_agent_client/__init__.py           258)             (b'ssh-ed25519', b'')
src/ssh_agent_client/__init__.py           259) 
src/ssh_agent_client/__init__.py           260)         """
src/ssh_agent_client/__init__.py           261)         n = len(bytestring)

src/ssh_agent_client/__init__.py           262)         msg = 'malformed SSH byte string'
src/ssh_agent_client/__init__.py           263)         if n < HEAD_LEN:
src/ssh_agent_client/__init__.py           264)             raise ValueError(msg)
src/ssh_agent_client/__init__.py           265)         m = int.from_bytes(bytestring[:HEAD_LEN], 'big', signed=False)
src/ssh_agent_client/__init__.py           266)         if m + HEAD_LEN > n:
src/ssh_agent_client/__init__.py           267)             raise ValueError(msg)

src/ssh_agent_client/__init__.py           268)         return (
src/ssh_agent_client/__init__.py           269)             bytestring[HEAD_LEN : m + HEAD_LEN],
src/ssh_agent_client/__init__.py           270)             bytestring[m + HEAD_LEN :],
src/ssh_agent_client/__init__.py           271)         )

src/ssh_agent_client/__init__.py           272) 

src/derivepassphrase/ssh_agent.py          273)     @overload
src/derivepassphrase/ssh_agent.py          274)     def request(  # pragma: no cover
src/derivepassphrase/ssh_agent.py          275)         self,
src/derivepassphrase/ssh_agent.py          276)         code: int | _types.SSH_AGENTC,
src/derivepassphrase/ssh_agent.py          277)         payload: bytes | bytearray,
src/derivepassphrase/ssh_agent.py          278)         /,
src/derivepassphrase/ssh_agent.py          279)         *,
src/derivepassphrase/ssh_agent.py          280)         response_code: None = None,
src/derivepassphrase/ssh_agent.py          281)     ) -> tuple[int, bytes | bytearray]: ...
src/derivepassphrase/ssh_agent.py          282) 
src/derivepassphrase/ssh_agent.py          283)     @overload
src/derivepassphrase/ssh_agent.py          284)     def request(  # pragma: no cover
src/derivepassphrase/ssh_agent.py          285)         self,
src/derivepassphrase/ssh_agent.py          286)         code: int | _types.SSH_AGENTC,
src/derivepassphrase/ssh_agent.py          287)         payload: bytes | bytearray,
src/derivepassphrase/ssh_agent.py          288)         /,
src/derivepassphrase/ssh_agent.py          289)         *,
src/derivepassphrase/ssh_agent.py          290)         response_code: Iterable[_types.SSH_AGENT | int] = frozenset({
src/derivepassphrase/ssh_agent.py          291)             _types.SSH_AGENT.SUCCESS
src/derivepassphrase/ssh_agent.py          292)         }),
src/derivepassphrase/ssh_agent.py          293)     ) -> tuple[int, bytes | bytearray]: ...
src/derivepassphrase/ssh_agent.py          294) 
src/derivepassphrase/ssh_agent.py          295)     @overload
src/derivepassphrase/ssh_agent.py          296)     def request(  # pragma: no cover
src/derivepassphrase/ssh_agent.py          297)         self,
src/derivepassphrase/ssh_agent.py          298)         code: int | _types.SSH_AGENTC,
src/derivepassphrase/ssh_agent.py          299)         payload: bytes | bytearray,
src/derivepassphrase/ssh_agent.py          300)         /,
src/derivepassphrase/ssh_agent.py          301)         *,
src/derivepassphrase/ssh_agent.py          302)         response_code: _types.SSH_AGENT | int = _types.SSH_AGENT.SUCCESS,
src/derivepassphrase/ssh_agent.py          303)     ) -> bytes | bytearray: ...
src/derivepassphrase/ssh_agent.py          304) 

src/ssh_agent_client/__init__.py           305)     def request(

src/derivepassphrase/ssh_agent.py          306)         self,
src/derivepassphrase/ssh_agent.py          307)         code: int | _types.SSH_AGENTC,
src/derivepassphrase/ssh_agent.py          308)         payload: bytes | bytearray,
src/derivepassphrase/ssh_agent.py          309)         /,
src/derivepassphrase/ssh_agent.py          310)         *,
src/derivepassphrase/ssh_agent.py          311)         response_code: (
src/derivepassphrase/ssh_agent.py          312)             Iterable[_types.SSH_AGENT | int] | _types.SSH_AGENT | int | None
src/derivepassphrase/ssh_agent.py          313)         ) = None,
src/derivepassphrase/ssh_agent.py          314)     ) -> tuple[int, bytes | bytearray] | bytes | bytearray:

src/ssh_agent_client/__init__.py           315)         """Issue a generic request to the SSH agent.
src/ssh_agent_client/__init__.py           316) 
src/ssh_agent_client/__init__.py           317)         Args:
src/ssh_agent_client/__init__.py           318)             code:
src/ssh_agent_client/__init__.py           319)                 The request code.  See the SSH agent protocol for
src/ssh_agent_client/__init__.py           320)                 protocol numbers to use here (and which protocol numbers
src/ssh_agent_client/__init__.py           321)                 to expect in a response).
src/ssh_agent_client/__init__.py           322)             payload:
src/ssh_agent_client/__init__.py           323)                 A byte string containing the payload, or "contents", of
src/ssh_agent_client/__init__.py           324)                 the request.  Request-specific.  `request` will add any
src/ssh_agent_client/__init__.py           325)                 necessary wire framing around the request code and the
src/ssh_agent_client/__init__.py           326)                 payload.

src/derivepassphrase/ssh_agent.py          327)             response_code:
src/derivepassphrase/ssh_agent.py          328)                 An optional response code, or a set of response codes,
src/derivepassphrase/ssh_agent.py          329)                 that we expect.  If given, and the actual response code
src/derivepassphrase/ssh_agent.py          330)                 does not match, raise an error.

src/ssh_agent_client/__init__.py           331) 
src/ssh_agent_client/__init__.py           332)         Returns:
src/ssh_agent_client/__init__.py           333)             A 2-tuple consisting of the response code and the payload,
src/ssh_agent_client/__init__.py           334)             with all wire framing removed.
src/ssh_agent_client/__init__.py           335) 
src/ssh_agent_client/__init__.py           336)         Raises:
src/ssh_agent_client/__init__.py           337)             EOFError:
src/ssh_agent_client/__init__.py           338)                 The response from the SSH agent is truncated or missing.

src/derivepassphrase/ssh_agent.py          339)             OSError:
src/derivepassphrase/ssh_agent.py          340)                 There was a communication error with the SSH agent.
src/derivepassphrase/ssh_agent.py          341)             SSHAgentFailedError:
src/derivepassphrase/ssh_agent.py          342)                 We expected specific response codes, but did not receive
src/derivepassphrase/ssh_agent.py          343)                 any of them.

src/ssh_agent_client/__init__.py           344) 
src/ssh_agent_client/__init__.py           345)         """

src/derivepassphrase/ssh_agent.py          346)         if isinstance(  # pragma: no branch
src/derivepassphrase/ssh_agent.py          347)             response_code, int | _types.SSH_AGENT
src/derivepassphrase/ssh_agent.py          348)         ):
src/derivepassphrase/ssh_agent.py          349)             response_code = frozenset({response_code})
src/derivepassphrase/ssh_agent.py          350)         if response_code is not None:  # pragma: no branch
src/derivepassphrase/ssh_agent.py          351)             response_code = frozenset({
src/derivepassphrase/ssh_agent.py          352)                 c if isinstance(c, int) else c.value for c in response_code
src/derivepassphrase/ssh_agent.py          353)             })
src/derivepassphrase/ssh_agent.py          354)         request_message = bytearray([
src/derivepassphrase/ssh_agent.py          355)             code if isinstance(code, int) else code.value
src/derivepassphrase/ssh_agent.py          356)         ])

src/ssh_agent_client/__init__.py           357)         request_message.extend(payload)

src/ssh_agent_client/__init__.py           358)         self._connection.sendall(self.string(request_message))

src/ssh_agent_client/__init__.py           359)         chunk = self._connection.recv(HEAD_LEN)
src/ssh_agent_client/__init__.py           360)         if len(chunk) < HEAD_LEN:
src/ssh_agent_client/__init__.py           361)             msg = 'cannot read response length'
src/ssh_agent_client/__init__.py           362)             raise EOFError(msg)

src/ssh_agent_client/__init__.py           363)         response_length = int.from_bytes(chunk, 'big', signed=False)

src/ssh_agent_client/__init__.py           364)         response = self._connection.recv(response_length)

src/ssh_agent_client/__init__.py           365)         if len(response) < response_length:

src/ssh_agent_client/__init__.py           366)             msg = 'truncated response from SSH agent'
src/ssh_agent_client/__init__.py           367)             raise EOFError(msg)

src/derivepassphrase/ssh_agent.py          368)         if not response_code:  # pragma: no cover
src/derivepassphrase/ssh_agent.py          369)             return response[0], response[1:]
src/derivepassphrase/ssh_agent.py          370)         if response[0] not in response_code:
src/derivepassphrase/ssh_agent.py          371)             raise SSHAgentFailedError(response[0], response[1:])
src/derivepassphrase/ssh_agent.py          372)         return response[1:]

src/ssh_agent_client/__init__.py           373) 

src/derivepassphrase/ssh_agent/__init__.py 374)     def list_keys(self) -> Sequence[_types.KeyCommentPair]:

src/ssh_agent_client/__init__.py           375)         """Request a list of keys known to the SSH agent.
src/ssh_agent_client/__init__.py           376) 
src/ssh_agent_client/__init__.py           377)         Returns:
src/ssh_agent_client/__init__.py           378)             A read-only sequence of key/comment pairs.
src/ssh_agent_client/__init__.py           379) 
src/ssh_agent_client/__init__.py           380)         Raises:
src/ssh_agent_client/__init__.py           381)             EOFError:
src/ssh_agent_client/__init__.py           382)                 The response from the SSH agent is truncated or missing.

src/derivepassphrase/ssh_agent.py          383)             OSError:
src/derivepassphrase/ssh_agent.py          384)                 There was a communication error with the SSH agent.

src/ssh_agent_client/__init__.py           385)             TrailingDataError:

src/ssh_agent_client/__init__.py           386)                 The response from the SSH agent is too long.

src/derivepassphrase/ssh_agent.py          387)             SSHAgentFailedError:

src/ssh_agent_client/__init__.py           388)                 The agent failed to complete the request.

src/ssh_agent_client/__init__.py           389) 
src/ssh_agent_client/__init__.py           390)         """

src/derivepassphrase/ssh_agent.py          391)         response = self.request(
src/derivepassphrase/ssh_agent.py          392)             _types.SSH_AGENTC.REQUEST_IDENTITIES.value,
src/derivepassphrase/ssh_agent.py          393)             b'',
src/derivepassphrase/ssh_agent.py          394)             response_code=_types.SSH_AGENT.IDENTITIES_ANSWER,

src/ssh_agent_client/__init__.py           395)         )

src/ssh_agent_client/__init__.py           396)         response_stream = collections.deque(response)

src/ssh_agent_client/__init__.py           397) 

src/ssh_agent_client/__init__.py           398)         def shift(num: int) -> bytes:

src/ssh_agent_client/__init__.py           399)             buf = collections.deque(b'')
src/ssh_agent_client/__init__.py           400)             for _ in range(num):

src/ssh_agent_client/__init__.py           401)                 try:
src/ssh_agent_client/__init__.py           402)                     val = response_stream.popleft()
src/ssh_agent_client/__init__.py           403)                 except IndexError:
src/ssh_agent_client/__init__.py           404)                     response_stream.extendleft(reversed(buf))

src/ssh_agent_client/__init__.py           405)                     msg = 'truncated response from SSH agent'
src/ssh_agent_client/__init__.py           406)                     raise EOFError(msg) from None

src/ssh_agent_client/__init__.py           407)                 buf.append(val)
src/ssh_agent_client/__init__.py           408)             return bytes(buf)

src/ssh_agent_client/__init__.py           409) 

src/ssh_agent_client/__init__.py           410)         key_count = int.from_bytes(shift(4), 'big')

src/derivepassphrase/ssh_agent/__init__.py 411)         keys: collections.deque[_types.KeyCommentPair]

src/ssh_agent_client/__init__.py           412)         keys = collections.deque()

src/ssh_agent_client/__init__.py           413)         for _ in range(key_count):

src/ssh_agent_client/__init__.py           414)             key_size = int.from_bytes(shift(4), 'big')
src/ssh_agent_client/__init__.py           415)             key = shift(key_size)
src/ssh_agent_client/__init__.py           416)             comment_size = int.from_bytes(shift(4), 'big')
src/ssh_agent_client/__init__.py           417)             comment = shift(comment_size)
src/ssh_agent_client/__init__.py           418)             # Both `key` and `comment` are not wrapped as SSH strings.

src/derivepassphrase/ssh_agent/__init__.py 419)             keys.append(_types.KeyCommentPair(key, comment))

src/ssh_agent_client/__init__.py           420)         if response_stream:

src/ssh_agent_client/__init__.py           421)             raise TrailingDataError

src/ssh_agent_client/__init__.py           422)         return keys
src/ssh_agent_client/__init__.py           423) 
src/ssh_agent_client/__init__.py           424)     def sign(

src/ssh_agent_client/__init__.py           425)         self,
src/ssh_agent_client/__init__.py           426)         /,
src/ssh_agent_client/__init__.py           427)         key: bytes | bytearray,
src/ssh_agent_client/__init__.py           428)         payload: bytes | bytearray,
src/ssh_agent_client/__init__.py           429)         *,
src/ssh_agent_client/__init__.py           430)         flags: int = 0,
src/ssh_agent_client/__init__.py           431)         check_if_key_loaded: bool = False,

src/ssh_agent_client/__init__.py           432)     ) -> bytes | bytearray:
src/ssh_agent_client/__init__.py           433)         """Request the SSH agent sign the payload with the key.
src/ssh_agent_client/__init__.py           434) 
src/ssh_agent_client/__init__.py           435)         Args:
src/ssh_agent_client/__init__.py           436)             key:
src/ssh_agent_client/__init__.py           437)                 The public SSH key to sign the payload with, in the same

src/derivepassphrase/ssh_agent.py          438)                 format as returned by, e.g., the [`list_keys`][] method.

src/ssh_agent_client/__init__.py           439)                 The corresponding private key must have previously been
src/ssh_agent_client/__init__.py           440)                 loaded into the agent to successfully issue a signature.
src/ssh_agent_client/__init__.py           441)             payload:
src/ssh_agent_client/__init__.py           442)                 A byte string of data to sign.
src/ssh_agent_client/__init__.py           443)             flags:
src/ssh_agent_client/__init__.py           444)                 Optional flags for the signing request.  Currently
src/ssh_agent_client/__init__.py           445)                 passed on as-is to the agent.  In real-world usage, this
src/ssh_agent_client/__init__.py           446)                 could be used, e.g., to request more modern hash
src/ssh_agent_client/__init__.py           447)                 algorithms when signing with RSA keys.  (No such
src/ssh_agent_client/__init__.py           448)                 real-world usage is currently implemented.)
src/ssh_agent_client/__init__.py           449)             check_if_key_loaded:

src/derivepassphrase/ssh_agent.py          450)                 If true, check beforehand (via [`list_keys`][]) if the

src/ssh_agent_client/__init__.py           451)                 corresponding key has been loaded into the agent.
src/ssh_agent_client/__init__.py           452) 
src/ssh_agent_client/__init__.py           453)         Returns:
src/ssh_agent_client/__init__.py           454)             The binary signature of the payload under the given key.
src/ssh_agent_client/__init__.py           455) 
src/ssh_agent_client/__init__.py           456)         Raises:
src/ssh_agent_client/__init__.py           457)             EOFError:
src/ssh_agent_client/__init__.py           458)                 The response from the SSH agent is truncated or missing.

src/derivepassphrase/ssh_agent.py          459)             OSError:
src/derivepassphrase/ssh_agent.py          460)                 There was a communication error with the SSH agent.

src/ssh_agent_client/__init__.py           461)             TrailingDataError:

src/ssh_agent_client/__init__.py           462)                 The response from the SSH agent is too long.

src/derivepassphrase/ssh_agent.py          463)             SSHAgentFailedError:

src/ssh_agent_client/__init__.py           464)                 The agent failed to complete the request.

src/ssh_agent_client/__init__.py           465)             KeyError:

src/ssh_agent_client/__init__.py           466)                 `check_if_key_loaded` is true, and the `key` was not
src/ssh_agent_client/__init__.py           467)                 loaded into the agent.
src/ssh_agent_client/__init__.py           468) 
src/ssh_agent_client/__init__.py           469)         """
src/ssh_agent_client/__init__.py           470)         if check_if_key_loaded:
src/ssh_agent_client/__init__.py           471)             loaded_keys = frozenset({pair.key for pair in self.list_keys()})
src/ssh_agent_client/__init__.py           472)             if bytes(key) not in loaded_keys:

src/ssh_agent_client/__init__.py           473)                 msg = 'target SSH key not loaded into agent'
src/ssh_agent_client/__init__.py           474)                 raise KeyError(msg)

src/ssh_agent_client/__init__.py           475)         request_data = bytearray(self.string(key))
src/ssh_agent_client/__init__.py           476)         request_data.extend(self.string(payload))
src/ssh_agent_client/__init__.py           477)         request_data.extend(self.uint32(flags))

src/derivepassphrase/ssh_agent.py          478)         return self.unstring(
src/derivepassphrase/ssh_agent.py          479)             self.request(
src/derivepassphrase/ssh_agent.py          480)                 _types.SSH_AGENTC.SIGN_REQUEST.value,
src/derivepassphrase/ssh_agent.py          481)                 request_data,
src/derivepassphrase/ssh_agent.py          482)                 response_code=_types.SSH_AGENT.SIGN_RESPONSE,
src/derivepassphrase/ssh_agent.py          483)             )