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[tests.RunningSSHAgentInfo]:

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)         :
249)             A 2-tuple `(ssh_auth_sock, agent_type)`, where
250)             `ssh_auth_sock` is the value of the `SSH_AUTH_SOCK`
251)             environment variable, to be used to connect to the running
252)             agent, and `agent_type` is the agent type.

253) 
254)     Raises:
255)         pytest.skip.Exception:
256)             If no agent is running or can be spawned, skip this test.
257) 
258)     """

259)     del skip_if_no_af_unix_support

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

284)         for exec_name, spawn_func, agent_type in _spawn_handlers:

285)             # Use match/case here once Python 3.9 becomes unsupported.
286)             if exec_name == '(system)':
287)                 assert (
288)                     os.environ.get('SSH_AUTH_SOCK', None)
289)                     == startup_ssh_auth_sock
290)                 ), 'SSH_AUTH_SOCK mismatch when checking for running agent'
291)                 try:
292)                     with ssh_agent.SSHAgentClient() as client:
293)                         client.list_keys()
294)                 except (KeyError, OSError):
295)                     continue

296)                 yield tests.RunningSSHAgentInfo(
297)                     os.environ['SSH_AUTH_SOCK'], agent_type
298)                 )

299)                 assert (
300)                     os.environ.get('SSH_AUTH_SOCK', None)
301)                     == startup_ssh_auth_sock
302)                 ), 'SSH_AUTH_SOCK mismatch after returning from running agent'
303)             else:
304)                 assert (
305)                     os.environ.get('SSH_AUTH_SOCK', None)
306)                     == startup_ssh_auth_sock
307)                 ), f'SSH_AUTH_SOCK mismatch when checking for spawnable {exec_name}'  # noqa: E501
308)                 spawn_data = spawn_func(  # type: ignore[operator]
309)                     executable=shutil.which(exec_name), env={}
310)                 )
311)                 if spawn_data is None:
312)                     continue
313)                 proc: subprocess.Popen[str]
314)                 emits_debug_output: bool
315)                 proc, emits_debug_output = spawn_data
316)                 with exit_stack:
317)                     exit_stack.enter_context(terminate_on_exit(proc))

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

321)                     ), f'SSH_AUTH_SOCK mismatch after spawning {exec_name}'
322)                     assert proc.stdout is not None
323)                     ssh_auth_sock_line = proc.stdout.readline()

324)                     try:

325)                         ssh_auth_sock = tests.parse_sh_export_line(
326)                             ssh_auth_sock_line, env_name='SSH_AUTH_SOCK'
327)                         )
328)                     except ValueError:  # pragma: no cover

329)                         continue

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

338)                     )

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

341)                     assert (
342)                         os.environ.get('SSH_AUTH_SOCK', None)
343)                         == startup_ssh_auth_sock

344)                     ), f'SSH_AUTH_SOCK mismatch after spawning {exec_name} helper'  # noqa: E501
345)                     monkeypatch2 = exit_stack.enter_context(
346)                         pytest.MonkeyPatch.context()
347)                     )
348)                     monkeypatch2.setenv('SSH_AUTH_SOCK', ssh_auth_sock)

349)                     yield tests.RunningSSHAgentInfo(ssh_auth_sock, agent_type)

350)                 assert (
351)                     os.environ.get('SSH_AUTH_SOCK', None)
352)                     == startup_ssh_auth_sock
353)                 ), f'SSH_AUTH_SOCK mismatch after tearing down {exec_name}'

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

394) def spawn_ssh_agent(  # noqa: C901

395)     request: pytest.FixtureRequest,

396)     skip_if_no_af_unix_support: None,

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

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

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

416)     del skip_if_no_af_unix_support

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

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

460)                 assert (
461)                     os.environ.get('SSH_AUTH_SOCK', None)
462)                     == startup_ssh_auth_sock

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

487)                 try:

488)                     ssh_auth_sock = tests.parse_sh_export_line(
489)                         ssh_auth_sock_line, env_name='SSH_AUTH_SOCK'
490)                     )
491)                 except ValueError:  # pragma: no cover

492)                     pytest.skip(

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

494)                     )

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

501)                 assert (
502)                     os.environ.get('SSH_AUTH_SOCK', None)
503)                     == startup_ssh_auth_sock

504)                 ), f'SSH_AUTH_SOCK mismatch before spawning {exec_name} helper'
505)                 proc2 = _spawn_data_sink(
506)                     emits_debug_output=emits_debug_output, proc=proc

507)                 )

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

510)                 assert (
511)                     os.environ.get('SSH_AUTH_SOCK', None)
512)                     == startup_ssh_auth_sock

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

524)                     )

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

531) 
532) 
533) @pytest.fixture
534) def ssh_agent_client_with_test_keys_loaded(  # noqa: C901
535)     spawn_ssh_agent: tests.SpawnedSSHAgentInfo,
536) ) -> Iterator[ssh_agent.SSHAgentClient]:

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

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

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