Coverage for tests / conftest.py: 100.000%

1# SPDX-FileCopyrightText: 2026 Marco Ricci <software@the13thletter.info> 

2# 

3# SPDX-License-Identifier: Zlib 

4 

5from __future__ import annotations (empty)

6 

7import base64 (empty)

8import contextlib (empty)

9import ctypes (empty)

10import os (empty)

11import shutil (empty)

12import socket (empty)

13import subprocess (empty)

14from typing import TYPE_CHECKING, Any, Protocol, TypeVar (empty)

15 

16import packaging.version (empty)

17import pytest (empty)

18from typing_extensions import NamedTuple (empty)

19 

20from derivepassphrase import _types, ssh_agent (empty)

21from derivepassphrase.ssh_agent import socketprovider (empty)

22from tests import data, machinery (empty)

23from tests.data import callables (empty)

24from tests.machinery import hypothesis as hypothesis_machinery (empty)

25from tests.machinery import pytest as pytest_machinery (empty)

26 

27if TYPE_CHECKING: (empty)

28 from collections.abc import Generator, Sequence 

29 

30startup_ssh_auth_sock = os.environ.get("SSH_AUTH_SOCK", None) (empty)

31 

32 

33def pytest_configure(config: pytest.Config) -> None: (empty)

34 """Configure `pytest`: add custom markers.""" 

35 config.addinivalue_line( (empty)

36 "markers", ("heavy_duty: mark test as a slow, heavy-duty test") 

37 ) 

38 config.addinivalue_line( (empty)

39 "markers", 

40 ( 

41 "correctness_focused: " 

42 "mark test as a potentially slow, correctness-focused test" 

43 ), 

44 ) 

45 config.addinivalue_line( (empty)

46 "markers", 

47 ( 

48 "external_agent: " 

49 "mark the agent (fixture output) as external to the test suite" 

50 ), 

51 ) 

52 config.addinivalue_line( (empty)

53 "markers", 

54 ( 

55 "non_reentrant_agent: " 

56 "mark the agent (fixture output) as non-reentrant" 

57 ), 

58 ) 

59 hypothesis_machinery._hypothesis_settings_setup() (empty)

60 

61 

62# https://docs.pytest.org/en/stable/explanation/fixtures.html#a-note-about-fixture-cleanup 

63# https://github.com/pytest-dev/pytest/issues/5243#issuecomment-491522595 

64@pytest.fixture(scope="session", autouse=False) (empty)

65def term_handler() -> Generator[ (empty)

66 None, None, None 

67]: # pragma: no cover [external] 

68 try: 

69 import signal # noqa: PLC0415 

70 

71 sigint_handler = signal.getsignal(signal.SIGINT) 

72 except (ImportError, OSError): 

73 return 

74 else: 

75 orig_term = signal.signal(signal.SIGTERM, sigint_handler) 

76 yield 

77 signal.signal(signal.SIGTERM, orig_term) 

78 

79 

80@pytest.fixture(scope="session") (empty)

81def skip_if_no_af_unix_support() -> None: # pragma: no cover [external] (empty)

82 """Skip the test if Python does not support `AF_UNIX`. 

83 

84 Implemented as a fixture instead of a mark because another "autouse" 

85 session fixture may want to force/simulate non-support of 

86 [`socket.AF_UNIX`][]. 

87 

88 """ 

89 if not hasattr(socket, "AF_UNIX"): (empty)

90 pytest.skip("No Python or system support for UNIX domain sockets.") (empty)

91 

92 

93@pytest.fixture(scope="session") (empty)

94def skip_if_no_ctypes_windll_support() -> None: # pragma: no cover [external] (empty)

95 """Skip the test if Python does not support WinDLL in `ctypes`. 

96 

97 Implemented as a fixture instead of a mark because another "autouse" 

98 session fixture may want to force/simulate non-support of 

99 [`ctypes.WinDLL`][]. 

100 

101 """ 

102 if not hasattr(ctypes, "WinDLL"): (empty)

103 pytest.skip( (empty)

104 "No Python or system support for interfacing Windows DLLs." (empty)

105 ) 

106 

107 

108@pytest.fixture (empty)

109def use_stub_agent() -> Generator[None, None, None]: (empty)

110 """Enforce use of the stubbed SSH agent.""" 

111 with pytest.MonkeyPatch.context() as monkeypatch: (empty)

112 monkeypatch.setattr( (empty)

113 ssh_agent.SSHAgentClient, 

114 "SOCKET_PROVIDERS", 

115 (_types.BuiltinSSHAgentSocketProvider.STUB_AGENT,), 

116 ) 

117 monkeypatch.delenv("SSH_AUTH_SOCK", raising=False) (empty)

118 yield (empty)

119 

120 

121@pytest.fixture (empty)

122def use_stub_agent_with_address() -> Generator[None, None, None]: (empty)

123 """Enforce use of the stub SSH agent with socket address.""" 

124 with pytest.MonkeyPatch.context() as monkeypatch: (empty)

125 monkeypatch.setattr( (empty)

126 ssh_agent.SSHAgentClient, 

127 "SOCKET_PROVIDERS", 

128 (_types.BuiltinSSHAgentSocketProvider.STUB_AGENT_WITH_ADDRESS,), 

129 ) 

