derivepassphrase.git

1) # SPDX-FileCopyrightText: 2024 Marco Ricci <m@the13thletter.info>
2) #
3) # SPDX-License-Identifier: MIT
4) 
5) """A bare-bones SSH agent client supporting signing and key listing."""
6) 
7) from __future__ import annotations
8) 
9) import collections
10) import enum
11) import errno
12) import os
13) import pathlib
14) import socket
15) 
16) from collections.abc import Sequence, MutableSequence
17) from typing import Any, NamedTuple, Self, TypeAlias
18) from ssh_agent_client.types import KeyCommentPair, SSH_AGENT, SSH_AGENTC
19) 
20) __all__ = ('SSHAgentClient',)

21) __author__ = 'Marco Ricci <m@the13thletter.info>'
22) __version__ = "0.1.0"

23) 
24) _socket = socket
25) 

26) class TrailingDataError(RuntimeError):
27)     """The result contained trailing data."""
28) 

29) class SSHAgentClient:
30)     """A bare-bones SSH agent client supporting signing and key listing.
31) 
32)     The main use case is requesting the agent sign some data, after
33)     checking that the necessary key is already loaded.
34) 
35)     The main fleshed out methods are `list_keys` and `sign`, which
36)     implement the `REQUEST_IDENTITIES` and `SIGN_REQUEST` requests.  If
37)     you *really* wanted to, there is enough infrastructure in place to
38)     issue other requests as defined in the protocol---it's merely the
39)     wrapper functions and the protocol numbers table that are missing.
40) 
41)     """

42)     _connection: socket.socket

43)     def __init__(
44)         self, /, *, socket: socket.socket | None = None, timeout: int = 125
45)     ) -> None:
46)         """Initialize the client.
47) 
48)         Args:
49)             socket:
50)                 An optional socket, connected to the SSH agent.  If not
51)                 given, we query the `SSH_AUTH_SOCK` environment
52)                 variable to auto-discover the correct socket address.
53)             timeout:
54)                 A connection timeout for the SSH agent.  Only used if
55)                 the socket is not yet connected.  The default value
56)                 gives ample time for agent connections forwarded via
57)                 SSH on high-latency networks (e.g. Tor).
58) 

59)         Raises:
60)             KeyError:
61)                 The `SSH_AUTH_SOCK` environment was not found.
62)             OSError:
63)                 There was an error setting up a socket connection to the
64)                 agent.
65) 

66)         """
67)         if socket is not None:

68)             self._connection = socket

69)         else:

70)             self._connection = _socket.socket(family=_socket.AF_UNIX)

71)         try:

72)             # Test whether the socket is connected.
73)             self._connection.getpeername()

74)         except OSError as e:

75)             # This condition is hard to test purposefully, so exclude
76)             # from coverage.
77)             if e.errno != errno.ENOTCONN:  # pragma: no cover

78)                 raise

79)             if 'SSH_AUTH_SOCK' not in os.environ:
80)                 raise KeyError('SSH_AUTH_SOCK environment variable')
81)             ssh_auth_sock = os.environ['SSH_AUTH_SOCK']

82)             self._connection.settimeout(timeout)

83)             self._connection.connect(ssh_auth_sock)

84) 
85)     def __enter__(self) -> Self:

86)         """Close socket connection upon context manager completion."""
87)         self._connection.__enter__()

88)         return self
89) 
90)     def __exit__(
91)         self, exc_type: Any, exc_val: Any, exc_tb: Any
92)     ) -> bool:

93)         """Close socket connection upon context manager completion."""

94)         return bool(

95)             self._connection.__exit__(

96)                 exc_type, exc_val, exc_tb)  # type: ignore[func-returns-value]
97)         )
98) 
99)     @staticmethod

100)     def uint32(num: int, /) -> bytes:
101)         r"""Format the number as a `uint32`, as per the agent protocol.
102) 
103)         Args:
104)             num: A number.
105) 
106)         Returns:
107)             The number in SSH agent wire protocol format, i.e. as
108)             a 32-bit big endian number.
109) 
110)         Raises:
111)             OverflowError:
112)                 As per [`int.to_bytes`][].
113) 
114)         Examples:
115)             >>> SSHAgentClient.uint32(16777216)
116)             b'\x01\x00\x00\x00'
117) 
118)         """

