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) import sys
15) import textwrap
16) from typing import TYPE_CHECKING, TypeVar
17) 

18) import hypothesis

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

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

32) hypothesis.settings.register_profile('ci', max_examples=1000)
33) hypothesis.settings.register_profile('dev', max_examples=10)

34) hypothesis.settings.register_profile(

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

36) )
37) 

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

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

69) def _spawn_pageant(  # pragma: no cover
70)     executable: str | None, env: dict[str, str]
71) ) -> tuple[subprocess.Popen[str], bool] | None:
72)     """Spawn an isolated Pageant, if possible.
73) 
74)     We attempt to detect whether Pageant is usable, i.e. whether Pageant
75)     has output buffering problems when announcing its authentication
76)     socket.
77) 
78)     Args:
79)         executable:
80)             The path to the Pageant executable.
81)         env:
82)             The new environment for Pageant.  Should typically not
83)             include an SSH_AUTH_SOCK variable.
84) 
85)     Returns:
86)         (tuple[subprocess.Popen, bool] | None):
87)         A 2-tuple `(proc, debug_output)`, where `proc` is the spawned
88)         Pageant subprocess, and `debug_output` indicates whether Pageant
89)         will continue to emit debug output (that needs to be actively
90)         read) or not.
91) 
92)         It is the caller's responsibility to clean up the spawned
93)         subprocess.
94) 
95)         If the executable is `None`, or if we detect that Pageant is too
96)         old to properly flush its output, which prevents readers from
97)         learning the SSH_AUTH_SOCK setting needed to connect to Pageant
98)         in the first place, then return `None` directly.
99) 
100)     """
101)     if executable is None:  # pragma: no cover
102)         return None
103) 
104)     pageant_features = {'flush': False, 'foreground': False}
105) 
106)     # Apparently, Pageant 0.81 and lower running in debug mode does
107)     # not actively flush its output.  As a result, the first two
108)     # lines, which set the SSH_AUTH_SOCK and the SSH_AGENT_PID, only
109)     # print once the output buffer is flushed, whenever that is.
110)     #
111)     # This has been reported to the PuTTY developers.
112)     #
113)     # For testing purposes, I currently build a version of Pageant with
114)     # output flushing fixed and with a `--foreground` option.  This is
115)     # detected here.
116)     help_output = subprocess.run(
117)         ['pageant', '--help'],
118)         executable=executable,
119)         env=env,
120)         capture_output=True,
121)         text=True,
122)         check=False,
123)     ).stdout
124)     help_lines = help_output.splitlines(True)
125)     pageant_version_string = (
126)         help_lines[1].strip().removeprefix('Release ')
127)         if len(help_lines) >= 2
128)         else ''
129)     )
130)     v0_81 = packaging.version.Version('0.81')
131)     if pageant_version_string not in {'', 'Unidentified build'}:
132)         # TODO(the-13th-letter): Once a fixed Pageant is released,
133)         # remove the check for build information in the version string.
134)         # https://github.com/the-13th-letter/derivepassphrase/issues/14
135)         pageant_version_string_numeric, local_segment_list = (
136)             pageant_version_string.split('+', 1)
137)             if '+' in pageant_version_string
138)             else (pageant_version_string, '')
139)         )
140)         local_segments = frozenset(local_segment_list.split('+'))
141)         pageant_version = packaging.version.Version(
142)             pageant_version_string_numeric
143)         )
144)         for key in pageant_features:
145)             pageant_features[key] = pageant_version > v0_81 or (
146)                 pageant_version == v0_81 and key in local_segments
147)             )
148) 
149)     if not pageant_features['flush']:  # pragma: no cover
150)         return None
151) 
152)     # Because Pageant's debug mode prints debugging information on
153)     # standard output, and because we yield control to a different
154)     # thread of execution, we cannot read-and-discard Pageant's output
155)     # here.  Instead, spawn a consumer process and connect it to
156)     # Pageant's standard output; see _spawn_data_sink.
157)     #
158)     # This will hopefully not be necessary with newer Pageants:
159)     # a feature request for a `--foreground` option that just avoids the
160)     # forking behavior has been submitted.
161) 
162)     return subprocess.Popen(
163)         [
164)             'pageant',
165)             '--foreground' if pageant_features['foreground'] else '--debug',
166)             '-s',
167)         ],
168)         executable=executable,
169)         stdin=subprocess.DEVNULL,
170)         stdout=subprocess.PIPE,
171)         shell=False,
172)         env=env,
173)         text=True,
174)         bufsize=1,
175)     ), not pageant_features['foreground']
176) 
177) 
178) def _spawn_openssh_agent(  # pragma: no cover
179)     executable: str | None, env: dict[str, str]
180) ) -> tuple[subprocess.Popen[str], Literal[False]] | None:
181)     """Spawn an isolated OpenSSH agent, if possible.
182) 
183)     We attempt to detect whether Pageant is usable, i.e. whether Pageant
184)     has output buffering problems when announcing its authentication
185)     socket.
186) 
187)     Args:
188)         executable:
189)             The path to the OpenSSH agent executable.
190)         env:
191)             The new environment for the OpenSSH agent.  Should typically
192)             not include an SSH_AUTH_SOCK variable.
193) 
194)     Returns:
195)         (tuple[subprocess.Popen, Literal[False]] | None):
196)         A 2-tuple `(proc, debug_output)`, where `proc` is the spawned
197)         OpenSSH agent subprocess, and `debug_output` indicates whether
198)         the OpenSSH agent will continue to emit debug output that needs
199)         to be actively read (which it doesn't, so this is always false).
200) 
201)         It is the caller's responsibility to clean up the spawned
202)         subprocess.
203) 
204)         If the executable is `None`, then return `None` directly.
205) 
206)     """
207)     if executable is None:
208)         return None
209)     return subprocess.Popen(
210)         ['ssh-agent', '-D', '-s'],
211)         executable=executable,
212)         stdin=subprocess.DEVNULL,
213)         stdout=subprocess.PIPE,
214)         shell=False,
215)         env=env,
216)         text=True,
217)         bufsize=1,
218)     ), False
219) 
220) 
221) def _spawn_system_agent(  # pragma: no cover
222)     executable: str | None, env: dict[str, str]
223) ) -> None:
224)     """Placeholder function. Does nothing."""
225) 
226) 
227) _spawn_handlers = [
228)     ('pageant', _spawn_pageant, tests.KnownSSHAgent.Pageant),
229)     ('ssh-agent', _spawn_openssh_agent, tests.KnownSSHAgent.OpenSSHAgent),
230)     ('(system)', _spawn_system_agent, tests.KnownSSHAgent.UNKNOWN),
231) ]
232) 
233) 
234) @pytest.fixture