130 monkeypatch.setenv( (empty)

131 "SSH_AUTH_SOCK", machinery.StubbedSSHAgentSocketWithAddress.ADDRESS 

132 ) 

133 yield (empty)

134 

135 

136class SSHAgentSpawnFunc(Protocol): (empty)

137 """Spawns an SSH agent, if possible.""" (empty)

138 

139 def __call__( (empty)

140 self, 

141 executable: str | None, (empty)

142 env: dict[str, str], (empty)

143 ) -> subprocess.Popen[str] | None: (empty)

144 """Spawn the SSH agent. 

145 

146 Args: 

147 executable: 

148 The respective SSH agent executable. 

149 env: 

150 The new environment for the respective agent. Should 

151 typically not include an SSH_AUTH_SOCK variable. 

152 

153 Returns: 

154 The spawned SSH agent subprocess. If the executable is 

155 `None`, then return `None` directly. 

156 

157 It is the caller's responsibility to clean up the spawned 

158 subprocess. 

159 

160 Raises: 

161 OSError: 

162 The [`subprocess.Popen`][] call failed. See there. 

163 

164 """ 

165 

166 

167class SSHAgentInterfaceFunc(Protocol): (empty)

168 """Interface an SSH agent, if possible.""" (empty)

169 

170 def __call__( (empty)

171 self, 

172 executable: str | None, (empty)

173 env: dict[str, Any], (empty)

174 ) -> tuple[_types.SSHAgentSocket, str] | None: (empty)

175 """Interface the SSH agent. 

176 

177 Args: 

178 env: 

179 A configuration environment to interface the agent. 

180 Will typically include some kind of address setting 

181 detailing how to communicate with the agent. The 

182 exact contents depend on the specific agent. 

183 

184 Other Args: 

185 executable: 

186 (Ignored. Not relevant for interfacing functions.) 

187 

188 Returns: 

189 A 2-tuple, if possible, containing an abstract socket 

190 connected to the specified SSH agent and the associated 

191 address; otherwise, `None`. 

192 

193 It is the caller's responsibility to clean up the socket 

194 (through use of the context manager protocol). 

195 

196 """ 

197 

198 

199def spawn_pageant_on_posix( # pragma: no cover [external] (empty)

200 executable: str | None, env: dict[str, str] (empty)

201) -> subprocess.Popen[str] | None: (empty)

202 """Spawn an isolated (UNIX) Pageant, if possible. 

203 

204 We attempt to detect whether Pageant is usable, i.e. whether Pageant 

205 has output buffering problems when announcing its authentication 

206 socket. This is the case for Pageant 0.81 and earlier. 

207 

208 Args: 

209 executable: 

210 The path to the Pageant executable. 

211 env: 

212 The new environment for Pageant. Should typically not 

213 include an SSH_AUTH_SOCK variable. 

214 

215 Returns: 

216 The spawned Pageant subprocess. If the executable is `None`, or 

217 if we detect that Pageant cannot be sensibly controlled as 

218 a subprocess, then return `None` directly. 

219 

220 It is the caller's responsibility to clean up the spawned 

221 subprocess. 

222 

223 """ 

224 if os.name == "nt": # pragma: no cover [external] (empty)

225 return None (empty)

226 if executable is None: # pragma: no cover [external] (empty)

227 executable = shutil.which("pageant") (empty)

228 if executable is None: (empty)

229 return None 

230 

231 # Apparently, Pageant 0.81 and lower running in debug mode does 

232 # not actively flush its output. As a result, the first two 

233 # lines, which set the SSH_AUTH_SOCK and the SSH_AGENT_PID, only 

234 # print once the output buffer is flushed, whenever that is. 

235 # 

236 # This has been reported to the PuTTY developers. It is fixed in 

237 # version 0.82, though the PuTTY developers consider this to be an 

238 # abuse of debug mode. A new foreground mode (`--foreground`), also 

239 # introduced in 0.82, provides the desired behavior: no forking, and 

240 # immediately parsable instructions for SSH_AUTH_SOCK and 

241 # SSH_AGENT_PID. 

242 

243 help_output = subprocess.run( (empty)

244 ["pageant", "--help"], (empty)

245 executable=executable, (empty)

246 env=env, (empty)

247 capture_output=True, (empty)

248 text=True, (empty)

249 check=False, (empty)

250 ).stdout (empty)

251 help_lines = help_output.splitlines(True) (empty)

252 pageant_version_string = ( (empty)

253 help_lines[1].strip().removeprefix("Release ") (empty)

254 if len(help_lines) >= 2 (empty)

255 else "" 

256 ) 

257 v0_82 = packaging.version.Version("0.82") (empty)

258 pageant_version = packaging.version.Version(pageant_version_string) (empty)

259 

260 if pageant_version < v0_82: # pragma: no cover [external] (empty)

261 return None 

262 

