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

17) import hypothesis

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

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

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

33) hypothesis.settings.register_profile(

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

35) )
36) 

37) 
38) # https://docs.pytest.org/en/stable/explanation/fixtures.html#a-note-about-fixture-cleanup
39) # https://github.com/pytest-dev/pytest/issues/5243#issuecomment-491522595
40) @pytest.fixture(scope='session', autouse=True)
41) def term_handler() -> Iterator[None]:  # pragma: no cover
42)     try:
43)         import signal  # noqa: PLC0415
44) 
45)         sigint_handler = signal.getsignal(signal.SIGINT)
46)     except (ImportError, OSError):
47)         return
48)     else:
49)         orig_term = signal.signal(signal.SIGTERM, sigint_handler)
50)         yield
51)         signal.signal(signal.SIGTERM, orig_term)
52) 
53) 
54) def _spawn_pageant(  # pragma: no cover
55)     executable: str | None, env: dict[str, str]
56) ) -> tuple[subprocess.Popen[str], bool] | None:
57)     """Spawn an isolated Pageant, if possible.
58) 
59)     We attempt to detect whether Pageant is usable, i.e. whether Pageant
60)     has output buffering problems when announcing its authentication
61)     socket.
62) 
63)     Args:
64)         executable:
65)             The path to the Pageant executable.
66)         env:
67)             The new environment for Pageant.  Should typically not
68)             include an SSH_AUTH_SOCK variable.
69) 
70)     Returns:
71)         (tuple[subprocess.Popen, bool] | None):
72)         A 2-tuple `(proc, debug_output)`, where `proc` is the spawned
73)         Pageant subprocess, and `debug_output` indicates whether Pageant
74)         will continue to emit debug output (that needs to be actively
75)         read) or not.
76) 
77)         It is the caller's responsibility to clean up the spawned
78)         subprocess.
79) 
80)         If the executable is `None`, or if we detect that Pageant is too
81)         old to properly flush its output, which prevents readers from
82)         learning the SSH_AUTH_SOCK setting needed to connect to Pageant
83)         in the first place, then return `None` directly.
84) 
85)     """
86)     if executable is None:  # pragma: no cover
87)         return None
88) 
89)     pageant_features = {'flush': False, 'foreground': False}
90) 
91)     # Apparently, Pageant 0.81 and lower running in debug mode does
92)     # not actively flush its output.  As a result, the first two
93)     # lines, which set the SSH_AUTH_SOCK and the SSH_AGENT_PID, only
94)     # print once the output buffer is flushed, whenever that is.
95)     #
96)     # This has been reported to the PuTTY developers.
97)     #
98)     # For testing purposes, I currently build a version of Pageant with
99)     # output flushing fixed and with a `--foreground` option.  This is
100)     # detected here.
101)     help_output = subprocess.run(
102)         ['pageant', '--help'],
103)         executable=executable,
104)         env=env,
105)         capture_output=True,
106)         text=True,
107)         check=False,
108)     ).stdout
109)     help_lines = help_output.splitlines(True)
110)     pageant_version_string = (
111)         help_lines[1].strip().removeprefix('Release ')
112)         if len(help_lines) >= 2
113)         else ''
114)     )
115)     v0_81 = packaging.version.Version('0.81')
116)     if pageant_version_string not in {'', 'Unidentified build'}:
117)         # TODO(the-13th-letter): Once a fixed Pageant is released,
118)         # remove the check for build information in the version string.
119)         # https://github.com/the-13th-letter/derivepassphrase/issues/14
120)         pageant_version_string_numeric, local_segment_list = (
121)             pageant_version_string.split('+', 1)
122)             if '+' in pageant_version_string
123)             else (pageant_version_string, '')
124)         )
125)         local_segments = frozenset(local_segment_list.split('+'))
126)         pageant_version = packaging.version.Version(
127)             pageant_version_string_numeric
128)         )
129)         for key in pageant_features:
130)             pageant_features[key] = pageant_version > v0_81 or (
131)                 pageant_version == v0_81 and key in local_segments
132)             )
133) 
134)     if not pageant_features['flush']:  # pragma: no cover
135)         return None
136) 
137)     # Because Pageant's debug mode prints debugging information on
138)     # standard output, and because we yield control to a different
139)     # thread of execution, we cannot read-and-discard Pageant's output
140)     # here.  Instead, spawn a consumer process and connect it to
141)     # Pageant's standard output; see _spawn_data_sink.
142)     #
143)     # This will hopefully not be necessary with newer Pageants:
144)     # a feature request for a `--foreground` option that just avoids the
145)     # forking behavior has been submitted.
146) 
147)     return subprocess.Popen(
148)         [
149)             'pageant',
150)             '--foreground' if pageant_features['foreground'] else '--debug',
151)             '-s',
152)         ],
153)         executable=executable,
154)         stdin=subprocess.DEVNULL,
155)         stdout=subprocess.PIPE,
156)         shell=False,
157)         env=env,
158)         text=True,
159)         bufsize=1,
160)     ), not pageant_features['foreground']
161) 
162) 
163) def _spawn_openssh_agent(  # pragma: no cover
164)     executable: str | None, env: dict[str, str]
165) ) -> tuple[subprocess.Popen[str], Literal[False]] | None:
166)     """Spawn an isolated OpenSSH agent, if possible.
167) 
168)     We attempt to detect whether Pageant is usable, i.e. whether Pageant
169)     has output buffering problems when announcing its authentication
170)     socket.
171) 
172)     Args:
173)         executable:
174)             The path to the OpenSSH agent executable.
175)         env:
176)             The new environment for the OpenSSH agent.  Should typically
177)             not include an SSH_AUTH_SOCK variable.
178) 
179)     Returns:
180)         (tuple[subprocess.Popen, Literal[False]] | None):
181)         A 2-tuple `(proc, debug_output)`, where `proc` is the spawned
182)         OpenSSH agent subprocess, and `debug_output` indicates whether
183)         the OpenSSH agent will continue to emit debug output that needs
184)         to be actively read (which it doesn't, so this is always false).
185) 
186)         It is the caller's responsibility to clean up the spawned
187)         subprocess.
188) 
189)         If the executable is `None`, then return `None` directly.
190) 
191)     """
192)     if executable is None:
193)         return None
194)     return subprocess.Popen(
195)         ['ssh-agent', '-D', '-s'],
196)         executable=executable,
197)         stdin=subprocess.DEVNULL,
198)         stdout=subprocess.PIPE,
199)         shell=False,
200)         env=env,
201)         text=True,
202)         bufsize=1,
203)     ), False
204) 
205) 
206) def _spawn_system_agent(  # pragma: no cover
207)     executable: str | None, env: dict[str, str]
208) ) -> None:
209)     """Placeholder function. Does nothing."""
210) 
211) 
212) _spawn_handlers = [
213)     ('pageant', _spawn_pageant, tests.KnownSSHAgent.Pageant),
214)     ('ssh-agent', _spawn_openssh_agent, tests.KnownSSHAgent.OpenSSHAgent),
215)     ('(system)', _spawn_system_agent, tests.KnownSSHAgent.UNKNOWN),
216) ]
217) 
218) 
219) @pytest.fixture
220) def running_ssh_agent() -> Iterator[str]:  # pragma: no cover
221)     """Ensure a running SSH agent, if possible, as a pytest fixture.
222) 
223)     Check for a running SSH agent, or spawn a new one if possible.  We
224)     know how to spawn OpenSSH's agent and PuTTY's Pageant.  If spawned
225)     this way, the agent does not persist beyond the test.
226) 
227)     This fixture can neither guarantee a particular running agent, nor
228)     can it guarantee a particular set of loaded keys.
229) 
230)     Yields:
231)         str:
232)             The value of the SSH_AUTH_SOCK environment variable, to be
233)             used to connect to the running agent.
234) 
235)     Raises:
236)         pytest.skip.Exception:
237)             If no agent is running or can be spawned, skip this test.
238) 
239)     """
240)     exit_stack = contextlib.ExitStack()
241)     Popen = TypeVar('Popen', bound=subprocess.Popen)
242) 
243)     @contextlib.contextmanager
244)     def terminate_on_exit(proc: Popen) -> Iterator[Popen]:
245)         try:
246)             yield proc
247)         finally:
248)             proc.terminate()
249)             proc.wait()
250) 
251)     with pytest.MonkeyPatch.context() as monkeypatch:
252)         # pytest's fixture system does not seem to guarantee that
253)         # environment variables are set up correctly if nested and
254)         # parametrized fixtures are used: it is possible that "outer"
255)         # parametrized fixtures are torn down only after other "outer"
256)         # fixtures of the same parameter set have run.  So set
257)         # SSH_AUTH_SOCK explicitly to the value saved at interpreter
258)         # startup.  This is then verified with *a lot* of further assert
259)         # statements.
260)         if startup_ssh_auth_sock:  # pragma: no cover
261)             monkeypatch.setenv('SSH_AUTH_SOCK', startup_ssh_auth_sock)
262)         else:  # pragma: no cover
263)             monkeypatch.delenv('SSH_AUTH_SOCK', raising=False)
264)         for exec_name, spawn_func, _ in _spawn_handlers:

265)             # Use match/case here once Python 3.9 becomes unsupported.
266)             if exec_name == '(system)':
267)                 assert (
268)                     os.environ.get('SSH_AUTH_SOCK', None)
269)                     == startup_ssh_auth_sock
270)                 ), 'SSH_AUTH_SOCK mismatch when checking for running agent'
271)                 try:
272)                     with ssh_agent.SSHAgentClient() as client:
273)                         client.list_keys()
274)                 except (KeyError, OSError):
275)                     continue
276)                 yield os.environ['SSH_AUTH_SOCK']
277)                 assert (
278)                     os.environ.get('SSH_AUTH_SOCK', None)
279)                     == startup_ssh_auth_sock
280)                 ), 'SSH_AUTH_SOCK mismatch after returning from running agent'
281)             else:
282)                 assert (
283)                     os.environ.get('SSH_AUTH_SOCK', None)
284)                     == startup_ssh_auth_sock
285)                 ), f'SSH_AUTH_SOCK mismatch when checking for spawnable {exec_name}'  # noqa: E501
286)                 spawn_data = spawn_func(  # type: ignore[operator]
287)                     executable=shutil.which(exec_name), env={}
288)                 )
289)                 if spawn_data is None:
290)                     continue
291)                 proc: subprocess.Popen[str]
292)                 emits_debug_output: bool
293)                 proc, emits_debug_output = spawn_data
294)                 with exit_stack:
295)                     exit_stack.enter_context(terminate_on_exit(proc))

