Coverage for src / derivepassphrase / ssh_agent / socketprovider.py: 100.000%

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

2# 

3# SPDX-License-Identifier: Zlib 

4 

5"""Machinery for providing sockets connected to SSH agents.""" 

6 

7from __future__ import annotations (empty)

8 

9import collections (empty)

10import ctypes (empty)

11import enum (empty)

12import errno (empty)

13import graphlib (empty)

14import hashlib (empty)

15import itertools (empty)

16import operator (empty)

17import os (empty)

18import socket (empty)

19from ctypes.wintypes import ( # type: ignore[attr-defined] (empty)

20 BOOL, 

21 DWORD, 

22 HANDLE, 

23 LPCVOID, 

24 LPCWSTR, 

25 LPDWORD, 

26 LPVOID, 

27) 

28from typing import TYPE_CHECKING, cast (empty)

29 

30from derivepassphrase import _types (empty)

31 

32if TYPE_CHECKING: (empty)

33 from collections.abc import Callable, Mapping 

34 from collections.abc import Set as AbstractSet 

35 from typing import ClassVar 

36 

37 from typing_extensions import ( 

38 Any, 

39 Buffer, 

40 Literal, 

41 Protocol, 

42 Self, 

43 TypeAlias, 

44 TypeVar, 

45 ) 

46 

47 SSHAgentSocketProviderT = TypeVar( 

48 "SSHAgentSocketProviderT", bound=_types.SSHAgentSocketProvider 

49 ) 

50 """A parametrized SSH agent socket provider.""" 

51 

52 kernel32: ctypes.WinDLL # type: ignore[name-defined] 

53 crypt32: ctypes.WinDLL # type: ignore[name-defined] 

54 

55 CryptProtectMemory: Callable[[LPVOID, DWORD, DWORD], BOOL] 

56 CreateFile: Callable[ 

57 [ 

58 LPCWSTR, 

59 DWORD, 

60 DWORD, 

61 None, 

62 DWORD, 

63 DWORD, 

64 HANDLE | None, 

65 ], 

66 HANDLE, 

67 ] 

68 ReadFile: Callable[[HANDLE, LPCVOID, DWORD, LPDWORD, LPOVERLAPPED], BOOL] 

69 WriteFile: Callable[[HANDLE, LPCVOID, DWORD, LPDWORD, LPOVERLAPPED], BOOL] 

70 CloseHandle: Callable[[HANDLE], BOOL] 

71 GetOverlappedResult: Callable[[HANDLE, LPOVERLAPPED, LPDWORD, BOOL], BOOL] 

72 CreateEvent: Callable[[None, BOOL, BOOL, None], HANDLE] 

73 

74 class WinErrorProtocol(Protocol): 

75 def __call__(self, err: int | None = None, /) -> OSError: ... 

76 

77 GetLastError: Callable[[], int] 

78 WinError: WinErrorProtocol 

79 

80 

81class _OverlappedDummyStruct(ctypes.Structure): (empty)

82 """The DUMMYSTRUCTNAME structure in the definition of OVERLAPPED.""" 

83 

84 _fields_ = (("Offset", DWORD), ("OffsetHigh", DWORD)) (empty)

85 

86 

87class _OverlappedDummyUnion(ctypes.Union): (empty)

88 """The DUMMYUNIONNAME union in the definition of OVERLAPPED.""" 

89 

90 _fields_ = ( (empty)

91 ("DUMMYSTRUCTNAME", _OverlappedDummyStruct), 

92 ("Pointer", ctypes.c_void_p), 

93 ) 

94 

95 

96class OVERLAPPED(ctypes.Structure): (empty)

97 """The data structure for overlapped (async) I/O on The Annoying OS.""" 

98 

99 _fields_ = ( (empty)

100 ("Internal", ctypes.POINTER(ctypes.c_ulong)), 

101 ("InternalHigh", ctypes.POINTER(ctypes.c_ulong)), 

102 ("DUMMYUNIONNAME", _OverlappedDummyUnion), 

103 ("hEvent", HANDLE), 

104 ) 

105 

106 

107if TYPE_CHECKING: (empty)

108 LPOVERLAPPED: TypeAlias = ctypes._Pointer[OVERLAPPED] # noqa: SLF001 

109else: 

110 LPOVERLAPPED = ctypes.POINTER(OVERLAPPED) (empty)

111 

112__all__ = ("SocketProvider",) (empty)

113 

114 

115class NoSuchProviderError(KeyError): (empty)

116 """No such SSH agent socket provider is known.""" 

117 

118 

119class UnixDomainSocketsNotAvailableError(NotImplementedError): (empty)

120 """This Python installation does not support UNIX domain sockets. 

121 

122 On this installation, the standard library symbol 

123 [`socket.AF_UNIX`][] is not available, necessary to create UNIX 

124 domain sockets. 

125 

126 """ 

127 

128 

129class WindowsNamedPipesNotAvailableError(NotImplementedError): (empty)

130 """This Python installation does not support Windows named pipes. 

131 

132 On this installation, the standard library symbol 

133 [`ctypes.WinDLL`][] is not available, necessary to interface with 

134 the Annoying Operating System's standard library to create Windows 

135 named pipes. 

136 

137 """ 

