derivepassphrase.git

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

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

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

src/ssh_agent_client/__init__.py            13) 

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

src/ssh_agent_client/__init__.py            15) 

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

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

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

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

src/ssh_agent_client/__init__.py            21) 

src/derivepassphrase/ssh_agent.py           22)     from typing_extensions import Buffer
src/derivepassphrase/ssh_agent.py           23) 

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

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

src/ssh_agent_client/__init__.py            26) 

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

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

src/ssh_agent_client/__init__.py            33) 

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

src/ssh_agent_client/__init__.py            36) 

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

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

src/ssh_agent_client/__init__.py            39) 

src/ssh_agent_client/__init__.py            40) 

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

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

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

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

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

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

src/ssh_agent_client/__init__.py            77) 

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

src/ssh_agent_client/__init__.py            79) 

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

src/ssh_agent_client/__init__.py           203)         try:

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

src/ssh_agent_client/__init__.py           242) 

src/ssh_agent_client/__init__.py           243)     @classmethod

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

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

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

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

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

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

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

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

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

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

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

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

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

src/ssh_agent_client/__init__.py           280)         return (

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

src/ssh_agent_client/__init__.py           283)         )

src/ssh_agent_client/__init__.py           284) 

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

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

src/derivepassphrase/ssh_agent.py          290)         /,
src/derivepassphrase/ssh_agent.py          291)         *,
src/derivepassphrase/ssh_agent.py          292)         response_code: None = None,

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

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

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

src/derivepassphrase/ssh_agent.py          300)         /,
src/derivepassphrase/ssh_agent.py          301)         *,
src/derivepassphrase/ssh_agent.py          302)         response_code: Iterable[_types.SSH_AGENT | int] = frozenset({
src/derivepassphrase/ssh_agent.py          303)             _types.SSH_AGENT.SUCCESS
src/derivepassphrase/ssh_agent.py          304)         }),

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

src/derivepassphrase/ssh_agent.py          306) 
src/derivepassphrase/ssh_agent.py          307)     @overload
src/derivepassphrase/ssh_agent.py          308)     def request(  # pragma: no cover
src/derivepassphrase/ssh_agent.py          309)         self,
src/derivepassphrase/ssh_agent.py          310)         code: int | _types.SSH_AGENTC,

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

src/derivepassphrase/ssh_agent.py          312)         /,
src/derivepassphrase/ssh_agent.py          313)         *,
src/derivepassphrase/ssh_agent.py          314)         response_code: _types.SSH_AGENT | int = _types.SSH_AGENT.SUCCESS,

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

src/derivepassphrase/ssh_agent.py          316) 

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

src/derivepassphrase/ssh_agent.py          318)         self,
src/derivepassphrase/ssh_agent.py          319)         code: int | _types.SSH_AGENTC,

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

src/derivepassphrase/ssh_agent.py          321)         /,
src/derivepassphrase/ssh_agent.py          322)         *,
src/derivepassphrase/ssh_agent.py          323)         response_code: (
src/derivepassphrase/ssh_agent.py          324)             Iterable[_types.SSH_AGENT | int] | _types.SSH_AGENT | int | None
src/derivepassphrase/ssh_agent.py          325)         ) = None,

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

src/ssh_agent_client/__init__.py           327)         """Issue a generic request to the SSH agent.
src/ssh_agent_client/__init__.py           328) 
src/ssh_agent_client/__init__.py           329)         Args:
src/ssh_agent_client/__init__.py           330)             code:
src/ssh_agent_client/__init__.py           331)                 The request code.  See the SSH agent protocol for
src/ssh_agent_client/__init__.py           332)                 protocol numbers to use here (and which protocol numbers
src/ssh_agent_client/__init__.py           333)                 to expect in a response).
src/ssh_agent_client/__init__.py           334)             payload:

src/derivepassphrase/ssh_agent.py          335)                 A bytes-like object containing the payload, or
src/derivepassphrase/ssh_agent.py          336)                 "contents", of the request.  Request-specific.
src/derivepassphrase/ssh_agent.py          337) 
src/derivepassphrase/ssh_agent.py          338)                 It is our responsibility to add any necessary wire
src/derivepassphrase/ssh_agent.py          339)                 framing around the request code and the payload,
src/derivepassphrase/ssh_agent.py          340)                 not the caller's.

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

src/ssh_agent_client/__init__.py           345) 
src/ssh_agent_client/__init__.py           346)         Returns:
src/ssh_agent_client/__init__.py           347)             A 2-tuple consisting of the response code and the payload,
src/ssh_agent_client/__init__.py           348)             with all wire framing removed.
src/ssh_agent_client/__init__.py           349) 

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

src/ssh_agent_client/__init__.py           352)         Raises:
src/ssh_agent_client/__init__.py           353)             EOFError:
src/ssh_agent_client/__init__.py           354)                 The response from the SSH agent is truncated or missing.

src/derivepassphrase/ssh_agent.py          355)             OSError:
src/derivepassphrase/ssh_agent.py          356)                 There was a communication error with the SSH agent.
src/derivepassphrase/ssh_agent.py          357)             SSHAgentFailedError:
src/derivepassphrase/ssh_agent.py          358)                 We expected specific response codes, but did not receive
src/derivepassphrase/ssh_agent.py          359)                 any of them.

src/ssh_agent_client/__init__.py           360) 
src/ssh_agent_client/__init__.py           361)         """

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

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