119)         return int.to_bytes(num, 4, 'big', signed=False)
120) 
121)     @classmethod
122)     def string(cls, payload: bytes | bytearray, /) -> bytes | bytearray:

123)         r"""Format the payload as an SSH string, as per the agent protocol.
124) 
125)         Args:
126)             payload: A byte string.
127) 
128)         Returns:
129)             The payload, framed in the SSH agent wire protocol format.
130) 
131)         Examples:
132)             >>> bytes(SSHAgentClient.string(b'ssh-rsa'))
133)             b'\x00\x00\x00\x07ssh-rsa'
134) 
135)         """

136)         try:
137)             ret = bytearray()
138)             ret.extend(cls.uint32(len(payload)))
139)             ret.extend(payload)
140)             return ret
141)         except Exception as e:
142)             raise TypeError('invalid payload type') from e

143) 
144)     @classmethod
145)     def unstring(cls, bytestring: bytes | bytearray, /) -> bytes | bytearray:

146)         r"""Unpack an SSH string.
147) 
148)         Args:
149)             bytestring: A framed byte string.
150) 
151)         Returns:
152)             The unframed byte string, i.e., the payload.
153) 
154)         Raises:
155)             ValueError:

156)                 The byte string is not an SSH string.

157) 
158)         Examples:
159)             >>> bytes(SSHAgentClient.unstring(b'\x00\x00\x00\x07ssh-rsa'))
160)             b'ssh-rsa'
161)             >>> bytes(SSHAgentClient.unstring(SSHAgentClient.string(b'ssh-ed25519')))
162)             b'ssh-ed25519'
163) 
164)         """

165)         n = len(bytestring)
166)         if n < 4:
167)             raise ValueError('malformed SSH byte string')
168)         elif n != 4 + int.from_bytes(bytestring[:4], 'big', signed=False):
169)             raise ValueError('malformed SSH byte string')
170)         return bytestring[4:]
171) 

172)     @classmethod
173)     def unstring_prefix(
174)         cls, bytestring: bytes | bytearray, /
175)     ) -> tuple[bytes | bytearray, bytes | bytearray]:
176)         r"""Unpack an SSH string at the beginning of the byte string.
177) 
178)         Args:
179)             bytestring:
180)                 A (general) byte string, beginning with a framed/SSH
181)                 byte string.
182) 
183)         Returns:
184)             A 2-tuple `(a, b)`, where `a` is the unframed byte
185)             string/payload at the beginning of input byte string, and
186)             `b` is the remainder of the input byte string.
187) 
188)         Raises:
189)             ValueError:
190)                 The byte string does not begin with an SSH string.
191) 
192)         Examples:
193)             >>> a, b = SSHAgentClient.unstring_prefix(
194)             ...     b'\x00\x00\x00\x07ssh-rsa____trailing data')
195)             >>> (bytes(a), bytes(b))
196)             (b'ssh-rsa', b'____trailing data')
197)             >>> a, b = SSHAgentClient.unstring_prefix(
198)             ...     SSHAgentClient.string(b'ssh-ed25519'))
199)             >>> (bytes(a), bytes(b))
200)             (b'ssh-ed25519', b'')
201) 
202)         """
203)         n = len(bytestring)
204)         if n < 4:
205)             raise ValueError('malformed SSH byte string')
206)         m = int.from_bytes(bytestring[:4], 'big', signed=False)
207)         if m + 4 > n:
208)             raise ValueError('malformed SSH byte string')
209)         return (bytestring[4:m + 4], bytestring[m + 4:])
210) 