263 return subprocess.Popen( (empty)

264 ["pageant", "--foreground", "-s"], (empty)

265 executable=executable, (empty)

266 stdin=subprocess.DEVNULL, (empty)

267 stdout=subprocess.PIPE, (empty)

268 shell=False, (empty)

269 env=env, (empty)

270 text=True, (empty)

271 bufsize=1, (empty)

272 ) 

273 

274 

275def spawn_openssh_agent_on_posix( # pragma: no cover [external] (empty)

276 executable: str | None, env: dict[str, str] (empty)

277) -> subprocess.Popen[str] | None: (empty)

278 """Spawn an isolated OpenSSH agent (on UNIX), if possible. 

279 

280 Args: 

281 executable: 

282 The path to the OpenSSH agent executable. 

283 env: 

284 The new environment for the OpenSSH agent. Should typically 

285 not include an SSH_AUTH_SOCK variable. 

286 

287 Returns: 

288 The spawned OpenSSH agent subprocess. If the executable is 

289 `None`, then return `None` directly. 

290 

291 It is the caller's responsibility to clean up the spawned 

292 subprocess. 

293 

294 """ 

295 if os.name == "nt": # pragma: no cover [external] (empty)

296 return None (empty)

297 if executable is None: # pragma: no cover [external] (empty)

298 executable = shutil.which("ssh-agent") (empty)

299 if executable is None: (empty)

300 return None 

301 return subprocess.Popen( (empty)

302 ["ssh-agent", "-D", "-s"], (empty)

303 executable=executable, (empty)

304 stdin=subprocess.DEVNULL, (empty)

305 stdout=subprocess.PIPE, (empty)

306 shell=False, (empty)

307 env=env, (empty)

308 text=True, (empty)

309 bufsize=1, (empty)

310 ) 

311 

312 

313def spawn_noop( # pragma: no cover [unused] (empty)

314 executable: str | None, env: dict[str, str] (empty)

315) -> None: (empty)

316 """Placeholder function. Does nothing.""" (empty)

317 

318 

319def interface_pageant_on_the_annoying_os( (empty)

320 executable: str | None, (empty)

321 env: dict[str, Any], (empty)

322) -> tuple[_types.SSHAgentSocket, str] | None: # pragma: no cover [external] (empty)

323 """Interface a Pageant instance on The Annoying OS, if possible. 

324 

325 Args: 

326 env: 

327 (Ignored.) 

328 

329 Returns: 

330 An abstract socket connected to Pageant, if possible; else 

331 `None`. 

332 

333 It is the caller's responsibility to clean up the socket 

334 (through use of the context manager protocol). 

335 

336 """ 

337 del executable, env (empty)

338 try: (empty)

339 socket = socketprovider.SocketProvider.windows_named_pipe_for_pageant() (empty)

340 except socketprovider.WindowsNamedPipesNotAvailableError: 

341 return None 

342 else: 

343 return (socket, socket.named_pipe_name()) (empty)

344 

345 

346def interface_openssh_agent_on_the_annoying_os( (empty)

347 executable: str | None, (empty)

348 env: dict[str, Any], (empty)

349) -> tuple[_types.SSHAgentSocket, str] | None: # pragma: no cover [external] (empty)

350 """Interface an OpenSSH agent instance on The Annoying OS, if possible. 

351 

352 Args: 

353 env: 

354 (Ignored.) 

355 

356 Returns: 

357 An abstract socket connected to the OpenSSH agent, if possible; 

358 else `None`. 

359 

360 It is the caller's responsibility to clean up the socket 

361 (through use of the context manager protocol). 

362 

363 """ 

364 del executable, env (empty)

365 try: (empty)

366 socket = socketprovider.SocketProvider.windows_named_pipe_for_openssh() (empty)

367 except socketprovider.WindowsNamedPipesNotAvailableError: (empty)

368 return None 

369 else: 

370 return (socket, socket.named_pipe_name()) 

371 

372 

373class SpawnHandler(NamedTuple): (empty)

374 """A handler for a spawned or interfaced SSH agent. 

375 

376 (The choice of terminology "spawn" instead of the more accurate 

377 "spawn or interface" is due to historical reasons.) 

378 

379 Attributes: 

380 instance: 

381 The SSH agent instance type under which this handler is 

382 registered. 

383 

384 This is a full enum, not just a string. In particular, the 

385 enum can be queried for the type of agent represented by 

386 this entry. 

387 executable: 

388 The (optional) full path to the executable. 

389 spawn_func: 

390 The spawn function. 

391 agent_name: 

392 The (optional) display name for this handler. 

393 

394 """ 

395 

396 instance: data.KnownSSHAgentInstance (empty)

397 executable: str | None (empty)

398 spawn_func: SSHAgentSpawnFunc | SSHAgentInterfaceFunc (empty)

399 agent_name: str | None (empty)

400 

401 