138 

139 

140GENERIC_READ = 0x80000000 (empty)

141GENERIC_WRITE = 0x40000000 (empty)

142 

143FILE_SHARE_READ = 0x00000001 (empty)

144FILE_SHARE_WRITE = 0x00000002 (empty)

145FILE_FLAG_OVERLAPPED = 0x40000000 (empty)

146 

147OPEN_EXISTING = 0x3 (empty)

148 

149INVALID_HANDLE_VALUE = -1 (empty)

150 

151CRYPTPROTECTMEMORY_BLOCK_SIZE = 0x10 (empty)

152CRYPTPROTECTMEMORY_CROSS_PROCESS = 0x1 (empty)

153 

154PIPE_PREFIX = "//./pipe/".replace("/", "\\") (empty)

155 

156ERROR_PIPE_BUSY = 231 (empty)

157ERROR_IO_PENDING = 997 (empty)

158 

159 

160# The type checker and the standard library use different 

161# (finer-grained?) types and type definitions on The Annoying Operating 

162# System to annotate things derived from ctypes.WinDLL. Furthermore, on 

163# other OSes, ctypes.WinDLL is missing entirely. So the type checker 

164# attempts to validate our code against *very* different type 

165# hierarchies, depending on what OS it is running on, and so we require 

166# a lot of type checking exclusions to keep the types reasonably narrow. 

167# 

168# Additionally, the official documentation for kernel32.dll and 

169# crypt32.dll (and The Annoying OS in general) uses CamelCase for both 

170# function names and function parameters. This clashes with the 

171# standard Python naming conventions, so we need linting exclusions for 

172# each such name. 

173# 

174# Finally, since this code is mostly platform-specific, we use coverage 

175# pragmas for the respective branches to avoid surprises about missing 

176# coverage. 

177# 

178# All these factors necessitate *a lot* of "type: ignore", "noqa:" and 

179# "pragma: ... no cover" comments below. 

180 

181try: # pragma: unless the-annoying-os no cover (empty)

182 import ctypes (empty)

183 

184 GetLastError = ctypes.GetLastError # type: ignore[attr-defined] (empty)

185 WinError = ctypes.WinError # type: ignore[attr-defined] (empty)

186except ( (empty)

187 AttributeError, 

188 FileNotFoundError, 

189 ImportError, 

190): # pragma: unless posix no cover 

191 

192 def GetLastError() -> int: # noqa: N802 # pragma: no cover [failsafe] (empty)

193 raise NotImplementedError 

194 

195 def WinError(err: int | None = None, /) -> OSError: # noqa: N802 # pragma: no cover [failsafe] (empty)

196 raise NotImplementedError 

197 

198 

199try: # pragma: unless the-annoying-os no cover (empty)

200 crypt32 = ctypes.WinDLL("crypt32.dll") # type: ignore[attr-defined] (empty)

201 CryptProtectMemory = crypt32.CryptProtectMemory # type: ignore[attr-defined] (empty)

202except ( (empty)

203 AttributeError, 

204 FileNotFoundError, 

205 ImportError, 

206): # pragma: unless posix no cover 

207 

208 def CryptProtectMemory( # noqa: N802 (empty)

209 pDataIn: LPVOID, # noqa: N803 (empty)

210 cbDataIn: DWORD, # noqa: N803 (empty)

211 dwFlags: DWORD, # noqa: N803 (empty)

212 ) -> BOOL: # pragma: no cover [external] (empty)

213 del pDataIn, cbDataIn, dwFlags 

214 raise NotImplementedError 

215 

216else: # pragma: unless the-annoying-os no cover 

217 CryptProtectMemory.argtypes = [LPVOID, DWORD, DWORD] # type: ignore[attr-defined] (empty)

218 CryptProtectMemory.restype = BOOL # type: ignore[attr-defined] (empty)

219 

220 

221def _errcheck( (empty)

222 result: int, (empty)

223 func: Callable, (empty)

224 arguments: Any, # noqa: ANN401 (empty)

225) -> int: # pragma: no cover [external] (empty)

226 del func 9 ctx1abcdefgih

227 del arguments 9 ctx1abcdefgih

228 # Even if they have the same value, ctypes pointers don't 

229 # compare equal to each other. We need to compare the values 

230 # explicitly, keeping the ranges in mind as well. 

231 result_value = HANDLE(result).value 9 ctx1abcdefgih

232 invalid_value = HANDLE(INVALID_HANDLE_VALUE).value 9 ctx1abcdefgih

233 if result_value == invalid_value: 9 ctx1abcdefgih

234 raise WinError() 4 ctx1abci

235 assert result_value is not None # for the type checker 8 ctx1abcdefgh

236 return result_value 8 ctx1abcdefgh

237 

238 

239try: # pragma: unless the-annoying-os no cover (empty)

240 kernel32 = ctypes.WinDLL("Kernel32.dll") # type: ignore[attr-defined] (empty)

241 CreateFile = kernel32.CreateFileW # type: ignore[attr-defined] (empty)

242 ReadFile = kernel32.ReadFile # type: ignore[attr-defined] (empty)

243 WriteFile = kernel32.WriteFile # type: ignore[attr-defined] (empty)

