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/derivepassphrase/ssh_agent.py           10) import contextlib

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/derivepassphrase/ssh_agent.py           15) from typing_extensions import Self, assert_never

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, Iterator, 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/derivepassphrase/ssh_agent.py           88)                 An optional socket, already connected to the SSH agent.
src/derivepassphrase/ssh_agent.py           89)                 If not given, we query the `SSH_AUTH_SOCK` environment

src/ssh_agent_client/__init__.py            90)                 variable to auto-discover the correct socket address.

src/derivepassphrase/ssh_agent.py           91) 
src/derivepassphrase/ssh_agent.py           92)                 [We currently only support connecting via UNIX domain
src/derivepassphrase/ssh_agent.py           93)                 sockets][issue13], and only on platforms with support
src/derivepassphrase/ssh_agent.py           94)                 for [`socket.AF_UNIX`][AF_UNIX].
src/derivepassphrase/ssh_agent.py           95) 
src/derivepassphrase/ssh_agent.py           96)                 [issue13]: https://github.com/the-13th-letter/derivepassphrase/issues/13
src/derivepassphrase/ssh_agent.py           97)                 [AF_UNIX]: https://docs.python.org/3/library/socket.html#socket.AF_UNIX

src/ssh_agent_client/__init__.py            98)             timeout:
src/ssh_agent_client/__init__.py            99)                 A connection timeout for the SSH agent.  Only used if
src/ssh_agent_client/__init__.py           100)                 the socket is not yet connected.  The default value
src/ssh_agent_client/__init__.py           101)                 gives ample time for agent connections forwarded via
src/ssh_agent_client/__init__.py           102)                 SSH on high-latency networks (e.g. Tor).
src/ssh_agent_client/__init__.py           103) 

src/ssh_agent_client/__init__.py           104)         Raises:
src/ssh_agent_client/__init__.py           105)             KeyError:

src/derivepassphrase/ssh_agent.py          106)                 The `SSH_AUTH_SOCK` environment variable was not found.
src/derivepassphrase/ssh_agent.py          107)             NotImplementedError:
src/derivepassphrase/ssh_agent.py          108)                 This Python version does not support UNIX domain
src/derivepassphrase/ssh_agent.py          109)                 sockets, necessary to automatically connect to a running
src/derivepassphrase/ssh_agent.py          110)                 SSH agent via the `SSH_AUTH_SOCK` environment variable.

src/ssh_agent_client/__init__.py           111)             OSError:
src/ssh_agent_client/__init__.py           112)                 There was an error setting up a socket connection to the
src/ssh_agent_client/__init__.py           113)                 agent.
src/ssh_agent_client/__init__.py           114) 

src/ssh_agent_client/__init__.py           115)         """
src/ssh_agent_client/__init__.py           116)         if socket is not None:

src/ssh_agent_client/__init__.py           117)             self._connection = socket
src/ssh_agent_client/__init__.py           118)             # Test whether the socket is connected.
src/ssh_agent_client/__init__.py           119)             self._connection.getpeername()

src/derivepassphrase/ssh_agent.py          120)         else:
src/derivepassphrase/ssh_agent.py          121)             if not hasattr(_socket, 'AF_UNIX'):
src/derivepassphrase/ssh_agent.py          122)                 msg = (
src/derivepassphrase/ssh_agent.py          123)                     'This Python version does not support UNIX domain sockets'
src/derivepassphrase/ssh_agent.py          124)                 )
src/derivepassphrase/ssh_agent.py          125)                 raise NotImplementedError(msg)
src/derivepassphrase/ssh_agent.py          126)             self._connection = _socket.socket(family=_socket.AF_UNIX)

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

src/ssh_agent_client/__init__.py           128)                 msg = 'SSH_AUTH_SOCK environment variable'

src/derivepassphrase/ssh_agent.py          129)                 raise KeyError(msg)

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

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

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

src/ssh_agent_client/__init__.py           133) 
src/ssh_agent_client/__init__.py           134)     def __enter__(self) -> Self:

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)         Returns:
src/derivepassphrase/ssh_agent.py          138)             Self.
src/derivepassphrase/ssh_agent.py          139) 
src/derivepassphrase/ssh_agent.py          140)         """

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

