diff --git a/CITATION.cff b/CITATION.cff index 64ed0c9..4a5d897 100644 --- a/CITATION.cff +++ b/CITATION.cff @@ -16,7 +16,7 @@ authors: affiliation: "Brookhaven National Laboratory" license: BSD-3-Clause repository-code: "https://github.com/NSLS2/PowderLine" -version: 0.1.1 +version: 0.1.2 identifiers: - type: doi value: "10.48550/arXiv.2608.17009" diff --git a/README.md b/README.md index f987289..ed83665 100644 --- a/README.md +++ b/README.md @@ -112,10 +112,18 @@ needed), which is a large speedup across many refinements. Manage it directly, or opt out per run with `--no-server`: ```bash -pixi run gsas-server status # start | status | stop | restart +pixi run gsas-server status # start | status | stop | restart | logs pixi run kicker --no-server # force an isolated one-shot subprocess ``` +The server is per-user: it listens on loopback only and accepts only requests +signed with a secret token kept in a directory only you can read; your client +in turn only accepts results signed with that token, so other users on the +same machine can neither use your server nor impersonate it. To switch it off +— e.g. on a shared workstation — set `POWDERLINE_NO_SERVER=1`: default +(`auto`) runs then go in-process, explicit server mode (`--use-server` / +`execution_mode='server'`) fails, and nothing is auto-started. + **Simulate a pattern from Materials Project** (requires an MP API key in `.powderline_config.yaml` — see `.powderline_config.yaml.example` — or in the `MP_API_KEY` environment variable): diff --git a/docs/DEVELOPMENT.md b/docs/DEVELOPMENT.md index 46f538d..d5bdf4c 100644 --- a/docs/DEVELOPMENT.md +++ b/docs/DEVELOPMENT.md @@ -470,6 +470,7 @@ class SizeBroadening(BaseModel): **src/powderline/gsas_server.py / gsas_client.py**: - Persistent FastAPI server keeping GSAS-II loaded in memory, and the HTTP client that talks to it (with in-process subprocess fallback) +- Per-user and authenticated by signed requests/responses (private state dir, loopback, ephemeral port, single instance via a kernel lock); `POWDERLINE_NO_SERVER=1` disables it - **Role**: Execution backends for `run()`'s `server`/`auto` modes **src/powderline/topas/ + src/powderline/easydiff/ + src/powderline/engine.py**: @@ -809,20 +810,47 @@ pixi run gsas-server status # Check if running pixi run gsas-server stop # Shutdown server ``` -**Configure server port (if default port 19471 is in use):** +**Server state and security (per user):** the server binds `127.0.0.1` on a +free ephemeral port and writes its lock, PID, log, and an endpoint file (port + +random secret token) — all mode 0600 — to a private per-user state directory: +`$XDG_RUNTIME_DIR/powderline` (if it is a real directory owned by you), else +`/powderline-` on POSIX (mode 0700; refused if it is a symlink, +owned by someone else, or group/other-accessible, or if the temp directory +would let others replace it), and `%LOCALAPPDATA%\powderline` on Windows. On +POSIX the files are opened relative to a re-verified, no-follow directory +handle. + +The token never travels over the wire: every request carries a fresh nonce, a +timestamp and an HMAC-SHA256 over method, path, nonce, timestamp and body, and +the server rejects anything unsigned, wrongly signed, stale, or replayed — +on every route, including 404/405/422 — before reading the body. Every +response is signed back over the nonce, status and body, and the client +discards unsigned responses. Other users on a shared host therefore can +neither submit jobs to your server nor impersonate it (e.g. by taking over the +port of a crashed server). `gsas-server start` takes a kernel-held lock +(`flock` / `msvcrt.locking` on `server.lock`) for the server's lifetime; the +OS releases it when the process exits, so concurrent auto-starts never spawn +duplicates and leftover PID files can never make a dead server look alive. + +**Pin the server port** (only if needed — the default ephemeral port avoids +collisions between users): ```bash -# Set custom port before starting server export POWDERLINE_SERVER_PORT=19472 pixi run gsas-server start +pixi run kicker input.json # client reads the port from your state directory +``` -# Client will automatically detect the port from the powderline_gsas_server.port -# file in the platform temp directory (e.g. /tmp on Linux, %TEMP% on Windows) -pixi run kicker input.json # No additional configuration needed +**Disable the server entirely** (shared workstations, locked-down deployments): +```bash +export POWDERLINE_NO_SERVER=1 # also: true / yes / on ``` +With it set, `execution_mode='auto'` (and the CLI default) runs in-process and +never starts a server, `execution_mode='server'` / `--use-server` fail with a +clear error, and `pixi run gsas-server start|restart` refuse. -**View server logs** (`powderline_gsas_server.log` in the platform temp directory): +**View server logs** (`server.log` in the state directory; `status` prints the path): ```bash -tail -f /tmp/powderline_gsas_server.log # Linux; on Windows: %TEMP%\powderline_gsas_server.log +pixi run gsas-server logs ``` **Server benefits:** diff --git a/docs/TROUBLESHOOTING.md b/docs/TROUBLESHOOTING.md index 18b5dca..c407cd8 100644 --- a/docs/TROUBLESHOOTING.md +++ b/docs/TROUBLESHOOTING.md @@ -654,11 +654,30 @@ Not necessarily! High correlation is common and often acceptable. pixi run kicker input.json --use-server ``` -4. **Check server logs if it's not working** — `powderline_gsas_server.log` lives in the platform temp directory (e.g. `/tmp` on Linux, `%TEMP%` on Windows): +4. **Check server logs if it's not working** — `server.log` lives in your + private per-user state directory (`pixi run gsas-server status` prints the + path): ```bash - tail -f /tmp/powderline_gsas_server.log # Linux + pixi run gsas-server logs ``` +5. **Check that the server isn't disabled** — if `POWDERLINE_NO_SERVER` is set + (e.g. by a facility launcher), default (`auto`) runs are in-process by + design, explicit server mode (`--use-server` / `execution_mode='server'`) + fails, and `gsas-server start` refuses. + +6. **Upgrading from v0.1.1 or earlier?** Older servers used fixed, shared + files in the temp directory and port 19471, and accepted unauthenticated + requests. The current client never uses them, but `gsas-server status|stop` + cannot see them either, so one may still be running (it may have been + auto-started without you noticing). Stop it once: + ```bash + pkill -u "$USER" -f gsas_server.py # Linux/macOS + ``` + On Windows, end the `python ... gsas_server.py start` process in Task + Manager (or `Get-CimInstance Win32_Process | Where-Object CommandLine -like + '*gsas_server.py*' | ForEach-Object { Stop-Process -Id $_.ProcessId }`). + **Performance comparison (post-fix):** - Server mode: 0.1-0.3s per simulation (10x+ faster) - Subprocess mode: 1-2s per simulation (GSAS-II loaded every time) @@ -676,7 +695,18 @@ If you see "Server communication failed" errors, check the server logs. ❌ Server mode failed: Server not available and fallback disabled ``` -**Cause:** Server process failed to start or port already in use. +**Cause:** Server process failed to start, the server is disabled +(`POWDERLINE_NO_SERVER`), or your per-user state directory is unusable. + +**How the server is found:** each user's server binds `127.0.0.1` on a free +(ephemeral) port and records its port, PID and a random secret token in a +private per-user state directory — `$XDG_RUNTIME_DIR/powderline` or +`/powderline-` on Linux/macOS, `%LOCALAPPDATA%\powderline` on +Windows. Clients read only their own user's directory, sign every request +with the token (the token itself is never sent), and only accept responses +signed with it — so another user's server, or another process that took over +the port of a crashed server, is never used. An endpoint whose server process +is gone is ignored. **Solution:** @@ -688,30 +718,29 @@ If you see "Server communication failed" errors, check the server logs. 2. **Stop any hung server processes:** ```bash pixi run gsas-server stop - # Or manually kill: - pkill -f gsas_server.py + # Or manually kill your own server: + pkill -u "$USER" -f gsas_server.py ``` -3. **Check if the port is in use:** +3. **"Refusing to use GSAS-II server state directory ..."** — the directory is + a symlink, owned by another user, or readable/writable by others (e.g. + pre-created in a shared `/tmp`). The server and auto-start are skipped: + default runs fall back to in-process, explicit server mode fails. Remove the + directory named in the message (if it is not yours, ask an administrator) + and it is recreated privately on the next start. **"Refusing to keep the + GSAS-II server state under ..."** means the parent (your `TMPDIR`) would let + other users replace that directory — world-writable without the sticky bit, + or owned by someone else; point `TMPDIR` (or `XDG_RUNTIME_DIR`) at a private + directory. + +4. **Pin a port only if you must** (e.g. a firewall rule): by default the + server uses a free ephemeral port, so users never collide. To force one: ```bash - # See if anything is listening on port 19471: - ss -tlnp | grep 19471 - ``` - -4. **Use a different port if 19471 is unavailable:** - ```bash - # Set custom port via environment variable export POWDERLINE_SERVER_PORT=19472 pixi run gsas-server start - - # Client will automatically detect the port - pixi run kicker input.json # No changes needed + pixi run kicker input.json # the client finds the port automatically ``` - The server writes its port to `powderline_gsas_server.port` in the platform temp - directory (e.g. `/tmp` on Linux, `%TEMP%` on Windows), which the client reads - automatically. No code changes needed when using a custom port. - 5. **Start server in foreground to see errors:** ```bash pixi shell diff --git a/docs/cross-platform-guide.md b/docs/cross-platform-guide.md index c7b4876..47679bb 100644 --- a/docs/cross-platform-guide.md +++ b/docs/cross-platform-guide.md @@ -549,3 +549,15 @@ If you prefer native tools, the platform equivalents are: - **Unix (Linux/macOS)**: `tail -f ` - **Windows PowerShell**: `Get-Content -Wait -Tail 50` + +`` is `server.log` in the per-user state directory +(`pixi run gsas-server status` prints it): `$XDG_RUNTIME_DIR/powderline` or +`/powderline-` on Linux/macOS (POSIX mode 0700, ownership +checked), `%LOCALAPPDATA%\powderline` on Windows. The server's port and token +live in the same directory, and the protocol is the same everywhere: loopback +TCP with requests and responses signed by a token only the owning user can +read. What protects that token differs: POSIX mode/ownership checks (enforced +by PowderLine) on Linux/macOS, the user-profile ACL of `%LOCALAPPDATA%` on +Windows (relied on, not checked — a profile whose ACL was loosened by an +administrator is not detected). The start lock uses `flock` on POSIX and +`msvcrt.locking` on Windows. diff --git a/docs/integration-guide.md b/docs/integration-guide.md index 98ecc7d..4d75626 100644 --- a/docs/integration-guide.md +++ b/docs/integration-guide.md @@ -193,6 +193,12 @@ The difference is startup overhead: not running and fall back to subprocess mode if the server fails to start. `powderline.run()` uses `GSASClient` internally. +The server is per-user: it is started as, and only accepts jobs from, the +calling user (requests and responses signed with a secret token kept in a +private state directory; loopback only), so it is safe on shared hosts. Integrations that must never run a background server +can set `POWDERLINE_NO_SERVER=1`: `auto` then runs in-process, `server` mode +returns an error result, and no server is ever started. + --- ## Example Scripts diff --git a/docs/known_issues.md b/docs/known_issues.md index 08e103e..61fb48c 100644 --- a/docs/known_issues.md +++ b/docs/known_issues.md @@ -326,6 +326,18 @@ discovery/lifecycle mismatch below remains open. HTTP shutdown endpoint, or have `stop` fall back to the PID reported by `/health`). +**Update (v0.1.2, KI-13).** The split discovery is resolved. Server state +moved to a per-user directory (`server.lock`, `server.pid`, `server.json` = +port + token; ephemeral port), and `status`/`stop` and clients now agree: +"running" means holding the kernel lock on `server.lock` (released by the OS +on exit, so a leftover PID file or a recycled PID no longer fools `status` or +`stop`), and clients use only the endpoint in `server.json`, verified by +signed responses. **Residual (why this stays open):** if the state directory +itself is deleted while a server runs (e.g. `$XDG_RUNTIME_DIR` removed at +logout, or a tempdir cleaner), that server keeps running unused — clients can +no longer find or authenticate to it — but `status`/`stop` cannot see it +either; it must be killed by hand (`pkill -u "$USER" -f gsas_server.py`). + **Revisit.** Next branch touching `gsas_server.py`/`gsas_client.py`. --- @@ -356,3 +368,61 @@ supported topology. Needs protocol versioning for already-running servers. **Revisit.** Next branch touching the server protocol, or when the HPC deployment work starts. + +--- + +## KI-13 — GSAS-II server was shared across users on multi-user hosts `implemented · critical` + +**What.** Through v0.1.1 the server kept machine-global state +(`/powderline_gsas_server.{port,pid,log}`), listened on a fixed port +(19471) and had no authentication. On a shared workstation: user B's client +found user A's server and ran B's job as A (files owned by A, possibly written +where B could not write); any local user could POST a job with an +`output_dir` inside another user's directories and have it written with that +user's identity; users collided on the port and the shared files; and there +was no global way to keep `powderline.run()` from auto-starting a server. + +**Evidence.** `gsas_server.py` (v0.1.1) `PORT_FILE`/`PID_FILE`/`LOG_FILE`, +`DEFAULT_PORT = 19471`, unauthenticated `/simulate`; `gsas_client.py` probed +`/health` on that port and submitted to whatever answered. + +**Decision.** Fixed in v0.1.2: + +- *Per-user state.* Lock, PID, log and endpoint (port + token), all mode + 0600, live in `$XDG_RUNTIME_DIR/powderline` (only if that is a real + directory owned by the user) or `/powderline-` (POSIX: 0700; + refused if a symlink, foreign-owned or group/other-accessible, or if the + temp directory would let other users replace it; files opened via a + re-verified `O_NOFOLLOW` directory handle), or `%LOCALAPPDATA%\powderline` + (Windows, protected by the profile ACL). +- *Authentication both ways, token never on the wire.* The server binds + `127.0.0.1` on an ephemeral port. Every request must carry a fresh nonce, a + timestamp and an HMAC-SHA256 (keyed by a per-start `secrets.token_urlsafe(32)` + token) over method, path, nonce, timestamp and body; unsigned, wrongly + signed, stale or replayed requests get 401 on every route before the body is + read. Every response is signed over the nonce, status and body, and the + client discards unsigned ones — so a process that takes over the port of a + crashed server learns nothing reusable and cannot fake a result. Clients + also ignore an endpoint whose server process is gone. +- *Single instance.* A kernel-held lock (`flock` / `msvcrt.locking`) on + `server.lock` for the server's lifetime replaces the PID-file lock, which a + stale PID file let several concurrent starters win. +- *Off switch.* `POWDERLINE_NO_SERVER=1|true|yes|on` disables the server: + `auto` runs in-process without starting or contacting one, `server` mode / + `--use-server` error, `gsas-server start|restart` refuse. +- *Upgrade.* A v0.1.1 server still running after the upgrade is never used, + but is invisible to `gsas-server status|stop`; stop it by hand (see + TROUBLESHOOTING). + +**Evidence.** `src/powderline/gsas_server.py:119-315` (state directory and +file access), `:318-450` (signing and `SignedRequestMiddleware`), `:476-562` +(start lock), `:847` (`is_server_running`); `src/powderline/gsas_client.py:59-136` +(endpoint, signed requests, identity check). Tests: +`tests/test_gsas_server_unit.py` (state-dir checks, lock incl. a concurrent +start race, signing on every route), `tests/test_gsas_server_http.py` (live +server), `tests/test_gsas_client_visibility.py` (impostor listeners, forged +results, `POWDERLINE_NO_SERVER`), `tests/test_api.py`. + +**Revisit.** Closed; kept for history. A Unix-domain-socket / named-pipe +transport would give OS-level peer authentication but needs separate POSIX and +Windows code paths; revisit if the HTTP transport is replaced (see KI-12). diff --git a/pixi.toml b/pixi.toml index 27d717e..8f23b6e 100644 --- a/pixi.toml +++ b/pixi.toml @@ -3,7 +3,7 @@ authors = ["Dan Olds ", "Adam Corrao <95718914+AdamCorrao@user channels = ["conda-forge"] name = "PowderLine" platforms = ["linux-64", "win-64", "osx-arm64"] -version = "0.1.1" +version = "0.1.2" preview = ["pixi-build"] [tasks] diff --git a/src/powderline/__init__.py b/src/powderline/__init__.py index efd69dd..53ab582 100644 --- a/src/powderline/__init__.py +++ b/src/powderline/__init__.py @@ -1,6 +1,6 @@ """PowderLine: Automated powder diffraction analysis using GSAS-II.""" -__version__ = "0.1.1" +__version__ = "0.1.2" # The GSAS-II-backed public API is imported eagerly when GSAS-II is available, # preserving the original behaviour exactly (including `powderline.kicker` being diff --git a/src/powderline/_code_hash.json b/src/powderline/_code_hash.json index 5e3820a..d336688 100644 --- a/src/powderline/_code_hash.json +++ b/src/powderline/_code_hash.json @@ -1,4 +1,4 @@ { "schema_version": "0.26.0", - "kicker_hash": "67826378ba97ed89c55cfb1df7e32ee4" + "kicker_hash": "c5e25670cb609186fedd6ac2e2401b9e" } diff --git a/src/powderline/gsas_client.py b/src/powderline/gsas_client.py index a28401b..ff298fa 100644 --- a/src/powderline/gsas_client.py +++ b/src/powderline/gsas_client.py @@ -5,19 +5,43 @@ running, or when a "successful" server run produced output files that are not visible to this process (server with a divergent filesystem view, e.g. on another node or inside a sandbox with a private /tmp). + +Only the calling user's own server is ever used: its port and secret token +are read from the user's private state directory. The token itself is never +sent; every request is signed with it and every response must carry a valid +signature, so a listener that does not hold the token (e.g. another user's +process on the port of a crashed server) is never trusted — it learns nothing +it could replay and cannot fake a result. ``POWDERLINE_NO_SERVER`` disables +server use entirely (see :func:`powderline.gsas_server.server_disabled`). """ +import json +import secrets import subprocess import sys import os from pathlib import Path -from typing import Dict, Any -import tempfile +from typing import Dict, Any, Optional + +from powderline import gsas_server +from powderline.gsas_server import ( + HOST, + NO_SERVER_ENV, + NO_SERVER_MESSAGE, + NONCE_HEADER, + RESPONSE_SIGNATURE_HEADER, + SIGNATURE_HEADER, + TIMESTAMP_HEADER, + ServerStateError, + read_endpoint, + server_disabled, + sign_request, + sign_response, +) -HOST = "127.0.0.1" -DEFAULT_PORT = 19471 -PORT = int(os.environ.get('POWDERLINE_SERVER_PORT', DEFAULT_PORT)) -PORT_FILE = Path(tempfile.gettempdir()) / 'powderline_gsas_server.port' + +class ServerIdentityError(RuntimeError): + """A response was not signed with this user's server token.""" class GSASClient: @@ -30,26 +54,86 @@ def __init__(self, fallback_to_subprocess: bool = True): fallback_to_subprocess: If True, use subprocess when server unavailable """ self.fallback_to_subprocess = fallback_to_subprocess + self._state_warning_shown = False - def _get_port(self) -> int: - """Read the server port from the port file.""" - if PORT_FILE.exists(): - try: - return int(PORT_FILE.read_text().strip()) - except (ValueError, OSError): - pass - return PORT + def _endpoint(self) -> Optional[Dict[str, Any]]: + """This user's live server endpoint (``pid``/``port``/``token``), or None. + + An endpoint whose recorded server process is gone is stale (the server + crashed without cleaning up) and is ignored. + """ + try: + endpoint = read_endpoint() + except ServerStateError as e: + if not self._state_warning_shown: + print(f" WARNING: {e}") + self._state_warning_shown = True + return None + if endpoint is not None and not gsas_server._pid_alive(endpoint['pid']): + return None + return endpoint + + @staticmethod + def _signed_request(endpoint: Dict[str, Any], method: str, path: str, + payload: Optional[dict] = None, timeout: float = 2.0): + """Send a request signed with the endpoint's token; verify the response. + + The token never leaves this process: the request carries a fresh + nonce, a timestamp and an HMAC, and the response must be signed over + that nonce, its status and its body. + + Raises: + ServerIdentityError: If the response is not signed by the token holder. + httpx.HTTPError: On connection/transport errors. + """ + import httpx + import time + + body = b'' if payload is None else json.dumps(payload).encode() + nonce = secrets.token_hex(16) + timestamp = str(int(time.time())) + token = endpoint['token'] + headers = { + NONCE_HEADER: nonce, + TIMESTAMP_HEADER: timestamp, + SIGNATURE_HEADER: sign_request(token, method, path, b'', nonce, timestamp, body), + } + url = f"http://{HOST}:{endpoint['port']}{path}" + # trust_env=False: loopback only, never route via a proxy + if method == 'GET': + resp = httpx.get(url, headers=headers, timeout=timeout, trust_env=False) + else: + headers['Content-Type'] = 'application/json' + resp = httpx.post(url, content=body, headers=headers, timeout=timeout, + trust_env=False) + got = resp.headers.get(RESPONSE_SIGNATURE_HEADER, '') + expected = sign_response(token, nonce, resp.status_code, resp.content) + if not secrets.compare_digest(got.encode('utf-8', 'replace'), expected.encode()): + raise ServerIdentityError( + f"The listener on port {endpoint['port']} did not prove it is your " + f"GSAS-II server (HTTP {resp.status_code}); not using it.") + return resp def is_server_available(self) -> bool: - """Check if server is running and accepting connections.""" + """Check that THIS user's server is up and holds this user's token. + + ``/health`` is sent as a signed request and its response must be + signed with the token, so a listener that does not know the token + (e.g. another user's process on a recycled port) is never treated as + available. Always False when ``POWDERLINE_NO_SERVER`` is set. + """ import httpx - port = self._get_port() + if server_disabled(): + return False + endpoint = self._endpoint() + if endpoint is None: + return False try: - resp = httpx.get(f"http://{HOST}:{port}/health", timeout=2.0) - return resp.status_code == 200 - except (httpx.ConnectError, httpx.TimeoutException, OSError): + resp = self._signed_request(endpoint, 'GET', '/health', timeout=2.0) + except (httpx.HTTPError, OSError, ServerIdentityError): return False + return resp.status_code == 200 @staticmethod def _stat_fit_profile(output_dir: Path) -> tuple | None: @@ -153,10 +237,22 @@ def submit_simulation(self, recipe, output_dir: Path, execution with a warning — or, when fallback_to_subprocess is False, returns a structured error explaining the filesystem-view mismatch. + + With ``POWDERLINE_NO_SERVER`` set, the server is never probed or + started: the run goes in-process, or — when fallback is disabled + (``execution_mode='server'``) — returns a structured error. """ recipe_dict = self._normalize_recipe(recipe) - if not use_server: + if use_server and server_disabled(): + if not self.fallback_to_subprocess: + return { + 'success': False, + 'error': NO_SERVER_MESSAGE, + 'method': 'none' + } + print(f" Using in-process mode ({NO_SERVER_ENV} is set)") + elif not use_server: if not verbose: print(" Using in-process mode (use_server=False)") # Try server first @@ -269,6 +365,15 @@ def _start_server_background(self, verbose: bool = False) -> bool: import os import time + if server_disabled(): + return False + try: + gsas_server.state_dir() # a server could not start with an unsafe state dir + except ServerStateError as e: + if verbose: + print(f" Could not auto-start server: {e}") + return False + # Check if already running if self.is_server_available(): return True @@ -330,7 +435,6 @@ def _submit_to_server(self, recipe_dict: dict, output_dir: Path) -> Dict[str, An import httpx import time - port = self._get_port() payload = { 'recipe_data': recipe_dict, 'output_dir': str(output_dir.absolute()), @@ -342,12 +446,13 @@ def _submit_to_server(self, recipe_dict: dict, output_dir: Path) -> Dict[str, An last_error = None for attempt in range(max_retries): + # Re-read each attempt: a restarted server has a new port and token. + endpoint = self._endpoint() try: - resp = httpx.post( - f"http://{HOST}:{port}/simulate", - json=payload, - timeout=300.0, - ) + if endpoint is None: + raise httpx.ConnectError("No GSAS-II server endpoint for this user") + resp = self._signed_request(endpoint, 'POST', '/simulate', + payload, timeout=300.0) resp.raise_for_status() return resp.json() except httpx.ConnectError as e: diff --git a/src/powderline/gsas_server.py b/src/powderline/gsas_server.py index 72cd717..14ad2f9 100644 --- a/src/powderline/gsas_server.py +++ b/src/powderline/gsas_server.py @@ -4,14 +4,24 @@ Listens on an HTTP port via FastAPI and processes simulation requests sequentially. Security notes: - - The server binds exclusively to ``127.0.0.1`` (loopback). It is not - accessible from remote hosts and requires no authentication by design. - - No rate limiting is applied. For a shared or multi-user environment, - add rate limiting via a reverse proxy or a middleware library such as - ``slowapi`` before exposing this server to untrusted callers. - - ``output_dir`` paths are sent by the client and used as-is; path - traversal is only exploitable by a local user who already has file-system - access equivalent to the server process. + - The server binds exclusively to ``127.0.0.1`` (loopback) on an ephemeral + port (or ``POWDERLINE_SERVER_PORT`` if set). It is not accessible from + remote hosts. + - Every request must be signed with a per-start secret token (an HMAC over + method, path, nonce, timestamp and body), and every response is signed + back (see :class:`SignedRequestMiddleware`). The token is stored, with + the port and PID, in a per-user state directory (see :func:`state_dir`) + that only the owning user can read, and it never travels over the wire. + Other local users therefore cannot submit jobs (which would write files + as the server's owner) or even probe the server, and a client never + trusts a listener that cannot sign with its token — e.g. another user's + process that took over the port of a crashed server. + - A kernel-held lock on ``server.lock`` (released by the OS when the server + exits) ensures a single server per user, even with concurrent starts. + - Set ``POWDERLINE_NO_SERVER=1`` (``true``/``yes``/``on``) to disable the + server: ``auto`` mode runs in-process, ``server`` mode returns an error, + and ``gsas-server start``/``restart`` refuse. + - No rate limiting is applied. - The server resolves ``output_dir`` in ITS OWN filesystem view. A client on another node — or a server started inside a sandbox/container with a private /tmp — will never see the output files even though the run @@ -30,11 +40,11 @@ # Stop server pixi run gsas-server stop - # Check status + # Check status (also prints the log file location) pixi run gsas-server status # View logs - tail -f /tmp/powderline_gsas_server.log + pixi run gsas-server logs # Restart server pixi run gsas-server restart @@ -44,9 +54,14 @@ import sys import os import signal +import socket +import stat import time import subprocess import logging +import hashlib +import hmac +import secrets from pathlib import Path from typing import Dict, Any, Optional, List import tempfile @@ -57,11 +72,523 @@ # Server configuration HOST = "127.0.0.1" -DEFAULT_PORT = 19471 -PORT = int(os.environ.get('POWDERLINE_SERVER_PORT', DEFAULT_PORT)) -PORT_FILE = Path(tempfile.gettempdir()) / 'powderline_gsas_server.port' -PID_FILE = Path(tempfile.gettempdir()) / 'powderline_gsas_server.pid' -LOG_FILE = Path(tempfile.gettempdir()) / 'powderline_gsas_server.log' +PORT_ENV = 'POWDERLINE_SERVER_PORT' +NO_SERVER_ENV = 'POWDERLINE_NO_SERVER' +NO_SERVER_VALUES = ('1', 'true', 'yes', 'on') +NO_SERVER_MESSAGE = ( + f"The GSAS-II server is disabled ({NO_SERVER_ENV} is set). " + f"Run in-process instead (execution_mode='subprocess' / --no-server), " + f"or unset {NO_SERVER_ENV}." +) + +# Request/response signing (see sign_request / sign_response) +NONCE_HEADER = 'X-PowderLine-Nonce' +TIMESTAMP_HEADER = 'X-PowderLine-Timestamp' +SIGNATURE_HEADER = 'X-PowderLine-Signature' +RESPONSE_SIGNATURE_HEADER = 'X-PowderLine-Response-Signature' +MAX_CLOCK_SKEW = 120 # seconds a signed request stays valid +MAX_BODY_BYTES = 64 * 1024 * 1024 # larger requests are refused unread + +# File names inside the per-user state directory +PID_NAME = 'server.pid' +ENDPOINT_NAME = 'server.json' +LOG_NAME = 'server.log' +LOCK_NAME = 'server.lock' + + +class ServerStateError(RuntimeError): + """The per-user server state directory is unsafe or unusable.""" + + +def server_disabled() -> bool: + """True if ``POWDERLINE_NO_SERVER`` is set to a truthy value (1/true/yes/on).""" + return os.environ.get(NO_SERVER_ENV, '').strip().lower() in NO_SERVER_VALUES + + +def _private_parent(path: str) -> bool: + """True if ``path`` is a real directory (not a symlink) owned by us that + others cannot write to.""" + try: + st = os.lstat(path) + except OSError: + return False + return (stat.S_ISDIR(st.st_mode) and st.st_uid == os.getuid() + and not st.st_mode & 0o022) + + +def _check_tempdir_parent(path: Path) -> None: + """Refuse a parent in which other users could rename or replace our entry. + + Safe parents are owned by root or by us and are either not writable by + group/other or sticky (like ``/tmp``). Symlinks are followed here on + purpose (macOS ``/tmp`` -> ``/private/tmp``). + """ + try: + st = os.stat(path) + except OSError as e: + raise ServerStateError( + f"Cannot use {path} for the GSAS-II server state directory: {e}") from e + if not stat.S_ISDIR(st.st_mode): + problem = "is not a directory" + elif st.st_uid not in (0, os.getuid()): + problem = f"is owned by uid {st.st_uid}" + elif st.st_mode & 0o022 and not st.st_mode & stat.S_ISVTX: + problem = "is writable by other users without the sticky bit" + else: + return + raise ServerStateError( + f"Refusing to keep the GSAS-II server state under {path}: it {problem}, " + f"so other users could replace the state directory. Point TMPDIR (or " + f"XDG_RUNTIME_DIR) at a private directory.") + + +def _state_dir_path() -> Path: + """Location of the per-user state directory (not created or checked). + + POSIX: ``$XDG_RUNTIME_DIR/powderline`` when that directory is ours, else + ``/powderline-`` — host-local, never the (possibly + NFS-shared) home directory, because the server is per-host. + Windows: ``%LOCALAPPDATA%\\powderline`` (protected by the profile ACL). + """ + if os.name == 'nt': + base = os.environ.get('LOCALAPPDATA') or str(Path.home() / 'AppData' / 'Local') + return Path(base) / 'powderline' + runtime_dir = os.environ.get('XDG_RUNTIME_DIR') + if runtime_dir and _private_parent(runtime_dir): + return Path(runtime_dir) / 'powderline' + return Path(tempfile.gettempdir()) / f'powderline-{os.getuid()}' + + +def _unsafe_state_dir(st: os.stat_result) -> Optional[str]: + """Why a stat of the state directory makes it unsafe, or None if it is fine.""" + if stat.S_ISLNK(st.st_mode) or not stat.S_ISDIR(st.st_mode): + return "is not a directory (or is a symlink)" + if st.st_uid != os.getuid(): + return f"is owned by uid {st.st_uid}, not by you (uid {os.getuid()})" + if st.st_mode & 0o077: + return f"is accessible to other users (mode {stat.S_IMODE(st.st_mode):o})" + return None + + +def _refuse(path: Path, problem: str) -> ServerStateError: + return ServerStateError( + f"Refusing to use GSAS-II server state directory {path}: it {problem}. " + f"Remove it so it can be recreated privately." + ) + + +def state_dir() -> Path: + """Return the per-user server state directory, creating it if needed. + + Holds the lock, PID, endpoint (port + token) and log files. On POSIX it is + created with mode 0700 and refused if it is a symlink, not owned by the + current user, or accessible to group/other, or if its parent would let + another user replace it — a directory pre-created by another user in a + shared /tmp must never be trusted. Files inside are then opened relative + to a re-verified directory handle (see :func:`_open_state_file`). + + Raises: + ServerStateError: If the directory exists but is unsafe to use. + """ + path = _state_dir_path() + if os.name == 'nt': + try: + path.mkdir(parents=True, exist_ok=True) + except OSError as e: + raise ServerStateError( + f"Cannot create GSAS-II server state directory {path}: {e}") from e + return path + _check_tempdir_parent(path.parent) + try: + try: + os.mkdir(path, 0o700) + os.chmod(path, 0o700) # umask may have stripped owner bits + except FileExistsError: + pass + st = os.lstat(path) + except OSError as e: + raise ServerStateError( + f"Cannot create GSAS-II server state directory {path}: {e}") from e + problem = _unsafe_state_dir(st) + if problem: + raise _refuse(path, problem) + return path + + +def _open_state_dir() -> int: + """Open the verified state directory without following symlinks (POSIX). + + The returned descriptor is re-checked with ``fstat``, so the directory the + caller works in is exactly the one verified — it cannot be swapped for a + symlink between the check and the use. + """ + path = state_dir() + try: + dfd = os.open(path, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) + except OSError as e: + raise _refuse(path, f"cannot be opened safely ({e})") from e + problem = _unsafe_state_dir(os.fstat(dfd)) + if problem: + os.close(dfd) + raise _refuse(path, problem) + return dfd + + +def _open_state_file(name: str, flags: int, mode: int = 0o600) -> int: + """``os.open`` a file in the state directory (never through a symlink).""" + if os.name == 'nt': + return os.open(state_dir() / name, flags | getattr(os, 'O_BINARY', 0), mode) + dfd = _open_state_dir() + try: + return os.open(name, flags | os.O_NOFOLLOW, mode, dir_fd=dfd) + finally: + os.close(dfd) + + +def _read_state_file(name: str) -> Optional[str]: + """Contents of a state file, or None if it does not exist.""" + try: + fd = _open_state_file(name, os.O_RDONLY) + except FileNotFoundError: + return None + with os.fdopen(fd, 'rb') as f: + return f.read().decode('utf-8', 'replace') + + +def _unlink_state_file(name: str) -> None: + """Remove a state file if present.""" + try: + if os.name == 'nt': + (state_dir() / name).unlink() + return + dfd = _open_state_dir() + try: + os.unlink(name, dir_fd=dfd) + finally: + os.close(dfd) + except FileNotFoundError: + pass + + +def _write_private(name: str, text: str) -> None: + """Atomically write ``text`` to state file ``name``, readable by the owner only.""" + tmp = name + '.tmp' + _unlink_state_file(tmp) + fd = _open_state_file(tmp, os.O_WRONLY | os.O_CREAT | os.O_EXCL) + with os.fdopen(fd, 'w') as f: + f.write(text) + if os.name == 'nt': + d = state_dir() + os.replace(d / tmp, d / name) + return + dfd = _open_state_dir() + try: + os.replace(tmp, name, src_dir_fd=dfd, dst_dir_fd=dfd) + finally: + os.close(dfd) + + +def log_file() -> Path: + """Path of this user's server log file.""" + return state_dir() / LOG_NAME + + +def read_endpoint() -> Optional[Dict[str, Any]]: + """Return this user's server endpoint ``{'pid', 'port', 'token'}``, or None. + + Raises: + ServerStateError: If the state directory is unsafe to use. + """ + try: + data = json.loads(_read_state_file(ENDPOINT_NAME) or '') + return {'pid': int(data['pid']), 'port': int(data['port']), + 'token': str(data['token'])} + except (OSError, ValueError, KeyError, TypeError): + return None + + +def read_pid() -> Optional[int]: + """PID recorded by the server holding the start lock, or None.""" + try: + return int(_read_state_file(PID_NAME) or '') + except (OSError, ValueError): + return None + + +# --- Request/response signing --- +# +# The token never travels over the wire. Each request carries a fresh nonce, +# a timestamp and an HMAC (keyed by the token) over the method, path, nonce, +# timestamp and body; the server answers with an HMAC over the nonce, status +# and response body. A listener that does not hold the token — e.g. another +# user's process that grabbed the port of a crashed server — can neither +# produce a valid response nor learn anything that would let it. + +def _mac(token: str, *parts) -> str: + msg = b'\n'.join(p if isinstance(p, bytes) else str(p).encode() for p in parts) + return hmac.new(token.encode(), msg, hashlib.sha256).hexdigest() + + +def sign_request(token: str, method: str, path: str, query: bytes, + nonce: str, timestamp: str, body: bytes) -> str: + """HMAC authenticating one request (method, target, nonce, time and body).""" + return _mac(token, b'powderline-request-v1', method.upper(), path, query, + nonce, timestamp, hashlib.sha256(body).hexdigest()) + + +def sign_response(token: str, nonce: str, status_code: int, body: bytes) -> str: + """HMAC proving a response came from the token holder, bound to the request nonce.""" + return _mac(token, b'powderline-response-v1', nonce, status_code, + hashlib.sha256(body).hexdigest()) + + +class SignedRequestMiddleware: + """ASGI wrapper: only correctly signed requests reach the app; every + response it produces is signed. + + It wraps the whole application, so 404/405/422 and error responses are + covered too, and unsigned requests are refused before their body is read. + """ + + def __init__(self, app, token: str): + self.app = app + self.token = token + self._seen_nonces: Dict[str, float] = {} + + async def __call__(self, scope, receive, send): + if scope['type'] == 'lifespan': + return await self.app(scope, receive, send) + if scope['type'] != 'http': + if scope['type'] == 'websocket': + await send({'type': 'websocket.close', 'code': 1008}) + return + + headers = {k.decode('latin-1').lower(): v.decode('latin-1') + for k, v in scope.get('headers', [])} + nonce = headers.get(NONCE_HEADER.lower(), '') + timestamp = headers.get(TIMESTAMP_HEADER.lower(), '') + signature = headers.get(SIGNATURE_HEADER.lower(), '') + now = time.time() + try: + fresh = abs(now - int(timestamp)) <= MAX_CLOCK_SKEW + except ValueError: + fresh = False + if not (fresh and 16 <= len(nonce) <= 128 + and all(c in '0123456789abcdef' for c in nonce) + and signature and nonce not in self._seen_nonces): + return await self._reject(send, 401) + + parts, size, more = [], 0, True + while more: + message = await receive() + if message['type'] == 'http.disconnect': + return + parts.append(message.get('body', b'')) + size += len(parts[-1]) + more = message.get('more_body', False) + if size > MAX_BODY_BYTES: + return await self._reject(send, 413) + body = b''.join(parts) + + expected = sign_request(self.token, scope['method'], scope['path'], + scope.get('query_string', b''), nonce, timestamp, body) + if not hmac.compare_digest(signature.encode('latin-1'), expected.encode()): + return await self._reject(send, 401) + # Another copy of this request may have claimed the nonce while this one + # awaited its body; re-check and record it with no await in between. + if nonce in self._seen_nonces: + return await self._reject(send, 401) + self._remember(nonce, now) + + body_sent = False + + async def replay_receive(): + nonlocal body_sent + if not body_sent: + body_sent = True + return {'type': 'http.request', 'body': body, 'more_body': False} + return await receive() + + start = None + chunks = [] + + async def signing_send(message): + nonlocal start + if message['type'] == 'http.response.start': + start = message + return + if message['type'] != 'http.response.body': + return await send(message) + chunks.append(message.get('body', b'')) + if message.get('more_body', False): + return + data = b''.join(chunks) + response_headers = [ + (k, v) for k, v in start.get('headers', []) + if k.lower() not in (b'content-length', + RESPONSE_SIGNATURE_HEADER.lower().encode()) + ] + response_headers += [ + (b'content-length', str(len(data)).encode()), + (RESPONSE_SIGNATURE_HEADER.lower().encode(), + sign_response(self.token, nonce, start['status'], data).encode()), + ] + await send({**start, 'headers': response_headers}) + await send({'type': 'http.response.body', 'body': data, 'more_body': False}) + + await self.app(scope, replay_receive, signing_send) + + def _remember(self, nonce: str, now: float) -> None: + """Record a used nonce (replay protection), forgetting expired ones.""" + if len(self._seen_nonces) > 1024: + cutoff = now - 2 * MAX_CLOCK_SKEW + self._seen_nonces = {n: t for n, t in self._seen_nonces.items() if t > cutoff} + self._seen_nonces[nonce] = now + + @staticmethod + async def _reject(send, status: int) -> None: + detail = b'{"detail":"Unauthorized"}' if status == 401 else b'{"detail":"Request too large"}' + await send({'type': 'http.response.start', 'status': status, 'headers': [ + (b'content-type', b'application/json'), + (b'content-length', str(len(detail)).encode()), + ]}) + await send({'type': 'http.response.body', 'body': detail}) + + +def _requested_port() -> int: + """Port from ``POWDERLINE_SERVER_PORT``, else 0 (an ephemeral port).""" + value = os.environ.get(PORT_ENV, '').strip() + return int(value) if value else 0 + + +def _bind_listener(port: int) -> socket.socket: + """Bind and listen on ``HOST:port`` (0 = ephemeral).""" + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + if hasattr(socket, 'SO_EXCLUSIVEADDRUSE'): + # Windows: forbid other sockets from binding the same port. + sock.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1) + elif port: + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + try: + sock.bind((HOST, port)) + sock.listen(128) + except OSError: + sock.close() + raise + return sock + + +# --- Start lock --- +# +# A kernel-held lock on server.lock, kept for the server's whole lifetime. The +# OS drops it when the process exits — however it exits — so there is no stale +# lock to reclaim, and holding it is the definition of "a server is running". +# The lock file itself is never deleted (deleting lock files reintroduces races). + +_lock_fd: Optional[int] = None # held by this process while it is the server + + +def _try_lock(fd: int) -> bool: + try: + if os.name == 'nt': + import msvcrt + os.lseek(fd, 0, os.SEEK_SET) + msvcrt.locking(fd, msvcrt.LK_NBLCK, 1) + else: + import fcntl + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + return True + except OSError: + return False + + +def _unlock(fd: int) -> None: + if os.name == 'nt': + import msvcrt + os.lseek(fd, 0, os.SEEK_SET) + msvcrt.locking(fd, msvcrt.LK_UNLCK, 1) + else: + import fcntl + fcntl.flock(fd, fcntl.LOCK_UN) + + +def _claim_server_lock(attempts: int = 20, delay: float = 0.05) -> bool: + """Take the start lock for the life of this process; False if a server holds it. + + A few retries absorb the instant during which ``is_server_running`` probes + the lock. Once held, any PID/endpoint files left over are stale (their + server is gone), so they are replaced. + """ + global _lock_fd + if _lock_fd is not None: + return True + fd = _open_state_file(LOCK_NAME, os.O_RDWR | os.O_CREAT) + for attempt in range(attempts): + if _try_lock(fd): + _lock_fd = fd + _unlink_state_file(ENDPOINT_NAME) + _write_private(PID_NAME, str(os.getpid())) + return True + if attempt < attempts - 1: + time.sleep(delay) + os.close(fd) + return False + + +def _lock_held() -> bool: + """True if some process (this one included) holds the start lock.""" + if _lock_fd is not None: + return True + fd = _open_state_file(LOCK_NAME, os.O_RDWR | os.O_CREAT) + try: + if _try_lock(fd): + _unlock(fd) + return False + return True + finally: + os.close(fd) + + +def _clear_stale_state_files() -> bool: + """Remove PID/endpoint files left by a server that is gone. + + Done while briefly holding the start lock, which proves no server is + running or starting, so a live server's files are never touched. Needed + where a server cannot clean up after itself — e.g. after ``stop`` on + Windows, which terminates the process. Returns False if the lock is held. + """ + if _lock_fd is not None: + return False + fd = _open_state_file(LOCK_NAME, os.O_RDWR | os.O_CREAT) + try: + if not _try_lock(fd): + return False + try: + _unlink_state_file(ENDPOINT_NAME) + _unlink_state_file(PID_NAME) + finally: + _unlock(fd) + return True + finally: + os.close(fd) + + +def _release_state_files() -> None: + """Remove the PID and endpoint files and drop the lock — only if we hold it.""" + global _lock_fd + if _lock_fd is None: + return + try: + _unlink_state_file(ENDPOINT_NAME) + _unlink_state_file(PID_NAME) + except (OSError, ServerStateError): + pass + try: + _unlock(_lock_fd) + except OSError: + pass + os.close(_lock_fd) + _lock_fd = None # --- Pydantic request/response models --- @@ -91,13 +618,15 @@ class SimulationResponse(BaseModel): # --- Logging --- -def setup_logging(log_file: Path) -> logging.Logger: - """Set up logging to both console and file.""" +def setup_logging() -> logging.Logger: + """Set up logging to both console and this user's server log file.""" logger = logging.getLogger('powderline.server') logger.setLevel(logging.DEBUG) - # File handler (always log everything) - file_handler = logging.FileHandler(log_file, mode='a') + # File handler (always log everything); owner-only, opened in the + # verified state directory + log_fd = _open_state_file(LOG_NAME, os.O_WRONLY | os.O_CREAT | os.O_APPEND) + file_handler = logging.StreamHandler(os.fdopen(log_fd, 'a', encoding='utf-8')) file_handler.setLevel(logging.DEBUG) # Console handler (info level for cleaner output) @@ -132,9 +661,11 @@ def __init__(self, logger: logging.Logger = None): self.running = False self.request_count = 0 self.start_time = None + # Per-start credential; clients read it from the owner-only state dir. + self.token = secrets.token_urlsafe(32) # Set up logging - self.logger = logger or setup_logging(LOG_FILE) + self.logger = logger or setup_logging() # Import GSAS-II (this is the slow part we want to do once) self.logger.info("Loading GSAS-II libraries...") @@ -156,10 +687,18 @@ def __init__(self, logger: logging.Logger = None): sys.exit(1) def create_app(self): - """Build and return the FastAPI application.""" + """Build and return the ASGI application. + + The FastAPI app is wrapped in :class:`SignedRequestMiddleware`: every + route — including ``/health`` — requires a request signed with + ``self.token`` (anything else gets 401), and every response is signed + so the client can verify it reached the token holder. + """ from fastapi import FastAPI - app = FastAPI(title="PowderLine GSAS-II Server") + # No interactive docs/schema routes: nothing is served unauthenticated. + app = FastAPI(title="PowderLine GSAS-II Server", + docs_url=None, redoc_url=None, openapi_url=None) server = self # capture for closures @app.post("/simulate", response_model=SimulationResponse) @@ -194,7 +733,7 @@ def health(): "request_count": server.request_count, } - return app + return SignedRequestMiddleware(app, self.token) def start(self): """Start the server with uvicorn.""" @@ -205,9 +744,12 @@ def start(self): self.logger.info("GSAS-II Server Starting") self.logger.info("=" * 60) - # Write PID and port files - PID_FILE.write_text(str(os.getpid())) - PORT_FILE.write_text(str(PORT)) + # Bind first so the endpoint file records the real (ephemeral) port + # and connections queue from the moment clients can discover it. + sock = _bind_listener(_requested_port()) + port = sock.getsockname()[1] + _write_private(ENDPOINT_NAME, json.dumps( + {'pid': os.getpid(), 'port': port, 'token': self.token})) # Register cleanup atexit.register(self.cleanup) @@ -216,15 +758,15 @@ def start(self): self.running = True self.logger.info(f"GSAS-II server started") - self.logger.info(f" URL: http://{HOST}:{PORT}") - self.logger.info(f" Log file: {LOG_FILE}") + self.logger.info(f" URL: http://{HOST}:{port}") + self.logger.info(f" Log file: {log_file()}") self.logger.info(f" PID: {os.getpid()}") self.logger.info(f" Started: {self.start_time.strftime('%Y-%m-%d %H:%M:%S')}") self.logger.info(f" Listening for simulation requests...") self.logger.info("-" * 60) app = self.create_app() - uvicorn.run(app, host=HOST, port=PORT, log_level="warning") + uvicorn.Server(uvicorn.Config(app, log_level="warning")).run(sockets=[sock]) def _run_simulation(self, request: Dict[str, Any]) -> Dict[str, Any]: """Run a GSAS-II simulation using the loaded kicker module. @@ -287,11 +829,8 @@ def _signal_handler(self, signum, frame): sys.exit(0) def cleanup(self): - """Clean up port file and PID file.""" - if PORT_FILE.exists(): - PORT_FILE.unlink() - if PID_FILE.exists(): - PID_FILE.unlink() + """Clean up the endpoint and PID files.""" + _release_state_files() if self.start_time: uptime = datetime.now() - self.start_time @@ -335,21 +874,12 @@ def _pid_alive(pid: int) -> bool: def is_server_running() -> bool: - """Check if server is currently running.""" - if not PID_FILE.exists(): - return False + """True if this user's server is running (or starting): it holds the start lock. - try: - pid = int(PID_FILE.read_text()) - except ValueError: - PID_FILE.unlink() - return False - - if _pid_alive(pid): - return True - # PID file exists but the process does not - PID_FILE.unlink() - return False + The lock is released by the OS when the server exits, however it exits, so + leftover PID files and recycled PIDs cannot make a dead server look alive. + """ + return _lock_held() def stop_server() -> bool: @@ -358,7 +888,10 @@ def stop_server() -> bool: print("Server is not running") return False - pid = int(PID_FILE.read_text()) + pid = read_pid() + if pid is None: + print("Server is starting up (no PID recorded yet); try again in a moment") + return False print(f"Stopping server (PID {pid})...") try: @@ -368,6 +901,7 @@ def stop_server() -> bool: for _ in range(50): # 5 seconds timeout time.sleep(0.1) if not is_server_running(): + _clear_stale_state_files() print("Server stopped") return True @@ -380,6 +914,7 @@ def stop_server() -> bool: time.sleep(0.5) if not is_server_running(): + _clear_stale_state_files() print("Server stopped (forced)") return True else: @@ -388,10 +923,16 @@ def stop_server() -> bool: except ProcessLookupError: print("Server process not found") - PID_FILE.unlink() return False +def _server_url() -> str: + endpoint = read_endpoint() + if endpoint is None: + return "(not listening yet)" + return f"http://{HOST}:{endpoint['port']}" + + def get_server_info() -> Optional[Dict[str, Any]]: """Get information about the running server. @@ -401,10 +942,10 @@ def get_server_info() -> Optional[Dict[str, Any]]: if not is_server_running(): return None + pid = read_pid() try: - pid = int(PID_FILE.read_text()) - port = int(PORT_FILE.read_text()) if PORT_FILE.exists() else PORT - + if pid is None: + raise LookupError("no PID recorded yet (server starting)") # Try to get process info import psutil proc = psutil.Process(pid) @@ -412,22 +953,38 @@ def get_server_info() -> Optional[Dict[str, Any]]: return { 'pid': pid, - 'url': f"http://{HOST}:{port}", - 'log_file': str(LOG_FILE), + 'url': _server_url(), + 'log_file': str(log_file()), 'uptime': str(uptime).split('.')[0], # Remove microseconds 'cpu_percent': proc.cpu_percent(interval=0.1), 'memory_mb': proc.memory_info().rss / 1024 / 1024, } except Exception: - # psutil not available, return basic info - port = int(PORT_FILE.read_text()) if PORT_FILE.exists() else PORT + # psutil not available (or no PID recorded yet), return basic info return { 'pid': pid, - 'url': f"http://{HOST}:{port}", - 'log_file': str(LOG_FILE), + 'url': _server_url(), + 'log_file': str(log_file()), } +def _start_foreground() -> None: + """Take the start lock, load GSAS-II and serve until stopped.""" + if not _claim_server_lock(): + print("Server is already running") + print(f" PID: {read_pid() or '(starting)'}") + print(f" URL: {_server_url()}") + print(f"\n To view logs: pixi run gsas-server logs") + sys.exit(1) + # Release the lock and state files even if the GSAS-II import below fails. + atexit.register(_release_state_files) + + print("Starting GSAS-II server...") + logger = setup_logging() + server = GSASServer(logger) + server.start() + + def main(): """Main entry point for server management.""" import argparse @@ -444,19 +1001,20 @@ def main(): args = parser.parse_args() action = args.action or 'status' # Default to status if no action - if action == 'start': - if is_server_running(): - print("Server is already running") - print(f" PID: {PID_FILE.read_text()}") - port = int(PORT_FILE.read_text()) if PORT_FILE.exists() else PORT - print(f" URL: http://{HOST}:{port}") - print(f"\n To view logs: tail -f {LOG_FILE}") - sys.exit(1) + if action in ('start', 'restart') and server_disabled(): + print(f"Not starting: {NO_SERVER_MESSAGE}") + sys.exit(1) - print("Starting GSAS-II server...") - logger = setup_logging(LOG_FILE) - server = GSASServer(logger) - server.start() + try: + _run_action(action) + except ServerStateError as e: + print(f"Error: {e}") + sys.exit(1) + + +def _run_action(action: str) -> None: + if action == 'start': + _start_foreground() elif action == 'stop': stop_server() @@ -474,11 +1032,11 @@ def main(): print(f" CPU: {info['cpu_percent']:.1f}%") if 'memory_mb' in info: print(f" Memory: {info['memory_mb']:.1f} MB") - print(f"\n To view logs: tail -f {info['log_file']}") + print(f"\n To view logs: pixi run gsas-server logs") else: print("Server is not running") print(f"\n To start: pixi run gsas-server start") - print(f" Log file: {LOG_FILE}") + print(f" Log file: {log_file()}") sys.exit(1) elif action == 'restart': @@ -486,32 +1044,33 @@ def main(): stop_server() time.sleep(1) - print("Starting GSAS-II server...") - logger = setup_logging(LOG_FILE) - server = GSASServer(logger) - server.start() + _start_foreground() elif action == 'logs': - if not LOG_FILE.exists(): - print(f"Log file not created yet: {LOG_FILE}") + path = log_file() + try: + text = _read_state_file(LOG_NAME) + except OSError as e: + print(f"Failed to read log file: {e}") + sys.exit(1) + if text is None: + print(f"Log file not created yet: {path}") sys.exit(0) - print(f"Recent logs from {LOG_FILE}:") + print(f"Recent logs from {path}:") print(" (showing last 50 lines)") print("-" * 60) try: # Show last 50 lines - with open(LOG_FILE, 'r') as f: - lines = f.readlines() - for line in lines[-50:]: - print(line.rstrip()) + for line in text.splitlines()[-50:]: + print(line.rstrip()) except Exception as e: print(f"Failed to read log file: {e}") sys.exit(1) print("-" * 60) - print(f" To follow logs in real-time: tail -f {LOG_FILE}") + print(f" To follow logs in real-time: tail -f {path}") if __name__ == '__main__': diff --git a/src/powderline/kicker.py b/src/powderline/kicker.py index 3a38b77..3f288bc 100644 --- a/src/powderline/kicker.py +++ b/src/powderline/kicker.py @@ -3393,6 +3393,11 @@ def run( subprocess. Suitable for batch/HPC workflows where each refinement is isolated. + Only the calling user's own server is used. If the + ``POWDERLINE_NO_SERVER`` environment variable is truthy + (``1``/``true``/``yes``/``on``), ``'auto'`` runs in-process without + starting a server and ``'server'`` returns an error result. + Returns: **Normal refinement** (``validate_only=False``): @@ -3878,6 +3883,7 @@ def run_refinement(recipe: RecipeModel, output_dir: Path, verbose: bool = False, Default (no flags): Try server → auto-start if needed → fall back to subprocess --use-server: Use server only (fail if unavailable, auto-start enabled) --no-server: Use subprocess only (skip server entirely) + POWDERLINE_NO_SERVER=1 (env): never use a server; default mode runs in-process Examples: pixi run kicker recipe.json # Smart mode (auto-detect) @@ -3937,6 +3943,10 @@ def run_refinement(recipe: RecipeModel, output_dir: Path, verbose: bool = False, if args.no_server: execution_mode = 'subprocess' elif args.use_server: + from powderline.gsas_server import NO_SERVER_MESSAGE, server_disabled + if server_disabled(): + print(f"\n{CROSS} --use-server: {NO_SERVER_MESSAGE}\n") + sys.exit(1) execution_mode = 'server' else: execution_mode = 'auto' diff --git a/tests/conftest.py b/tests/conftest.py index 3b54f39..0a88505 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -7,6 +7,25 @@ import shutil +@pytest.fixture(autouse=True) +def _private_server_state(request, tmp_path_factory, monkeypatch): + """Keep every test away from the developer's real GSAS-II server state. + + A probe there would create the per-user state directory and could submit + jobs to the developer's own running server. Each test gets a private + directory instead (in-process only; subprocesses are unaffected). The + state-path logic itself is tested in test_gsas_server_unit.py, which + isolates every case explicitly. + """ + if request.module.__name__.endswith("test_gsas_server_unit"): + return + from powderline import gsas_server + + root = tmp_path_factory.mktemp("server_state") + root.chmod(0o700) + monkeypatch.setattr(gsas_server, "_state_dir_path", lambda: root / "powderline") + + @pytest.fixture def example_LaB6_dir(): """Path to example_LaB6 directory.""" diff --git a/tests/test_api.py b/tests/test_api.py index 0c83c6e..543ef89 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -407,8 +407,72 @@ def test_run_server_mode_failure_normalizes_result(recipe_LaB6_dict, tmp_path): assert set(result["refined_parameters"].columns) == expected_cols +# ─── POWDERLINE_NO_SERVER (deployment off switch) ───────────────────────────── + +def _no_server_paths(): + """Patches that fail the test if any server path is touched.""" + from powderline import gsas_client as _gc + + def _fail(*a, **k): + raise AssertionError("server path used despite POWDERLINE_NO_SERVER") + + return (patch.object(_gc.GSASClient, "is_server_available", side_effect=_fail), + patch.object(_gc.GSASClient, "_start_server_background", side_effect=_fail), + patch.object(_gc.GSASClient, "_submit_to_server", side_effect=_fail)) + + +def test_run_auto_mode_with_no_server_env_runs_in_process( + recipe_LaB6_dict, tmp_path, monkeypatch +): + """POWDERLINE_NO_SERVER: run(execution_mode='auto') never probes or starts a server.""" + monkeypatch.setenv("POWDERLINE_NO_SERVER", "1") + probe, start, submit = _no_server_paths() + with probe, start, submit: + result = powderline.run(recipe_LaB6_dict, tmp_path / "out", execution_mode="auto") + + assert result["success"] is True, f"Refinement failed: {result.get('error')}" + assert result["method"] == "subprocess" + + +def test_run_server_mode_with_no_server_env_errors(recipe_LaB6_dict, tmp_path, monkeypatch): + """POWDERLINE_NO_SERVER: run(execution_mode='server') returns a clear error result.""" + monkeypatch.setenv("POWDERLINE_NO_SERVER", "true") + probe, start, submit = _no_server_paths() + with probe, start, submit: + result = powderline.run(recipe_LaB6_dict, tmp_path / "out", execution_mode="server") + + assert result["success"] is False + assert "POWDERLINE_NO_SERVER" in result["error"] + assert isinstance(result["fit_profile"], pd.DataFrame) + + # ─── HTTP retry logic tests ──────────────────────────────────────────────────── +def test_cli_use_server_with_no_server_env_exits(tmp_path): + """POWDERLINE_NO_SERVER: `kicker --use-server` exits 1 with a clear message.""" + import os + import subprocess + import sys + + repo = Path(__file__).parent.parent + env = {**os.environ, "POWDERLINE_NO_SERVER": "1", + "PYTHONPATH": str(repo / "src") + os.pathsep + os.environ.get("PYTHONPATH", "")} + proc = subprocess.run( + [sys.executable, str(repo / "src" / "powderline" / "kicker.py"), + str(repo / "examples" / "example_LaB6" / "input.json"), + "--use-server", "--output", str(tmp_path / "out")], + env=env, capture_output=True, text=True, encoding="utf-8", errors="replace", + timeout=300, + ) + assert proc.returncode == 1, proc.stdout + proc.stderr + assert "POWDERLINE_NO_SERVER" in proc.stdout + assert not (tmp_path / "out" / "fit_profile.txt").exists() + + +# This user's server endpoint as read from the private state dir. +_FAKE_ENDPOINT = {"pid": 4242, "port": 50123, "token": "t0ken"} + + def test_submit_to_server_retries_on_connect_error(recipe_LaB6_dict, tmp_path): """_submit_to_server() retries up to 3 times on ConnectError before raising.""" import httpx @@ -417,7 +481,8 @@ def test_submit_to_server_retries_on_connect_error(recipe_LaB6_dict, tmp_path): client = GSASClient() connect_error = httpx.ConnectError("Simulated connection refused") - with patch("httpx.post", side_effect=connect_error) as mock_post: + with patch.object(GSASClient, "_endpoint", return_value=_FAKE_ENDPOINT), \ + patch("httpx.post", side_effect=connect_error) as mock_post: with pytest.raises(httpx.ConnectError): client._submit_to_server(recipe_LaB6_dict, tmp_path / "out") @@ -440,7 +505,8 @@ def test_submit_to_server_does_not_retry_on_http_status_error( "500 Server Error", request=MagicMock(), response=mock_response ) - with patch("httpx.post", side_effect=http_error) as mock_post: + with patch.object(GSASClient, "_endpoint", return_value=_FAKE_ENDPOINT), \ + patch("httpx.post", side_effect=http_error) as mock_post: with pytest.raises(httpx.HTTPStatusError): client._submit_to_server(recipe_LaB6_dict, tmp_path / "out") diff --git a/tests/test_gsas_client_visibility.py b/tests/test_gsas_client_visibility.py index c4df80e..a2e8f85 100644 --- a/tests/test_gsas_client_visibility.py +++ b/tests/test_gsas_client_visibility.py @@ -8,21 +8,31 @@ not freshly written after a "successful" server run — a stale file from a previous run into the same output_dir does not count), warn, and fall back to in-process execution — or return a structured error when fallback is -disabled. Also covers the use_server bypass and the keyword-signature -regression guard. +disabled. Also covers the use_server bypass, the keyword-signature +regression guard, server identity (a server must prove it holds this user's +token) and the POWDERLINE_NO_SERVER switch. """ from __future__ import annotations +import json +import os from pathlib import Path import pytest +from powderline import gsas_client, gsas_server from powderline.gsas_client import GSASClient RECIPE = {"schema_name": "GSASII_Rietveld", "schema_version": "0.26.0", "payload": {}} +@pytest.fixture(autouse=True) +def _server_enabled(monkeypatch): + """Tests assume server use is allowed unless they opt out explicitly.""" + monkeypatch.delenv("POWDERLINE_NO_SERVER", raising=False) + + @pytest.fixture def client(monkeypatch): c = GSASClient() @@ -189,3 +199,210 @@ def test_submit_simulation_wrong_keyword_raises(): client = GSASClient() with pytest.raises(TypeError, match="recipe_path"): client.submit_simulation(recipe_path=Path("dummy.json"), output_dir=Path(".")) + + +# --- Server identity: only this user's server, proven by its token --- + +class _FakeResponse: + """Minimal httpx.Response stand-in.""" + + def __init__(self, status_code, body, headers=None): + self.status_code = status_code + self._body = body + self.content = json.dumps(body).encode() + self.headers = headers or {} + + def json(self): + return self._body + + def raise_for_status(self): + pass + + +def _signed_response(token, request_headers, body, status_code=200): + """A response signed with ``token`` for the request that carried ``request_headers``.""" + resp = _FakeResponse(status_code, body) + resp.headers = {gsas_server.RESPONSE_SIGNATURE_HEADER: gsas_server.sign_response( + token, request_headers[gsas_server.NONCE_HEADER], status_code, resp.content)} + return resp + + +@pytest.fixture +def own_endpoint(monkeypatch, tmp_path): + monkeypatch.setattr(gsas_server, "state_dir", lambda: tmp_path) + # A live PID: endpoints of dead servers are ignored before any probe. + (tmp_path / "server.json").write_text( + json.dumps({"pid": os.getpid(), "port": 50123, "token": "my-token"})) + return tmp_path + + +def _fake_health(monkeypatch, respond): + """Patch httpx.get; ``respond(headers) -> _FakeResponse``. Returns call log.""" + import httpx + calls = [] + + def fake_get(url, headers=None, **kwargs): + calls.append({"url": url, "headers": headers, **kwargs}) + return respond(headers) + + monkeypatch.setattr(httpx, "get", fake_get) + return calls + + +def _assert_token_not_sent(call): + sent = json.dumps({k: v for k, v in call.items() if k != "content"}, default=str) + sent += str(call.get("content", b"")) + assert "my-token" not in sent, "the token itself must never be sent" + + +def test_server_unavailable_without_endpoint(monkeypatch, tmp_path): + monkeypatch.setattr(gsas_server, "state_dir", lambda: tmp_path) + calls = _fake_health(monkeypatch, lambda h: pytest.fail("must not probe")) + assert GSASClient().is_server_available() is False + assert calls == [] + + +def test_endpoint_of_dead_server_is_not_probed(monkeypatch, own_endpoint): + monkeypatch.setattr(gsas_server, "_pid_alive", lambda pid: False) + calls = _fake_health(monkeypatch, lambda h: pytest.fail("must not probe")) + assert GSASClient().is_server_available() is False + assert calls == [] + + +def test_server_available_when_response_signed_with_token(monkeypatch, own_endpoint): + calls = _fake_health(monkeypatch, lambda h: _signed_response("my-token", h, {"status": "ok"})) + assert GSASClient().is_server_available() is True + (call,) = calls + assert call["url"] == "http://127.0.0.1:50123/health" + assert call["trust_env"] is False # never sent through a proxy + _assert_token_not_sent(call) + h = call["headers"] + assert h[gsas_server.SIGNATURE_HEADER] == gsas_server.sign_request( + "my-token", "GET", "/health", b"", h[gsas_server.NONCE_HEADER], + h[gsas_server.TIMESTAMP_HEADER], b"") + + +def test_nonce_is_fresh_per_request(monkeypatch, own_endpoint): + calls = _fake_health(monkeypatch, lambda h: _signed_response("my-token", h, {})) + client = GSASClient() + client.is_server_available() + client.is_server_available() + assert calls[0]["headers"][gsas_server.NONCE_HEADER] != \ + calls[1]["headers"][gsas_server.NONCE_HEADER] + + +def _echo_impostor(headers): + """A listener without the token that signs with whatever secret it was sent + (a bearer token, if the client ever sent one again).""" + key = headers.get("Authorization", "").removeprefix("Bearer ") \ + or headers.get(gsas_server.SIGNATURE_HEADER, "") + return _signed_response(key, headers, {"status": "ok"}) + + +@pytest.mark.parametrize("respond", [ + lambda h: _FakeResponse(200, {"status": "ok"}), # unsigned / old server + lambda h: _FakeResponse(200, {"status": "ok"}, + {gsas_server.RESPONSE_SIGNATURE_HEADER: "0" * 64}), + lambda h: _signed_response("other-token", h, {"status": "ok"}), # another user's server + lambda h: _signed_response("my-token", {gsas_server.NONCE_HEADER: "replayed"}, {}), + lambda h: _FakeResponse(401, {"detail": "Unauthorized"}), + _echo_impostor, # port squatter +]) +def test_server_unavailable_without_valid_response_signature(monkeypatch, own_endpoint, respond): + _fake_health(monkeypatch, respond) + assert GSASClient().is_server_available() is False + + +def test_unsafe_state_dir_means_no_server(monkeypatch, capsys): + def unsafe(): + raise gsas_server.ServerStateError("unsafe state dir") + + monkeypatch.setattr(gsas_server, "state_dir", unsafe) + _fake_health(monkeypatch, lambda h: pytest.fail("must not probe")) + c = GSASClient() + assert c.is_server_available() is False + assert c.is_server_available() is False + assert c._start_server_background() is False + assert capsys.readouterr().out.count("unsafe state dir") == 1 + + +def _fake_post(monkeypatch, respond): + import httpx + seen = {} + + def fake_post(url, content=None, headers=None, **kwargs): + seen.update(url=url, content=content, headers=headers, **kwargs) + return respond(headers) + + monkeypatch.setattr(httpx, "post", fake_post) + return seen + + +def test_submit_is_signed_and_response_verified(monkeypatch, own_endpoint, tmp_path): + seen = _fake_post(monkeypatch, lambda h: _signed_response("my-token", h, {"success": True})) + assert GSASClient()._submit_to_server(RECIPE, tmp_path / "out") == {"success": True} + assert seen["url"] == "http://127.0.0.1:50123/simulate" + assert seen["trust_env"] is False + _assert_token_not_sent(seen) + h = seen["headers"] + assert h[gsas_server.SIGNATURE_HEADER] == gsas_server.sign_request( + "my-token", "POST", "/simulate", b"", h[gsas_server.NONCE_HEADER], + h[gsas_server.TIMESTAMP_HEADER], seen["content"]) + + +def test_forged_server_result_is_never_trusted(monkeypatch, own_endpoint, tmp_path): + """A listener that took over the port answers 'success' without the token: + the run must not report success from it.""" + _fake_post(monkeypatch, lambda h: _FakeResponse(200, {"success": True, "rwp": 1.0})) + monkeypatch.setattr(GSASClient, "is_server_available", lambda self: True) + c = GSASClient(fallback_to_subprocess=False) + with pytest.raises(gsas_client.ServerIdentityError): + c._submit_to_server(RECIPE, tmp_path / "out") + result = c.submit_simulation(RECIPE, tmp_path / "out", auto_start_server=False) + assert result["success"] is False + assert "did not prove" in result["error"] + + +# --- POWDERLINE_NO_SERVER --- + +@pytest.fixture +def no_server(monkeypatch): + monkeypatch.setenv("POWDERLINE_NO_SERVER", "1") + + +def _forbid_server(monkeypatch, c): + for name in ("is_server_available", "_submit_to_server"): + monkeypatch.setattr(c, name, lambda *a, **k: pytest.fail(f"{name} called")) + monkeypatch.setattr(c, "_start_server_background", + lambda *a, **k: pytest.fail("server auto-started")) + + +def test_no_server_env_auto_runs_in_process(no_server, monkeypatch, tmp_path, capsys): + c = GSASClient() + _forbid_server(monkeypatch, c) + monkeypatch.setattr(c, "_submit_via_subprocess", + lambda rd, od, verbose: {"success": True}) + result = c.submit_simulation(RECIPE, tmp_path, auto_start_server=True) + assert result == {"success": True, "method": "subprocess"} + assert "POWDERLINE_NO_SERVER" in capsys.readouterr().out + + +def test_no_server_env_server_mode_errors(no_server, monkeypatch, tmp_path): + c = GSASClient(fallback_to_subprocess=False) + _forbid_server(monkeypatch, c) + monkeypatch.setattr(c, "_submit_via_subprocess", + lambda *a: pytest.fail("must not run in-process")) + result = c.submit_simulation(RECIPE, tmp_path) + assert result["success"] is False + assert "POWDERLINE_NO_SERVER" in result["error"] + + +def test_no_server_env_blocks_background_start(no_server, monkeypatch): + import subprocess as sp + monkeypatch.setattr(sp, "Popen", lambda *a, **k: pytest.fail("server spawned")) + assert GSASClient()._start_server_background() is False + + +def test_no_server_env_direct_probe_contacts_nothing(no_server, monkeypatch, own_endpoint): + _fake_health(monkeypatch, lambda h: pytest.fail("must not contact a server")) + assert GSASClient().is_server_available() is False diff --git a/tests/test_gsas_server_http.py b/tests/test_gsas_server_http.py index 59efcf6..46e21a8 100644 --- a/tests/test_gsas_server_http.py +++ b/tests/test_gsas_server_http.py @@ -1,22 +1,17 @@ """Tests for GSAS-II server HTTP endpoints. -These tests verify that the FastAPI server exposes the correct HTTP API -and handles requests properly. - -NOTE: These are integration tests that require a running server. -They can be run in two ways: - -1. Manual server start (recommended): - ```bash - pixi run gsas-server start - pixi run pytest tests/test_gsas_server_http.py -v - pixi run gsas-server stop - ``` - -2. Auto-start in fixture (slower, may be flaky): - Just run pytest and the fixture will attempt to start a test server. +These tests verify that the FastAPI server exposes the correct HTTP API, +handles requests properly, and rejects callers that do not hold its token. + +These are integration tests: the module fixture starts a real server in a +private, per-test state directory (via ``XDG_RUNTIME_DIR`` / ``LOCALAPPDATA``), +so it never touches — or reuses — the developer's own server, and reads the +server's port and secret token from that directory. They skip if the server +cannot start (e.g. GSAS-II unavailable). """ +import json +import secrets import pytest import time import subprocess @@ -25,59 +20,56 @@ import tempfile import os +from powderline import gsas_server +from powderline.gsas_client import GSASClient -def is_server_running(port: int = 19471) -> bool: - """Check if server is running on given port.""" - import httpx + +def _endpoint_in(state_root: Path): + """The endpoint (pid/port/token) the server wrote under ``state_root``.""" + try: + return json.loads((state_root / 'powderline' / 'server.json').read_text()) + except (OSError, ValueError): + return None + + +def _signed(port: int, token: str, method: str, path: str, payload=None): + """Send a request signed with ``token``; the response signature is verified + (raises ``ServerIdentityError`` otherwise).""" + return GSASClient._signed_request({'pid': 0, 'port': port, 'token': token}, + method, path, payload, timeout=5.0) + + +def _health_ok(port: int, token: str) -> bool: try: - resp = httpx.get(f"http://127.0.0.1:{port}/health", timeout=1.0) - return resp.status_code == 200 - except (httpx.ConnectError, httpx.TimeoutException, OSError): + return _signed(port, token, 'GET', '/health').status_code == 200 + except Exception: return False @pytest.fixture(scope="module") -def server_port(): - """Use default port or find available port for testing.""" - # First try default port (19471) - if is_server_running(19471): - yield 19471 - return - - # If default port not available, try to find a free port - import socket - sock = socket.socket() - sock.bind(('', 0)) - port = sock.getsockname()[1] - sock.close() - yield port +def server_state_root(tmp_path_factory): + """A private (0700) directory standing in for this user's runtime dir.""" + root = tmp_path_factory.mktemp("server_state") + root.chmod(0o700) + return root @pytest.fixture(scope="module") -def test_server(server_port): - """Ensure test server is running on the specified port. - - If server is already running on the port, use it. - Otherwise, attempt to start one for testing. - """ - # Check if already running - if is_server_running(server_port): - yield server_port - return - - # Try to start test server +def test_server(server_state_root): + """Start a server in an isolated state dir; yield ``(port, token)``.""" src_dir = Path(__file__).parent.parent / 'src' server_script = src_dir / 'powderline' / 'gsas_server.py' if not server_script.exists(): pytest.skip(f"Server script not found: {server_script}") - # Set custom port via environment env = os.environ.copy() env['PYTHONPATH'] = str(src_dir) + os.pathsep + env.get('PYTHONPATH', '') - env['POWDERLINE_SERVER_PORT'] = str(server_port) + env['XDG_RUNTIME_DIR'] = str(server_state_root) # POSIX state dir + env['LOCALAPPDATA'] = str(server_state_root) # Windows state dir + env.pop('POWDERLINE_NO_SERVER', None) + env.pop('POWDERLINE_SERVER_PORT', None) # exercise the ephemeral port - # Start server process proc = subprocess.Popen( [sys.executable, str(server_script), 'start'], env=env, @@ -85,24 +77,21 @@ def test_server(server_port): stderr=subprocess.DEVNULL ) - # Wait for server to start (up to 15 seconds for GSAS-II import) - started = False - for attempt in range(150): + # Wait for server to start (up to 30 seconds for GSAS-II import) + endpoint = None + for attempt in range(300): time.sleep(0.1) - if is_server_running(server_port): - started = True + endpoint = _endpoint_in(server_state_root) + if endpoint and _health_ok(endpoint['port'], endpoint['token']): break - # Check if process died if proc.poll() is not None: pytest.skip(f"Test server exited unexpectedly (exit code: {proc.returncode})") - - if not started: + else: proc.kill() - pytest.skip(f"Test server failed to start within 15 seconds on port {server_port}") + pytest.skip("Test server failed to start within 30 seconds") - yield server_port + yield endpoint['port'], endpoint['token'] - # Cleanup - only kill if we started it proc.terminate() try: proc.wait(timeout=5) @@ -114,8 +103,8 @@ def test_health_endpoint(test_server): """Test /health endpoint returns correct format.""" import httpx - port = test_server - resp = httpx.get(f"http://127.0.0.1:{port}/health", timeout=2.0) + port, token = test_server + resp = _signed(port, token, 'GET', '/health') assert resp.status_code == 200, "Health endpoint should return 200" @@ -136,8 +125,8 @@ def test_health_endpoint_format(test_server): """Verify health endpoint returns proper JSON structure.""" import httpx - port = test_server - resp = httpx.get(f"http://127.0.0.1:{port}/health", timeout=2.0) + port, token = test_server + resp = _signed(port, token, 'GET', '/health') # Verify response is valid JSON data = resp.json() @@ -152,29 +141,20 @@ def test_simulate_endpoint_missing_fields(test_server): """Test /simulate endpoint validates required fields.""" import httpx - port = test_server + port, token = test_server # Missing recipe_data - resp = httpx.post( - f"http://127.0.0.1:{port}/simulate", - json={"output_dir": "/tmp/test"}, - timeout=5.0 + resp = _signed(port, token, 'POST', '/simulate', {"output_dir": "/tmp/test"} ) assert resp.status_code == 422, "Should return validation error for missing recipe_data" # Missing output_dir - resp = httpx.post( - f"http://127.0.0.1:{port}/simulate", - json={"recipe_data": {}}, - timeout=5.0 + resp = _signed(port, token, 'POST', '/simulate', {"recipe_data": {}} ) assert resp.status_code == 422, "Should return validation error for missing output_dir" # Empty request - resp = httpx.post( - f"http://127.0.0.1:{port}/simulate", - json={}, - timeout=5.0 + resp = _signed(port, token, 'POST', '/simulate', {} ) assert resp.status_code == 422, "Should return validation error for empty request" @@ -183,17 +163,14 @@ def test_simulate_endpoint_invalid_recipe(test_server): """Test /simulate endpoint handles invalid recipe_data gracefully.""" import httpx - port = test_server + port, token = test_server # Invalid (empty) recipe_data — server should return 200 with success=False with tempfile.TemporaryDirectory() as tmpdir: - resp = httpx.post( - f"http://127.0.0.1:{port}/simulate", - json={ + resp = _signed(port, token, 'POST', '/simulate', { "recipe_data": {}, "output_dir": tmpdir - }, - timeout=5.0 + } ) assert resp.status_code == 200, "Should return 200 even for errors (error in response body)" @@ -210,31 +187,25 @@ def test_simulate_endpoint_optional_verbose(test_server): """Test /simulate endpoint accepts optional verbose parameter.""" import httpx - port = test_server + port, token = test_server with tempfile.TemporaryDirectory() as tmpdir: # Request with verbose=true — should accept and return 200 (even if recipe invalid) - resp = httpx.post( - f"http://127.0.0.1:{port}/simulate", - json={ + resp = _signed(port, token, 'POST', '/simulate', { "recipe_data": {}, "output_dir": tmpdir, "verbose": True - }, - timeout=5.0 + } ) assert resp.status_code == 200, "Should accept verbose parameter" # Request with verbose=false - resp = httpx.post( - f"http://127.0.0.1:{port}/simulate", - json={ + resp = _signed(port, token, 'POST', '/simulate', { "recipe_data": {}, "output_dir": tmpdir, "verbose": False - }, - timeout=5.0 + } ) assert resp.status_code == 200, "Should accept verbose parameter" @@ -244,10 +215,10 @@ def test_health_endpoint_performance(test_server): """Test that health endpoint responds quickly.""" import httpx - port = test_server + port, token = test_server start = time.time() - resp = httpx.get(f"http://127.0.0.1:{port}/health", timeout=2.0) + resp = _signed(port, token, 'GET', '/health') elapsed = time.time() - start assert resp.status_code == 200 @@ -258,11 +229,11 @@ def test_multiple_health_checks(test_server): """Test that multiple health checks work correctly.""" import httpx - port = test_server + port, token = test_server # Multiple rapid health checks for _ in range(10): - resp = httpx.get(f"http://127.0.0.1:{port}/health", timeout=2.0) + resp = _signed(port, token, 'GET', '/health') assert resp.status_code == 200 data = resp.json() assert data["status"] == "ok" @@ -272,25 +243,104 @@ def test_server_increments_request_count(test_server): """Test that server tracks request count correctly.""" import httpx - port = test_server + port, token = test_server # Get initial count (might not be 0 due to startup checks) - resp = httpx.get(f"http://127.0.0.1:{port}/health", timeout=2.0) + resp = _signed(port, token, 'GET', '/health') initial_count = resp.json()["request_count"] # Make a simulation request (will fail validation but should still increment counter) with tempfile.TemporaryDirectory() as tmpdir: - httpx.post( - f"http://127.0.0.1:{port}/simulate", - json={ + _signed(port, token, 'POST', '/simulate', { "recipe_data": {}, "output_dir": tmpdir - }, - timeout=5.0 + } ) # Check count increased - resp = httpx.get(f"http://127.0.0.1:{port}/health", timeout=2.0) + resp = _signed(port, token, 'GET', '/health') new_count = resp.json()["request_count"] assert new_count > initial_count, "Request count should increment after simulation request" + + +# --- Multi-user safety: only the token holder can use the server --- + +def _forged_headers(token: str, method: str, path: str, body: bytes) -> dict: + """Headers signed with ``token`` (here: a token that is not the server's).""" + nonce, ts = secrets.token_hex(16), str(int(time.time())) + return {gsas_server.NONCE_HEADER: nonce, gsas_server.TIMESTAMP_HEADER: ts, + gsas_server.SIGNATURE_HEADER: gsas_server.sign_request( + token, method, path, b'', nonce, ts, body), + "Content-Type": "application/json"} + + +def test_requests_without_token_rejected(test_server, tmp_path): + """Unsigned, bearer-style, or wrongly signed requests -> 401 on every route, + and nothing is written.""" + import httpx + + port, token = test_server + out = tmp_path / "other_users_output" + body = json.dumps({"recipe_data": {}, "output_dir": str(out)}).encode() + for headers in ({}, + {"Authorization": f"Bearer {token}"}, # the old scheme: refused + _forged_headers("not-the-token", "POST", "/simulate", body)): + resp = httpx.post(f"http://127.0.0.1:{port}/simulate", headers=headers, + content=body, timeout=5.0, trust_env=False) + assert resp.status_code == 401 + resp = httpx.get(f"http://127.0.0.1:{port}/health", headers=headers, + timeout=2.0, trust_env=False) + assert resp.status_code == 401 + assert "pid" not in resp.text + assert gsas_server.RESPONSE_SIGNATURE_HEADER not in resp.headers + assert not out.exists(), "an unauthenticated request must not create output_dir" + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX permission bits") +def test_state_files_are_owner_only(test_server, server_state_root): + state = server_state_root / "powderline" + assert (state.stat().st_mode & 0o777) == 0o700 + for name in ("server.json", "server.pid", "server.lock", "server.log"): + assert (state / name).stat().st_mode & 0o777 == 0o600, name + + +def test_live_server_holds_start_lock(test_server, server_state_root, monkeypatch): + """A second start is refused while the server runs; status sees it running.""" + monkeypatch.setattr(gsas_server, "state_dir", lambda: server_state_root / "powderline") + assert gsas_server.is_server_running() is True + assert gsas_server._claim_server_lock(attempts=2, delay=0) is False + + +def test_client_uses_own_server(test_server, server_state_root, monkeypatch): + monkeypatch.delenv("POWDERLINE_NO_SERVER", raising=False) + monkeypatch.setattr(gsas_server, "state_dir", lambda: server_state_root / "powderline") + assert GSASClient().is_server_available() is True + + +def test_client_ignores_another_users_server(test_server, tmp_path, monkeypatch): + """A client whose state dir lacks the token never treats the server as its own, + even when it knows the port (here: a forged endpoint with a guessed token).""" + port, _ = test_server + monkeypatch.setattr(gsas_server, "state_dir", lambda: tmp_path) + client = GSASClient() + assert client.is_server_available() is False + + (tmp_path / "server.json").write_text( + json.dumps({"pid": os.getpid(), "port": port, "token": "guessed"})) + assert client.is_server_available() is False + with pytest.raises(Exception): + client._submit_to_server({}, tmp_path / "out") + assert not (tmp_path / "out").exists() + + +def test_refinement_through_authenticated_server(test_server, server_state_root, + monkeypatch, recipe_LaB6_dict, tmp_path): + """End to end: the owner's client runs a real refinement via the server.""" + monkeypatch.delenv("POWDERLINE_NO_SERVER", raising=False) + monkeypatch.setattr(gsas_server, "state_dir", lambda: server_state_root / "powderline") + client = GSASClient(fallback_to_subprocess=False) + result = client.submit_simulation(recipe_LaB6_dict, tmp_path, auto_start_server=False) + assert result["success"] is True, result.get("error") + assert result["method"] == "server" + assert (tmp_path / "fit_profile.txt").exists() diff --git a/tests/test_gsas_server_unit.py b/tests/test_gsas_server_unit.py index 34c418d..a1c4519 100644 --- a/tests/test_gsas_server_unit.py +++ b/tests/test_gsas_server_unit.py @@ -1,18 +1,23 @@ """Unit tests for gsas_server process-management helpers. -Covers the cross-platform liveness probe (_pid_alive), is_server_running(), -and stop_server() PID-file handling. All os.kill / PID-file interactions are -mocked with monkeypatch + tmp_path so NO real server is ever started and no -signals reach real processes. +Covers the cross-platform liveness probe (_pid_alive), is_server_running() +and stop_server(), plus the multi-user safety layer: the per-user state +directory checks, the kernel-held start lock, request/response signing on +every route, and the POWDERLINE_NO_SERVER switch. All os.kill interactions are +mocked and state files live under tmp_path, so NO real server is ever started +and no signals reach real processes. Import note: gsas_server.py does NOT import GSAS-II at module top (only stdlib + pydantic); GSAS-II is imported lazily inside GSASServer.__init__ / the request handlers. A plain module import is therefore safe without GSAS-II. """ +import json import os +import secrets import signal import subprocess import sys +import time import pytest @@ -34,12 +39,20 @@ @pytest.fixture def isolated_pid_files(monkeypatch, tmp_path): - """Point PID_FILE / PORT_FILE at a throwaway tmp dir (nonexistent by default).""" - pid_file = tmp_path / "server.pid" - port_file = tmp_path / "server.port" - monkeypatch.setattr(gsas_server, "PID_FILE", pid_file) - monkeypatch.setattr(gsas_server, "PORT_FILE", port_file) - return pid_file, port_file + """Point the state dir at a throwaway tmp dir (state files absent by default).""" + monkeypatch.setattr(gsas_server, "state_dir", lambda: tmp_path) + yield tmp_path / "server.pid", tmp_path / "server.json" + gsas_server._release_state_files() # drop a lock a test may have claimed + + +@pytest.fixture +def other_holder(isolated_pid_files): + """Hold the start lock through a separate descriptor, like another server would.""" + fd = gsas_server._open_state_file(gsas_server.LOCK_NAME, os.O_RDWR | os.O_CREAT) + assert gsas_server._try_lock(fd) + yield + gsas_server._unlock(fd) + os.close(fd) # --- _pid_alive: cross-platform contract (real PIDs, no mocking) --- @@ -83,69 +96,64 @@ def _raise(pid, sig): assert gsas_server._pid_alive(1) is True -# --- is_server_running --- +# --- is_server_running: holding the start lock is the definition --- -def test_is_server_running_false_without_pid_file(isolated_pid_files): - pid_file, _ = isolated_pid_files - assert not pid_file.exists() +def test_is_server_running_false_without_any_state(isolated_pid_files): assert gsas_server.is_server_running() is False -def test_is_server_running_true_for_live_pid(isolated_pid_files, monkeypatch): - pid_file, _ = isolated_pid_files - pid_file.write_text("4242") - monkeypatch.setattr(gsas_server, "_pid_alive", lambda pid: True) +def test_is_server_running_true_while_lock_held(other_holder): assert gsas_server.is_server_running() is True - # Live PID file is left in place. - assert pid_file.exists() -def test_is_server_running_stale_pid_removes_file(isolated_pid_files, monkeypatch): +def test_is_server_running_ignores_stale_or_recycled_pid(isolated_pid_files): + # A leftover PID file naming a live (recycled) PID does not make a dead + # server look alive: nobody holds the lock. pid_file, _ = isolated_pid_files - pid_file.write_text("4242") - monkeypatch.setattr(gsas_server, "_pid_alive", lambda pid: False) + pid_file.write_text(str(os.getpid())) assert gsas_server.is_server_running() is False - # Stale PID file is cleaned up. - assert not pid_file.exists() -def test_is_server_running_garbage_pid_removes_file(isolated_pid_files): - pid_file, _ = isolated_pid_files - pid_file.write_text("not-an-int") +def test_is_server_running_probe_does_not_keep_the_lock(isolated_pid_files): assert gsas_server.is_server_running() is False - assert not pid_file.exists() + assert gsas_server._claim_server_lock(attempts=1) is True # --- stop_server --- def test_stop_server_when_not_running(isolated_pid_files, capsys): - # No PID file at all -> graceful no-op, returns False, no exception. + # No server at all -> graceful no-op, returns False, no exception. result = gsas_server.stop_server() assert result is False assert "not running" in capsys.readouterr().out.lower() -def test_stop_server_stale_pid_graceful(isolated_pid_files, monkeypatch, capsys): - # PID file present but process dead: is_server_running() detects the stale - # PID, removes the file, and stop_server() reports "not running" -> False. +def test_stop_server_never_signals_a_stale_pid(isolated_pid_files, monkeypatch, capsys): + # PID file left by a crashed server (its PID possibly recycled by another + # process): no lock holder -> "not running", and nothing is signalled. pid_file, _ = isolated_pid_files - pid_file.write_text("999999") - monkeypatch.setattr(gsas_server, "_pid_alive", lambda pid: False) + pid_file.write_text(str(os.getpid())) - # Guard: os.kill must never be invoked on a dead PID here. def _boom(pid, sig): - raise AssertionError("os.kill should not be called for a stale PID") + raise AssertionError("os.kill must not be called without a lock holder") monkeypatch.setattr(gsas_server.os, "kill", _boom) result = gsas_server.stop_server() assert result is False assert "not running" in capsys.readouterr().out.lower() - assert not pid_file.exists() + + +def test_stop_server_while_starting_up(other_holder, monkeypatch, capsys): + # Lock held but no PID recorded yet: nothing to signal, report and return. + monkeypatch.setattr(gsas_server.os, "kill", + lambda pid, sig: pytest.fail("nothing to signal yet")) + assert gsas_server.stop_server() is False + assert "starting up" in capsys.readouterr().out.lower() def test_stop_server_sends_sigterm_to_live_pid(isolated_pid_files, monkeypatch, capsys): - # Live PID: stop_server should signal it, then observe it gone on next poll. + # Running server: stop_server should signal it, then observe it gone on next poll. pid_file, _ = isolated_pid_files pid_file.write_text("4242") @@ -154,9 +162,9 @@ def test_stop_server_sends_sigterm_to_live_pid(isolated_pid_files, monkeypatch, def fake_kill(pid, sig): calls["signals"].append((pid, sig)) - # First is_server_running() (guard) True; after SIGTERM, report dead. - alive_states = iter([True, False]) - monkeypatch.setattr(gsas_server, "_pid_alive", lambda pid: next(alive_states)) + # First is_server_running() (guard) True; after SIGTERM, report stopped. + running_states = iter([True, False]) + monkeypatch.setattr(gsas_server, "_lock_held", lambda: next(running_states)) monkeypatch.setattr(gsas_server.os, "kill", fake_kill) monkeypatch.setattr(gsas_server.time, "sleep", lambda s: None) @@ -166,6 +174,26 @@ def fake_kill(pid, sig): assert calls["signals"][0][0] == 4242 assert calls["signals"][0][1] == signal.SIGTERM assert "stopped" in capsys.readouterr().out.lower() + # A terminated server (e.g. TerminateProcess on Windows) cannot clean up; + # stop does it, under the lock. + assert not pid_file.exists() + + +def test_clear_stale_state_files_never_touches_a_live_server(isolated_pid_files, other_holder): + pid_file, endpoint_file = isolated_pid_files + pid_file.write_text("4242") + endpoint_file.write_text("{}") + assert gsas_server._clear_stale_state_files() is False + assert pid_file.exists() and endpoint_file.exists() + + +def test_clear_stale_state_files_after_crash(isolated_pid_files): + pid_file, endpoint_file = isolated_pid_files + pid_file.write_text("4242") + endpoint_file.write_text("{}") + assert gsas_server._clear_stale_state_files() is True + assert not pid_file.exists() and not endpoint_file.exists() + assert gsas_server.is_server_running() is False # lock released again # The escalation signal stop_server() uses when the graceful SIGTERM window @@ -185,8 +213,8 @@ def test_stop_server_force_kill_escalation(isolated_pid_files, monkeypatch, caps def fake_kill(pid, sig): calls["signals"].append((pid, sig)) - # Alive until the second (force) signal has been sent, then dead. - monkeypatch.setattr(gsas_server, "_pid_alive", lambda pid: len(calls["signals"]) < 2) + # Running until the second (force) signal has been sent, then gone. + monkeypatch.setattr(gsas_server, "_lock_held", lambda: len(calls["signals"]) < 2) monkeypatch.setattr(gsas_server.os, "kill", fake_kill) monkeypatch.setattr(gsas_server.time, "sleep", lambda s: None) @@ -207,7 +235,7 @@ def test_stop_server_force_kill_never_dies_returns_false(isolated_pid_files, mon def fake_kill(pid, sig): calls["signals"].append((pid, sig)) - monkeypatch.setattr(gsas_server, "_pid_alive", lambda pid: True) + monkeypatch.setattr(gsas_server, "_lock_held", lambda: True) monkeypatch.setattr(gsas_server.os, "kill", fake_kill) monkeypatch.setattr(gsas_server.time, "sleep", lambda s: None) @@ -215,3 +243,470 @@ def fake_kill(pid, sig): assert result is False assert calls["signals"] == [(4242, signal.SIGTERM), (4242, _FORCE_SIGNAL)] assert "failed to stop" in capsys.readouterr().out.lower() + + +# --- POWDERLINE_NO_SERVER --- + +@pytest.mark.parametrize("value", ["1", "true", "TRUE", "yes", " Yes ", "on", "ON"]) +def test_server_disabled_truthy(monkeypatch, value): + monkeypatch.setenv("POWDERLINE_NO_SERVER", value) + assert gsas_server.server_disabled() is True + + +@pytest.mark.parametrize("value", [None, "", "0", "false", "no", "off"]) +def test_server_disabled_falsy(monkeypatch, value): + if value is None: + monkeypatch.delenv("POWDERLINE_NO_SERVER", raising=False) + else: + monkeypatch.setenv("POWDERLINE_NO_SERVER", value) + assert gsas_server.server_disabled() is False + + +@pytest.mark.parametrize("action", ["start", "restart"]) +def test_main_start_refuses_when_disabled(monkeypatch, capsys, action): + monkeypatch.setenv("POWDERLINE_NO_SERVER", "1") + monkeypatch.setattr(sys, "argv", ["gsas_server.py", action]) + monkeypatch.setattr(gsas_server, "GSASServer", + lambda *a, **k: pytest.fail("server must not be created")) + monkeypatch.setattr(gsas_server, "stop_server", + lambda: pytest.fail("restart must not stop the server first")) + with pytest.raises(SystemExit) as exc: + gsas_server.main() + assert exc.value.code == 1 + assert "POWDERLINE_NO_SERVER" in capsys.readouterr().out + + +# --- Per-user state directory --- + +@posix_only +def test_state_dir_uses_private_xdg_runtime_dir(monkeypatch, tmp_path): + runtime = tmp_path / "runtime" + runtime.mkdir(mode=0o700) + monkeypatch.setenv("XDG_RUNTIME_DIR", str(runtime)) + path = gsas_server.state_dir() + assert path == runtime / "powderline" + assert (path.stat().st_mode & 0o777) == 0o700 + + +@posix_only +def test_state_dir_falls_back_to_per_uid_tempdir(monkeypatch, tmp_path): + monkeypatch.delenv("XDG_RUNTIME_DIR", raising=False) + monkeypatch.setattr(gsas_server.tempfile, "gettempdir", lambda: str(tmp_path)) + path = gsas_server.state_dir() + assert path == tmp_path / f"powderline-{os.getuid()}" + assert (path.stat().st_mode & 0o777) == 0o700 + + +@posix_only +def test_state_dir_ignores_shared_xdg_runtime_dir(monkeypatch, tmp_path): + # An XDG_RUNTIME_DIR others can write to (e.g. inherited after su) is not used. + shared = tmp_path / "shared" + shared.mkdir() + shared.chmod(0o777) + monkeypatch.setenv("XDG_RUNTIME_DIR", str(shared)) + monkeypatch.setattr(gsas_server.tempfile, "gettempdir", lambda: str(tmp_path)) + assert gsas_server.state_dir() == tmp_path / f"powderline-{os.getuid()}" + + +@posix_only +def test_state_dir_path_is_distinct_per_uid(monkeypatch, tmp_path): + monkeypatch.delenv("XDG_RUNTIME_DIR", raising=False) + monkeypatch.setattr(gsas_server.tempfile, "gettempdir", lambda: str(tmp_path)) + paths = set() + for uid in (1001, 1002): + monkeypatch.setattr(gsas_server.os, "getuid", lambda uid=uid: uid) + paths.add(gsas_server._state_dir_path()) + assert paths == {tmp_path / "powderline-1001", tmp_path / "powderline-1002"} + + +@posix_only +def test_state_dir_refuses_dir_owned_by_another_user(monkeypatch, tmp_path): + # The dir exists (created by "someone else"): we are not its owner. + monkeypatch.delenv("XDG_RUNTIME_DIR", raising=False) + monkeypatch.setattr(gsas_server.tempfile, "gettempdir", lambda: str(tmp_path)) + other_uid = os.getuid() + 1 + (tmp_path / f"powderline-{other_uid}").mkdir(mode=0o700) + monkeypatch.setattr(gsas_server.os, "getuid", lambda: other_uid) + # Isolate the directory-owner check from the parent check. + monkeypatch.setattr(gsas_server, "_check_tempdir_parent", lambda path: None) + with pytest.raises(gsas_server.ServerStateError, match="not by you"): + gsas_server.state_dir() + + +@posix_only +def test_state_dir_refuses_symlink(monkeypatch, tmp_path): + monkeypatch.delenv("XDG_RUNTIME_DIR", raising=False) + monkeypatch.setattr(gsas_server.tempfile, "gettempdir", lambda: str(tmp_path)) + target = tmp_path / "elsewhere" + target.mkdir(mode=0o700) + (tmp_path / f"powderline-{os.getuid()}").symlink_to(target) + with pytest.raises(gsas_server.ServerStateError, match="symlink"): + gsas_server.state_dir() + + +@posix_only +@pytest.mark.parametrize("mode", [0o750, 0o705, 0o777]) +def test_state_dir_refuses_group_or_other_access(monkeypatch, tmp_path, mode): + monkeypatch.delenv("XDG_RUNTIME_DIR", raising=False) + monkeypatch.setattr(gsas_server.tempfile, "gettempdir", lambda: str(tmp_path)) + path = tmp_path / f"powderline-{os.getuid()}" + path.mkdir() + path.chmod(mode) + with pytest.raises(gsas_server.ServerStateError, match="accessible to other users"): + gsas_server.state_dir() + + +@pytest.mark.skipif(os.name != "nt", reason="Windows state-dir location") +def test_state_dir_windows_uses_localappdata(monkeypatch, tmp_path): + monkeypatch.setenv("LOCALAPPDATA", str(tmp_path)) + assert gsas_server.state_dir() == tmp_path / "powderline" + + +@posix_only +def test_state_dir_ignores_symlinked_xdg_runtime_dir(monkeypatch, tmp_path): + # A symlink (which its owner could retarget later) is never trusted, even + # if it currently points at a private directory of ours. + private = tmp_path / "private" + private.mkdir(mode=0o700) + link = tmp_path / "runtime-link" + link.symlink_to(private) + monkeypatch.setenv("XDG_RUNTIME_DIR", str(link)) + monkeypatch.setattr(gsas_server.tempfile, "gettempdir", lambda: str(tmp_path)) + assert gsas_server.state_dir() == tmp_path / f"powderline-{os.getuid()}" + + +@posix_only +def test_state_dir_refuses_shared_non_sticky_tempdir(monkeypatch, tmp_path): + # Others could rename/replace our directory in a world-writable parent + # that lacks the sticky bit. + shared = tmp_path / "shared" + shared.mkdir() + shared.chmod(0o777) + monkeypatch.delenv("XDG_RUNTIME_DIR", raising=False) + monkeypatch.setattr(gsas_server.tempfile, "gettempdir", lambda: str(shared)) + with pytest.raises(gsas_server.ServerStateError, match="sticky"): + gsas_server.state_dir() + + +@posix_only +def test_state_dir_accepts_sticky_shared_tempdir(monkeypatch, tmp_path): + shared = tmp_path / "tmp" + shared.mkdir() + shared.chmod(0o1777) # like /tmp + monkeypatch.delenv("XDG_RUNTIME_DIR", raising=False) + monkeypatch.setattr(gsas_server.tempfile, "gettempdir", lambda: str(shared)) + assert gsas_server.state_dir() == shared / f"powderline-{os.getuid()}" + + +@posix_only +def test_state_dir_refuses_tempdir_owned_by_another_user(monkeypatch, tmp_path): + # The owner of a parent directory can always rename entries in it. + monkeypatch.delenv("XDG_RUNTIME_DIR", raising=False) + monkeypatch.setattr(gsas_server.tempfile, "gettempdir", lambda: str(tmp_path)) + monkeypatch.setattr(gsas_server.os, "getuid", lambda: os.stat(tmp_path).st_uid + 1) + with pytest.raises(gsas_server.ServerStateError, match="is owned by uid"): + gsas_server.state_dir() + + +@posix_only +def test_state_files_never_opened_through_a_swapped_symlink(monkeypatch, tmp_path): + # Even if the directory is replaced by a symlink after state_dir() checked + # it, files are opened relative to an O_NOFOLLOW, fstat-verified handle. + elsewhere = tmp_path / "elsewhere" + elsewhere.mkdir(mode=0o700) + (elsewhere / "server.json").write_text('{"pid": 1, "port": 2, "token": "planted"}') + swapped = tmp_path / "state" + swapped.symlink_to(elsewhere) + monkeypatch.setattr(gsas_server, "state_dir", lambda: swapped) + with pytest.raises(gsas_server.ServerStateError): + gsas_server.read_endpoint() + + +@posix_only +def test_write_private_is_owner_only(isolated_pid_files): + _, target = isolated_pid_files + target.write_text("old") + target.chmod(0o644) + gsas_server._write_private(gsas_server.ENDPOINT_NAME, "new") + assert target.read_text() == "new" + assert (target.stat().st_mode & 0o777) == 0o600 + + +def test_read_endpoint_absent_or_garbage_is_none(isolated_pid_files): + _, endpoint_file = isolated_pid_files + assert gsas_server.read_endpoint() is None + endpoint_file.write_text("{not json") + assert gsas_server.read_endpoint() is None + endpoint_file.write_text('{"pid": 1, "port": 2}') # no token + assert gsas_server.read_endpoint() is None + endpoint_file.write_text('{"pid": 1, "port": 2, "token": "t"}') + assert gsas_server.read_endpoint() == {"pid": 1, "port": 2, "token": "t"} + + +# --- Start lock / state-file lifecycle --- + +def test_claim_server_lock_when_free(isolated_pid_files): + pid_file, _ = isolated_pid_files + assert gsas_server._claim_server_lock(attempts=1) is True + assert pid_file.read_text() == str(os.getpid()) + assert gsas_server.is_server_running() is True + + +def test_claim_server_lock_refused_while_held(isolated_pid_files, other_holder): + pid_file, _ = isolated_pid_files + pid_file.write_text("4242") + assert gsas_server._claim_server_lock(attempts=2, delay=0) is False + assert pid_file.read_text() == "4242" + + +def test_claim_server_lock_replaces_stale_files(isolated_pid_files): + # A crashed server left its files behind; the lock is free, so they are stale. + pid_file, endpoint_file = isolated_pid_files + pid_file.write_text("4242") + endpoint_file.write_text('{"pid": 4242, "port": 1, "token": "old"}') + assert gsas_server._claim_server_lock(attempts=1) is True + assert pid_file.read_text() == str(os.getpid()) + assert not endpoint_file.exists() + + +def test_release_state_files_only_when_holding_the_lock(isolated_pid_files): + pid_file, endpoint_file = isolated_pid_files + pid_file.write_text("4242") + endpoint_file.write_text("{}") + gsas_server._release_state_files() # not the holder: touches nothing + assert pid_file.exists() and endpoint_file.exists() + + assert gsas_server._claim_server_lock(attempts=1) is True + endpoint_file.write_text("{}") + gsas_server._release_state_files() + assert not pid_file.exists() and not endpoint_file.exists() + assert gsas_server.is_server_running() is False + + +_CLAIM_SCRIPT = r""" +import importlib.util, sys, time +spec = importlib.util.spec_from_file_location("gs", sys.argv[1]) +gs = importlib.util.module_from_spec(spec); spec.loader.exec_module(gs) +while time.time() < float(sys.argv[2]): + pass +won = gs._claim_server_lock(attempts=1) +print("WON" if won else "LOST", flush=True) +if won: + sys.stdin.read() # hold the lock until the test has every result +""" + + +def test_concurrent_starts_yield_exactly_one_holder(tmp_path): + # Many starters at once, with stale files from a crashed server present + # (the case that defeated a PID-file lock): exactly one may win. + state = tmp_path / "powderline" + state.mkdir(mode=0o700) + (state / "server.pid").write_text("999999") + (state / "server.json").write_text('{"pid": 999999, "port": 1, "token": "old"}') + env = {**os.environ, "XDG_RUNTIME_DIR": str(tmp_path), "LOCALAPPDATA": str(tmp_path)} + n = 8 + go = time.time() + 2.0 # start together; a late starter still finds the lock held + procs = [subprocess.Popen( + [sys.executable, "-c", _CLAIM_SCRIPT, gsas_server.__file__, str(go)], + env=env, stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True) + for _ in range(n)] + try: + results = [p.stdout.readline().strip() for p in procs] + finally: + for p in procs: + p.stdin.close() + p.wait(timeout=60) + assert sorted(results) == ["LOST"] * (n - 1) + ["WON"], results + + +# --- Signed requests on the ASGI app --- + +TOKEN = "s3cret-token" + + +def _signed_headers(method, path, body=b"", *, token=TOKEN, nonce=None, ts=None): + nonce = nonce or secrets.token_hex(16) + ts = ts if ts is not None else str(int(time.time())) + return nonce, { + gsas_server.NONCE_HEADER: nonce, + gsas_server.TIMESTAMP_HEADER: ts, + gsas_server.SIGNATURE_HEADER: gsas_server.sign_request( + token, method, path, b"", nonce, ts, body), + "content-type": "application/json", + } + + +def _response_signed(resp, nonce, token=TOKEN): + expected = gsas_server.sign_response(token, nonce, resp.status_code, resp.content) + return resp.headers.get(gsas_server.RESPONSE_SIGNATURE_HEADER) == expected + + +@pytest.fixture +def app_client(): + """TestClient for the real app, without importing GSAS-II or binding a port.""" + pytest.importorskip("fastapi") + pytest.importorskip("httpx") + import logging + from fastapi.testclient import TestClient + + server = gsas_server.GSASServer.__new__(gsas_server.GSASServer) + server.request_count = 0 + server.start_time = None + server.token = TOKEN + server.logger = logging.getLogger("powderline.server.test") + calls = [] + server._run_simulation = lambda req: calls.append(req) or { + "success": False, "error": "stub", "method": "server"} + return TestClient(server.create_app()), calls + + +def _job_body(out): + return json.dumps({"recipe_data": {}, "output_dir": str(out)}).encode() + + +@pytest.mark.parametrize("case", [ + "unsigned", "bearer_token", "wrong_token", "stale_timestamp", "future_timestamp", + "tampered_body", "wrong_path", "non_hex_nonce", "short_nonce", "non_ascii_signature", +]) +def test_unauthenticated_requests_rejected(app_client, case, tmp_path): + client, calls = app_client + out = tmp_path / "out" + body = _job_body(out) + if case == "unsigned": + headers = {} + elif case == "bearer_token": + headers = {"Authorization": f"Bearer {TOKEN}"} # the token alone is not enough + elif case == "wrong_token": + _, headers = _signed_headers("POST", "/simulate", body, token="other-token") + elif case == "stale_timestamp": + _, headers = _signed_headers("POST", "/simulate", body, ts=str(int(time.time()) - 3600)) + elif case == "future_timestamp": + _, headers = _signed_headers("POST", "/simulate", body, ts=str(int(time.time()) + 3600)) + elif case == "tampered_body": + _, headers = _signed_headers("POST", "/simulate", body) + body = _job_body(tmp_path / "elsewhere") + elif case == "wrong_path": + _, headers = _signed_headers("POST", "/health", body) + elif case == "non_hex_nonce": + _, headers = _signed_headers("POST", "/simulate", body, nonce="z" * 32) + elif case == "short_nonce": + _, headers = _signed_headers("POST", "/simulate", body, nonce="ab") + else: + _, headers = _signed_headers("POST", "/simulate", body) + headers[gsas_server.SIGNATURE_HEADER] = "s\u00e9".encode("latin-1") + resp = client.post("/simulate", headers=headers, content=body) + assert resp.status_code == 401 + assert gsas_server.RESPONSE_SIGNATURE_HEADER not in resp.headers + assert calls == [], "no job may run without a valid signature" + assert not out.exists() + + +def test_signed_requests_accepted_and_responses_signed(app_client, tmp_path): + client, calls = app_client + body = _job_body(tmp_path) + nonce, headers = _signed_headers("POST", "/simulate", body) + resp = client.post("/simulate", headers=headers, content=body) + assert resp.status_code == 200 + assert len(calls) == 1 + assert _response_signed(resp, nonce) + assert not _response_signed(resp, nonce, token="other-token") + + nonce, headers = _signed_headers("GET", "/health") + resp = client.get("/health", headers=headers) + assert resp.status_code == 200 and resp.json()["status"] == "ok" + assert _response_signed(resp, nonce) + + +def test_replayed_request_rejected(app_client, tmp_path): + client, calls = app_client + body = _job_body(tmp_path) + _, headers = _signed_headers("POST", "/simulate", body) + assert client.post("/simulate", headers=headers, content=body).status_code == 200 + assert client.post("/simulate", headers=headers, content=body).status_code == 401 + assert len(calls) == 1 + + +def test_concurrent_replays_run_at_most_once(): + """Two copies of one signed request arriving together: the nonce is + re-checked after the body is read, so only one reaches the app.""" + import asyncio + + calls = [] + + async def app(scope, receive, send): + calls.append(scope["path"]) + await send({"type": "http.response.start", "status": 200, "headers": []}) + await send({"type": "http.response.body", "body": b"{}"}) + + middleware = gsas_server.SignedRequestMiddleware(app, TOKEN) + body = b'{"recipe_data": {}, "output_dir": "x"}' + _, headers = _signed_headers("POST", "/simulate", body) + scope = {"type": "http", "method": "POST", "path": "/simulate", "query_string": b"", + "headers": [(k.lower().encode(), v.encode()) for k, v in headers.items()]} + + async def main(): + body_ready = asyncio.Event() + statuses = [] + + async def receive(): + await body_ready.wait() # both copies are suspended here together + return {"type": "http.request", "body": body, "more_body": False} + + def sender(): + async def send(message): + if message["type"] == "http.response.start": + statuses.append(message["status"]) + return send + + tasks = [asyncio.create_task(middleware(dict(scope), receive, sender())) + for _ in range(2)] + await asyncio.sleep(0.05) + body_ready.set() + await asyncio.gather(*tasks) + return statuses + + statuses = asyncio.run(main()) + assert sorted(statuses) == [200, 401] + assert calls == ["/simulate"] + + +@pytest.mark.parametrize("method, path, body, status", [ + ("GET", "/nope", b"", 404), + ("PUT", "/simulate", b"", 405), + ("POST", "/simulate", b"not json", 422), + ("GET", "/docs", b"", 404), + ("GET", "/redoc", b"", 404), + ("GET", "/openapi.json", b"", 404), +]) +def test_every_route_and_error_needs_a_signature(app_client, method, path, body, status): + client, calls = app_client + assert client.request(method, path, content=body).status_code == 401 + nonce, headers = _signed_headers(method, path, body) + resp = client.request(method, path, headers=headers, content=body) + assert resp.status_code == status + assert _response_signed(resp, nonce) # error responses are signed too + assert calls == [] + + +def test_oversized_request_refused_unread(app_client, monkeypatch): + client, calls = app_client + monkeypatch.setattr(gsas_server, "MAX_BODY_BYTES", 16) + body = b"x" * 64 + _, headers = _signed_headers("POST", "/simulate", body) + assert client.post("/simulate", headers=headers, content=body).status_code == 413 + assert calls == [] + + +@posix_only +def test_state_dir_uncreatable_raises_state_error(monkeypatch, tmp_path): + # e.g. an unwritable/missing tempdir: callers must get ServerStateError, not OSError. + monkeypatch.delenv("XDG_RUNTIME_DIR", raising=False) + monkeypatch.setattr(gsas_server.tempfile, "gettempdir", + lambda: str(tmp_path / "missing" / "deeper")) + with pytest.raises(gsas_server.ServerStateError, match="Cannot use"): + gsas_server.state_dir() + readonly = tmp_path / "readonly" + readonly.mkdir(mode=0o500) + monkeypatch.setattr(gsas_server.tempfile, "gettempdir", lambda: str(readonly)) + with pytest.raises(gsas_server.ServerStateError, match="Cannot create"): + gsas_server.state_dir()