402spawn_handlers: dict[str, SpawnHandler] = { (empty)

403 data.KnownSSHAgentInstance.unix_pageant: SpawnHandler( 

404 data.KnownSSHAgentInstance.unix_pageant, 

405 None, 

406 spawn_pageant_on_posix, 

407 "Pageant (UNIX)", 

408 ), 

409 data.KnownSSHAgentInstance.openssh: SpawnHandler( 

410 data.KnownSSHAgentInstance.openssh, 

411 None, 

412 spawn_openssh_agent_on_posix, 

413 "ssh-agent (OpenSSH)", 

414 ), 

415 data.KnownSSHAgentInstance.pageant: SpawnHandler( 

416 data.KnownSSHAgentInstance.pageant, 

417 None, 

418 interface_pageant_on_the_annoying_os, 

419 "Pageant", 

420 ), 

421 data.KnownSSHAgentInstance.openssh_on_windows: SpawnHandler( 

422 data.KnownSSHAgentInstance.openssh_on_windows, 

423 None, 

424 interface_openssh_agent_on_the_annoying_os, 

425 "ssh-agent (OpenSSH on Windows)", 

426 ), 

427 data.KnownSSHAgentInstance.stub_agent_with_address: SpawnHandler( 

428 data.KnownSSHAgentInstance.stub_agent_with_address, 

429 None, 

430 spawn_noop, 

431 "stub_agent_with_address (derivepassphrase test suite)", 

432 ), 

433 data.KnownSSHAgentInstance.stub_agent_with_address_and_deterministic_dsa: SpawnHandler( 

434 data.KnownSSHAgentInstance.stub_agent_with_address_and_deterministic_dsa, 

435 None, 

436 spawn_noop, 

437 "stub_agent_with_address_and_deterministic_dsa " 

438 "(derivepassphrase test suite)", 

439 ), 

440 data.KnownSSHAgentInstance.system_agent: SpawnHandler( 

441 data.KnownSSHAgentInstance.system_agent, 

442 None, 

443 spawn_noop, 

444 None, 

445 ), 

446} 

447""" (empty)

448The standard registry of agent spawning/interfacing functions. 

449 

450(The choice of terminology "spawn" instead of the more accurate "spawn 

451or interface" is due to historical reasons.) 

452""" 

453 

454 

455def external_agent_restriction( (empty)

456 agent_type: data.KnownSSHAgentType, (empty)

457 env_var: str, (empty)

458 *, 

459 default: bool = True, (empty)

460) -> bool: # pragma: no cover [external] (empty)

461 """Classify an SSH agent according to some environment variable. 

462 

463 Look up the given SSH agent type in a certain environment variable, 

464 and report on whether the agent type is listed in that variable. 

465 This can be used to check if the agent is e.g. in a list of 

466 permitted agents, or if it is in the list of agents requiring 

467 special workarounds. Invalid agent type names are silently ignored. 

468 

469 If the environment variable is empty, return a default value. 

470 

471 By design, the stubbed agents have a fixed classification (the 

472 default value), unaffected by the environment variable. 

473 

474 Args: 

475 agent_type: 

476 The agent type to classify. 

477 env_var: 

478 The environment variable in which to look up the agent type 

479 name. 

480 default: 

481 The value to return if the environment variable is unset, or 

482 if a stub agent is to be classified. 

483 

484 Returns: 

485 If the agent type is not a stub agent, and the environment 

486 variable is non-empty, then `True` if the agent type is 

487 mentioned in the environment variable, and `False` otherwise; 

488 the default value otherwise. 

489 

490 See also: 

491 [`is_agent_permitted`][] and [`is_agent_non_reentrant`][], which 

492 make use of this function internally. 

493 

494 """ 

495 if agent_type == data.KnownSSHAgentType.StubbedSSHAgent: (empty)

496 return default (empty)

497 if not os.environ.get(env_var): (empty)

498 return default (empty)

499 marked_agents = { (empty)

500 data.KnownSSHAgentType(x) (empty)

501 for x in os.environ[env_var].split(",") (empty)

502 if x in data.KnownSSHAgentType.__members__ (empty)

503 } 

504 return agent_type in marked_agents (empty)

505 

506 

507def is_agent_permitted( (empty)

508 agent_type: data.KnownSSHAgentType, (empty)

509) -> bool: # pragma: no cover [external] (empty)

510 """May the given SSH agent be spawned by the test harness? 

511 

512 If the environment variable `PERMITTED_SSH_AGENTS` is given, it 

513 names a comma-separated list of known SSH agent names that the test 

514 harness may spawn. Invalid names are silently ignored. If not 

515 given, or empty, then any agent may be spawned. 

516 

517 (To not allow any agent to be spawned, specify a single comma as the 

518 list. But see below.) 

519 

520 The stubbed agents cannot be restricted in this manner, as the test 

521 suite depends on their availability. 

522 

523 """ 

524 return external_agent_restriction( (empty)

525 agent_type, "PERMITTED_SSH_AGENTS", default=True (empty)

526 ) 

527 

528 

529def is_agent_non_reentrant( (empty)

530 agent_type: data.KnownSSHAgentType, (empty)

531) -> bool: # pragma: no cover [external] (empty)

