diff --git a/README.md b/README.md
index 73fc2b1..ceee50a 100644
--- a/README.md
+++ b/README.md
@@ -227,9 +227,14 @@ Direct mode does not use SSH. It requires Imp's authenticated gateway behind a
publicly trusted TLS reverse proxy plus a pairing token.
Direct WSS is deliberately not enabled by the normal server installation.
-
-See [VPS deployment](deploy/README.md#direct-wss-gateway-advancedmanual) for
-gateway, TLS, and pairing setup.
+Provision pairing with the installed bundle's
+[`imp-direct-wss`](deploy/README.md#direct-wss-gateway-advanced) `setup`,
+`rotate`, and `status` commands. The command prints a token only for deliberate
+one-time terminal handoff; never capture or redirect that output.
+
+See [VPS deployment](deploy/README.md#direct-wss-gateway-advanced) for the
+supported service boundary, TLS/proxy operator responsibilities, bounded
+public verification, and separately authorized live acceptance plan.
## What Imp provides
diff --git a/apps/desktop/src/components/ConnectionDialog.svelte b/apps/desktop/src/components/ConnectionDialog.svelte
index 8590078..72d9f9e 100644
--- a/apps/desktop/src/components/ConnectionDialog.svelte
+++ b/apps/desktop/src/components/ConnectionDialog.svelte
@@ -208,10 +208,18 @@
/>
+
+ Obtain a token with imp-direct-wss setup on the server. Enter the public WSS URL and token
+ separately; never append the token to the URL.
+
+
{settings.hasPairingToken
? 'Leave the token blank to keep the currently stored credential.'
- : 'Paste the 43-character pairing token supplied by the gateway operator.'}
+ : 'Paste the 43-character pairing token from the server setup command.'}
+
+
+ After server rotation, paste the replacement token. Save and restart Imp to use it.
{/if}
diff --git a/deploy/README.md b/deploy/README.md
index c597413..aae23ab 100644
--- a/deploy/README.md
+++ b/deploy/README.md
@@ -145,70 +145,102 @@ sudo loginctl enable-linger "$USER"
The installer reports a warning when lingering is not enabled; it does not
silently perform privileged system configuration.
-## Direct WSS gateway (advanced/manual)
+## Direct WSS gateway (advanced)
-Direct WSS access uses one 256-bit pairing token. Imp's gateway stores only
-the SHA-256 digest of the token. The plaintext token is entered once in the
-desktop Connection settings later; do not put it in a URL, shell history,
-service unit, repository file, or reverse-proxy configuration.
+Direct WSS uses one 256-bit pairing token. The gateway stores only its
+SHA-256 digest. The plaintext token is shown once by the bundle's provisioning
+command for deliberate terminal handoff to the desktop. On the VPS, never put
+the plaintext token in arguments, environment variables, files, URLs, service
+units, reverse-proxy configuration, or logs.
-Until pairing/rotation UX is automated, generate a token and its digest
-manually:
+The bundled `imp-direct-wss` command is installed at the stable user link
+`~/.local/bin/imp-direct-wss`. Invoke that link from the installed Linux
+target user with a working user systemd manager; it operates only on that
+user's installed bundle and fixed HOME paths. From a source checkout, invoke
+`uv run --directory services/relay imp-direct-wss OPERATION`, replacing
+`OPERATION` with exactly one of `setup`, `rotate`, or `status` (for example,
+`uv run --directory services/relay imp-direct-wss setup`). Both forms retain
+the installed-target-user safety boundary; this is not a general systemd manager.
-```bash
-mkdir -p ~/.config/imp
-chmod 700 ~/.config/imp
-
-read -r token digest < ~/.config/imp/gateway.env
-chmod 600 ~/.config/imp/gateway.env
+Run setup from an interactive terminal:
-printf 'Pairing token: %s\n' "$token"
+```bash
+~/.local/bin/imp-direct-wss setup
```
-Record the displayed pairing token securely for the desktop Connection step,
-then clear the shell variables:
+Setup requires stdin and stdout terminals before mutation. With no existing
+configuration it securely generates a token, writes only its digest to
+`~/.config/imp/gateway.env` (private `0700` config directory and `0600` file),
+enables/starts the supported gateway user service, checks exact minimal HTTP
+health `{"status":"ok"}`, and authenticates `/state` requiring the first
+protocol-v2 `hello`. The token is printed once to the terminal only after
+success; do not capture or redirect command output. If a secure, syntactically
+valid single-digest configuration already exists, setup is idempotent: it
+generates no token, makes no service/network calls, and changes nothing.
+If configuration is missing while the gateway is already active, setup refuses
+before mutation; use `rotate` only when rotation is deliberate. Unsafe existing
+files or service arrangements are rejected without modification. Paths must not
+be symlinks; unsupported custom gateway unit files, systemd drop-ins, linked or
+masked units, and manager state requiring daemon reload are rejected rather
+than adopted. The supported unit must match the installed shipped service;
+systemd's resolved fragment, executable, environment-file, and restart policy
+must agree with that unit.
+
+Rotate a compromised or otherwise replaced token using:
```bash
-unset token digest
+~/.local/bin/imp-direct-wss rotate
```
-`gateway.env` contains only the digest, not the plaintext pairing token. The
-gateway refuses to start without a valid 64-hex-character digest.
+Rotate has the same terminal requirement and handoff discipline. If the
+gateway was active, it is restarted and its exact minimal HTTP health plus
+authenticated `/state` protocol-v2 hello are verified. If it was inactive, it
+remains inactive (and retains its previous enabled/disabled state); no
+network preflight or activation occurs.
+
+Handled failures attempt to restore the previous configuration bytes and
+enabled/active service state. Rollback is best effort: if recovery is incomplete,
+the command reports recovery-required and attempts to stop the gateway, rather
+than claiming success. These protections cover handled operation failures, not
+power loss or cross-filesystem/systemd atomicity.
+
+If the prior digest is restored after a failed rotation, the old token
+authenticates again. Treat that as immediate containment work: stop the gateway
+and address the failure before retrying rotation. Do not assume failed rotation
+invalidated a compromised credential.
-The release installer does not enable Direct WSS. After deliberately
-configuring `gateway.env`, enable the gateway separately:
+Check without changing files, locks, credentials, or service state:
```bash
-systemctl --user enable --now imp-gateway.service
-systemctl --user status imp-gateway.service
+~/.local/bin/imp-direct-wss status
```
+Status reports configuration syntax validity, private-file safety, systemd
+enabled/active state, and minimal health without disclosing credential
+material. The fixed paths are `~/.config/imp/gateway.env`,
+`~/.config/systemd/user/imp-gateway.service`, and
+`~/.local/share/imp/current/.venv/bin/imp-gateway`. The config's only accepted
+content is exactly one `IMP_GATEWAY_TOKEN_SHA256=<64 hex characters>`
+assignment, with full-line comments and surrounding whitespace allowed; NUL
+bytes are rejected anywhere, including comments. It is not a general environment
+file. The config directory and file must be owned by the target user with modes
+`0700` and `0600`, respectively.
+
+The normal release installer does not create pairing credentials or enable
+`imp-gateway.service`. Upgrade preserves the gateway's existing enablement and
+restarts it only if it was already active; see the installer behavior above.
+Provision the gateway separately and deliberately with `setup`.
+
The gateway listens only on `127.0.0.1:8788` and connects only to the relay at
`ws://127.0.0.1:8787`. Never proxy port 8787 or the relay's `/ingest` or
`/action-consumer` endpoints to the Internet.
### Public TLS/WSS edge for Direct mode
-Direct WSS requires a publicly trusted TLS endpoint. TLS termination belongs to
-a normal reverse proxy such as Caddy or nginx, not to Imp's Python
-gateway.
-
-The public proxy must expose only `/state`, `/action`, and optionally
-`/healthz`, forwarding them to `127.0.0.1:8788`. All other paths should be
-rejected.
+The operator owns the public hostname, DNS, publicly trusted TLS certificate
+acquisition/renewal, and reverse-proxy lifecycle. Imp's Python gateway does
+not terminate TLS. The reverse proxy must forward only `/state`, `/action`,
+and optionally `/healthz` to `127.0.0.1:8788`; reject every other path.
A minimal Caddy route shape is:
@@ -226,33 +258,258 @@ https://imp.example {
}
```
-Certificate acquisition and renewal are currently operator-owned. If
-certificates are provisioned outside the reverse proxy, renewal must also reload
-the proxy after replacing its readable certificate/key copies.
+If certificates are provisioned outside the reverse proxy, renewal must also
+reload the proxy after replacing its readable certificate/key copies.
The Slice 11 live acceptance used Caddy and a publicly trusted certificate and
verified the public `/healthz`, `/state`, and `/action` path while
-`/ingest` and `/action-consumer` remained unreachable. Automated
-reverse-proxy/certificate provisioning remains future distribution work.
+`/ingest` and `/action-consumer` remained unreachable. This is historical
+evidence only; it does not verify Slice 24 provisioning or current live
+acceptance.
+
+### Bounded public verification
+
+Run this from a machine with network reachability to the public hostname, using
+the installed bundle's CPython 3.12 environment (which includes the Imp relay
+package and `websockets`), or another environment that already has both
+dependencies on its import path. It probes authenticated state, auth
+rejection, and HTTP plus WebSocket 404s for privileged/unknown routes without
+an `Origin` header, and optionally minimal health. It sends no action frame.
+The token is read from /dev/tty using `getpass`, never argv or environment.
+Both secret prompts fail closed if terminal echo control is unavailable, before
+reading fallback input or starting network activity. Errors are intentionally
+generic so exception text cannot expose credentials or response payloads.
+
+Save the snippet as a local script (not in a repository or shared location),
+then run it interactively with the approved WSS state endpoint as its sole
+argument. It rejects credentials, query/fragment, non-WSS URLs, and endpoints
+not ending in `/state`; derives sibling routes while retaining any path prefix.
+The optional previous-token prompt proves revocation after rotation; its value
+is read only from /dev/tty and never appears in output. TLS verification stays
+enabled, redirects are refused, and no insecure TLS options are used. Each HTTP
+probe has a five-second monotonic elapsed deadline covering headers and body,
+in addition to the socket timeout. It uses `SIGALRM` to interrupt even a
+drip-fed response, so it requires Unix, the main thread, and an unarmed
+`ITIMER_REAL`; it refuses to probe if a real-time timer is already armed.
+
+For example, after saving the snippet as `public_wss_check.py`:
-### Direct desktop configuration
+```bash
+~/.local/share/imp/current/.venv/bin/python public_wss_check.py 'wss://imp.example/state'
+```
-On the workstation, open Imp and use **Settings -> Connection -> Direct**.
+```python
+import base64
+import getpass
+import json
+import secrets
+import signal
+import sys
+import time
+import warnings
+from urllib.parse import urlsplit, urlunsplit
+from urllib.error import HTTPError
+from urllib.request import HTTPSHandler, HTTPRedirectHandler, ProxyHandler, Request, build_opener
+
+from imp_relay.websocket_logging import websocket_logger
+from websockets.exceptions import ConnectionClosed, InvalidStatus
+from websockets.sync.client import connect
+
+logger = websocket_logger()
+HTTP_DEADLINE = 5
+
+
+class NoRedirect(HTTPRedirectHandler):
+ def redirect_request(self, req, fp, code, msg, headers, new_url):
+ return None
+
+
+def http_get(url):
+ def expired(signum, frame):
+ raise TimeoutError
+
+ if any(signal.getitimer(signal.ITIMER_REAL)):
+ fail()
+ deadline = time.monotonic() + HTTP_DEADLINE
+ previous_handler = signal.signal(signal.SIGALRM, expired)
+ try:
+ signal.setitimer(signal.ITIMER_REAL, max(deadline - time.monotonic(), 0.000001))
+ with opener.open(Request(url, method="GET"), timeout=5) as response:
+ return response.status, response.read(128)
+ except HTTPError as exc:
+ exc.close()
+ return exc.code, b""
+ except Exception:
+ fail()
+ finally:
+ signal.setitimer(signal.ITIMER_REAL, 0)
+ signal.signal(signal.SIGALRM, previous_handler)
+
+
+def fail():
+ print("Verification failed.")
+ raise SystemExit(1)
+
+
+def endpoint(path):
+ return urlunsplit(("wss", parts.netloc, path, "", ""))
+
+
+def canonical_token(value):
+ try:
+ raw = base64.urlsafe_b64decode(value + "=")
+ return (len(value) == 43 and len(raw) == 32
+ and base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii") == value)
+ except Exception:
+ return False
+
+
+def rejection(path, token):
+ try:
+ with connect(endpoint(path), proxy=None, open_timeout=5, logger=logger) as ws:
+ if token is not None:
+ ws.send('{"type":"auth","token":"' + token + '"}')
+ ws.recv(timeout=10)
+ except ConnectionClosed as exc:
+ if exc.code == 1008:
+ return
+ except Exception:
+ pass
+ fail()
+
+
+def hidden(path):
+ try:
+ with connect(endpoint(path), proxy=None, open_timeout=5, logger=logger):
+ pass
+ except InvalidStatus as exc:
+ if exc.response.status_code == 404:
+ return
+ except Exception:
+ pass
+ fail()
+
+
+if len(sys.argv) != 2:
+ fail()
+try:
+ parts = urlsplit(sys.argv[1])
+ valid_url = (parts.scheme == "wss" and parts.hostname
+ and parts.username is None and parts.password is None
+ and not parts.query and not parts.fragment
+ and parts.path.endswith("/state"))
+ parts.port
+except ValueError:
+ fail()
+if (not valid_url or "?" in sys.argv[1] or "#" in sys.argv[1]):
+ fail()
+prefix = parts.path[:-len("/state")]
+state_path = prefix + "/state"
+action_path = prefix + "/action"
+health_path = prefix + "/healthz"
+try:
+ with open("/dev/tty", "w") as tty, warnings.catch_warnings():
+ warnings.simplefilter("error", getpass.GetPassWarning)
+ token = getpass.getpass("Pairing token: ", stream=tty)
+ previous_token = getpass.getpass(
+ "Previous token to verify revocation (optional): ", stream=tty)
+ if not canonical_token(token) or (previous_token and not canonical_token(previous_token)):
+ fail()
+except Exception:
+ fail()
+
+try:
+ with connect(endpoint(state_path), proxy=None, open_timeout=5, logger=logger) as ws:
+ ws.send('{"type":"auth","token":"' + token + '"}')
+ hello = json.loads(ws.recv(timeout=5))
+ if hello.get("type") != "hello" or hello.get("protocol") != 2:
+ fail()
+except SystemExit:
+ raise
+except Exception:
+ fail()
+
+wrong_token = base64.urlsafe_b64encode(secrets.token_bytes(32)).rstrip(b"=").decode("ascii")
+while wrong_token in (token, previous_token):
+ wrong_token = base64.urlsafe_b64encode(secrets.token_bytes(32)).rstrip(b"=").decode("ascii")
+for route in (state_path, action_path):
+ rejection(route, wrong_token)
+ rejection(route, None)
+ if previous_token:
+ rejection(route, previous_token)
+
+opener = build_opener(ProxyHandler({}), HTTPSHandler(), NoRedirect())
+for route in ("/ingest", "/action-consumer", "/unknown"):
+ status, _ = http_get(f"https://{parts.netloc}{prefix}{route}")
+ if status != 404:
+ fail()
+ hidden(prefix + route)
+
+status, body = http_get(f"https://{parts.netloc}{health_path}")
+if status not in (200, 404):
+ fail()
+if status == 200:
+ try:
+ if json.loads(body) != {"status": "ok"}:
+ fail()
+ except Exception:
+ fail()
+
+print("Public gateway verification passed.")
+```
-Enter:
+Use only a disposable local copy of the script and remove it after use. Do not
+enable websocket protocol/frame logging. The configured Imp `websockets`
+logger's INFO security floor is part of the credential confidentiality
+boundary.
+
+### Separate live acceptance authorization
+
+This is an operator-run acceptance plan only; no live change is authorized by
+this document, and none is being performed now. Before any mutation, obtain
+explicit approval naming the existing VPS, desktop, public hostname,
+maintenance window, and one harmless MUD action. Approval must specifically
+cover securely parking the exact existing `gateway.env` bytes in a mode-`0600`
+backup containing only the digest assignment plus permitted comments/whitespace,
+stopping/disabling the gateway to establish the clean initial state,
+setup/rotation/reinstall and desktop changes, and restoring the prior approved
+configuration and service state. Any old plaintext token
+may be held only in the operator's approved secure terminal/native credential
+store, never in a VPS file or log. If approval or the existing Caddy
+environment is absent, stop before mutation. Reuse the existing service and
+port `8788`; do not provision infrastructure or change Caddy, DNS, or TLS.
+
+1. Establish and record the installed-bundle baseline with `~/.local/bin/imp-direct-wss status`: no gateway digest, service disabled/inactive, and Direct WSS not enabled; preserve the prior digest-only configuration and enabled/active state as authorized.
+2. Run the supported setup command `~/.local/bin/imp-direct-wss setup` interactively.
+3. Receive the newly generated token once through the intentional terminal handoff; never redirect, capture, or log output.
+4. Confirm `~/.config/imp/gateway.env` contains only the SHA-256 digest assignment, with the supported private ownership/modes and no plaintext token.
+5. Confirm `imp-gateway.service` is enabled and active on the existing user manager.
+6. Confirm setup completed its local minimal-health check and authenticated `/state` protocol-v2 hello; record `~/.local/bin/imp-direct-wss status`.
+7. Run the bounded verifier against the existing public `wss:` `/state` URL and confirm trusted TLS plus authenticated state access.
+8. Confirm the verifier receives 404 for privileged/unknown HTTP and WebSocket routes without an `Origin` header.
+9. Configure the existing desktop's Direct pairing through Settings with the public `/state` URL and the new token; save and restart.
+10. Confirm live MUD state arrives over Direct WSS.
+11. Send only the separately approved action through the desktop UI and independently observe its execution; do not synthesize an action in a probe.
+12. While the gateway is active, run `~/.local/bin/imp-direct-wss rotate`; receive the replacement token only by terminal handoff and confirm the service remains active.
+13. Rerun the bounded verifier, entering the replacement token and old token at its secure prompts; confirm the old credential and missing/wrong credentials close with `1008`, while the replacement succeeds.
+14. Save the replacement token in the existing desktop Direct settings, restart, and confirm Direct state succeeds.
+15. Stop and restart the gateway once; confirm the desktop's existing reconnect behavior recovers state without action replay.
+16. Reinstall/upgrade from the already-approved extracted bundle on the same VPS by running `./install.sh`; confirm the digest, enabled state, and active state are preserved and the active gateway is restarted as expected.
+17. Confirm `ss -ltnp` shows relay `8787` and gateway `8788` listening only on loopback; restore the parked digest-only configuration, prior service enablement/active state, and approved desktop credential, then record only observed results and restore outcome.
-- the public `wss:` state endpoint, for example
- `wss://imp.example/state`; and
-- the exact 43-character plaintext pairing token whose SHA-256 digest is stored
- in `gateway.env`.
+### Direct desktop configuration
+
+On the workstation, open Imp and use **Settings -> Connection -> Direct**.
-Save the connection and restart Imp. The URL must use `wss:`, contain no
+Enter the public `wss:` state endpoint (for example `wss://imp.example/state`)
+and the 43-character pairing token displayed by `imp-direct-wss setup` or
+`rotate`. Save and restart Imp. The URL must use `wss:`, contain no
credentials, query, or fragment, and end in `/state`.
-The pairing token is persisted by the native application. Imp's
-settings-read path does not return the stored plaintext token merely to populate
-the form; an existing Direct token can therefore remain unchanged when editing
-other Direct settings.
+The pairing token is persisted by the native application. Imp's settings-read
+path does not return the stored plaintext token merely to populate the form;
+an existing Direct token can therefore remain unchanged when editing other
+Direct settings.
Do not put the token in a URL, reverse-proxy configuration, `VITE_*`
environment value, or WebView `localStorage`.
diff --git a/deploy/install.sh b/deploy/install.sh
index 36d4e29..2a8abee 100755
--- a/deploy/install.sh
+++ b/deploy/install.sh
@@ -164,6 +164,7 @@ BIN_DIR="$HOME/.local/bin"
SYSTEMD_DIR="$HOME/.config/systemd/user"
ACTION_LINK="$BIN_DIR/imp-action-consumer"
+DIRECT_WSS_LINK="$BIN_DIR/imp-direct-wss"
if [[ -e "$CURRENT_LINK" && ! -L "$CURRENT_LINK" ]]; then
die "$CURRENT_LINK exists but is not a symlink"
@@ -173,6 +174,15 @@ if [[ -e "$ACTION_LINK" && ! -L "$ACTION_LINK" ]]; then
die "$ACTION_LINK exists but is not a symlink"
fi
+if [[ -e "$DIRECT_WSS_LINK" && ! -L "$DIRECT_WSS_LINK" ]]; then
+ die "$DIRECT_WSS_LINK exists but is not a symlink"
+fi
+
+if [[ -L "$DIRECT_WSS_LINK" &&
+ "$(readlink "$DIRECT_WSS_LINK")" != "$CURRENT_LINK/.venv/bin/imp-direct-wss" ]]; then
+ die "$DIRECT_WSS_LINK points somewhere other than this install's runtime"
+fi
+
mkdir -p \
"$RELEASES_DIR" \
"$CONFIG_DIR" \
@@ -251,7 +261,7 @@ for package in ("imp-relay", "imp-adapter", "imp-tinyfugue"):
)
PY_VERIFY_RUNTIME
- for entry in imp-relay imp-gateway imp-feed imp-action-consumer; do
+ for entry in imp-relay imp-gateway imp-feed imp-action-consumer imp-direct-wss; do
script="$release/.venv/bin/$entry"
[[ -x "$script" ]] ||
@@ -325,6 +335,7 @@ cleanup() {
restore_path "$SYSTEMD_DIR/imp-relay.service" imp-relay.service
restore_path "$SYSTEMD_DIR/imp-feed.service" imp-feed.service
restore_path "$SYSTEMD_DIR/imp-gateway.service" imp-gateway.service
+ restore_path "$DIRECT_WSS_LINK" direct-wss
fi
if ((TF_STARTUP_CHANGED)) &&
@@ -424,6 +435,8 @@ rm -rf -- "$FILES_BACKUP_DIR"
mkdir -p "$FILES_BACKUP_DIR"
backup_path "$ACTION_LINK" action-consumer
+backup_path "$DIRECT_WSS_LINK" direct-wss
+
backup_path "$CONFIG_DIR/capture.tf" capture.tf
backup_path "$SYSTEMD_DIR/imp-relay.service" imp-relay.service
backup_path "$SYSTEMD_DIR/imp-feed.service" imp-feed.service
@@ -434,6 +447,9 @@ FILES_BACKUP_READY=1
ln -sfn \
"$CURRENT_LINK/.venv/bin/imp-action-consumer" \
"$ACTION_LINK"
+ln -sfn \
+ "$CURRENT_LINK/.venv/bin/imp-direct-wss" \
+ "$DIRECT_WSS_LINK"
install -m 600 \
"$BUNDLE_DIR/capture.tf" \
@@ -524,4 +540,5 @@ if command -v loginctl >/dev/null 2>&1; then
fi
fi
-printf '\nDirect WSS gateway enablement was not changed.\n'
+printf '\nDirect WSS setup is off on a clean install; existing gateway enablement is unchanged.\n'
+printf 'To configure it interactively, run: %s setup\n' "$DIRECT_WSS_LINK"
diff --git a/docs/architecture/boundaries/trust-boundary.md b/docs/architecture/boundaries/trust-boundary.md
index a2f40a1..fbbd582 100644
--- a/docs/architecture/boundaries/trust-boundary.md
+++ b/docs/architecture/boundaries/trust-boundary.md
@@ -154,7 +154,14 @@ gateway:
direct WSS connection;
- possession of the token grants remote state observation and context-bound
action requests for that Imp installation, so compromise requires
- rotation.
+ rotation;
+- installed Linux bundle provisioning reads no token from argv or environment
+ and stores only the digest in `~/.config/imp/gateway.env`; plaintext is
+ emitted only for deliberate terminal handoff during successful setup or
+ rotation;
+- a handled rotation failure may restore the prior digest, in which case the
+ prior token authenticates again and must be treated as immediate containment
+ work.
The relay itself remains unauthenticated and permanently loopback-only.
`/ingest` and `/action-consumer` are not exposed by the gateway. See
diff --git a/docs/architecture/decisions/0010-authenticated-remote-gateway.md b/docs/architecture/decisions/0010-authenticated-remote-gateway.md
index 820474f..0a1c7b0 100644
--- a/docs/architecture/decisions/0010-authenticated-remote-gateway.md
+++ b/docs/architecture/decisions/0010-authenticated-remote-gateway.md
@@ -194,8 +194,8 @@ TLS alone is not authentication to Imp.
- The reverse proxy never needs the Imp pairing token; it forwards the
WebSocket traffic without interpreting Imp authentication.
- Protocol-v2 remains unchanged.
-- Automated reverse-proxy provisioning, token generation/rotation UX, VPS
- installation scripts, and end-user onboarding belong to the following
- deployment/distribution slice.
+- The installed Linux bundle provides target-user gateway setup, rotation, and
+ status commands; hostname, DNS, public TLS, and reverse-proxy provisioning
+ remain operator-owned.
- Multi-user hosting and mutually untrusted local users remain outside the
current supported trust boundary.
diff --git a/docs/architecture/objects/gateway.md b/docs/architecture/objects/gateway.md
index 8dd46c2..c30f328 100644
--- a/docs/architecture/objects/gateway.md
+++ b/docs/architecture/objects/gateway.md
@@ -11,6 +11,14 @@ The gateway does not replace the relay and does not make the relay public. It
authenticates a remote desktop before opening any connection to the existing
loopback relay.
+The installed Linux bundle's `imp-direct-wss setup|rotate|status` command
+manages only the supported target-user service and fixed configuration paths.
+Setup and rotation require terminal stdin/stdout before mutation; plaintext
+token output is reserved for deliberate one-time handoff. Status does not
+mutate or disclose credentials. The normal installer leaves the gateway
+disabled, and upgrades preserve its prior enablement, restarting it only when
+it was active. The operator owns hostname, DNS, TLS, and public proxy setup.
+
## Source
- `services/relay/src/imp_relay/gateway.py` - listener, authentication,
@@ -19,6 +27,8 @@ loopback relay.
listener/upstream configuration and pairing-token digest input.
- `services/relay/src/imp_relay/gateway_main.py` - process entry point and
bounded lifecycle logging.
+- `services/relay/src/imp_relay/direct_wss.py` - installed-bundle setup,
+ rotation, and read-only status for the target user's gateway.
- `deploy/systemd/imp-gateway.service` - VPS user-service shape.
- `apps/desktop/src-tauri/src/tunnel_config.rs` - native Direct-WSS endpoint
and pairing-token configuration.
diff --git a/docs/architecture/processes/managed-runtime.md b/docs/architecture/processes/managed-runtime.md
index 463db24..d2bc12a 100644
--- a/docs/architecture/processes/managed-runtime.md
+++ b/docs/architecture/processes/managed-runtime.md
@@ -28,6 +28,12 @@ client-adapter transport.
| healthy pre-existing relay on `8789` | whoever started it | Managed mode may adopt and monitor it; never owns or signals it |
| WebSocket reconnect and HUD state | `RelayStateSource` / HUD model | transport-independent freshness presentation |
+The installed-bundle `imp-direct-wss` command provisions the VPS gateway under
+the same target user's `systemd --user` manager; it does not configure the
+public hostname, DNS, TLS certificate, or reverse proxy. A clean bundle install
+leaves this service disabled; upgrades preserve its existing enablement and
+active state.
+
A pre-existing listener on the desktop node port `8787` is never adopted,
regardless of whether it looks like an Imp relay. The desktop node supervisor
either owns the child itself or reports the port unavailable.
diff --git a/docs/install-windows.md b/docs/install-windows.md
index 3fd6796..eaf8b01 100644
--- a/docs/install-windows.md
+++ b/docs/install-windows.md
@@ -80,10 +80,13 @@ reach every forwarded relay route.
## Direct WSS
-Direct mode does not use SSH. It requires an Imp authenticated gateway
-behind a publicly trusted TLS reverse proxy.
+Direct mode does not use SSH. It requires an Imp authenticated gateway behind
+a publicly trusted TLS reverse proxy. On the Linux VPS, provision its pairing
+token with the installed bundle's `imp-direct-wss setup` command. The command
+prints the token only for deliberate one-time terminal handoff; never capture
+or redirect its output. Normal installation does not enable the gateway.
-In **Settings -> Connection**:
+In **Settings -> Connection -> Direct**:
1. Select **Direct**.
2. Enter the public `wss:` state endpoint, ending in `/state`.
@@ -91,8 +94,9 @@ In **Settings -> Connection**:
4. Save.
5. Restart Imp.
-Gateway deployment and pairing-token setup are documented in
-[`../deploy/README.md`](../deploy/README.md).
+Gateway setup/rotation, safe public verification, and operator-owned
+hostname/DNS/TLS/proxy boundaries are documented in
+[`../deploy/README.md`](../deploy/README.md#direct-wss-gateway-advanced).
## Mudlet integration
diff --git a/docs/roadmap.md b/docs/roadmap.md
index 66130bf..f338c3c 100644
--- a/docs/roadmap.md
+++ b/docs/roadmap.md
@@ -42,6 +42,7 @@ claims belong in `status.md` and the relevant architecture documents.
| Slice 21 - Native runtime ownership | Complete | `native-runtime-ownership` |
| Slice 22 - Client-neutral UI and docs polish | Complete | `client-neutral-ui-docs` |
| Slice 23 - ActionBroker timeout validation | Complete | `actionbroker-timeout-validation` |
+| Slice 24 - Direct WSS provisioning and onboarding | In progress | No marker; live acceptance pending |
Slice numbers and milestone tags are development-history markers. They have no
relationship to release versions. Release versions are derived independently
@@ -49,8 +50,8 @@ from Conventional Commits as documented in [`releases.md`](releases.md).
The Slices 17–23 remediation program is complete; see the separate
[remediation closure](audits/2026-10-05-remediation-closure.md) for landed
-PR identities, verification, and retained limitations. No next slice is selected;
-the candidate work below remains provisional.
+PR identities, verification, and retained limitations. Slice 24 is selected
+and in progress; candidate work remains provisional.
## ✅ Slice 1 - Bootstrap baseline
@@ -700,6 +701,20 @@ Higher-level timing validation, action semantics, and the historical audit remai
unchanged. Program closure is recorded separately; no broader action hardening
is claimed.
+### Slice 24 - Direct WSS provisioning and onboarding
+
+In progress; no completion marker. Implement installed-bundle-only
+`imp-direct-wss setup|rotate|status` for the Linux target user's fixed config
+and user-systemd service paths. The normal installer stays default-off and
+upgrades preserve existing gateway enablement and active state. Terminal-only
+token handoff, safe existing-config handling, active/inactive rotation,
+credential-free status, and handled rollback are in scope.
+
+Public hostname, DNS, TLS certificate lifecycle, and reverse-proxy
+configuration remain operator-owned. The separately authorized live acceptance
+plan is in the [deployment guide](../deploy/README.md#separate-live-acceptance-authorization).
+Live acceptance is pending; no native or live completion is claimed.
+
## Candidate work
Unordered, deliberately without slice numbers, and provisional. Live evidence
@@ -786,17 +801,14 @@ schemas.
### Distribution and onboarding
-#### Direct WSS provisioning and onboarding
-
-Reduce the operator-owned setup around the authenticated Direct WSS transport.
+#### Direct WSS public-edge automation
-Today the released server installer deliberately does not enable Direct WSS,
-pairing-token generation and rotation are manual, and the public TLS reverse
-proxy and certificate lifecycle remain operator-owned.
+Slice 24 is adding target-user setup, rotation, and status commands while
+leaving the normal installer default-off. Public hostname/DNS configuration,
+TLS certificate lifecycle, and reverse-proxy setup remain operator-owned.
+Automating those public-edge responsibilities is separate future work.
-A future design can evaluate how much of gateway enablement, token lifecycle,
-reverse-proxy/TLS setup, and first-run guidance Imp should automate while
-preserving the existing boundaries:
+A future design must preserve the existing boundaries:
- the relay remains loopback-only;
- the gateway exposes only the intended remote capabilities;
diff --git a/docs/status.md b/docs/status.md
index 9998fd4..a8deb54 100644
--- a/docs/status.md
+++ b/docs/status.md
@@ -1075,4 +1075,29 @@ TypeScript/Svelte checks with no warnings, the frontend production build, all
101 TinyFugue tests, formatting/lint/docs/version checks,
`git diff --check origin/main`, and the complete `npm run check` gate.
+### Slice 24 - Direct WSS provisioning (in progress)
+
+The installed-bundle `imp-direct-wss setup|rotate|status` CLI, stable installer
+link, Direct WSS onboarding text, and operator documentation are implemented.
+Observed local validation: 79 focused provisioning/gateway tests passed, and
+the rebuilt server bundle passed its acceptance script. The latter exercises
+the installed CLI through a PTY with real relay/gateway processes and a
+stateful fake user-systemd manager: setup, idempotence, active/inactive rotation,
+credential retention across distinct releases, read-only status, and rollback.
+It does not establish real user-systemd acceptance.
+The full `npm run check` gate also passed: 64 protocol, 152 desktop, 271 relay,
+31 shared-adapter, 32 Mudlet, and 101 TinyFugue tests, strict typing/lint/docs
+checks, cross-component roundtrip, and the production frontend build.
+
+The Direct connection dialog was inspected in a browser at desktop and narrow
+viewport widths. The documented bounded verifier passed against a disposable
+locally trusted TLS edge with real relay/gateway processes, including old-token
+rejection and privileged-route 404s without sending an action. Neither check
+establishes Windows-native persistence, public-CA TLS, or public deployment.
+
+The separate 17-step live acceptance plan remains unauthorized and pending.
+Do not infer Slice 24 live success from the historical Slice 11 WSS evidence
+or Slice 15 bundle acceptance above. No live VPS, DNS, proxy, certificate, or
+native Windows changes were made.
+
For future candidate work, see [`roadmap.md`](roadmap.md).
diff --git a/scripts/test-server-bundle.sh b/scripts/test-server-bundle.sh
index aa06d22..b592a70 100755
--- a/scripts/test-server-bundle.sh
+++ b/scripts/test-server-bundle.sh
@@ -99,6 +99,11 @@ EXPECTED_RELEASE="$HOME_DIR/.local/share/imp/releases/$VERSION"
)" == "$HOME_DIR/.local/share/imp/current/.venv/bin/imp-action-consumer" ]] ||
die "action-consumer symlink is incorrect"
+[[ "$(
+ readlink "$HOME_DIR/.local/bin/imp-direct-wss"
+)" == "$HOME_DIR/.local/share/imp/current/.venv/bin/imp-direct-wss" ]] ||
+ die "direct-wss symlink is incorrect"
+
mapfile -t WEBSOCKETS_WHEELS < <(
find "$BUNDLE_DIR/wheels" \
-maxdepth 1 \
@@ -147,7 +152,7 @@ asyncio.run(verify_hello_version())
print("package and hello versions: OK")
PY_VERIFY
-for entry in imp-relay imp-gateway imp-feed imp-action-consumer; do
+for entry in imp-relay imp-gateway imp-feed imp-action-consumer imp-direct-wss; do
script="$EXPECTED_RELEASE/.venv/bin/$entry"
[[ -x "$script" ]] ||
@@ -208,10 +213,570 @@ grep -Fxq \
"$HOME_DIR/.config/systemd/user/imp-gateway.service"
assert_no_debris "$HOME_DIR"
+[[ ! -e "$HOME_DIR/.config/imp/gateway.env" ]] ||
+ die "ordinary install unexpectedly created gateway credentials"
printf 'clean install: OK\n'
printf 'idempotent reinstall: OK\n'
+printf '\n=== credential retention across same-version and distinct-release installs ===\n'
+
+UPGRADE_HOME="$WORK_DIR/upgrade-home"
+SYNTHETIC_BUNDLE="$WORK_DIR/synthetic-release"
+mkdir -p "$UPGRADE_HOME"
+printf '/load ~/.config/imp/capture.tf\n' > "$UPGRADE_HOME/.tfrc"
+HOME="$UPGRADE_HOME" "$BUNDLE_DIR/install.sh" --no-start
+
+mkdir -p "$UPGRADE_HOME/.config/imp"
+printf 'IMP_GATEWAY_TOKEN_SHA256=%064d\n' 0 \
+ > "$UPGRADE_HOME/.config/imp/gateway.env"
+CONFIG_SHA="$(sha256sum "$UPGRADE_HOME/.config/imp/gateway.env")"
+
+HOME="$UPGRADE_HOME" "$BUNDLE_DIR/install.sh" --no-start
+[[ "$(sha256sum "$UPGRADE_HOME/.config/imp/gateway.env")" == "$CONFIG_SHA" ]] ||
+ die "same-version reinstall changed gateway credentials"
+
+cp -a "$BUNDLE_DIR" "$SYNTHETIC_BUNDLE"
+SYNTHETIC_VERSION="$(
+ python3.12 - "$VERSION" "$SYNTHETIC_BUNDLE" <<'PY_SYNTHETIC_RELEASE'
+import base64
+import csv
+from hashlib import sha256
+from io import StringIO
+from pathlib import Path
+import re
+import sys
+from zipfile import ZIP_DEFLATED, ZipFile
+
+old_version, root_value = sys.argv[1:]
+root = Path(root_value)
+parts = old_version.rsplit(".", 1)
+new_version = f"{parts[0]}.{int(parts[1]) + 1}"
+wheel_dir = root / "wheels"
+
+for wheel in wheel_dir.glob("*.whl"):
+ if wheel.name.startswith("websockets-"):
+ continue
+ rewritten = {}
+ with ZipFile(wheel) as source:
+ for name in source.namelist():
+ renamed = name.replace(f"-{old_version}.dist-info/", f"-{new_version}.dist-info/")
+ data = source.read(name)
+ if name.endswith(".dist-info/METADATA"):
+ data = re.sub(
+ rb"(?m)^Version: " + re.escape(old_version.encode()) + rb"$",
+ f"Version: {new_version}".encode(),
+ data,
+ )
+ rewritten[renamed] = data
+
+ record_path = next(name for name in rewritten if name.endswith(".dist-info/RECORD"))
+ rows = []
+ for name, data in rewritten.items():
+ if name == record_path:
+ rows.append((name, "", ""))
+ else:
+ digest = base64.urlsafe_b64encode(sha256(data).digest()).rstrip(b"=").decode()
+ rows.append((name, f"sha256={digest}", str(len(data))))
+ record = StringIO()
+ csv.writer(record, lineterminator="\n").writerows(rows)
+ rewritten[record_path] = record.getvalue().encode()
+
+ replacement = wheel.with_name(wheel.name.replace(f"-{old_version}-", f"-{new_version}-", 1))
+ with ZipFile(replacement, "w", ZIP_DEFLATED) as target:
+ for name, data in rewritten.items():
+ target.writestr(name, data)
+ wheel.unlink()
+
+(root / "VERSION").write_text(new_version + "\n")
+files = [
+ root / "VERSION",
+ root / "install.sh",
+ root / "capture.tf",
+ *sorted((root / "systemd").glob("*.service")),
+ *sorted(wheel_dir.glob("*.whl")),
+]
+(root / "SHA256SUMS").write_text(
+ "".join(
+ f"{sha256(path.read_bytes()).hexdigest()} {path.relative_to(root)}\n"
+ for path in files
+ )
+)
+print(new_version)
+PY_SYNTHETIC_RELEASE
+)"
+
+HOME="$UPGRADE_HOME" "$SYNTHETIC_BUNDLE/install.sh" --no-start
+[[ "$(readlink "$UPGRADE_HOME/.local/share/imp/current")" == \
+ "$UPGRADE_HOME/.local/share/imp/releases/$SYNTHETIC_VERSION" ]] ||
+ die "distinct-release install did not select the new runtime"
+[[ "$(readlink "$UPGRADE_HOME/.local/bin/imp-direct-wss")" == \
+ "$UPGRADE_HOME/.local/share/imp/current/.venv/bin/imp-direct-wss" ]] ||
+ die "distinct-release install did not preserve the stable CLI link"
+[[ "$(sha256sum "$UPGRADE_HOME/.config/imp/gateway.env")" == "$CONFIG_SHA" ]] ||
+ die "distinct-release install changed gateway credentials"
+printf 'credential retention and synthetic release selection: OK\n'
+
+printf '\n=== installed Direct WSS setup, rotation, and status ===\n'
+
+python3.12 - "$HOME_DIR" "$BUNDLE_DIR" "$SYNTHETIC_BUNDLE" <<'PY_DIRECT_WSS_BUNDLE'
+import hashlib
+import json
+import os
+from pathlib import Path
+import pty
+import re
+import select
+import socket
+import subprocess
+import sys
+import tempfile
+import threading
+import time
+
+home, bundle, synthetic = map(Path, sys.argv[1:])
+runtime = home / ".local/share/imp/current/.venv/bin"
+cli = home / ".local/bin/imp-direct-wss"
+unit_path = home / ".config/systemd/user/imp-gateway.service"
+config_path = home / ".config/imp/gateway.env"
+state = {
+ "enabled": False,
+ "gateway": None,
+ "gateway_release": None,
+ "fail_next_gateway_start": False,
+ "starts": 0,
+ "relay": False,
+ "feed": False,
+}
+state_lock = threading.Lock()
+temporary = tempfile.TemporaryDirectory()
+socket_path = str(Path(temporary.name) / "systemd.sock")
+server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+server.bind(socket_path)
+server.listen()
+server.settimeout(0.2)
+events = []
+relay = None
+stop_server = threading.Event()
+
+
+def stop_gateway():
+ process = state["gateway"]
+ state["gateway"] = None
+ if process is not None and process.poll() is None:
+ process.terminate()
+ try:
+ process.wait(timeout=5)
+ except subprocess.TimeoutExpired:
+ process.kill()
+ process.wait()
+
+
+def start_gateway():
+ stop_gateway()
+ assignment = config_path.read_text().strip().split("=", 1)[1]
+ if state["fail_next_gateway_start"]:
+ assignment = "0" * 64
+ state["fail_next_gateway_start"] = False
+ env = os.environ.copy()
+ env["IMP_GATEWAY_TOKEN_SHA256"] = assignment
+ binary = os.path.realpath(runtime / "imp-gateway")
+ state["gateway_release"] = Path(binary).parent.parent.parent.name
+ process = subprocess.Popen(
+ [
+ binary,
+ "--host", "127.0.0.1", "--port", "8788",
+ "--relay-url", "ws://127.0.0.1:8787",
+ ],
+ env=env,
+ stdout=subprocess.DEVNULL,
+ stderr=subprocess.DEVNULL,
+ )
+ state["gateway"] = process
+ state["starts"] += 1
+ deadline = time.monotonic() + 5
+ while time.monotonic() < deadline:
+ if process.poll() is not None:
+ raise RuntimeError("disposable gateway exited during startup")
+ try:
+ with socket.create_connection(("127.0.0.1", 8788), timeout=0.1):
+ return
+ except OSError:
+ time.sleep(0.05)
+ raise RuntimeError("disposable gateway did not listen")
+
+
+def active(unit):
+ if unit == "imp-gateway.service":
+ process = state["gateway"]
+ return process is not None and process.poll() is None
+ return state["relay"] if unit == "imp-relay.service" else state["feed"]
+
+
+def handle(arguments):
+ args = arguments[1:] if arguments and arguments[0] == "--user" else arguments
+ if len(args) >= 2 and args[0] == "show":
+ binary = str(home / ".local/share/imp/current/.venv/bin/imp-gateway")
+ command = [binary, "--host", "127.0.0.1", "--port", "8788",
+ "--relay-url", "ws://127.0.0.1:8787"]
+ exec_start = f"{{ path={binary} ; argv[]={' '.join(command)} ; ignore_errors=no }}"
+ values = {
+ "LoadState": "loaded",
+ "FragmentPath": str(unit_path),
+ "DropInPaths": "",
+ "NeedDaemonReload": "no",
+ "Type": "simple",
+ "ExecStart": exec_start,
+ "EnvironmentFiles": (
+ f"{home}/.config/imp/gateway.env (ignore_errors=no)"
+ ),
+ "Restart": "on-failure",
+ "RestartUSec": "2s",
+ "NoNewPrivileges": "yes",
+ "UnitFileState": "enabled" if state["enabled"] else "disabled",
+ "ActiveState": "active" if active("imp-gateway.service") else "inactive",
+ }
+ return 0, "".join(f"{key}={value}\n" for key, value in values.items())
+
+ operation = args[0] if args else ""
+ unit = args[-1] if args else ""
+ events.append(tuple(args))
+ if operation == "is-active":
+ return (0 if active(unit) else 3), ""
+ if operation == "is-enabled":
+ if unit == "imp-gateway.service":
+ enabled = state["enabled"]
+ else:
+ enabled = state["relay"] if unit == "imp-relay.service" else state["feed"]
+ return (0 if enabled else 1), ""
+ if operation == "enable":
+ for name in args[1:]:
+ if name == "--now":
+ continue
+ if name == "imp-gateway.service":
+ state["enabled"] = True
+ elif name == "imp-relay.service":
+ state["relay"] = True
+ elif name == "imp-feed.service":
+ state["feed"] = True
+ if "--now" in args:
+ start_gateway()
+ elif operation == "disable":
+ if unit == "imp-gateway.service":
+ state["enabled"] = False
+ elif unit == "imp-relay.service":
+ state["relay"] = False
+ elif unit == "imp-feed.service":
+ state["feed"] = False
+ elif operation == "restart":
+ if unit == "imp-gateway.service":
+ start_gateway()
+ elif unit == "imp-relay.service":
+ state["relay"] = True
+ elif unit == "imp-feed.service":
+ state["feed"] = True
+ elif operation == "stop":
+ if unit == "imp-gateway.service":
+ stop_gateway()
+ elif unit == "imp-relay.service":
+ state["relay"] = False
+ elif unit == "imp-feed.service":
+ state["feed"] = False
+ return 0, ""
+
+
+def serve():
+ while not stop_server.is_set():
+ try:
+ connection, _ = server.accept()
+ except socket.timeout:
+ continue
+ with connection:
+ request = json.loads(connection.makefile("rb").readline())
+ try:
+ with state_lock:
+ code, output = handle(request)
+ response = {"code": code, "output": output}
+ except Exception:
+ response = {"code": 1, "output": ""}
+ connection.sendall(json.dumps(response).encode() + b"\n")
+
+
+client = Path(temporary.name) / "systemctl"
+client.write_text(
+ "#!/usr/bin/env python3\n"
+ "import json, os, socket, sys\n"
+ "s=socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)\n"
+ "s.connect(os.environ['IMP_TEST_SYSTEMD_SOCKET'])\n"
+ "s.sendall(json.dumps(sys.argv[1:]).encode()+b'\\n')\n"
+ "r=json.loads(s.makefile('rb').readline())\n"
+ "sys.stdout.write(r['output'])\n"
+ "raise SystemExit(r['code'])\n"
+)
+client.chmod(0o755)
+server_thread = threading.Thread(target=serve, daemon=True)
+server_thread.start()
+environment = os.environ.copy()
+environment.update(
+ HOME=str(home),
+ PATH=f"{temporary.name}:{os.environ['PATH']}",
+ IMP_TEST_SYSTEMD_SOCKET=socket_path,
+)
+environment.pop("PYTHONPATH", None)
+environment.pop("PYTHONHOME", None)
+environment.pop("IMP_GATEWAY_TOKEN_SHA256", None)
+
+
+def run_tty(command, expected=0):
+ master, slave = pty.openpty()
+ process = subprocess.Popen(
+ [str(arg) for arg in command],
+ env=environment,
+ stdin=slave,
+ stdout=slave,
+ stderr=slave,
+ close_fds=True,
+ )
+ os.close(slave)
+ output = bytearray()
+ deadline = time.monotonic() + 40
+ timed_out = False
+ try:
+ while True:
+ if not timed_out and time.monotonic() >= deadline and process.poll() is None:
+ timed_out = True
+ process.kill()
+ process.wait()
+ ready, _, _ = select.select([master], [], [], 0.2)
+ if not ready:
+ continue
+ try:
+ chunk = os.read(master, 4096)
+ except OSError:
+ break
+ if not chunk:
+ break
+ output.extend(chunk)
+ status = process.wait()
+ if timed_out:
+ raise AssertionError("CLI timed out")
+ if status != expected:
+ raise AssertionError(f"CLI returned {status}, expected {expected}")
+ return bytes(output)
+ finally:
+ os.close(master)
+ if process.poll() is None:
+ process.kill()
+ process.wait()
+
+
+def token_from(output):
+ values = re.findall(
+ rb"(?:Pairing token:|Direct WSS pairing token \(shown once; not stored\):\r?\n)"
+ rb"([A-Za-z0-9_-]{43})(?:\r?\n|$)",
+ output,
+ )
+ if len(values) != 1:
+ raise AssertionError("successful PTY handoff did not contain exactly one token")
+ token = values[0]
+ digest = config_path.read_text().strip().split("=", 1)[1]
+ if hashlib.sha256(token).hexdigest() != digest:
+ raise AssertionError("terminal token did not match the stored digest")
+ return token
+
+
+try:
+ for port in (8787, 8788):
+ try:
+ with socket.socket() as probe:
+ probe.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+ probe.bind(("127.0.0.1", port))
+ except OSError:
+ print(f"Direct WSS PTY integration requires free loopback port {port}", file=sys.stderr)
+ raise SystemExit(1)
+ relay = subprocess.Popen(
+ [str(runtime / "imp-relay"), "--host", "127.0.0.1", "--port", "8787"],
+ env=environment,
+ stdout=subprocess.DEVNULL,
+ stderr=subprocess.DEVNULL,
+ )
+ deadline = time.monotonic() + 5
+ relay_ready = False
+ while time.monotonic() < deadline and relay.poll() is None:
+ try:
+ with socket.create_connection(("127.0.0.1", 8787), timeout=0.1):
+ relay_ready = True
+ break
+ except OSError:
+ time.sleep(0.05)
+ if relay.poll() is not None or not relay_ready:
+ raise AssertionError("disposable relay did not become ready")
+ setup_event_start = len(events)
+ state["fail_next_gateway_start"] = True
+ failed_setup = run_tty([cli, "setup"], expected=1)
+ assert not re.search(
+ rb"(?:Pairing token:|Direct WSS pairing token \(shown once; not stored\):\r?\n)"
+ rb"[A-Za-z0-9_-]{43}",
+ failed_setup,
+ )
+ assert not config_path.exists()
+ assert not state["enabled"] and not active("imp-gateway.service")
+ setup_actions = [
+ event for event in events[setup_event_start:]
+ if event and event[0] in {"enable", "disable", "restart", "stop"}
+ ]
+ assert setup_actions == [
+ ("enable", "--now", "imp-gateway.service"),
+ ("stop", "imp-gateway.service"),
+ ("disable", "imp-gateway.service"),
+ ]
+
+ token = token_from(run_tty([cli, "setup"]))
+ initial_config = config_path.read_bytes()
+ assert b"=" in initial_config and token not in initial_config
+ assert state["enabled"] and active("imp-gateway.service")
+ assert state["starts"] == 2
+
+ log_count = len(events)
+
+ repeated = run_tty([cli, "setup"])
+ assert token not in repeated
+ assert config_path.read_bytes() == initial_config
+ assert len(events) == log_count
+
+ config_path.write_bytes(b"IMP_GATEWAY_TOKEN_SHA256=broken\n")
+ malformed = config_path.read_bytes()
+ log_count = len(events)
+ run_tty([cli, "setup"], expected=1)
+ assert config_path.read_bytes() == malformed
+ assert len(events) == log_count
+ config_path.write_bytes(initial_config)
+
+ before_reinstall = config_path.read_bytes()
+ completed = subprocess.run(
+ [str(bundle / "install.sh"), "--no-start"],
+ env=environment,
+ stdout=subprocess.DEVNULL,
+ stderr=subprocess.DEVNULL,
+ check=False,
+ )
+ assert completed.returncode == 0
+ assert config_path.read_bytes() == before_reinstall
+
+ rotated = token_from(run_tty([cli, "rotate"]))
+ assert rotated != token
+ assert state["starts"] == 3 and active("imp-gateway.service")
+
+ before_failed_rotation = config_path.read_bytes()
+ failure_event_start = len(events)
+ starts_before_failure = state["starts"]
+ state["fail_next_gateway_start"] = True
+ failed_output = run_tty([cli, "rotate"], expected=1)
+ assert not re.search(
+ rb"(?:Pairing token:|Direct WSS pairing token \(shown once; not stored\):\r?\n)"
+ rb"[A-Za-z0-9_-]{43}",
+ failed_output,
+ )
+ assert config_path.read_bytes() == before_failed_rotation
+ assert state["starts"] == starts_before_failure + 2
+ assert active("imp-gateway.service") and state["enabled"]
+ failure_actions = [
+ event for event in events[failure_event_start:]
+ if event and event[0] in {"enable", "disable", "restart", "stop"}
+ ]
+ assert failure_actions == [
+ ("restart", "imp-gateway.service"),
+ ("restart", "imp-gateway.service"),
+ ("enable", "imp-gateway.service"),
+ ]
+
+ import asyncio
+ import logging
+
+ sys.path.insert(0, str(runtime.parent / "lib/python3.12/site-packages"))
+ from websockets.asyncio.client import connect
+
+ async def verify_restored_credentials():
+ async with connect(
+ "ws://127.0.0.1:8788/state",
+ proxy=None,
+ logger=logging.getLogger("bundle-acceptance"),
+ ) as connection:
+ await connection.send(json.dumps({"type": "auth", "token": rotated.decode()}))
+ hello = json.loads(await connection.recv())
+ assert hello.get("type") == "hello" and hello.get("protocol") == 2
+
+ asyncio.run(verify_restored_credentials())
+
+
+ before_upgrade = config_path.read_bytes()
+ upgraded = subprocess.run(
+ [str(synthetic / "install.sh")],
+ env=environment,
+ stdout=subprocess.DEVNULL,
+ stderr=subprocess.DEVNULL,
+ check=False,
+ )
+ assert upgraded.returncode == 0
+ assert config_path.read_bytes() == before_upgrade
+ assert state["gateway_release"] == (synthetic / "VERSION").read_text().strip()
+ assert state["starts"] == 6 and active("imp-gateway.service") and state["enabled"]
+
+ manager = subprocess.run(
+ [str(client), "--user", "stop", "imp-gateway.service"],
+ env=environment,
+ stdout=subprocess.DEVNULL,
+ stderr=subprocess.DEVNULL,
+ check=False,
+ )
+ assert manager.returncode == 0
+ assert not active("imp-gateway.service") and state["enabled"]
+ starts = state["starts"]
+ mutations_before_inactive_rotate = [
+ event for event in events
+ if event and event[0] in {"enable", "disable", "restart", "stop"}
+ ]
+
+ inactive_token = token_from(run_tty([cli, "rotate"]))
+ assert inactive_token != rotated
+ mutations_after_inactive_rotate = [
+ event for event in events
+ if event and event[0] in {"enable", "disable", "restart", "stop"}
+ ]
+ assert mutations_after_inactive_rotate == mutations_before_inactive_rotate
+ assert state["starts"] == starts
+ assert not active("imp-gateway.service") and state["enabled"]
+
+ status_before = config_path.stat()
+ lock_before = (config_path.parent / "gateway.env.lock").stat()
+ status = subprocess.run(
+ [str(cli), "status"],
+ env=environment,
+ stdin=subprocess.DEVNULL,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE,
+ check=False,
+ )
+ assert status.returncode == 0
+ assert inactive_token not in status.stdout and inactive_token not in status.stderr
+ assert config_path.stat().st_mtime_ns == status_before.st_mtime_ns
+ assert (config_path.parent / "gateway.env.lock").stat().st_mtime_ns == lock_before.st_mtime_ns
+finally:
+ with state_lock:
+ stop_gateway()
+ stop_server.set()
+ server_thread.join(timeout=2)
+ server.close()
+ if relay is not None and relay.poll() is None:
+ relay.terminate()
+ try:
+ relay.wait(timeout=5)
+ except subprocess.TimeoutExpired:
+ relay.kill()
+ relay.wait()
+ temporary.cleanup()
+PY_DIRECT_WSS_BUNDLE
+
printf '\n=== custom TinyFugue startup override ===\n'
CUSTOM_HOME="$WORK_DIR/custom-home"
@@ -324,6 +889,23 @@ grep -Fq \
printf 'unlisted bundle content rejection: OK\n'
+printf '\n=== unsafe Direct WSS link rejection ===\n'
+
+LINK_HOME="$WORK_DIR/unsafe-link-home"
+mkdir -p "$LINK_HOME/.local/bin"
+printf '/echo startup\n' > "$LINK_HOME/.tfrc"
+ln -s /tmp/unrelated-command "$LINK_HOME/.local/bin/imp-direct-wss"
+
+if HOME="$LINK_HOME" "$BUNDLE_DIR/install.sh" --no-start \
+ >/dev/null 2>&1; then
+ die "installer accepted an unrelated imp-direct-wss symlink"
+fi
+
+[[ "$(readlink "$LINK_HOME/.local/bin/imp-direct-wss")" == /tmp/unrelated-command ]] ||
+ die "unsafe imp-direct-wss symlink was modified"
+[[ ! -e "$LINK_HOME/.local/share/imp" ]] ||
+ die "unsafe imp-direct-wss link rejection changed the install home"
+
printf '\n=== failed activation rollback ===\n'
ROLLBACK_HOME="$WORK_DIR/rollback-home"
@@ -355,44 +937,140 @@ printf 'old-gateway-unit\n' \
ln -sfn /tmp/old-action-consumer \
"$ROLLBACK_HOME/.local/bin/imp-action-consumer"
+
printf '/echo original startup\n' > "$ROLLBACK_HOME/.tfrc"
ROLLBACK_SYSTEMCTL_LOG="$WORK_DIR/rollback-systemctl.log"
ROLLBACK_RELAY_FAIL_MARKER="$WORK_DIR/rollback-relay-failed-once"
: > "$ROLLBACK_SYSTEMCTL_LOG"
+ROLLBACK_SYSTEMCTL_STATE="$WORK_DIR/rollback-systemctl-state"
+mkdir -p "$ROLLBACK_SYSTEMCTL_STATE"
+for unit in imp-relay.service imp-feed.service imp-gateway.service; do
+ : > "$ROLLBACK_SYSTEMCTL_STATE/active-$unit"
+done
+for unit in imp-relay.service imp-feed.service; do
+ : > "$ROLLBACK_SYSTEMCTL_STATE/enabled-$unit"
+done
+
cat > "$FAKE_BIN/systemctl" <<'SH_SYSTEMCTL'
#!/usr/bin/env bash
set -eu
printf '%s\n' "$*" >> "${IMP_TEST_SYSTEMCTL_LOG:?}"
+state="${IMP_TEST_SYSTEMCTL_STATE:?}"
-case "$*" in
- *"is-active"*)
- exit 0
+case "$2" in
+ is-active)
+ [[ -e "$state/active-${4:?}" ]]
+ ;;
+ is-enabled)
+ [[ -e "$state/enabled-${4:?}" ]]
+ ;;
+ enable)
+ for unit in "${@:3}"; do : > "$state/enabled-$unit"; done
;;
- *"is-enabled"*)
- exit 0
+ disable)
+ for unit in "${@:3}"; do rm -f "$state/enabled-$unit"; done
;;
- *"restart imp-relay.service"*)
- if [[ ! -e "${IMP_TEST_RELAY_FAIL_MARKER:?}" ]]; then
+ restart)
+ unit="${3:?}"
+ if [[ "$unit" == imp-relay.service &&
+ ! -e "${IMP_TEST_RELAY_FAIL_MARKER:?}" ]]; then
: > "$IMP_TEST_RELAY_FAIL_MARKER"
exit 1
fi
- exit 0
+ : > "$state/active-$unit"
;;
- *)
- exit 0
+ stop)
+ rm -f "$state/active-${3:?}"
;;
esac
SH_SYSTEMCTL
chmod +x "$FAKE_BIN/systemctl"
+printf '\n=== ordinary install service state ===\n'
+
+START_HOME="$WORK_DIR/start-home"
+START_STATE="$WORK_DIR/start-state"
+mkdir -p "$START_HOME" "$START_STATE"
+printf '/load ~/.config/imp/capture.tf\n' > "$START_HOME/.tfrc"
+: > "$START_STATE/enabled-imp-gateway.service"
+: > "$WORK_DIR/no-relay-failure"
+: > "$WORK_DIR/start-systemctl.log"
+
+HOME="$START_HOME" \
+ PATH="$FAKE_BIN:$PATH" \
+ IMP_TEST_SYSTEMCTL_LOG="$WORK_DIR/start-systemctl.log" \
+ IMP_TEST_SYSTEMCTL_STATE="$START_STATE" \
+ IMP_TEST_RELAY_FAIL_MARKER="$WORK_DIR/no-relay-failure" \
+ "$BUNDLE_DIR/install.sh" \
+ >"$WORK_DIR/start.out" 2>&1
+
+[[ -e "$START_STATE/active-imp-relay.service" &&
+ -e "$START_STATE/active-imp-feed.service" ]] ||
+ die "ordinary install did not activate relay and feed"
+[[ -e "$START_STATE/enabled-imp-relay.service" &&
+ -e "$START_STATE/enabled-imp-feed.service" ]] ||
+ die "ordinary install did not enable relay and feed"
+[[ -e "$START_STATE/enabled-imp-gateway.service" &&
+ ! -e "$START_STATE/active-imp-gateway.service" ]] ||
+ die "ordinary install changed gateway enabled/active state"
+[[ ! -e "$START_HOME/.config/imp/gateway.env" ]] ||
+ die "ordinary install unexpectedly created gateway credentials"
+
+for call in enable start restart stop disable; do
+ if grep -E -- "--user $call( .*|$)imp-gateway.service" \
+ "$WORK_DIR/start-systemctl.log"; then
+ die "ordinary install mutated gateway service state"
+ fi
+done
+
+INACTIVE_DISABLED_HOME="$WORK_DIR/inactive-disabled-home"
+INACTIVE_DISABLED_STATE="$WORK_DIR/inactive-disabled-state"
+mkdir -p "$INACTIVE_DISABLED_HOME" "$INACTIVE_DISABLED_STATE"
+printf '/load ~/.config/imp/capture.tf\n' > "$INACTIVE_DISABLED_HOME/.tfrc"
+HOME="$INACTIVE_DISABLED_HOME" \
+ PATH="$FAKE_BIN:$PATH" \
+ IMP_TEST_SYSTEMCTL_LOG="$WORK_DIR/inactive-disabled.log" \
+ IMP_TEST_SYSTEMCTL_STATE="$INACTIVE_DISABLED_STATE" \
+ IMP_TEST_RELAY_FAIL_MARKER="$WORK_DIR/no-relay-failure" \
+ "$BUNDLE_DIR/install.sh" \
+ >"$WORK_DIR/inactive-disabled.out" 2>&1
+[[ ! -e "$INACTIVE_DISABLED_STATE/active-imp-gateway.service" &&
+ ! -e "$INACTIVE_DISABLED_STATE/enabled-imp-gateway.service" ]] ||
+ die "ordinary install changed inactive/disabled gateway state"
+
+ACTIVE_DISABLED_HOME="$WORK_DIR/active-disabled-home"
+ACTIVE_DISABLED_STATE="$WORK_DIR/active-disabled-state"
+mkdir -p "$ACTIVE_DISABLED_HOME" "$ACTIVE_DISABLED_STATE"
+printf '/load ~/.config/imp/capture.tf\n' > "$ACTIVE_DISABLED_HOME/.tfrc"
+: > "$ACTIVE_DISABLED_STATE/active-imp-gateway.service"
+HOME="$ACTIVE_DISABLED_HOME" \
+ PATH="$FAKE_BIN:$PATH" \
+ IMP_TEST_SYSTEMCTL_LOG="$WORK_DIR/active-disabled.log" \
+ IMP_TEST_SYSTEMCTL_STATE="$ACTIVE_DISABLED_STATE" \
+ IMP_TEST_RELAY_FAIL_MARKER="$WORK_DIR/no-relay-failure" \
+ "$BUNDLE_DIR/install.sh" \
+ >"$WORK_DIR/active-disabled.out" 2>&1
+[[ -e "$ACTIVE_DISABLED_STATE/active-imp-gateway.service" &&
+ ! -e "$ACTIVE_DISABLED_STATE/enabled-imp-gateway.service" ]] ||
+ die "ordinary install changed active/disabled gateway state"
+grep -Fxq -- '--user restart imp-gateway.service' "$WORK_DIR/active-disabled.log" ||
+ die "ordinary install did not restart an active gateway"
+for call in enable disable; do
+ if grep -Fxq -- "--user $call imp-gateway.service" "$WORK_DIR/active-disabled.log"; then
+ die "ordinary install changed gateway enablement"
+ fi
+done
+
+
if HOME="$ROLLBACK_HOME" \
PATH="$FAKE_BIN:$PATH" \
IMP_TEST_SYSTEMCTL_LOG="$ROLLBACK_SYSTEMCTL_LOG" \
+ IMP_TEST_SYSTEMCTL_STATE="$ROLLBACK_SYSTEMCTL_STATE" \
IMP_TEST_RELAY_FAIL_MARKER="$ROLLBACK_RELAY_FAIL_MARKER" \
"$BUNDLE_DIR/install.sh" \
>"$WORK_DIR/rollback.out" 2>&1; then
@@ -478,8 +1156,23 @@ mapfile -t ROLLBACK_BACKUPS < <(
assert_no_debris "$ROLLBACK_HOME"
+[[ -e "$ROLLBACK_SYSTEMCTL_STATE/active-imp-relay.service" &&
+ -e "$ROLLBACK_SYSTEMCTL_STATE/active-imp-feed.service" &&
+ -e "$ROLLBACK_SYSTEMCTL_STATE/active-imp-gateway.service" ]] ||
+ die "fake systemd did not restore the previous active state"
+[[ -e "$ROLLBACK_SYSTEMCTL_STATE/enabled-imp-relay.service" &&
+ -e "$ROLLBACK_SYSTEMCTL_STATE/enabled-imp-feed.service" &&
+ ! -e "$ROLLBACK_SYSTEMCTL_STATE/enabled-imp-gateway.service" ]] ||
+ die "fake systemd did not retain the previous enabled state"
+
+[[ "$(
+ readlink "$ROLLBACK_HOME/.local/bin/imp-direct-wss"
+)" == "$ROLLBACK_HOME/.local/share/imp/current/.venv/bin/imp-direct-wss" ]] ||
+ die "direct-wss link rollback failed"
+
printf 'runtime rollback: OK\n'
printf 'integration-file rollback: OK\n'
printf 'TinyFugue startup rollback: OK\n'
+printf 'Fake systemd models state and launches real CLI-fixture gateway/relay processes; it does not validate real user-systemd behavior.\n'
printf '\nserver bundle acceptance: PASS\n'
diff --git a/services/relay/README.md b/services/relay/README.md
index de27411..c5d46e9 100644
--- a/services/relay/README.md
+++ b/services/relay/README.md
@@ -23,3 +23,20 @@ snapshot/text state and performs no action retry or replay.
`GET /healthz` on the relay returns feed/state health.
`GET /healthz` on the gateway returns only `{"status":"ok"}`.
+
+## Installed gateway provisioning
+
+The bundled Linux target-user CLI `imp-direct-wss setup|rotate|status` manages
+only the supported user-systemd gateway installation and fixed user paths.
+Invoke the stable installed link `~/.local/bin/imp-direct-wss`; from a source
+checkout invoke `uv run --directory services/relay imp-direct-wss OPERATION`,
+replacing `OPERATION` with one of `setup`, `rotate`, or `status` (for example,
+`uv run --directory services/relay imp-direct-wss setup`). Setup and rotation
+require interactive stdin/stdout before mutation and show a newly issued token
+only for intentional terminal handoff. Status does not write or reveal
+credentials.
+
+The normal installer leaves the gateway disabled; upgrades preserve
+enablement and restart it only if already active. The operator remains
+responsible for public hostname, DNS, TLS, and the reverse-proxy route allowlist.
+See the [deployment guide](../../deploy/README.md#direct-wss-gateway-advanced).
diff --git a/services/relay/pyproject.toml b/services/relay/pyproject.toml
index d5cdfc0..92b06f9 100644
--- a/services/relay/pyproject.toml
+++ b/services/relay/pyproject.toml
@@ -13,6 +13,7 @@ dependencies = ["websockets>=16.0,<17"]
[project.scripts]
imp-relay = "imp_relay.__main__:main"
imp-gateway = "imp_relay.gateway_main:main"
+imp-direct-wss = "imp_relay.direct_wss:main"
[dependency-groups]
dev = [
diff --git a/services/relay/src/imp_relay/direct_wss.py b/services/relay/src/imp_relay/direct_wss.py
new file mode 100644
index 0000000..dd97dbb
--- /dev/null
+++ b/services/relay/src/imp_relay/direct_wss.py
@@ -0,0 +1,810 @@
+"""Provision the installed loopback gateway without exporting or storing its pairing token."""
+
+from __future__ import annotations
+
+import asyncio
+import http.client
+import json
+import os
+import re
+import secrets
+import shlex
+import stat
+import subprocess
+import sys
+import time
+from collections.abc import Iterator
+from contextlib import contextmanager, suppress
+from dataclasses import dataclass
+from pathlib import Path
+from typing import IO
+
+from websockets.asyncio.client import connect
+
+from .gateway import token_digest, valid_pairing_token
+from .protocol import HelloMessage, decode_server_message
+from .websocket_logging import websocket_logger
+
+try:
+ import fcntl
+except ImportError: # Keep desktop imports/builds working outside Linux.
+ fcntl = None # type: ignore[assignment]
+
+_UNIT = "imp-gateway.service"
+_CONFIG_NAME = "gateway.env"
+_LOCK_NAME = "gateway.env.lock"
+_CONFIG_KEY = "IMP_GATEWAY_TOKEN_SHA256"
+_GATEWAY_HOST = "127.0.0.1"
+_GATEWAY_PORT = 8788
+_RELAY_PORT = 8787
+_TIMEOUT = 5
+_MAX_CONFIG_BYTES = 4096
+_DIGEST_LINE = re.compile(r"^IMP_GATEWAY_TOKEN_SHA256[ \t]*=[ \t]*([0-9a-fA-F]{64})$")
+_PROPERTIES = (
+ "LoadState",
+ "FragmentPath",
+ "DropInPaths",
+ "NeedDaemonReload",
+ "Type",
+ "ExecStart",
+ "EnvironmentFiles",
+ "Restart",
+ "RestartUSec",
+ "NoNewPrivileges",
+ "UnitFileState",
+ "ActiveState",
+)
+_SHIPPED_UNIT = (
+ "[Unit]\n"
+ "Description=Imp authenticated remote gateway\n"
+ "After=network.target imp-relay.service\n"
+ "Wants=imp-relay.service\n"
+ "\n"
+ "[Service]\n"
+ "Type=simple\n"
+ "EnvironmentFile=%h/.config/imp/gateway.env\n"
+ "ExecStart=%h/.local/share/imp/current/.venv/bin/imp-gateway --host 127.0.0.1 --port 8788 --relay-url ws://127.0.0.1:8787\n"
+ "Restart=on-failure\n"
+ "RestartSec=2\n"
+ "NoNewPrivileges=true\n"
+ "\n"
+ "[Install]\n"
+ "WantedBy=default.target\n"
+).encode("ascii")
+
+
+class DirectWssError(Exception):
+ """A concise, credential-safe command failure."""
+
+
+class AtomicWriteError(DirectWssError):
+ def __init__(self, *, replaced: bool) -> None:
+ super().__init__("could not atomically write gateway.env")
+ self.replaced = replaced
+
+
+@dataclass(frozen=True)
+class ServiceState:
+ active: bool
+ enabled: bool
+
+
+@dataclass(frozen=True)
+class ConfigInspection:
+ syntax: str
+ mode: str
+ directory_mode: str
+ digest: bytes | None
+ data: bytes | None
+ exists: bool
+
+
+def _supported_platform() -> None:
+ if not sys.platform.startswith("linux") or fcntl is None:
+ raise DirectWssError("Direct WSS requires Linux user-systemd")
+ if os.geteuid() == 0:
+ raise DirectWssError("run Direct WSS as the installed target user, not root")
+
+
+def _user_home() -> tuple[int, str]:
+ uid = os.geteuid()
+ home = str(Path.home())
+ if not home.startswith("/") or home == "/":
+ raise DirectWssError("the target user's home directory is unavailable")
+ return uid, home
+
+
+def _directory_flags() -> int:
+ return os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC | os.O_NOFOLLOW
+
+
+def _open_home_component(parent_fd: int, name: str, uid: int) -> int:
+ try:
+ fd = os.open(name, _directory_flags(), dir_fd=parent_fd)
+ except OSError:
+ raise DirectWssError("the target user's home directory is unsafe") from None
+ try:
+ info = os.fstat(fd)
+ except OSError:
+ os.close(fd)
+ raise DirectWssError("the target user's home directory is unsafe") from None
+ mode = stat.S_IMODE(info.st_mode)
+ if (
+ not stat.S_ISDIR(info.st_mode)
+ or info.st_uid not in {0, uid}
+ or (mode & 0o022 and not (info.st_uid == 0 and mode & stat.S_ISVTX))
+ ):
+ os.close(fd)
+ raise DirectWssError("the target user's home directory is unsafe")
+ return fd
+
+
+def _open_directory(parent_fd: int, name: str, uid: int, *, create: bool = False) -> int:
+ created = False
+ try:
+ fd = os.open(name, _directory_flags(), dir_fd=parent_fd)
+ except FileNotFoundError:
+ if not create:
+ raise
+ try:
+ os.mkdir(name, 0o700, dir_fd=parent_fd)
+ created = True
+ except FileExistsError:
+ pass
+ try:
+ fd = os.open(name, _directory_flags(), dir_fd=parent_fd)
+ except OSError:
+ raise DirectWssError("configuration path contains an unsafe directory") from None
+ except OSError:
+ raise DirectWssError("configuration path contains an unsafe directory") from None
+ try:
+ info = os.fstat(fd)
+ except OSError:
+ os.close(fd)
+ raise DirectWssError("configuration path contains an unsafe directory") from None
+ mode = stat.S_IMODE(info.st_mode)
+ if not stat.S_ISDIR(info.st_mode) or info.st_uid != uid or mode & 0o022:
+ os.close(fd)
+ raise DirectWssError("configuration path contains an unsafe directory")
+ if created:
+ try:
+ os.fchmod(fd, 0o700)
+ except OSError:
+ os.close(fd)
+ raise DirectWssError("could not secure the configuration directory") from None
+ return fd
+
+
+@contextmanager
+def _open_home() -> Iterator[tuple[int, str, int]]:
+ uid, home = _user_home()
+ fd = os.open("/", _directory_flags())
+ try:
+ for component in home.split("/")[1:]:
+ if not component or component in {".", ".."}:
+ raise DirectWssError("the target user's home directory is unsafe")
+ next_fd = _open_home_component(fd, component, uid)
+ os.close(fd)
+ fd = next_fd
+ if os.fstat(fd).st_uid != uid:
+ raise DirectWssError("the target user's home directory is not owned by that user")
+ yield uid, home, fd
+ finally:
+ os.close(fd)
+
+
+def _config_directory(home_fd: int, uid: int, *, create: bool, private: bool = True) -> int:
+ config_fd = _open_directory(home_fd, ".config", uid, create=create)
+ try:
+ imp_fd = _open_directory(config_fd, "imp", uid, create=create)
+ finally:
+ os.close(config_fd)
+ if private and stat.S_IMODE(os.fstat(imp_fd).st_mode) != 0o700:
+ os.close(imp_fd)
+ raise DirectWssError("~/.config/imp must be owned by the target user with mode 0700")
+ return imp_fd
+
+
+def _read_regular_file(dir_fd: int, name: str, uid: int, *, max_bytes: int) -> tuple[bytes, os.stat_result]:
+ try:
+ fd = os.open(name, os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=dir_fd)
+ except FileNotFoundError:
+ raise
+ except OSError:
+ raise DirectWssError("configuration file is not a safe regular file") from None
+ try:
+ info = os.fstat(fd)
+ if not stat.S_ISREG(info.st_mode) or info.st_uid != uid or info.st_nlink != 1:
+ raise DirectWssError("configuration file is not a safe regular file")
+ chunks: list[bytes] = []
+ size = 0
+ while size <= max_bytes:
+ chunk = os.read(fd, min(1024, max_bytes + 1 - size))
+ if not chunk:
+ break
+ chunks.append(chunk)
+ size += len(chunk)
+ if size > max_bytes:
+ raise DirectWssError("configuration file is too large")
+ return b"".join(chunks), info
+ finally:
+ os.close(fd)
+
+
+def _parse_config(data: bytes) -> bytes:
+ if b"\x00" in data:
+ raise DirectWssError("gateway.env has unsupported or malformed content")
+ try:
+ text = data.decode("utf-8")
+ except UnicodeDecodeError:
+ raise DirectWssError("gateway.env has unsupported or malformed content") from None
+ digests: list[str] = []
+ for line in text.split("\n"):
+ stripped = line.removesuffix("\r").strip(" \t")
+ if not stripped or stripped.startswith("#"):
+ continue
+ match = _DIGEST_LINE.fullmatch(stripped)
+ if match is None:
+ raise DirectWssError("gateway.env has unsupported or malformed content")
+ digests.append(match.group(1))
+ if len(digests) != 1:
+ raise DirectWssError("gateway.env must contain exactly one token digest assignment")
+ return bytes.fromhex(digests[0])
+
+
+def _inspect_config(dir_fd: int, uid: int, *, require_private: bool) -> ConfigInspection:
+ try:
+ data, info = _read_regular_file(dir_fd, _CONFIG_NAME, uid, max_bytes=_MAX_CONFIG_BYTES)
+ except FileNotFoundError:
+ return ConfigInspection("missing", "unavailable", "0700 private", None, None, False)
+ mode = stat.S_IMODE(info.st_mode)
+ mode_text = f"{mode:04o} {'private' if mode == 0o600 else 'not private'}"
+ if require_private and mode != 0o600:
+ raise DirectWssError("gateway.env must be owned by the target user with mode 0600")
+ digest = _parse_config(data)
+ return ConfigInspection("valid", mode_text, "0700 private", digest, data, True)
+
+
+def _lock_mutations(dir_fd: int, uid: int) -> int:
+ flags = os.O_RDWR | os.O_CLOEXEC | os.O_NOFOLLOW | os.O_NONBLOCK
+ created = False
+ try:
+ fd = os.open(_LOCK_NAME, flags | os.O_CREAT | os.O_EXCL, 0o600, dir_fd=dir_fd)
+ created = True
+ except FileExistsError:
+ try:
+ fd = os.open(_LOCK_NAME, flags, dir_fd=dir_fd)
+ except OSError:
+ raise DirectWssError("could not safely open the Direct WSS mutation lock") from None
+ except OSError:
+ raise DirectWssError("could not safely open the Direct WSS mutation lock") from None
+ if created:
+ try:
+ os.fchmod(fd, 0o600)
+ except OSError:
+ os.close(fd)
+ raise DirectWssError("could not secure the Direct WSS mutation lock") from None
+ info = os.fstat(fd)
+ if (
+ not stat.S_ISREG(info.st_mode)
+ or info.st_uid != uid
+ or info.st_nlink != 1
+ or stat.S_IMODE(info.st_mode) != 0o600
+ ):
+ os.close(fd)
+ raise DirectWssError("Direct WSS mutation lock has unsafe ownership or permissions")
+ assert fcntl is not None
+ try:
+ fcntl.flock(fd, fcntl.LOCK_EX)
+ except OSError:
+ os.close(fd)
+ raise DirectWssError("could not acquire the Direct WSS mutation lock") from None
+ return fd
+
+
+def _atomic_replace(dir_fd: int, data: bytes) -> None:
+ staged = f".gateway.env.tmp.{secrets.token_hex(8)}"
+ fd: int | None = None
+ staged_created = False
+ replaced = False
+ try:
+ fd = os.open(
+ staged,
+ os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_CLOEXEC | os.O_NOFOLLOW,
+ 0o600,
+ dir_fd=dir_fd,
+ )
+ staged_created = True
+ os.fchmod(fd, 0o600)
+ view = memoryview(data)
+ while view:
+ written = os.write(fd, view)
+ if written <= 0:
+ raise OSError("short write")
+ view = view[written:]
+ os.fsync(fd)
+ os.close(fd)
+ fd = None
+ os.replace(staged, _CONFIG_NAME, src_dir_fd=dir_fd, dst_dir_fd=dir_fd)
+ replaced = True
+ except BaseException:
+ if staged_created and not replaced:
+ try:
+ os.stat(staged, dir_fd=dir_fd, follow_symlinks=False)
+ except FileNotFoundError:
+ replaced = True
+ except OSError:
+ replaced = True
+ raise AtomicWriteError(replaced=replaced) from None
+ finally:
+ if fd is not None:
+ with suppress(OSError):
+ os.close(fd)
+ if staged_created:
+ try:
+ os.unlink(staged, dir_fd=dir_fd)
+ except FileNotFoundError:
+ pass
+ except OSError:
+ pass
+
+
+def _restore_config(dir_fd: int, previous: bytes | None) -> None:
+ if previous is None:
+ try:
+ os.unlink(_CONFIG_NAME, dir_fd=dir_fd)
+ except FileNotFoundError:
+ return
+ except OSError:
+ raise DirectWssError("could not remove the generated gateway.env during rollback") from None
+ else:
+ _atomic_replace(dir_fd, previous)
+
+
+def _systemctl(*args: str) -> subprocess.CompletedProcess[str]:
+ try:
+ return subprocess.run(
+ ("systemctl", "--user", *args),
+ capture_output=True,
+ text=True,
+ timeout=_TIMEOUT,
+ check=False,
+ )
+ except (OSError, subprocess.TimeoutExpired):
+ raise DirectWssError("the user systemd manager could not be reached") from None
+
+
+def _property_map(output: str) -> dict[str, str]:
+ result: dict[str, str] = {}
+ for line in output.splitlines():
+ key, separator, value = line.partition("=")
+ if separator:
+ result[key] = value
+ return result
+
+
+def _unit_file(home_fd: int, uid: int, home: str) -> str:
+ try:
+ config_fd = _open_directory(home_fd, ".config", uid)
+ try:
+ systemd_fd = _open_directory(config_fd, "systemd", uid)
+ finally:
+ os.close(config_fd)
+ try:
+ user_fd = _open_directory(systemd_fd, "user", uid)
+ finally:
+ os.close(systemd_fd)
+ except FileNotFoundError:
+ raise DirectWssError("the installed gateway unit path is missing") from None
+ try:
+ try:
+ data, info = _read_regular_file(user_fd, _UNIT, uid, max_bytes=16_384)
+ except FileNotFoundError:
+ raise DirectWssError("the installed gateway unit is missing") from None
+ if stat.S_IMODE(info.st_mode) & 0o022:
+ raise DirectWssError("the installed gateway unit has unsafe permissions")
+ if data != _SHIPPED_UNIT:
+ raise DirectWssError("the installed gateway unit differs from the supported bundle unit")
+ try:
+ os.stat(f"{_UNIT}.d", dir_fd=user_fd, follow_symlinks=False)
+ except FileNotFoundError:
+ pass
+ else:
+ raise DirectWssError("gateway unit drop-ins are not supported")
+ return os.path.join(home, ".config/systemd/user", _UNIT)
+ finally:
+ os.close(user_fd)
+
+
+def _expected_exec_start(value: str, home: str) -> bool:
+ binary = os.path.join(home, ".local/share/imp/current/.venv/bin/imp-gateway")
+ paths = [path.strip() for path in re.findall(r"(?:^|\{ )path=([^;}]*)", value)]
+ raw_argv = re.findall(r"argv\[\]=(.*?)(?=\s*;\s*|\s*\})", value)
+ if len(raw_argv) != 1:
+ return False
+ try:
+ argv = shlex.split(raw_argv[0])
+ except ValueError:
+ return False
+ return paths == [binary] and argv == [
+ binary,
+ "--host",
+ _GATEWAY_HOST,
+ "--port",
+ str(_GATEWAY_PORT),
+ "--relay-url",
+ f"ws://{_GATEWAY_HOST}:{_RELAY_PORT}",
+ ]
+
+
+def _expected_environment_files(value: str, home: str) -> bool:
+ expected = os.path.join(home, ".config/imp", _CONFIG_NAME)
+ return value.strip() == f"{expected} (ignore_errors=no)"
+
+
+def _service_preflight(home_fd: int, home: str) -> ServiceState:
+ expected_fragment = _unit_file(home_fd, os.geteuid(), home)
+ shown = _systemctl("show", _UNIT, "--property=" + ",".join(_PROPERTIES))
+ if shown.returncode != 0:
+ raise DirectWssError("the gateway unit is not available in the user systemd manager")
+ props = _property_map(shown.stdout)
+ expected_values = {
+ "LoadState": "loaded",
+ "FragmentPath": expected_fragment,
+ "DropInPaths": "",
+ "NeedDaemonReload": "no",
+ "Type": "simple",
+ "Restart": "on-failure",
+ "RestartUSec": "2s",
+ "NoNewPrivileges": "yes",
+ }
+ if any(props.get(key) != value for key, value in expected_values.items()):
+ raise DirectWssError("the manager's loaded gateway unit is unsupported or differs from disk")
+ if not _expected_exec_start(props.get("ExecStart", ""), home):
+ raise DirectWssError("the manager's loaded gateway command is not the supported loopback command")
+ if not _expected_environment_files(props.get("EnvironmentFiles", ""), home):
+ raise DirectWssError("the manager's loaded gateway environment file is unsupported")
+ unit_state = props.get("UnitFileState")
+ active_state = props.get("ActiveState")
+ if unit_state not in {"enabled", "disabled"} or active_state not in {"active", "inactive"}:
+ raise DirectWssError("the gateway must be in a normal enabled or disabled, active or inactive state")
+ active_check = _systemctl("is-active", "--quiet", _UNIT)
+ enabled_check = _systemctl("is-enabled", "--quiet", _UNIT)
+ active = active_state == "active"
+ enabled = unit_state == "enabled"
+ if active_check.returncode != (0 if active else 3) or enabled_check.returncode != (0 if enabled else 1):
+ raise DirectWssError("the gateway manager state is inconsistent")
+ return ServiceState(active, enabled)
+
+
+def _systemctl_action(*args: str) -> None:
+ if _systemctl(*args).returncode != 0:
+ raise DirectWssError("a gateway systemd operation failed")
+
+
+def _restore_service(
+ before: ServiceState,
+ *,
+ operation_attempted: bool,
+ config_restored: bool,
+) -> list[str]:
+ if not operation_attempted and config_restored:
+ return []
+ failures: list[str] = []
+ active_after = before.active and config_restored
+ if operation_attempted or not config_restored:
+ try:
+ _systemctl_action("restart" if active_after else "stop", _UNIT)
+ except DirectWssError:
+ failures.append("restore the previous gateway active state")
+ if operation_attempted:
+ try:
+ _systemctl_action("enable" if before.enabled else "disable", _UNIT)
+ except DirectWssError:
+ failures.append("restore the previous gateway enabled state")
+ try:
+ active_check = _systemctl("is-active", "--quiet", _UNIT)
+ if active_check.returncode != (0 if active_after else 3):
+ failures.append("verify the previous gateway active state")
+ except DirectWssError:
+ failures.append("verify the previous gateway active state")
+ if operation_attempted:
+ try:
+ enabled_check = _systemctl("is-enabled", "--quiet", _UNIT)
+ if enabled_check.returncode != (0 if before.enabled else 1):
+ failures.append("verify the previous gateway enabled state")
+ except DirectWssError:
+ failures.append("verify the previous gateway enabled state")
+ if failures:
+ try:
+ _systemctl_action("stop", _UNIT)
+ except DirectWssError:
+ failures.append("stop the gateway")
+ return failures
+
+
+def _gateway_health() -> bool:
+ connection = http.client.HTTPConnection(_GATEWAY_HOST, _GATEWAY_PORT, timeout=_TIMEOUT)
+ try:
+ connection.request("GET", "/healthz")
+ response = connection.getresponse()
+ body = response.read(1024)
+ if response.status != 200:
+ return False
+ try:
+ return bool(json.loads(body, object_pairs_hook=lambda pairs: pairs) == [("status", "ok")])
+ except (json.JSONDecodeError, UnicodeDecodeError):
+ return False
+ except (OSError, http.client.HTTPException):
+ return False
+ finally:
+ connection.close()
+
+
+async def _state_hello(port: int, token: str | None = None) -> bool:
+ try:
+ async with connect(
+ f"ws://{_GATEWAY_HOST}:{port}/state",
+ open_timeout=_TIMEOUT,
+ close_timeout=2,
+ proxy=None,
+ logger=websocket_logger(),
+ ) as connection:
+ if token is not None:
+ await connection.send(json.dumps({"type": "auth", "token": token}, separators=(",", ":")))
+ frame = await asyncio.wait_for(connection.recv(), timeout=_TIMEOUT)
+ except Exception:
+ return False
+ if not isinstance(frame, str):
+ return False
+ decoded = decode_server_message(frame)
+ return decoded.ok and isinstance(decoded.value, HelloMessage) and decoded.value.protocol == 2
+
+
+def _verify_gateway(token: str) -> None:
+ deadline = time.monotonic() + _TIMEOUT
+ while time.monotonic() < deadline:
+ if _gateway_health() and asyncio.run(_state_hello(_GATEWAY_PORT, token)):
+ return
+ time.sleep(0.1)
+ raise DirectWssError("gateway did not become healthy and authenticate within the readiness timeout")
+
+
+def _require_tty() -> None:
+ if not sys.stdin.isatty() or not sys.stdout.isatty():
+ raise DirectWssError("setup and rotate require interactive stdin and stdout")
+
+
+def _write_pairing_token(token: str, stdout: IO[str]) -> None:
+ try:
+ stdout.write("Direct WSS pairing token (shown once; not stored):\n")
+ stdout.write(token + "\n")
+ stdout.flush()
+ except OSError:
+ if stdout is sys.stdout:
+ with suppress(OSError, ValueError):
+ fd = os.open(os.devnull, os.O_WRONLY)
+ try:
+ os.dup2(fd, stdout.fileno())
+ finally:
+ os.close(fd)
+ raise
+
+
+def _mutate(command: str, stdout: IO[str]) -> None:
+ _supported_platform()
+ with _open_home() as (uid, home, home_fd):
+ try:
+ config_fd = _config_directory(home_fd, uid, create=False)
+ except FileNotFoundError:
+ config_fd = None
+ if config_fd is None:
+ if command == "rotate":
+ raise DirectWssError("rotate requires an existing supported gateway.env")
+ else:
+ try:
+ existing = _inspect_config(config_fd, uid, require_private=True)
+ if command == "setup" and existing.exists:
+ stdout.write("Direct WSS is already configured; no changes made.\n")
+ return
+ if command == "rotate" and not existing.exists:
+ raise DirectWssError("rotate requires an existing supported gateway.env")
+ finally:
+ os.close(config_fd)
+ _require_tty()
+ config_fd = _config_directory(home_fd, uid, create=True)
+ lock_fd: int | None = None
+ try:
+ lock_fd = _lock_mutations(config_fd, uid)
+ existing = _inspect_config(config_fd, uid, require_private=True)
+ if command == "setup" and existing.exists:
+ stdout.write("Direct WSS is already configured; no changes made.\n")
+ return
+ if command == "rotate" and not existing.exists:
+ raise DirectWssError("rotate requires an existing supported gateway.env")
+ before = _service_preflight(home_fd, home)
+ if command == "setup" and before.active:
+ raise DirectWssError("setup cannot replace credentials while the gateway is already active")
+ if (command == "setup" or before.active) and not asyncio.run(_state_hello(_RELAY_PORT)):
+ raise DirectWssError("the local relay is unavailable or did not send a protocol-v2 hello")
+ token = secrets.token_urlsafe(32)
+ if not valid_pairing_token(token):
+ raise DirectWssError("secure pairing token generation failed")
+ digest = token_digest(token)
+ old_data = existing.data
+ new_data = f"{_CONFIG_KEY}={digest.hex()}\n".encode("ascii")
+ config_replaced = False
+ service_operation_attempted = False
+ try:
+ _atomic_replace(config_fd, new_data)
+ config_replaced = True
+ if command == "setup":
+ service_operation_attempted = True
+ _systemctl_action("enable", "--now", _UNIT)
+ elif before.active:
+ service_operation_attempted = True
+ _systemctl_action("restart", _UNIT)
+ if command == "setup" or before.active:
+ if _systemctl("is-active", "--quiet", _UNIT).returncode != 0:
+ raise DirectWssError("the gateway service did not become active")
+ expected_enabled = command == "setup" or before.enabled
+ if _systemctl("is-enabled", "--quiet", _UNIT).returncode != (
+ 0 if expected_enabled else 1
+ ):
+ raise DirectWssError("the gateway enabled state changed unexpectedly")
+ _verify_gateway(token)
+ stdout.write(
+ "Gateway: locally verified; enabled for boot.\n"
+ if expected_enabled
+ else "Gateway: locally verified; boot enablement remains disabled.\n"
+ )
+ else:
+ stdout.write("Credential rotated; gateway remains stopped and enablement is unchanged.\n")
+ stdout.write(
+ "Public state URL: wss:///state\n"
+ "Configure your TLS proxy to forward approved routes to 127.0.0.1:8788.\n"
+ "Paste the URL and token separately into Settings -> Connection -> Direct.\n"
+ "Save and restart Imp.\n"
+ )
+ _write_pairing_token(token, stdout)
+ except BaseException as error:
+ rollback_failures: list[str] = []
+ config_may_be_replaced = config_replaced or (
+ isinstance(error, AtomicWriteError) and error.replaced
+ )
+ config_restored = not config_may_be_replaced
+ if config_may_be_replaced:
+ try:
+ _restore_config(config_fd, old_data)
+ config_restored = True
+ except DirectWssError:
+ rollback_failures.append("restore the previous gateway.env bytes")
+ rollback_failures.extend(
+ _restore_service(
+ before,
+ operation_attempted=service_operation_attempted,
+ config_restored=config_restored,
+ )
+ )
+ if rollback_failures:
+ raise DirectWssError(
+ "operation failed; restoration failed ("
+ + ", ".join(rollback_failures)
+ + "); gateway stop was attempted"
+ ) from None
+ if isinstance(error, (KeyboardInterrupt, SystemExit)):
+ raise DirectWssError(
+ "operation interrupted; previous configuration and service state restored"
+ ) from None
+ if isinstance(error, DirectWssError):
+ raise DirectWssError(
+ f"{error}; previous configuration and service state restored"
+ ) from None
+ raise DirectWssError(
+ "operation failed; previous configuration and service state restored"
+ ) from None
+ finally:
+ if lock_fd is not None:
+ os.close(lock_fd)
+ os.close(config_fd)
+
+
+def _status_config(home_fd: int, uid: int) -> ConfigInspection:
+ try:
+ config_fd = _config_directory(home_fd, uid, create=False, private=False)
+ except FileNotFoundError:
+ return ConfigInspection("missing", "unavailable", "unavailable", None, None, False)
+ except DirectWssError:
+ return ConfigInspection("unsafe", "unavailable", "unavailable", None, None, False)
+ try:
+ directory_value = stat.S_IMODE(os.fstat(config_fd).st_mode)
+ directory_mode = f"{directory_value:04o} {'private' if directory_value == 0o700 else 'not private'}"
+ try:
+ data, info = _read_regular_file(config_fd, _CONFIG_NAME, uid, max_bytes=_MAX_CONFIG_BYTES)
+ except FileNotFoundError:
+ return ConfigInspection("missing", "unavailable", directory_mode, None, None, False)
+ except DirectWssError:
+ try:
+ info = os.stat(_CONFIG_NAME, dir_fd=config_fd, follow_symlinks=False)
+ mode = stat.S_IMODE(info.st_mode)
+ mode_text = f"{mode:04o} {'private' if mode == 0o600 else 'not private'}"
+ except OSError:
+ mode_text = "unavailable"
+ return ConfigInspection("unsafe", mode_text, directory_mode, None, None, True)
+ mode = stat.S_IMODE(info.st_mode)
+ mode_text = f"{mode:04o} {'private' if mode == 0o600 else 'not private'}"
+ try:
+ digest = _parse_config(data)
+ except DirectWssError:
+ return ConfigInspection("invalid", mode_text, directory_mode, None, data, True)
+ return ConfigInspection("valid", mode_text, directory_mode, digest, data, True)
+ finally:
+ os.close(config_fd)
+
+
+def _status_service(home_fd: int, home: str) -> tuple[str, str, str]:
+ enabled = "unknown"
+ active = "unknown"
+ policy = "unsupported"
+ try:
+ state = _service_preflight(home_fd, home)
+ enabled = "yes" if state.enabled else "no"
+ active = "yes" if state.active else "no"
+ policy = "supported"
+ except DirectWssError:
+ try:
+ shown = _systemctl("show", _UNIT, "--property=" + ",".join(_PROPERTIES))
+ except DirectWssError:
+ return enabled, active, policy
+ if shown.returncode == 0:
+ props = _property_map(shown.stdout)
+ enabled_state = props.get("UnitFileState")
+ active_state = props.get("ActiveState")
+ if enabled_state in {"enabled", "disabled"}:
+ enabled = "yes" if enabled_state == "enabled" else "no"
+ if active_state in {"active", "inactive"}:
+ active = "yes" if active_state == "active" else "no"
+ return enabled, active, policy
+
+
+def _status(stdout: IO[str]) -> None:
+ _supported_platform()
+ with _open_home() as (uid, home, home_fd):
+ inspection = _status_config(home_fd, uid)
+ enabled, active, policy = _status_service(home_fd, home)
+ health = "healthy" if active == "yes" and _gateway_health() else "unavailable"
+ stdout.write(f"gateway.env syntax: {inspection.syntax}\n")
+ stdout.write(f"gateway.env mode: {inspection.mode}\n")
+ stdout.write(f"gateway.env directory mode: {inspection.directory_mode}\n")
+ stdout.write(f"gateway service enabled: {enabled}\n")
+ stdout.write(f"gateway service active: {active}\n")
+ stdout.write(f"gateway unit policy: {policy}\n")
+ stdout.write(f"gateway health: {health}\n")
+
+
+def main(argv: list[str] | None = None) -> int:
+ args = sys.argv[1:] if argv is None else argv
+ if len(args) != 1 or args[0] not in {"setup", "rotate", "status"}:
+ with suppress(BaseException):
+ print("usage: imp-direct-wss {setup|rotate|status}", file=sys.stderr)
+ return 2
+ try:
+ command = args[0]
+ if command == "status":
+ _status(sys.stdout)
+ else:
+ _mutate(command, sys.stdout)
+ return 0
+ except DirectWssError as error:
+ message = f"error: {error}"
+ except KeyboardInterrupt:
+ return 130
+ except Exception:
+ message = "error: Direct WSS operation failed"
+ with suppress(BaseException):
+ print(message, file=sys.stderr)
+ return 1
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/services/relay/tests/test_direct_wss.py b/services/relay/tests/test_direct_wss.py
new file mode 100644
index 0000000..b1b2695
--- /dev/null
+++ b/services/relay/tests/test_direct_wss.py
@@ -0,0 +1,1005 @@
+from __future__ import annotations
+
+import base64
+import hashlib
+import http.server
+import io
+import os
+import secrets
+import shlex
+import stat
+import subprocess
+import sys
+import threading
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any
+
+import pytest
+
+import imp_relay.direct_wss as direct_wss
+from imp_relay.gateway import token_digest
+
+TOKEN = base64.urlsafe_b64encode(bytes(range(32))).rstrip(b"=").decode("ascii")
+OLD_TOKEN = base64.urlsafe_b64encode(bytes(range(1, 33))).rstrip(b"=").decode("ascii")
+
+
+class TTY(io.StringIO):
+ def isatty(self) -> bool:
+ return True
+
+
+def _home(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
+ home = tmp_path / "home"
+ home.mkdir()
+ monkeypatch.setattr(direct_wss, "_supported_platform", lambda: None)
+ monkeypatch.setattr(direct_wss, "_user_home", lambda: (os.geteuid(), str(home)))
+ return home
+
+
+def _config_dir(home: Path) -> Path:
+ path = home / ".config" / "imp"
+ path.mkdir(parents=True, exist_ok=True)
+ os.chmod(path.parent, 0o700)
+ os.chmod(path, 0o700)
+ return path
+
+
+def _write_config(home: Path, data: bytes) -> Path:
+ path = _config_dir(home) / "gateway.env"
+ path.write_bytes(data)
+ os.chmod(path, 0o600)
+ return path
+
+
+def _install_unit(home: Path) -> None:
+ unit_dir = home / ".config" / "systemd" / "user"
+ unit_dir.mkdir(parents=True)
+ for directory in (unit_dir.parent, unit_dir):
+ os.chmod(directory, 0o700)
+ unit = unit_dir / direct_wss._UNIT
+ unit.write_bytes(direct_wss._SHIPPED_UNIT)
+ os.chmod(unit, 0o644)
+
+
+def _tty_streams(monkeypatch: pytest.MonkeyPatch) -> tuple[TTY, io.StringIO]:
+ monkeypatch.setattr(sys, "stdin", TTY())
+ stdout = TTY()
+ stderr = io.StringIO()
+ monkeypatch.setattr(sys, "stdout", stdout)
+ monkeypatch.setattr(sys, "stderr", stderr)
+ return stdout, stderr
+
+
+class FakeSystemctl:
+ def __init__(
+ self,
+ home: Path,
+ *,
+ active: bool = False,
+ enabled: bool = False,
+ properties: dict[str, str] | None = None,
+ failed_restarts: set[int] | None = None,
+ ) -> None:
+ self.home = home
+ self.active = active
+ self.enabled = enabled
+ self.properties = properties or {}
+ self.failed_restarts = failed_restarts or set()
+ self.restart_count = 0
+ self.calls: list[tuple[str, ...]] = []
+ self.actions: list[tuple[str, ...]] = []
+
+ def _show(self) -> str:
+ binary = str(self.home / ".local/share/imp/current/.venv/bin/imp-gateway")
+ argv = (
+ binary,
+ "--host",
+ "127.0.0.1",
+ "--port",
+ "8788",
+ "--relay-url",
+ "ws://127.0.0.1:8787",
+ )
+ exec_start = (
+ f"{{ path={binary} ; argv[]={shlex.join(argv)} ; ignore_errors=no ; "
+ "start_time=[n/a] ; stop_time=[n/a] ; pid=0 }"
+ )
+ environment = f"{self.home}/.config/imp/gateway.env (ignore_errors=no)"
+ values = {
+ "LoadState": "loaded",
+ "FragmentPath": str(self.home / ".config/systemd/user/imp-gateway.service"),
+ "DropInPaths": "",
+ "NeedDaemonReload": "no",
+ "Type": "simple",
+ "ExecStart": exec_start,
+ "EnvironmentFiles": environment,
+ "Restart": "on-failure",
+ "RestartUSec": "2s",
+ "NoNewPrivileges": "yes",
+ "UnitFileState": "enabled" if self.enabled else "disabled",
+ "ActiveState": "active" if self.active else "inactive",
+ }
+ values.update(self.properties)
+ return "".join(f"{key}={value}\n" for key, value in values.items())
+
+ def __call__(self, *args: str) -> subprocess.CompletedProcess[str]:
+ self.calls.append(args)
+ op = args[0]
+ if op == "show":
+ return subprocess.CompletedProcess(args, 0, self._show(), "")
+ if op == "is-active":
+ return subprocess.CompletedProcess(args, 0 if self.active else 3, "", "")
+ if op == "is-enabled":
+ return subprocess.CompletedProcess(args, 0 if self.enabled else 1, "", "")
+ self.actions.append(args)
+ if op == "enable" and "--now" in args:
+ self.active = True
+ self.enabled = True
+ elif op == "enable":
+ self.enabled = True
+ elif op == "disable":
+ self.enabled = False
+ elif op == "restart":
+ self.restart_count += 1
+ if self.restart_count in self.failed_restarts:
+ return subprocess.CompletedProcess(args, 1, "", "private service error")
+ self.active = True
+ elif op == "stop":
+ self.active = False
+ else:
+ return subprocess.CompletedProcess(args, 1, "", "unsupported fake command")
+ return subprocess.CompletedProcess(args, 0, "", "")
+
+
+def _fake_systemctl(
+ home: Path,
+ monkeypatch: pytest.MonkeyPatch,
+ **kwargs: Any,
+) -> FakeSystemctl:
+ _install_unit(home)
+ fake = FakeSystemctl(home, **kwargs)
+ monkeypatch.setattr(direct_wss, "_systemctl", fake)
+
+ async def relay_ready(port: int, token: str | None = None) -> bool:
+ return True
+
+ monkeypatch.setattr(direct_wss, "_state_hello", relay_ready)
+ return fake
+
+
+def test_config_parser_accepts_only_one_digest_with_whitespace_and_comments() -> None:
+ digest = token_digest(TOKEN)
+ assert (
+ direct_wss._parse_config(
+ f" # supported comment\n\n {_config_line(digest.hex().upper())} \n".encode()
+ )
+ == digest
+ )
+
+ for content in (
+ b"",
+ f"{_config_line(digest.hex())}\n{_config_line(digest.hex())}\n".encode(),
+ f"{_config_line(digest.hex())}\nUNSUPPORTED=value\n".encode(),
+ f"export {_config_line(digest.hex())}\n".encode(),
+ f'{direct_wss._CONFIG_KEY}="{digest.hex()}"\n'.encode(),
+ f"{_config_line(digest.hex())} # inline comment\n".encode(),
+ b"IMP_GATEWAY_TOKEN_SHA256=not-a-digest\n",
+ b"IMP_GATEWAY_TOKEN_SHA256=\xff\n",
+ f"\u00a0{_config_line(digest.hex())}\n".encode(),
+ f"# comment\u2028{_config_line(digest.hex())}\n".encode(),
+ ):
+ with pytest.raises(direct_wss.DirectWssError):
+ direct_wss._parse_config(content)
+
+
+def _config_line(digest: str) -> str:
+ return f"{direct_wss._CONFIG_KEY} = {digest}"
+
+
+def test_setup_persists_only_digest_with_private_modes_and_hands_token_off_after_verify(
+ tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ home = _home(tmp_path, monkeypatch)
+ service = _fake_systemctl(home, monkeypatch)
+ monkeypatch.setattr(secrets, "token_urlsafe", lambda length: TOKEN)
+ verified: list[str] = []
+ monkeypatch.setattr(direct_wss, "_verify_gateway", lambda token: verified.append(token))
+ stdout, stderr = _tty_streams(monkeypatch)
+
+ assert direct_wss.main(["setup"]) == 0
+
+ config = home / ".config/imp/gateway.env"
+ assert config.read_bytes() == f"{direct_wss._CONFIG_KEY}={token_digest(TOKEN).hex()}\n".encode()
+ assert stat.S_IMODE(config.stat().st_mode) == 0o600
+ assert stat.S_IMODE(config.parent.stat().st_mode) == 0o700
+ assert stat.S_IMODE((config.parent / direct_wss._LOCK_NAME).stat().st_mode) == 0o600
+ assert verified == [TOKEN]
+ assert stdout.getvalue().endswith(TOKEN + "\n")
+ assert TOKEN not in config.read_text()
+ assert all(TOKEN not in argument for call in service.calls for argument in call)
+ assert TOKEN not in stderr.getvalue()
+ assert TOKEN not in " ".join(os.environ.values())
+ assert TOKEN not in " ".join(sys.argv)
+ assert service.active and service.enabled
+ assert not list(config.parent.glob(".gateway.env.tmp.*"))
+
+
+def test_setup_idempotent_does_not_generate_lock_or_call_service(
+ tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ home = _home(tmp_path, monkeypatch)
+ config = _write_config(home, f"{_config_line(token_digest(TOKEN).hex())}\n".encode())
+ original = config.read_bytes()
+ monkeypatch.setattr(secrets, "token_urlsafe", lambda length: pytest.fail("generated token"))
+ monkeypatch.setattr(direct_wss, "_systemctl", lambda *args: pytest.fail("called systemctl"))
+ monkeypatch.setattr(sys, "stdin", io.StringIO())
+ monkeypatch.setattr(sys, "stdout", io.StringIO())
+
+ assert direct_wss.main(["setup"]) == 0
+ assert config.read_bytes() == original
+ assert not (config.parent / direct_wss._LOCK_NAME).exists()
+
+
+@pytest.mark.parametrize(
+ "content",
+ [
+ b"",
+ b"IMP_GATEWAY_TOKEN_SHA256=" + b"a" * 64 + b"\nIMP_GATEWAY_TOKEN_SHA256=" + b"b" * 64 + b"\n",
+ b"IMP_GATEWAY_TOKEN_SHA256=" + b"a" * 64 + b"\nUNKNOWN=value\n",
+ b"export IMP_GATEWAY_TOKEN_SHA256=" + b"a" * 64 + b"\n",
+ b"IMP_GATEWAY_TOKEN_SHA256='" + b"a" * 64 + b"'\n",
+ ],
+)
+def test_setup_rejects_unsupported_config_unchanged(
+ tmp_path: Path, monkeypatch: pytest.MonkeyPatch, content: bytes
+) -> None:
+ home = _home(tmp_path, monkeypatch)
+ config = _write_config(home, content)
+ _tty_streams(monkeypatch)
+ monkeypatch.setattr(direct_wss, "_systemctl", lambda *args: pytest.fail("called systemctl"))
+
+ assert direct_wss.main(["setup"]) == 1
+ assert config.read_bytes() == content
+ assert not (config.parent / direct_wss._LOCK_NAME).exists()
+
+
+@pytest.mark.parametrize("operation", ["setup", "rotate", "status"])
+@pytest.mark.parametrize("nul_at", ["comment", "assignment_end", "assignment_inside"])
+def test_nul_in_config_is_invalid_for_every_operation(
+ tmp_path: Path, monkeypatch: pytest.MonkeyPatch, operation: str, nul_at: str
+) -> None:
+ home = _home(tmp_path, monkeypatch)
+ digest = token_digest(TOKEN).hex()
+ line = _config_line(digest)
+ content = {
+ "comment": f"# note\x00\n{line}\n",
+ "assignment_end": f"{line}\x00\n",
+ "assignment_inside": f"{line[:-1]}\x00{line[-1]}\n",
+ }[nul_at].encode()
+ config = _write_config(home, content)
+ before = config.stat()
+ service = _fake_systemctl(home, monkeypatch)
+ monkeypatch.setattr(secrets, "token_urlsafe", lambda length: pytest.fail("generated token"))
+ stdout, stderr = _tty_streams(monkeypatch)
+
+ assert direct_wss.main([operation]) == (0 if operation == "status" else 1)
+ output = stdout.getvalue() + stderr.getvalue()
+ if operation == "status":
+ assert "gateway.env syntax: invalid" in output
+ for secret in (digest, TOKEN, "\x00", "# note"):
+ assert secret not in output
+ after = config.stat()
+ assert config.read_bytes() == content
+ assert (after.st_ino, after.st_mtime_ns, stat.S_IMODE(after.st_mode)) == (
+ before.st_ino,
+ before.st_mtime_ns,
+ stat.S_IMODE(before.st_mode),
+ )
+ assert not (config.parent / direct_wss._LOCK_NAME).exists()
+ assert not service.actions
+
+
+def test_setup_rejects_permissive_config_file_mode_unchanged(
+ tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ home = _home(tmp_path, monkeypatch)
+ config = _write_config(home, f"{_config_line(token_digest(TOKEN).hex())}\n".encode())
+ os.chmod(config, 0o644)
+ before = config.read_bytes()
+ _tty_streams(monkeypatch)
+ monkeypatch.setattr(direct_wss, "_systemctl", lambda *args: pytest.fail("called systemctl"))
+
+ assert direct_wss.main(["setup"]) == 1
+ assert config.read_bytes() == before
+ assert stat.S_IMODE(config.stat().st_mode) == 0o644
+
+
+def test_setup_rejects_symlink_config_and_preserves_target(
+ tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ home = _home(tmp_path, monkeypatch)
+ config_dir = _config_dir(home)
+ target = tmp_path / "external.env"
+ target.write_bytes(b"external bytes\n")
+ (config_dir / "gateway.env").symlink_to(target)
+ _tty_streams(monkeypatch)
+ monkeypatch.setattr(direct_wss, "_systemctl", lambda *args: pytest.fail("called systemctl"))
+
+ assert direct_wss.main(["setup"]) == 1
+ assert target.read_bytes() == b"external bytes\n"
+ assert (config_dir / "gateway.env").is_symlink()
+
+
+def test_setup_rejects_symlink_config_ancestor_unchanged(
+ tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ home = _home(tmp_path, monkeypatch)
+ outside = tmp_path / "outside"
+ _write_config(outside, f"{_config_line(token_digest(TOKEN).hex())}\n".encode())
+ (home / ".config").symlink_to(outside / ".config", target_is_directory=True)
+ _tty_streams(monkeypatch)
+ monkeypatch.setattr(direct_wss, "_systemctl", lambda *args: pytest.fail("called systemctl"))
+
+ assert direct_wss.main(["setup"]) == 1
+ assert not (outside / ".config/imp/gateway.env.lock").exists()
+
+
+def test_setup_rejects_wrong_file_owner(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+ home = _home(tmp_path, monkeypatch)
+ config = _write_config(home, f"{_config_line(token_digest(TOKEN).hex())}\n".encode())
+ original_fstat = os.fstat
+ target = config.stat()
+
+ def foreign_owner(fd: int) -> Any:
+ info = original_fstat(fd)
+ if (info.st_dev, info.st_ino) == (target.st_dev, target.st_ino):
+ return SimpleNamespace(
+ st_mode=info.st_mode,
+ st_uid=os.geteuid() + 1,
+ st_nlink=info.st_nlink,
+ )
+ return info
+
+ monkeypatch.setattr(os, "fstat", foreign_owner)
+ _tty_streams(monkeypatch)
+ monkeypatch.setattr(direct_wss, "_systemctl", lambda *args: pytest.fail("called systemctl"))
+
+ assert direct_wss.main(["setup"]) == 1
+ assert config.exists()
+
+
+def test_rotate_rejects_unsafe_lock_without_touching_target(
+ tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ home = _home(tmp_path, monkeypatch)
+ config = _write_config(home, f"{_config_line(token_digest(OLD_TOKEN).hex())}\n".encode())
+ target = tmp_path / "outside.lock"
+ target.write_bytes(b"leave alone")
+ (config.parent / direct_wss._LOCK_NAME).symlink_to(target)
+ _tty_streams(monkeypatch)
+ monkeypatch.setattr(secrets, "token_urlsafe", lambda length: pytest.fail("generated token"))
+
+ assert direct_wss.main(["rotate"]) == 1
+ assert target.read_bytes() == b"leave alone"
+ assert token_digest(OLD_TOKEN).hex().encode() in config.read_bytes()
+
+
+def test_atomic_replace_is_private_and_failure_keeps_previous_bytes(
+ tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ home = _home(tmp_path, monkeypatch)
+ config = _write_config(home, f"{_config_line(token_digest(OLD_TOKEN).hex())}\n".encode())
+ old_data = config.read_bytes()
+ service = _fake_systemctl(home, monkeypatch, active=False, enabled=False)
+ monkeypatch.setattr(secrets, "token_urlsafe", lambda length: TOKEN)
+ _tty_streams(monkeypatch)
+
+ def fail_replace(*args: Any, **kwargs: Any) -> None:
+ raise OSError("private replacement failure")
+
+ monkeypatch.setattr(os, "replace", fail_replace)
+ assert direct_wss.main(["rotate"]) == 1
+ assert config.read_bytes() == old_data
+ assert not list(config.parent.glob(".gateway.env.tmp.*"))
+ assert not service.actions
+
+
+def test_setup_failure_restores_missing_config_and_inactive_disabled_state(
+ tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ home = _home(tmp_path, monkeypatch)
+ service = _fake_systemctl(home, monkeypatch)
+ monkeypatch.setattr(secrets, "token_urlsafe", lambda length: TOKEN)
+ monkeypatch.setattr(
+ direct_wss,
+ "_verify_gateway",
+ lambda token: (_ for _ in ()).throw(direct_wss.DirectWssError("readiness failed")),
+ )
+ stdout, stderr = _tty_streams(monkeypatch)
+
+ assert direct_wss.main(["setup"]) == 1
+ assert not (home / ".config/imp/gateway.env").exists()
+ assert not stdout.getvalue()
+ assert TOKEN not in stderr.getvalue()
+ assert not service.active and not service.enabled
+ assert service.actions[-2:] == [("stop", direct_wss._UNIT), ("disable", direct_wss._UNIT)]
+
+
+def test_active_rotate_readiness_failure_restores_exact_config_and_service_state(
+ tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ home = _home(tmp_path, monkeypatch)
+ old_data = b"# keep comments\n" + f"{_config_line(token_digest(OLD_TOKEN).hex())}\n".encode()
+ config = _write_config(home, old_data)
+ service = _fake_systemctl(home, monkeypatch, active=True, enabled=True)
+ monkeypatch.setattr(secrets, "token_urlsafe", lambda length: TOKEN)
+ monkeypatch.setattr(
+ direct_wss,
+ "_verify_gateway",
+ lambda token: (_ for _ in ()).throw(direct_wss.DirectWssError("gateway readiness failed")),
+ )
+ stdout, stderr = _tty_streams(monkeypatch)
+
+ assert direct_wss.main(["rotate"]) == 1
+ assert config.read_bytes() == old_data
+ assert service.active and service.enabled
+ assert service.restart_count == 2
+ assert not stdout.getvalue()
+ assert TOKEN not in stderr.getvalue()
+
+
+def test_restore_failure_is_explicit_and_gateway_is_stopped(
+ tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ home = _home(tmp_path, monkeypatch)
+ config = _write_config(home, f"{_config_line(token_digest(OLD_TOKEN).hex())}\n".encode())
+ service = _fake_systemctl(home, monkeypatch, active=True, enabled=True, failed_restarts={2})
+ monkeypatch.setattr(secrets, "token_urlsafe", lambda length: TOKEN)
+ monkeypatch.setattr(
+ direct_wss,
+ "_verify_gateway",
+ lambda token: (_ for _ in ()).throw(direct_wss.DirectWssError("readiness failed")),
+ )
+ _, stderr = _tty_streams(monkeypatch)
+
+ assert direct_wss.main(["rotate"]) == 1
+ assert config.read_bytes() == f"{_config_line(token_digest(OLD_TOKEN).hex())}\n".encode()
+ assert "restoration failed" in stderr.getvalue()
+ assert TOKEN not in stderr.getvalue()
+ assert not service.active
+ assert service.actions[-1] == ("stop", direct_wss._UNIT)
+
+
+@pytest.mark.parametrize("enabled", [False, True])
+def test_inactive_rotate_preserves_enabled_state_without_activation_or_readiness(
+ tmp_path: Path, monkeypatch: pytest.MonkeyPatch, enabled: bool
+) -> None:
+ home = _home(tmp_path, monkeypatch)
+ config = _write_config(home, f"{_config_line(token_digest(OLD_TOKEN).hex())}\n".encode())
+ service = _fake_systemctl(home, monkeypatch, active=False, enabled=enabled)
+ monkeypatch.setattr(secrets, "token_urlsafe", lambda length: TOKEN)
+ monkeypatch.setattr(direct_wss, "_verify_gateway", lambda token: pytest.fail("readiness must be skipped"))
+ stdout, _ = _tty_streams(monkeypatch)
+
+ assert direct_wss.main(["rotate"]) == 0
+ assert direct_wss._parse_config(config.read_bytes()) == token_digest(TOKEN)
+ assert not service.active and service.enabled is enabled
+ assert not service.actions
+ assert stdout.getvalue().endswith(TOKEN + "\n")
+
+
+def test_rotate_requires_supported_existing_config(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+ home = _home(tmp_path, monkeypatch)
+ _config_dir(home)
+ _tty_streams(monkeypatch)
+ monkeypatch.setattr(direct_wss, "_systemctl", lambda *args: pytest.fail("called systemctl"))
+
+ assert direct_wss.main(["rotate"]) == 1
+ assert not (home / ".config/imp/gateway.env").exists()
+ assert not (home / ".config/imp/gateway.env.lock").exists()
+
+
+def test_tty_refusal_happens_before_mutation(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+ home = _home(tmp_path, monkeypatch)
+ _fake_systemctl(home, monkeypatch)
+ monkeypatch.setattr(sys, "stdin", io.StringIO())
+ monkeypatch.setattr(sys, "stdout", io.StringIO())
+ monkeypatch.setattr(secrets, "token_urlsafe", lambda length: pytest.fail("generated token"))
+
+ assert direct_wss.main(["setup"]) == 1
+ assert not (home / ".config/imp/gateway.env").exists()
+ assert not (home / ".config/imp/gateway.env.lock").exists()
+ assert not (home / ".config/imp").exists()
+
+
+def test_status_is_read_only_and_separates_syntax_from_file_permissions(
+ tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ home = _home(tmp_path, monkeypatch)
+ config = _write_config(home, f"{_config_line(token_digest(TOKEN).hex())}\n".encode())
+ os.chmod(config, 0o644)
+ service = _fake_systemctl(home, monkeypatch, active=False, enabled=True)
+ before = config.read_bytes()
+ out = io.StringIO()
+ monkeypatch.setattr(sys, "stdout", out)
+
+ assert direct_wss.main(["status"]) == 0
+ assert "gateway.env syntax: valid" in out.getvalue()
+ assert "gateway.env mode: 0644 not private" in out.getvalue()
+ assert "gateway.env directory mode: 0700 private" in out.getvalue()
+ assert "gateway service enabled: yes" in out.getvalue()
+ assert "gateway service active: no" in out.getvalue()
+ assert "gateway health: unavailable" in out.getvalue()
+ assert token_digest(TOKEN).hex() not in out.getvalue()
+ assert TOKEN not in out.getvalue()
+ assert config.read_bytes() == before
+ assert not (config.parent / direct_wss._LOCK_NAME).exists()
+ assert not service.actions
+
+
+def test_status_reports_malformed_config_without_exposing_content(
+ tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ home = _home(tmp_path, monkeypatch)
+ config = _write_config(home, b"SECRETISH unsupported content\n")
+ _fake_systemctl(home, monkeypatch)
+ out = io.StringIO()
+ monkeypatch.setattr(sys, "stdout", out)
+
+ assert direct_wss.main(["status"]) == 0
+ assert "gateway.env syntax: invalid" in out.getvalue()
+ assert "SECRETISH" not in out.getvalue()
+ assert not (config.parent / direct_wss._LOCK_NAME).exists()
+
+
+def test_systemd_preflight_accepts_only_shipped_manager_and_disk_policy(
+ tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ home = _home(tmp_path, monkeypatch)
+ service = _fake_systemctl(home, monkeypatch, active=True, enabled=False)
+ with direct_wss._open_home() as (_, path, home_fd):
+ assert direct_wss._service_preflight(home_fd, path) == direct_wss.ServiceState(True, False)
+
+ for name, value in (
+ ("NeedDaemonReload", "yes"),
+ ("FragmentPath", "/etc/systemd/user/imp-gateway.service"),
+ ("DropInPaths", "/tmp/override.conf"),
+ ("Restart", "always"),
+ ("EnvironmentFiles", "/tmp/other.env (ignore_errors=no)"),
+ ("ExecStart", "{ path=/usr/bin/other ; argv[]=/usr/bin/other }"),
+ ):
+ service.properties[name] = value
+ with direct_wss._open_home() as (_, path, home_fd), pytest.raises(direct_wss.DirectWssError):
+ direct_wss._service_preflight(home_fd, path)
+ service.properties.clear()
+
+
+def test_systemd_preflight_rejects_disk_drift_dropins_and_symlink_unit(
+ tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ home = _home(tmp_path, monkeypatch)
+ _fake_systemctl(home, monkeypatch)
+ unit_dir = home / ".config/systemd/user"
+ unit = unit_dir / direct_wss._UNIT
+ unit.write_bytes(direct_wss._SHIPPED_UNIT + b"# changed\n")
+ with direct_wss._open_home() as (_, path, home_fd), pytest.raises(direct_wss.DirectWssError):
+ direct_wss._service_preflight(home_fd, path)
+ unit.write_bytes(direct_wss._SHIPPED_UNIT)
+ (unit_dir / f"{direct_wss._UNIT}.d").mkdir()
+ with direct_wss._open_home() as (_, path, home_fd), pytest.raises(direct_wss.DirectWssError):
+ direct_wss._service_preflight(home_fd, path)
+ (unit_dir / f"{direct_wss._UNIT}.d").rmdir()
+ unit.unlink()
+ unit.symlink_to(tmp_path / "missing-unit")
+ with direct_wss._open_home() as (_, path, home_fd), pytest.raises(direct_wss.DirectWssError):
+ direct_wss._service_preflight(home_fd, path)
+
+
+def test_concurrent_setup_issues_one_credential_without_overwriting(
+ tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ home = _home(tmp_path, monkeypatch)
+ service = _fake_systemctl(home, monkeypatch)
+ _tty_streams(monkeypatch)
+ entered = threading.Event()
+ second_waiting = threading.Event()
+ release = threading.Event()
+ generated: list[str] = []
+ outputs = [TTY(), TTY()]
+ failures: list[BaseException] = []
+ original_lock = direct_wss._lock_mutations
+
+ def lock(dir_fd: int, uid: int) -> int:
+ if threading.current_thread().name == "second-setup":
+ second_waiting.set()
+ return original_lock(dir_fd, uid)
+
+ def generate(length: int) -> str:
+ generated.append(TOKEN)
+ entered.set()
+ assert release.wait(timeout=3)
+ return TOKEN
+
+ def setup(output: TTY) -> None:
+ try:
+ direct_wss._mutate("setup", output)
+ except BaseException as error:
+ failures.append(error)
+
+ monkeypatch.setattr(direct_wss, "_lock_mutations", lock)
+ monkeypatch.setattr(secrets, "token_urlsafe", generate)
+ monkeypatch.setattr(direct_wss, "_verify_gateway", lambda token: None)
+ first = threading.Thread(target=setup, args=(outputs[0],))
+ second = threading.Thread(target=setup, args=(outputs[1],), name="second-setup")
+ first.start()
+ try:
+ assert entered.wait(timeout=3)
+ second.start()
+ assert second_waiting.wait(timeout=3)
+ finally:
+ release.set()
+ first.join(timeout=3)
+ if second.ident is not None:
+ second.join(timeout=3)
+ assert not failures and not first.is_alive() and not second.is_alive()
+ assert generated == [TOKEN]
+ assert outputs[0].getvalue().count(TOKEN) == 1
+ assert TOKEN not in outputs[1].getvalue()
+ assert direct_wss._parse_config((home / ".config/imp/gateway.env").read_bytes()) == token_digest(TOKEN)
+ assert service.actions == [("enable", "--now", direct_wss._UNIT)]
+
+
+def test_created_modes_ignore_a_permissive_umask(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+ home = _home(tmp_path, monkeypatch)
+ service = _fake_systemctl(home, monkeypatch)
+ monkeypatch.setattr(secrets, "token_urlsafe", lambda length: TOKEN)
+ monkeypatch.setattr(direct_wss, "_verify_gateway", lambda token: None)
+ _tty_streams(monkeypatch)
+ old_umask = os.umask(0)
+ try:
+ assert direct_wss.main(["setup"]) == 0
+ finally:
+ os.umask(old_umask)
+
+ config_dir = home / ".config/imp"
+ assert stat.S_IMODE(config_dir.stat().st_mode) == 0o700
+ assert stat.S_IMODE((config_dir / "gateway.env").stat().st_mode) == 0o600
+ assert stat.S_IMODE((config_dir / direct_wss._LOCK_NAME).stat().st_mode) == 0o600
+ assert service.active and service.enabled
+
+
+def test_setup_handles_stdout_failure_without_handing_off_or_retaining_new_credentials(
+ tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ home = _home(tmp_path, monkeypatch)
+ service = _fake_systemctl(home, monkeypatch)
+ monkeypatch.setattr(secrets, "token_urlsafe", lambda length: TOKEN)
+ monkeypatch.setattr(direct_wss, "_verify_gateway", lambda token: None)
+
+ class BrokenTTY(TTY):
+ def write(self, value: str) -> int:
+ if value.startswith("Direct WSS pairing token"):
+ return super().write(value)
+ raise BrokenPipeError
+
+ monkeypatch.setattr(sys, "stdin", TTY())
+ stdout = BrokenTTY()
+ stderr = io.StringIO()
+ monkeypatch.setattr(sys, "stdout", stdout)
+ monkeypatch.setattr(sys, "stderr", stderr)
+
+ assert direct_wss.main(["setup"]) == 1
+ assert TOKEN not in stdout.getvalue()
+ assert TOKEN not in stderr.getvalue()
+ assert not (home / ".config/imp/gateway.env").exists()
+ assert not service.active and not service.enabled
+
+
+def test_staged_file_sync_failure_leaves_previous_bytes_unchanged(
+ tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ home = _home(tmp_path, monkeypatch)
+ old_data = f"{_config_line(token_digest(OLD_TOKEN).hex())}\n".encode()
+ config = _write_config(home, old_data)
+ _fake_systemctl(home, monkeypatch, active=False, enabled=True)
+ monkeypatch.setattr(secrets, "token_urlsafe", lambda length: TOKEN)
+ _tty_streams(monkeypatch)
+
+ def fail_sync(fd: int) -> None:
+ raise OSError("sync failure")
+
+ monkeypatch.setattr(os, "fsync", fail_sync)
+
+ assert direct_wss.main(["rotate"]) == 1
+ assert config.read_bytes() == old_data
+ assert not list(config.parent.glob(".gateway.env.tmp.*"))
+
+
+def test_config_restore_failure_is_explicit_and_stops_active_gateway(
+ tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ home = _home(tmp_path, monkeypatch)
+ old_data = f"{_config_line(token_digest(OLD_TOKEN).hex())}\n".encode()
+ config = _write_config(home, old_data)
+ service = _fake_systemctl(home, monkeypatch, active=True, enabled=True)
+ monkeypatch.setattr(secrets, "token_urlsafe", lambda length: TOKEN)
+ monkeypatch.setattr(
+ direct_wss,
+ "_verify_gateway",
+ lambda token: (_ for _ in ()).throw(direct_wss.DirectWssError("readiness failed")),
+ )
+ _, stderr = _tty_streams(monkeypatch)
+ original_replace = os.replace
+ replacements = 0
+
+ def fail_rollback_replace(source: str, destination: str, **kwargs: Any) -> None:
+ nonlocal replacements
+ replacements += 1
+ if replacements == 2:
+ raise OSError("rollback sync failed")
+ original_replace(source, destination, **kwargs)
+
+ monkeypatch.setattr(os, "replace", fail_rollback_replace)
+
+ assert direct_wss.main(["rotate"]) == 1
+ assert replacements == 2
+ assert "restoration failed" in stderr.getvalue()
+ assert "gateway stop was attempted" in stderr.getvalue()
+ assert not service.active and service.enabled
+ assert token_digest(TOKEN).hex().encode() in config.read_bytes()
+
+
+def test_unsafe_gateway_env_file_types_and_config_directory_are_rejected(
+ tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ home = _home(tmp_path, monkeypatch)
+ config_dir = _config_dir(home)
+ config = config_dir / "gateway.env"
+ config.mkdir()
+ _tty_streams(monkeypatch)
+ monkeypatch.setattr(direct_wss, "_systemctl", lambda *args: pytest.fail("called systemctl"))
+
+ assert direct_wss.main(["setup"]) == 1
+ assert config.is_dir()
+ config.rmdir()
+ os.mkfifo(config)
+ assert direct_wss.main(["setup"]) == 1
+ assert config.is_fifo()
+ config.unlink()
+ config.write_bytes(f"{_config_line(token_digest(TOKEN).hex())}\n".encode())
+ os.chmod(config, 0o600)
+ os.chmod(config_dir, 0o755)
+ assert direct_wss.main(["setup"]) == 1
+ assert stat.S_IMODE(config_dir.stat().st_mode) == 0o755
+ assert config.exists()
+
+
+def test_systemd_preflight_rejects_masked_linked_and_mismatched_unit_states(
+ tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ home = _home(tmp_path, monkeypatch)
+ service = _fake_systemctl(home, monkeypatch)
+ for name, value in (("LoadState", "masked"), ("UnitFileState", "linked"), ("ActiveState", "failed")):
+ service.properties[name] = value
+ with direct_wss._open_home() as (_, path, home_fd), pytest.raises(direct_wss.DirectWssError):
+ direct_wss._service_preflight(home_fd, path)
+ service.properties.clear()
+
+
+def test_symlinked_home_is_rejected_before_any_mutation(
+ tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ actual_home = tmp_path / "actual-home"
+ actual_home.mkdir()
+ alias = tmp_path / "home-link"
+ alias.symlink_to(actual_home, target_is_directory=True)
+ monkeypatch.setattr(direct_wss, "_supported_platform", lambda: None)
+ monkeypatch.setattr(direct_wss, "_user_home", lambda: (os.geteuid(), str(alias)))
+ _tty_streams(monkeypatch)
+
+ assert direct_wss.main(["setup"]) == 1
+ assert list(actual_home.iterdir()) == []
+
+
+def test_status_command_never_creates_config_directory(
+ tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ home = _home(tmp_path, monkeypatch)
+ _install_unit(home)
+ fake = FakeSystemctl(home)
+ monkeypatch.setattr(direct_wss, "_systemctl", fake)
+ monkeypatch.setattr(direct_wss, "_gateway_health", lambda: False)
+ monkeypatch.setattr(sys, "stdout", io.StringIO())
+
+ assert direct_wss.main(["status"]) == 0
+ assert not (home / ".config/imp").exists()
+
+
+def test_setup_refuses_active_gateway_without_generation_or_service_change(
+ tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ home = _home(tmp_path, monkeypatch)
+ service = _fake_systemctl(home, monkeypatch, active=True, enabled=True)
+ _tty_streams(monkeypatch)
+ monkeypatch.setattr(secrets, "token_urlsafe", lambda length: pytest.fail("generated token"))
+
+ assert direct_wss.main(["setup"]) == 1
+ assert not (home / ".config/imp/gateway.env").exists()
+ assert not service.actions
+
+
+def test_stdout_token_handoff_failure_rolls_back_active_rotation(
+ tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ home = _home(tmp_path, monkeypatch)
+ old_data = f"{_config_line(token_digest(OLD_TOKEN).hex())}\n".encode()
+ config = _write_config(home, old_data)
+ service = _fake_systemctl(home, monkeypatch, active=True, enabled=False)
+ monkeypatch.setattr(secrets, "token_urlsafe", lambda length: TOKEN)
+ monkeypatch.setattr(direct_wss, "_verify_gateway", lambda token: None)
+
+ class BrokenTTY(TTY):
+ def write(self, value: str) -> int:
+ if value.startswith("Direct WSS pairing token"):
+ return super().write(value)
+ raise BrokenPipeError
+
+ monkeypatch.setattr(sys, "stdin", TTY())
+ monkeypatch.setattr(sys, "stdout", BrokenTTY())
+ stderr = io.StringIO()
+ monkeypatch.setattr(sys, "stderr", stderr)
+
+ assert direct_wss.main(["rotate"]) == 1
+ assert config.read_bytes() == old_data
+ assert service.active and not service.enabled
+ assert TOKEN not in stderr.getvalue()
+
+
+def test_readiness_interrupt_restores_config_and_service_state(
+ tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ home = _home(tmp_path, monkeypatch)
+ old_data = f"{_config_line(token_digest(OLD_TOKEN).hex())}\n".encode()
+ config = _write_config(home, old_data)
+ service = _fake_systemctl(home, monkeypatch, active=True, enabled=True)
+ monkeypatch.setattr(secrets, "token_urlsafe", lambda length: TOKEN)
+ monkeypatch.setattr(
+ direct_wss,
+ "_verify_gateway",
+ lambda token: (_ for _ in ()).throw(KeyboardInterrupt()),
+ )
+ stdout, stderr = _tty_streams(monkeypatch)
+
+ assert direct_wss.main(["rotate"]) == 1
+ assert config.read_bytes() == old_data
+ assert service.active and service.enabled
+ assert not stdout.getvalue()
+ assert "interrupted" in stderr.getvalue()
+ assert TOKEN not in stderr.getvalue()
+
+
+def test_captured_systemctl_errors_never_echo_credentials(
+ tmp_path: Path, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+ home = _home(tmp_path, monkeypatch)
+ service = _fake_systemctl(home, monkeypatch)
+ original = service
+
+ def failing_start(*args: str) -> subprocess.CompletedProcess[str]:
+ result = original(*args)
+ if args[0] == "enable" and "--now" in args:
+ return subprocess.CompletedProcess(args, 1, "", TOKEN)
+ return result
+
+ monkeypatch.setattr(direct_wss, "_systemctl", failing_start)
+ monkeypatch.setattr(secrets, "token_urlsafe", lambda length: TOKEN)
+ monkeypatch.setattr(direct_wss, "_verify_gateway", lambda token: None)
+ stdout, stderr = _tty_streams(monkeypatch)
+
+ assert direct_wss.main(["setup"]) == 1
+ assert not stdout.getvalue()
+ assert TOKEN not in stderr.getvalue()
+ assert TOKEN not in caplog.text
+ assert TOKEN not in " ".join(os.environ.values())
+ assert not (home / ".config/imp/gateway.env").exists()
+ assert not original.active and not original.enabled
+
+
+def test_cli_platform_guard_rejects_windows_and_root(monkeypatch: pytest.MonkeyPatch) -> None:
+ stderr = io.StringIO()
+ monkeypatch.setattr(sys, "stderr", stderr)
+ monkeypatch.setattr(sys, "platform", "win32")
+ monkeypatch.setattr(sys, "stdout", io.StringIO())
+ assert direct_wss.main(["status"]) == 1
+
+ monkeypatch.setattr(sys, "platform", "linux")
+ monkeypatch.setattr(os, "geteuid", lambda: 0)
+ assert direct_wss.main(["status"]) == 1
+
+
+def test_systemd_execstart_uses_one_quoted_argv_field_and_exact_environment_file() -> None:
+ home = "/home/test user"
+ binary = f"{home}/.local/share/imp/current/.venv/bin/imp-gateway"
+ argv = (
+ binary,
+ "--host",
+ "127.0.0.1",
+ "--port",
+ "8788",
+ "--relay-url",
+ "ws://127.0.0.1:8787",
+ )
+ property_value = f"{{ path={binary} ; argv[]={shlex.join(argv)} ; ignore_errors=no }}"
+
+ assert direct_wss._expected_exec_start(property_value, home)
+ assert direct_wss._expected_environment_files(f"{home}/.config/imp/gateway.env (ignore_errors=no)", home)
+ repeated_fields = f"{{ path={binary} ; argv[]={binary} ; argv[]=--host ; ignore_errors=no }}"
+ assert not direct_wss._expected_exec_start(repeated_fields, home)
+
+
+def test_gateway_health_requires_exact_minimal_response(monkeypatch: pytest.MonkeyPatch) -> None:
+ class Handler(http.server.BaseHTTPRequestHandler):
+ body = b'{"status":"ok"}'
+ status = 200
+
+ def do_GET(self) -> None:
+ self.send_response(self.status)
+ self.send_header("Content-Length", str(len(self.body)))
+ self.end_headers()
+ self.wfile.write(self.body)
+
+ def log_message(self, format: str, *args: Any) -> None:
+ return
+
+ server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+ thread = threading.Thread(target=server.serve_forever, daemon=True)
+ monkeypatch.setattr(direct_wss, "_GATEWAY_PORT", server.server_address[1])
+ thread.start()
+ try:
+ assert direct_wss._gateway_health()
+ Handler.body = b'{"status":"ok","extra":true}'
+ assert not direct_wss._gateway_health()
+ Handler.body = b'{"status":"bad","status":"ok"}'
+ assert not direct_wss._gateway_health()
+ Handler.status = 503
+ Handler.body = b'{"status":"ok"}'
+ assert not direct_wss._gateway_health()
+ finally:
+ server.shutdown()
+ server.server_close()
+ thread.join(timeout=3)
+
+
+def test_unavailable_relay_aborts_before_token_generation_or_activation(
+ tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ home = _home(tmp_path, monkeypatch)
+ service = _fake_systemctl(home, monkeypatch)
+ _tty_streams(monkeypatch)
+
+ async def unavailable(port: int, token: str | None = None) -> bool:
+ return False
+
+ monkeypatch.setattr(direct_wss, "_state_hello", unavailable)
+ monkeypatch.setattr(secrets, "token_urlsafe", lambda length: pytest.fail("generated token"))
+ assert direct_wss.main(["setup"]) == 1
+ assert not service.actions
+ assert not (home / ".config/imp/gateway.env").exists()
+
+
+def test_generated_token_has_canonical_256_bit_encoding_and_textual_digest(
+ tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ home = _home(tmp_path, monkeypatch)
+ _fake_systemctl(home, monkeypatch)
+ monkeypatch.setattr(direct_wss, "_verify_gateway", lambda token: None)
+ stdout, _ = _tty_streams(monkeypatch)
+ assert direct_wss.main(["setup"]) == 0
+ token = stdout.getvalue().splitlines()[-1]
+ raw = base64.urlsafe_b64decode(token + "=")
+ assert len(token) == 43 and len(raw) == 32
+ assert base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii") == token
+ config = (home / ".config/imp/gateway.env").read_bytes()
+ assert direct_wss._parse_config(config) == hashlib.sha256(token.encode("ascii")).digest()
+ assert token.encode("ascii") not in config
diff --git a/services/relay/tests/test_gateway.py b/services/relay/tests/test_gateway.py
index d336f29..db0771d 100644
--- a/services/relay/tests/test_gateway.py
+++ b/services/relay/tests/test_gateway.py
@@ -3,8 +3,17 @@
import asyncio
import base64
import hashlib
+import io
import json
+import logging
+import os
+import secrets
import socket
+import subprocess
+import sys
+import threading
+import time
+from pathlib import Path
from typing import Any, cast
import pytest
@@ -12,6 +21,7 @@
from websockets.exceptions import ConnectionClosed, ConnectionClosedOK, InvalidStatus
from websockets.typing import Origin
+import imp_relay.direct_wss as direct_wss
import imp_relay.gateway as gateway_module
from imp_relay.gateway import (
AUTH_POLICY_CLOSE_CODE,
@@ -601,3 +611,235 @@ async def send_snapshots() -> None:
await relay.close()
asyncio.run(scenario())
+
+
+class _InteractiveStream(io.StringIO):
+ def isatty(self) -> bool:
+ return True
+
+
+class _GatewayHarness:
+ def __init__(self, gateway_port: int, relay_port: int) -> None:
+ self.gateway_port = gateway_port
+ self.relay_port = relay_port
+ self.loop = asyncio.new_event_loop()
+ self.ready = threading.Event()
+ self.thread = threading.Thread(target=self._run, daemon=True)
+ self.relay: RelayServer | None = None
+ self.gateway: GatewayServer | None = None
+
+ def _run(self) -> None:
+ asyncio.set_event_loop(self.loop)
+ self.ready.set()
+ self.loop.run_forever()
+
+ def call(self, coroutine: Any) -> Any:
+ return asyncio.run_coroutine_threadsafe(coroutine, self.loop).result(timeout=5)
+
+ def start(self, digest: bytes, context: StateContext) -> None:
+ self.thread.start()
+ assert self.ready.wait(timeout=5)
+ self.call(self._start(digest, context))
+
+ async def _start(self, digest: bytes, context: StateContext) -> None:
+ self.relay = RelayServer(port=self.relay_port)
+ self.relay.state.apply_select(context, _state("Before rotation"), now=1.0)
+ await self.relay.start()
+ await self._replace_gateway(digest)
+
+ async def _replace_gateway(self, digest: bytes) -> None:
+ if self.gateway is not None:
+ await self.gateway.close()
+ self.gateway = GatewayServer(
+ digest,
+ port=self.gateway_port,
+ relay_url=f"ws://127.0.0.1:{self.relay_port}",
+ )
+ await self.gateway.start()
+
+ async def replace_digest(self, digest: bytes) -> None:
+ await self._replace_gateway(digest)
+
+ async def select(self, context: StateContext) -> None:
+ assert self.relay is not None
+ self.relay.state.apply_select(context, _state("After rotation"), now=2.0)
+
+ async def _close(self) -> None:
+ if self.gateway is not None:
+ await self.gateway.close()
+ if self.relay is not None:
+ await self.relay.close()
+
+ def close(self) -> None:
+ self.call(self._close())
+ self.loop.call_soon_threadsafe(self.loop.stop)
+ self.thread.join(timeout=5)
+ self.loop.close()
+
+
+def test_direct_wss_rotation_replaces_gateway_auth_session_and_preserves_relay_semantics(
+ tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+ relay_blocker, relay_port = _reserve_unavailable_port()
+ gateway_blocker, gateway_port = _reserve_unavailable_port()
+ relay_blocker.close()
+ gateway_blocker.close()
+ old_context = StateContext("old_session", 1, 1)
+ new_context = StateContext("new_session", 2, 2)
+ harness = _GatewayHarness(gateway_port, relay_port)
+ home = tmp_path / "home"
+ home.mkdir()
+ config_dir = home / ".config" / "imp"
+ config_dir.mkdir(parents=True)
+ os.chmod(home / ".config", 0o700)
+ os.chmod(config_dir, 0o700)
+ config = config_dir / "gateway.env"
+ config.write_text(f"IMP_GATEWAY_TOKEN_SHA256={token_digest(WRONG_TOKEN).hex()}\n")
+ os.chmod(config, 0o600)
+
+ monkeypatch.setattr(direct_wss, "_supported_platform", lambda: None)
+ monkeypatch.setattr(direct_wss, "_user_home", lambda: (os.geteuid(), str(home)))
+ monkeypatch.setattr(direct_wss, "_GATEWAY_PORT", gateway_port)
+ monkeypatch.setattr(direct_wss, "_RELAY_PORT", relay_port)
+ monkeypatch.setattr(
+ direct_wss,
+ "_service_preflight",
+ lambda home_fd, path: direct_wss.ServiceState(True, True),
+ )
+ monkeypatch.setattr(secrets, "token_urlsafe", lambda length: TOKEN)
+ monkeypatch.setattr(sys, "stdin", _InteractiveStream())
+ output = _InteractiveStream()
+ monkeypatch.setattr(sys, "stdout", _InteractiveStream())
+
+ def restart_gateway(*args: str) -> subprocess.CompletedProcess[str]:
+ if args[0] == "restart":
+ harness.call(harness.replace_digest(direct_wss._parse_config(config.read_bytes())))
+ return subprocess.CompletedProcess(args, 0, "", "")
+
+ monkeypatch.setattr(direct_wss, "_systemctl", restart_gateway)
+
+ async def scenario() -> None:
+ try:
+ await asyncio.to_thread(harness.start, token_digest(WRONG_TOKEN), old_context)
+ except BaseException:
+ await asyncio.to_thread(harness.close)
+ raise
+ try:
+ async with connect(f"ws://127.0.0.1:{gateway_port}/state") as previous_session:
+ await previous_session.send(_auth(WRONG_TOKEN))
+ initial_hello = await _receive_type(previous_session, "hello")
+ initial_snapshot = await _receive_type(previous_session, "snapshot")
+ assert initial_hello["protocol"] == 2
+ assert initial_snapshot["context"] == {
+ "session": "old_session",
+ "foreground": 1,
+ "connection": 1,
+ }
+ await asyncio.to_thread(harness.call, harness.select(new_context))
+
+ await asyncio.to_thread(direct_wss._mutate, "rotate", output)
+ rotated_token = output.getvalue().strip().splitlines()[-1]
+ assert valid_pairing_token(rotated_token)
+ assert config.read_text() == (
+ f"IMP_GATEWAY_TOKEN_SHA256={token_digest(rotated_token).hex()}\n"
+ )
+
+ async with asyncio.timeout(5):
+ while True:
+ try:
+ await previous_session.recv()
+ except ConnectionClosed:
+ break
+
+ async with connect(f"ws://127.0.0.1:{gateway_port}/state") as rejected:
+ await rejected.send(_auth(WRONG_TOKEN))
+ with pytest.raises(ConnectionClosed):
+ await rejected.recv()
+ assert rejected.close_code == AUTH_POLICY_CLOSE_CODE
+
+ async with connect(f"ws://127.0.0.1:{gateway_port}/state") as current:
+ await current.send(_auth(rotated_token))
+ hello = await _receive_type(current, "hello")
+ snapshot = await _receive_type(current, "snapshot")
+ assert hello["protocol"] == 2
+ assert snapshot["context"] == {
+ "session": "new_session",
+ "foreground": 2,
+ "connection": 2,
+ }
+ state = cast(dict[str, object], snapshot["state"])
+ character = cast(dict[str, object], state["character"])
+ assert character["name"] == "After rotation"
+
+ async with connect(f"ws://127.0.0.1:{gateway_port}/action") as rejected_action:
+ await rejected_action.send(_auth(WRONG_TOKEN))
+ with pytest.raises(ConnectionClosed):
+ await rejected_action.recv()
+ assert rejected_action.close_code == AUTH_POLICY_CLOSE_CODE
+
+ async with connect(f"ws://127.0.0.1:{gateway_port}/action") as stale_action:
+ await stale_action.send(_auth(rotated_token))
+ await stale_action.send(encode_action(old_context, "look"))
+ stale_result = await _receive_type(stale_action, "action-result")
+ assert stale_result["status"] == "rejected"
+
+ async with connect(f"ws://127.0.0.1:{relay_port}/action-consumer") as consumer:
+ await consumer.send(encode_consumer(new_context))
+ await _receive_type(consumer, "consumer-ready")
+ async with connect(f"ws://127.0.0.1:{gateway_port}/action") as current_action:
+ await current_action.send(_auth(rotated_token))
+ await current_action.send(encode_action(new_context, "look"))
+ dispatch = await _receive_type(consumer, "dispatch")
+ assert dispatch["command"] == "look"
+ await consumer.send(encode_consumer_result(cast(str, dispatch["id"]), "forwarded"))
+ result = await _receive_type(current_action, "action-result")
+ assert result["status"] == "forwarded"
+
+ status_output = _InteractiveStream()
+ monkeypatch.setattr(sys, "stdout", status_output)
+ assert direct_wss.main(["status"]) == 0
+ assert "gateway service active: yes" in status_output.getvalue()
+ assert "gateway health: healthy" in status_output.getvalue()
+ assert token_digest(rotated_token).hex() not in status_output.getvalue()
+ finally:
+ await asyncio.to_thread(harness.close)
+
+ asyncio.run(scenario())
+
+
+def test_direct_wss_readiness_waits_for_startup_without_logging_credentials(
+ monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+ relay_blocker, relay_port = _reserve_unavailable_port()
+ gateway_blocker, gateway_port = _reserve_unavailable_port()
+ relay_blocker.close()
+ gateway_blocker.close()
+ harness = _GatewayHarness(gateway_port, relay_port)
+ monkeypatch.setattr(direct_wss, "_GATEWAY_PORT", gateway_port)
+ monkeypatch.setattr(direct_wss, "_TIMEOUT", 1)
+ monkeypatch.setenv("ALL_PROXY", "http://127.0.0.1:1")
+ caplog.set_level(logging.DEBUG)
+ caplog.set_level(logging.DEBUG, logger="websockets.client")
+ errors: list[BaseException] = []
+
+ def delayed_start() -> None:
+ try:
+ time.sleep(0.15)
+ harness.start(token_digest(TOKEN), CONTEXT)
+ except BaseException as error:
+ errors.append(error)
+
+ starter = threading.Thread(target=delayed_start)
+ starter.start()
+ try:
+ direct_wss._verify_gateway(TOKEN)
+ assert not errors
+ assert asyncio.run(direct_wss._state_hello(gateway_port, TOKEN))
+ assert TOKEN not in caplog.text
+ finally:
+ starter.join(timeout=5)
+ if harness.thread.is_alive():
+ harness.close()
+ monkeypatch.setattr(direct_wss, "_TIMEOUT", 0.1)
+ with pytest.raises(direct_wss.DirectWssError):
+ direct_wss._verify_gateway(TOKEN)
diff --git a/services/relay/tests/test_public_verifier.py b/services/relay/tests/test_public_verifier.py
new file mode 100644
index 0000000..2e2c896
--- /dev/null
+++ b/services/relay/tests/test_public_verifier.py
@@ -0,0 +1,186 @@
+from __future__ import annotations
+
+import ast
+import builtins
+import getpass
+import http.server
+import io
+import os
+import signal
+import sys
+import termios
+import threading
+import time
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+TOKEN = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
+
+
+def _snippet() -> str:
+ readme = Path(__file__).resolve().parents[3] / "deploy" / "README.md"
+ section = readme.read_text().split("### Bounded public verification", 1)[1]
+ return section.split("```python\n", 1)[1].split("```", 1)[0]
+
+
+@pytest.mark.parametrize("failed_prompt", [1, 2])
+def test_secure_prompt_failure_never_reads_fallback_or_uses_network(
+ monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], failed_prompt: int
+) -> None:
+ original_open = os.open
+ real_getpass = getpass.getpass
+ calls = 0
+
+ def prompt(*args: Any, **kwargs: Any) -> str:
+ nonlocal calls
+ calls += 1
+ return real_getpass(*args, **kwargs) if calls == failed_prompt else TOKEN
+
+ def echo_failure(fd: int) -> None:
+ raise termios.error("injected echo-control failure")
+
+ def forbidden(*args: Any, **kwargs: Any) -> Any:
+ pytest.fail("fallback credential read or network activity")
+
+ monkeypatch.setattr(builtins, "open", lambda *args, **kwargs: io.StringIO())
+ monkeypatch.setattr(os, "open", lambda *args, **kwargs: original_open(os.devnull, os.O_RDWR))
+ monkeypatch.setattr(termios, "tcgetattr", echo_failure)
+ monkeypatch.setattr(getpass, "getpass", prompt)
+ monkeypatch.setattr(getpass, "_raw_input", forbidden)
+ monkeypatch.setattr("websockets.sync.client.connect", forbidden)
+ monkeypatch.setattr("urllib.request.OpenerDirector.open", forbidden)
+ monkeypatch.setattr(sys, "argv", ["public_wss_check.py", "wss://localhost/state"])
+
+ with pytest.raises(SystemExit) as exc:
+ exec(compile(_snippet(), "public_wss_check.py", "exec"), {})
+
+ assert exc.value.code == 1
+ assert calls == failed_prompt
+ output = capsys.readouterr()
+ assert output.out == "Verification failed.\n"
+ assert output.err == ""
+ assert TOKEN not in output.out + output.err
+
+
+def _definitions() -> dict[str, Any]:
+ # Execute the snippet's imports/definitions, stopping before interactive input.
+ tree = ast.parse(_snippet())
+ boundary = next(i for i, node in enumerate(tree.body) if isinstance(node, ast.If))
+ tree.body = tree.body[:boundary]
+ namespace: dict[str, Any] = {}
+ exec(compile(tree, "public_wss_check.py", "exec"), namespace)
+ namespace["opener"] = namespace["build_opener"](
+ namespace["ProxyHandler"]({}), namespace["HTTPSHandler"](), namespace["NoRedirect"]()
+ )
+ return namespace
+
+
+@pytest.mark.parametrize("interval", [0.0, 60.0])
+def test_http_refuses_armed_real_timer_without_changing_it(
+ monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], interval: float
+) -> None:
+ namespace = _definitions()
+
+ def forbidden(*args: Any, **kwargs: Any) -> Any:
+ pytest.fail("network activity")
+
+ def handler(signum: int, frame: Any) -> None:
+ pass
+
+ monkeypatch.setattr("urllib.request.OpenerDirector.open", forbidden)
+ previous_handler = signal.signal(signal.SIGALRM, handler)
+ try:
+ signal.setitimer(signal.ITIMER_REAL, 30.0, interval)
+ with pytest.raises(SystemExit) as exc:
+ namespace["http_get"]("https://127.0.0.1/healthz")
+ delay, remaining_interval = signal.getitimer(signal.ITIMER_REAL)
+ assert exc.value.code == 1
+ assert 29.0 < delay <= 30.0
+ assert remaining_interval == pytest.approx(interval)
+ assert signal.getsignal(signal.SIGALRM) is handler
+ assert capsys.readouterr().out == "Verification failed.\n"
+ finally:
+ signal.setitimer(signal.ITIMER_REAL, 0)
+ signal.signal(signal.SIGALRM, previous_handler)
+
+
+@pytest.mark.parametrize("status", [200, 302])
+def test_http_fast_response_and_redirect_release_timer_on_repeat(status: int) -> None:
+ namespace = _definitions()
+
+ class Handler(http.server.BaseHTTPRequestHandler):
+ def do_GET(self) -> None:
+ self.send_response(status)
+ self.send_header("Location", "/elsewhere")
+ self.send_header("Content-Length", "15")
+ self.end_headers()
+ self.wfile.write(b'{"status":"ok"}')
+
+ def log_message(self, format: str, *args: Any) -> None:
+ pass
+
+ server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+ thread = threading.Thread(target=server.serve_forever)
+ thread.start()
+ previous_handler = signal.getsignal(signal.SIGALRM)
+ try:
+ for _ in range(2):
+ result = namespace["http_get"](f"http://127.0.0.1:{server.server_port}/healthz")
+ assert result == ((200, b'{"status":"ok"}') if status == 200 else (302, b""))
+ assert signal.getsignal(signal.SIGALRM) == previous_handler
+ assert signal.getitimer(signal.ITIMER_REAL) == (0.0, 0.0)
+ finally:
+ server.shutdown()
+ server.server_close()
+ thread.join()
+
+
+@pytest.mark.parametrize("drip", ["headers", "body"])
+def test_http_deadline_interrupts_drip_fed_response(capsys: pytest.CaptureFixture[str], drip: str) -> None:
+ namespace = _definitions()
+ stop = threading.Event()
+ sent = 0
+
+ class Handler(http.server.BaseHTTPRequestHandler):
+ def do_GET(self) -> None:
+ nonlocal sent
+ header = b"HTTP/1.1 200 OK\r\nContent-Length: 15\r\n\r\n"
+ body = b'{"status":"ok"}'
+ if drip == "body":
+ self.wfile.write(header)
+ self.wfile.flush()
+ try:
+ for byte in header if drip == "headers" else body:
+ self.wfile.write(bytes([byte]))
+ self.wfile.flush()
+ sent += 1
+ if stop.wait(0.6):
+ break
+ except (BrokenPipeError, ConnectionResetError):
+ pass
+
+ def log_message(self, format: str, *args: Any) -> None:
+ pass
+
+ server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+ thread = threading.Thread(target=server.serve_forever)
+ thread.start()
+ previous_handler = signal.getsignal(signal.SIGALRM)
+ start = time.monotonic()
+ try:
+ with pytest.raises(SystemExit) as exc:
+ namespace["http_get"](f"http://127.0.0.1:{server.server_port}/healthz")
+ elapsed = time.monotonic() - start
+ assert exc.value.code == 1
+ assert namespace["HTTP_DEADLINE"] <= elapsed < namespace["HTTP_DEADLINE"] + 1
+ assert sent >= 5
+ assert signal.getsignal(signal.SIGALRM) == previous_handler
+ assert signal.getitimer(signal.ITIMER_REAL) == (0.0, 0.0)
+ assert capsys.readouterr().out == "Verification failed.\n"
+ finally:
+ stop.set()
+ server.shutdown()
+ server.server_close()
+ thread.join()