derivepassphrase.git

1) # SPDX-FileCopyrightText: 2024 Marco Ricci <software@the13thletter.info>
2) #
3) # SPDX-License-Identifier: MIT
4) 
5) from __future__ import annotations
6) 
7) import base64
8) import contextlib
9) import operator
10) import os
11) import shutil

12) import socket

13) import subprocess
14) from typing import TYPE_CHECKING, TypeVar
15) 

16) import hypothesis

17) import packaging.version
18) import pytest
19) 
20) import tests
21) from derivepassphrase import _types, ssh_agent
22) 
23) if TYPE_CHECKING:
24)     from collections.abc import Iterator
25) 
26) startup_ssh_auth_sock = os.environ.get('SSH_AUTH_SOCK', None)
27) 

28) # https://hypothesis.readthedocs.io/en/latest/settings.html#settings-profiles

29) hypothesis.settings.register_profile('ci', max_examples=1000)
30) hypothesis.settings.register_profile('dev', max_examples=10)

31) hypothesis.settings.register_profile(

32)     'debug', max_examples=10, verbosity=hypothesis.Verbosity.verbose

33) )
34) 

35) 
36) # https://docs.pytest.org/en/stable/explanation/fixtures.html#a-note-about-fixture-cleanup
37) # https://github.com/pytest-dev/pytest/issues/5243#issuecomment-491522595
38) @pytest.fixture(scope='session', autouse=True)
39) def term_handler() -> Iterator[None]:  # pragma: no cover
40)     try:
41)         import signal  # noqa: PLC0415
42) 
43)         sigint_handler = signal.getsignal(signal.SIGINT)
44)     except (ImportError, OSError):
45)         return
46)     else:
47)         orig_term = signal.signal(signal.SIGTERM, sigint_handler)
48)         yield
49)         signal.signal(signal.SIGTERM, orig_term)
50) 
51) 

52) @pytest.fixture(scope='session')
53) def skip_if_no_af_unix_support() -> None:  # pragma: no cover
54)     """Skip the test if Python does not support AF_UNIX.
55) 
56)     Implemented as a fixture instead of a mark because it has
57)     consequences for other fixtures, and because another "autouse"
58)     session fixture may want to force/simulate non-support of
59)     [`socket.AF_UNIX`][].
60) 
61)     """
62)     if not hasattr(socket, 'AF_UNIX'):
63)         pytest.skip('socket module does not support AF_UNIX')
64) 
65) 

66) def _spawn_pageant(  # pragma: no cover
67)     executable: str | None, env: dict[str, str]

68) ) -> subprocess.Popen[str] | None:

69)     """Spawn an isolated Pageant, if possible.
70) 
71)     We attempt to detect whether Pageant is usable, i.e. whether Pageant
72)     has output buffering problems when announcing its authentication

73)     socket.  This is the case for Pageant 0.81 and earlier.

74) 
75)     Args:
76)         executable:
77)             The path to the Pageant executable.
78)         env:
79)             The new environment for Pageant.  Should typically not
80)             include an SSH_AUTH_SOCK variable.
81) 
82)     Returns:

83)         The spawned Pageant subprocess.  If the executable is `None`, or
84)         if we detect that Pageant cannot be sensibly controlled as
85)         a subprocess, then return `None` directly.

86) 
87)         It is the caller's responsibility to clean up the spawned
88)         subprocess.
89) 
90)     """
91)     if executable is None:  # pragma: no cover
92)         return None
93) 
94)     # Apparently, Pageant 0.81 and lower running in debug mode does
95)     # not actively flush its output.  As a result, the first two
96)     # lines, which set the SSH_AUTH_SOCK and the SSH_AGENT_PID, only
97)     # print once the output buffer is flushed, whenever that is.
98)     #

99)     # This has been reported to the PuTTY developers.  It is fixed in
100)     # version 0.82, though the PuTTY developers consider this to be an
101)     # abuse of debug mode.  A new foreground mode (`--foreground`), also
102)     # introduced in 0.82, provides the desired behavior: no forking, and
103)     # immediately parsable instructions for SSH_AUTH_SOCK and
104)     # SSH_AGENT_PID.
105) 

