Marco Ricci commited on 2026-08-30 18:14:44
Zeige 11 geänderte Dateien mit 514 Einfügungen und 8 Löschungen.
There is a three-part example project for how to write an SSH agent socket provider; the project is a full-fledged Python package in its own right. (One of the three variants is also presented in the README.) There is also a single integration test in the SSH agent socket provider API package that verifies that this example project can be installed (into an isolated virtual environment), and that it correctly exposes the three SSH agent socket providers. Preparing this example package alone and getting the test to pass has already uncovered several misconfigurations and misspecifications in both the API package and the example package. We also set up a separate, reduced pytest configuration for the SSH agent socket provider API package, because otherwise pytest determines the wrong root directory, and the example project package files cannot be correctly installed. We explicitly mention that we currently do not measure coverage, because the test runs inside a temporary virtual environment (which coverage measurement would have to account for, both when measuring and when looking up configuration) and because there are currently no code branches in the API package, so the coverage information is uninteresting if we already know the example project was installable.
| ... | ... |
@@ -31,14 +31,43 @@ The abstract channel must also be a context manager, which closes itself upon le |
| 31 | 31 |
|
| 32 | 32 |
The `SSHAgentSocketProvider` is a callable that returns an `SSHAgentSocket` when called without arguments. |
| 33 | 33 |
|
| 34 |
-To then actually register an SSH agent socket provider, build an `SSHAgentSocketProviderEntry` struct as follows: |
|
| 34 |
+To then actually register an SSH agent socket provider, build an `SSHAgentSocketProviderEntry` struct. |
|
| 35 |
+For example: |
|
| 35 | 36 |
|
| 36 | 37 |
~~~ python |
| 37 |
-def ssh_agent_over_stdin_stdout_provider() -> SSHAgentSocket: ... |
|
| 38 |
- |
|
| 39 |
- |
|
| 40 |
-stdin_stdout_entry_point = SSHAgentSocketProviderEntry( |
|
| 41 |
- provider=ssh_agent_over_stdin_stdout_provider, |
|
| 38 |
+# The class is an SSHAgentSocket and has an empty constructor, so the |
|
| 39 |
+# constructor is a valid SSHAgentSocketProvider. |
|
| 40 |
+class SSHAgentOverStdinStdoutSocket: |
|
| 41 |
+ """Forwarding STDIN/STDOUT, as if connected to an SSH agent.""" |
|
| 42 |
+ |
|
| 43 |
+ FLAGS_ARE_UNSUPPORTED = "flags argument is unsupported" |
|
| 44 |
+ """Common error message.""" |
|
| 45 |
+ |
|
| 46 |
+ def __enter__(self) -> Self: |
|
| 47 |
+ """Return self.""" |
|
| 48 |
+ return self |
|
| 49 |
+ |
|
| 50 |
+ def __exit__(self, *args: object) -> bool | None: |
|
| 51 |
+ """Close stdin/stdout.""" |
|
| 52 |
+ sys.stdin.close() |
|
| 53 |
+ sys.stdout.close() |
|
| 54 |
+ return None |
|
| 55 |
+ |
|
| 56 |
+ def send(self, data: Buffer, flags: int = 0, /) -> None: |
|
| 57 |
+ """Send data to agent.""" |
|
| 58 |
+ if flags: |
|
| 59 |
+ raise ValueError(self.FLAGS_ARE_UNSUPPORTED) |
|
| 60 |
+ sys.stdout.buffer.write(data) |
|
| 61 |
+ |
|
| 62 |
+ def recv(self, bufsize: int, flags: int = 0, /) -> bytes: |
|
| 63 |
+ """Receive data from agent.""" |
|
| 64 |
+ if flags: |
|
| 65 |
+ raise ValueError(self.FLAGS_ARE_UNSUPPORTED) |
|
| 66 |
+ return sys.stdin.buffer.read(bufsize) |
|
| 67 |
+ |
|
| 68 |
+ |
|
| 69 |
+ENTRY_POINT = SSHAgentSocketProviderEntry( |
|
| 70 |
+ provider=SSHAgentOverStdinStdoutSocket, |
|
| 42 | 71 |
key="stdin_stdout", |
| 43 | 72 |
aliases=("stdin", "stdout"),
|
| 44 | 73 |
) |
| ... | ... |
@@ -48,15 +77,17 @@ Then add an appropriate entry point definition in your `pyproject.toml`, using t |
| 48 | 77 |
|
| 49 | 78 |
~~~ toml |
| 50 | 79 |
[project.entry."derivepassphrase.ssh_agent_socket_providers"] |
| 51 |
-first_provider = "mymodule: stdin_stdout_entry_point" |
|
| 80 |
+first_provider = "mymodule: ENTRY_POINT" |
|
| 52 | 81 |
~~~ |
| 53 | 82 |
|
| 54 | 83 |
`derivepassphrase` will then pick up `stdin_stdout` as a new SSH agent socket provider, with aliases `stdin` and `stdout`.[^entry_point_name] |
| 55 | 84 |
|
| 56 | 85 |
[^entry_point_name]: |
| 57 |
- Only the fields in `SSHAgentSocketProviderEntry` matter, not what names are used to declare the entry point. |
|
| 86 |
+ Only the fields in the `SSHAgentSocketProviderEntry` matter, not what names are used to declare the entry point. |
|
| 58 | 87 |
By convention, however, you would choose `stdin_stdout` as the entry point name, not `first_provider`. |
| 59 | 88 |
|
| 89 |
+(See the tests and the test data for two further examples.) |
|
| 90 |
+ |
|
| 60 | 91 |
## License |
| 61 | 92 |
|
| 62 | 93 |
Like `derivepassphrase`, `derivepassphrase-sshagentsocketprovider` is distributed under the terms of the [zlib/libpng license](https://spdx.org/licenses/Zlib.html). |
| ... | ... |
@@ -38,3 +38,17 @@ dependencies = [ |
| 38 | 38 |
Documentation = "https://the13thletter.info/derivepassphrase/" |
| 39 | 39 |
Issues = "https://the13thletter.info/derivepassphrase/latest/wishlist/" |
| 40 | 40 |
Source = "https://git.schokokeks.org/derivepassphrase.git" |
| 41 |
+ |
|
| 42 |
+# Don't attempt to collect coverage data at this point. There is only |
|
| 43 |
+# one integration test, with linear execution, and it creates virtual |
|
| 44 |
+# environments in subprocesses. So it is hard to get coverage installed |
|
| 45 |
+# and configured in the target environment to actually collect data, and |
|
| 46 |
+# even if that works, there is not much data to collect, because there |
|
| 47 |
+# is no branching execution. |
|
| 48 |
+ |
|
| 49 |
+[tool.pytest.ini_options] |
|
| 50 |
+addopts = '--import-mode=importlib' |
|
| 51 |
+testpaths = [ |
|
| 52 |
+ 'tests', |
|
| 53 |
+] |
|
| 54 |
+xfail_strict = true |
| ... | ... |
@@ -0,0 +1,22 @@ |
| 1 |
+This is free and unencumbered software released into the public domain. |
|
| 2 |
+ |
|
| 3 |
+Anyone is free to copy, modify, publish, use, compile, sell, or |
|
| 4 |
+distribute this software, either in source code form or as a compiled |
|
| 5 |
+binary, for any purpose, commercial or non-commercial, and by any means. |
|
| 6 |
+ |
|
| 7 |
+In jurisdictions that recognize copyright laws, the author or authors of |
|
| 8 |
+this software dedicate any and all copyright interest in the software to |
|
| 9 |
+the public domain. We make this dedication for the benefit of the public |
|
| 10 |
+at large and to the detriment of our heirs and successors. We intend |
|
| 11 |
+this dedication to be an overt act of relinquishment in perpetuity of |
|
| 12 |
+all present and future rights to this software under copyright law. |
|
| 13 |
+ |
|
| 14 |
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS |
|
| 15 |
+OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF |
|
| 16 |
+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. |
|
| 17 |
+IN NO EVENT SHALL THE AUTHORS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER |
|
| 18 |
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING |
|
| 19 |
+FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER |
|
| 20 |
+DEALINGS IN THE SOFTWARE. |
|
| 21 |
+ |
|
| 22 |
+For more information, please refer to <http://unlicense.org/> |
| ... | ... |
@@ -0,0 +1,11 @@ |
| 1 |
+# derivepassphrase-sshagentsocketprovider-example |
|
| 2 |
+ |
|
| 3 |
+Three full examples of writing an [SSH agent socket provider][SASP]. |
|
| 4 |
+ |
|
| 5 |
+One example is rather trivial, the other two much less so. |
|
| 6 |
+ |
|
| 7 |
+[SASP]: https://pypi.org/project/derivepassphrase-sshagentsocketprovider/ |
|
| 8 |
+ |
|
| 9 |
+## License |
|
| 10 |
+ |
|
| 11 |
+This example code is distributed under the terms of the [Unlicense](https://spdx.org/licenses/Unlicense.html). |
| ... | ... |
@@ -0,0 +1,44 @@ |
| 1 |
+[build-system] |
|
| 2 |
+requires = ["hatchling"] |
|
| 3 |
+build-backend = "hatchling.build" |
|
| 4 |
+ |
|
| 5 |
+[project] |
|
| 6 |
+name = "derivepassphrase-sshagentsocketprovider-example" |
|
| 7 |
+description = 'An example SSH agent socket provider for derivepassphrase.' |
|
| 8 |
+readme = "README.md" |
|
| 9 |
+version = "0.0.0" |
|
| 10 |
+requires-python = ">= 3.9" |
|
| 11 |
+license = "Unlicense" |
|
| 12 |
+keywords = [] |
|
| 13 |
+authors = [ |
|
| 14 |
+ { name = "Marco Ricci", email = "software@the13thletter.info" },
|
|
| 15 |
+] |
|
| 16 |
+classifiers = [ |
|
| 17 |
+ "Development Status :: 4 - Beta", |
|
| 18 |
+ "Intended Audience :: Developers", |
|
| 19 |
+ "Operating System :: OS Independent", |
|
| 20 |
+ "Private :: Do Not Upload", |
|
| 21 |
+ "Programming Language :: Python :: 3", |
|
| 22 |
+ "Programming Language :: Python :: 3.9", |
|
| 23 |
+ "Programming Language :: Python :: 3.10", |
|
| 24 |
+ "Programming Language :: Python :: 3.11", |
|
| 25 |
+ "Programming Language :: Python :: 3.12", |
|
| 26 |
+ "Programming Language :: Python :: 3.13", |
|
| 27 |
+ "Programming Language :: Python :: 3.14", |
|
| 28 |
+ "Programming Language :: Python :: 3.15", |
|
| 29 |
+ "Programming Language :: Python :: Implementation :: CPython", |
|
| 30 |
+ "Programming Language :: Python :: Implementation :: PyPy", |
|
| 31 |
+ "Topic :: Software Development :: Libraries", |
|
| 32 |
+ "Typing :: Typed", |
|
| 33 |
+] |
|
| 34 |
+dependencies = [ |
|
| 35 |
+ "derivepassphrase-sshagentsocketprovider >= 1.0a1", |
|
| 36 |
+ "typing-extensions", |
|
| 37 |
+] |
|
| 38 |
+ |
|
| 39 |
+[project.urls] |
|
| 40 |
+ |
|
| 41 |
+[project.entry-points."derivepassphrase.ssh_agent_socket_providers"] |
|
| 42 |
+stdin_stdout = "derivepassphrase_sshagentsocketprovider_example.stdin_stdout: ENTRY_POINT" |
|
| 43 |
+failure_response = "derivepassphrase_sshagentsocketprovider_example.failure_response: ENTRY_POINT" |
|
| 44 |
+always_fail = "derivepassphrase_sshagentsocketprovider_example.always_fail: ENTRY_POINT" |
| ... | ... |
@@ -0,0 +1,23 @@ |
| 1 |
+# SPDX-FileCopyrightText: 2026 Marco Ricci <software@the13thletter.info> |
|
| 2 |
+# |
|
| 3 |
+# SPDX-License-Identifier: Unlicense |
|
| 4 |
+ |
|
| 5 |
+"""Example SSH agent providers. |
|
| 6 |
+ |
|
| 7 |
+We package three examples: |
|
| 8 |
+ |
|
| 9 |
+ 1. The `stdin_stdout` example is the simplest useful one. The provider |
|
| 10 |
+ exposes a virtual socket such that reading from the virtual socket |
|
| 11 |
+ reads from `sys.stdin`, and writing to the virtual socket writes to |
|
| 12 |
+ `sys.stdout`. |
|
| 13 |
+ 2. The `failure_response` example is a somewhat more complicated, but |
|
| 14 |
+ also more useful one. The provider creates a pseudo socket which |
|
| 15 |
+ answers every request with an `SSH_AGENT_FAILURE` response. |
|
| 16 |
+ 3. The `always_fail` example is very simple, but not very useful. The |
|
| 17 |
+ provider always fails to construct a socket. |
|
| 18 |
+ |
|
| 19 |
+See the respective submodule for the code. |
|
| 20 |
+ |
|
| 21 |
+""" |
|
| 22 |
+ |
|
| 23 |
+__version__ = "0.0.0" |
| ... | ... |
@@ -0,0 +1,26 @@ |
| 1 |
+# SPDX-FileCopyrightText: 2026 Marco Ricci <software@the13thletter.info> |
|
| 2 |
+# |
|
| 3 |
+# SPDX-License-Identifier: Unlicense |
|
| 4 |
+ |
|
| 5 |
+"""The `always_fail` example SSH agent provider. |
|
| 6 |
+ |
|
| 7 |
+The provider always fails to construct a socket. |
|
| 8 |
+ |
|
| 9 |
+""" |
|
| 10 |
+ |
|
| 11 |
+from __future__ import annotations |
|
| 12 |
+ |
|
| 13 |
+import derivepassphrase_sshagentsocketprovider as d_sasp |
|
| 14 |
+ |
|
| 15 |
+ |
|
| 16 |
+# Very simple, but uninteresting example. |
|
| 17 |
+def always_fail_socket_provider() -> d_sasp.SSHAgentSocket: |
|
| 18 |
+ """A provider that always fails to provide an SSH agent socket.""" |
|
| 19 |
+ msg = "Deliberately failing to construct an SSH agent socket." |
|
| 20 |
+ raise RuntimeError(msg) |
|
| 21 |
+ |
|
| 22 |
+ |
|
| 23 |
+assert isinstance(always_fail_socket_provider, d_sasp.SSHAgentSocketProvider) |
|
| 24 |
+ENTRY_POINT = d_sasp.SSHAgentSocketProviderEntry( |
|
| 25 |
+ provider=always_fail_socket_provider, key="always_fail", aliases=("fail",)
|
|
| 26 |
+) |
| ... | ... |
@@ -0,0 +1,157 @@ |
| 1 |
+# SPDX-FileCopyrightText: 2026 Marco Ricci <software@the13thletter.info> |
|
| 2 |
+# |
|
| 3 |
+# SPDX-License-Identifier: Unlicense |
|
| 4 |
+ |
|
| 5 |
+"""The `failure_response` example SSH agent provider. |
|
| 6 |
+ |
|
| 7 |
+The provider creates a pseudo socket which answers every request with an |
|
| 8 |
+`SSH_AGENT_FAILURE` response. |
|
| 9 |
+ |
|
| 10 |
+""" |
|
| 11 |
+ |
|
| 12 |
+from __future__ import annotations |
|
| 13 |
+ |
|
| 14 |
+import errno |
|
| 15 |
+import os |
|
| 16 |
+import struct |
|
| 17 |
+import sys |
|
| 18 |
+from typing import TYPE_CHECKING, cast |
|
| 19 |
+ |
|
| 20 |
+import derivepassphrase_sshagentsocketprovider as d_sasp |
|
| 21 |
+ |
|
| 22 |
+if TYPE_CHECKING: |
|
| 23 |
+ from typing_extensions import Buffer, Self |
|
| 24 |
+ |
|
| 25 |
+ |
|
| 26 |
+# Slightly less simple example, mocking the entire SSH agent. |
|
| 27 |
+class FailureSSHAgentSocket: |
|
| 28 |
+ """A pseudo socket that always returns failure responses.""" |
|
| 29 |
+ |
|
| 30 |
+ FLAGS_ARE_UNSUPPORTED = "flags argument is unsupported" |
|
| 31 |
+ """Common error message.""" |
|
| 32 |
+ FAILURE_RESPONSE = b"\x00\x00\x00\x01\x05" |
|
| 33 |
+ """Common protocol response: SSH_AGENT_FAILURE.""" |
|
| 34 |
+ |
|
| 35 |
+ def __init__(self) -> None: |
|
| 36 |
+ """Init self.""" |
|
| 37 |
+ self.closed = False |
|
| 38 |
+ """Track whether the channel is already closed.""" |
|
| 39 |
+ self.send_queue = bytearray() |
|
| 40 |
+ """The queue of bytes from the client, filled by [`send`][].""" |
|
| 41 |
+ self.recv_queue = bytearray() |
|
| 42 |
+ """The queue of bytes to the client, filled by [`recv`][].""" |
|
| 43 |
+ self.header_len_struct = struct.Struct(">I")
|
|
| 44 |
+ """ |
|
| 45 |
+ A struct used in the parsing of requests. |
|
| 46 |
+ Cached for efficiency reasons. |
|
| 47 |
+ """ |
|
| 48 |
+ |
|
| 49 |
+ def __enter__(self) -> Self: |
|
| 50 |
+ """Return self.""" |
|
| 51 |
+ return self |
|
| 52 |
+ |
|
| 53 |
+ def __exit__(self, *args: object) -> bool | None: |
|
| 54 |
+ """Close the pseudo socket.""" |
|
| 55 |
+ self.closed = True |
|
| 56 |
+ return None |
|
| 57 |
+ |
|
| 58 |
+ def send(self, data: Buffer, flags: int = 0, /) -> None: |
|
| 59 |
+ """Send data to agent.""" |
|
| 60 |
+ if self.closed: |
|
| 61 |
+ raise OSError(errno.EBADF, os.strerror(errno.EBADF)) |
|
| 62 |
+ if flags: |
|
| 63 |
+ raise ValueError(self.FLAGS_ARE_UNSUPPORTED) |
|
| 64 |
+ data_view = memoryview(data) |
|
| 65 |
+ self.send_queue.extend(data_view) |
|
| 66 |
+ while self._read_one_request(): |
|
| 67 |
+ self.recv_queue.extend(self.FAILURE_RESPONSE) |
|
| 68 |
+ |
|
| 69 |
+ def _read_one_request(self) -> bool: |
|
| 70 |
+ """Gobble a complete request from the send_queue, if possible. |
|
| 71 |
+ |
|
| 72 |
+ Returns: |
|
| 73 |
+ True if a complete request could be gobbled, else False. |
|
| 74 |
+ |
|
| 75 |
+ """ |
|
| 76 |
+ # Protocol requests are framed as [n: = UINT32, PAYLOAD[n]], |
|
| 77 |
+ # i.e. a length indicator (as uint32), then a bytes payload of |
|
| 78 |
+ # that length. A request is thus incomplete if and only if it |
|
| 79 |
+ # is shorter than 4 bytes (so the UINT32 doesn't fit) or the |
|
| 80 |
+ # payload is shorter than the declared length. |
|
| 81 |
+ header_size = self.header_len_struct.size |
|
| 82 |
+ if len(self.send_queue) < header_size: |
|
| 83 |
+ return False |
|
| 84 |
+ destructured = cast( |
|
| 85 |
+ "tuple[int]", self.header_len_struct.unpack_from(self.send_queue) |
|
| 86 |
+ ) |
|
| 87 |
+ payload_size = destructured[0] |
|
| 88 |
+ if len(self.send_queue) < header_size + payload_size: |
|
| 89 |
+ return False |
|
| 90 |
+ |
|
| 91 |
+ # The queue contains at least header_size + payload_size bytes, |
|
| 92 |
+ # so it contains a full request that can be trimmed/gobbled. |
|
| 93 |
+ del self.send_queue[: header_size + payload_size] |
|
| 94 |
+ return True |
|
| 95 |
+ |
|
| 96 |
+ def recv(self, bufsize: int, flags: int = 0, /) -> bytes: |
|
| 97 |
+ """Receive data from agent.""" |
|
| 98 |
+ if self.closed: |
|
| 99 |
+ raise OSError(errno.EBADF, os.strerror(errno.EBADF)) |
|
| 100 |
+ if flags: |
|
| 101 |
+ raise ValueError(self.FLAGS_ARE_UNSUPPORTED) |
|
| 102 |
+ return self.take_bytes(self.recv_queue, bufsize) |
|
| 103 |
+ |
|
| 104 |
+ @staticmethod |
|
| 105 |
+ def take_bytes(array: bytearray, n: int | None = None, /) -> bytes: |
|
| 106 |
+ """Implementation of `bytearray.take_bytes(n)` from Python 3.15. |
|
| 107 |
+ |
|
| 108 |
+ Provided for compatibility with older Pythons. |
|
| 109 |
+ |
|
| 110 |
+ Args: |
|
| 111 |
+ array: |
|
| 112 |
+ The bytearray to take bytes from. |
|
| 113 |
+ n: |
|
| 114 |
+ The count of bytes to take from the bytearray. If |
|
| 115 |
+ out-of-bounds (when read as an array index), raise |
|
| 116 |
+ `IndexError`. Otherwise, if positive, then take that |
|
| 117 |
+ many bytes from the start, or if negative, then leave |
|
| 118 |
+ the last `abs(n)` many bytes in the array, and take the |
|
| 119 |
+ rest. |
|
| 120 |
+ |
|
| 121 |
+ Returns: |
|
| 122 |
+ A portion of the original bytearray, as a bytes object. The |
|
| 123 |
+ original bytearray will have that section of bytes removed. |
|
| 124 |
+ |
|
| 125 |
+ Raises: |
|
| 126 |
+ IndexError: |
|
| 127 |
+ The index `n` is invalid. |
|
| 128 |
+ |
|
| 129 |
+ Note: |
|
| 130 |
+ The compatibility implementation makes no attempt to be a |
|
| 131 |
+ zero-copy operation. |
|
| 132 |
+ |
|
| 133 |
+ """ |
|
| 134 |
+ if sys.version_info >= (3, 15): |
|
| 135 |
+ return array.take_bytes(n) |
|
| 136 |
+ if n is None: |
|
| 137 |
+ result = bytes(array) |
|
| 138 |
+ array.clear() |
|
| 139 |
+ return result |
|
| 140 |
+ n2 = n + len(array) if n < 0 else n |
|
| 141 |
+ if not (0 <= n2 < len(array)): |
|
| 142 |
+ msg = ( |
|
| 143 |
+ f"Index {n} is out of range for length {len(array)} bytearray"
|
|
| 144 |
+ ) |
|
| 145 |
+ raise IndexError(msg) |
|
| 146 |
+ result = bytes(array[:n]) |
|
| 147 |
+ del array[:n] |
|
| 148 |
+ return result |
|
| 149 |
+ |
|
| 150 |
+ |
|
| 151 |
+FAILURE_SSH_AGENT_PROVIDER = FailureSSHAgentSocket |
|
| 152 |
+assert isinstance(FAILURE_SSH_AGENT_PROVIDER, d_sasp.SSHAgentSocketProvider) |
|
| 153 |
+ENTRY_POINT = d_sasp.SSHAgentSocketProviderEntry( |
|
| 154 |
+ provider=FAILURE_SSH_AGENT_PROVIDER, |
|
| 155 |
+ key="failure_agent", |
|
| 156 |
+ aliases=(), |
|
| 157 |
+) |
| ... | ... |
@@ -0,0 +1,62 @@ |
| 1 |
+# SPDX-FileCopyrightText: 2026 Marco Ricci <software@the13thletter.info> |
|
| 2 |
+# |
|
| 3 |
+# SPDX-License-Identifier: Unlicense |
|
| 4 |
+ |
|
| 5 |
+"""The `stdin_stdout` example SSH agent provider. |
|
| 6 |
+ |
|
| 7 |
+The provider exposes a virtual socket such that reading from the virtual |
|
| 8 |
+socket reads from `sys.stdin`, and writing to the virtual socket writes |
|
| 9 |
+to `sys.stdout`. |
|
| 10 |
+ |
|
| 11 |
+""" |
|
| 12 |
+ |
|
| 13 |
+from __future__ import annotations |
|
| 14 |
+ |
|
| 15 |
+import sys |
|
| 16 |
+from typing import TYPE_CHECKING |
|
| 17 |
+ |
|
| 18 |
+import derivepassphrase_sshagentsocketprovider as d_sasp |
|
| 19 |
+ |
|
| 20 |
+if TYPE_CHECKING: |
|
| 21 |
+ from typing_extensions import Buffer, Self |
|
| 22 |
+ |
|
| 23 |
+ |
|
| 24 |
+# Simple example, forwarding access to certain file-like objects. |
|
| 25 |
+class SSHAgentOverStdinStdoutSocket: |
|
| 26 |
+ """Forwarding STDIN/STDOUT, as if connected to an SSH agent.""" |
|
| 27 |
+ |
|
| 28 |
+ FLAGS_ARE_UNSUPPORTED = "flags argument is unsupported" |
|
| 29 |
+ """Common error message.""" |
|
| 30 |
+ |
|
| 31 |
+ def __enter__(self) -> Self: |
|
| 32 |
+ """Return self.""" |
|
| 33 |
+ return self |
|
| 34 |
+ |
|
| 35 |
+ def __exit__(self, *args: object) -> bool | None: |
|
| 36 |
+ """Close stdin/stdout.""" |
|
| 37 |
+ sys.stdin.close() |
|
| 38 |
+ sys.stdout.close() |
|
| 39 |
+ return None |
|
| 40 |
+ |
|
| 41 |
+ def send(self, data: Buffer, flags: int = 0, /) -> None: |
|
| 42 |
+ """Send data to agent.""" |
|
| 43 |
+ if flags: |
|
| 44 |
+ raise ValueError(self.FLAGS_ARE_UNSUPPORTED) |
|
| 45 |
+ sys.stdout.buffer.write(data) |
|
| 46 |
+ |
|
| 47 |
+ def recv(self, bufsize: int, flags: int = 0, /) -> bytes: |
|
| 48 |
+ """Receive data from agent.""" |
|
| 49 |
+ if flags: |
|
| 50 |
+ raise ValueError(self.FLAGS_ARE_UNSUPPORTED) |
|
| 51 |
+ return sys.stdin.buffer.read(bufsize) |
|
| 52 |
+ |
|
| 53 |
+ |
|
| 54 |
+SSH_AGENT_OVER_STDIN_STDOUT_PROVIDER = SSHAgentOverStdinStdoutSocket |
|
| 55 |
+assert isinstance( |
|
| 56 |
+ SSH_AGENT_OVER_STDIN_STDOUT_PROVIDER, d_sasp.SSHAgentSocketProvider |
|
| 57 |
+) |
|
| 58 |
+ENTRY_POINT = d_sasp.SSHAgentSocketProviderEntry( |
|
| 59 |
+ provider=SSH_AGENT_OVER_STDIN_STDOUT_PROVIDER, |
|
| 60 |
+ key="stdin_stdout", |
|
| 61 |
+ aliases=("stdin", "stdout"),
|
|
| 62 |
+) |
| ... | ... |
@@ -0,0 +1,116 @@ |
| 1 |
+# SPDX-FileCopyrightText: 2026 Marco Ricci <software@the13thletter.info> |
|
| 2 |
+# |
|
| 3 |
+# SPDX-License-Identifier: Zlib |
|
| 4 |
+ |
|
| 5 |
+from __future__ import annotations |
|
| 6 |
+ |
|
| 7 |
+import subprocess |
|
| 8 |
+import sys |
|
| 9 |
+from typing import TYPE_CHECKING |
|
| 10 |
+ |
|
| 11 |
+import pytest |
|
| 12 |
+ |
|
| 13 |
+if TYPE_CHECKING: |
|
| 14 |
+ import pathlib |
|
| 15 |
+ |
|
| 16 |
+try: |
|
| 17 |
+ import venv |
|
| 18 |
+except ImportError: # pragma: no cover |
|
| 19 |
+ pytest.skip( |
|
| 20 |
+ "Cannot test example package installation without virtualenv support", |
|
| 21 |
+ allow_module_level=True, |
|
| 22 |
+ ) |
|
| 23 |
+ |
|
| 24 |
+ |
|
| 25 |
+@pytest.fixture |
|
| 26 |
+def isolated_venv_python(tmp_path: pathlib.Path) -> pathlib.Path: |
|
| 27 |
+ """Yield the path of the python executable in a newly created venv.""" |
|
| 28 |
+ venv.create(tmp_path / "venv", with_pip=True) |
|
| 29 |
+ path_windows = tmp_path / "venv" / "Scripts" / "Python.exe" |
|
| 30 |
+ path_unix = tmp_path / "venv" / "bin" / "python" |
|
| 31 |
+ return path_windows if sys.platform == "win32" else path_unix |
|
| 32 |
+ |
|
| 33 |
+ |
|
| 34 |
+def is_project(path: pathlib.Path) -> bool: |
|
| 35 |
+ pyproject_toml = path / "pyproject.toml" |
|
| 36 |
+ return pyproject_toml.exists() and pyproject_toml.is_file() |
|
| 37 |
+ |
|
| 38 |
+ |
|
| 39 |
+def test_example_provider( |
|
| 40 |
+ request: pytest.FixtureRequest, isolated_venv_python: pathlib.Path |
|
| 41 |
+) -> None: |
|
| 42 |
+ rootdir = request.config.rootpath |
|
| 43 |
+ if not is_project(rootdir): |
|
| 44 |
+ pytest.skip( |
|
| 45 |
+ "Cannot install this project into a virtualenv " |
|
| 46 |
+ "without access to the project's pyproject.toml, " |
|
| 47 |
+ "but could not find any pyproject.toml at the pytest rootdir." |
|
| 48 |
+ ) |
|
| 49 |
+ example_provider_project = ( |
|
| 50 |
+ rootdir / "tests" / "fixtures" / "example-provider" |
|
| 51 |
+ ) |
|
| 52 |
+ assert is_project(example_provider_project) |
|
| 53 |
+ |
|
| 54 |
+ install_step = subprocess.run( |
|
| 55 |
+ [ |
|
| 56 |
+ str(isolated_venv_python), |
|
| 57 |
+ "-m", |
|
| 58 |
+ "pip", |
|
| 59 |
+ "-q", |
|
| 60 |
+ "--", |
|
| 61 |
+ "install", |
|
| 62 |
+ str(rootdir), |
|
| 63 |
+ str(example_provider_project), |
|
| 64 |
+ ], |
|
| 65 |
+ check=False, |
|
| 66 |
+ capture_output=True, |
|
| 67 |
+ text=True, |
|
| 68 |
+ ) |
|
| 69 |
+ # Pass on output for debugging, *then* check the return code. |
|
| 70 |
+ sys.stdout.write(install_step.stdout) |
|
| 71 |
+ sys.stderr.write(install_step.stderr) |
|
| 72 |
+ install_step.check_returncode() |
|
| 73 |
+ |
|
| 74 |
+ check_file = isolated_venv_python.parent / "check.py" |
|
| 75 |
+ check_file.write_text(r""" |
|
| 76 |
+import importlib.metadata |
|
| 77 |
+ |
|
| 78 |
+import derivepassphrase_sshagentsocketprovider as d_sasp |
|
| 79 |
+from derivepassphrase_sshagentsocketprovider_example import stdin_stdout, failure_response, always_fail |
|
| 80 |
+ |
|
| 81 |
+expected_providers = [ |
|
| 82 |
+ stdin_stdout.SSHAgentOverStdinStdoutSocket, |
|
| 83 |
+ failure_response.FailureSSHAgentSocket, |
|
| 84 |
+ always_fail.always_fail_socket_provider, |
|
| 85 |
+] |
|
| 86 |
+n = len(expected_providers) |
|
| 87 |
+ |
|
| 88 |
+entry_points = list( |
|
| 89 |
+ importlib.metadata.entry_points(group=d_sasp.ENTRY_POINT_GROUP_NAME) |
|
| 90 |
+) |
|
| 91 |
+assert len(entry_points) == n, f"expected exactly {n} entry points: {entry_points!r}"
|
|
| 92 |
+ |
|
| 93 |
+for ep in entry_points: |
|
| 94 |
+ entry = ep.load() |
|
| 95 |
+ assert isinstance(entry, d_sasp.SSHAgentSocketProviderEntry), f"not an SSHAgentSocketProviderEntry: {entry!r}"
|
|
| 96 |
+ assert isinstance(entry.provider, d_sasp.SSHAgentSocketProvider), f"not an SSHAgentSocketProvider: {entry.provider!r}"
|
|
| 97 |
+ assert isinstance(entry.key, str), f"not a str: {entry.key!r}"
|
|
| 98 |
+ assert isinstance(entry.aliases, tuple) and all(isinstance(x, str) for x in entry.aliases), f"not a tuple[str, ...]: {entry.aliases!r}"
|
|
| 99 |
+ assert entry.provider in expected_providers, f"Unexpected provider: {entry.provider!r}"
|
|
| 100 |
+ expected_providers.remove(entry.provider) |
|
| 101 |
+ |
|
| 102 |
+assert not expected_providers, f"Missing providers in entry points: {expected_providers!r}"
|
|
| 103 |
+print("OK")
|
|
| 104 |
+""") |
|
| 105 |
+ check_step = subprocess.run( |
|
| 106 |
+ [str(isolated_venv_python), str(check_file)], |
|
| 107 |
+ check=False, |
|
| 108 |
+ capture_output=True, |
|
| 109 |
+ text=True, |
|
| 110 |
+ ) |
|
| 111 |
+ # Pass on output for debugging, *then* check the return code. |
|
| 112 |
+ sys.stdout.write(check_step.stdout) |
|
| 113 |
+ sys.stderr.write(check_step.stderr) |
|
| 114 |
+ check_step.check_returncode() |
|
| 115 |
+ assert not check_step.stderr.strip() |
|
| 116 |
+ assert check_step.stdout.strip() == "OK" |
|
| 0 | 117 |