211)     def request(
212)         self, code: int, payload: bytes | bytearray, /
213)     ) -> tuple[int, bytes | bytearray]:
214)         """Issue a generic request to the SSH agent.
215) 
216)         Args:
217)             code:
218)                 The request code.  See the SSH agent protocol for
219)                 protocol numbers to use here (and which protocol numbers
220)                 to expect in a response).
221)             payload:
222)                 A byte string containing the payload, or "contents", of
223)                 the request.  Request-specific.  `request` will add any
224)                 necessary wire framing around the request code and the
225)                 payload.
226) 
227)         Returns:
228)             A 2-tuple consisting of the response code and the payload,
229)             with all wire framing removed.
230) 
231)         Raises:
232)             EOFError:
233)                 The response from the SSH agent is truncated or missing.
234) 
235)         """
236)         request_message = bytearray([code])
237)         request_message.extend(payload)

238)         self._connection.sendall(self.string(request_message))
239)         chunk = self._connection.recv(4)

240)         if len(chunk) < 4:
241)             raise EOFError('cannot read response length')
242)         response_length = int.from_bytes(chunk, 'big', signed=False)

243)         response = self._connection.recv(response_length)

244)         if len(response) < response_length:
245)             raise EOFError('truncated response from SSH agent')
246)         return response[0], response[1:]
247) 
248)     def list_keys(self) -> Sequence[KeyCommentPair]:
249)         """Request a list of keys known to the SSH agent.
250) 
251)         Returns:
252)             A read-only sequence of key/comment pairs.
253) 
254)         Raises:
255)             EOFError:
256)                 The response from the SSH agent is truncated or missing.

257)             TrailingDataError:

258)                 The response from the SSH agent is too long.

259)             RuntimeError:
260)                 The agent failed to complete the request.

261) 
262)         """
263)         response_code, response = self.request(
264)             SSH_AGENTC.REQUEST_IDENTITIES.value, b'')
265)         if response_code != SSH_AGENT.IDENTITIES_ANSWER.value:
266)             raise RuntimeError(
267)                 f'error return from SSH agent: '
268)                 f'{response_code = }, {response = }'
269)             )
270)         response_stream = collections.deque(response)
271)         def shift(num: int) -> bytes:
272)             buf = collections.deque(bytes())
273)             for i in range(num):
274)                 try:
275)                     val = response_stream.popleft()
276)                 except IndexError:
277)                     response_stream.extendleft(reversed(buf))
278)                     raise EOFError(
279)                         'truncated response from SSH agent'
280)                     ) from None
281)                 buf.append(val)
282)             return bytes(buf)
283)         key_count = int.from_bytes(shift(4), 'big')
284)         keys: collections.deque[KeyCommentPair] = collections.deque()
285)         for i in range(key_count):
286)             key_size = int.from_bytes(shift(4), 'big')
287)             key = shift(key_size)
288)             comment_size = int.from_bytes(shift(4), 'big')
289)             comment = shift(comment_size)
290)             # Both `key` and `comment` are not wrapped as SSH strings.
291)             keys.append(KeyCommentPair(key, comment))
292)         if response_stream:

293)             raise TrailingDataError('overlong response from SSH agent')

294)         return keys
295) 
296)     def sign(
297)         self, /, key: bytes | bytearray, payload: bytes | bytearray,
298)         *, flags: int = 0, check_if_key_loaded: bool = False,
299)     ) -> bytes | bytearray:
300)         """Request the SSH agent sign the payload with the key.
301) 
302)         Args:
303)             key:
304)                 The public SSH key to sign the payload with, in the same
305)                 format as returned by, e.g., the `list_keys` method.
306)                 The corresponding private key must have previously been
307)                 loaded into the agent to successfully issue a signature.
308)             payload:
309)                 A byte string of data to sign.
310)             flags:
311)                 Optional flags for the signing request.  Currently
312)                 passed on as-is to the agent.  In real-world usage, this
313)                 could be used, e.g., to request more modern hash
314)                 algorithms when signing with RSA keys.  (No such
315)                 real-world usage is currently implemented.)
316)             check_if_key_loaded:
317)                 If true, check beforehand (via `list_keys`) if the
318)                 corresponding key has been loaded into the agent.
319) 
320)         Returns:
321)             The binary signature of the payload under the given key.
322) 
323)         Raises:
324)             EOFError:
325)                 The response from the SSH agent is truncated or missing.

326)             TrailingDataError: