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/derivepassphrase/ssh_agent.py           21)     from collections.abc import Set as AbstractSet

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

src/ssh_agent_client/__init__.py            23) 

src/derivepassphrase/ssh_agent.py           24)     from typing_extensions import Buffer
src/derivepassphrase/ssh_agent.py           25) 

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

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

src/ssh_agent_client/__init__.py            28) 

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

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

src/ssh_agent_client/__init__.py            35) 

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

src/ssh_agent_client/__init__.py            38) 

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

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

src/ssh_agent_client/__init__.py            41) 

src/ssh_agent_client/__init__.py            42) 

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

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

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

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

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

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

src/ssh_agent_client/__init__.py            79) 

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

src/ssh_agent_client/__init__.py            81) 

src/ssh_agent_client/__init__.py            82)     def __init__(
src/ssh_agent_client/__init__.py            83)         self, /, *, socket: socket.socket | None = None, timeout: int = 125
src/ssh_agent_client/__init__.py            84)     ) -> None:
src/ssh_agent_client/__init__.py            85)         """Initialize the client.
src/ssh_agent_client/__init__.py            86) 
src/ssh_agent_client/__init__.py            87)         Args:
src/ssh_agent_client/__init__.py            88)             socket:

src/derivepassphrase/ssh_agent.py           89)                 An optional socket, already connected to the SSH agent.
src/derivepassphrase/ssh_agent.py           90)                 If not given, we query the `SSH_AUTH_SOCK` environment

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

src/derivepassphrase/ssh_agent.py          136)         """Close socket connection upon context manager completion.
src/derivepassphrase/ssh_agent.py          137) 
src/derivepassphrase/ssh_agent.py          138)         Returns:
src/derivepassphrase/ssh_agent.py          139)             Self.
src/derivepassphrase/ssh_agent.py          140) 
src/derivepassphrase/ssh_agent.py          141)         """

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

src/ssh_agent_client/__init__.py           205)         try:

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

src/ssh_agent_client/__init__.py           244) 

src/ssh_agent_client/__init__.py           245)     @classmethod

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

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

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

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

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

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

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

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

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

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

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

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

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

src/ssh_agent_client/__init__.py           282)         return (

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

src/ssh_agent_client/__init__.py           285)         )

src/ssh_agent_client/__init__.py           286) 

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

src/derivepassphrase/ssh_agent.py          342)     def _agent_is_pageant(self) -> bool:
src/derivepassphrase/ssh_agent.py          343)         """Return True if we are connected to Pageant.
src/derivepassphrase/ssh_agent.py          344) 
src/derivepassphrase/ssh_agent.py          345)         Warning:
src/derivepassphrase/ssh_agent.py          346)             This is a heuristic, not a verified query or computation.
src/derivepassphrase/ssh_agent.py          347) 
src/derivepassphrase/ssh_agent.py          348)         """
src/derivepassphrase/ssh_agent.py          349)         return (
src/derivepassphrase/ssh_agent.py          350)             b'list-extended@putty.projects.tartarus.org'
src/derivepassphrase/ssh_agent.py          351)             in self.query_extensions()
src/derivepassphrase/ssh_agent.py          352)         )
src/derivepassphrase/ssh_agent.py          353) 

src/derivepassphrase/ssh_agent.py          354)     def has_deterministic_signatures(self) -> bool:
src/derivepassphrase/ssh_agent.py          355)         """Check whether the agent returns deterministic signatures.
src/derivepassphrase/ssh_agent.py          356) 
src/derivepassphrase/ssh_agent.py          357)         Returns:
src/derivepassphrase/ssh_agent.py          358)             True if a known agent was detected where signatures are
src/derivepassphrase/ssh_agent.py          359)             deterministic for all SSH key types, false otherwise.
src/derivepassphrase/ssh_agent.py          360) 
src/derivepassphrase/ssh_agent.py          361)         Note: Known agents with deterministic signatures
src/derivepassphrase/ssh_agent.py          362)             | agent           | detected via                                                  |
src/derivepassphrase/ssh_agent.py          363)             |:----------------|:--------------------------------------------------------------|
src/derivepassphrase/ssh_agent.py          364)             | Pageant (PuTTY) | `list-extended@putty.projects.tartarus.org` extension request |
src/derivepassphrase/ssh_agent.py          365) 
src/derivepassphrase/ssh_agent.py          366)         """  # noqa: E501

src/derivepassphrase/ssh_agent.py          367)         known_good_agents = {
src/derivepassphrase/ssh_agent.py          368)             'Pageant': self._agent_is_pageant,
src/derivepassphrase/ssh_agent.py          369)         }
src/derivepassphrase/ssh_agent.py          370)         return any(  # pragma: no branch
src/derivepassphrase/ssh_agent.py          371)             v() for v in known_good_agents.values()

src/derivepassphrase/ssh_agent.py          372)         )
src/derivepassphrase/ssh_agent.py          373) 

src/derivepassphrase/ssh_agent.py          374)     @overload
src/derivepassphrase/ssh_agent.py          375)     def request(  # pragma: no cover
src/derivepassphrase/ssh_agent.py          376)         self,
src/derivepassphrase/ssh_agent.py          377)         code: int | _types.SSH_AGENTC,

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

src/derivepassphrase/ssh_agent.py          379)         /,
src/derivepassphrase/ssh_agent.py          380)         *,
src/derivepassphrase/ssh_agent.py          381)         response_code: None = None,

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

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

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

src/derivepassphrase/ssh_agent.py          389)         /,
src/derivepassphrase/ssh_agent.py          390)         *,
src/derivepassphrase/ssh_agent.py          391)         response_code: Iterable[_types.SSH_AGENT | int] = frozenset({
src/derivepassphrase/ssh_agent.py          392)             _types.SSH_AGENT.SUCCESS
src/derivepassphrase/ssh_agent.py          393)         }),

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

src/derivepassphrase/ssh_agent.py          395) 
src/derivepassphrase/ssh_agent.py          396)     @overload
src/derivepassphrase/ssh_agent.py          397)     def request(  # pragma: no cover
src/derivepassphrase/ssh_agent.py          398)         self,
src/derivepassphrase/ssh_agent.py          399)         code: int | _types.SSH_AGENTC,

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

src/derivepassphrase/ssh_agent.py          401)         /,
src/derivepassphrase/ssh_agent.py          402)         *,
src/derivepassphrase/ssh_agent.py          403)         response_code: _types.SSH_AGENT | int = _types.SSH_AGENT.SUCCESS,

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

src/derivepassphrase/ssh_agent.py          405) 

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

src/derivepassphrase/ssh_agent.py          407)         self,
src/derivepassphrase/ssh_agent.py          408)         code: int | _types.SSH_AGENTC,

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

src/derivepassphrase/ssh_agent.py          410)         /,
src/derivepassphrase/ssh_agent.py          411)         *,
src/derivepassphrase/ssh_agent.py          412)         response_code: (
src/derivepassphrase/ssh_agent.py          413)             Iterable[_types.SSH_AGENT | int] | _types.SSH_AGENT | int | None
src/derivepassphrase/ssh_agent.py          414)         ) = None,

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

src/ssh_agent_client/__init__.py           416)         """Issue a generic request to the SSH agent.
src/ssh_agent_client/__init__.py           417) 
src/ssh_agent_client/__init__.py           418)         Args:
src/ssh_agent_client/__init__.py           419)             code:
src/ssh_agent_client/__init__.py           420)                 The request code.  See the SSH agent protocol for
src/ssh_agent_client/__init__.py           421)                 protocol numbers to use here (and which protocol numbers
src/ssh_agent_client/__init__.py           422)                 to expect in a response).
src/ssh_agent_client/__init__.py           423)             payload:

src/derivepassphrase/ssh_agent.py          424)                 A bytes-like object containing the payload, or
src/derivepassphrase/ssh_agent.py          425)                 "contents", of the request.  Request-specific.
src/derivepassphrase/ssh_agent.py          426) 
src/derivepassphrase/ssh_agent.py          427)                 It is our responsibility to add any necessary wire
src/derivepassphrase/ssh_agent.py          428)                 framing around the request code and the payload,
src/derivepassphrase/ssh_agent.py          429)                 not the caller's.

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

src/ssh_agent_client/__init__.py           434) 
src/ssh_agent_client/__init__.py           435)         Returns:
src/ssh_agent_client/__init__.py           436)             A 2-tuple consisting of the response code and the payload,
src/ssh_agent_client/__init__.py           437)             with all wire framing removed.
src/ssh_agent_client/__init__.py           438) 

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

src/ssh_agent_client/__init__.py           441)         Raises:
src/ssh_agent_client/__init__.py           442)             EOFError:
src/ssh_agent_client/__init__.py           443)                 The response from the SSH agent is truncated or missing.

src/derivepassphrase/ssh_agent.py          444)             OSError:
src/derivepassphrase/ssh_agent.py          445)                 There was a communication error with the SSH agent.
src/derivepassphrase/ssh_agent.py          446)             SSHAgentFailedError:
src/derivepassphrase/ssh_agent.py          447)                 We expected specific response codes, but did not receive
src/derivepassphrase/ssh_agent.py          448)                 any of them.