src/ssh_agent_client/__init__.py           142)         return self
src/ssh_agent_client/__init__.py           143) 
src/ssh_agent_client/__init__.py           144)     def __exit__(

src/ssh_agent_client/__init__.py           145)         self,
src/ssh_agent_client/__init__.py           146)         exc_type: type[BaseException] | None,
src/ssh_agent_client/__init__.py           147)         exc_val: BaseException | None,

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

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

src/derivepassphrase/ssh_agent.py          150)         """Close socket connection upon context manager completion.
src/derivepassphrase/ssh_agent.py          151) 
src/derivepassphrase/ssh_agent.py          152)         Args:
src/derivepassphrase/ssh_agent.py          153)             exc_type: An optional exception type.
src/derivepassphrase/ssh_agent.py          154)             exc_val: An optional exception value.
src/derivepassphrase/ssh_agent.py          155)             exc_tb: An optional exception traceback.
src/derivepassphrase/ssh_agent.py          156) 
src/derivepassphrase/ssh_agent.py          157)         Returns:
src/derivepassphrase/ssh_agent.py          158)             True if the exception was handled, false if it should
src/derivepassphrase/ssh_agent.py          159)             propagate.
src/derivepassphrase/ssh_agent.py          160) 
src/derivepassphrase/ssh_agent.py          161)         """

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

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

src/ssh_agent_client/__init__.py           164)         )
src/ssh_agent_client/__init__.py           165) 
src/ssh_agent_client/__init__.py           166)     @staticmethod

src/ssh_agent_client/__init__.py           167)     def uint32(num: int, /) -> bytes:
src/ssh_agent_client/__init__.py           168)         r"""Format the number as a `uint32`, as per the agent protocol.
src/ssh_agent_client/__init__.py           169) 
src/ssh_agent_client/__init__.py           170)         Args:
src/ssh_agent_client/__init__.py           171)             num: A number.
src/ssh_agent_client/__init__.py           172) 
src/ssh_agent_client/__init__.py           173)         Returns:
src/ssh_agent_client/__init__.py           174)             The number in SSH agent wire protocol format, i.e. as
src/ssh_agent_client/__init__.py           175)             a 32-bit big endian number.
src/ssh_agent_client/__init__.py           176) 
src/ssh_agent_client/__init__.py           177)         Raises:
src/ssh_agent_client/__init__.py           178)             OverflowError:
src/ssh_agent_client/__init__.py           179)                 As per [`int.to_bytes`][].
src/ssh_agent_client/__init__.py           180) 
src/ssh_agent_client/__init__.py           181)         Examples:
src/ssh_agent_client/__init__.py           182)             >>> SSHAgentClient.uint32(16777216)
src/ssh_agent_client/__init__.py           183)             b'\x01\x00\x00\x00'
src/ssh_agent_client/__init__.py           184) 
src/ssh_agent_client/__init__.py           185)         """

src/ssh_agent_client/__init__.py           186)         return int.to_bytes(num, 4, 'big', signed=False)
src/ssh_agent_client/__init__.py           187) 
src/ssh_agent_client/__init__.py           188)     @classmethod

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

src/ssh_agent_client/__init__.py           190)         r"""Format the payload as an SSH string, as per the agent protocol.
src/ssh_agent_client/__init__.py           191) 
src/ssh_agent_client/__init__.py           192)         Args:

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

src/ssh_agent_client/__init__.py           194) 
src/ssh_agent_client/__init__.py           195)         Returns:

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

src/ssh_agent_client/__init__.py           198) 
src/ssh_agent_client/__init__.py           199)         Examples:

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

src/ssh_agent_client/__init__.py           201)             b'\x00\x00\x00\x07ssh-rsa'
src/ssh_agent_client/__init__.py           202) 
src/ssh_agent_client/__init__.py           203)         """