244 CloseHandle = kernel32.CloseHandle # type: ignore[attr-defined] (empty)

245 GetOverlappedResult = kernel32.GetOverlappedResult # type: ignore[attr-defined] (empty)

246 CreateEvent = kernel32.CreateEventW # type: ignore[attr-defined] (empty)

247except ( (empty)

248 AttributeError, 

249 FileNotFoundError, 

250): # pragma: unless posix no cover 

251 

252 def CreateFile( # noqa: N802, PLR0913, PLR0917 (empty)

253 lpFilename: LPCWSTR, # noqa: N803 (empty)

254 dwDesiredAccess: DWORD, # noqa: N803 (empty)

255 dwShareMode: DWORD, # noqa: N803 (empty)

256 lpSecurityAttributes: None, # noqa: N803 (empty)

257 dwCreationDisposition: DWORD, # noqa: N803 (empty)

258 dwFlagsAndAttributes: DWORD, # noqa: N803 (empty)

259 hTemplateFile: HANDLE | None, # noqa: N803 (empty)

260 ) -> HANDLE: # pragma: no cover [external] (empty)

261 del lpFilename 

262 del dwDesiredAccess 

263 del dwShareMode 

264 del lpSecurityAttributes 

265 del dwCreationDisposition 

266 del dwFlagsAndAttributes 

267 del hTemplateFile 

268 msg = "This Python version does not support Windows named pipes." 

269 raise WindowsNamedPipesNotAvailableError(msg) 

270 

271 def ReadFile( # noqa: N802 (empty)

272 hFile: HANDLE, # noqa: N803 (empty)

273 lpBuffer: LPVOID, # noqa: N803 (empty)

274 nNumberOfBytesToRead: DWORD, # noqa: N803 (empty)

275 lpNumberOfBytesRead: LPDWORD, # noqa: N803 (empty)

276 lpOverlapped: LPOVERLAPPED, # noqa: N803 (empty)

277 ) -> BOOL: # pragma: no cover [external] (empty)

278 del hFile 

279 del lpBuffer 

280 del nNumberOfBytesToRead 

281 del lpNumberOfBytesRead 

282 del lpOverlapped 

283 raise OSError(errno.ENOTSUP, os.strerror(errno.ENOTSUP)) 

284 

285 def WriteFile( # noqa: N802 (empty)

286 hFile: HANDLE, # noqa: N803 (empty)

287 lpBuffer: LPVOID, # noqa: N803 (empty)

288 nNumberOfBytesToWrite: DWORD, # noqa: N803 (empty)

289 lpNumberOfBytesWritten: LPDWORD, # noqa: N803 (empty)

290 lpOverlapped: LPOVERLAPPED, # noqa: N803 (empty)

291 ) -> BOOL: # pragma: no cover [external] (empty)

292 del hFile 

293 del lpBuffer 

294 del nNumberOfBytesToWrite 

295 del lpNumberOfBytesWritten 

296 del lpOverlapped 

297 raise OSError(errno.ENOTSUP, os.strerror(errno.ENOTSUP)) 

298 

299 def CloseHandle( # noqa: N802 (empty)

300 hHandle: HANDLE, # noqa: N803 (empty)

301 ) -> BOOL: # pragma: no cover [external] (empty)

302 del hHandle 

303 raise OSError(errno.ENOTSUP, os.strerror(errno.ENOTSUP)) 

304 

305 def GetOverlappedResult( # noqa: N802 (empty)

306 hFile: HANDLE, # noqa: N803 (empty)

307 lpOverlapped: LPOVERLAPPED, # noqa: N803 (empty)

308 lpNumberOfBytesTransferred: LPDWORD, # noqa: N803 (empty)

309 bWait: BOOL, # noqa: N803 (empty)

310 ) -> BOOL: # pragma: no cover [external] (empty)

311 del hFile, lpOverlapped, lpNumberOfBytesTransferred, bWait 

312 raise OSError(errno.ENOTSUP, os.strerror(errno.ENOTSUP)) 

313 

314 def CreateEvent( # noqa: N802 (empty)

315 lpEventAttributes: None, # noqa: N803 (empty)

316 bManualReset: BOOL, # noqa: N803 (empty)

317 bInitialState: BOOL, # noqa: N803 (empty)

318 lpName: None, # noqa: N803 (empty)

319 ) -> HANDLE: # pragma: no cover [external] (empty)

320 del lpEventAttributes, bManualReset, bInitialState, lpName 

321 raise OSError(errno.ENOTSUP, os.strerror(errno.ENOTSUP)) 

322 

323 

324else: # pragma: unless the-annoying-os no cover 

325 CreateFile.argtypes = [ # type: ignore[attr-defined] (empty)

326 LPCWSTR, 

327 DWORD, 

328 DWORD, 

329 # Actually, LPSECURITY_ATTRIBUTES, but we always pass 

330 # None/NULL. 

331 ctypes.c_void_p, 

332 DWORD, 

333 DWORD, 

334 # We always pass None/NULL. 

335 HANDLE, 

336 ] 

337 CreateEvent.argtypes = [ # type: ignore[attr-defined] (empty)

338 # Actually, LPSECURITY_ATTRIBUTES, but we always pass 

339 # None/NULL. 

340 ctypes.c_void_p, 

341 BOOL, 

342 BOOL, 

343 # We always pass None/NULL. 

344 LPCWSTR, 

345 ] 

