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)     @pytest.mark.parametrize(
28)         ['public_key', 'public_key_data'],
29)         [
30)             (val['public_key'], val['public_key_data'])
31)             for val in tests.SUPPORTED_KEYS.values()
32)         ],
33)     )

34)     def test_100_key_decoding(self, public_key, public_key_data):
35)         keydata = base64.b64decode(public_key.split(None, 2)[1])
36)         assert (
37)             keydata == public_key_data
38)         ), "recorded public key data doesn't match"
39) 
40)     def test_200_constructor_no_running_agent(self, monkeypatch):
41)         monkeypatch.delenv('SSH_AUTH_SOCK', raising=False)
42)         sock = socket.socket(family=socket.AF_UNIX)

43)         with pytest.raises(
44)             KeyError, match='SSH_AUTH_SOCK environment variable'
45)         ):

46)             ssh_agent_client.SSHAgentClient(socket=sock)
47) 

48)     @pytest.mark.parametrize(
49)         ['input', 'expected'],
50)         [
51)             (16777216, b'\x01\x00\x00\x00'),
52)         ],
53)     )

54)     def test_210_uint32(self, input, expected):
55)         uint32 = ssh_agent_client.SSHAgentClient.uint32
56)         assert uint32(input) == expected
57) 

58)     @pytest.mark.parametrize(
59)         ['input', 'expected'],
60)         [
61)             (b'ssh-rsa', b'\x00\x00\x00\x07ssh-rsa'),
62)             (b'ssh-ed25519', b'\x00\x00\x00\x0bssh-ed25519'),
63)             (
64)                 ssh_agent_client.SSHAgentClient.string(b'ssh-ed25519'),
65)                 b'\x00\x00\x00\x0f\x00\x00\x00\x0bssh-ed25519',
66)             ),
67)         ],
68)     )

69)     def test_211_string(self, input, expected):
70)         string = ssh_agent_client.SSHAgentClient.string
71)         assert bytes(string(input)) == expected
72) 

73)     @pytest.mark.parametrize(
74)         ['input', 'expected'],
75)         [
76)             (b'\x00\x00\x00\x07ssh-rsa', b'ssh-rsa'),
77)             (
78)                 ssh_agent_client.SSHAgentClient.string(b'ssh-ed25519'),
79)                 b'ssh-ed25519',
80)             ),
81)         ],
82)     )

83)     def test_212_unstring(self, input, expected):
84)         unstring = ssh_agent_client.SSHAgentClient.unstring
85)         unstring_prefix = ssh_agent_client.SSHAgentClient.unstring_prefix
86)         assert bytes(unstring(input)) == expected

87)         assert tuple(bytes(x) for x in unstring_prefix(input)) == (
88)             expected,
89)             b'',
90)         )

91) 

92)     @pytest.mark.parametrize(
93)         ['value', 'exc_type', 'exc_pattern'],
94)         [
95)             (10000000000000000, OverflowError, 'int too big to convert'),
96)             (-1, OverflowError, "can't convert negative int to unsigned"),
97)         ],
98)     )

99)     def test_310_uint32_exceptions(self, value, exc_type, exc_pattern):
100)         uint32 = ssh_agent_client.SSHAgentClient.uint32
101)         with pytest.raises(exc_type, match=exc_pattern):
102)             uint32(value)
103) 

104)     @pytest.mark.parametrize(
105)         ['input', 'exc_type', 'exc_pattern'],
106)         [
107)             ('some string', TypeError, 'invalid payload type'),
108)         ],
109)     )

110)     def test_311_string_exceptions(self, input, exc_type, exc_pattern):
111)         string = ssh_agent_client.SSHAgentClient.string
112)         with pytest.raises(exc_type, match=exc_pattern):
113)             string(input)
114) 
115)     @pytest.mark.parametrize(

116)         ['input', 'exc_type', 'exc_pattern', 'has_trailer', 'parts'],
117)         [

118)             (b'ssh', ValueError, 'malformed SSH byte string', False, None),
119)             (
120)                 b'\x00\x00\x00\x08ssh-rsa',

121)                 ValueError,
122)                 'malformed SSH byte string',
123)                 False,
124)                 None,

125)             ),
126)             (
127)                 b'\x00\x00\x00\x04XXX trailing text',

128)                 ValueError,
129)                 'malformed SSH byte string',
130)                 True,
131)                 (b'XXX ', b'trailing text'),

132)             ),

133)         ],
134)     )
135)     def test_312_unstring_exceptions(
136)         self, input, exc_type, exc_pattern, has_trailer, parts
137)     ):

