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/derivepassphrase/ssh_agent.py           23)     from typing_extensions import Buffer
src/derivepassphrase/ssh_agent.py           24) 

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

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

src/ssh_agent_client/__init__.py            27) 

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

src/ssh_agent_client/__init__.py            32) _socket = socket
src/ssh_agent_client/__init__.py            33) 

src/ssh_agent_client/__init__.py            34) 

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

src/ssh_agent_client/__init__.py            37) 

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

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

src/ssh_agent_client/__init__.py            40) 

src/ssh_agent_client/__init__.py            41) 

src/derivepassphrase/ssh_agent.py           42) class SSHAgentFailedError(RuntimeError):
src/derivepassphrase/ssh_agent.py           43)     """The SSH agent failed to complete the requested operation."""
src/derivepassphrase/ssh_agent.py           44) 
src/derivepassphrase/ssh_agent.py           45)     def __str__(self) -> str:

src/derivepassphrase/ssh_agent.py           46)         # Use match/case here once Python 3.9 becomes unsupported.
src/derivepassphrase/ssh_agent.py           47)         if self.args == (  # pragma: no branch
src/derivepassphrase/ssh_agent.py           48)             _types.SSH_AGENT.FAILURE.value,
src/derivepassphrase/ssh_agent.py           49)             b'',
src/derivepassphrase/ssh_agent.py           50)         ):
src/derivepassphrase/ssh_agent.py           51)             return 'The SSH agent failed to complete the request'
src/derivepassphrase/ssh_agent.py           52)         elif self.args[1]:  # noqa: RET505  # pragma: no cover
src/derivepassphrase/ssh_agent.py           53)             code = self.args[0]
src/derivepassphrase/ssh_agent.py           54)             msg = self.args[1].decode('utf-8', 'surrogateescape')
src/derivepassphrase/ssh_agent.py           55)             return f'[Code {code:d}] {msg:s}'
src/derivepassphrase/ssh_agent.py           56)         else:  # pragma: no cover
src/derivepassphrase/ssh_agent.py           57)             return repr(self)

src/derivepassphrase/ssh_agent.py           58) 
src/derivepassphrase/ssh_agent.py           59)     def __repr__(self) -> str:  # pragma: no cover
src/derivepassphrase/ssh_agent.py           60)         return f'{self.__class__.__name__}{self.args!r}'
src/derivepassphrase/ssh_agent.py           61) 
src/derivepassphrase/ssh_agent.py           62) 

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

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

src/ssh_agent_client/__init__.py            76) 
src/ssh_agent_client/__init__.py            77)     """

src/ssh_agent_client/__init__.py            78) 

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

src/ssh_agent_client/__init__.py            80) 

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

src/ssh_agent_client/__init__.py            97)         Raises:
src/ssh_agent_client/__init__.py            98)             KeyError:
src/ssh_agent_client/__init__.py            99)                 The `SSH_AUTH_SOCK` environment was not found.
src/ssh_agent_client/__init__.py           100)             OSError:
src/ssh_agent_client/__init__.py           101)                 There was an error setting up a socket connection to the
src/ssh_agent_client/__init__.py           102)                 agent.
src/ssh_agent_client/__init__.py           103) 

src/ssh_agent_client/__init__.py           104)         """
src/ssh_agent_client/__init__.py           105)         if socket is not None:

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

src/ssh_agent_client/__init__.py           107)         else:

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

src/ssh_agent_client/__init__.py           109)         try:

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

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

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

src/ssh_agent_client/__init__.py           116)                 raise

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

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

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

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

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

src/ssh_agent_client/__init__.py           123) 
src/ssh_agent_client/__init__.py           124)     def __enter__(self) -> Self:

src/derivepassphrase/ssh_agent.py          125)         """Close socket connection upon context manager completion.
src/derivepassphrase/ssh_agent.py          126) 
src/derivepassphrase/ssh_agent.py          127)         Returns:
src/derivepassphrase/ssh_agent.py          128)             Self.
src/derivepassphrase/ssh_agent.py          129) 
src/derivepassphrase/ssh_agent.py          130)         """

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