346 CreateFile.restype = HANDLE # type: ignore[attr-defined] (empty)

347 CreateEvent.restype = HANDLE # type: ignore[attr-defined] (empty)

348 CreateFile.errcheck = _errcheck # type: ignore[assignment, attr-defined] (empty)

349 CreateEvent.errcheck = _errcheck # type: ignore[assignment, attr-defined] (empty)

350 ReadFile.argtypes = [ # type: ignore[attr-defined] (empty)

351 HANDLE, 

352 LPVOID, 

353 DWORD, 

354 LPDWORD, 

355 LPOVERLAPPED, 

356 ] 

357 WriteFile.argtypes = [ # type: ignore[attr-defined] (empty)

358 HANDLE, 

359 LPVOID, 

360 DWORD, 

361 LPDWORD, 

362 LPOVERLAPPED, 

363 ] 

364 CloseHandle.argtypes = [HANDLE] # type: ignore[attr-defined] (empty)

365 GetOverlappedResult.argtypes = [ # type: ignore[attr-defined] (empty)

366 HANDLE, 

367 LPOVERLAPPED, 

368 LPDWORD, 

369 # We always pass True. 

370 BOOL, 

371 ] 

372 ReadFile.restype = BOOL # type: ignore[attr-defined] (empty)

373 WriteFile.restype = BOOL # type: ignore[attr-defined] (empty)

374 CloseHandle.restype = BOOL # type: ignore[attr-defined] (empty)

375 GetOverlappedResult.restype = BOOL # type: ignore[attr-defined] (empty)

376 

377 

378class WindowsNamedPipeHandle: (empty)

379 """A Windows named pipe handle. 

380 

381 This handle implements the [`SSHAgentSocket`][_types.SSHAgentSocket] 

382 interface. It is only constructable if the Python installation can 

383 successfully call into the Annoying OS `kernel32.dll` library via 

384 [`ctypes`][]. 

385 

386 """ 

387 

388 _IO_ON_CLOSED_FILE = "I/O operation on closed file." (empty)

389 

390 def __init__( (empty)

391 self, name: str 

392 ) -> None: # pragma: unless the-annoying-os no cover 

393 """Create a named pipe handle. 

394 

395 Args: 

396 name: 

397 The named pipe's name. 

398 

399 Raises: 

400 OSError: 

401 The system failed to open the named pipe. 

402 ValueError: 

403 The named pipe's name does not designate a named pipe. 

404 WindowsNamedPipesNotAvailableError: 

405 This Python version does not support Windows named 

406 pipes. 

407 

408 """ 

409 if not name.replace("/", "\\").startswith(PIPE_PREFIX): 10 ctx1abcdefiqPr

410 msg = f"Invalid named pipe name: {name!r}" 1 ctx1P

411 raise ValueError(msg) 1 ctx1P

412 self.handle: HANDLE | None = None 9 ctx1abcdefiqr

413 while self.handle is None: 9 ctx1abcdefiqr

414 try: 9 ctx1abcdefiqr

415 self.handle = CreateFile( 9 ctx1abcdefiqr

416 ctypes.c_wchar_p(name), 

417 ctypes.c_ulong(GENERIC_READ | GENERIC_WRITE), 

418 ctypes.c_ulong(FILE_SHARE_READ | FILE_SHARE_WRITE), 

419 None, 

420 ctypes.c_ulong(OPEN_EXISTING), 

421 ctypes.c_ulong(FILE_FLAG_OVERLAPPED), 

422 None, 

423 ) 

424 except BlockingIOError: # pragma: no cover [external] 4 ctx1abci

425 continue # retry 

426 except OSError as exc: # pragma: no cover [external] 4 ctx1abci

427 if getattr(exc, "winerror", None) == ERROR_PIPE_BUSY: 4 ctx1abci

428 continue # retry 

429 raise 4 ctx1abci

430 self._name = name 8 ctx1abcdefqr

431 

432 def __enter__(self) -> Self: # pragma: unless the-annoying-os no cover (empty)

433 return self 8 ctx1abcdefqr

434 

435 def __exit__( (empty)

436 self, *args: object 

437 ) -> Literal[False]: # pragma: unless the-annoying-os no cover 

438 try: 8 ctx1abcdefqr

439 if self.handle is not None: 8 ctx1abcdefqr

440 CloseHandle(self.handle) 8 ctx1abcdefqr

441 return False 8 ctx1abcdefqr

442 finally: 

443 self.handle = None 8 ctx1abcdefqr

444 

445 def recv( (empty)

446 self, data: int, flags: int = 0, / 

447 ) -> bytes: # pragma: unless the-annoying-os no cover 

448 if self.handle is None: 6 ctx1adghqr

449 raise ValueError(self._IO_ON_CLOSED_FILE) 2 ctx1qr

450 del flags 4 ctx1adgh

451 result = bytearray() 4 ctx1adgh

452 read_count = DWORD(0) 4 ctx1adgh

453 buffer = (ctypes.c_char * 65536)() 4 ctx1adgh

454 while data > 0: 4 ctx1adgh