src/ssh_agent_client/__init__.py           449) 
src/ssh_agent_client/__init__.py           450)         """

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

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

src/derivepassphrase/ssh_agent.py          453)         ):
src/derivepassphrase/ssh_agent.py          454)             response_code = frozenset({response_code})
src/derivepassphrase/ssh_agent.py          455)         if response_code is not None:  # pragma: no branch
src/derivepassphrase/ssh_agent.py          456)             response_code = frozenset({
src/derivepassphrase/ssh_agent.py          457)                 c if isinstance(c, int) else c.value for c in response_code
src/derivepassphrase/ssh_agent.py          458)             })

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

src/derivepassphrase/ssh_agent.py          460)         request_message = bytearray([
src/derivepassphrase/ssh_agent.py          461)             code if isinstance(code, int) else code.value
src/derivepassphrase/ssh_agent.py          462)         ])

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

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

src/ssh_agent_client/__init__.py           465)         chunk = self._connection.recv(HEAD_LEN)
src/ssh_agent_client/__init__.py           466)         if len(chunk) < HEAD_LEN:
src/ssh_agent_client/__init__.py           467)             msg = 'cannot read response length'
src/ssh_agent_client/__init__.py           468)             raise EOFError(msg)

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

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

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

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

src/derivepassphrase/ssh_agent.py          474)         if not response_code:  # pragma: no cover
src/derivepassphrase/ssh_agent.py          475)             return response[0], response[1:]
src/derivepassphrase/ssh_agent.py          476)         if response[0] not in response_code:
src/derivepassphrase/ssh_agent.py          477)             raise SSHAgentFailedError(response[0], response[1:])
src/derivepassphrase/ssh_agent.py          478)         return response[1:]

src/ssh_agent_client/__init__.py           479) 

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

src/ssh_agent_client/__init__.py           481)         """Request a list of keys known to the SSH agent.
src/ssh_agent_client/__init__.py           482) 
src/ssh_agent_client/__init__.py           483)         Returns:
src/ssh_agent_client/__init__.py           484)             A read-only sequence of key/comment pairs.
src/ssh_agent_client/__init__.py           485) 
src/ssh_agent_client/__init__.py           486)         Raises:
src/ssh_agent_client/__init__.py           487)             EOFError:
src/ssh_agent_client/__init__.py           488)                 The response from the SSH agent is truncated or missing.

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

src/ssh_agent_client/__init__.py           491)             TrailingDataError:

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

src/derivepassphrase/ssh_agent.py          493)             SSHAgentFailedError:

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

src/ssh_agent_client/__init__.py           495) 
src/ssh_agent_client/__init__.py           496)         """

src/derivepassphrase/ssh_agent.py          497)         response = self.request(
src/derivepassphrase/ssh_agent.py          498)             _types.SSH_AGENTC.REQUEST_IDENTITIES.value,
src/derivepassphrase/ssh_agent.py          499)             b'',
src/derivepassphrase/ssh_agent.py          500)             response_code=_types.SSH_AGENT.IDENTITIES_ANSWER,

src/ssh_agent_client/__init__.py           501)         )

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

src/ssh_agent_client/__init__.py           503) 

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

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

src/ssh_agent_client/__init__.py           507)                 try:
src/ssh_agent_client/__init__.py           508)                     val = response_stream.popleft()
src/ssh_agent_client/__init__.py           509)                 except IndexError:
src/ssh_agent_client/__init__.py           510)                     response_stream.extendleft(reversed(buf))

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

src/ssh_agent_client/__init__.py           513)                 buf.append(val)
src/ssh_agent_client/__init__.py           514)             return bytes(buf)

src/ssh_agent_client/__init__.py           515) 

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

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

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

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

src/ssh_agent_client/__init__.py           520)             key_size = int.from_bytes(shift(4), 'big')
src/ssh_agent_client/__init__.py           521)             key = shift(key_size)
src/ssh_agent_client/__init__.py           522)             comment_size = int.from_bytes(shift(4), 'big')
src/ssh_agent_client/__init__.py           523)             comment = shift(comment_size)
src/ssh_agent_client/__init__.py           524)             # Both `key` and `comment` are not wrapped as SSH strings.

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

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

src/ssh_agent_client/__init__.py           527)             raise TrailingDataError

src/ssh_agent_client/__init__.py           528)         return keys
src/ssh_agent_client/__init__.py           529) 
src/ssh_agent_client/__init__.py           530)     def sign(

src/ssh_agent_client/__init__.py           531)         self,
src/ssh_agent_client/__init__.py           532)         /,

src/derivepassphrase/ssh_agent.py          533)         key: Buffer,
src/derivepassphrase/ssh_agent.py          534)         payload: Buffer,

src/ssh_agent_client/__init__.py           535)         *,
src/ssh_agent_client/__init__.py           536)         flags: int = 0,
src/ssh_agent_client/__init__.py           537)         check_if_key_loaded: bool = False,

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

src/ssh_agent_client/__init__.py           539)         """Request the SSH agent sign the payload with the key.
src/ssh_agent_client/__init__.py           540) 
src/ssh_agent_client/__init__.py           541)         Args:
src/ssh_agent_client/__init__.py           542)             key:
src/ssh_agent_client/__init__.py           543)                 The public SSH key to sign the payload with, in the same

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