532 """Is the given SSH agent assumed to be non-reentrant? 

533 

534 If the environment variable `NON_REENTRANT_SSH_AGENTS` is given, it 

535 names a comma-separated list of known SSH agent names that are 

536 assumed to be non-reentrant, i.e., where the SSH agent cannot 

537 tolerate multiple clients connecting to it at the same time, and 

538 thus no client for the same agent may be constructed while another 

539 one is already connected. Invalid names are silently ignored. If 

540 not given, or empty, then all agents are considered reentrant, i.e., 

541 all agents are not considered non-reentrant. 

542 

543 (To consider all agents non-reentrant, specify a single comma as the 

544 list. But see below.) 

545 

546 The stubbed agents cannot be influenced in this manner, as the test 

547 suite depends on their reentrancy. 

548 

549 """ 

550 return external_agent_restriction( (empty)

551 agent_type, "NON_REENTRANT_SSH_AGENTS", default=False (empty)

552 ) 

553 

554 

555spawn_handlers_params: list[Sequence] = [] (empty)

556""" (empty)

557The standard registry of agent spawning functions, annotated as 

558a `pytest` parameter set. (In particular, this includes the conditional 

559skip marks.) Used by some test fixtures. 

560 

561(The choice of terminology "spawn" instead of the more accurate "spawn 

562or interface" is due to historical reasons.) 

563""" 

564for instance, handler in spawn_handlers.items(): (empty)

565 marks = [ (empty)

566 pytest.mark.skipif( 

567 not is_agent_permitted(handler.instance.agent_type), 

568 reason="SSH agent excluded via PERMITTED_AGENTS " 

569 "environment variable.", 

570 ), 

571 ] 

572 agent_type = handler.instance.agent_type (empty)

573 # Distinguish spawned/interfaced agents ("external agents") from the 

574 # stub agents ("internal agents"). While this could be inferred from 

575 # the other marks, it is sensible to have a dedicated mark for this. 

576 # Tests that unnecessarily use external agents without testing anything 

577 # that is specific to external agents can be found more easily using 

578 # this mark, and converted to using internal agents only, with speed and 

579 # flakiness-resistance benefits. 

580 if agent_type != data.KnownSSHAgentType.StubbedSSHAgent: (empty)

581 marks.append(pytest.mark.external_agent) (empty)

582 # Allow agents to be marked as non-reentrant. Workaround for an 

583 # issue with GnuPG 2.4.8 on The Annoying OS when `gpg-agent` 

584 # masquerades as OpenSSH's `ssh-agent`. 

585 if is_agent_non_reentrant(agent_type): # pragma: no cover [external] (empty)

586 marks.append(pytest.mark.non_reentrant_agent) (empty)

587 # Mark non-isolated agents as, well, using non-isolated agents. 

588 # Assume by default that an agent is potentially non-isolated unless 

589 # we can be absolutely sure it is isolated, by design. (The "stub" 

590 # agents fall into the latter category.) 

591 if agent_type != data.KnownSSHAgentType.StubbedSSHAgent: (empty)

592 marks.append(pytest_machinery.non_isolated_agent_use) (empty)

593 # Some agents require UNIX domain socket support. 

594 if instance in { (empty)

595 data.KnownSSHAgentInstance.unix_pageant, 

596 data.KnownSSHAgentInstance.openssh, 

597 }: 

598 marks.append( (empty)

599 pytest.mark.skipif( 

600 not hasattr(socket, "AF_UNIX"), 

601 reason="No Python or system support for UNIX domain sockets.", 

602 ) 

603 ) 

604 # Others require Windows named pipe support. 

605 if instance in { (empty)

606 data.KnownSSHAgentInstance.pageant, 

607 data.KnownSSHAgentInstance.openssh_on_windows, 

608 }: 

609 marks.append( (empty)

610 pytest.mark.skipif( 

611 not hasattr(ctypes, "WinDLL"), 

612 reason="No Python or system support for Windows named pipes.", 

613 ) 

614 ) 

615 spawn_handlers_params.append( (empty)

616 pytest.param(handler, id=instance, marks=marks) 

617 ) 

618 

619 

620Popen = TypeVar("Popen", bound=subprocess.Popen) (empty)

621 

622 

623# As of v0.6, all SSH agents we interact with on The Annoying OS are 

624# interfaced, not spawned. Therefore, this context manager is unused on 

625# The Annoying OS. 

626@contextlib.contextmanager (empty)

627def terminate_on_exit( (empty)

628 proc: Popen, 

629) -> Generator[Popen, None, None]: # pragma: unless posix no cover [unused] 

630 """Terminate and wait for the subprocess upon exiting the context. 

631 

632 Args: 

633 proc: 

634 The subprocess to manage. 

635 

636 Returns: 

637 A context manager. Upon entering the manager, return the 

638 managed subprocess. Upon exiting the manager, terminate the 

639 process and wait for it. 

640 

641 """ 

642 try: (empty)

643 yield proc (empty)

644 finally: 

645 proc.terminate() (empty)

646 proc.wait() (empty)

647 

648 

649class CannotSpawnError(RuntimeError): (empty)

650 """Cannot spawn the SSH agent.""" 

651 

652 

653def spawn_named_agent( # noqa: C901 (empty)

654 executable: str | None, (empty)

655 spawn_func: SSHAgentSpawnFunc | SSHAgentInterfaceFunc, (empty)

656 agent_type: data.KnownSSHAgentType, (empty)

657 agent_name: str, (empty)

658) -> Generator[ (empty)

659 data.SpawnedSSHAgentInfo, None, None 

660]: # pragma: no cover [external] 