455 block_size = min(max(0, data), 65536) 4 ctx1adgh

456 wait_event = CreateEvent(None, BOOL(True), BOOL(False), None) # noqa: FBT003 4 ctx1adgh

457 try: 4 ctx1adgh

458 overlapped_struct = OVERLAPPED(hEvent=wait_event) 4 ctx1adgh

459 success = ReadFile( 4 ctx1adgh

460 self.handle, 

461 ctypes.cast(ctypes.byref(buffer), ctypes.c_void_p), 

462 DWORD(block_size), 

463 ctypes.cast(ctypes.byref(read_count), LPDWORD), 

464 ctypes.cast(ctypes.byref(overlapped_struct), LPOVERLAPPED), 

465 ) 

466 if ( 4 ctx1adgh

467 not success and GetLastError() == ERROR_IO_PENDING 

468 ): # pragma: no cover [external] 

469 success = GetOverlappedResult( 4 ctx1adgh

470 self.handle, 

471 ctypes.cast( 

472 ctypes.byref(overlapped_struct), LPOVERLAPPED 

473 ), 

474 ctypes.cast(ctypes.byref(read_count), LPDWORD), 

475 BOOL(True), # noqa: FBT003 

476 ) 

477 if ( 4 ctx1adgh

478 not success or read_count.value == 0 

479 ): # pragma: no cover [external] 

480 raise WinError() 

481 finally: 

482 CloseHandle(wait_event) 4 ctx1adgh

483 result.extend(buffer.raw[:block_size]) 4 ctx1adgh

484 data -= read_count.value 4 ctx1adgh

485 read_count.value = 0 4 ctx1adgh

486 return bytes(result) 4 ctx1adgh

487 

488 def sendall( (empty)

489 self, data: Buffer, flags: int = 0, / 

490 ) -> None: # pragma: unless the-annoying-os no cover 

491 if self.handle is None: 6 ctx1adghqr

492 raise ValueError(self._IO_ON_CLOSED_FILE) 2 ctx1qr

493 del flags 4 ctx1adgh

494 data = bytes(data) 4 ctx1adgh

495 databuf = (ctypes.c_char * len(data))() 4 ctx1adgh

496 for i, x in enumerate(data): 4 ctx1adgh

497 databuf[i] = ctypes.c_char(x) 4 ctx1adgh

498 write_count = DWORD(0) 4 ctx1adgh

499 wait_event = CreateEvent(None, BOOL(True), BOOL(False), None) # noqa: FBT003 4 ctx1adgh

500 try: 4 ctx1adgh

501 overlapped_struct = OVERLAPPED(hEvent=wait_event) 4 ctx1adgh

502 success = WriteFile( 4 ctx1adgh

503 self.handle, 

504 ctypes.cast(ctypes.byref(databuf), ctypes.c_void_p), 

505 DWORD(len(data)), 

506 ctypes.cast(ctypes.byref(write_count), LPDWORD), 

507 ctypes.cast(ctypes.byref(overlapped_struct), LPOVERLAPPED), 

508 ) 

509 if ( 4 ctx1adgh

510 not success and GetLastError() == ERROR_IO_PENDING 

511 ): # pragma: no cover [external] 

512 success = GetOverlappedResult( 

513 self.handle, 

514 ctypes.cast(ctypes.byref(overlapped_struct), LPOVERLAPPED), 

515 ctypes.cast(ctypes.byref(write_count), LPDWORD), 

516 BOOL(True), # noqa: FBT003 

517 ) 

518 if ( 4 ctx1adgh

519 not success or write_count.value == 0 

520 ): # pragma: no cover [external] 

521 raise WinError() 

522 finally: 

523 CloseHandle(wait_event) 4 ctx1adgh

524 

525 def named_pipe_name( (empty)

526 self, 

527 ) -> str: # pragma: unless the-annoying-os no cover 

528 """Return the named pipe name this socket is connected to.""" 

529 return self._name (empty)

530 

531 @classmethod (empty)

532 def for_openssh(cls) -> Self: # pragma: unless the-annoying-os no cover (empty)

533 """Construct a named pipe for use with OpenSSH on The Annoying OS. 

534 

535 Returns: 

536 A new named pipe handle, using OpenSSH's pipe name. 

537 

538 """ 

539 return cls(f"{PIPE_PREFIX}openssh-ssh-agent") 3 ctx1abc

540 

541 @classmethod (empty)

542 def for_pageant(cls) -> Self: # pragma: unless the-annoying-os no cover (empty)

543 """Construct a named pipe for use with Pageant. 

544 

545 Returns: 

546 A new named pipe handle, using Pageant's pipe name. 

547 

548 """ 

549 return cls(cls.pageant_named_pipe_name()) 3 ctx1abc

550 

551 @staticmethod (empty)

552 def pageant_named_pipe_name( (empty)

553 *, 

554 require_cryptprotectmemory: bool = False, 

555 ) -> str: # pragma: unless the-annoying-os no cover 