src/ssh_agent_client/__init__.py           204)         try:

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

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

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

src/derivepassphrase/ssh_agent.py          209)         ret = bytearray()
src/derivepassphrase/ssh_agent.py          210)         ret.extend(cls.uint32(len(payload)))
src/derivepassphrase/ssh_agent.py          211)         ret.extend(payload)
src/derivepassphrase/ssh_agent.py          212)         return bytes(ret)

src/ssh_agent_client/__init__.py           213) 
src/ssh_agent_client/__init__.py           214)     @classmethod

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

src/ssh_agent_client/__init__.py           216)         r"""Unpack an SSH string.
src/ssh_agent_client/__init__.py           217) 
src/ssh_agent_client/__init__.py           218)         Args:

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

src/ssh_agent_client/__init__.py           220) 
src/ssh_agent_client/__init__.py           221)         Returns:

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

src/ssh_agent_client/__init__.py           223) 
src/ssh_agent_client/__init__.py           224)         Raises:
src/ssh_agent_client/__init__.py           225)             ValueError:

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

src/ssh_agent_client/__init__.py           227) 
src/ssh_agent_client/__init__.py           228)         Examples:

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

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

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

src/ssh_agent_client/__init__.py           232)             b'ssh-ed25519'
src/ssh_agent_client/__init__.py           233) 

src/derivepassphrase/ssh_agent.py          234)         """
src/derivepassphrase/ssh_agent.py          235)         bytestring = memoryview(bytestring)

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

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

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

src/ssh_agent_client/__init__.py           240)         ):
src/ssh_agent_client/__init__.py           241)             raise ValueError(msg)

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

src/ssh_agent_client/__init__.py           243) 

src/ssh_agent_client/__init__.py           244)     @classmethod

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

src/ssh_agent_client/__init__.py           246)         r"""Unpack an SSH string at the beginning of the byte string.
src/ssh_agent_client/__init__.py           247) 
src/ssh_agent_client/__init__.py           248)         Args:
src/ssh_agent_client/__init__.py           249)             bytestring:

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

src/ssh_agent_client/__init__.py           252) 
src/ssh_agent_client/__init__.py           253)         Returns:
src/ssh_agent_client/__init__.py           254)             A 2-tuple `(a, b)`, where `a` is the unframed byte
src/ssh_agent_client/__init__.py           255)             string/payload at the beginning of input byte string, and
src/ssh_agent_client/__init__.py           256)             `b` is the remainder of the input byte string.
src/ssh_agent_client/__init__.py           257) 
src/ssh_agent_client/__init__.py           258)         Raises:
src/ssh_agent_client/__init__.py           259)             ValueError:
src/ssh_agent_client/__init__.py           260)                 The byte string does not begin with an SSH string.
src/ssh_agent_client/__init__.py           261) 
src/ssh_agent_client/__init__.py           262)         Examples:

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

src/ssh_agent_client/__init__.py           264)             ...     b'\x00\x00\x00\x07ssh-rsa____trailing data'
src/ssh_agent_client/__init__.py           265)             ... )

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

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

src/ssh_agent_client/__init__.py           268)             ...     SSHAgentClient.string(b'ssh-ed25519')
src/ssh_agent_client/__init__.py           269)             ... )

src/ssh_agent_client/__init__.py           270)             (b'ssh-ed25519', b'')
src/ssh_agent_client/__init__.py           271) 
src/ssh_agent_client/__init__.py           272)         """

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

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

src/ssh_agent_client/__init__.py           275)         msg = 'malformed SSH byte string'
src/ssh_agent_client/__init__.py           276)         if n < HEAD_LEN:
src/ssh_agent_client/__init__.py           277)             raise ValueError(msg)
src/ssh_agent_client/__init__.py           278)         m = int.from_bytes(bytestring[:HEAD_LEN], 'big', signed=False)
src/ssh_agent_client/__init__.py           279)         if m + HEAD_LEN > n:
src/ssh_agent_client/__init__.py           280)             raise ValueError(msg)

