Coverage for tests / data / callables.py: 100.000%
63 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-07-06 21:34 +0200
« prev ^ index » next coverage.py v7.13.1, created at 2026-07-06 21:34 +0200
1# SPDX-FileCopyrightText: 2026 Marco Ricci <software@the13thletter.info>
2#
3# SPDX-License-Identifier: Zlib
5"""General machinery for `derivepassphrase` testing data.
7This module contains machinery supplementary to the data from the
8`tests.data` module, such as common dummy functions, or simple parsers
9for (the restricted format in use by) the data. It also contains some
10actual test data that needs to be generated programmatically, or that
11e.g. includes a function from this module as an embedded callback
12parameter.
14"""
16from __future__ import annotations
18import os
19import shlex
20import stat
21from typing import TYPE_CHECKING
23import tests.data
24from derivepassphrase import _types, ssh_agent, vault
26__all__ = ()
28if TYPE_CHECKING:
29 import socket
30 from collections.abc import Generator
32 from typing_extensions import Any
35# Stubbed actions
36# ===============
39# SSH agent client
40# ----------------
43def list_keys(self: Any = None) -> list[_types.SSHKeyCommentPair]:
44 """Return a list of all SSH test keys, as key/comment pairs.
46 Intended as a monkeypatching replacement for
47 [`ssh_agent.SSHAgentClient.list_keys`][].
49 """
50 del self # Unused. 1hde
51 Pair = _types.SSHKeyCommentPair # noqa: N806 1hde
52 return [ 1hde
53 Pair(value.public_key_data, f"{key} test key".encode("ASCII"))
54 for key, value in tests.data.ALL_KEYS.items()
55 ]
58def sign(
59 self: Any, key: bytes | bytearray, message: bytes | bytearray
60) -> bytes:
61 """Return the signature of `message` under `key`.
63 Can only handle keys in [`SUPPORTED_KEYS`][], and only the vault
64 UUID as the message.
66 Intended as a monkeypatching replacement for
67 [`ssh_agent.SSHAgentClient.sign`][].
69 """
70 del self # Unused. 1dg
71 assert message == vault.Vault.UUID 1dg
72 for value in tests.data.SUPPORTED_KEYS.values(): 1dg
73 if value.public_key_data == key: # pragma: no branch 1dg
74 return value.expected_signatures[ 1dg
75 tests.data.SSHTestKeyDeterministicSignatureClass.SPEC
76 ].signature
77 raise AssertionError
80def list_keys_singleton(self: Any = None) -> list[_types.SSHKeyCommentPair]:
81 """Return a singleton list of the first supported SSH test key.
83 The key is returned as a key/comment pair.
85 Intended as a monkeypatching replacement for
86 [`ssh_agent.SSHAgentClient.list_keys`][].
88 """
89 del self # Unused. 1ae
90 Pair = _types.SSHKeyCommentPair # noqa: N806 1ae
91 list1 = [ 1ae
92 Pair(value.public_key_data, f"{key} test key".encode("ASCII"))
93 for key, value in tests.data.SUPPORTED_KEYS.items()
94 ]
95 return list1[:1] 1ae
98# CLI machinery
99# -------------
102def suitable_ssh_keys(
103 conn: Any,
104) -> Generator[_types.SSHKeyCommentPair, None, None]:
105 """Return a two-item list of SSH test keys (key/comment pairs).
107 Intended as a monkeypatching replacement for
108 `cli_machinery.get_suitable_ssh_keys` to better script and test the
109 interactive key selection. When used this way, `derivepassphrase`
110 believes that only those two keys are loaded and suitable.
112 """
113 del conn # Unused. 1fijk
114 Pair = _types.SSHKeyCommentPair # noqa: N806 1fijk
115 yield from [ 1fijk
116 Pair(tests.data.DUMMY_KEY1, b"no comment"),
117 Pair(tests.data.DUMMY_KEY2, b"a comment"),
118 ]
121def auto_prompt(*args: Any, **kwargs: Any) -> str:
122 """Return [`DUMMY_PASSPHRASE`][].
124 Intended as a monkeypatching replacement for
125 `cli.prompt_for_passphrase` to better script and test the
126 interactive passphrase queries.
128 """
129 del args, kwargs # Unused. 1nopqr
130 return tests.data.DUMMY_PASSPHRASE 1nopqr
133# `vault` module
134# --------------
137def phrase_from_key(
138 key: bytes,
139 /,
140 *,
141 conn: ssh_agent.SSHAgentClient | socket.socket | None = None,
142) -> bytes:
143 """Return the "equivalent master passphrase" for key.
145 Only works for key [`DUMMY_KEY1`][].
147 Intended as a monkeypatching replacement for
148 [`vault.Vault.phrase_from_key`][], bypassing communication with an
149 actual SSH agent.
151 """
152 del conn 1flm
153 if key == tests.data.DUMMY_KEY1: # pragma: no branch 1flm
154 return tests.data.DUMMY_PHRASE_FROM_KEY1 1flm
155 raise KeyError(key) # pragma: no cover
158# SSH agent socket provider data (with callables)
159# ===============================================
162def provider_entry_provider() -> _types.SSHAgentSocket: # pragma: no cover
163 """A pseudo provider for a [`_types.SSHAgentSocketProviderEntry`][]."""
164 msg = "We are not supposed to be called!"
165 raise AssertionError(msg)
168provider_entry1 = _types.SSHAgentSocketProviderEntry(
169 provider_entry_provider, "entry1", ("entry1a", "entry1b", "entry1c")
170)
171"""A sample [`_types.SSHAgentSocketProviderEntry`][]."""
173provider_entry2 = _types.SSHAgentSocketProviderEntry(
174 provider_entry_provider, "entry2", ("entry2d", "entry2e")
175)
178# SSH agent output parsing
179# ========================
182def parse_sh_export_line(line: str, *, env_name: str) -> str:
183 """Parse the output of typical SSH agents' SSH_AUTH_SOCK lines.
185 Intentionally parses only a small subset of sh(1) syntax which works
186 with current OpenSSH and PuTTY output. We require exactly one
187 variable setting, and one export instruction, both on the same line,
188 and perhaps combined into one statement. Terminating semicolons
189 after each command are ignored.
191 Args:
192 line:
193 A line of sh(1) script to parse.
194 env_name:
195 The name of the environment variable to expect.
197 Returns:
198 The parsed environment variable value.
200 Raises:
201 ValueError:
202 Cannot parse the sh script. Perhaps it is too complex,
203 perhaps it is malformed.
205 """
206 line = line.rstrip("\r\n") 1ab
207 shlex_parser = shlex.shlex( 1ab
208 instream=line, posix=True, punctuation_chars=True
209 )
210 shlex_parser.whitespace = " \t" 1ab
211 tokens = list(shlex_parser) 1ab
212 orig_tokens = tokens.copy() 1ab
213 if tokens[-1] == ";": 1ab
214 tokens.pop() 1ab
215 if tokens[-3:] == [";", "export", env_name]: 1ab
216 tokens[-3:] = [] 1ab
217 tokens[:0] = ["export"] 1ab
218 if not ( 1ab
219 len(tokens) == 2
220 and tokens[0] == "export"
221 and tokens[1].startswith(f"{env_name}=")
222 ):
223 msg = f"Cannot parse sh line: {orig_tokens!r} -> {tokens!r}" 1b
224 raise ValueError(msg) 1b
225 return tokens[1].split("=", 1)[1] 1ab
228# General file system actions
229# ===========================
232def make_file_readonly(
233 pathname: str | bytes | os.PathLike[str],
234 /,
235 *,
236 try_race_free_implementation: bool = True,
237) -> None:
238 """Mark a file as read-only.
240 On POSIX, this entails removing the write permission bits for user,
241 group and other, and ensuring the read permission bit for user is
242 set.
244 Unfortunately, The Annoying OS (a.k.a. Microsoft Windows) has its
245 own rules: Set exactly(?) the read permission bit for user to make
246 the file read-only, and set exactly(?) the write permission bit for
247 user to make the file read/write; all other permission bit settings
248 are ignored.
250 The cross-platform procedure therefore is:
252 1. Call `os.stat` on the file, noting the permission bits.
253 2. Calculate the new permission bits POSIX-style.
254 3. Call `os.chmod` with permission bit `stat.S_IREAD`.
255 4. Call `os.chmod` with the correct POSIX-style permissions.
257 If the platform supports it, we use a file descriptor instead of
258 a path name. Otherwise, we use the same path name multiple times,
259 and are susceptible to race conditions.
261 """
262 fname: int | str | bytes | os.PathLike
263 if try_race_free_implementation and {os.stat, os.chmod} <= os.supports_fd: 1c
264 # The Annoying OS (v11 at least) supports fstat and fchmod, but
265 # does not support changing the file mode on file descriptors
266 # for read-only files.
267 fname = os.open( 1c
268 pathname,
269 os.O_RDWR
270 | getattr(os, "O_CLOEXEC", 0)
271 | getattr(os, "O_NOCTTY", 0),
272 )
273 else:
274 fname = pathname 1c
275 try: 1c
276 orig_mode = os.stat(fname).st_mode # noqa: PTH116 1c
277 new_mode = ( 1c
278 orig_mode & ~stat.S_IWUSR & ~stat.S_IWGRP & ~stat.S_IWOTH
279 | stat.S_IREAD
280 )
281 os.chmod(fname, stat.S_IREAD) # noqa: PTH101 1c
282 os.chmod(fname, new_mode) # noqa: PTH101 1c
283 finally:
284 if isinstance(fname, int): 1c
285 os.close(fname) 1c