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) 
15) import click
16) import click.testing

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

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

25) 

26) class TestStaticFunctionality:
27) 
28)     @pytest.mark.parametrize(['public_key', 'public_key_data'],
29)                              [(val['public_key'], val['public_key_data'])
30)                               for val in tests.SUPPORTED_KEYS.values()])
31)     def test_100_key_decoding(self, public_key, public_key_data):
32)         keydata = base64.b64decode(public_key.split(None, 2)[1])
33)         assert (
34)             keydata == public_key_data
35)         ), "recorded public key data doesn't match"
36) 
37)     def test_200_constructor_no_running_agent(self, monkeypatch):
38)         monkeypatch.delenv('SSH_AUTH_SOCK', raising=False)
39)         sock = socket.socket(family=socket.AF_UNIX)

40)         with pytest.raises(KeyError,

41)                            match='SSH_AUTH_SOCK environment variable'):
42)             ssh_agent_client.SSHAgentClient(socket=sock)
43) 
44)     @pytest.mark.parametrize(['input', 'expected'], [
45)         (16777216, b'\x01\x00\x00\x00'),
46)     ])
47)     def test_210_uint32(self, input, expected):
48)         uint32 = ssh_agent_client.SSHAgentClient.uint32
49)         assert uint32(input) == expected
50) 
51)     @pytest.mark.parametrize(['input', 'expected'], [
52)         (b'ssh-rsa', b'\x00\x00\x00\x07ssh-rsa'),
53)         (b'ssh-ed25519', b'\x00\x00\x00\x0bssh-ed25519'),
54)         (
55)             ssh_agent_client.SSHAgentClient.string(b'ssh-ed25519'),
56)             b'\x00\x00\x00\x0f\x00\x00\x00\x0bssh-ed25519',
57)         ),
58)     ])
59)     def test_211_string(self, input, expected):
60)         string = ssh_agent_client.SSHAgentClient.string
61)         assert bytes(string(input)) == expected
62) 
63)     @pytest.mark.parametrize(['input', 'expected'], [
64)         (b'\x00\x00\x00\x07ssh-rsa', b'ssh-rsa'),
65)         (
66)             ssh_agent_client.SSHAgentClient.string(b'ssh-ed25519'),
67)             b'ssh-ed25519',
68)         ),
69)     ])
70)     def test_212_unstring(self, input, expected):
71)         unstring = ssh_agent_client.SSHAgentClient.unstring
72)         unstring_prefix = ssh_agent_client.SSHAgentClient.unstring_prefix
73)         assert bytes(unstring(input)) == expected
74)         assert tuple(
75)             bytes(x) for x in unstring_prefix(input)
76)         ) == (expected, b'')
77) 
78)     @pytest.mark.parametrize(['value', 'exc_type', 'exc_pattern'], [
79)         (10000000000000000, OverflowError, 'int too big to convert'),
80)         (-1, OverflowError, "can't convert negative int to unsigned"),
81)     ])
82)     def test_310_uint32_exceptions(self, value, exc_type, exc_pattern):
83)         uint32 = ssh_agent_client.SSHAgentClient.uint32
84)         with pytest.raises(exc_type, match=exc_pattern):
85)             uint32(value)
86) 
87)     @pytest.mark.parametrize(['input', 'exc_type', 'exc_pattern'], [
88)         ('some string', TypeError, 'invalid payload type'),
89)     ])
90)     def test_311_string_exceptions(self, input, exc_type, exc_pattern):
91)         string = ssh_agent_client.SSHAgentClient.string
92)         with pytest.raises(exc_type, match=exc_pattern):
93)             string(input)
94) 
95)     @pytest.mark.parametrize(
96)         ['input', 'exc_type', 'exc_pattern', 'has_trailer', 'parts'], [
97)             (b'ssh', ValueError, 'malformed SSH byte string', False, None),
98)             (
99)                 b'\x00\x00\x00\x08ssh-rsa',
100)                 ValueError, 'malformed SSH byte string',
101)                 False, None,
102)             ),
103)             (
104)                 b'\x00\x00\x00\x04XXX trailing text',
105)                 ValueError, 'malformed SSH byte string',
106)                 True, (b'XXX ', b'trailing text'),
107)             ),
108)     ])
109)     def test_312_unstring_exceptions(self, input, exc_type, exc_pattern,
110)                                      has_trailer, parts):
111)         unstring = ssh_agent_client.SSHAgentClient.unstring
112)         unstring_prefix = ssh_agent_client.SSHAgentClient.unstring_prefix
113)         with pytest.raises(exc_type, match=exc_pattern):
114)             unstring(input)
115)         if has_trailer:
116)             assert tuple(bytes(x) for x in unstring_prefix(input)) == parts
117)         else:
118)             with pytest.raises(exc_type, match=exc_pattern):
119)                 unstring_prefix(input)
120) 
121) @tests.skip_if_no_agent
122) class TestAgentInteraction:
123) 
124)     @pytest.mark.parametrize(['keytype', 'data_dict'],
125)                              list(tests.SUPPORTED_KEYS.items()))
126)     def test_200_sign_data_via_agent(self, keytype, data_dict):

127)         del keytype  # Unused.

128)         private_key = data_dict['private_key']
129)         try:

130)             _ = subprocess.run(['ssh-add', '-t', '30', '-q', '-'],
131)                                input=private_key, check=True,
132)                                capture_output=True)

133)         except subprocess.CalledProcessError as e:
134)             pytest.skip(
135)                 f"uploading test key: {e!r}, stdout={e.stdout!r}, "
136)                 f"stderr={e.stderr!r}"
137)             )
138)         else:
139)             try:
140)                 client = ssh_agent_client.SSHAgentClient()
141)             except OSError:  # pragma: no cover
142)                 pytest.skip('communication error with the SSH agent')
143)         with client:
144)             key_comment_pairs = {bytes(k): bytes(c)
145)                                  for k, c in client.list_keys()}
146)             public_key_data = data_dict['public_key_data']
147)             expected_signature = data_dict['expected_signature']
148)             derived_passphrase = data_dict['derived_passphrase']
149)             if public_key_data not in key_comment_pairs:  # pragma: no cover
150)                 pytest.skip('prerequisite SSH key not loaded')
151)             signature = bytes(client.sign(
152)                 payload=derivepassphrase.Vault._UUID, key=public_key_data))
153)             assert signature == expected_signature, 'SSH signature mismatch'
154)             signature2 = bytes(client.sign(
155)                 payload=derivepassphrase.Vault._UUID, key=public_key_data))
156)             assert signature2 == expected_signature, 'SSH signature mismatch'
157)             assert (
158)                 derivepassphrase.Vault.phrase_from_key(public_key_data) ==
159)                 derived_passphrase
160)             ), 'SSH signature mismatch'
161) 
162)     @pytest.mark.parametrize(['keytype', 'data_dict'],
163)                              list(tests.UNSUITABLE_KEYS.items()))
164)     def test_201_sign_data_via_agent_unsupported(self, keytype, data_dict):

165)         del keytype  # Unused.

166)         private_key = data_dict['private_key']
167)         try:

168)             _ = subprocess.run(['ssh-add', '-t', '30', '-q', '-'],
169)                                input=private_key, check=True,
170)                                capture_output=True)

171)         except subprocess.CalledProcessError as e:  # pragma: no cover
172)             pytest.skip(
173)                 f"uploading test key: {e!r}, stdout={e.stdout!r}, "
174)                 f"stderr={e.stderr!r}"
175)             )
176)         else:
177)             try:
178)                 client = ssh_agent_client.SSHAgentClient()
179)             except OSError:  # pragma: no cover
180)                 pytest.skip('communication error with the SSH agent')
181)         with client:
182)             key_comment_pairs = {bytes(k): bytes(c)
183)                                  for k, c in client.list_keys()}
184)             public_key_data = data_dict['public_key_data']

185)             _ = data_dict['expected_signature']

186)             if public_key_data not in key_comment_pairs:  # pragma: no cover
187)                 pytest.skip('prerequisite SSH key not loaded')
188)             signature = bytes(client.sign(
189)                 payload=derivepassphrase.Vault._UUID, key=public_key_data))
190)             signature2 = bytes(client.sign(
191)                 payload=derivepassphrase.Vault._UUID, key=public_key_data))
192)             assert signature != signature2, 'SSH signature repeatable?!'
193)             with pytest.raises(ValueError, match='unsuitable SSH key'):
194)                 derivepassphrase.Vault.phrase_from_key(public_key_data)
195) 
196)     @staticmethod
197)     def _params():
198)         for value in tests.SUPPORTED_KEYS.values():
199)             key = value['public_key_data']
200)             yield (key, False)
201)         singleton_key = tests.list_keys_singleton()[0].key
202)         for value in tests.SUPPORTED_KEYS.values():
203)             key = value['public_key_data']
204)             if key == singleton_key:
205)                 yield (key, True)
206) 
207)     @pytest.mark.parametrize(['key', 'single'], list(_params()))
208)     def test_210_ssh_key_selector(self, monkeypatch, key, single):
209)         def key_is_suitable(key: bytes):
210)             return key in {v['public_key_data']
211)                            for v in tests.SUPPORTED_KEYS.values()}
212)         if single:
213)             monkeypatch.setattr(ssh_agent_client.SSHAgentClient,
214)                                 'list_keys', tests.list_keys_singleton)
215)             keys = [pair.key for pair in tests.list_keys_singleton()
216)                     if key_is_suitable(pair.key)]
217)             index = '1'

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

219)         else:
220)             monkeypatch.setattr(ssh_agent_client.SSHAgentClient,
221)                                 'list_keys', tests.list_keys)
222)             keys = [pair.key for pair in tests.list_keys()
223)                     if key_is_suitable(pair.key)]
224)             index = str(1 + keys.index(key))
225)             n = len(keys)
226)             text = f'Your selection? (1-{n}, leave empty to abort): {index}\n'
227)         b64_key = base64.standard_b64encode(key).decode('ASCII')
228) 
229)         @click.command()
230)         def driver():
231)             key = derivepassphrase.cli._select_ssh_key()
232)             click.echo(base64.standard_b64encode(key).decode('ASCII'))
233) 
234)         runner = click.testing.CliRunner(mix_stderr=True)
235)         result = runner.invoke(driver, [],
236)                                input=('yes\n' if single else f'{index}\n'),
237)                                catch_exceptions=True)
238)         assert result.stdout.startswith('Suitable SSH keys:\n'), (
239)             'missing expected output'
240)         )
241)         assert text in result.stdout, 'missing expected output'
242)         assert (
243)             result.stdout.endswith(f'\n{b64_key}\n')
244)         ), 'missing expected output'
245)         assert result.exit_code == 0, 'driver program failed?!'
246) 
247)     del _params
248) 
249)     def test_300_constructor_bad_running_agent(self, monkeypatch):
250)         monkeypatch.setenv('SSH_AUTH_SOCK',
251)                            os.environ['SSH_AUTH_SOCK'] + '~')
252)         sock = socket.socket(family=socket.AF_UNIX)

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

254)             ssh_agent_client.SSHAgentClient(socket=sock)
255) 

256)     @pytest.mark.parametrize('response', [
257)         b'\x00\x00',
258)         b'\x00\x00\x00\x1f some bytes missing',

259)     ])
260)     def test_310_truncated_server_response(self, monkeypatch, response):
261)         client = ssh_agent_client.SSHAgentClient()
262)         response_stream = io.BytesIO(response)

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

265)                 return None
266)             def recv(self, *args: Any, **kwargs: Any) -> Any:
267)                 return response_stream.read(*args, **kwargs)
268)         pseudo_socket = PseudoSocket()
269)         monkeypatch.setattr(client, '_connection', pseudo_socket)
270)         with pytest.raises(EOFError):
271)             client.request(255, b'')
272) 
273)     @tests.skip_if_no_agent
274)     @pytest.mark.parametrize(
275)         ['response_code', 'response', 'exc_type', 'exc_pattern'],
276)         [
277)             (255, b'', RuntimeError, 'error return from SSH agent:'),
278)             (12, b'\x00\x00\x00\x01', EOFError, 'truncated response'),

279)             (
280)                 12,
281)                 b'\x00\x00\x00\x00abc',
282)                 ssh_agent_client.TrailingDataError,

283)                 'Overlong response',

284)             ),

285)         ]
286)     )
287)     def test_320_list_keys_error_responses(self, monkeypatch, response_code,
288)                                            response, exc_type, exc_pattern):
289)         client = ssh_agent_client.SSHAgentClient()
290)         monkeypatch.setattr(client, 'request',

291)                             lambda *a, **kw: (response_code, response))  # noqa: ARG005

292)         with pytest.raises(exc_type, match=exc_pattern):
293)             client.list_keys()
294) 
295)     @tests.skip_if_no_agent
296)     @pytest.mark.parametrize(
297)         ['key', 'check', 'response', 'exc_type', 'exc_pattern'],
298)         [
299)             (
300)                 b'invalid-key',
301)                 True,
302)                 (255, b''),
303)                 KeyError,
304)                 'target SSH key not loaded into agent',
305)             ),
306)             (
307)                 tests.SUPPORTED_KEYS['ed25519']['public_key_data'],
308)                 True,
309)                 (255, b''),
310)                 RuntimeError,
311)                 'signing data failed:',
312)             )
313)         ]
314)     )
315)     def test_330_sign_error_responses(self, monkeypatch, key, check,
316)                                       response, exc_type, exc_pattern):
317)         client = ssh_agent_client.SSHAgentClient()

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