106)     help_output = subprocess.run(
107)         ['pageant', '--help'],
108)         executable=executable,
109)         env=env,
110)         capture_output=True,
111)         text=True,
112)         check=False,
113)     ).stdout
114)     help_lines = help_output.splitlines(True)
115)     pageant_version_string = (
116)         help_lines[1].strip().removeprefix('Release ')
117)         if len(help_lines) >= 2
118)         else ''
119)     )

120)     v0_82 = packaging.version.Version('0.82')
121)     pageant_version = packaging.version.Version(pageant_version_string)

122) 

123)     if pageant_version < v0_82:  # pragma: no cover

124)         return None
125) 
126)     return subprocess.Popen(

127)         ['pageant', '--foreground', '-s'],

128)         executable=executable,
129)         stdin=subprocess.DEVNULL,
130)         stdout=subprocess.PIPE,
131)         shell=False,
132)         env=env,
133)         text=True,
134)         bufsize=1,

135)     )

136) 
137) 
138) def _spawn_openssh_agent(  # pragma: no cover
139)     executable: str | None, env: dict[str, str]

140) ) -> subprocess.Popen[str] | None:

141)     """Spawn an isolated OpenSSH agent, if possible.
142) 
143)     Args:
144)         executable:
145)             The path to the OpenSSH agent executable.
146)         env:
147)             The new environment for the OpenSSH agent.  Should typically
148)             not include an SSH_AUTH_SOCK variable.
149) 
150)     Returns:

151)         The spawned OpenSSH agent subprocess.  If the executable is
152)         `None`, then return `None` directly.

153) 
154)         It is the caller's responsibility to clean up the spawned
155)         subprocess.
156) 
157)     """
158)     if executable is None:
159)         return None
160)     return subprocess.Popen(
161)         ['ssh-agent', '-D', '-s'],
162)         executable=executable,
163)         stdin=subprocess.DEVNULL,
164)         stdout=subprocess.PIPE,
165)         shell=False,
166)         env=env,
167)         text=True,
168)         bufsize=1,

169)     )

170) 
171) 
172) def _spawn_system_agent(  # pragma: no cover
173)     executable: str | None, env: dict[str, str]
174) ) -> None:
175)     """Placeholder function. Does nothing."""
176) 
177) 
178) _spawn_handlers = [
179)     ('pageant', _spawn_pageant, tests.KnownSSHAgent.Pageant),
180)     ('ssh-agent', _spawn_openssh_agent, tests.KnownSSHAgent.OpenSSHAgent),
181)     ('(system)', _spawn_system_agent, tests.KnownSSHAgent.UNKNOWN),
182) ]
183) 
184) 
185) @pytest.fixture

186) def running_ssh_agent(  # pragma: no cover
187)     skip_if_no_af_unix_support: None,

188) ) -> Iterator[tests.RunningSSHAgentInfo]:

189)     """Ensure a running SSH agent, if possible, as a pytest fixture.
190) 
191)     Check for a running SSH agent, or spawn a new one if possible.  We
192)     know how to spawn OpenSSH's agent and PuTTY's Pageant.  If spawned
193)     this way, the agent does not persist beyond the test.
194) 
195)     This fixture can neither guarantee a particular running agent, nor
196)     can it guarantee a particular set of loaded keys.
197) 
198)     Yields:

199)         A 2-tuple `(ssh_auth_sock, agent_type)`, where `ssh_auth_sock`
200)         is the value of the `SSH_AUTH_SOCK` environment variable, to be
201)         used to connect to the running agent, and `agent_type` is the
202)         agent type.

203) 
204)     Raises:
205)         pytest.skip.Exception:
206)             If no agent is running or can be spawned, skip this test.
207) 
208)     """

209)     del skip_if_no_af_unix_support

210)     exit_stack = contextlib.ExitStack()
211)     Popen = TypeVar('Popen', bound=subprocess.Popen)
212) 
213)     @contextlib.contextmanager
214)     def terminate_on_exit(proc: Popen) -> Iterator[Popen]:
215)         try:
216)             yield proc
217)         finally:
218)             proc.terminate()
219)             proc.wait()
220) 
221)     with pytest.MonkeyPatch.context() as monkeypatch:
222)         # pytest's fixture system does not seem to guarantee that
223)         # environment variables are set up correctly if nested and
224)         # parametrized fixtures are used: it is possible that "outer"
225)         # parametrized fixtures are torn down only after other "outer"
226)         # fixtures of the same parameter set have run.  So set
227)         # SSH_AUTH_SOCK explicitly to the value saved at interpreter
228)         # startup.  This is then verified with *a lot* of further assert
229)         # statements.
230)         if startup_ssh_auth_sock:  # pragma: no cover
231)             monkeypatch.setenv('SSH_AUTH_SOCK', startup_ssh_auth_sock)
232)         else:  # pragma: no cover
233)             monkeypatch.delenv('SSH_AUTH_SOCK', raising=False)