src/ssh_agent_client/__init__.py           281)         return (

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

src/ssh_agent_client/__init__.py           284)         )

src/ssh_agent_client/__init__.py           285) 

src/derivepassphrase/ssh_agent.py          286)     @classmethod
src/derivepassphrase/ssh_agent.py          287)     @contextlib.contextmanager
src/derivepassphrase/ssh_agent.py          288)     def ensure_agent_subcontext(
src/derivepassphrase/ssh_agent.py          289)         cls,
src/derivepassphrase/ssh_agent.py          290)         conn: SSHAgentClient | socket.socket | None = None,
src/derivepassphrase/ssh_agent.py          291)     ) -> Iterator[SSHAgentClient]:
src/derivepassphrase/ssh_agent.py          292)         """Return an SSH agent client subcontext.
src/derivepassphrase/ssh_agent.py          293) 
src/derivepassphrase/ssh_agent.py          294)         If necessary, construct an SSH agent client first using the
src/derivepassphrase/ssh_agent.py          295)         connection hint.
src/derivepassphrase/ssh_agent.py          296) 
src/derivepassphrase/ssh_agent.py          297)         Args:
src/derivepassphrase/ssh_agent.py          298)             conn:
src/derivepassphrase/ssh_agent.py          299)                 If an existing SSH agent client, then enter a context
src/derivepassphrase/ssh_agent.py          300)                 within this client's scope.  After exiting the context,
src/derivepassphrase/ssh_agent.py          301)                 the client persists, including its socket.
src/derivepassphrase/ssh_agent.py          302) 
src/derivepassphrase/ssh_agent.py          303)                 If a socket, then construct a client using this socket,
src/derivepassphrase/ssh_agent.py          304)                 then enter a context within this client's scope.  After
src/derivepassphrase/ssh_agent.py          305)                 exiting the context, the client is destroyed and the
src/derivepassphrase/ssh_agent.py          306)                 socket is closed.
src/derivepassphrase/ssh_agent.py          307) 
src/derivepassphrase/ssh_agent.py          308)                 If `None`, construct a client using agent
src/derivepassphrase/ssh_agent.py          309)                 auto-discovery, then enter a context within this
src/derivepassphrase/ssh_agent.py          310)                 client's scope.  After exiting the context, both the
src/derivepassphrase/ssh_agent.py          311)                 client and its socket are destroyed.
src/derivepassphrase/ssh_agent.py          312) 
src/derivepassphrase/ssh_agent.py          313)         Yields:
src/derivepassphrase/ssh_agent.py          314)             When entering this context, return the SSH agent client.
src/derivepassphrase/ssh_agent.py          315) 
src/derivepassphrase/ssh_agent.py          316)         Raises:
src/derivepassphrase/ssh_agent.py          317)             KeyError:
src/derivepassphrase/ssh_agent.py          318)                 `conn` was `None`, and the `SSH_AUTH_SOCK` environment
src/derivepassphrase/ssh_agent.py          319)                 variable was not found.
src/derivepassphrase/ssh_agent.py          320)             NotImplementedError:
src/derivepassphrase/ssh_agent.py          321)                 `conn` was `None`, and this Python does not support
src/derivepassphrase/ssh_agent.py          322)                 [`socket.AF_UNIX`][], so the SSH agent client cannot be
src/derivepassphrase/ssh_agent.py          323)                 automatically set up.
src/derivepassphrase/ssh_agent.py          324)             OSError:
src/derivepassphrase/ssh_agent.py          325)                 `conn` was a socket or `None`, and there was an error
src/derivepassphrase/ssh_agent.py          326)                 setting up a socket connection to the agent.
src/derivepassphrase/ssh_agent.py          327) 
src/derivepassphrase/ssh_agent.py          328)         """
src/derivepassphrase/ssh_agent.py          329)         # Use match/case here once Python 3.9 becomes unsupported.
src/derivepassphrase/ssh_agent.py          330)         if isinstance(conn, SSHAgentClient):
src/derivepassphrase/ssh_agent.py          331)             with contextlib.nullcontext():
src/derivepassphrase/ssh_agent.py          332)                 yield conn
src/derivepassphrase/ssh_agent.py          333)         elif isinstance(conn, socket.socket) or conn is None:
src/derivepassphrase/ssh_agent.py          334)             with SSHAgentClient(socket=conn) as client:
src/derivepassphrase/ssh_agent.py          335)                 yield client
src/derivepassphrase/ssh_agent.py          336)         else:  # pragma: no cover
src/derivepassphrase/ssh_agent.py          337)             assert_never(conn)
src/derivepassphrase/ssh_agent.py          338)             msg = f'invalid connection hint: {conn!r}'
src/derivepassphrase/ssh_agent.py          339)             raise TypeError(msg)  # noqa: DOC501
src/derivepassphrase/ssh_agent.py          340) 