556 """Return the pipe name that Pageant on The Annoying OS would use. 

557 

558 Args: 

559 require_cryptprotectmemory: 

560 Pageant normally attempts to use the 

561 `CryptProtectMemory` system function from the 

562 `crypt32.dll` library on The Annoying OS, ignoring any 

563 errors. This in turn influences the resulting named 

564 pipe's filename. 

565 

566 If true, and if we fail to call `CryptProtectMemory`, we 

567 abort; otherwise we ignore any errors from calling or 

568 failing to call `CryptProtectMemory`. 

569 

570 Raises: 

571 NotImplementedError: 

572 `require_cryptprotectmemory` was `True`, but this Python 

573 version cannot call `CryptProtectMemory` due to lack of 

574 system support. 

575 

576 """ 

577 realname = b"Pageant\x00" 3 ctx1abc

578 num_blocks = ( 3 ctx1abc

579 len(realname) + CRYPTPROTECTMEMORY_BLOCK_SIZE - 1 

580 ) // CRYPTPROTECTMEMORY_BLOCK_SIZE 

581 assert num_blocks == 1 3 ctx1abc

582 buf_decl = ctypes.c_char * (CRYPTPROTECTMEMORY_BLOCK_SIZE * num_blocks) 3 ctx1abc

583 buf = buf_decl() 3 ctx1abc

584 for i, x in enumerate(realname): 3 ctx1abc

585 buf[i] = ctypes.c_char(x) 3 ctx1abc

586 try: 3 ctx1abc

587 CryptProtectMemory( 3 ctx1abc

588 ctypes.cast(ctypes.byref(buf), ctypes.c_void_p), 

589 DWORD(len(buf.raw)), 

590 DWORD(CRYPTPROTECTMEMORY_CROSS_PROCESS), 

591 ) 

592 except NotImplementedError: # pragma: no cover [external] 

593 if require_cryptprotectmemory: 

594 raise 

595 buf_as_ssh_string = bytearray() 3 ctx1abc

596 buf_as_ssh_string.extend(int.to_bytes(len(buf.raw), 4, "big")) 3 ctx1abc

597 buf_as_ssh_string.extend(buf.raw) 3 ctx1abc

598 digest = hashlib.sha256(buf_as_ssh_string).hexdigest() 3 ctx1abc

599 return f"{PIPE_PREFIX}pageant.{os.getlogin()}.{digest}" 3 ctx1abc

600 

601 

602class _WindowsNamedPipeSocketAddress(enum.Enum): (empty)

603 """Internal enum for the socket address for Windows named pipes. 

604 

605 Attributes: 

606 PAGEANT: The dynamic address for PuTTY/Pageant. 

607 OPENSSH: The static address for OpenSSH-on-Windows. 

608 

609 """ 

610 

611 PAGEANT = "PuTTY/Pageant on The Annoying OS" (empty)

612 """""" (empty)

613 OPENSSH = "OpenSSH on The Annoying OS" (empty)

614 """""" (empty)

615 

616 

617class SocketProvider: (empty)

618 """Static functionality for providing sockets.""" 

619 

620 @staticmethod (empty)

621 def unix_domain_ssh_auth_sock(*, timeout: int = 125) -> socket.socket: (empty)

622 """Return a UNIX domain socket connected to `SSH_AUTH_SOCK`. 

623 

624 Args: 

625 timeout: 

626 A connection timeout for the SSH agent. Only used for 

627 "true" sockets, and only if the socket is not yet connected. 

628 The default value gives ample time for agent connections 

629 forwarded via SSH on high-latency networks (e.g. Tor). 

630 

631 Returns: 

632 A connected UNIX domain socket. 

633 

634 Raises: 

635 KeyError: 

636 The `SSH_AUTH_SOCK` environment variable was not found. 

637 UnixDomainSocketsNotAvailableError: 

638 This Python version does not support UNIX domain 

639 sockets, necessary to automatically connect to 

640 a running SSH agent via the `SSH_AUTH_SOCK` environment 

641 variable. 

642 OSError: 

643 There was an error setting up a socket connection to the 

644 agent. 

645 

646 """ 

647 if not hasattr(socket, "AF_UNIX"): 11 ctx1asbcdtefiuv

648 msg = "This Python version does not support UNIX domain sockets." 5 ctx1sbctu

649 raise UnixDomainSocketsNotAvailableError(msg) 5 ctx1sbctu

650 else: # noqa: RET506 # pragma: unless posix no cover 

651 sock = socket.socket(family=socket.AF_UNIX) 8 ctx1abcdefiv

652 if "SSH_AUTH_SOCK" not in os.environ: 8 ctx1abcdefiv

653 msg = "SSH_AUTH_SOCK environment variable" 1 ctx1v

654 raise KeyError(msg) 1 ctx1v

655 ssh_auth_sock = os.environ["SSH_AUTH_SOCK"] 7 ctx1abcdefi

656 sock.settimeout(timeout) 7 ctx1abcdefi

657 sock.connect(ssh_auth_sock) 7 ctx1abcdefi

658 return sock 6 ctx1abcdef

659 

660 @staticmethod (empty)

661 def _windows_named_pipe( (empty)

662 pipe_name: _WindowsNamedPipeSocketAddress | str | None, 

663 ) -> WindowsNamedPipeHandle: 