234)         for exec_name, spawn_func, agent_type in _spawn_handlers:

235)             # Use match/case here once Python 3.9 becomes unsupported.
236)             if exec_name == '(system)':
237)                 assert (
238)                     os.environ.get('SSH_AUTH_SOCK', None)
239)                     == startup_ssh_auth_sock
240)                 ), 'SSH_AUTH_SOCK mismatch when checking for running agent'
241)                 try:
242)                     with ssh_agent.SSHAgentClient() as client:
243)                         client.list_keys()
244)                 except (KeyError, OSError):
245)                     continue

246)                 yield tests.RunningSSHAgentInfo(
247)                     os.environ['SSH_AUTH_SOCK'], agent_type
248)                 )

249)                 assert (
250)                     os.environ.get('SSH_AUTH_SOCK', None)
251)                     == startup_ssh_auth_sock
252)                 ), 'SSH_AUTH_SOCK mismatch after returning from running agent'
253)             else:
254)                 assert (
255)                     os.environ.get('SSH_AUTH_SOCK', None)
256)                     == startup_ssh_auth_sock
257)                 ), f'SSH_AUTH_SOCK mismatch when checking for spawnable {exec_name}'  # noqa: E501

258)                 proc = spawn_func(executable=shutil.which(exec_name), env={})
259)                 if proc is None:

260)                     continue
261)                 with exit_stack:
262)                     exit_stack.enter_context(terminate_on_exit(proc))

263)                     assert (
264)                         os.environ.get('SSH_AUTH_SOCK', None)
265)                         == startup_ssh_auth_sock

266)                     ), f'SSH_AUTH_SOCK mismatch after spawning {exec_name}'
267)                     assert proc.stdout is not None
268)                     ssh_auth_sock_line = proc.stdout.readline()

269)                     try:

270)                         ssh_auth_sock = tests.parse_sh_export_line(
271)                             ssh_auth_sock_line, env_name='SSH_AUTH_SOCK'
272)                         )
273)                     except ValueError:  # pragma: no cover

274)                         continue

275)                     pid_line = proc.stdout.readline()
276)                     if (
277)                         'pid' not in pid_line.lower()
278)                         and '_pid' not in pid_line.lower()
279)                     ):  # pragma: no cover
280)                         pytest.skip(f'Cannot parse agent output: {pid_line!r}')
281)                     monkeypatch2 = exit_stack.enter_context(
282)                         pytest.MonkeyPatch.context()
283)                     )
284)                     monkeypatch2.setenv('SSH_AUTH_SOCK', ssh_auth_sock)

285)                     yield tests.RunningSSHAgentInfo(ssh_auth_sock, agent_type)

286)                 assert (
287)                     os.environ.get('SSH_AUTH_SOCK', None)
288)                     == startup_ssh_auth_sock
289)                 ), f'SSH_AUTH_SOCK mismatch after tearing down {exec_name}'

290)             return
291)         pytest.skip('No SSH agent running or spawnable')
292) 
293) 
294) @pytest.fixture(params=_spawn_handlers, ids=operator.itemgetter(0))

295) def spawn_ssh_agent(

296)     request: pytest.FixtureRequest,

297)     skip_if_no_af_unix_support: None,

298) ) -> Iterator[tests.SpawnedSSHAgentInfo]:
299)     """Spawn an isolated SSH agent, if possible, as a pytest fixture.
300) 
301)     Spawn a new SSH agent isolated from other SSH use by other
302)     processes, if possible.  We know how to spawn OpenSSH's agent and
303)     PuTTY's Pageant, and the "(system)" fallback agent.
304) 
305)     Yields:

306)         A [named tuple][collections.namedtuple] containing information
307)         about the spawned agent, e.g. the software product, a client
308)         connected to the agent, and whether the agent is isolated from
309)         other clients.

310) 
311)     Raises:
312)         pytest.skip.Exception:
313)             If the agent cannot be spawned, skip this test.
314) 
315)     """