src/derivepassphrase/ssh_agent.py          364)         ):
src/derivepassphrase/ssh_agent.py          365)             response_code = frozenset({response_code})
src/derivepassphrase/ssh_agent.py          366)         if response_code is not None:  # pragma: no branch
src/derivepassphrase/ssh_agent.py          367)             response_code = frozenset({
src/derivepassphrase/ssh_agent.py          368)                 c if isinstance(c, int) else c.value for c in response_code
src/derivepassphrase/ssh_agent.py          369)             })

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

src/derivepassphrase/ssh_agent.py          371)         request_message = bytearray([
src/derivepassphrase/ssh_agent.py          372)             code if isinstance(code, int) else code.value
src/derivepassphrase/ssh_agent.py          373)         ])

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

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

src/ssh_agent_client/__init__.py           376)         chunk = self._connection.recv(HEAD_LEN)
src/ssh_agent_client/__init__.py           377)         if len(chunk) < HEAD_LEN:
src/ssh_agent_client/__init__.py           378)             msg = 'cannot read response length'
src/ssh_agent_client/__init__.py           379)             raise EOFError(msg)

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

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

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

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

src/derivepassphrase/ssh_agent.py          385)         if not response_code:  # pragma: no cover
src/derivepassphrase/ssh_agent.py          386)             return response[0], response[1:]
src/derivepassphrase/ssh_agent.py          387)         if response[0] not in response_code:
src/derivepassphrase/ssh_agent.py          388)             raise SSHAgentFailedError(response[0], response[1:])
src/derivepassphrase/ssh_agent.py          389)         return response[1:]

src/ssh_agent_client/__init__.py           390) 

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

src/ssh_agent_client/__init__.py           392)         """Request a list of keys known to the SSH agent.
src/ssh_agent_client/__init__.py           393) 
src/ssh_agent_client/__init__.py           394)         Returns:
src/ssh_agent_client/__init__.py           395)             A read-only sequence of key/comment pairs.
src/ssh_agent_client/__init__.py           396) 
src/ssh_agent_client/__init__.py           397)         Raises:
src/ssh_agent_client/__init__.py           398)             EOFError:
src/ssh_agent_client/__init__.py           399)                 The response from the SSH agent is truncated or missing.

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

src/ssh_agent_client/__init__.py           402)             TrailingDataError:

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

src/derivepassphrase/ssh_agent.py          404)             SSHAgentFailedError:

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

src/ssh_agent_client/__init__.py           406) 
src/ssh_agent_client/__init__.py           407)         """

src/derivepassphrase/ssh_agent.py          408)         response = self.request(
src/derivepassphrase/ssh_agent.py          409)             _types.SSH_AGENTC.REQUEST_IDENTITIES.value,
src/derivepassphrase/ssh_agent.py          410)             b'',
src/derivepassphrase/ssh_agent.py          411)             response_code=_types.SSH_AGENT.IDENTITIES_ANSWER,

src/ssh_agent_client/__init__.py           412)         )

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

src/ssh_agent_client/__init__.py           414) 

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

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

src/ssh_agent_client/__init__.py           418)                 try:
src/ssh_agent_client/__init__.py           419)                     val = response_stream.popleft()
src/ssh_agent_client/__init__.py           420)                 except IndexError:
src/ssh_agent_client/__init__.py           421)                     response_stream.extendleft(reversed(buf))

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

src/ssh_agent_client/__init__.py           424)                 buf.append(val)
src/ssh_agent_client/__init__.py           425)             return bytes(buf)

src/ssh_agent_client/__init__.py           426) 

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

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

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

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