664 r"""Return a socket wrapper around a Windows named pipe. 

665 

666 Args: 

667 pipe_name: 

668 The named pipe's name, or a well-known named pipe socket 

669 address, or `None` to look up the true name in 

670 `SSH_AUTH_SOCK`. In the latter case, the `SSH_AUTH_SOCK` 

671 environment variable is assumed to contain a valid named 

672 pipe name, i.e., a path starting with `\\.\pipe\` or 

673 `//./pipe/`. 

674 

675 Raises: 

676 KeyError: 

677 The `SSH_AUTH_SOCK` environment variable was not found. 

678 OSError: 

679 There was an error setting up a connection to the agent. 

680 ValueError: 

681 The path clearly does not name a valid named pipe name. 

682 WindowsNamedPipesNotAvailableError: 

683 This Python version does not support Windows named 

684 pipes. 

685 

686 """ 

687 # We must raise WindowsNamedPipesNotAvailableError in preference 

688 # to other errors, so we must duplicate the system support check 

689 # here (even though WindowsNamedPipeHandle would raise as well). 

690 if not hasattr(ctypes, "WinDLL"): 11 ctx1asbcdtefiuo

691 msg = "This Python version does not support Windows named pipes." 5 ctx1sbctu

692 raise WindowsNamedPipesNotAvailableError(msg) 5 ctx1sbctu

693 else: # noqa: RET506 # pragma: unless the-annoying-os no cover 

694 # TODO(the-13th-letter): Rewrite using structural pattern 

695 # matching (3 cases). 

696 # https://the13thletter.info/derivepassphrase/latest/pycompatibility/#after-eol-py3.9 

697 if pipe_name == _WindowsNamedPipeSocketAddress.PAGEANT: 8 ctx1abcdefio

698 return WindowsNamedPipeHandle.for_pageant() 3 ctx1abc

699 if pipe_name == _WindowsNamedPipeSocketAddress.OPENSSH: 8 ctx1abcdefio

700 return WindowsNamedPipeHandle.for_openssh() 3 ctx1abc

701 if pipe_name is None: # pragma: no branch 8 ctx1abcdefio

702 pipe_name = os.environ["SSH_AUTH_SOCK"] 8 ctx1abcdefio

703 if not pipe_name.replace("/", "\\").startswith(PIPE_PREFIX): 5 ctx1defio

704 msg = f"Invalid named pipe name: {pipe_name!r}" 1 ctx1o

705 raise ValueError(msg) 1 ctx1o

706 return WindowsNamedPipeHandle(pipe_name) 4 ctx1defi

707 

708 @classmethod (empty)

709 def windows_named_pipe_for_pageant(cls) -> WindowsNamedPipeHandle: (empty)

710 """Return a socket wrapper connected to Pageant on The Annoying OS. 

711 

712 Raises: 

713 OSError: 

714 There was an error setting up a connection to Pageant. 

715 WindowsNamedPipesNotAvailableError: 

716 This Python version does not support Windows named 

717 pipes. 

718 

719 """ 

720 return cls._windows_named_pipe(_WindowsNamedPipeSocketAddress.PAGEANT) 3 ctx1abc

721 

722 @classmethod (empty)

723 def windows_named_pipe_for_openssh(cls) -> WindowsNamedPipeHandle: (empty)

724 """Return a socket wrapper connected to the OpenSSH agent on The Annoying OS. 

725 

726 Raises: 

727 OSError: 

728 There was an error setting up a connection to the OpenSSH 

729 agent. 

730 WindowsNamedPipesNotAvailableError: 

731 This Python version does not support Windows named 

732 pipes. 

733 

734 """ # noqa: E501 

735 return cls._windows_named_pipe(_WindowsNamedPipeSocketAddress.OPENSSH) 3 ctx1abc

736 

737 @classmethod (empty)

738 def windows_named_pipe_ssh_auth_sock(cls) -> WindowsNamedPipeHandle: (empty)

739 r"""Return a socket wrapper connected to the agent in `SSH_AUTH_SOCK`. 

740 

741 The `SSH_AUTH_SOCK` environment variable is assumed to contain 

742 a valid named pipe name, i.e., a path starting with `\\.\pipe\` 

743 or `//./pipe/`. 

744 

745 Raises: 

746 KeyError: 

747 The `SSH_AUTH_SOCK` environment variable was not found. 

748 ValueError: 

749 The path in `SSH_AUTH_SOCK` clearly does not name a valid 

750 named pipe name. 

751 OSError: 

752 There was an error setting up a connection to the OpenSSH 

753 agent. 

754 WindowsNamedPipesNotAvailableError: 

755 This Python version does not support Windows named 

756 pipes. 

757 

758 """ 

759 return cls._windows_named_pipe(None) 11 ctx1asbcdtefiuo

760 

761 registry: ClassVar[ (empty)

762 dict[str, _types.SSHAgentSocketProvider | str | None] 

763 ] = {} 