316)     del skip_if_no_af_unix_support

317)     agent_env = os.environ.copy()
318)     agent_env.pop('SSH_AUTH_SOCK', None)
319)     exit_stack = contextlib.ExitStack()
320)     Popen = TypeVar('Popen', bound=subprocess.Popen)
321) 
322)     @contextlib.contextmanager
323)     def terminate_on_exit(proc: Popen) -> Iterator[Popen]:
324)         try:
325)             yield proc
326)         finally:
327)             proc.terminate()
328)             proc.wait()
329) 
330)     with pytest.MonkeyPatch.context() as monkeypatch:
331)         # pytest's fixture system does not seem to guarantee that
332)         # environment variables are set up correctly if nested and
333)         # parametrized fixtures are used: it is possible that "outer"
334)         # parametrized fixtures are torn down only after other "outer"
335)         # fixtures of the same parameter set have run.  So set
336)         # SSH_AUTH_SOCK explicitly to the value saved at interpreter
337)         # startup.  This is then verified with *a lot* of further assert
338)         # statements.
339)         if startup_ssh_auth_sock:  # pragma: no cover
340)             monkeypatch.setenv('SSH_AUTH_SOCK', startup_ssh_auth_sock)
341)         else:  # pragma: no cover
342)             monkeypatch.delenv('SSH_AUTH_SOCK', raising=False)
343)         exec_name, spawn_func, agent_type = request.param

344)         # Use match/case here once Python 3.9 becomes unsupported.
345)         if exec_name == '(system)':
346)             assert (
347)                 os.environ.get('SSH_AUTH_SOCK', None) == startup_ssh_auth_sock
348)             ), 'SSH_AUTH_SOCK mismatch when checking for running agent'
349)             try:
350)                 client = ssh_agent.SSHAgentClient()
351)                 client.list_keys()
352)             except KeyError:  # pragma: no cover
353)                 pytest.skip('SSH agent is not running')
354)             except OSError as exc:  # pragma: no cover
355)                 pytest.skip(
356)                     f'Cannot talk to SSH agent: '
357)                     f'{exc.strerror}: {exc.filename!r}'
358)                 )
359)             with client:

360)                 assert (
361)                     os.environ.get('SSH_AUTH_SOCK', None)
362)                     == startup_ssh_auth_sock

363)                 ), 'SSH_AUTH_SOCK mismatch before setting up for running agent'
364)                 yield tests.SpawnedSSHAgentInfo(agent_type, client, False)
365)             assert (
366)                 os.environ.get('SSH_AUTH_SOCK', None) == startup_ssh_auth_sock
367)             ), 'SSH_AUTH_SOCK mismatch after returning from running agent'
368)             return
369) 
370)         else:
371)             assert (
372)                 os.environ.get('SSH_AUTH_SOCK', None) == startup_ssh_auth_sock
373)             ), f'SSH_AUTH_SOCK mismatch when checking for spawnable {exec_name}'  # noqa: E501

374)             proc = spawn_func(

375)                 executable=shutil.which(exec_name), env=agent_env
376)             )
377)             assert (
378)                 os.environ.get('SSH_AUTH_SOCK', None) == startup_ssh_auth_sock
379)             ), f'SSH_AUTH_SOCK mismatch after spawning {exec_name}'

380)             if proc is None:  # pragma: no cover

381)                 pytest.skip(f'Cannot spawn usable {exec_name}')
382)             with exit_stack:
383)                 exit_stack.enter_context(terminate_on_exit(proc))
384)                 assert proc.stdout is not None
385)                 ssh_auth_sock_line = proc.stdout.readline()

386)                 try:

387)                     ssh_auth_sock = tests.parse_sh_export_line(
388)                         ssh_auth_sock_line, env_name='SSH_AUTH_SOCK'
389)                     )
390)                 except ValueError:  # pragma: no cover

391)                     pytest.skip(

392)                         f'Cannot parse agent output: {ssh_auth_sock_line}'

393)                     )

394)                 pid_line = proc.stdout.readline()
395)                 if (
396)                     'pid' not in pid_line.lower()
397)                     and '_pid' not in pid_line.lower()
398)                 ):  # pragma: no cover
399)                     pytest.skip(f'Cannot parse agent output: {pid_line!r}')