src/derivepassphrase/ssh_agent.py          341)     def has_deterministic_signatures(self) -> bool:
src/derivepassphrase/ssh_agent.py          342)         """Check whether the agent returns deterministic signatures.
src/derivepassphrase/ssh_agent.py          343) 
src/derivepassphrase/ssh_agent.py          344)         Returns:
src/derivepassphrase/ssh_agent.py          345)             True if a known agent was detected where signatures are
src/derivepassphrase/ssh_agent.py          346)             deterministic for all SSH key types, false otherwise.
src/derivepassphrase/ssh_agent.py          347) 
src/derivepassphrase/ssh_agent.py          348)         Note: Known agents with deterministic signatures
src/derivepassphrase/ssh_agent.py          349)             | agent           | detected via                                                  |
src/derivepassphrase/ssh_agent.py          350)             |:----------------|:--------------------------------------------------------------|
src/derivepassphrase/ssh_agent.py          351)             | Pageant (PuTTY) | `list-extended@putty.projects.tartarus.org` extension request |
src/derivepassphrase/ssh_agent.py          352) 
src/derivepassphrase/ssh_agent.py          353)         """  # noqa: E501
src/derivepassphrase/ssh_agent.py          354)         returncode, _payload = self.request(
src/derivepassphrase/ssh_agent.py          355)             _types.SSH_AGENTC.EXTENSION,
src/derivepassphrase/ssh_agent.py          356)             self.string(b'list-extended@putty.projects.tartarus.org'),
src/derivepassphrase/ssh_agent.py          357)         )
src/derivepassphrase/ssh_agent.py          358)         return returncode == _types.SSH_AGENT.SUCCESS.value
src/derivepassphrase/ssh_agent.py          359) 

src/derivepassphrase/ssh_agent.py          360)     @overload
src/derivepassphrase/ssh_agent.py          361)     def request(  # pragma: no cover
src/derivepassphrase/ssh_agent.py          362)         self,
src/derivepassphrase/ssh_agent.py          363)         code: int | _types.SSH_AGENTC,

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

src/derivepassphrase/ssh_agent.py          365)         /,
src/derivepassphrase/ssh_agent.py          366)         *,
src/derivepassphrase/ssh_agent.py          367)         response_code: None = None,

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

src/derivepassphrase/ssh_agent.py          369) 
src/derivepassphrase/ssh_agent.py          370)     @overload
src/derivepassphrase/ssh_agent.py          371)     def request(  # pragma: no cover
src/derivepassphrase/ssh_agent.py          372)         self,
src/derivepassphrase/ssh_agent.py          373)         code: int | _types.SSH_AGENTC,

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

src/derivepassphrase/ssh_agent.py          375)         /,
src/derivepassphrase/ssh_agent.py          376)         *,
src/derivepassphrase/ssh_agent.py          377)         response_code: Iterable[_types.SSH_AGENT | int] = frozenset({
src/derivepassphrase/ssh_agent.py          378)             _types.SSH_AGENT.SUCCESS
src/derivepassphrase/ssh_agent.py          379)         }),

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

