derivepassphrase.git

1) # SPDX-FileCopyrightText: 2024 Marco Ricci <m@the13thletter.info>
2) #
3) # SPDX-License-Identifier: MIT
4) 
5) """Test OpenSSH key loading and signing."""
6) 
7) from __future__ import annotations
8) 
9) import base64
10) import io
11) import os
12) import socket
13) import subprocess

14) from typing import TYPE_CHECKING

15) 
16) import click
17) import click.testing

18) import pytest
19) from typing_extensions import Any
20) 

21) import derivepassphrase
22) import derivepassphrase.cli
23) import ssh_agent_client
24) import tests
25) 

26) if TYPE_CHECKING:
27)     from collections.abc import Iterator
28) 

29) 

30) class TestStaticFunctionality:

31)     @pytest.mark.parametrize(
32)         ['public_key', 'public_key_data'],
33)         [
34)             (val['public_key'], val['public_key_data'])
35)             for val in tests.SUPPORTED_KEYS.values()
36)         ],
37)     )

38)     def test_100_key_decoding(
39)         self, public_key: bytes, public_key_data: bytes
40)     ) -> None:

41)         keydata = base64.b64decode(public_key.split(None, 2)[1])
42)         assert (
43)             keydata == public_key_data
44)         ), "recorded public key data doesn't match"
45) 

46)     def test_200_constructor_no_running_agent(self, monkeypatch: Any) -> None:

47)         monkeypatch.delenv('SSH_AUTH_SOCK', raising=False)
48)         sock = socket.socket(family=socket.AF_UNIX)

49)         with pytest.raises(
50)             KeyError, match='SSH_AUTH_SOCK environment variable'
51)         ):

52)             ssh_agent_client.SSHAgentClient(socket=sock)
53) 

54)     @pytest.mark.parametrize(
55)         ['input', 'expected'],
56)         [
57)             (16777216, b'\x01\x00\x00\x00'),
58)         ],
59)     )

60)     def test_210_uint32(self, input: int, expected: bytes | bytearray) -> None:

61)         uint32 = ssh_agent_client.SSHAgentClient.uint32
62)         assert uint32(input) == expected
63) 

64)     @pytest.mark.parametrize(
65)         ['input', 'expected'],
66)         [
67)             (b'ssh-rsa', b'\x00\x00\x00\x07ssh-rsa'),
68)             (b'ssh-ed25519', b'\x00\x00\x00\x0bssh-ed25519'),
69)             (
70)                 ssh_agent_client.SSHAgentClient.string(b'ssh-ed25519'),
71)                 b'\x00\x00\x00\x0f\x00\x00\x00\x0bssh-ed25519',
72)             ),
73)         ],
74)     )

75)     def test_211_string(
76)         self, input: bytes | bytearray, expected: bytes | bytearray
77)     ) -> None:

78)         string = ssh_agent_client.SSHAgentClient.string
79)         assert bytes(string(input)) == expected
80) 

81)     @pytest.mark.parametrize(
82)         ['input', 'expected'],
83)         [
84)             (b'\x00\x00\x00\x07ssh-rsa', b'ssh-rsa'),
85)             (
86)                 ssh_agent_client.SSHAgentClient.string(b'ssh-ed25519'),
87)                 b'ssh-ed25519',
88)             ),
89)         ],
90)     )

91)     def test_212_unstring(
92)         self, input: bytes | bytearray, expected: bytes | bytearray
93)     ) -> None:

94)         unstring = ssh_agent_client.SSHAgentClient.unstring
95)         unstring_prefix = ssh_agent_client.SSHAgentClient.unstring_prefix
96)         assert bytes(unstring(input)) == expected

97)         assert tuple(bytes(x) for x in unstring_prefix(input)) == (
98)             expected,
99)             b'',
100)         )

101) 

102)     @pytest.mark.parametrize(
103)         ['value', 'exc_type', 'exc_pattern'],
104)         [
105)             (10000000000000000, OverflowError, 'int too big to convert'),
106)             (-1, OverflowError, "can't convert negative int to unsigned"),
107)         ],
108)     )

109)     def test_310_uint32_exceptions(
110)         self, value: int, exc_type: type[Exception], exc_pattern: str
111)     ) -> None:

112)         uint32 = ssh_agent_client.SSHAgentClient.uint32
113)         with pytest.raises(exc_type, match=exc_pattern):
114)             uint32(value)
115) 

116)     @pytest.mark.parametrize(
117)         ['input', 'exc_type', 'exc_pattern'],
118)         [
119)             ('some string', TypeError, 'invalid payload type'),
120)         ],
121)     )

122)     def test_311_string_exceptions(
123)         self, input: Any, exc_type: type[Exception], exc_pattern: str
124)     ) -> None:

125)         string = ssh_agent_client.SSHAgentClient.string
126)         with pytest.raises(exc_type, match=exc_pattern):
127)             string(input)
128) 
129)     @pytest.mark.parametrize(

130)         ['input', 'exc_type', 'exc_pattern', 'has_trailer', 'parts'],
131)         [

132)             (b'ssh', ValueError, 'malformed SSH byte string', False, None),
133)             (
134)                 b'\x00\x00\x00\x08ssh-rsa',

135)                 ValueError,
136)                 'malformed SSH byte string',
137)                 False,
138)                 None,

139)             ),
140)             (
141)                 b'\x00\x00\x00\x04XXX trailing text',

142)                 ValueError,
143)                 'malformed SSH byte string',
144)                 True,
145)                 (b'XXX ', b'trailing text'),

146)             ),

147)         ],
148)     )
149)     def test_312_unstring_exceptions(

150)         self,
151)         input: bytes | bytearray,
152)         exc_type: type[Exception],
153)         exc_pattern: str,
154)         has_trailer: bool,
155)         parts: tuple[bytes | bytearray, bytes | bytearray] | None,
156)     ) -> None:

157)         unstring = ssh_agent_client.SSHAgentClient.unstring
158)         unstring_prefix = ssh_agent_client.SSHAgentClient.unstring_prefix
159)         with pytest.raises(exc_type, match=exc_pattern):
160)             unstring(input)
161)         if has_trailer:
162)             assert tuple(bytes(x) for x in unstring_prefix(input)) == parts
163)         else:
164)             with pytest.raises(exc_type, match=exc_pattern):
165)                 unstring_prefix(input)
166) 

167) 

168) @tests.skip_if_no_agent
169) class TestAgentInteraction:

170)     @pytest.mark.parametrize(
171)         ['keytype', 'data_dict'], list(tests.SUPPORTED_KEYS.items())
172)     )

173)     def test_200_sign_data_via_agent(
174)         self, keytype: str, data_dict: tests.SSHTestKey
175)     ) -> None:

176)         del keytype  # Unused.

177)         private_key = data_dict['private_key']
178)         try:

179)             _ = subprocess.run(
180)                 ['ssh-add', '-t', '30', '-q', '-'],
181)                 input=private_key,
182)                 check=True,
183)                 capture_output=True,
184)             )

185)         except subprocess.CalledProcessError as e:
186)             pytest.skip(

187)                 f'uploading test key: {e!r}, stdout={e.stdout!r}, '
188)                 f'stderr={e.stderr!r}'

189)             )
190)         else:
191)             try:
192)                 client = ssh_agent_client.SSHAgentClient()
193)             except OSError:  # pragma: no cover
194)                 pytest.skip('communication error with the SSH agent')
195)         with client:

196)             key_comment_pairs = {
197)                 bytes(k): bytes(c) for k, c in client.list_keys()
198)             }

199)             public_key_data = data_dict['public_key_data']
200)             expected_signature = data_dict['expected_signature']
201)             derived_passphrase = data_dict['derived_passphrase']
202)             if public_key_data not in key_comment_pairs:  # pragma: no cover
203)                 pytest.skip('prerequisite SSH key not loaded')

204)             signature = bytes(
205)                 client.sign(
206)                     payload=derivepassphrase.Vault._UUID, key=public_key_data
207)                 )
208)             )

209)             assert signature == expected_signature, 'SSH signature mismatch'

210)             signature2 = bytes(
211)                 client.sign(
212)                     payload=derivepassphrase.Vault._UUID, key=public_key_data
213)                 )
214)             )

215)             assert signature2 == expected_signature, 'SSH signature mismatch'
216)             assert (

217)                 derivepassphrase.Vault.phrase_from_key(public_key_data)
218)                 == derived_passphrase

219)             ), 'SSH signature mismatch'
220) 

221)     @pytest.mark.parametrize(
222)         ['keytype', 'data_dict'], list(tests.UNSUITABLE_KEYS.items())
223)     )

224)     def test_201_sign_data_via_agent_unsupported(
225)         self, keytype: str, data_dict: tests.SSHTestKey
226)     ) -> None:

227)         del keytype  # Unused.

228)         private_key = data_dict['private_key']
229)         try:

230)             _ = subprocess.run(
231)                 ['ssh-add', '-t', '30', '-q', '-'],
232)                 input=private_key,
233)                 check=True,
234)                 capture_output=True,
235)             )

236)         except subprocess.CalledProcessError as e:  # pragma: no cover
237)             pytest.skip(

238)                 f'uploading test key: {e!r}, stdout={e.stdout!r}, '
239)                 f'stderr={e.stderr!r}'

240)             )
241)         else:
242)             try:
243)                 client = ssh_agent_client.SSHAgentClient()
244)             except OSError:  # pragma: no cover
245)                 pytest.skip('communication error with the SSH agent')
246)         with client:

247)             key_comment_pairs = {
248)                 bytes(k): bytes(c) for k, c in client.list_keys()
249)             }

250)             public_key_data = data_dict['public_key_data']

251)             _ = data_dict['expected_signature']

252)             if public_key_data not in key_comment_pairs:  # pragma: no cover
253)                 pytest.skip('prerequisite SSH key not loaded')

254)             signature = bytes(
255)                 client.sign(
256)                     payload=derivepassphrase.Vault._UUID, key=public_key_data
257)                 )
258)             )
259)             signature2 = bytes(
260)                 client.sign(
261)                     payload=derivepassphrase.Vault._UUID, key=public_key_data
262)                 )
263)             )

264)             assert signature != signature2, 'SSH signature repeatable?!'
265)             with pytest.raises(ValueError, match='unsuitable SSH key'):
266)                 derivepassphrase.Vault.phrase_from_key(public_key_data)
267) 
268)     @staticmethod

269)     def _params() -> Iterator[tuple[bytes, bool]]:

270)         for value in tests.SUPPORTED_KEYS.values():
271)             key = value['public_key_data']
272)             yield (key, False)
273)         singleton_key = tests.list_keys_singleton()[0].key
274)         for value in tests.SUPPORTED_KEYS.values():
275)             key = value['public_key_data']
276)             if key == singleton_key:
277)                 yield (key, True)
278) 
279)     @pytest.mark.parametrize(['key', 'single'], list(_params()))

280)     def test_210_ssh_key_selector(
281)         self, monkeypatch: Any, key: bytes, single: bool
282)     ) -> None:
283)         def key_is_suitable(key: bytes) -> bool:

284)             return key in {
285)                 v['public_key_data'] for v in tests.SUPPORTED_KEYS.values()
286)             }
287) 

288)         if single:

289)             monkeypatch.setattr(
290)                 ssh_agent_client.SSHAgentClient,
291)                 'list_keys',
292)                 tests.list_keys_singleton,
293)             )
294)             keys = [
295)                 pair.key
296)                 for pair in tests.list_keys_singleton()
297)                 if key_is_suitable(pair.key)
298)             ]

299)             index = '1'

300)             text = 'Use this key? yes\n'

301)         else:

302)             monkeypatch.setattr(
303)                 ssh_agent_client.SSHAgentClient, 'list_keys', tests.list_keys
304)             )
305)             keys = [
306)                 pair.key
307)                 for pair in tests.list_keys()
308)                 if key_is_suitable(pair.key)
309)             ]

310)             index = str(1 + keys.index(key))
311)             n = len(keys)
312)             text = f'Your selection? (1-{n}, leave empty to abort): {index}\n'
313)         b64_key = base64.standard_b64encode(key).decode('ASCII')
314) 
315)         @click.command()

316)         def driver() -> None:

317)             key = derivepassphrase.cli._select_ssh_key()
318)             click.echo(base64.standard_b64encode(key).decode('ASCII'))
319) 
320)         runner = click.testing.CliRunner(mix_stderr=True)

321)         result = runner.invoke(
322)             driver,
323)             [],
324)             input=('yes\n' if single else f'{index}\n'),
325)             catch_exceptions=True,

326)         )