src/ssh_agent_client/__init__.py           431)             key_size = int.from_bytes(shift(4), 'big')
src/ssh_agent_client/__init__.py           432)             key = shift(key_size)
src/ssh_agent_client/__init__.py           433)             comment_size = int.from_bytes(shift(4), 'big')
src/ssh_agent_client/__init__.py           434)             comment = shift(comment_size)
src/ssh_agent_client/__init__.py           435)             # Both `key` and `comment` are not wrapped as SSH strings.

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

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

src/ssh_agent_client/__init__.py           438)             raise TrailingDataError

src/ssh_agent_client/__init__.py           439)         return keys
src/ssh_agent_client/__init__.py           440) 
src/ssh_agent_client/__init__.py           441)     def sign(

src/ssh_agent_client/__init__.py           442)         self,
src/ssh_agent_client/__init__.py           443)         /,

src/derivepassphrase/ssh_agent.py          444)         key: Buffer,
src/derivepassphrase/ssh_agent.py          445)         payload: Buffer,

src/ssh_agent_client/__init__.py           446)         *,
src/ssh_agent_client/__init__.py           447)         flags: int = 0,
src/ssh_agent_client/__init__.py           448)         check_if_key_loaded: bool = False,

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

src/ssh_agent_client/__init__.py           450)         """Request the SSH agent sign the payload with the key.
src/ssh_agent_client/__init__.py           451) 
src/ssh_agent_client/__init__.py           452)         Args:
src/ssh_agent_client/__init__.py           453)             key:
src/ssh_agent_client/__init__.py           454)                 The public SSH key to sign the payload with, in the same

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

src/ssh_agent_client/__init__.py           456)                 The corresponding private key must have previously been
src/ssh_agent_client/__init__.py           457)                 loaded into the agent to successfully issue a signature.
src/ssh_agent_client/__init__.py           458)             payload:
src/ssh_agent_client/__init__.py           459)                 A byte string of data to sign.
src/ssh_agent_client/__init__.py           460)             flags:
src/ssh_agent_client/__init__.py           461)                 Optional flags for the signing request.  Currently
src/ssh_agent_client/__init__.py           462)                 passed on as-is to the agent.  In real-world usage, this
src/ssh_agent_client/__init__.py           463)                 could be used, e.g., to request more modern hash
src/ssh_agent_client/__init__.py           464)                 algorithms when signing with RSA keys.  (No such
src/ssh_agent_client/__init__.py           465)                 real-world usage is currently implemented.)
src/ssh_agent_client/__init__.py           466)             check_if_key_loaded:

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

src/ssh_agent_client/__init__.py           468)                 corresponding key has been loaded into the agent.
src/ssh_agent_client/__init__.py           469) 
src/ssh_agent_client/__init__.py           470)         Returns:
src/ssh_agent_client/__init__.py           471)             The binary signature of the payload under the given key.
src/ssh_agent_client/__init__.py           472) 
src/ssh_agent_client/__init__.py           473)         Raises:
src/ssh_agent_client/__init__.py           474)             EOFError:
src/ssh_agent_client/__init__.py           475)                 The response from the SSH agent is truncated or missing.

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

src/ssh_agent_client/__init__.py           478)             TrailingDataError:

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

src/derivepassphrase/ssh_agent.py          480)             SSHAgentFailedError:

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

src/ssh_agent_client/__init__.py           482)             KeyError:

src/ssh_agent_client/__init__.py           483)                 `check_if_key_loaded` is true, and the `key` was not
src/ssh_agent_client/__init__.py           484)                 loaded into the agent.
src/ssh_agent_client/__init__.py           485) 
src/ssh_agent_client/__init__.py           486)         """

src/derivepassphrase/ssh_agent.py          487)         key = memoryview(key)
src/derivepassphrase/ssh_agent.py          488)         payload = memoryview(payload)

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

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

src/ssh_agent_client/__init__.py           494)         request_data = bytearray(self.string(key))
src/ssh_agent_client/__init__.py           495)         request_data.extend(self.string(payload))
src/ssh_agent_client/__init__.py           496)         request_data.extend(self.uint32(flags))

src/derivepassphrase/ssh_agent.py          497)         return bytes(
src/derivepassphrase/ssh_agent.py          498)             self.unstring(
src/derivepassphrase/ssh_agent.py          499)                 self.request(
src/derivepassphrase/ssh_agent.py          500)                     _types.SSH_AGENTC.SIGN_REQUEST.value,
src/derivepassphrase/ssh_agent.py          501)                     request_data,
src/derivepassphrase/ssh_agent.py          502)                     response_code=_types.SSH_AGENT.SIGN_RESPONSE,
src/derivepassphrase/ssh_agent.py          503)                 )

src/derivepassphrase/ssh_agent.py          504)             )