764 """A dictionary of callables that provide SSH agent sockets. (empty)

765 

766 Each entry in the dictionary points either to a callable, a string, 

767 or `None`: if a callable, then that callable returns a socket; if 

768 a string, then this entry is an alias for that other entry; if 

769 `None`, then the entry name is merely registered, but no 

770 implementation is available. 

771 

772 If a callable is not applicable to this platform, Python 

773 installation or `derivepassphrase` installation, then it MUST raise 

774 [`NotImplementedError`][]. Conversely, if the callable returns 

775 a value, or raises any other kind of exception, then the caller MAY 

776 assume that this platform, Python installation and 

777 `derivepassphrase` installation are sufficient for the callable to 

778 be able to return a working socket on this platform. (The latter 

779 may still be dependent on further, external circumstances, such as 

780 required configuration settings, or environment variables, or 

781 sufficient system resources, etc.) 

782 

783 (Interpretation of "MUST" and "MAY" as per IETF Best Current 

784 Practice #14; see [RFC 2119][] and [RFC 8174][].) 

785 

786 [RFC 2119]: https://www.rfc-editor.org/info/rfc2119 

787 [RFC 8174]: https://www.rfc-editor.org/info/rfc8174 

788 

789 """ 

790 

791 @classmethod (empty)

792 def register( (empty)

793 cls, 

794 name: str, 

795 *aliases: str, 

796 ) -> Callable[[SSHAgentSocketProviderT], SSHAgentSocketProviderT]: 

797 """Register a callable as an SSH agent socket provider (decorator). 

798 

799 Attempting to re-register an existing alias, or a name with an 

800 implementation, with a different implementation is an error. 

801 

802 Args: 

803 name: 

804 The principal name under which to register the passed 

805 callable. 

806 aliases: 

807 Alternate names to register as aliases for the principal 

808 name. 

809 

810 Returns: 

811 A decorator implementing the above. 

812 

813 """ 

814 

815 def decorator(f: SSHAgentSocketProviderT) -> SSHAgentSocketProviderT: 5 ctx1alpnm

816 """Register a callable as an SSH agent socket provider. 

817 

818 Attempting to re-register an existing alias, or a name with 

819 an implementation, with a different implementation is an 

820 error. 

821 

822 Args: 

823 f: The callable to decorate/register. 

824 

825 Returns: 

826 The callable. 

827 

828 Raises: 

829 ValueError: 

830 The name or alias is already in use. 

831 

832 """ 

833 for alias in [name, *aliases]: 5 ctx1alpnm

834 try: 5 ctx1alpnm

835 existing = cls.lookup(alias) 5 ctx1alpnm

836 except (NoSuchProviderError, NotImplementedError): 3 ctx1lnm

837 cls.registry[alias] = f if alias == name else name 3 ctx1lnm

838 else: 

839 if existing is None: 4 ctx1alpm

840 cls.registry[alias] = f if alias == name else name (empty)

841 elif existing != f: 3 ctx1lpm

842 msg = ( 1 ctx1p

843 f"The SSH agent socket provider {alias!r} " 

844 f"is already registered." 

845 ) 

846 raise ValueError(msg) 1 ctx1p

847 return f 4 ctx1alnm

848 

849 return decorator 5 ctx1alpnm

850 

851 @classmethod (empty)

852 def lookup( (empty)

853 cls, provider: _types.SSHAgentSocketProvider | str | None, / 

854 ) -> _types.SSHAgentSocketProvider | None: 

855 """Look up a socket provider entry. 

856 

857 Args: 

858 provider: The provider to look up. 

859 

860 Returns: 

861 The callable indicated by this provider, if it is 

862 implemented, or `None`, if it is merely registered. 

863 

864 Raises: 

865 NoSuchProviderError: 

866 The provider is not registered. 

867 

868 """ 

869 ret = provider 36 ctx1alsbcdtefAOBCDEFGHIiuzvjpnmwxJKLyMoN

870 while isinstance(ret, str): 36 ctx1alsbcdtefAOBCDEFGHIiuzvjpnmwxJKLyMoN

871 try: 36 ctx1alsbcdtefAOBCDEFGHIiuzvjpnmwxJKLyMoN

872 ret = cls.registry[ret] 36 ctx1alsbcdtefAOBCDEFGHIiuzvjpnmwxJKLyMoN

873 except KeyError as exc: 5 ctx1lOnmx

874 raise NoSuchProviderError(ret) from exc 5 ctx1lOnmx

875 return ret 35 ctx1alsbcdtefABCDEFGHIiuzvjpnmwxJKLyMoN

876 

877 @classmethod (empty)

878 def resolve( (empty)

879 cls, provider: _types.SSHAgentSocketProvider | str | None, / 

880 ) -> _types.SSHAgentSocketProvider: 

881 """Resolve a socket provider to a proper callable. 

882 

883 Args: 

884 provider: The provider to resolve. 

885 

886 Returns: 

887 The callable indicated by this provider. 

888 

889 Raises: 

890 NoSuchProviderError: 

891 The provider is not registered. 

892 NotImplementedError: 

893 The provider is registered, but is not functional or not 

894 applicable to this `derivepassphrase` installation. 

895 

896 """ 

897 ret = cls.lookup(provider) 36 ctx1alsbcdtefAOBCDEFGHIiuzvjpnmwxJKLyMoN

898 if ret is None: 35 ctx1alsbcdtefABCDEFGHIiuzvjpnmwxJKLyMoN

899 msg = ( 3 ctx1zwy

900 f"The {ret!r} socket provider is not functional on or " 

901 "not applicable to this derivepassphrase installation." 

902 ) 

903 raise NotImplementedError(msg) 3 ctx1zwy