src/ssh_agent_client/__init__.py           545)                 The corresponding private key must have previously been
src/ssh_agent_client/__init__.py           546)                 loaded into the agent to successfully issue a signature.
src/ssh_agent_client/__init__.py           547)             payload:
src/ssh_agent_client/__init__.py           548)                 A byte string of data to sign.
src/ssh_agent_client/__init__.py           549)             flags:
src/ssh_agent_client/__init__.py           550)                 Optional flags for the signing request.  Currently
src/ssh_agent_client/__init__.py           551)                 passed on as-is to the agent.  In real-world usage, this
src/ssh_agent_client/__init__.py           552)                 could be used, e.g., to request more modern hash
src/ssh_agent_client/__init__.py           553)                 algorithms when signing with RSA keys.  (No such
src/ssh_agent_client/__init__.py           554)                 real-world usage is currently implemented.)
src/ssh_agent_client/__init__.py           555)             check_if_key_loaded:

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

src/ssh_agent_client/__init__.py           557)                 corresponding key has been loaded into the agent.
src/ssh_agent_client/__init__.py           558) 
src/ssh_agent_client/__init__.py           559)         Returns:
src/ssh_agent_client/__init__.py           560)             The binary signature of the payload under the given key.
src/ssh_agent_client/__init__.py           561) 
src/ssh_agent_client/__init__.py           562)         Raises:
src/ssh_agent_client/__init__.py           563)             EOFError:
src/ssh_agent_client/__init__.py           564)                 The response from the SSH agent is truncated or missing.

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

src/ssh_agent_client/__init__.py           567)             TrailingDataError:

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

src/derivepassphrase/ssh_agent.py          569)             SSHAgentFailedError:

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

src/ssh_agent_client/__init__.py           571)             KeyError:

src/ssh_agent_client/__init__.py           572)                 `check_if_key_loaded` is true, and the `key` was not
src/ssh_agent_client/__init__.py           573)                 loaded into the agent.
src/ssh_agent_client/__init__.py           574) 
src/ssh_agent_client/__init__.py           575)         """

src/derivepassphrase/ssh_agent.py          576)         key = memoryview(key)
src/derivepassphrase/ssh_agent.py          577)         payload = memoryview(payload)

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

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

src/ssh_agent_client/__init__.py           583)         request_data = bytearray(self.string(key))
src/ssh_agent_client/__init__.py           584)         request_data.extend(self.string(payload))
src/ssh_agent_client/__init__.py           585)         request_data.extend(self.uint32(flags))

src/derivepassphrase/ssh_agent.py          586)         return bytes(
src/derivepassphrase/ssh_agent.py          587)             self.unstring(
src/derivepassphrase/ssh_agent.py          588)                 self.request(
src/derivepassphrase/ssh_agent.py          589)                     _types.SSH_AGENTC.SIGN_REQUEST.value,
src/derivepassphrase/ssh_agent.py          590)                     request_data,
src/derivepassphrase/ssh_agent.py          591)                     response_code=_types.SSH_AGENT.SIGN_RESPONSE,
src/derivepassphrase/ssh_agent.py          592)                 )

src/derivepassphrase/ssh_agent.py          593)             )

src/ssh_agent_client/__init__.py           594)         )