138)         unstring = ssh_agent_client.SSHAgentClient.unstring
139)         unstring_prefix = ssh_agent_client.SSHAgentClient.unstring_prefix
140)         with pytest.raises(exc_type, match=exc_pattern):
141)             unstring(input)
142)         if has_trailer:
143)             assert tuple(bytes(x) for x in unstring_prefix(input)) == parts
144)         else:
145)             with pytest.raises(exc_type, match=exc_pattern):
146)                 unstring_prefix(input)
147) 

148) 

149) @tests.skip_if_no_agent
150) class TestAgentInteraction:

151)     @pytest.mark.parametrize(
152)         ['keytype', 'data_dict'], list(tests.SUPPORTED_KEYS.items())
153)     )

154)     def test_200_sign_data_via_agent(self, keytype, data_dict):

155)         del keytype  # Unused.

156)         private_key = data_dict['private_key']
157)         try:

158)             _ = subprocess.run(
159)                 ['ssh-add', '-t', '30', '-q', '-'],
160)                 input=private_key,
161)                 check=True,
162)                 capture_output=True,
163)             )

164)         except subprocess.CalledProcessError as e:
165)             pytest.skip(

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

168)             )
169)         else:
170)             try:
171)                 client = ssh_agent_client.SSHAgentClient()
172)             except OSError:  # pragma: no cover
173)                 pytest.skip('communication error with the SSH agent')
174)         with client:

175)             key_comment_pairs = {
176)                 bytes(k): bytes(c) for k, c in client.list_keys()
177)             }

178)             public_key_data = data_dict['public_key_data']
179)             expected_signature = data_dict['expected_signature']
180)             derived_passphrase = data_dict['derived_passphrase']
181)             if public_key_data not in key_comment_pairs:  # pragma: no cover
182)                 pytest.skip('prerequisite SSH key not loaded')

183)             signature = bytes(
184)                 client.sign(
185)                     payload=derivepassphrase.Vault._UUID, key=public_key_data
186)                 )
187)             )

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

189)             signature2 = bytes(
190)                 client.sign(
191)                     payload=derivepassphrase.Vault._UUID, key=public_key_data
192)                 )
193)             )

194)             assert signature2 == expected_signature, 'SSH signature mismatch'
195)             assert (

196)                 derivepassphrase.Vault.phrase_from_key(public_key_data)
197)                 == derived_passphrase

198)             ), 'SSH signature mismatch'
199) 

200)     @pytest.mark.parametrize(
201)         ['keytype', 'data_dict'], list(tests.UNSUITABLE_KEYS.items())
202)     )

203)     def test_201_sign_data_via_agent_unsupported(self, keytype, data_dict):

204)         del keytype  # Unused.

205)         private_key = data_dict['private_key']
206)         try:

207)             _ = subprocess.run(
208)                 ['ssh-add', '-t', '30', '-q', '-'],
209)                 input=private_key,
210)                 check=True,
211)                 capture_output=True,
212)             )

213)         except subprocess.CalledProcessError as e:  # pragma: no cover
214)             pytest.skip(

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

217)             )
218)         else:
219)             try:
220)                 client = ssh_agent_client.SSHAgentClient()
221)             except OSError:  # pragma: no cover
222)                 pytest.skip('communication error with the SSH agent')
223)         with client:

224)             key_comment_pairs = {
225)                 bytes(k): bytes(c) for k, c in client.list_keys()
226)             }

227)             public_key_data = data_dict['public_key_data']

228)             _ = data_dict['expected_signature']

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

231)             signature = bytes(
232)                 client.sign(
233)                     payload=derivepassphrase.Vault._UUID, key=public_key_data
234)                 )
235)             )
236)             signature2 = bytes(
237)                 client.sign(
238)                     payload=derivepassphrase.Vault._UUID, key=public_key_data
239)                 )
240)             )

241)             assert signature != signature2, 'SSH signature repeatable?!'
242)             with pytest.raises(ValueError, match='unsuitable SSH key'):
243)                 derivepassphrase.Vault.phrase_from_key(public_key_data)
244) 
245)     @staticmethod
246)     def _params():
247)         for value in tests.SUPPORTED_KEYS.values():
248)             key = value['public_key_data']
249)             yield (key, False)
250)         singleton_key = tests.list_keys_singleton()[0].key
251)         for value in tests.SUPPORTED_KEYS.values():
252)             key = value['public_key_data']
253)             if key == singleton_key:
254)                 yield (key, True)
255) 
256)     @pytest.mark.parametrize(['key', 'single'], list(_params()))
257)     def test_210_ssh_key_selector(self, monkeypatch, key, single):
258)         def key_is_suitable(key: bytes):