src/derivepassphrase/ssh_agent.py          381) 
src/derivepassphrase/ssh_agent.py          382)     @overload
src/derivepassphrase/ssh_agent.py          383)     def request(  # pragma: no cover
src/derivepassphrase/ssh_agent.py          384)         self,
src/derivepassphrase/ssh_agent.py          385)         code: int | _types.SSH_AGENTC,

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

src/derivepassphrase/ssh_agent.py          387)         /,
src/derivepassphrase/ssh_agent.py          388)         *,
src/derivepassphrase/ssh_agent.py          389)         response_code: _types.SSH_AGENT | int = _types.SSH_AGENT.SUCCESS,

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

src/derivepassphrase/ssh_agent.py          391) 

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

src/derivepassphrase/ssh_agent.py          393)         self,
src/derivepassphrase/ssh_agent.py          394)         code: int | _types.SSH_AGENTC,

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

src/derivepassphrase/ssh_agent.py          396)         /,
src/derivepassphrase/ssh_agent.py          397)         *,
src/derivepassphrase/ssh_agent.py          398)         response_code: (
src/derivepassphrase/ssh_agent.py          399)             Iterable[_types.SSH_AGENT | int] | _types.SSH_AGENT | int | None
src/derivepassphrase/ssh_agent.py          400)         ) = None,

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

src/ssh_agent_client/__init__.py           402)         """Issue a generic request to the SSH agent.
src/ssh_agent_client/__init__.py           403) 
src/ssh_agent_client/__init__.py           404)         Args:
src/ssh_agent_client/__init__.py           405)             code:
src/ssh_agent_client/__init__.py           406)                 The request code.  See the SSH agent protocol for
src/ssh_agent_client/__init__.py           407)                 protocol numbers to use here (and which protocol numbers
src/ssh_agent_client/__init__.py           408)                 to expect in a response).
src/ssh_agent_client/__init__.py           409)             payload:

src/derivepassphrase/ssh_agent.py          410)                 A bytes-like object containing the payload, or
src/derivepassphrase/ssh_agent.py          411)                 "contents", of the request.  Request-specific.
src/derivepassphrase/ssh_agent.py          412) 
src/derivepassphrase/ssh_agent.py          413)                 It is our responsibility to add any necessary wire
src/derivepassphrase/ssh_agent.py          414)                 framing around the request code and the payload,
src/derivepassphrase/ssh_agent.py          415)                 not the caller's.

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

src/ssh_agent_client/__init__.py           420) 
src/ssh_agent_client/__init__.py           421)         Returns:
src/ssh_agent_client/__init__.py           422)             A 2-tuple consisting of the response code and the payload,
src/ssh_agent_client/__init__.py           423)             with all wire framing removed.
src/ssh_agent_client/__init__.py           424) 

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

src/ssh_agent_client/__init__.py           427)         Raises:
src/ssh_agent_client/__init__.py           428)             EOFError:
src/ssh_agent_client/__init__.py           429)                 The response from the SSH agent is truncated or missing.

src/derivepassphrase/ssh_agent.py          430)             OSError:
src/derivepassphrase/ssh_agent.py          431)                 There was a communication error with the SSH agent.
src/derivepassphrase/ssh_agent.py          432)             SSHAgentFailedError:
src/derivepassphrase/ssh_agent.py          433)                 We expected specific response codes, but did not receive
src/derivepassphrase/ssh_agent.py          434)                 any of them.

src/ssh_agent_client/__init__.py           435) 
src/ssh_agent_client/__init__.py           436)         """

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

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

src/derivepassphrase/ssh_agent.py          439)         ):
src/derivepassphrase/ssh_agent.py          440)             response_code = frozenset({response_code})
src/derivepassphrase/ssh_agent.py          441)         if response_code is not None:  # pragma: no branch
src/derivepassphrase/ssh_agent.py          442)             response_code = frozenset({
src/derivepassphrase/ssh_agent.py          443)                 c if isinstance(c, int) else c.value for c in response_code
src/derivepassphrase/ssh_agent.py          444)             })

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