327)         assert result.stdout.startswith(
328)             'Suitable SSH keys:\n'
329)         ), 'missing expected output'

330)         assert text in result.stdout, 'missing expected output'

331)         assert result.stdout.endswith(
332)             f'\n{b64_key}\n'

333)         ), 'missing expected output'
334)         assert result.exit_code == 0, 'driver program failed?!'
335) 
336)     del _params
337) 

338)     def test_300_constructor_bad_running_agent(self, monkeypatch: Any) -> None:

339)         monkeypatch.setenv('SSH_AUTH_SOCK', os.environ['SSH_AUTH_SOCK'] + '~')

340)         sock = socket.socket(family=socket.AF_UNIX)

341)         with pytest.raises(OSError):  # noqa: PT011

342)             ssh_agent_client.SSHAgentClient(socket=sock)
343) 

344)     @pytest.mark.parametrize(
345)         'response',
346)         [
347)             b'\x00\x00',
348)             b'\x00\x00\x00\x1f some bytes missing',
349)         ],
350)     )

351)     def test_310_truncated_server_response(
352)         self, monkeypatch: Any, response: bytes
353)     ) -> None:

354)         client = ssh_agent_client.SSHAgentClient()
355)         response_stream = io.BytesIO(response)

356) 

357)         class PseudoSocket:
358)             def sendall(self, *args: Any, **kwargs: Any) -> Any:  # noqa: ARG002

359)                 return None

360) 

361)             def recv(self, *args: Any, **kwargs: Any) -> Any:
362)                 return response_stream.read(*args, **kwargs)

363) 

364)         pseudo_socket = PseudoSocket()
365)         monkeypatch.setattr(client, '_connection', pseudo_socket)
366)         with pytest.raises(EOFError):
367)             client.request(255, b'')
368) 
369)     @tests.skip_if_no_agent
370)     @pytest.mark.parametrize(
371)         ['response_code', 'response', 'exc_type', 'exc_pattern'],
372)         [
373)             (255, b'', RuntimeError, 'error return from SSH agent:'),
374)             (12, b'\x00\x00\x00\x01', EOFError, 'truncated response'),

375)             (
376)                 12,
377)                 b'\x00\x00\x00\x00abc',
378)                 ssh_agent_client.TrailingDataError,

379)                 'Overlong response',

380)             ),

381)         ],

382)     )

383)     def test_320_list_keys_error_responses(

384)         self,
385)         monkeypatch: Any,
386)         response_code: int,
387)         response: bytes | bytearray,
388)         exc_type: type[Exception],
389)         exc_pattern: str,
390)     ) -> None:

391)         client = ssh_agent_client.SSHAgentClient()

392)         monkeypatch.setattr(
393)             client,
394)             'request',
395)             lambda *a, **kw: (response_code, response),  # noqa: ARG005
396)         )

397)         with pytest.raises(exc_type, match=exc_pattern):
398)             client.list_keys()
399) 
400)     @tests.skip_if_no_agent
401)     @pytest.mark.parametrize(
402)         ['key', 'check', 'response', 'exc_type', 'exc_pattern'],
403)         [
404)             (
405)                 b'invalid-key',
406)                 True,
407)                 (255, b''),
408)                 KeyError,
409)                 'target SSH key not loaded into agent',
410)             ),
411)             (
412)                 tests.SUPPORTED_KEYS['ed25519']['public_key_data'],
413)                 True,
414)                 (255, b''),
415)                 RuntimeError,
416)                 'signing data failed:',

417)             ),
418)         ],

419)     )

420)     def test_330_sign_error_responses(

421)         self,
422)         monkeypatch: Any,
423)         key: bytes | bytearray,
424)         check: bool,
425)         response: tuple[int, bytes | bytearray],
426)         exc_type: type[Exception],
427)         exc_pattern: str,
428)     ) -> None:

429)         client = ssh_agent_client.SSHAgentClient()

430)         monkeypatch.setattr(client, 'request', lambda a, b: response)  # noqa: ARG005
431)         KeyCommentPair = ssh_agent_client.types.KeyCommentPair  # noqa: N806

432)         loaded_keys = [
433)             KeyCommentPair(v['public_key_data'], b'no comment')
434)             for v in tests.SUPPORTED_KEYS.values()
435)         ]