259)             return key in {
260)                 v['public_key_data'] for v in tests.SUPPORTED_KEYS.values()
261)             }
262) 

263)         if single:

264)             monkeypatch.setattr(
265)                 ssh_agent_client.SSHAgentClient,
266)                 'list_keys',
267)                 tests.list_keys_singleton,
268)             )
269)             keys = [
270)                 pair.key
271)                 for pair in tests.list_keys_singleton()
272)                 if key_is_suitable(pair.key)
273)             ]

274)             index = '1'

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

276)         else:

277)             monkeypatch.setattr(
278)                 ssh_agent_client.SSHAgentClient, 'list_keys', tests.list_keys
279)             )
280)             keys = [
281)                 pair.key
282)                 for pair in tests.list_keys()
283)                 if key_is_suitable(pair.key)
284)             ]

285)             index = str(1 + keys.index(key))
286)             n = len(keys)
287)             text = f'Your selection? (1-{n}, leave empty to abort): {index}\n'
288)         b64_key = base64.standard_b64encode(key).decode('ASCII')
289) 
290)         @click.command()
291)         def driver():
292)             key = derivepassphrase.cli._select_ssh_key()
293)             click.echo(base64.standard_b64encode(key).decode('ASCII'))
294) 
295)         runner = click.testing.CliRunner(mix_stderr=True)

296)         result = runner.invoke(
297)             driver,
298)             [],
299)             input=('yes\n' if single else f'{index}\n'),
300)             catch_exceptions=True,

301)         )

302)         assert result.stdout.startswith(
303)             'Suitable SSH keys:\n'
304)         ), 'missing expected output'

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

306)         assert result.stdout.endswith(
307)             f'\n{b64_key}\n'

308)         ), 'missing expected output'
309)         assert result.exit_code == 0, 'driver program failed?!'
310) 
311)     del _params
312) 
313)     def test_300_constructor_bad_running_agent(self, monkeypatch):

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

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

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

317)             ssh_agent_client.SSHAgentClient(socket=sock)
318) 

319)     @pytest.mark.parametrize(
320)         'response',
321)         [
322)             b'\x00\x00',
323)             b'\x00\x00\x00\x1f some bytes missing',
324)         ],
325)     )

326)     def test_310_truncated_server_response(self, monkeypatch, response):
327)         client = ssh_agent_client.SSHAgentClient()
328)         response_stream = io.BytesIO(response)

329) 

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

332)                 return None

333) 

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

336) 

337)         pseudo_socket = PseudoSocket()
338)         monkeypatch.setattr(client, '_connection', pseudo_socket)
339)         with pytest.raises(EOFError):
340)             client.request(255, b'')
341) 
342)     @tests.skip_if_no_agent
343)     @pytest.mark.parametrize(
344)         ['response_code', 'response', 'exc_type', 'exc_pattern'],
345)         [
346)             (255, b'', RuntimeError, 'error return from SSH agent:'),
347)             (12, b'\x00\x00\x00\x01', EOFError, 'truncated response'),

348)             (
349)                 12,
350)                 b'\x00\x00\x00\x00abc',
351)                 ssh_agent_client.TrailingDataError,

352)                 'Overlong response',

353)             ),

354)         ],

355)     )

356)     def test_320_list_keys_error_responses(
357)         self, monkeypatch, response_code, response, exc_type, exc_pattern
358)     ):

359)         client = ssh_agent_client.SSHAgentClient()

360)         monkeypatch.setattr(
361)             client,
362)             'request',
363)             lambda *a, **kw: (response_code, response),  # noqa: ARG005
364)         )

365)         with pytest.raises(exc_type, match=exc_pattern):
366)             client.list_keys()
367) 
368)     @tests.skip_if_no_agent
369)     @pytest.mark.parametrize(
370)         ['key', 'check', 'response', 'exc_type', 'exc_pattern'],
371)         [
372)             (
373)                 b'invalid-key',
374)                 True,
375)                 (255, b''),
376)                 KeyError,
377)                 'target SSH key not loaded into agent',
378)             ),
379)             (
380)                 tests.SUPPORTED_KEYS['ed25519']['public_key_data'],
381)                 True,
382)                 (255, b''),
383)                 RuntimeError,
384)                 'signing data failed:',

385)             ),
386)         ],

387)     )

388)     def test_330_sign_error_responses(
389)         self, monkeypatch, key, check, response, exc_type, exc_pattern
390)     ):

391)         client = ssh_agent_client.SSHAgentClient()

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

394)         loaded_keys = [
395)             KeyCommentPair(v['public_key_data'], b'no comment')
396)             for v in tests.SUPPORTED_KEYS.values()
397)         ]