src/ssh_agent_client/__init__.py           132)         return self
src/ssh_agent_client/__init__.py           133) 
src/ssh_agent_client/__init__.py           134)     def __exit__(

src/ssh_agent_client/__init__.py           135)         self,
src/ssh_agent_client/__init__.py           136)         exc_type: type[BaseException] | None,
src/ssh_agent_client/__init__.py           137)         exc_val: BaseException | None,

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

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

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

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

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

src/ssh_agent_client/__init__.py           154)         )
src/ssh_agent_client/__init__.py           155) 
src/ssh_agent_client/__init__.py           156)     @staticmethod

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

src/ssh_agent_client/__init__.py           176)         return int.to_bytes(num, 4, 'big', signed=False)
src/ssh_agent_client/__init__.py           177) 
src/ssh_agent_client/__init__.py           178)     @classmethod

src/derivepassphrase/ssh_agent.py          179)     def string(cls, payload: Buffer, /) -> bytes:

src/ssh_agent_client/__init__.py           180)         r"""Format the payload as an SSH string, as per the agent protocol.
src/ssh_agent_client/__init__.py           181) 
src/ssh_agent_client/__init__.py           182)         Args:

src/derivepassphrase/ssh_agent.py          183)             payload: A bytes-like object.

src/ssh_agent_client/__init__.py           184) 
src/ssh_agent_client/__init__.py           185)         Returns:

src/derivepassphrase/ssh_agent.py          186)             The payload, framed in the SSH agent wire protocol format,
src/derivepassphrase/ssh_agent.py          187)             as a bytes object.

src/ssh_agent_client/__init__.py           188) 
src/ssh_agent_client/__init__.py           189)         Examples:

src/derivepassphrase/ssh_agent.py          190)             >>> SSHAgentClient.string(b'ssh-rsa')

src/ssh_agent_client/__init__.py           191)             b'\x00\x00\x00\x07ssh-rsa'
src/ssh_agent_client/__init__.py           192) 
src/ssh_agent_client/__init__.py           193)         """

src/ssh_agent_client/__init__.py           194)         try:

src/derivepassphrase/ssh_agent.py          195)             payload = memoryview(payload)
src/derivepassphrase/ssh_agent.py          196)         except TypeError as e:

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

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

src/derivepassphrase/ssh_agent.py          199)         ret = bytearray()
src/derivepassphrase/ssh_agent.py          200)         ret.extend(cls.uint32(len(payload)))
src/derivepassphrase/ssh_agent.py          201)         ret.extend(payload)
src/derivepassphrase/ssh_agent.py          202)         return bytes(ret)

src/ssh_agent_client/__init__.py           203) 
src/ssh_agent_client/__init__.py           204)     @classmethod

src/derivepassphrase/ssh_agent.py          205)     def unstring(cls, bytestring: Buffer, /) -> bytes:

src/ssh_agent_client/__init__.py           206)         r"""Unpack an SSH string.
src/ssh_agent_client/__init__.py           207) 
src/ssh_agent_client/__init__.py           208)         Args:

src/derivepassphrase/ssh_agent.py          209)             bytestring: A framed bytes-like object.

src/ssh_agent_client/__init__.py           210) 
src/ssh_agent_client/__init__.py           211)         Returns:

src/derivepassphrase/ssh_agent.py          212)             The payload, as a bytes object.

src/ssh_agent_client/__init__.py           213) 
src/ssh_agent_client/__init__.py           214)         Raises:
src/ssh_agent_client/__init__.py           215)             ValueError:

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

src/ssh_agent_client/__init__.py           217) 
src/ssh_agent_client/__init__.py           218)         Examples:

src/derivepassphrase/ssh_agent.py          219)             >>> SSHAgentClient.unstring(b'\x00\x00\x00\x07ssh-rsa')

src/ssh_agent_client/__init__.py           220)             b'ssh-rsa'

src/derivepassphrase/ssh_agent.py          221)             >>> SSHAgentClient.unstring(SSHAgentClient.string(b'ssh-ed25519'))

src/ssh_agent_client/__init__.py           222)             b'ssh-ed25519'
src/ssh_agent_client/__init__.py           223) 

src/derivepassphrase/ssh_agent.py          224)         """
src/derivepassphrase/ssh_agent.py          225)         bytestring = memoryview(bytestring)

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

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

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