235) def running_ssh_agent(  # pragma: no cover
236)     skip_if_no_af_unix_support: None,
237) ) -> Iterator[str]:

238)     """Ensure a running SSH agent, if possible, as a pytest fixture.
239) 
240)     Check for a running SSH agent, or spawn a new one if possible.  We
241)     know how to spawn OpenSSH's agent and PuTTY's Pageant.  If spawned
242)     this way, the agent does not persist beyond the test.
243) 
244)     This fixture can neither guarantee a particular running agent, nor
245)     can it guarantee a particular set of loaded keys.
246) 
247)     Yields:
248)         str:
249)             The value of the SSH_AUTH_SOCK environment variable, to be
250)             used to connect to the running agent.
251) 
252)     Raises:
253)         pytest.skip.Exception:
254)             If no agent is running or can be spawned, skip this test.
255) 
256)     """

257)     del skip_if_no_af_unix_support

258)     exit_stack = contextlib.ExitStack()
259)     Popen = TypeVar('Popen', bound=subprocess.Popen)
260) 
261)     @contextlib.contextmanager
262)     def terminate_on_exit(proc: Popen) -> Iterator[Popen]:
263)         try:
264)             yield proc
265)         finally:
266)             proc.terminate()
267)             proc.wait()
268) 
269)     with pytest.MonkeyPatch.context() as monkeypatch:
270)         # pytest's fixture system does not seem to guarantee that
271)         # environment variables are set up correctly if nested and
272)         # parametrized fixtures are used: it is possible that "outer"
273)         # parametrized fixtures are torn down only after other "outer"
274)         # fixtures of the same parameter set have run.  So set
275)         # SSH_AUTH_SOCK explicitly to the value saved at interpreter
276)         # startup.  This is then verified with *a lot* of further assert
277)         # statements.
278)         if startup_ssh_auth_sock:  # pragma: no cover
279)             monkeypatch.setenv('SSH_AUTH_SOCK', startup_ssh_auth_sock)
280)         else:  # pragma: no cover
281)             monkeypatch.delenv('SSH_AUTH_SOCK', raising=False)
282)         for exec_name, spawn_func, _ in _spawn_handlers:

283)             # Use match/case here once Python 3.9 becomes unsupported.
284)             if exec_name == '(system)':
285)                 assert (
286)                     os.environ.get('SSH_AUTH_SOCK', None)
287)                     == startup_ssh_auth_sock
288)                 ), 'SSH_AUTH_SOCK mismatch when checking for running agent'
289)                 try:
290)                     with ssh_agent.SSHAgentClient() as client:
291)                         client.list_keys()
292)                 except (KeyError, OSError):
293)                     continue
294)                 yield os.environ['SSH_AUTH_SOCK']
295)                 assert (
296)                     os.environ.get('SSH_AUTH_SOCK', None)
297)                     == startup_ssh_auth_sock
298)                 ), 'SSH_AUTH_SOCK mismatch after returning from running agent'
299)             else:
300)                 assert (
301)                     os.environ.get('SSH_AUTH_SOCK', None)
302)                     == startup_ssh_auth_sock
303)                 ), f'SSH_AUTH_SOCK mismatch when checking for spawnable {exec_name}'  # noqa: E501
304)                 spawn_data = spawn_func(  # type: ignore[operator]
305)                     executable=shutil.which(exec_name), env={}
306)                 )
307)                 if spawn_data is None:
308)                     continue
309)                 proc: subprocess.Popen[str]
310)                 emits_debug_output: bool
311)                 proc, emits_debug_output = spawn_data
312)                 with exit_stack:
313)                     exit_stack.enter_context(terminate_on_exit(proc))

314)                     assert (
315)                         os.environ.get('SSH_AUTH_SOCK', None)
316)                         == startup_ssh_auth_sock

317)                     ), f'SSH_AUTH_SOCK mismatch after spawning {exec_name}'
318)                     assert proc.stdout is not None
319)                     ssh_auth_sock_line = proc.stdout.readline()

320)                     try:

321)                         ssh_auth_sock = tests.parse_sh_export_line(
322)                             ssh_auth_sock_line, env_name='SSH_AUTH_SOCK'
323)                         )
324)                     except ValueError:  # pragma: no cover

325)                         continue

326)                     pid_line = proc.stdout.readline()
327)                     if (
328)                         'pid' not in pid_line.lower()
329)                         and '_pid' not in pid_line.lower()
330)                     ):  # pragma: no cover
331)                         pytest.skip(f'Cannot parse agent output: {pid_line!r}')
332)                     proc2 = _spawn_data_sink(
333)                         emits_debug_output=emits_debug_output, proc=proc

334)                     )

335)                     if proc2 is not None:  # pragma: no cover
336)                         exit_stack.enter_context(terminate_on_exit(proc2))

337)                     assert (
338)                         os.environ.get('SSH_AUTH_SOCK', None)
339)                         == startup_ssh_auth_sock

340)                     ), f'SSH_AUTH_SOCK mismatch after spawning {exec_name} helper'  # noqa: E501
341)                     monkeypatch2 = exit_stack.enter_context(
342)                         pytest.MonkeyPatch.context()
343)                     )
344)                     monkeypatch2.setenv('SSH_AUTH_SOCK', ssh_auth_sock)
345)                     yield ssh_auth_sock
346)                 assert (
347)                     os.environ.get('SSH_AUTH_SOCK', None)
348)                     == startup_ssh_auth_sock
349)                 ), f'SSH_AUTH_SOCK mismatch after tearing down {exec_name}'

350)             return
351)         pytest.skip('No SSH agent running or spawnable')
352) 
353) 
354) def _spawn_data_sink(  # pragma: no cover
355)     emits_debug_output: bool, *, proc: subprocess.Popen[str]
356) ) -> subprocess.Popen[str] | None:
357)     """Spawn a data sink to read and discard standard input.
358) 
359)     Necessary for certain SSH agents that emit copious debugging output.
360) 
361)     On UNIX, we can use `cat`, redirected to `/dev/null`.  Otherwise,
362)     the most robust thing to do is to spawn Python and repeatedly call
363)     `.read()` on `sys.stdin.buffer`.
364) 
365)     """
366)     if not emits_debug_output:
367)         return None
368)     if proc.stdout is None:
369)         return None
370)     sink_script = textwrap.dedent("""
371)     import sys
372)     while sys.stdin.buffer.read(4096):
373)         pass
374)     """)
375)     return subprocess.Popen(
376)         (
377)             ['cat']
378)             if os.name == 'posix'
379)             else [sys.executable or 'python3', '-c', sink_script]
380)         ),
381)         executable=sys.executable or None,
382)         stdin=proc.stdout.fileno(),
383)         stdout=subprocess.DEVNULL,
384)         shell=False,
385)         text=True,
386)     )
387) 
388) 
389) @pytest.fixture(params=_spawn_handlers, ids=operator.itemgetter(0))