src/derivepassphrase/ssh_agent.py          446)         request_message = bytearray([
src/derivepassphrase/ssh_agent.py          447)             code if isinstance(code, int) else code.value
src/derivepassphrase/ssh_agent.py          448)         ])

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

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

src/ssh_agent_client/__init__.py           451)         chunk = self._connection.recv(HEAD_LEN)
src/ssh_agent_client/__init__.py           452)         if len(chunk) < HEAD_LEN:
src/ssh_agent_client/__init__.py           453)             msg = 'cannot read response length'
src/ssh_agent_client/__init__.py           454)             raise EOFError(msg)

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

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

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

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

src/derivepassphrase/ssh_agent.py          460)         if not response_code:  # pragma: no cover
src/derivepassphrase/ssh_agent.py          461)             return response[0], response[1:]
src/derivepassphrase/ssh_agent.py          462)         if response[0] not in response_code:
src/derivepassphrase/ssh_agent.py          463)             raise SSHAgentFailedError(response[0], response[1:])
src/derivepassphrase/ssh_agent.py          464)         return response[1:]

src/ssh_agent_client/__init__.py           465) 

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

src/ssh_agent_client/__init__.py           467)         """Request a list of keys known to the SSH agent.
src/ssh_agent_client/__init__.py           468) 
src/ssh_agent_client/__init__.py           469)         Returns:
src/ssh_agent_client/__init__.py           470)             A read-only sequence of key/comment pairs.
src/ssh_agent_client/__init__.py           471) 
src/ssh_agent_client/__init__.py           472)         Raises:
src/ssh_agent_client/__init__.py           473)             EOFError:
src/ssh_agent_client/__init__.py           474)                 The response from the SSH agent is truncated or missing.

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

src/ssh_agent_client/__init__.py           477)             TrailingDataError:

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

src/derivepassphrase/ssh_agent.py          479)             SSHAgentFailedError:

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

src/ssh_agent_client/__init__.py           481) 
src/ssh_agent_client/__init__.py           482)         """

src/derivepassphrase/ssh_agent.py          483)         response = self.request(
src/derivepassphrase/ssh_agent.py          484)             _types.SSH_AGENTC.REQUEST_IDENTITIES.value,
src/derivepassphrase/ssh_agent.py          485)             b'',
src/derivepassphrase/ssh_agent.py          486)             response_code=_types.SSH_AGENT.IDENTITIES_ANSWER,

src/ssh_agent_client/__init__.py           487)         )

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

src/ssh_agent_client/__init__.py           489) 

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

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

src/ssh_agent_client/__init__.py           493)                 try:
src/ssh_agent_client/__init__.py           494)                     val = response_stream.popleft()
src/ssh_agent_client/__init__.py           495)                 except IndexError:
src/ssh_agent_client/__init__.py           496)                     response_stream.extendleft(reversed(buf))

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

src/ssh_agent_client/__init__.py           499)                 buf.append(val)
src/ssh_agent_client/__init__.py           500)             return bytes(buf)

src/ssh_agent_client/__init__.py           501) 

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

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

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

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

src/ssh_agent_client/__init__.py           506)             key_size = int.from_bytes(shift(4), 'big')
src/ssh_agent_client/__init__.py           507)             key = shift(key_size)
src/ssh_agent_client/__init__.py           508)             comment_size = int.from_bytes(shift(4), 'big')
src/ssh_agent_client/__init__.py           509)             comment = shift(comment_size)
src/ssh_agent_client/__init__.py           510)             # Both `key` and `comment` are not wrapped as SSH strings.

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

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

src/ssh_agent_client/__init__.py           513)             raise TrailingDataError

src/ssh_agent_client/__init__.py           514)         return keys
src/ssh_agent_client/__init__.py           515) 
src/ssh_agent_client/__init__.py           516)     def sign(

src/ssh_agent_client/__init__.py           517)         self,
src/ssh_agent_client/__init__.py           518)         /,