296)                     assert (
297)                         os.environ.get('SSH_AUTH_SOCK', None)
298)                         == startup_ssh_auth_sock

299)                     ), f'SSH_AUTH_SOCK mismatch after spawning {exec_name}'
300)                     assert proc.stdout is not None
301)                     ssh_auth_sock_line = proc.stdout.readline()

302)                     try:

303)                         ssh_auth_sock = tests.parse_sh_export_line(
304)                             ssh_auth_sock_line, env_name='SSH_AUTH_SOCK'
305)                         )
306)                     except ValueError:  # pragma: no cover

307)                         continue

308)                     pid_line = proc.stdout.readline()
309)                     if (
310)                         'pid' not in pid_line.lower()
311)                         and '_pid' not in pid_line.lower()
312)                     ):  # pragma: no cover
313)                         pytest.skip(f'Cannot parse agent output: {pid_line!r}')
314)                     proc2 = _spawn_data_sink(
315)                         emits_debug_output=emits_debug_output, proc=proc

316)                     )

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

319)                     assert (
320)                         os.environ.get('SSH_AUTH_SOCK', None)
321)                         == startup_ssh_auth_sock

322)                     ), f'SSH_AUTH_SOCK mismatch after spawning {exec_name} helper'  # noqa: E501
323)                     monkeypatch2 = exit_stack.enter_context(
324)                         pytest.MonkeyPatch.context()
325)                     )
326)                     monkeypatch2.setenv('SSH_AUTH_SOCK', ssh_auth_sock)
327)                     yield ssh_auth_sock
328)                 assert (
329)                     os.environ.get('SSH_AUTH_SOCK', None)
330)                     == startup_ssh_auth_sock
331)                 ), f'SSH_AUTH_SOCK mismatch after tearing down {exec_name}'