390) def spawn_ssh_agent(  # noqa: C901

391)     request: pytest.FixtureRequest,

392)     skip_if_no_af_unix_support: None,

393) ) -> Iterator[tests.SpawnedSSHAgentInfo]:
394)     """Spawn an isolated SSH agent, if possible, as a pytest fixture.
395) 
396)     Spawn a new SSH agent isolated from other SSH use by other
397)     processes, if possible.  We know how to spawn OpenSSH's agent and
398)     PuTTY's Pageant, and the "(system)" fallback agent.
399) 
400)     Yields:
401)         (tests.SpawnedSSHAgentInfo):

402)             A [named tuple][collections.namedtuple] containing

403)             information about the spawned agent, e.g. the software
404)             product, a client connected to the agent, and whether the
405)             agent is isolated from other clients.
406) 
407)     Raises:
408)         pytest.skip.Exception:
409)             If the agent cannot be spawned, skip this test.
410) 
411)     """

412)     del skip_if_no_af_unix_support

413)     agent_env = os.environ.copy()
414)     agent_env.pop('SSH_AUTH_SOCK', None)
415)     exit_stack = contextlib.ExitStack()
416)     Popen = TypeVar('Popen', bound=subprocess.Popen)
417) 
418)     @contextlib.contextmanager
419)     def terminate_on_exit(proc: Popen) -> Iterator[Popen]:
420)         try:
421)             yield proc
422)         finally:
423)             proc.terminate()
424)             proc.wait()
425) 
426)     with pytest.MonkeyPatch.context() as monkeypatch:
427)         # pytest's fixture system does not seem to guarantee that
428)         # environment variables are set up correctly if nested and
429)         # parametrized fixtures are used: it is possible that "outer"
430)         # parametrized fixtures are torn down only after other "outer"
431)         # fixtures of the same parameter set have run.  So set
432)         # SSH_AUTH_SOCK explicitly to the value saved at interpreter
433)         # startup.  This is then verified with *a lot* of further assert
434)         # statements.
435)         if startup_ssh_auth_sock:  # pragma: no cover
436)             monkeypatch.setenv('SSH_AUTH_SOCK', startup_ssh_auth_sock)
437)         else:  # pragma: no cover
438)             monkeypatch.delenv('SSH_AUTH_SOCK', raising=False)
439)         exec_name, spawn_func, agent_type = request.param

440)         # Use match/case here once Python 3.9 becomes unsupported.
441)         if exec_name == '(system)':
442)             assert (
443)                 os.environ.get('SSH_AUTH_SOCK', None) == startup_ssh_auth_sock
444)             ), 'SSH_AUTH_SOCK mismatch when checking for running agent'
445)             try:
446)                 client = ssh_agent.SSHAgentClient()
447)                 client.list_keys()
448)             except KeyError:  # pragma: no cover
449)                 pytest.skip('SSH agent is not running')
450)             except OSError as exc:  # pragma: no cover
451)                 pytest.skip(
452)                     f'Cannot talk to SSH agent: '
453)                     f'{exc.strerror}: {exc.filename!r}'
454)                 )
455)             with client:

456)                 assert (
457)                     os.environ.get('SSH_AUTH_SOCK', None)
458)                     == startup_ssh_auth_sock

459)                 ), 'SSH_AUTH_SOCK mismatch before setting up for running agent'
460)                 yield tests.SpawnedSSHAgentInfo(agent_type, client, False)
461)             assert (
462)                 os.environ.get('SSH_AUTH_SOCK', None) == startup_ssh_auth_sock
463)             ), 'SSH_AUTH_SOCK mismatch after returning from running agent'
464)             return
465) 
466)         else:
467)             assert (
468)                 os.environ.get('SSH_AUTH_SOCK', None) == startup_ssh_auth_sock
469)             ), f'SSH_AUTH_SOCK mismatch when checking for spawnable {exec_name}'  # noqa: E501
470)             spawn_data = spawn_func(
471)                 executable=shutil.which(exec_name), env=agent_env
472)             )
473)             assert (
474)                 os.environ.get('SSH_AUTH_SOCK', None) == startup_ssh_auth_sock
475)             ), f'SSH_AUTH_SOCK mismatch after spawning {exec_name}'
476)             if spawn_data is None:  # pragma: no cover
477)                 pytest.skip(f'Cannot spawn usable {exec_name}')
478)             proc, emits_debug_output = spawn_data
479)             with exit_stack:
480)                 exit_stack.enter_context(terminate_on_exit(proc))
481)                 assert proc.stdout is not None
482)                 ssh_auth_sock_line = proc.stdout.readline()