src/ssh_agent_client/__init__.py           230)         ):
src/ssh_agent_client/__init__.py           231)             raise ValueError(msg)

src/derivepassphrase/ssh_agent.py          232)         return bytes(bytestring[HEAD_LEN:])

src/ssh_agent_client/__init__.py           233) 

src/ssh_agent_client/__init__.py           234)     @classmethod

src/derivepassphrase/ssh_agent.py          235)     def unstring_prefix(cls, bytestring: Buffer, /) -> tuple[bytes, bytes]:

src/ssh_agent_client/__init__.py           236)         r"""Unpack an SSH string at the beginning of the byte string.
src/ssh_agent_client/__init__.py           237) 
src/ssh_agent_client/__init__.py           238)         Args:
src/ssh_agent_client/__init__.py           239)             bytestring:

src/derivepassphrase/ssh_agent.py          240)                 A bytes-like object, beginning with a framed/SSH byte
src/derivepassphrase/ssh_agent.py          241)                 string.

src/ssh_agent_client/__init__.py           242) 
src/ssh_agent_client/__init__.py           243)         Returns:
src/ssh_agent_client/__init__.py           244)             A 2-tuple `(a, b)`, where `a` is the unframed byte
src/ssh_agent_client/__init__.py           245)             string/payload at the beginning of input byte string, and
src/ssh_agent_client/__init__.py           246)             `b` is the remainder of the input byte string.
src/ssh_agent_client/__init__.py           247) 
src/ssh_agent_client/__init__.py           248)         Raises:
src/ssh_agent_client/__init__.py           249)             ValueError:
src/ssh_agent_client/__init__.py           250)                 The byte string does not begin with an SSH string.
src/ssh_agent_client/__init__.py           251) 
src/ssh_agent_client/__init__.py           252)         Examples:

src/derivepassphrase/ssh_agent.py          253)             >>> SSHAgentClient.unstring_prefix(

src/ssh_agent_client/__init__.py           254)             ...     b'\x00\x00\x00\x07ssh-rsa____trailing data'
src/ssh_agent_client/__init__.py           255)             ... )

src/ssh_agent_client/__init__.py           256)             (b'ssh-rsa', b'____trailing data')

src/derivepassphrase/ssh_agent.py          257)             >>> SSHAgentClient.unstring_prefix(

src/ssh_agent_client/__init__.py           258)             ...     SSHAgentClient.string(b'ssh-ed25519')
src/ssh_agent_client/__init__.py           259)             ... )

src/ssh_agent_client/__init__.py           260)             (b'ssh-ed25519', b'')
src/ssh_agent_client/__init__.py           261) 
src/ssh_agent_client/__init__.py           262)         """

src/derivepassphrase/ssh_agent.py          263)         bytestring = memoryview(bytestring).toreadonly()

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

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

src/ssh_agent_client/__init__.py           271)         return (

src/derivepassphrase/ssh_agent.py          272)             bytes(bytestring[HEAD_LEN : m + HEAD_LEN]),
src/derivepassphrase/ssh_agent.py          273)             bytes(bytestring[m + HEAD_LEN :]),

src/ssh_agent_client/__init__.py           274)         )

src/ssh_agent_client/__init__.py           275) 

src/derivepassphrase/ssh_agent.py          276)     @overload
src/derivepassphrase/ssh_agent.py          277)     def request(  # pragma: no cover
src/derivepassphrase/ssh_agent.py          278)         self,
src/derivepassphrase/ssh_agent.py          279)         code: int | _types.SSH_AGENTC,

src/derivepassphrase/ssh_agent.py          280)         payload: Buffer,

src/derivepassphrase/ssh_agent.py          281)         /,
src/derivepassphrase/ssh_agent.py          282)         *,
src/derivepassphrase/ssh_agent.py          283)         response_code: None = None,

src/derivepassphrase/ssh_agent.py          284)     ) -> tuple[int, bytes]: ...

src/derivepassphrase/ssh_agent.py          285) 
src/derivepassphrase/ssh_agent.py          286)     @overload
src/derivepassphrase/ssh_agent.py          287)     def request(  # pragma: no cover
src/derivepassphrase/ssh_agent.py          288)         self,
src/derivepassphrase/ssh_agent.py          289)         code: int | _types.SSH_AGENTC,