400)                 assert (
401)                     os.environ.get('SSH_AUTH_SOCK', None)
402)                     == startup_ssh_auth_sock

403)                 ), f'SSH_AUTH_SOCK mismatch before spawning {exec_name} helper'
404)                 monkeypatch2 = exit_stack.enter_context(
405)                     pytest.MonkeyPatch.context()
406)                 )
407)                 monkeypatch2.setenv('SSH_AUTH_SOCK', ssh_auth_sock)
408)                 try:
409)                     client = ssh_agent.SSHAgentClient()
410)                 except OSError as exc:  # pragma: no cover
411)                     pytest.skip(
412)                         f'Cannot talk to SSH agent: '
413)                         f'{exc.strerror}: {exc.filename!r}'

414)                     )

415)                 exit_stack.enter_context(client)
416)                 yield tests.SpawnedSSHAgentInfo(agent_type, client, True)
417)             assert (
418)                 os.environ.get('SSH_AUTH_SOCK', None) == startup_ssh_auth_sock
419)             ), f'SSH_AUTH_SOCK mismatch after tearing down {exec_name}'
420)             return

421) 
422) 
423) @pytest.fixture
424) def ssh_agent_client_with_test_keys_loaded(  # noqa: C901
425)     spawn_ssh_agent: tests.SpawnedSSHAgentInfo,
426) ) -> Iterator[ssh_agent.SSHAgentClient]:

427)     """Provide an SSH agent with loaded test keys, as a pytest fixture.
428) 
429)     Use the `spawn_ssh_agent` fixture to acquire a usable SSH agent,
430)     upload the known test keys into the agent, and return a connected
431)     client.
432) 
433)     The agent may reject several of the test keys due to unsupported or
434)     obsolete key types.  Rejected keys will be silently ignored, unless
435)     all keys are rejected; then the test will be skipped.  You must not
436)     automatically assume any particular key is present in the agent.
437) 
438)     Yields:

439)         A [named tuple][collections.namedtuple] containing
440)         information about the spawned agent, e.g. the software
441)         product, a client connected to the agent, and whether the
442)         agent is isolated from other clients.

443) 
444)     Raises:
445)         OSError:
446)             There was a communication or a socket setup error with the
447)             agent.
448)         pytest.skip.Exception:
449)             If the agent is unusable or if it rejected all test keys,
450)             skip this test.
451) 
452)     Warning:
453)         It is the fixture's responsibility to clean up the SSH agent
454)         client after the test.  Closing the client's socket connection
455)         beforehand (e.g. by using the client as a context manager) may
456)         lead to exceptions being thrown upon fixture teardown.
457) 
458)     """

459)     agent_type, client, isolated = spawn_ssh_agent
460)     all_test_keys = {**tests.SUPPORTED_KEYS, **tests.UNSUITABLE_KEYS}
461)     successfully_loaded_keys: set[str] = set()
462) 
463)     def prepare_payload(
464)         payload: bytes | bytearray,
465)         *,
466)         isolated: bool = True,
467)         time_to_live: int = 30,
468)     ) -> tuple[_types.SSH_AGENTC, bytes]:
469)         return_code = (
470)             _types.SSH_AGENTC.ADD_IDENTITY
471)             if isolated
472)             else _types.SSH_AGENTC.ADD_ID_CONSTRAINED
473)         )
474)         lifetime_constraint = (
475)             b''
476)             if isolated
477)             else b'\x01' + ssh_agent.SSHAgentClient.uint32(time_to_live)
478)         )
479)         return (return_code, bytes(payload) + lifetime_constraint)
480) 
481)     try:
482)         for key_type, key_struct in all_test_keys.items():
483)             try:
484)                 private_key_data = key_struct['private_key_blob']
485)             except KeyError:  # pragma: no cover
486)                 continue
487)             request_code, payload = prepare_payload(
488)                 private_key_data, isolated=isolated, time_to_live=30
489)             )
490)             try:
491)                 try:
492)                     client.request(
493)                         request_code,
494)                         payload,
495)                         response_code=_types.SSH_AGENT.SUCCESS,
496)                     )
497)                 except ssh_agent.SSHAgentFailedError:  # pragma: no cover
498)                     # Pageant can fail to accept a key for two separate
499)                     # reasons:
500)                     #
501)                     # - Pageant refuses to accept a key it already holds

502)                     #   in memory.  Verify this by listing keys.