661 """Spawn the named SSH agent and check that it is operational. 

662 

663 Using the correct agent-specific spawn function from the 

664 [`spawn_handlers`][] registry, spawn the named SSH agent (according 

665 to its declared type), then set up the communication channel and 

666 yield an SSH agent client connected to this agent. After resuming, 

667 tear down the communication channel and terminate the SSH agent. 

668 

669 The SSH agent's instructions for setting up the communication 

670 channel are parsed with [`callables.parse_sh_export_line`][]. See 

671 the caveats there. 

672 

673 (The choice of terminology "spawned" instead of the more accurate 

674 "spawned or interfaced" is due to historical reasons.) 

675 

676 Args: 

677 executable: 

678 The full path of the executable to spawn. If not given, we 

679 attempt to spawn the agent under its conventional executable 

680 name. 

681 spawn_func: 

682 The agent-specific spawn function. 

683 agent_type: 

684 The agent type. 

685 agent_name: 

686 The agent's display name. 

687 

688 Yields: 

689 A 3-tuple containing the agent type, an SSH agent client 

690 connected to this agent, and a boolean indicating whether this 

691 agent was actually spawned in an isolated manner. 

692 

693 Only one tuple will ever be yielded. After resuming, the 

694 connected client will be torn down, as will the agent if it was 

695 isolated. 

696 

697 Raises: 

698 CannotSpawnError: 

699 We failed to spawn the agent or otherwise set up the 

700 environment/communication channel/etc. 

701 

702 """ 

703 # pytest's fixture system does not seem to guarantee that 

704 # environment variables are set up correctly if nested and 

705 # parametrized fixtures are used: it is possible that "outer" 

706 # parametrized fixtures are torn down only after other "outer" 

707 # fixtures of the same parameter set have run. So our fixtures set 

708 # SSH_AUTH_SOCK explicitly to the value saved at interpreter 

709 # startup. 

710 # 

711 # Here, we verify at most major steps that SSH_AUTH_SOCK didn't 

712 # change under our nose. 

713 assert os.environ.get("SSH_AUTH_SOCK") == startup_ssh_auth_sock, ( (empty)

714 f"SSH_AUTH_SOCK mismatch when checking for spawnable {agent_name}" 

715 ) 

716 exit_stack = contextlib.ExitStack() (empty)

717 agent_env = os.environ.copy() (empty)

718 ssh_auth_sock = agent_env.pop("SSH_AUTH_SOCK", None) (empty)

719 proc_or_socket_data = spawn_func(executable=executable, env=agent_env) (empty)

720 with exit_stack: (empty)

721 if agent_type == data.KnownSSHAgentType.StubbedSSHAgent: (empty)

722 ssh_auth_sock = None (empty)

723 elif spawn_func is spawn_noop: (empty)

724 ssh_auth_sock = os.environ.get("SSH_AUTH_SOCK") (empty)

725 elif proc_or_socket_data is None: # pragma: no cover [external] (empty)

726 err_msg = f"Cannot spawn or interface with usable {agent_name}" (empty)

727 raise CannotSpawnError(err_msg) (empty)

728 elif isinstance(proc_or_socket_data, subprocess.Popen): (empty)

729 proc = proc_or_socket_data (empty)

730 exit_stack.enter_context(terminate_on_exit(proc)) (empty)

731 assert os.environ.get("SSH_AUTH_SOCK") == startup_ssh_auth_sock, ( (empty)

732 f"SSH_AUTH_SOCK mismatch after spawning {agent_name}" 

733 ) 

734 assert proc.stdout is not None (empty)

735 ssh_auth_sock_line = proc.stdout.readline() (empty)

736 try: (empty)

737 ssh_auth_sock = callables.parse_sh_export_line( (empty)

738 ssh_auth_sock_line, env_name="SSH_AUTH_SOCK" (empty)

739 ) 

740 except ValueError: # pragma: no cover [external] 

741 err_msg = f"Cannot parse agent output: {ssh_auth_sock_line!r}" 

742 raise CannotSpawnError(err_msg) from None 

743 pid_line = proc.stdout.readline() (empty)

744 if ( (empty)

745 "pid" not in pid_line.lower() (empty)

746 and "_pid" not in pid_line.lower() 

747 ): # pragma: no cover [external] 

748 err_msg = f"Cannot parse agent output: {pid_line!r}" 

749 raise CannotSpawnError(err_msg) 

750 else: 

751 ssh_auth_sock = proc_or_socket_data[1] (empty)

752 monkeypatch = exit_stack.enter_context(pytest.MonkeyPatch.context()) (empty)

753 if agent_type == data.KnownSSHAgentType.StubbedSSHAgent: (empty)

754 assert ssh_auth_sock is None (empty)

755 monkeypatch.setenv( (empty)

756 "SSH_AUTH_SOCK", (empty)

757 machinery.StubbedSSHAgentSocketWithAddress.ADDRESS, (empty)

758 ) 