src/derivepassphrase/ssh_agent.py          290)         payload: Buffer,

src/derivepassphrase/ssh_agent.py          291)         /,
src/derivepassphrase/ssh_agent.py          292)         *,
src/derivepassphrase/ssh_agent.py          293)         response_code: Iterable[_types.SSH_AGENT | int] = frozenset({
src/derivepassphrase/ssh_agent.py          294)             _types.SSH_AGENT.SUCCESS
src/derivepassphrase/ssh_agent.py          295)         }),

src/derivepassphrase/ssh_agent.py          296)     ) -> tuple[int, bytes]: ...

src/derivepassphrase/ssh_agent.py          297) 
src/derivepassphrase/ssh_agent.py          298)     @overload
src/derivepassphrase/ssh_agent.py          299)     def request(  # pragma: no cover
src/derivepassphrase/ssh_agent.py          300)         self,
src/derivepassphrase/ssh_agent.py          301)         code: int | _types.SSH_AGENTC,

src/derivepassphrase/ssh_agent.py          302)         payload: Buffer,

src/derivepassphrase/ssh_agent.py          303)         /,
src/derivepassphrase/ssh_agent.py          304)         *,
src/derivepassphrase/ssh_agent.py          305)         response_code: _types.SSH_AGENT | int = _types.SSH_AGENT.SUCCESS,

src/derivepassphrase/ssh_agent.py          306)     ) -> bytes: ...

src/derivepassphrase/ssh_agent.py          307) 

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

src/derivepassphrase/ssh_agent.py          309)         self,
src/derivepassphrase/ssh_agent.py          310)         code: int | _types.SSH_AGENTC,

src/derivepassphrase/ssh_agent.py          311)         payload: Buffer,

src/derivepassphrase/ssh_agent.py          312)         /,
src/derivepassphrase/ssh_agent.py          313)         *,
src/derivepassphrase/ssh_agent.py          314)         response_code: (
src/derivepassphrase/ssh_agent.py          315)             Iterable[_types.SSH_AGENT | int] | _types.SSH_AGENT | int | None
src/derivepassphrase/ssh_agent.py          316)         ) = None,

src/derivepassphrase/ssh_agent.py          317)     ) -> tuple[int, bytes] | bytes:

src/ssh_agent_client/__init__.py           318)         """Issue a generic request to the SSH agent.
src/ssh_agent_client/__init__.py           319) 
src/ssh_agent_client/__init__.py           320)         Args:
src/ssh_agent_client/__init__.py           321)             code:
src/ssh_agent_client/__init__.py           322)                 The request code.  See the SSH agent protocol for
src/ssh_agent_client/__init__.py           323)                 protocol numbers to use here (and which protocol numbers
src/ssh_agent_client/__init__.py           324)                 to expect in a response).
src/ssh_agent_client/__init__.py           325)             payload:

src/derivepassphrase/ssh_agent.py          326)                 A bytes-like object containing the payload, or
src/derivepassphrase/ssh_agent.py          327)                 "contents", of the request.  Request-specific.
src/derivepassphrase/ssh_agent.py          328) 
src/derivepassphrase/ssh_agent.py          329)                 It is our responsibility to add any necessary wire
src/derivepassphrase/ssh_agent.py          330)                 framing around the request code and the payload,
src/derivepassphrase/ssh_agent.py          331)                 not the caller's.

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

src/ssh_agent_client/__init__.py           336) 
src/ssh_agent_client/__init__.py           337)         Returns:
src/ssh_agent_client/__init__.py           338)             A 2-tuple consisting of the response code and the payload,
src/ssh_agent_client/__init__.py           339)             with all wire framing removed.
src/ssh_agent_client/__init__.py           340) 

src/derivepassphrase/ssh_agent.py          341)             If a response code was passed, then only return the payload.
src/derivepassphrase/ssh_agent.py          342) 

src/ssh_agent_client/__init__.py           343)         Raises:
src/ssh_agent_client/__init__.py           344)             EOFError:
src/ssh_agent_client/__init__.py           345)                 The response from the SSH agent is truncated or missing.

src/derivepassphrase/ssh_agent.py          346)             OSError:
src/derivepassphrase/ssh_agent.py          347)                 There was a communication error with the SSH agent.
src/derivepassphrase/ssh_agent.py          348)             SSHAgentFailedError:
src/derivepassphrase/ssh_agent.py          349)                 We expected specific response codes, but did not receive
src/derivepassphrase/ssh_agent.py          350)                 any of them.

src/ssh_agent_client/__init__.py           351) 
src/ssh_agent_client/__init__.py           352)         """