332)             return
333)         pytest.skip('No SSH agent running or spawnable')
334) 
335) 
336) def _spawn_data_sink(  # pragma: no cover
337)     emits_debug_output: bool, *, proc: subprocess.Popen[str]
338) ) -> subprocess.Popen[str] | None:
339)     """Spawn a data sink to read and discard standard input.
340) 
341)     Necessary for certain SSH agents that emit copious debugging output.
342) 
343)     On UNIX, we can use `cat`, redirected to `/dev/null`.  Otherwise,
344)     the most robust thing to do is to spawn Python and repeatedly call
345)     `.read()` on `sys.stdin.buffer`.
346) 
347)     """
348)     if not emits_debug_output:
349)         return None
350)     if proc.stdout is None:
351)         return None
352)     sink_script = textwrap.dedent("""
353)     import sys
354)     while sys.stdin.buffer.read(4096):
355)         pass
356)     """)
357)     return subprocess.Popen(
358)         (
359)             ['cat']
360)             if os.name == 'posix'
361)             else [sys.executable or 'python3', '-c', sink_script]
362)         ),
363)         executable=sys.executable or None,
364)         stdin=proc.stdout.fileno(),
365)         stdout=subprocess.DEVNULL,
366)         shell=False,
367)         text=True,
368)     )
369) 
370) 
371) @pytest.fixture(params=_spawn_handlers, ids=operator.itemgetter(0))

372) def spawn_ssh_agent(  # noqa: C901

373)     request: pytest.FixtureRequest,
374) ) -> Iterator[tests.SpawnedSSHAgentInfo]:
375)     """Spawn an isolated SSH agent, if possible, as a pytest fixture.
376) 
377)     Spawn a new SSH agent isolated from other SSH use by other
378)     processes, if possible.  We know how to spawn OpenSSH's agent and
379)     PuTTY's Pageant, and the "(system)" fallback agent.
380) 
381)     Yields:
382)         (tests.SpawnedSSHAgentInfo):

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

384)             information about the spawned agent, e.g. the software
385)             product, a client connected to the agent, and whether the
386)             agent is isolated from other clients.
387) 
388)     Raises:
389)         pytest.skip.Exception:
390)             If the agent cannot be spawned, skip this test.
391) 
392)     """
393)     agent_env = os.environ.copy()
394)     agent_env.pop('SSH_AUTH_SOCK', None)
395)     exit_stack = contextlib.ExitStack()
396)     Popen = TypeVar('Popen', bound=subprocess.Popen)
397) 
398)     @contextlib.contextmanager
399)     def terminate_on_exit(proc: Popen) -> Iterator[Popen]:
400)         try:
401)             yield proc
402)         finally:
403)             proc.terminate()
404)             proc.wait()
405) 
406)     with pytest.MonkeyPatch.context() as monkeypatch:
407)         # pytest's fixture system does not seem to guarantee that
408)         # environment variables are set up correctly if nested and
409)         # parametrized fixtures are used: it is possible that "outer"
410)         # parametrized fixtures are torn down only after other "outer"
411)         # fixtures of the same parameter set have run.  So set
412)         # SSH_AUTH_SOCK explicitly to the value saved at interpreter
413)         # startup.  This is then verified with *a lot* of further assert
414)         # statements.
415)         if startup_ssh_auth_sock:  # pragma: no cover
416)             monkeypatch.setenv('SSH_AUTH_SOCK', startup_ssh_auth_sock)
417)         else:  # pragma: no cover
418)             monkeypatch.delenv('SSH_AUTH_SOCK', raising=False)
419)         exec_name, spawn_func, agent_type = request.param

420)         # Use match/case here once Python 3.9 becomes unsupported.
421)         if exec_name == '(system)':
422)             assert (
423)                 os.environ.get('SSH_AUTH_SOCK', None) == startup_ssh_auth_sock
424)             ), 'SSH_AUTH_SOCK mismatch when checking for running agent'
425)             try:
426)                 client = ssh_agent.SSHAgentClient()
427)                 client.list_keys()
428)             except KeyError:  # pragma: no cover
429)                 pytest.skip('SSH agent is not running')
430)             except OSError as exc:  # pragma: no cover
431)                 pytest.skip(
432)                     f'Cannot talk to SSH agent: '
433)                     f'{exc.strerror}: {exc.filename!r}'
434)                 )
435)             with client:

436)                 assert (
437)                     os.environ.get('SSH_AUTH_SOCK', None)
438)                     == startup_ssh_auth_sock