483)                 try:

484)                     ssh_auth_sock = tests.parse_sh_export_line(
485)                         ssh_auth_sock_line, env_name='SSH_AUTH_SOCK'
486)                     )
487)                 except ValueError:  # pragma: no cover

488)                     pytest.skip(

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

490)                     )

491)                 pid_line = proc.stdout.readline()
492)                 if (
493)                     'pid' not in pid_line.lower()
494)                     and '_pid' not in pid_line.lower()
495)                 ):  # pragma: no cover
496)                     pytest.skip(f'Cannot parse agent output: {pid_line!r}')

497)                 assert (
498)                     os.environ.get('SSH_AUTH_SOCK', None)
499)                     == startup_ssh_auth_sock

500)                 ), f'SSH_AUTH_SOCK mismatch before spawning {exec_name} helper'
501)                 proc2 = _spawn_data_sink(
502)                     emits_debug_output=emits_debug_output, proc=proc

503)                 )

504)                 if proc2 is not None:  # pragma: no cover
505)                     exit_stack.enter_context(terminate_on_exit(proc2))

506)                 assert (
507)                     os.environ.get('SSH_AUTH_SOCK', None)
508)                     == startup_ssh_auth_sock

509)                 ), f'SSH_AUTH_SOCK mismatch after spawning {exec_name} helper'
510)                 monkeypatch2 = exit_stack.enter_context(
511)                     pytest.MonkeyPatch.context()
512)                 )
513)                 monkeypatch2.setenv('SSH_AUTH_SOCK', ssh_auth_sock)
514)                 try:
515)                     client = ssh_agent.SSHAgentClient()
516)                 except OSError as exc:  # pragma: no cover
517)                     pytest.skip(
518)                         f'Cannot talk to SSH agent: '
519)                         f'{exc.strerror}: {exc.filename!r}'

520)                     )

521)                 exit_stack.enter_context(client)
522)                 yield tests.SpawnedSSHAgentInfo(agent_type, client, True)
523)             assert (
524)                 os.environ.get('SSH_AUTH_SOCK', None) == startup_ssh_auth_sock
525)             ), f'SSH_AUTH_SOCK mismatch after tearing down {exec_name}'
526)             return

527) 
528) 
529) @pytest.fixture
530) def ssh_agent_client_with_test_keys_loaded(  # noqa: C901
531)     spawn_ssh_agent: tests.SpawnedSSHAgentInfo,
532) ) -> Iterator[ssh_agent.SSHAgentClient]:

533)     """Provide an SSH agent with loaded test keys, as a pytest fixture.
534) 
535)     Use the `spawn_ssh_agent` fixture to acquire a usable SSH agent,
536)     upload the known test keys into the agent, and return a connected
537)     client.
538) 
539)     The agent may reject several of the test keys due to unsupported or
540)     obsolete key types.  Rejected keys will be silently ignored, unless
541)     all keys are rejected; then the test will be skipped.  You must not
542)     automatically assume any particular key is present in the agent.
543) 
544)     Yields:
545)         (ssh_agent.SSHAgentClient):

546)             A [named tuple][collections.namedtuple] containing

547)             information about the spawned agent, e.g. the software
548)             product, a client connected to the agent, and whether the
549)             agent is isolated from other clients.
550) 
551)     Raises:
552)         OSError:
553)             There was a communication or a socket setup error with the
554)             agent.
555)         pytest.skip.Exception:
556)             If the agent is unusable or if it rejected all test keys,
557)             skip this test.
558) 
559)     Warning:
560)         It is the fixture's responsibility to clean up the SSH agent
561)         client after the test.  Closing the client's socket connection
562)         beforehand (e.g. by using the client as a context manager) may
563)         lead to exceptions being thrown upon fixture teardown.
564) 
565)     """