src/derivepassphrase/ssh_agent.py          353)         if isinstance(  # pragma: no branch

src/derivepassphrase/ssh_agent.py          354)             response_code, (int, _types.SSH_AGENT)

src/derivepassphrase/ssh_agent.py          355)         ):
src/derivepassphrase/ssh_agent.py          356)             response_code = frozenset({response_code})
src/derivepassphrase/ssh_agent.py          357)         if response_code is not None:  # pragma: no branch
src/derivepassphrase/ssh_agent.py          358)             response_code = frozenset({
src/derivepassphrase/ssh_agent.py          359)                 c if isinstance(c, int) else c.value for c in response_code
src/derivepassphrase/ssh_agent.py          360)             })

src/derivepassphrase/ssh_agent.py          361)         payload = memoryview(payload)

src/derivepassphrase/ssh_agent.py          362)         request_message = bytearray([
src/derivepassphrase/ssh_agent.py          363)             code if isinstance(code, int) else code.value
src/derivepassphrase/ssh_agent.py          364)         ])

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

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

src/ssh_agent_client/__init__.py           367)         chunk = self._connection.recv(HEAD_LEN)
src/ssh_agent_client/__init__.py           368)         if len(chunk) < HEAD_LEN:
src/ssh_agent_client/__init__.py           369)             msg = 'cannot read response length'
src/ssh_agent_client/__init__.py           370)             raise EOFError(msg)

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

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

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

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

src/derivepassphrase/ssh_agent.py          376)         if not response_code:  # pragma: no cover
src/derivepassphrase/ssh_agent.py          377)             return response[0], response[1:]
src/derivepassphrase/ssh_agent.py          378)         if response[0] not in response_code:
src/derivepassphrase/ssh_agent.py          379)             raise SSHAgentFailedError(response[0], response[1:])
src/derivepassphrase/ssh_agent.py          380)         return response[1:]

src/ssh_agent_client/__init__.py           381) 

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

src/ssh_agent_client/__init__.py           383)         """Request a list of keys known to the SSH agent.
src/ssh_agent_client/__init__.py           384) 
src/ssh_agent_client/__init__.py           385)         Returns:
src/ssh_agent_client/__init__.py           386)             A read-only sequence of key/comment pairs.
src/ssh_agent_client/__init__.py           387) 
src/ssh_agent_client/__init__.py           388)         Raises:
src/ssh_agent_client/__init__.py           389)             EOFError:
src/ssh_agent_client/__init__.py           390)                 The response from the SSH agent is truncated or missing.

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

src/ssh_agent_client/__init__.py           393)             TrailingDataError:

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

src/derivepassphrase/ssh_agent.py          395)             SSHAgentFailedError:

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

src/ssh_agent_client/__init__.py           397) 
src/ssh_agent_client/__init__.py           398)         """

src/derivepassphrase/ssh_agent.py          399)         response = self.request(
src/derivepassphrase/ssh_agent.py          400)             _types.SSH_AGENTC.REQUEST_IDENTITIES.value,
src/derivepassphrase/ssh_agent.py          401)             b'',
src/derivepassphrase/ssh_agent.py          402)             response_code=_types.SSH_AGENT.IDENTITIES_ANSWER,

src/ssh_agent_client/__init__.py           403)         )

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

src/ssh_agent_client/__init__.py           405) 

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

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

src/ssh_agent_client/__init__.py           409)                 try:
src/ssh_agent_client/__init__.py           410)                     val = response_stream.popleft()
src/ssh_agent_client/__init__.py           411)                 except IndexError:
src/ssh_agent_client/__init__.py           412)                     response_stream.extendleft(reversed(buf))

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

src/ssh_agent_client/__init__.py           415)                 buf.append(val)
src/ssh_agent_client/__init__.py           416)             return bytes(buf)

src/ssh_agent_client/__init__.py           417) 

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

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

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

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