439)                 ), 'SSH_AUTH_SOCK mismatch before setting up for running agent'
440)                 yield tests.SpawnedSSHAgentInfo(agent_type, client, False)
441)             assert (
442)                 os.environ.get('SSH_AUTH_SOCK', None) == startup_ssh_auth_sock
443)             ), 'SSH_AUTH_SOCK mismatch after returning from running agent'
444)             return
445) 
446)         else:
447)             assert (
448)                 os.environ.get('SSH_AUTH_SOCK', None) == startup_ssh_auth_sock
449)             ), f'SSH_AUTH_SOCK mismatch when checking for spawnable {exec_name}'  # noqa: E501
450)             spawn_data = spawn_func(
451)                 executable=shutil.which(exec_name), env=agent_env
452)             )
453)             assert (
454)                 os.environ.get('SSH_AUTH_SOCK', None) == startup_ssh_auth_sock
455)             ), f'SSH_AUTH_SOCK mismatch after spawning {exec_name}'
456)             if spawn_data is None:  # pragma: no cover
457)                 pytest.skip(f'Cannot spawn usable {exec_name}')
458)             proc, emits_debug_output = spawn_data
459)             with exit_stack:
460)                 exit_stack.enter_context(terminate_on_exit(proc))
461)                 assert proc.stdout is not None
462)                 ssh_auth_sock_line = proc.stdout.readline()

463)                 try:

464)                     ssh_auth_sock = tests.parse_sh_export_line(
465)                         ssh_auth_sock_line, env_name='SSH_AUTH_SOCK'
466)                     )
467)                 except ValueError:  # pragma: no cover

468)                     pytest.skip(

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

470)                     )

471)                 pid_line = proc.stdout.readline()
472)                 if (
473)                     'pid' not in pid_line.lower()
474)                     and '_pid' not in pid_line.lower()
475)                 ):  # pragma: no cover
476)                     pytest.skip(f'Cannot parse agent output: {pid_line!r}')

477)                 assert (
478)                     os.environ.get('SSH_AUTH_SOCK', None)
479)                     == startup_ssh_auth_sock

480)                 ), f'SSH_AUTH_SOCK mismatch before spawning {exec_name} helper'
481)                 proc2 = _spawn_data_sink(
482)                     emits_debug_output=emits_debug_output, proc=proc

483)                 )

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

486)                 assert (
487)                     os.environ.get('SSH_AUTH_SOCK', None)
488)                     == startup_ssh_auth_sock

489)                 ), f'SSH_AUTH_SOCK mismatch after spawning {exec_name} helper'
490)                 monkeypatch2 = exit_stack.enter_context(
491)                     pytest.MonkeyPatch.context()
492)                 )
493)                 monkeypatch2.setenv('SSH_AUTH_SOCK', ssh_auth_sock)
494)                 try:
495)                     client = ssh_agent.SSHAgentClient()
496)                 except OSError as exc:  # pragma: no cover
497)                     pytest.skip(
498)                         f'Cannot talk to SSH agent: '
499)                         f'{exc.strerror}: {exc.filename!r}'

500)                     )

501)                 exit_stack.enter_context(client)
502)                 yield tests.SpawnedSSHAgentInfo(agent_type, client, True)
503)             assert (
504)                 os.environ.get('SSH_AUTH_SOCK', None) == startup_ssh_auth_sock
505)             ), f'SSH_AUTH_SOCK mismatch after tearing down {exec_name}'
506)             return

507) 
508) 
509) @pytest.fixture
510) def ssh_agent_client_with_test_keys_loaded(  # noqa: C901
511)     spawn_ssh_agent: tests.SpawnedSSHAgentInfo,
512) ) -> Iterator[ssh_agent.SSHAgentClient]:

513)     """Provide an SSH agent with loaded test keys, as a pytest fixture.
514) 
515)     Use the `spawn_ssh_agent` fixture to acquire a usable SSH agent,
516)     upload the known test keys into the agent, and return a connected
517)     client.
518) 
519)     The agent may reject several of the test keys due to unsupported or
520)     obsolete key types.  Rejected keys will be silently ignored, unless
521)     all keys are rejected; then the test will be skipped.  You must not
522)     automatically assume any particular key is present in the agent.
523) 
524)     Yields:
525)         (ssh_agent.SSHAgentClient):

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

527)             information about the spawned agent, e.g. the software
528)             product, a client connected to the agent, and whether the
529)             agent is isolated from other clients.
530) 
531)     Raises:
532)         OSError:
533)             There was a communication or a socket setup error with the
534)             agent.
535)         pytest.skip.Exception:
536)             If the agent is unusable or if it rejected all test keys,
537)             skip this test.
538) 
539)     Warning:
540)         It is the fixture's responsibility to clean up the SSH agent
541)         client after the test.  Closing the client's socket connection
542)         beforehand (e.g. by using the client as a context manager) may
543)         lead to exceptions being thrown upon fixture teardown.
544) 
545)     """