src/derivepassphrase/ssh_agent.py          519)         key: Buffer,
src/derivepassphrase/ssh_agent.py          520)         payload: Buffer,

src/ssh_agent_client/__init__.py           521)         *,
src/ssh_agent_client/__init__.py           522)         flags: int = 0,
src/ssh_agent_client/__init__.py           523)         check_if_key_loaded: bool = False,

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

src/ssh_agent_client/__init__.py           525)         """Request the SSH agent sign the payload with the key.
src/ssh_agent_client/__init__.py           526) 
src/ssh_agent_client/__init__.py           527)         Args:
src/ssh_agent_client/__init__.py           528)             key:
src/ssh_agent_client/__init__.py           529)                 The public SSH key to sign the payload with, in the same

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

src/ssh_agent_client/__init__.py           531)                 The corresponding private key must have previously been
src/ssh_agent_client/__init__.py           532)                 loaded into the agent to successfully issue a signature.
src/ssh_agent_client/__init__.py           533)             payload:
src/ssh_agent_client/__init__.py           534)                 A byte string of data to sign.
src/ssh_agent_client/__init__.py           535)             flags:
src/ssh_agent_client/__init__.py           536)                 Optional flags for the signing request.  Currently
src/ssh_agent_client/__init__.py           537)                 passed on as-is to the agent.  In real-world usage, this
src/ssh_agent_client/__init__.py           538)                 could be used, e.g., to request more modern hash
src/ssh_agent_client/__init__.py           539)                 algorithms when signing with RSA keys.  (No such
src/ssh_agent_client/__init__.py           540)                 real-world usage is currently implemented.)
src/ssh_agent_client/__init__.py           541)             check_if_key_loaded:

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

src/ssh_agent_client/__init__.py           543)                 corresponding key has been loaded into the agent.
src/ssh_agent_client/__init__.py           544) 
src/ssh_agent_client/__init__.py           545)         Returns:
src/ssh_agent_client/__init__.py           546)             The binary signature of the payload under the given key.
src/ssh_agent_client/__init__.py           547) 
src/ssh_agent_client/__init__.py           548)         Raises:
src/ssh_agent_client/__init__.py           549)             EOFError:
src/ssh_agent_client/__init__.py           550)                 The response from the SSH agent is truncated or missing.

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

src/ssh_agent_client/__init__.py           553)             TrailingDataError:

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

src/derivepassphrase/ssh_agent.py          555)             SSHAgentFailedError:

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

src/ssh_agent_client/__init__.py           557)             KeyError:

src/ssh_agent_client/__init__.py           558)                 `check_if_key_loaded` is true, and the `key` was not
src/ssh_agent_client/__init__.py           559)                 loaded into the agent.
src/ssh_agent_client/__init__.py           560) 
src/ssh_agent_client/__init__.py           561)         """

src/derivepassphrase/ssh_agent.py          562)         key = memoryview(key)
src/derivepassphrase/ssh_agent.py          563)         payload = memoryview(payload)

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

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

src/ssh_agent_client/__init__.py           569)         request_data = bytearray(self.string(key))
src/ssh_agent_client/__init__.py           570)         request_data.extend(self.string(payload))
src/ssh_agent_client/__init__.py           571)         request_data.extend(self.uint32(flags))

src/derivepassphrase/ssh_agent.py          572)         return bytes(
src/derivepassphrase/ssh_agent.py          573)             self.unstring(
src/derivepassphrase/ssh_agent.py          574)                 self.request(
src/derivepassphrase/ssh_agent.py          575)                     _types.SSH_AGENTC.SIGN_REQUEST.value,
src/derivepassphrase/ssh_agent.py          576)                     request_data,
src/derivepassphrase/ssh_agent.py          577)                     response_code=_types.SSH_AGENT.SIGN_RESPONSE,
src/derivepassphrase/ssh_agent.py          578)                 )

src/derivepassphrase/ssh_agent.py          579)             )