src/ssh_agent_client/__init__.py           422)             key_size = int.from_bytes(shift(4), 'big')
src/ssh_agent_client/__init__.py           423)             key = shift(key_size)
src/ssh_agent_client/__init__.py           424)             comment_size = int.from_bytes(shift(4), 'big')
src/ssh_agent_client/__init__.py           425)             comment = shift(comment_size)
src/ssh_agent_client/__init__.py           426)             # Both `key` and `comment` are not wrapped as SSH strings.

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

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

src/ssh_agent_client/__init__.py           429)             raise TrailingDataError

src/ssh_agent_client/__init__.py           430)         return keys
src/ssh_agent_client/__init__.py           431) 
src/ssh_agent_client/__init__.py           432)     def sign(

src/ssh_agent_client/__init__.py           433)         self,
src/ssh_agent_client/__init__.py           434)         /,

src/derivepassphrase/ssh_agent.py          435)         key: Buffer,
src/derivepassphrase/ssh_agent.py          436)         payload: Buffer,

src/ssh_agent_client/__init__.py           437)         *,
src/ssh_agent_client/__init__.py           438)         flags: int = 0,
src/ssh_agent_client/__init__.py           439)         check_if_key_loaded: bool = False,

src/derivepassphrase/ssh_agent.py          440)     ) -> bytes:

src/ssh_agent_client/__init__.py           441)         """Request the SSH agent sign the payload with the key.
src/ssh_agent_client/__init__.py           442) 
src/ssh_agent_client/__init__.py           443)         Args:
src/ssh_agent_client/__init__.py           444)             key:
src/ssh_agent_client/__init__.py           445)                 The public SSH key to sign the payload with, in the same

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

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

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

src/ssh_agent_client/__init__.py           459)                 corresponding key has been loaded into the agent.
src/ssh_agent_client/__init__.py           460) 
src/ssh_agent_client/__init__.py           461)         Returns:
src/ssh_agent_client/__init__.py           462)             The binary signature of the payload under the given key.
src/ssh_agent_client/__init__.py           463) 
src/ssh_agent_client/__init__.py           464)         Raises:
src/ssh_agent_client/__init__.py           465)             EOFError:
src/ssh_agent_client/__init__.py           466)                 The response from the SSH agent is truncated or missing.

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

src/ssh_agent_client/__init__.py           469)             TrailingDataError:

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

src/derivepassphrase/ssh_agent.py          471)             SSHAgentFailedError:

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

src/ssh_agent_client/__init__.py           473)             KeyError:

src/ssh_agent_client/__init__.py           474)                 `check_if_key_loaded` is true, and the `key` was not
src/ssh_agent_client/__init__.py           475)                 loaded into the agent.
src/ssh_agent_client/__init__.py           476) 
src/ssh_agent_client/__init__.py           477)         """

src/derivepassphrase/ssh_agent.py          478)         key = memoryview(key)
src/derivepassphrase/ssh_agent.py          479)         payload = memoryview(payload)

src/ssh_agent_client/__init__.py           480)         if check_if_key_loaded:
src/ssh_agent_client/__init__.py           481)             loaded_keys = frozenset({pair.key for pair in self.list_keys()})
src/ssh_agent_client/__init__.py           482)             if bytes(key) not in loaded_keys:

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

src/ssh_agent_client/__init__.py           485)         request_data = bytearray(self.string(key))
src/ssh_agent_client/__init__.py           486)         request_data.extend(self.string(payload))
src/ssh_agent_client/__init__.py           487)         request_data.extend(self.uint32(flags))

src/derivepassphrase/ssh_agent.py          488)         return bytes(
src/derivepassphrase/ssh_agent.py          489)             self.unstring(
src/derivepassphrase/ssh_agent.py          490)                 self.request(
src/derivepassphrase/ssh_agent.py          491)                     _types.SSH_AGENTC.SIGN_REQUEST.value,
src/derivepassphrase/ssh_agent.py          492)                     request_data,
src/derivepassphrase/ssh_agent.py          493)                     response_code=_types.SSH_AGENT.SIGN_RESPONSE,
src/derivepassphrase/ssh_agent.py          494)                 )

src/derivepassphrase/ssh_agent.py          495)             )