759 monkeypatch.setattr( (empty)

760 ssh_agent.SSHAgentClient, (empty)

761 "SOCKET_PROVIDERS", (empty)

762 [ (empty)

763 data.KnownSSHAgentInstance.stub_agent_with_address_and_deterministic_dsa (empty)

764 ] 

765 if "stub_agent_with_address_and_deterministic_dsa" (empty)

766 in agent_name (empty)

767 else [data.KnownSSHAgentInstance.stub_agent_with_address], (empty)

768 ) 

769 client = exit_stack.enter_context( (empty)

770 ssh_agent.SSHAgentClient.ensure_agent_subcontext( (empty)

771 machinery.StubbedSSHAgentSocketWithAddressAndDeterministicDSA() (empty)

772 if "stub_agent_with_address_and_deterministic_dsa" (empty)

773 in agent_name (empty)

774 else machinery.StubbedSSHAgentSocketWithAddress() (empty)

775 ) 

776 ) 

777 elif ( (empty)

778 not isinstance(proc_or_socket_data, subprocess.Popen) (empty)

779 and proc_or_socket_data is not None (empty)

780 ): 

781 socket: _types.SSHAgentSocket 

782 socket, ssh_auth_sock = proc_or_socket_data (empty)

783 monkeypatch.setenv("SSH_AUTH_SOCK", ssh_auth_sock) (empty)

784 client = exit_stack.enter_context( (empty)

785 ssh_agent.SSHAgentClient.ensure_agent_subcontext(conn=socket) (empty)

786 ) 

787 else: 

788 if ssh_auth_sock: (empty)

789 monkeypatch.setenv("SSH_AUTH_SOCK", ssh_auth_sock) (empty)

790 client = exit_stack.enter_context( (empty)

791 ssh_agent.SSHAgentClient.ensure_agent_subcontext() (empty)

792 ) 

793 # We sanity-test the connected SSH agent if it is not one of our 

794 # test agents, because allowing the user to run the test suite 

795 # with a clearly faulty agent would likely and unfairly 

796 # misattribute the agent's protocol violations to our test 

797 # suite. On the flip side, for our own test agents, the 

798 # correctness tests are part of the test suite, so we don't want 

799 # the *setup* code here to already decide that the agent is 

800 # unsuitable. Therefore, do a sanity check if and only if the 

801 # agent is not one of our test agents, and if the check fails, 

802 # skip this agent. 

803 if ( (empty)

804 agent_type != data.KnownSSHAgentType.StubbedSSHAgent (empty)

805 ): # pragma: no cover [external] 

806 try: (empty)

807 client.list_keys() # sanity test (empty)

808 except ( 

809 EOFError, 

810 OSError, 

811 ssh_agent.SSHAgentFailedError, 

812 ) as exc: # pragma: no cover [failsafe] 

813 msg = f'agent failed the "list keys" sanity test: {exc!r}' 

814 raise CannotSpawnError(msg) from exc 

815 yield data.SpawnedSSHAgentInfo( (empty)

816 agent_type, (empty)

817 client, (empty)

818 spawn_func is not spawn_noop (empty)

819 if isinstance(proc_or_socket_data, subprocess.Popen) (empty)

820 else False, (empty)

821 ) 

822 assert os.environ.get("SSH_AUTH_SOCK", None) == startup_ssh_auth_sock, ( (empty)

823 f"SSH_AUTH_SOCK mismatch after tearing down {agent_name}" 

824 ) 

825 

826 

827# Hack: We cannot associate markers with fixtures (the pytest devs say 

828# there are many unresolved questions when it comes to fixture and 

829# marker overriding in such a case), so we fake this by trivially 

830# parametrizing the fixture. (The parametrizations *can* contain the 

831# necessary marks.) 

832@pytest.fixture( (empty)

833 params=[ (empty)

834 pytest.param( (empty)

835 "non_isolated_agent_use", (empty)

836 marks=pytest_machinery.non_isolated_agent_use, (empty)

837 ) 

838 ] 

839) 

840def running_ssh_agent( # pragma: no cover [external] (empty)

841 request: pytest.FixtureRequest, (empty)

842) -> Generator[data.RunningSSHAgentInfo, None, None]: (empty)

843 """Ensure a running SSH agent, if possible, as a pytest fixture. 

844 

845 Check for a running SSH agent, or spawn/interface a new one if 

846 possible. We know how to spawn OpenSSH's agent and PuTTY's Pageant 

847 on UNIX, or interface them on The Annoying OS. If spawned this way, 

848 the agent does not persist beyond the test. 

849 

850 This fixture can neither guarantee a particular running agent, nor 

851 can it guarantee a particular set of loaded keys. 

852 

853 Yields: 

854 A 2-tuple `(ssh_auth_sock, agent_type)`, where `ssh_auth_sock` 

855 is the value of the `SSH_AUTH_SOCK` environment variable, to be 

856 used to connect to the running agent, and `agent_type` is the 

857 agent type. 

858 

859 Raises: 

860 pytest.skip.Exception: 

861 If no agent is running or can be spawned, skip this test. 

862 

863 """ 

864 del request (empty)

865 

866 def prepare_environment( (empty)

867 agent_type: data.KnownSSHAgentType, (empty)

868 ) -> Generator[data.RunningSSHAgentInfo, None, None]: (empty)

869 with pytest.MonkeyPatch.context() as monkeypatch: (empty)

870 if agent_type == data.KnownSSHAgentType.StubbedSSHAgent: (empty)

871 monkeypatch.setattr( 

872 ssh_agent.SSHAgentClient, 

873 "SOCKET_PROVIDERS", 

874 [ 

875 _types.BuiltinSSHAgentSocketProvider.STUB_AGENT_WITH_ADDRESS 

876 ], 

877 ) 

878 monkeypatch.setenv( 

879 "SSH_AUTH_SOCK", 

880 machinery.StubbedSSHAgentSocketWithAddress.ADDRESS, 

881 ) 

882 yield data.RunningSSHAgentInfo( 

883 machinery.StubbedSSHAgentSocketWithAddressAndDeterministicDSA, 

884 data.KnownSSHAgentType.StubbedSSHAgent, 

885 ) 

886 else: 

887 yield data.RunningSSHAgentInfo( (empty)

888 os.environ["SSH_AUTH_SOCK"], (empty)

889 agent_type, (empty)

890 ) 

891 

892 with pytest.MonkeyPatch.context() as monkeypatch: (empty)

893 # pytest's fixture system does not seem to guarantee that 

894 # environment variables are set up correctly if nested and 

895 # parametrized fixtures are used: it is possible that "outer" 

896 # parametrized fixtures are torn down only after other "outer" 

897 # fixtures of the same parameter set have run. So set 

898 # SSH_AUTH_SOCK explicitly to the value saved at interpreter 

899 # startup. 

900 if startup_ssh_auth_sock: (empty)

901 monkeypatch.setenv("SSH_AUTH_SOCK", startup_ssh_auth_sock) (empty)

902 else: 

903 monkeypatch.delenv("SSH_AUTH_SOCK", raising=False) (empty)

904 for handler in spawn_handlers.values(): (empty)

905 if not is_agent_permitted(handler.instance.agent_type): (empty)

906 continue 

907 try: (empty)

908 for _agent_info in spawn_named_agent( (empty)

909 executable=handler.executable, (empty)

910 spawn_func=handler.spawn_func, (empty)

911 agent_type=handler.instance.agent_type, (empty)

912 agent_name=handler.agent_name or handler.instance, (empty)

913 ): 

914 yield from prepare_environment(handler.instance.agent_type) (empty)

915 except (KeyError, OSError, CannotSpawnError): (empty)

916 continue (empty)

917 return (empty)

918 pytest.skip("No SSH agent running or spawnable") 

919 

920 

921@pytest.fixture(params=spawn_handlers_params) (empty)

922def spawn_ssh_agent( (empty)

923 request: pytest.FixtureRequest, (empty)

924) -> Generator[ (empty)

925 data.SpawnedSSHAgentInfo, None, None 

926]: # pragma: no cover [external] 

927 """Spawn or interface an SSH agent, as a pytest fixture. 

928 

929 Spawn a new SSH agent (isolated from other SSH use by other 

930 processes, on a best-effort basis), or interface a running 

931 (non-isolated) SSH agent. We know how to spawn OpenSSH's agent (on 

932 UNIX) and PuTTY's Pageant (on UNIX) and the stubbed agents, and how 

933 to interface PuTTY's Pageant (on The Annoying OS), OpenSSH's agent 

934 (on The Annoying OS) and the "system" agent (on every OS). 

935 

936 (The choice of terminology "spawned" instead of the more accurate 

937 "spawned or interfaced" is due to historical reasons.) 

938 

939 Yields: 

940 A [named tuple][collections.namedtuple] containing information 

941 about the spawned agent, e.g. the software product, a client 

942 connected to the agent, and whether the agent is isolated from 

943 other clients. 

944 

945 Raises: 

946 pytest.skip.Exception: 

947 If the agent cannot be spawned or interfaced, skip this 

948 test. 

949 

950 """ 

951 

952 with pytest.MonkeyPatch.context() as monkeypatch: (empty)

953 # pytest's fixture system does not seem to guarantee that 

954 # environment variables are set up correctly if nested and 

955 # parametrized fixtures are used: it is possible that "outer" 

956 # parametrized fixtures are torn down only after other "outer" 

957 # fixtures of the same parameter set have run. So set 

958 # SSH_AUTH_SOCK explicitly to the value saved at interpreter 

959 # startup. 

960 if startup_ssh_auth_sock: # pragma: no cover [external] (empty)

961 monkeypatch.setenv("SSH_AUTH_SOCK", startup_ssh_auth_sock) (empty)

962 else: # pragma: no cover [external] 

963 monkeypatch.delenv("SSH_AUTH_SOCK", raising=False) (empty)

964 try: (empty)

965 yield from spawn_named_agent( (empty)

966 executable=request.param.executable, (empty)

967 spawn_func=request.param.spawn_func, (empty)

968 agent_type=request.param.instance.agent_type, (empty)

969 agent_name=request.param.agent_name or request.param.instance, (empty)