diff --git a/.env.example b/.env.example index 7be5648..e58f1a7 100644 --- a/.env.example +++ b/.env.example @@ -151,7 +151,17 @@ ENGRAPHIS_LLM_API_KEY=sk-your-key-here # ENGRAPHIS_CLOUD_COMPUTE_URL=https://compute.engraphis.com # ENGRAPHIS_CLOUD_ORGANIZATION_ID=org_replace_me -# Prefer the owner-only ~/.engraphis/cloud_session.json created during hosted onboarding. +# Prefer the owner-only ~/.engraphis/cloud_session.json written by `engraphis connect`: +# +# engraphis connect --token engr_ct_... # the command your account portal shows +# printf %s "$TOKEN" | engraphis connect --token - # keep the token out of shell history +# +# That is the supported way to connect a client. It redeems the one-time connect token, saves +# the rotating refresh credential with 0600 permissions, and keeps it rotated afterwards, so +# none of the variables below are needed on an interactive machine. Set +# ENGRAPHIS_CLOUD_COMPUTE_URL (or pass --compute-url) only if your account portal shows a +# compute endpoint different from the default. See docs/AGENT_CONNECT.md. +# # For non-interactive deployments a refresh credential may be injected as a bootstrap secret. # It rotates on use; the owner-only saved replacement takes precedence afterward, even while # the environment variable remains set. Never commit it. Bind environment-only bootstrap diff --git a/CHANGELOG.md b/CHANGELOG.md index 3902eb7..c2de406 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,32 @@ All notable changes to Engraphis are documented here. Format loosely follows ## [Unreleased] +### Added + +- **`engraphis connect --token engr_ct_…`** — the missing client half of device connect. + `cloud_session.save_bootstrap()` is the only writer of `~/.engraphis/cloud_session.json`, + and it had no production caller: the docs told paying customers to prefer a file nothing + created, so a purchased installation could not be connected without hand-writing state. + The new command redeems the one-time connect token from the account portal against + `POST /v1/devices/connect`, saves the returned session with owner-only permissions, and + verifies `cloud_session.configured()` before reporting success. The token is sent in the + request body and nowhere else — never printed, logged, or written to disk — and every + refusal maps to fixed, actionable copy (an expired or already-used token is not confused + with a lapsed subscription). Session storage is pre-flighted before the exchange, so an + unwritable state directory or a `cloud_session.json` replaced by a link fails the command + *without* spending the single-use token — the customer fixes the path and retries with the + same token instead of returning to the portal for a new one. Faults that can only happen + *after* the exchange — a reply truncated mid-body (`http.client.IncompleteRead`), or an + endpoint that stops resolving before the session is written (`CloudUrlUnresolved`) — are + reported as errors that say the token was already used, rather than escaping as tracebacks + that leave the customer unable to tell whether to retry. Also installed as + `engraphis-connect`. +- An `engraphis` front-door command that dispatches to the existing `engraphis-` + entry points, so the command the account portal displays is runnable as shown. +- A stable per-installation identity at `~/.engraphis/client_identity.json` (random ULIDs, + not a hardware fingerprint) so reconnecting a machine updates its existing installation + instead of registering a new device every time. + ### Removed - **"Signed compliance export" is no longer advertised on any surface.** It was granted to diff --git a/docs/AGENT_CONNECT.md b/docs/AGENT_CONNECT.md index 67e2b37..4ea833f 100644 --- a/docs/AGENT_CONNECT.md +++ b/docs/AGENT_CONNECT.md @@ -25,8 +25,9 @@ organization: 1. The organization owner starts Team or purchases a subscription in Engraphis Cloud. 2. The owner invites named members and assigns roles in the hosted dashboard. -3. A member accepts the invitation and creates a scoped agent/device credential. -4. The member configures their agent with the hosted URL and the one-time credential. +3. A member accepts the invitation and generates a one-time **connect token** (`engr_ct_…`). + The account portal shows the exact command to run. +4. The member runs that command on the machine being connected (see below). 5. The hosted service rechecks organization membership, role, scopes, entitlement version, and workspace binding on every request. @@ -37,6 +38,53 @@ The hosted onboarding flow provides the exact endpoint and client snippet for th organization. Do not substitute the URL of a public self-hosted image: that image intentionally has no Team identity backend. +## Connect a machine: `engraphis connect` + +Copy the command from your account portal and run it on the machine you are connecting: + +```bash +engraphis connect --token engr_ct_... +``` + +That redeems the token against `POST /v1/devices/connect` on the control plane and writes the +owner-only session file `~/.engraphis/cloud_session.json` (mode `0600`). The dashboard, the MCP +server, and Cloud Sync all read that file, so no environment secret is needed afterwards. Rerun +`engraphis connect` with a fresh token on every machine you want connected. + +Useful options: + +| Option | Effect | +| --- | --- | +| `--token -` | Read the token from stdin, so it never enters shell history. | +| `--workspace WS_ID` | Bind this device to a single workspace. | +| `--label TEXT` | Name this installation in your account portal. | +| `--device-name TEXT` | Override the device name (defaults to the hostname). | +| `--control-url URL` | Point at a non-default control plane. | +| `--compute-url URL` | Set the managed compute endpoint (also `ENGRAPHIS_CLOUD_COMPUTE_URL`). | +| `--json` | Print a redacted, machine-readable summary. | + +The same command is installed as `engraphis-connect`, matching the other `engraphis-*` scripts. + +Connect tokens are **single-use and short-lived**. The service answers every refusal — expired, +already redeemed, or never valid — with the same `401`, so the client reports all three +possibilities and the fix is always the same: generate a new token in the account portal. A `402` +means the subscription itself has lapsed; fix billing rather than the token. + +Because the token is single-use, the client checks that it can actually write the session file +*before* redeeming it. If the state directory is not writable, or `cloud_session.json` has been +replaced by a symlink, a hard link, or a directory, the command fails immediately, names the path +to fix, and sends nothing — your token is untouched, so you can correct the path and rerun the +same command rather than issuing a new token. + +The token is a credential. It is sent in the request body and nowhere else — it is never +printed, never logged, and never written to disk. What *is* written is the rotating refresh +credential the service returns, which is why the session file is owner-only. + +The command also mints a stable per-installation identity at `~/.engraphis/client_identity.json` +(random ULIDs, not a hardware fingerprint) so reconnecting the same machine updates the existing +installation instead of registering a new device every time. Both files honour +`ENGRAPHIS_STATE_DIR`; a distinct state directory is a distinct installation. + ## Credential lifecycle Hosted access uses short-lived access tokens plus rotating refresh credentials. A refresh family @@ -45,7 +93,10 @@ service; the raw replacement is returned once and must be kept in an owner-only or secrets manager. Customer-side environment variables are documented in [`.env.example`](../.env.example). Prefer -the onboarding-created `~/.engraphis/cloud_session.json` over long-lived environment secrets. +the `~/.engraphis/cloud_session.json` that `engraphis connect --token` writes over long-lived +environment secrets: it holds a rotating credential, it is owner-only, and it is the path the +client keeps up to date on its own. Environment secrets are for non-interactive deployments that +cannot run the connect command. ## Trial and grace diff --git a/engraphis/cloud_session.py b/engraphis/cloud_session.py index b539afe..50a0850 100644 --- a/engraphis/cloud_session.py +++ b/engraphis/cloud_session.py @@ -6,9 +6,11 @@ """ from __future__ import annotations +import http.client import json import os import stat +import tempfile import threading import time import urllib.error @@ -80,6 +82,15 @@ def _session_path() -> Path: return base / "cloud_session.json" +def _refresh_lock_path() -> Path: + return _session_path().with_name(".cloud_session.refresh.lock") + + +#: Exactly the flags ``_refresh_lock`` opens the lock with, so ``preflight_save`` cannot +#: drift from the access the real save needs. +_LOCK_OPEN_FLAGS = os.O_RDWR | getattr(os, "O_BINARY", 0) | getattr(os, "O_NOFOLLOW", 0) + + @contextmanager def _refresh_lock(): """Serialize spend-and-rotate of the single-use refresh credential. @@ -89,11 +100,11 @@ def _refresh_lock(): so every process coordinates on one stable filesystem object. """ with _REFRESH_THREAD_LOCK: - lock_path = _session_path().with_name(".cloud_session.refresh.lock") + lock_path = _refresh_lock_path() try: lock_path.parent.mkdir(parents=True, exist_ok=True) expected = private_file_stat(lock_path, allow_missing=True) - flags = os.O_RDWR | getattr(os, "O_BINARY", 0) | getattr(os, "O_NOFOLLOW", 0) + flags = _LOCK_OPEN_FLAGS if expected is None: try: descriptor = os.open( @@ -212,6 +223,98 @@ def _save(value: dict) -> None: atomic_private_text(path, json.dumps(value, sort_keys=True, separators=(",", ":"))) +def preflight_save() -> Path: + """Verify the session file can be saved, without saving anything. Returns its path. + + :func:`save_bootstrap` runs *after* the control plane has answered, so a state + directory that lost its permissions, or a ``cloud_session.json`` that is a symlink or + a hard link, was only discovered once a single-use connect token had already been + consumed: the customer was left with a spent token, no session, and a trip back to + the portal for a new one. Any caller about to spend a one-shot credential must run + this first, so a storage fault costs nothing. + + The checks are the ones the write path itself applies -- :func:`private_file_stat` on + the session leaf and on its refresh lock, plus the randomized sibling temp file + :func:`atomic_private_text` has to create -- so the preflight cannot drift from what + the real save will accept. It deliberately never opens, creates or replaces the + session leaf: an existing session survives a failed preflight untouched, and a first + connect is not turned into a half-written file. + """ + + path = _session_path() + try: + path.parent.mkdir(parents=True, exist_ok=True) + except OSError as exc: + raise CloudSessionError( + "The Engraphis state directory %s could not be created." % path.parent, + status=409, + ) from exc + for candidate in (path, _refresh_lock_path()): + try: + private_file_stat(candidate, allow_missing=True) + except UnsafeStateFile as exc: + raise CloudSessionError( + "%s is not a plain private file -- a symlink, hard link or directory is " + "in its place. Remove it." % candidate, + status=409, + ) from exc + except OSError as exc: + raise CloudSessionError( + "%s could not be inspected; check its permissions." % candidate, + status=409, + ) from exc + # ``private_file_stat`` above is an ``lstat``: it proves the lock is a plain, private, + # unlinked file, not that this process can *open* it. ``_refresh_lock`` opens it + # ``O_RDWR``, so a lock left behind by another UID -- or one whose mode changed -- + # passes the stat and fails the save, after the token is spent. Open it here with the + # same flags when it already exists. When it does not, ``_refresh_lock`` creates it in + # a directory the probe below proves writable, so there is nothing to pre-check and + # nothing is created here. + lock_path = _refresh_lock_path() + if lock_path.exists(): + try: + descriptor = os.open(str(lock_path), _LOCK_OPEN_FLAGS) + except OSError as exc: + raise CloudSessionError( + "The cloud session refresh lock %s exists but cannot be opened; check its " + "ownership and permissions." % lock_path, + status=409, + ) from exc + os.close(descriptor) + # Probe the directory the way ``atomic_private_text`` will, rather than touching + # ``cloud_session.json``: a read-only mount or a lost ACL fails here, while a valid + # existing session is never created, truncated or replaced. + try: + descriptor, probe = tempfile.mkstemp( + prefix=".%s.preflight." % path.name, dir=str(path.parent) + ) + except OSError as exc: + raise CloudSessionError( + "The Engraphis state directory %s is not writable, so the cloud session " + "cannot be saved there." % path.parent, + status=409, + ) from exc + os.close(descriptor) + try: + os.unlink(probe) + except OSError as exc: + # Not cosmetic, and not "the probe was just created so of course it can be + # removed": creating a file and removing one are separate rights. A directory ACL + # that grants add-file but denies delete lets ``mkstemp`` succeed and ``unlink`` + # fail, and ``atomic_private_text`` finishes with ``os.replace`` over the session + # leaf -- which needs exactly the delete/replace right that just failed. Swallowing + # this let the preflight pass on a directory the real save cannot use, redeeming the + # single-use token before failing, and left the probe behind on every attempt. That + # is precisely the drift the preflight exists to prevent. + raise CloudSessionError( + "The Engraphis state directory %s does not allow files to be replaced, so the " + "cloud session cannot be saved there. Check its permissions; a leftover %s may " + "need removing." % (path.parent, Path(probe).name), + status=409, + ) from exc + return path + + #: Plans that carry no paid cloud access, used only to default an absent activity flag. _UNPAID_PLANS = ("free", "local") #: Bounds on the entitlement fields before they are written into the session record. The @@ -337,12 +440,30 @@ def record_billing_denial() -> bool: return False +def text_field(response: dict, key: str) -> str: + """Return ``response[key]`` when it is a string, else ``""``. Never a ``repr``. + + ``str(response.get(key) or "")`` looks like a coercion but is not a validation: JSON + arrays and objects arrive as Python ``list``/``dict``, and ``str()`` renders their + *repr*, which is both truthy and non-empty. A control-plane reply carrying + ``"refresh_credential": ["tok"]`` was therefore stored as the literal text ``['tok']``; + ``configured()`` read that back as a usable session and connect reported success, while + the next refresh submitted the junk and failed -- after the single-use connect token had + already been spent, so the customer could not simply retry. + + Provider bodies are untrusted, so a field that must be a string is required to be one. + """ + + value = response.get(key) + return value.strip() if isinstance(value, str) else "" + + def save_bootstrap(response: dict, *, control_url: str, compute_url: Optional[str] = None) -> None: """Persist the one-time bootstrap/refresh material returned by the control plane.""" - refresh = str(response.get("refresh_credential") or "").strip() - organization_id = str(response.get("organization_id") or "").strip() + refresh = text_field(response, "refresh_credential") + organization_id = text_field(response, "organization_id") if not refresh or not organization_id: raise CloudSessionError("Cloud bootstrap did not return a refresh credential.") value = { @@ -350,11 +471,11 @@ def save_bootstrap(response: dict, *, control_url: str, "control_url": validate_cloud_base_url(control_url), "compute_url": validate_cloud_base_url(compute_url) if compute_url else "", "organization_id": organization_id, - "installation_id": str(response.get("installation_id") or ""), - "device_id": str(response.get("device_id") or ""), - "member_id": str(response.get("member_id") or ""), + "installation_id": text_field(response, "installation_id"), + "device_id": text_field(response, "device_id"), + "member_id": text_field(response, "member_id"), "refresh_credential": refresh, - "refresh_expires_at": str(response.get("refresh_expires_at") or ""), + "refresh_expires_at": text_field(response, "refresh_expires_at"), "token_subject": _validated_token_subject( response.get("token_subject") or "member" ), @@ -398,6 +519,11 @@ def _refresh_http_error(status: int) -> CloudSessionError: return CloudSessionError("Engraphis Cloud could not refresh this session.") +#: What a best-effort drain of an error body is allowed to fail with. See the comment in +#: :func:`_post_refresh`; ``engraphis.device_connect`` guards its drain with the same tuple. +_DRAIN_FAILURES = (OSError, ValueError, http.client.HTTPException) + + def _post_refresh(control_url: str, refresh: str, workspace_id: Optional[str], token_subject: str) -> dict: # An org-scoped entitlement read asks for an unbound token, so it passes no workspace. @@ -417,37 +543,110 @@ def _post_refresh(control_url: str, refresh: str, workspace_id: Optional[str], }, method="POST", ) + # Split for the same reason as ``device_connect.post_connect``, and with sharper + # consequences here. Once ``open`` returns, a success status line has been parsed, so + # the control plane processed the refresh and the single-use credential it was given is + # spent -- but the rotated replacement only reaches disk after the body parses, below. try: - with build_pinned_https_opener(_NoRedirect()).open( - request, timeout=10.0 - ) as response: - raw = response.read(_MAX_RESPONSE_BYTES + 1) + response = build_pinned_https_opener(_NoRedirect()).open(request, timeout=10.0) except urllib.error.HTTPError as exc: code = exc.code # Draining and closing the error body can itself time out or reset. A sibling # ``except`` clause of this ``try`` does NOT cover an exception raised inside this # handler, so an unguarded read escapes as an unhandled traceback whenever the # cloud is flaky -- exactly the launch-day condition this path exists for. + # + # ``HTTPException`` is named explicitly: a truncated chunked error body raises + # ``http.client.IncompleteRead``, whose MRO is ``(IncompleteRead, HTTPException, + # Exception, BaseException, object)`` -- neither an ``OSError`` nor a ``ValueError``, + # so the pair alone let it through. try: exc.read(_MAX_RESPONSE_BYTES + 1) - except (OSError, ValueError): + except _DRAIN_FAILURES: pass finally: try: exc.close() - except (OSError, ValueError): + except _DRAIN_FAILURES: pass raise _refresh_http_error(code) - except (urllib.error.URLError, TimeoutError, OSError) as exc: + # urllib wraps failures before or while sending the request in URLError. That is the one + # distinguishable pre-send path, so the credential was not spent and retry remains safe. + except urllib.error.URLError as exc: raise CloudSessionError("Engraphis Cloud is temporarily unreachable.") from exc + except (TimeoutError, http.client.RemoteDisconnected, OSError) as exc: + # These escape directly from getresponse() after urllib wrote the POST. The control + # plane may have spent the single-use credential even though no status line arrived; + # retrying the unchanged on-disk value risks revoking its entire credential family. + raise CloudSessionError( + "Engraphis Cloud did not complete this refresh response, so the rotated " + "credential could not be saved. Connect this installation again.", + status=409, + ) from exc + except (http.client.BadStatusLine, http.client.LineTooLong) as exc: + # ``getresponse()`` raises these only after urllib has sent the POST. The control + # plane may therefore have consumed the single-use refresh credential, while its + # replacement never reached disk. Retrying would replay the stale credential and can + # revoke its whole family, so prefer reconnecting over an unsafe transient retry. + # ``RemoteDisconnected`` is also a ``BadStatusLine`` but reaches the earlier OSError + # transport clause through its ``ConnectionResetError`` base. + raise CloudSessionError( + "Engraphis Cloud returned a malformed refresh response, so the rotated " + "credential could not be saved. Connect this installation again.", + status=409, + ) from exc + except http.client.HTTPException as exc: + # Other malformed HTTP replies have no useful protocol status, but unlike a malformed + # status line they do not establish that this request reached the control plane. + raise CloudSessionError("Engraphis Cloud is temporarily unreachable.") from exc + + try: + with response: + raw = response.read(_MAX_RESPONSE_BYTES + 1) + except (OSError, ValueError, http.client.HTTPException) as exc: + # Post-response, and therefore NOT a transient outage. The server answered, so the + # credential just submitted is spent, but the rotation it returned never reached + # ``_save`` -- the stale value is still on disk. ``_public_session_error`` maps 503 + # to ``transient=True``, which ``CloudFeatureClient.run_job`` acts on by retrying; + # that retry would resubmit the spent credential, and this module already documents + # (see ``_note_denied``) that the control plane treats replay by revoking the whole + # credential family. 409 is the existing "saved session is unusable; connect this + # installation again" bucket, which is the honest answer here. + # + # This deliberately prefers a false "reconnect" over a replay: if the truncation + # happened before the server committed the rotation the old credential was still + # good and the reconnect was unnecessary, but the opposite mistake revokes every + # credential in the family and forces the same reconnect anyway, from a worse state. + raise CloudSessionError( + "Engraphis Cloud answered this session refresh but the reply was incomplete, " + "so the rotated credential could not be saved. Connect this installation " + "again.", + status=409, + ) from exc + + # These are post-response too, so they carry the same replay hazard as the truncated + # body above and take the same non-transient status: the server consumed the credential + # it was given, and a body this client cannot parse means the rotation never landed. if len(raw) > _MAX_RESPONSE_BYTES: - raise CloudSessionError("Engraphis Cloud returned an oversized session response.") + raise CloudSessionError( + "Engraphis Cloud returned an oversized session response, so the rotated " + "credential could not be saved. Connect this installation again.", + status=409, + ) try: body = json.loads(raw.decode("utf-8")) except (UnicodeDecodeError, ValueError, RecursionError) as exc: - raise CloudSessionError("Engraphis Cloud returned an invalid session response.") from exc + raise CloudSessionError( + "Engraphis Cloud returned an invalid session response, so the rotated " + "credential could not be saved. Connect this installation again.", + status=409, + ) from exc if not isinstance(body, dict): - raise CloudSessionError("Engraphis Cloud returned an invalid session response.") + raise CloudSessionError( + "Engraphis Cloud returned an invalid session response, so the rotated " + "credential could not be saved. Connect this installation again.", + status=409, + ) return body @@ -519,13 +718,21 @@ def access_for_workspace( compute = _reachable_cloud_base_url(compute) if compute else "" token_subject = _token_subject(saved) body = _post_refresh(control, refresh, workspace_id, token_subject) - access = str(body.get("access_token") or "").strip() - organization_id = str( - body.get("organization_id") or saved.get("organization_id") or "" - ).strip() - rotated = str(body.get("refresh_credential") or "").strip() + # Same untrusted-provider boundary as ``save_bootstrap``: a non-string credential + # would otherwise be stored as its ``repr`` and submitted on the next refresh. + access = text_field(body, "access_token") + organization_id = ( + text_field(body, "organization_id") or text_field(saved, "organization_id") + ) + rotated = text_field(body, "refresh_credential") if not access or not organization_id or not rotated: - raise CloudSessionError("Engraphis Cloud returned incomplete session credentials.") + # Also post-response: the submitted credential is spent and no rotation was + # saved, so this must not be reported as a retryable outage either. + raise CloudSessionError( + "Engraphis Cloud returned incomplete session credentials, so the rotated " + "credential could not be saved. Connect this installation again.", + status=409, + ) response_subject = _validated_token_subject( body.get("token_subject") or token_subject ) @@ -536,7 +743,7 @@ def access_for_workspace( "compute_url": compute, "organization_id": organization_id, "refresh_credential": rotated, - "refresh_expires_at": str(body.get("refresh_expires_at") or ""), + "refresh_expires_at": text_field(body, "refresh_expires_at"), "token_subject": response_subject, }) # The refresh response carries the same entitlement fields as registration, so the diff --git a/engraphis/device_connect.py b/engraphis/device_connect.py new file mode 100644 index 0000000..0002787 --- /dev/null +++ b/engraphis/device_connect.py @@ -0,0 +1,800 @@ +"""Client half of the Engraphis Cloud device-connect flow. + +The account portal issues a one-time connect token (``engr_ct_...``) and shows the +customer a single command. This module exchanges that token for the durable session +material :func:`engraphis.cloud_session.save_bootstrap` persists, so +``~/.engraphis/cloud_session.json`` -- the file :doc:`AGENT_CONNECT` and ``.env.example`` +tell paying customers to prefer -- is finally created by something. Before this existed +``save_bootstrap`` had no production caller at all and a paid installation could not be +connected without hand-writing the state file. + +Design constraints, all of which have teeth: + +* **The connect token is a bearer credential.** It travels in the request body and + nowhere else: never in a log line, never in an exception message, never in the saved + session, never echoed back to the terminal. Callers get a redacted summary. +* **The vetted transport only.** Requests go through + :func:`engraphis.hosted_client.build_pinned_https_opener` with redirects blocked and an + explicit timeout, exactly as :mod:`engraphis.cloud_session` and + :mod:`engraphis.update_check` do. A bare ``urllib.request.build_opener`` would drop the + DNS-rebinding guard on a credential-bearing call. +* **Fixed copy keyed on status.** Provider bodies are untrusted -- they may carry + internal URLs -- so nothing from the response is reflected into an error message. The + control plane deliberately makes expired / consumed / invalid tokens indistinguishable + (all ``401``), so the client says all three at once rather than guessing. +""" +from __future__ import annotations + +import http.client +import json +import math +import os +import platform +import re +import socket +import urllib.error +import urllib.request +from pathlib import Path +from urllib.parse import urlsplit +from typing import Optional, Tuple + +from engraphis import cloud_session +from engraphis.hosted_client import ( + CloudUrlUnresolved, + build_pinned_https_opener, + upgrade_url, + validate_cloud_base_url, +) +from engraphis.private_state import ( + UnsafeStateFile, + atomic_private_text, + publish_private_text_if_absent, + read_private_text, +) + +try: # installed distribution -> real version; source tree -> harmless fallback + from engraphis import __version__ as CURRENT_VERSION +except Exception: # pragma: no cover - engraphis is always importable in practice + CURRENT_VERSION = "0" + +#: Path segment of the unauthenticated connect endpoint on the control plane. +CONNECT_PATH = "/v1/devices/connect" + +#: One interactive request. Longer than the 10s refresh budget because this runs at a +#: human's prompt, once, and a spurious timeout costs the customer a fresh token. +DEFAULT_TIMEOUT_SECONDS = 15.0 + +#: Upper bound on ``--timeout``. Not a policy number -- a platform one. ``socket`` +#: converts a timeout into an absolute deadline, so a large but *finite* value raises +#: ``OverflowError`` ("timeout doesn't fit into C timeval", "timestamp out of range for +#: platform time_t") from inside ``urllib``: ``1e9`` already overflows on CPython 3.12. +#: One hour is orders of magnitude more than a single interactive POST can need and orders +#: of magnitude below the first value any supported platform rejects. +_MAX_TIMEOUT_SECONDS = 3600.0 + +#: The control plane answers with a small fixed record; anything larger is not ours. +_MAX_RESPONSE_BYTES = 64 * 1024 + +#: Copy for a reply that began but never completed. Every other failure in this module can +#: promise the connect token is untouched; this one cannot, because the request reached a +#: control plane that started to answer, and a 200 consumes the token as it is written. +#: Saying "try again" here would be a lie that costs the customer a second failed attempt, +#: so the copy sends them to the one place that shows whether the device landed. +_TRUNCATED_REPLY = ( + "Engraphis Cloud started answering this connect request but the connection closed " + "before the reply was complete. Your connect token may already have been used: check " + "your account portal, and generate a new one if this device is not listed there." +) + +#: Appended to every failure that can only happen once the control plane has answered +#: successfully. ``urllib`` raises ``HTTPError`` for every status >= 400, and ``_NoRedirect`` +#: turns a 3xx into one too, so any fault past ``opener.open()`` follows a 2xx -- and a +#: successful connect consumes the single-use token as it is written. Telling the customer +#: to "try again" there would send them into a guaranteed 401 with no session; the only +#: action that can work is a fresh token from the portal. +_SPENT_TOKEN_SUFFIX = ( + " This connect token has now been used -- generate a new one in your Engraphis account " + "portal and try again, and contact support if it repeats." +) + +#: The portal always mints tokens with this prefix. Checking it locally turns a +#: mistyped or truncated paste into an instant, free error instead of a round trip that +#: consumes rate budget and returns the same opaque 401 as a genuinely dead token. +CONNECT_TOKEN_PREFIX = "engr_ct_" +#: Shortest credible token: the prefix plus enough entropy to be a real secret. +_MIN_TOKEN_CHARS = len(CONNECT_TOKEN_PREFIX) + 16 +_MAX_TOKEN_CHARS = 512 +#: ``secrets.token_urlsafe`` alphabet. Anything else is a paste accident (a shell- +#: mangled quote, a wrapped line, a whole command copied in). +_TOKEN_BODY = re.compile(r"\A[A-Za-z0-9_-]+\Z") + +#: Identity file for this installation. Kept beside ``cloud_session.json`` in the same +#: owner-only state directory and honouring the same ``ENGRAPHIS_STATE_DIR`` override. +_IDENTITY_FILENAME = "client_identity.json" +_IDENTITY_SCHEMA = "engraphis-client-identity/v1" + +#: Compute endpoint for the production deployment. ``DeviceRegistrationResponse`` does +#: not carry one, and ``cloud_session.configured()`` requires it, so a connect against the +#: shipped control plane would otherwise leave a customer "connected" but not configured. +#: Applied *only* when the control plane is the shipped one: a self-hosted or staging +#: control URL gets no guess, and the caller is told to supply the compute URL. +#: +#: This is the *fallback*, not the authority: ``default_compute_url`` prefers the +#: manifest's ``compute_plane`` whenever it carries one. The shipped +#: ``engraphis-commercial/v2`` manifest does not declare that key, so without a constant +#: here a production connect would resolve an empty compute URL and save an unconfigured +#: session. +DEFAULT_COMPUTE_URL = "https://compute.engraphis.com" + + +class DeviceConnectError(RuntimeError): + """A connect attempt failed with public, actionable copy. + + ``status`` mirrors the control-plane status where there was one, and uses ``400`` for + a request this client refused to send at all. + """ + + def __init__(self, message: str, *, status: int = 503) -> None: + super().__init__(message) + self.status = status + + +class _NoRedirect(urllib.request.HTTPRedirectHandler): + """Refuse redirects: a 3xx would replay the connect token at an unvetted host.""" + + def redirect_request(self, req, fp, code, msg, headers, newurl): + return None + + +def normalize_connect_token(value: object) -> str: + """Return the trimmed token, or raise before any network call happens. + + Never quotes the offending value. A rejected token is still a credential -- an + error message containing it would land in shell history, CI logs and bug reports. + """ + + token = str(value or "").strip() + if not token: + raise DeviceConnectError( + "No connect token was supplied. Copy the `engraphis connect --token ...` " + "command shown in your Engraphis account portal.", + status=400, + ) + if ( + not token.startswith(CONNECT_TOKEN_PREFIX) + or len(token) < _MIN_TOKEN_CHARS + or len(token) > _MAX_TOKEN_CHARS + or not _TOKEN_BODY.match(token[len(CONNECT_TOKEN_PREFIX):]) + ): + raise DeviceConnectError( + "That does not look like an Engraphis connect token (they start with " + "`%s`). Copy the whole command from your account portal -- a token split " + "across lines or missing its last characters will not work." + % CONNECT_TOKEN_PREFIX, + status=400, + ) + return token + + +def _state_dir() -> Path: + root = os.environ.get("ENGRAPHIS_STATE_DIR", "").strip() + return Path(root).expanduser() if root else Path.home() / ".engraphis" + + +def _identity_path() -> Path: + try: + return _state_dir() / _IDENTITY_FILENAME + except RuntimeError as exc: + # ``Path.home()`` raises ``RuntimeError("Could not determine home directory")`` on a + # service account or container with no resolvable home. This module promises every + # failure is a ``DeviceConnectError``, and the CLI catches only that, so the bare + # ``RuntimeError`` reached the terminal as a traceback -- from a command whose whole + # selling point is being non-interactive and scriptable. ``ENGRAPHIS_STATE_DIR`` is + # the documented answer, so the copy names it. + raise DeviceConnectError( + "Engraphis could not determine a home directory for its state files. Set " + "ENGRAPHIS_STATE_DIR to a writable directory and run `engraphis connect " + "--token ...` again.", + status=400, + ) from exc + + +def _new_client_id(prefix: str) -> str: + # The package's own ULID minter, so these ids sort chronologically and read the same + # way as every other prefixed id in the client. + from engraphis.core import ids + + return "%s_%s" % (prefix, ids.ulid()) + + +def _read_identity() -> dict: + try: + raw = read_private_text(_identity_path(), max_bytes=8 * 1024, allow_missing=True) + except DeviceConnectError: + # Already actionable copy (an unresolvable home directory). Must come first: + # ``DeviceConnectError`` is a ``RuntimeError``, so the broad clause below would + # otherwise swallow it and let the *next*, unguarded ``_identity_path()`` call + # raise the raw error instead. + raise + except UnsafeStateFile as exc: + raise DeviceConnectError( + "The saved client identity has unsafe filesystem permissions. Remove %s and " + "connect again." % _identity_path(), + status=409, + ) from exc + except (OSError, RuntimeError): + return {} + if not raw: + return {} + try: + value = json.loads(raw) + except (ValueError, RecursionError): + return {} + return value if isinstance(value, dict) else {} + + +def client_identity() -> Tuple[str, str]: + """Return ``(installation_client_id, device_client_id)`` for this machine. + + Minted once and persisted at ``/client_identity.json`` with owner-only + permissions, so reconnecting the same machine re-presents the same pair and the + control plane recognises the installation instead of accumulating a new phantom + device on every connect. + + They are random ULIDs rather than a hardware fingerprint on purpose: a MAC- or + hostname-derived id is a stable cross-account identifier the customer never agreed to + ship, and it changes under exactly the conditions (a new NIC, a rename) where + stability was the point. "Stable per installation" is what the server needs, and a + minted-once file in the state directory is precisely that. A second state directory + (``ENGRAPHIS_STATE_DIR``, a container, another user account) is a second installation, + which is the correct reading. + """ + + saved = _read_identity() + installation = str(saved.get("installation_client_id") or "").strip() + device = str(saved.get("device_client_id") or "").strip() + if installation and device: + return installation, device + + minted = { + "schema": _IDENTITY_SCHEMA, + "installation_client_id": installation or _new_client_id("inst"), + "device_client_id": device or _new_client_id("dev"), + } + payload = json.dumps(minted, sort_keys=True, separators=(",", ":")) + path = _identity_path() + try: + if not publish_private_text_if_absent(path, payload): + # Another process won the create, or a half-written file already existed. + # Re-read: the winner's ids are the ones the control plane will see. + existing = _read_identity() + installation = str(existing.get("installation_client_id") or "").strip() + device = str(existing.get("device_client_id") or "").strip() + if installation and device: + return installation, device + atomic_private_text(path, payload) + except UnsafeStateFile as exc: + raise DeviceConnectError( + "The client identity file at %s could not be written safely." % path, + status=409, + ) from exc + except OSError as exc: + raise DeviceConnectError( + "Could not write the client identity file at %s." % path, status=409 + ) from exc + return minted["installation_client_id"], minted["device_client_id"] + + +def default_control_url() -> str: + """Resolve the control plane: explicit env override, else the shipped manifest.""" + + configured = os.environ.get("ENGRAPHIS_CLOUD_CONTROL_URL", "").strip() + if configured: + return configured + try: + from engraphis.commercial import manifest + + return str(manifest().get("control_plane") or "").strip() + except Exception: # pragma: no cover - manifest ships with the package + return "" + + +def _canonical_endpoint(value: object) -> str: + """Canonical ``scheme://host:port/path`` for comparing two endpoint URLs. + + ``validate_cloud_base_url`` returns ``urlunsplit((scheme, parts.netloc, ...))``: it + lower-cases the *scheme* but passes the netloc through verbatim, and it never removes a + redundant default port. So ``HTTPS://API.ENGRAPHIS.COM`` normalises to + ``https://API.ENGRAPHIS.COM``, and ``https://api.engraphis.com:443`` keeps its ``:443`` + -- both are the shipped control plane, and both compared unequal to it as raw strings. + The consequence was silent and expensive: :func:`default_compute_url` returned ``""``, + so a customer who typed the production URL with a different case or an explicit port + connected successfully and then had every hosted feature switched off, because + ``cloud_session.configured()`` needs a compute endpoint. + + Returns ``""`` for anything unparseable, which never equals another canonical form. + """ + + try: + parts = urlsplit(str(value or "").strip()) + except ValueError: + return "" + scheme = parts.scheme.lower() + host = (parts.hostname or "").lower() + if not scheme or not host: + return "" + try: + port = parts.port + except ValueError: + return "" + if port is None: + port = 443 if scheme == "https" else 80 + return "%s://%s:%d%s" % (scheme, host, port, parts.path.rstrip("/")) + + +def default_compute_url(control_url: str) -> str: + """Resolve the compute plane, guessing only for the shipped control plane. + + The manifest is the authority for shipped endpoint metadata, so a ``compute_plane`` + key wins if one is ever published -- a manifest-only endpoint change then needs no + code change here. The current ``engraphis-commercial/v2`` manifest declares only + ``control_plane``, so :data:`DEFAULT_COMPUTE_URL` is what production actually + resolves; reading the absent key alone would yield ``""`` and save a session + ``cloud_session.configured()`` rejects. + """ + + configured = os.environ.get("ENGRAPHIS_CLOUD_COMPUTE_URL", "").strip() + if configured: + return configured + shipped = "" + shipped_compute = "" + try: + from engraphis.commercial import manifest + + data = manifest() + shipped = str(data.get("control_plane") or "").strip() + shipped_compute = str(data.get("compute_plane") or "").strip() + except Exception: # pragma: no cover - manifest ships with the package + return "" + if shipped and _canonical_endpoint(control_url) == _canonical_endpoint(shipped): + return shipped_compute or DEFAULT_COMPUTE_URL + return "" + + +def _validated_timeout(value: object) -> float: + """Reject a timeout the socket layer cannot use, before any request is started. + + ``float("nan")`` and ``float("inf")`` are values ``argparse``'s ``type=float`` + accepts happily, but they reach the pinned opener's deadline arithmetic and raise + ``ValueError``/``OverflowError`` from inside ``urllib`` -- neither of which + :func:`post_connect` catches. That breaks this module's contract that every failure + is a :class:`DeviceConnectError` with actionable copy, so ``--timeout nan`` printed a + traceback instead of an error and an exit code. Non-positive values are refused for + the same reason: a ``0`` or negative socket timeout is not a shorter wait, it is a + different (non-blocking) mode the caller did not ask for. + + ``math.isfinite`` is necessary but *not sufficient*: ``socket.settimeout`` turns the + value into an absolute deadline, so a merely large finite number + (``--timeout 10000000000``, and in fact anything from about ``1e9`` up) raises + ``OverflowError`` from the same uncaught place. Hence the explicit + :data:`_MAX_TIMEOUT_SECONDS` ceiling rather than a finiteness check alone. + """ + + try: + timeout = float(value) + except (TypeError, ValueError) as exc: + raise DeviceConnectError( + "The connect timeout must be a number of seconds.", status=400 + ) from exc + if not math.isfinite(timeout) or timeout <= 0: + raise DeviceConnectError( + "The connect timeout must be a positive, finite number of seconds.", + status=400, + ) + if timeout > _MAX_TIMEOUT_SECONDS: + raise DeviceConnectError( + "The connect timeout must be at most %d seconds." % int(_MAX_TIMEOUT_SECONDS), + status=400, + ) + return timeout + + +def _validated_control_url(value: str) -> str: + if not str(value or "").strip(): + raise DeviceConnectError( + "No Engraphis Cloud control URL is configured. Set " + "ENGRAPHIS_CLOUD_CONTROL_URL or pass --control-url.", + status=400, + ) + try: + return validate_cloud_base_url(value) + except CloudUrlUnresolved as exc: + # "Offline" is not "misconfigured": validate_cloud_base_url resolves the host, so + # a customer on a plane would otherwise be told their URL is invalid forever. + raise DeviceConnectError( + "Engraphis Cloud is temporarily unreachable. Check your network and try " + "again.", + status=503, + ) from exc + except ValueError as exc: + raise DeviceConnectError( + "The Engraphis Cloud control URL is not a valid HTTPS endpoint.", status=400 + ) from exc + + +def _default_device_name() -> str: + try: + name = socket.gethostname().strip() + except OSError: + name = "" + return name[:100] + + +def _default_platform() -> str: + try: + return ("%s %s" % (platform.system(), platform.machine())).strip()[:100] + except Exception: # pragma: no cover - platform is stdlib and total + return "" + + +def _connect_http_error(status: int) -> DeviceConnectError: + """Map a control-plane status to fixed, actionable copy. + + Only the status is used. ``401`` deliberately covers expired, already-consumed and + never-valid tokens with one indistinguishable answer, so the copy names all three + instead of asserting one -- the fix is the same in every case. + """ + + if status == 401: + return DeviceConnectError( + "That connect token has expired, was already used, or is not valid. " + "Generate a new one in your Engraphis account portal and run " + "`engraphis connect --token ...` again.", + status=401, + ) + if status == 402: + return DeviceConnectError( + "This Engraphis Cloud subscription is not active, so no new device can be " + "connected. Update billing at %s and try again." % upgrade_url(), + status=402, + ) + if status == 403: + return DeviceConnectError( + "Engraphis Cloud refused this connect request. Check with the organization " + "owner that your account may still add devices.", + status=403, + ) + if status == 404: + return DeviceConnectError( + "This Engraphis Cloud control plane has no device-connect endpoint. Check " + "ENGRAPHIS_CLOUD_CONTROL_URL points at the URL shown in your account portal.", + status=404, + ) + if status == 422: + return DeviceConnectError( + "Engraphis Cloud rejected this connect request as malformed. Upgrade the " + "client (`pip install -U engraphis`) and try again.", + status=422, + ) + if status == 429: + return DeviceConnectError( + "Too many connect attempts. Wait a minute and try again.", status=429 + ) + if status == 503: + return DeviceConnectError( + "Engraphis Cloud is not accepting new device activations right now. Try " + "again shortly; your connect token is unaffected.", + status=503, + ) + return DeviceConnectError( + "Engraphis Cloud could not connect this device. Try again shortly.", status=503 + ) + + +def post_connect(control_url: str, token: str, *, installation_client_id: str, + device_client_id: str, installation_label: Optional[str] = None, + device_name: Optional[str] = None, app_platform: Optional[str] = None, + app_version: Optional[str] = None, workspace_id: Optional[str] = None, + timeout: float = DEFAULT_TIMEOUT_SECONDS) -> dict: + """POST the connect token and return the ``DeviceRegistrationResponse`` body. + + *control_url* must already be validated. The endpoint rejects unknown fields with a + ``422``, so optional values are omitted rather than sent empty, and there is + deliberately no ``organization_id``: the token carries the organization. + """ + + timeout = _validated_timeout(timeout) + body = { + "connect_token": token, + "installation_client_id": installation_client_id, + "device_client_id": device_client_id, + } + for key, value in ( + ("installation_label", installation_label), + ("device_name", device_name), + ("platform", app_platform), + ("app_version", app_version), + ("workspace_id", workspace_id), + ): + cleaned = str(value or "").strip() + if cleaned: + body[key] = cleaned + + payload = json.dumps(body, sort_keys=True, separators=(",", ":")).encode("utf-8") + request = urllib.request.Request( + control_url + CONNECT_PATH, + data=payload, + headers={ + "Accept": "application/json", + "Content-Type": "application/json", + "User-Agent": "Engraphis/%s device-connect" % CURRENT_VERSION, + }, + method="POST", + ) + # The open and the body read are deliberately separate ``try`` blocks. They look like + # one operation but they sit on opposite sides of the point of no return: once ``open`` + # returns, urllib has already parsed a success status line, so the control plane + # answered and the single-use token is spent. A ``TimeoutError`` or + # ``ConnectionResetError`` is an ``OSError`` in both phases, and while they shared one + # ``try`` the body-read case inherited the connection-phase copy and told the customer + # to retry a token that was already gone. + try: + response = build_pinned_https_opener(_NoRedirect()).open( + request, timeout=timeout + ) # nosec B310 - scheme validated by validate_cloud_base_url + except urllib.error.HTTPError as exc: + status = exc.code + # Draining the error body can itself raise, and this runs *inside* an ``except`` + # block, so the sibling ``except http.client.HTTPException`` clause below cannot + # cover it -- an unguarded read escapes as a raw traceback exactly when the cloud + # is flaky, replacing the status copy the customer needs. + # + # ``HTTPException`` has to be named explicitly: a truncated chunked error body + # raises ``http.client.IncompleteRead``, whose MRO is ``(IncompleteRead, + # HTTPException, Exception, BaseException, object)`` -- it is neither an ``OSError`` + # nor a ``ValueError``, so an ``(OSError, ValueError)`` guard let it straight + # through. ``tests/test_device_connect.py`` pins that MRO so the mistake cannot + # come back. Same shape as cloud_session._post_refresh. + _DRAIN_FAILURES = (OSError, ValueError, http.client.HTTPException) + try: + exc.read(_MAX_RESPONSE_BYTES + 1) + except _DRAIN_FAILURES: + pass + finally: + try: + exc.close() + except _DRAIN_FAILURES: + pass + raise _connect_http_error(status) + except urllib.error.URLError as exc: + # ``exc`` may quote an internal host or a proxy URL; never reflect it. + # + # ``URLError`` is the honest "nothing was consumed" case, and the only one. + # ``AbstractHTTPHandler.do_open`` wraps failures from establishing the connection + # and from ``h.request(...)`` -- DNS, connection refused, TLS, a write that never + # completed -- in ``URLError``, but lets anything raised by ``h.getresponse()`` + # propagate unwrapped. So reaching *this* clause means the request never finished + # going out, and "try again" with the same token is correct. + raise DeviceConnectError( + "Engraphis Cloud is temporarily unreachable. Check your network and try " + "again.", + status=503, + ) from exc + except (TimeoutError, OSError, http.client.HTTPException) as exc: + # Everything else in this phase comes out of ``h.getresponse()``, which runs only + # after the request has been written in full. The control plane may therefore have + # received and processed it -- and a processed connect spends the token -- so this + # is ambiguous, not a clean miss. + # + # This clause used to claim the opposite for ``RemoteDisconnected`` ("the peer + # closed without answering at all. Nothing was consumed"). That was an overclaim: + # ``RemoteDisconnected`` is raised when ``getresponse()`` reads zero bytes for the + # status line, which is *after* the POST went out. Telling that customer to retry + # sent them into a 401 that reads as "the token I just generated is invalid". + # ``_TRUNCATED_REPLY`` is the right copy under ambiguity: it asks them to check the + # portal first rather than asserting either outcome. + # + # ``LineTooLong``/``BadStatusLine`` from a mangled status line, and ``IncompleteRead`` + # from a truncated body, land here for the same reason -- none of them are a + # ``URLError``, so before this clause existed they escaped as a raw traceback. + raise DeviceConnectError(_TRUNCATED_REPLY, status=502) from exc + + # Past this line the control plane has answered with a success status, so the token is + # spent no matter what goes wrong next. ``OSError`` covers a socket that times out or + # resets mid-body, ``HTTPException`` a truncated chunked body, ``ValueError`` a read + # from an already-closed response; all three mean the same thing to the customer. + try: + with response: + raw = response.read(_MAX_RESPONSE_BYTES + 1) + except (OSError, ValueError, http.client.HTTPException) as exc: + raise DeviceConnectError(_TRUNCATED_REPLY, status=502) from exc + + # Everything below here runs only after a 2xx, so the token is already spent -- see + # ``_SPENT_TOKEN_SUFFIX``. A bare "invalid response" would leave the customer retrying + # a consumed token forever. + if len(raw) > _MAX_RESPONSE_BYTES: + raise DeviceConnectError( + "Engraphis Cloud returned an oversized connect response, so no session was " + "saved." + _SPENT_TOKEN_SUFFIX, + status=502, + ) + try: + parsed = json.loads(raw.decode("utf-8")) + except (UnicodeDecodeError, ValueError, RecursionError) as exc: + raise DeviceConnectError( + "Engraphis Cloud returned an invalid connect response, so no session was " + "saved." + _SPENT_TOKEN_SUFFIX, + status=502, + ) from exc + if not isinstance(parsed, dict): + raise DeviceConnectError( + "Engraphis Cloud returned an invalid connect response, so no session was " + "saved." + _SPENT_TOKEN_SUFFIX, + status=502, + ) + return parsed + + +#: Fields worth echoing back to the customer. Deliberately excludes +#: ``refresh_credential`` and ``access_token``: the summary is printed. +_SUMMARY_FIELDS = ( + "organization_id", + "installation_id", + "device_id", + "member_id", + "workspace_id", + "token_subject", + "plan", + "cloud_access_active", + "cloud_features", + "entitlement_version", + "expires_in_seconds", +) + + +def summarize(response: dict) -> dict: + """Return the non-secret fields of a registration response, for display.""" + + summary = {} + for key in _SUMMARY_FIELDS: + if key in response: + summary[key] = response[key] + return summary + + +def _preflight_session_storage() -> Path: + """Refuse a connect *before* the token is spent when the session cannot be saved. + + The exchange is the point of no return: the control plane consumes the single-use + connect token as it answers, so any storage fault discovered afterwards costs the + customer a fresh token from the portal. Delegated to :mod:`engraphis.cloud_session` + because that module owns the paths, the lock and the atomic write this is checking -- + a private copy of those rules here would drift from the save it is meant to predict. + """ + + try: + return cloud_session.preflight_save() + except cloud_session.CloudSessionError as exc: + raise DeviceConnectError( + "%s Your connect token has not been used, so you can fix this and run " + "`engraphis connect --token ...` again with the same token." % exc, + status=getattr(exc, "status", 409), + ) from exc + + +def connect(token: object, *, control_url: Optional[str] = None, + compute_url: Optional[str] = None, workspace_id: Optional[str] = None, + installation_label: Optional[str] = None, device_name: Optional[str] = None, + timeout: float = DEFAULT_TIMEOUT_SECONDS) -> dict: + """Exchange a connect token for a saved cloud session. + + Returns the redacted summary -- it is safe to print. Raises + :class:`DeviceConnectError` for every failure, with copy the customer can act on and + never containing the token. Nothing is written unless the exchange succeeded. + """ + + # Argument checks first: a bad ``--timeout`` must be reported as a bad timeout, not + # masked by whatever the identity or storage pre-flight happens to hit on the way to + # the same rejection inside ``post_connect``. + timeout = _validated_timeout(timeout) + normalized = normalize_connect_token(token) + resolved_control = _validated_control_url( + control_url if control_url is not None else default_control_url() + ) + resolved_compute = ( + compute_url if compute_url is not None else default_compute_url(resolved_control) + ) + if resolved_compute: + try: + resolved_compute = validate_cloud_base_url(resolved_compute) + except CloudUrlUnresolved as exc: + raise DeviceConnectError( + "The Engraphis Cloud compute endpoint is temporarily unreachable.", + status=503, + ) from exc + except ValueError as exc: + raise DeviceConnectError( + "The Engraphis Cloud compute URL is not a valid HTTPS endpoint.", + status=400, + ) from exc + + installation_client_id, device_client_id = client_identity() + # Last check before the point of no return. ``client_identity`` may have written its + # file minutes or months ago, so a writable state directory then is no evidence of one + # now; prove the session can land *before* the POST spends the token, not after. + session_path = _preflight_session_storage() + response = post_connect( + resolved_control, + normalized, + installation_client_id=installation_client_id, + device_client_id=device_client_id, + installation_label=installation_label, + device_name=device_name if device_name is not None else _default_device_name(), + app_platform=_default_platform(), + app_version=str(CURRENT_VERSION), + workspace_id=workspace_id, + timeout=timeout, + ) + # ``text_field`` and not ``str(... or "")``: a JSON array or object arrives as a Python + # ``list``/``dict`` whose ``repr`` is truthy and non-empty, so the coercion accepted a + # credential that is not a credential, wrote it, and reported a connection that could + # never refresh. Checked with the same helper the writer uses so the two cannot + # disagree about what counts as present. + if not cloud_session.text_field(response, "refresh_credential"): + # Reaching here means a 200 was parsed, and the control plane consumes the + # single-use connect token as it writes one. So "try again" would be actively + # wrong: re-running the same command deterministically returns 401 and still leaves + # no session. Point at the portal, the same way the truncated-reply copy does. + raise DeviceConnectError( + "Engraphis Cloud accepted the token but returned no session credential, so " + "no session was saved." + _SPENT_TOKEN_SUFFIX, + status=502, + ) + try: + cloud_session.save_bootstrap( + response, control_url=resolved_control, compute_url=resolved_compute or None + ) + except cloud_session.CloudSessionError as exc: + # Also post-redemption. The pre-flight proved this path writable moments ago, so + # arriving here is a race -- typically the refresh lock disappearing or turning + # unsafe, which ``cloud_session`` wraps in a ``CloudSessionError``, which is exactly + # why it does not reach the ``OSError`` clause below that carries the spent-token + # warning. Forwarding the bare lock message left the customer retrying a consumed + # token. ``str(exc)`` is this package's own fixed copy, never provider text. + raise DeviceConnectError( + "Engraphis Cloud accepted the token, but the session could not be saved: " + "%s%s" % (exc, _SPENT_TOKEN_SUFFIX), + status=getattr(exc, "status", 503), + ) from exc + except OSError as exc: + # The pre-flight cleared this exact path moments ago, so arriving here means the + # state directory changed underneath the exchange. ``UnsafeStateFile`` is an + # ``OSError`` and not a ``CloudSessionError``, so without this clause it escapes + # as a raw traceback at the worst possible moment -- the token is already spent, + # and the customer needs to be told that plainly rather than shown a stack. + raise DeviceConnectError( + "Engraphis Cloud accepted the token, but the session could not be written to " + "%s. Fix that path, then connect again with a new token from your account " + "portal -- this one has been used." % session_path, + status=409, + ) from exc + except ValueError as exc: + # ``save_bootstrap`` re-runs ``validate_cloud_base_url`` on both endpoints, and + # that helper *resolves* the host. A resolver that dies between the pre-POST check + # and this line raises ``CloudUrlUnresolved``; an endpoint that starts resolving to + # a rejected address raises a bare ``ValueError``. Both are ``ValueError``, so + # neither the ``CloudSessionError`` nor the ``OSError`` clause above covered them + # and they escaped as a traceback -- again after the token was already spent. + raise DeviceConnectError( + "Engraphis Cloud accepted the token, but its endpoints could not be verified " + "in time to save the session, so nothing was written. Check your network, " + "then connect again with a new token from your account portal -- this one " + "has been used.", + status=409, + ) from exc + + summary = summarize(response) + summary["control_url"] = resolved_control + summary["compute_url"] = resolved_compute + summary["session_path"] = str(session_path) + return summary diff --git a/pyproject.toml b/pyproject.toml index da56949..5af2743 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -164,6 +164,11 @@ Repository = "https://github.com/Coding-Dev-Tools/engraphis" Issues = "https://github.com/Coding-Dev-Tools/engraphis/issues" [project.scripts] +# The front door. The account portal tells customers to run `engraphis connect --token ...`, +# so the bare verb form has to exist; scripts.entry dispatches to the same main() each +# engraphis- script below calls. +engraphis = "scripts.entry:main" +engraphis-connect = "scripts.connect:main" engraphis-server = "scripts.start_server:main" engraphis-cli = "scripts.cli:main" engraphis-mcp = "engraphis.mcp_cli:main" diff --git a/scripts/connect.py b/scripts/connect.py new file mode 100644 index 0000000..6652b0e --- /dev/null +++ b/scripts/connect.py @@ -0,0 +1,167 @@ +"""engraphis connect - redeem the connect token from your account portal. + +Your Engraphis account portal shows a one-time token and the exact command to run: + + engraphis connect --token engr_ct_... + +That exchanges the token with the control plane and writes the owner-only session file +``~/.engraphis/cloud_session.json`` that the dashboard, MCP server, and Cloud Sync all +read. Until this command existed the file had no writer at all, so a paying customer had +no supported way to connect a client. + + engraphis connect --token engr_ct_... # the normal case + engraphis connect --token - # read the token from stdin + engraphis connect --token ... --workspace ws_1 # bind the device to one workspace + engraphis connect --token ... --label "CI runner" + engraphis connect --token ... --json # redacted machine-readable summary + +The token is a credential. It is sent in the request body and nowhere else: it is never +printed, never logged, and never written to disk. Passing it as an argument does put it +in your shell history, so ``--token -`` is available for scripted and shared machines. + +Non-interactive by design (no prompts): safe in scripts, CI, and agent shells. +""" +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + +from engraphis import cloud_session +from engraphis.device_connect import ( + DEFAULT_TIMEOUT_SECONDS, + DeviceConnectError, + connect, +) + + +def _prog(argv_zero: str) -> str: + """Name this command the way the caller actually invoked it.""" + + stem = Path(argv_zero or "").stem + return stem if stem.endswith("connect") and stem != "connect" else "engraphis connect" + + +def _read_token(value: str) -> str: + """Resolve ``--token -`` to one line of stdin, keeping it off the command line.""" + + if value != "-": + return value + if sys.stdin is None or sys.stdin.isatty(): + # A bare `--token -` on a terminal would silently hang waiting for input. + raise DeviceConnectError( + "--token - reads the token from stdin; pipe it in, for example " + "`printf %s \"$ENGRAPHIS_CONNECT_TOKEN\" | engraphis connect --token -`.", + status=400, + ) + return sys.stdin.readline() + + +def _print_summary(summary: dict) -> None: + rows = [ + ("organization", summary.get("organization_id")), + ("installation", summary.get("installation_id")), + ("device", summary.get("device_id")), + ("member", summary.get("member_id")), + ("workspace", summary.get("workspace_id")), + ("subject", summary.get("token_subject")), + ] + plan = str(summary.get("plan") or "").strip() + if plan: + active = summary.get("cloud_access_active") + rows.append(("plan", plan + ("" if active is None else + " (active)" if active else " (inactive)"))) + features = summary.get("cloud_features") + if isinstance(features, (list, tuple)) and features: + rows.append(("features", ", ".join(str(item) for item in features))) + rows.append(("control url", summary.get("control_url"))) + rows.append(("compute url", summary.get("compute_url") or "(not configured)")) + rows.append(("session file", summary.get("session_path"))) + + print("Connected this device to Engraphis Cloud.") + print() + for label, value in rows: + text = str(value or "").strip() + if text: + print(" %-14s %s" % (label, text)) + + +def main(argv=None) -> int: + ap = argparse.ArgumentParser( + prog=_prog(sys.argv[0] if sys.argv else ""), + description=__doc__, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + ap.add_argument("--token", required=True, metavar="TOKEN", + help="the connect token from your account portal, or - for stdin") + ap.add_argument("--control-url", default=None, metavar="URL", + help="control plane to connect to (default: the shipped endpoint, " + "or ENGRAPHIS_CLOUD_CONTROL_URL)") + ap.add_argument("--compute-url", default=None, metavar="URL", + help="managed compute endpoint (default: ENGRAPHIS_CLOUD_COMPUTE_URL, " + "or the shipped endpoint)") + ap.add_argument("--workspace", default=None, metavar="WORKSPACE_ID", + help="bind this device to a single workspace") + ap.add_argument("--label", default=None, metavar="TEXT", + help="label for this installation in your account portal") + ap.add_argument("--device-name", default=None, metavar="TEXT", + help="device name shown in your account portal (default: hostname)") + ap.add_argument("--timeout", type=float, default=DEFAULT_TIMEOUT_SECONDS, + metavar="SECONDS", help="network timeout (default: %(default)s)") + ap.add_argument("--json", action="store_true", + help="print the redacted summary as JSON instead of a report") + args = ap.parse_args(argv) + + try: + summary = connect( + _read_token(args.token), + control_url=args.control_url, + compute_url=args.compute_url, + workspace_id=args.workspace, + installation_label=args.label, + device_name=args.device_name, + timeout=args.timeout, + ) + except DeviceConnectError as exc: + # ``str(exc)`` is fixed public copy from device_connect; it never carries the + # token, the response body, or an internal hostname. + print("%s: %s" % (ap.prog, exc), file=sys.stderr) + return 1 + + # Prove the write actually produced a session the rest of the client will use, rather + # than reporting success on a file nothing can load. + # + # This runs *before* any success output is emitted, and that ordering is load-bearing. + # ``configured()`` reads the session back and can raise -- an invalid + # ``ENGRAPHIS_CLOUD_TOKEN_SUBJECT``, or the file changing under the read. Printing + # first meant a complete ``--json`` success object was already on stdout when the + # command then wrote an error and exited 1, so a consumer parsing stdout accepted a + # connect that had failed, and a human got two contradictory answers. + try: + usable = cloud_session.configured() + except cloud_session.CloudSessionError as exc: + print("%s: the session was written but is not usable: %s" % (ap.prog, exc), + file=sys.stderr) + return 1 + + if args.json: + print(json.dumps(summary, sort_keys=True, indent=2)) + else: + _print_summary(summary) + print() + if usable: + print("Next steps:") + print(" engraphis-dashboard # your plan's features are now unlocked") + print(" engraphis-init --check # verify the installation") + else: + print("Note: hosted compute is not configured yet, so managed compute " + "features stay off.") + print(" Set ENGRAPHIS_CLOUD_COMPUTE_URL (or rerun with --compute-url) " + "using the endpoint") + print(" shown in your account portal. Everything else is connected.") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/entry.py b/scripts/entry.py new file mode 100644 index 0000000..c486055 --- /dev/null +++ b/scripts/entry.py @@ -0,0 +1,89 @@ +"""engraphis - the single front door, so the portal's copy-paste command actually runs. + +Every capability already ships as its own ``engraphis-`` console script; this adds +the spelling customers are *shown*. The account portal hands out + + engraphis connect --token engr_ct_... + +and until there was an ``engraphis`` executable that string was not a runnable command. + +This is a dispatcher, not a second CLI: it rewrites ``sys.argv`` and calls the same +``main()`` the matching ``engraphis-`` script calls, so a verb behaves identically +either way and there is exactly one implementation of each command. The import is lazy +so ``engraphis connect`` never drags in the memory/embedding stack that ``engraphis cli`` +needs. +""" +from __future__ import annotations + +import sys +from importlib import import_module + +#: verb -> "module:function", mirroring ``[project.scripts]`` with the prefix dropped. +COMMANDS = { + "connect": "scripts.connect:main", + "init": "scripts.init:main", + "cli": "scripts.cli:main", + "mcp": "engraphis.mcp_cli:main", + "server": "scripts.start_server:main", + "dashboard": "scripts.start_dashboard:main", + "inspector": "scripts.inspector:main", + "consolidate": "scripts.consolidate:main", + "graph": "scripts.graph_cli:main", + "graph-server": "scripts.graph_server:main", + "update": "scripts.update:main", +} + +_USAGE = """usage: engraphis [options] + +commands: + connect redeem the connect token from your account portal + init write a project .env and print agent setup snippets + cli store and recall memories from the terminal + mcp run the MCP server (Claude Code, Cursor, Cline, Zed) + server run the REST server + dashboard run the product dashboard + inspector inspect the local database + consolidate run consolidation over stored memories + graph query the knowledge graph + graph-server run the graph server + update check for and install a newer Engraphis release + +Run `engraphis --help` for a command's options. +Every command is also installed as `engraphis-`.""" + + +def main(argv=None) -> int: + args = list(sys.argv[1:] if argv is None else argv) + if not args or args[0] in {"-h", "--help", "help"}: + print(_USAGE) + return 0 if args else 2 + if args[0] in {"-V", "--version"}: + from engraphis import __version__ + + print(__version__) + return 0 + + verb = args[0] + target = COMMANDS.get(verb) + if target is None: + print("engraphis: unknown command %r\n" % verb, file=sys.stderr) + print(_USAGE, file=sys.stderr) + return 2 + + module_name, _, attribute = target.partition(":") + command = getattr(import_module(module_name), attribute) + + # Hand the verb its own argv. Some targets take ``main(argv)`` and some read + # ``sys.argv`` directly, so rewriting is the one approach that works for all of + # them -- and it also puts "engraphis connect" in that command's --help/usage. + saved = sys.argv + sys.argv = ["engraphis %s" % verb] + args[1:] + try: + result = command() + finally: + sys.argv = saved + return 0 if result is None else int(result) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_cloud_session.py b/tests/test_cloud_session.py index f5ac02e..4f46407 100644 --- a/tests/test_cloud_session.py +++ b/tests/test_cloud_session.py @@ -1,5 +1,6 @@ from __future__ import annotations +import http.client from io import BytesIO import threading from concurrent.futures import ThreadPoolExecutor @@ -282,6 +283,154 @@ def open(self, request, timeout): assert caught.value.status == status +@pytest.mark.parametrize("failure", [ + http.client.LineTooLong("header line"), + http.client.BadStatusLine("garbage"), +]) +def test_a_mangled_refresh_status_line_requires_reconnect(monkeypatch, failure) -> None: + """The request was sent, so an unparsable reply may hide a spent credential. + + These ``HTTPException`` variants escape from ``getresponse()`` after urllib has sent the + refresh POST. A control plane or proxy may have consumed the one-time credential before + the malformed reply prevented its rotation from reaching disk; classifying that as 503 + would retry the stale credential and can revoke its entire family. + """ + + class _Opener: + def open(self, request, timeout): + raise failure + + monkeypatch.setattr( + cloud_session.urllib.request, "build_opener", lambda *handlers: _Opener() + ) + with pytest.raises(cloud_session.CloudSessionError) as caught: + cloud_session._post_refresh( + "https://control.example.test", "refresh", "ws", "member" + ) + + assert caught.value.status == 409 + assert "Connect this installation again" in str(caught.value) + + +@pytest.mark.parametrize("failure", [ + http.client.IncompleteRead(b'{"access_'), + TimeoutError("read timed out"), + ConnectionResetError("reset mid-body"), +]) +def test_a_truncated_refresh_body_is_not_reported_as_a_retryable_outage( + monkeypatch, failure +) -> None: + """A post-response failure must not invite a retry that replays a spent credential. + + Once the status line parses, the control plane consumed the submitted credential, but + the rotation it returned only reaches disk after the body parses -- so the stale value + is still there. ``_public_session_error`` maps 503 to ``transient=True``, which + ``CloudFeatureClient.run_job`` acts on by retrying, and this module documents that the + control plane answers a replayed credential by revoking the whole family. 409 is the + existing non-transient "connect this installation again" bucket. + """ + + class _Truncated: + def __enter__(self): + return self + + def __exit__(self, *exc) -> bool: + return False + + def read(self, size: int = -1) -> bytes: + raise failure + + class _Opener: + def open(self, request, timeout): + return _Truncated() + + monkeypatch.setattr( + cloud_session.urllib.request, "build_opener", lambda *handlers: _Opener() + ) + with pytest.raises(cloud_session.CloudSessionError) as caught: + cloud_session._post_refresh( + "https://control.example.test", "refresh", "ws", "member" + ) + + assert caught.value.status == 409 + assert "Connect this installation again" in str(caught.value) + + # The status must survive translation to the public error as non-transient, or the + # retry this whole change exists to prevent happens anyway. + from engraphis.cloud_features import _public_session_error + + _message, transient = _public_session_error(caught.value.status) + assert transient is False + + +@pytest.mark.parametrize("payload", [ + b"{not json", + b'"a string"', + b"[]", +]) +def test_an_unparseable_refresh_body_is_also_non_transient(monkeypatch, payload) -> None: + """Same hazard as a truncated body: answered, credential spent, rotation not saved.""" + + class _Response: + def __enter__(self): + return self + + def __exit__(self, *exc) -> bool: + return False + + def read(self, size: int = -1) -> bytes: + return payload + + class _Opener: + def open(self, request, timeout): + return _Response() + + monkeypatch.setattr( + cloud_session.urllib.request, "build_opener", lambda *handlers: _Opener() + ) + with pytest.raises(cloud_session.CloudSessionError) as caught: + cloud_session._post_refresh( + "https://control.example.test", "refresh", "ws", "member" + ) + + assert caught.value.status == 409 + assert "invalid session response" in str(caught.value) + + +def test_a_truncated_refresh_error_body_still_reports_the_status(monkeypatch) -> None: + """The drain runs inside the ``except HTTPError`` block, so it needs its own guard. + + An exception raised there cannot reach the sibling ``except HTTPException`` clause of + the same ``try``, so an ``(OSError, ValueError)`` guard let ``IncompleteRead`` replace + the 401 copy with a traceback. + """ + + error = urllib.error.HTTPError( + "https://control.example.test/v1/tokens/refresh", + 401, "denied", {}, BytesIO(b'{"detail":"private"}'), + ) + + def _boom(*args, **kwargs): + raise http.client.IncompleteRead(b'{"detail":"pri') + + error.read = _boom + error.close = _boom + + class _Opener: + def open(self, request, timeout): + raise error + + monkeypatch.setattr( + cloud_session.urllib.request, "build_opener", lambda *handlers: _Opener() + ) + with pytest.raises(cloud_session.CloudSessionError) as caught: + cloud_session._post_refresh( + "https://control.example.test", "refresh", "ws", "member" + ) + + assert caught.value.status == 401 + + def test_refresh_network_error_is_service_unavailable(monkeypatch) -> None: class _Opener: def open(self, request, timeout): diff --git a/tests/test_device_connect.py b/tests/test_device_connect.py new file mode 100644 index 0000000..849fc6b --- /dev/null +++ b/tests/test_device_connect.py @@ -0,0 +1,1503 @@ +"""Tests for the client half of the device-connect flow. + +``cloud_session.save_bootstrap`` had no production caller: nothing in the shipped client +ever created ``~/.engraphis/cloud_session.json``, while the docs told customers to prefer +it. These tests pin the command that closes that gap, and in particular that the connect +token -- a bearer credential -- never escapes the request body. +""" +from __future__ import annotations + +import http.client +import json +import os +import socket +import urllib.error +import urllib.request +from io import BytesIO +from pathlib import Path + +import pytest + +from engraphis import cloud_session, device_connect +from engraphis.hosted_client import CloudUrlUnresolved +from engraphis.private_state import UnsafeStateFile +from scripts import connect as connect_cli + +CONTROL_URL = "https://control.example.test" +COMPUTE_URL = "https://compute.example.test" +TOKEN = "engr_ct_R3plAceMeWithRealEntropy_0123456789" + +#: Every key the server's ``DeviceRegistrationResponse`` carries, fed to the client +#: verbatim the way a real 200 would be. +REGISTRATION = { + "organization_id": "org_alpha", + "installation_id": "instl_alpha", + "device_id": "devc_alpha", + "member_id": "mem_alpha", + "workspace_id": "ws_alpha", + "access_token": "short-lived-access-token", + "token_type": "Bearer", + "expires_in_seconds": 900, + "refresh_credential": "rotating-refresh-credential", + "refresh_expires_at": "2026-08-21T00:00:00Z", + "token_subject": "device", + "entitlement_version": 7, + "plan": "team", + "cloud_access_active": True, + # No ``export``: the signed compliance export was never implemented and is no longer + # in either repo's plan->feature table, so a fixture claiming the cloud grants it + # would re-establish exactly the drift that removal cleaned up. + "cloud_features": ["analytics", "automation", "sync", "team"], +} + + +class _Response: + """Minimal stand-in for the ``http.client`` response the opener yields.""" + + def __init__(self, payload: bytes) -> None: + self._payload = payload + + def __enter__(self): + return self + + def __exit__(self, *exc) -> bool: + return False + + def read(self, size: int = -1) -> bytes: + return self._payload + + +class _Opener: + """Fake of the pinned opener, recording exactly what was put on the wire.""" + + def __init__(self, *, body=None, error=None) -> None: + self._body = body + self._error = error + self.calls = [] + + def open(self, request, timeout): + self.calls.append({ + "url": request.full_url, + "method": request.get_method(), + "headers": dict(request.headers), + "body": request.data, + "timeout": timeout, + }) + if self._error is not None: + raise self._error + return _Response(json.dumps(self._body).encode("utf-8")) + + +def _install_opener(monkeypatch, opener) -> None: + # ``build_pinned_https_opener`` is a thin wrapper over ``build_opener``; patching the + # stdlib factory keeps the module's own call site (and its redirect handler) intact. + monkeypatch.setattr(urllib.request, "build_opener", lambda *handlers: opener) + + +def _http_error(status: int, body: bytes = b'{"detail":"private"}'): + return urllib.error.HTTPError( + CONTROL_URL + device_connect.CONNECT_PATH, status, "denied", {}, BytesIO(body) + ) + + +@pytest.fixture(autouse=True) +def _isolated_client_state(monkeypatch, tmp_path): + """A private state directory and a control plane that never needs DNS.""" + + monkeypatch.setenv("ENGRAPHIS_STATE_DIR", str(tmp_path)) + for name in ( + "ENGRAPHIS_CLOUD_CONTROL_URL", + "ENGRAPHIS_CLOUD_COMPUTE_URL", + "ENGRAPHIS_CLOUD_ACCESS_TOKEN", + "ENGRAPHIS_CLOUD_ORGANIZATION_ID", + "ENGRAPHIS_CLOUD_REFRESH_CREDENTIAL", + "ENGRAPHIS_CLOUD_TOKEN_SUBJECT", + ): + monkeypatch.delenv(name, raising=False) + # ``validate_cloud_base_url`` resolves the hostname, so the offline gate cannot use + # the real one against ``.test`` names. Normalization is what callers depend on. + normalize = lambda value: str(value).rstrip("/") # noqa: E731 + monkeypatch.setattr(device_connect, "validate_cloud_base_url", normalize) + monkeypatch.setattr(cloud_session, "validate_cloud_base_url", normalize) + return tmp_path + + +def _state_files(root: Path): + return [path for path in Path(root).rglob("*") if path.is_file()] + + +# --------------------------------------------------------------------------- happy path + + +def test_connect_writes_a_session_the_rest_of_the_client_can_use(monkeypatch, tmp_path): + """The whole point: after connect, ``cloud_session.configured()`` is true. + + Before this command shipped, the only writer of the session file was a function with + zero production callers, so this assertion could not hold for any customer. + """ + + opener = _Opener(body=REGISTRATION) + _install_opener(monkeypatch, opener) + + summary = device_connect.connect( + TOKEN, control_url=CONTROL_URL, compute_url=COMPUTE_URL + ) + + assert cloud_session.configured() is True + saved = json.loads((tmp_path / "cloud_session.json").read_text(encoding="utf-8")) + assert saved["schema"] == "engraphis-cloud-session/v1" + assert saved["organization_id"] == "org_alpha" + assert saved["installation_id"] == "instl_alpha" + assert saved["device_id"] == "devc_alpha" + assert saved["member_id"] == "mem_alpha" + assert saved["refresh_credential"] == "rotating-refresh-credential" + assert saved["token_subject"] == "device" + assert saved["control_url"] == CONTROL_URL + assert saved["compute_url"] == COMPUTE_URL + # The entitlement travelled with the registration, so the dashboard knows the plan on + # its very first boot instead of showing a paying Team customer the free core. + assert saved["plan"] == "team" + assert saved["cloud_access_active"] is True + assert "team" in saved["cloud_features"] + assert summary["organization_id"] == "org_alpha" + + +def test_connect_posts_exactly_the_documented_request(monkeypatch): + """The endpoint 422s on unknown fields, and takes no ``organization_id`` at all.""" + + opener = _Opener(body=REGISTRATION) + _install_opener(monkeypatch, opener) + + device_connect.connect( + TOKEN, + control_url=CONTROL_URL, + compute_url=COMPUTE_URL, + workspace_id="ws_alpha", + installation_label="CI runner", + device_name="build-box", + ) + + call = opener.calls[0] + assert call["url"] == CONTROL_URL + "/v1/devices/connect" + assert call["method"] == "POST" + assert call["timeout"] == device_connect.DEFAULT_TIMEOUT_SECONDS + body = json.loads(call["body"].decode("utf-8")) + assert body["connect_token"] == TOKEN + assert body["installation_client_id"].startswith("inst_") + assert body["device_client_id"].startswith("dev_") + assert body["workspace_id"] == "ws_alpha" + assert body["installation_label"] == "CI runner" + assert body["device_name"] == "build-box" + # The token carries the organization; sending one would be rejected. + assert "organization_id" not in body + assert set(body) <= { + "connect_token", "installation_client_id", "device_client_id", + "installation_label", "device_name", "platform", "app_version", "workspace_id", + } + + +def test_optional_fields_are_omitted_rather_than_sent_empty(monkeypatch): + opener = _Opener(body=REGISTRATION) + _install_opener(monkeypatch, opener) + + device_connect.connect( + TOKEN, control_url=CONTROL_URL, compute_url=COMPUTE_URL, device_name="" + ) + + body = json.loads(opener.calls[0]["body"].decode("utf-8")) + assert "workspace_id" not in body + assert "installation_label" not in body + assert "device_name" not in body + + +def test_client_ids_are_stable_across_connects(monkeypatch): + """A reconnect must re-present the same installation, not mint a phantom device.""" + + opener = _Opener(body=REGISTRATION) + _install_opener(monkeypatch, opener) + + first = device_connect.client_identity() + device_connect.connect(TOKEN, control_url=CONTROL_URL, compute_url=COMPUTE_URL) + device_connect.connect(TOKEN, control_url=CONTROL_URL, compute_url=COMPUTE_URL) + + bodies = [json.loads(call["body"].decode("utf-8")) for call in opener.calls] + assert {body["installation_client_id"] for body in bodies} == {first[0]} + assert {body["device_client_id"] for body in bodies} == {first[1]} + assert device_connect.client_identity() == first + + +# ------------------------------------------------------------------------ failure modes + + +def test_expired_or_consumed_token_is_actionable_and_writes_nothing( + monkeypatch, tmp_path, capsys +): + """401 is the control plane's single answer for expired / consumed / invalid. + + The copy therefore has to name all three and point at the one fix, and a refused + connect must not leave a half-written session behind for the dashboard to load. + """ + + _install_opener(monkeypatch, _Opener(error=_http_error(401))) + + exit_code = connect_cli.main([ + "--token", TOKEN, "--control-url", CONTROL_URL, "--compute-url", COMPUTE_URL, + ]) + + assert exit_code == 1 + captured = capsys.readouterr() + assert "expired" in captured.err + assert "already used" in captured.err + assert "account portal" in captured.err + assert not (tmp_path / "cloud_session.json").exists() + assert cloud_session.configured() is False + + +def test_lapsed_subscription_is_distinguishable_from_a_dead_token(monkeypatch, capsys): + """402 must not be flattened into "your token is bad" -- the fix is billing.""" + + _install_opener(monkeypatch, _Opener(error=_http_error(402))) + + exit_code = connect_cli.main([ + "--token", TOKEN, "--control-url", CONTROL_URL, "--compute-url", COMPUTE_URL, + ]) + + assert exit_code == 1 + message = capsys.readouterr().err + assert "subscription is not active" in message + assert "billing" in message + assert "expired" not in message + + with pytest.raises(device_connect.DeviceConnectError) as caught: + device_connect.connect(TOKEN, control_url=CONTROL_URL, compute_url=COMPUTE_URL) + assert caught.value.status == 402 + + +@pytest.mark.parametrize( + "status,fragment", + [ + (403, "organization owner"), + (422, "pip install -U engraphis"), + (429, "Too many connect attempts"), + (503, "not accepting new device activations"), + (500, "could not connect this device"), + ], +) +def test_each_failure_status_gets_its_own_copy(monkeypatch, status, fragment): + _install_opener(monkeypatch, _Opener(error=_http_error(status))) + + with pytest.raises(device_connect.DeviceConnectError) as caught: + device_connect.connect(TOKEN, control_url=CONTROL_URL, compute_url=COMPUTE_URL) + + assert fragment in str(caught.value) + assert caught.value.status == (status if status != 500 else 503) + + +def test_error_body_is_never_reflected_into_the_message(monkeypatch): + """Provider bodies are untrusted and may quote internal hosts.""" + + error = _http_error(503, b'{"detail":"upstream 10.0.0.7 activation freeze"}') + _install_opener(monkeypatch, _Opener(error=error)) + + with pytest.raises(device_connect.DeviceConnectError) as caught: + device_connect.connect(TOKEN, control_url=CONTROL_URL, compute_url=COMPUTE_URL) + + assert "10.0.0.7" not in str(caught.value) + + +def test_http_error_body_is_drained_and_closed(monkeypatch): + error = _http_error(429) + closed = [] + original_close = error.close + error.close = lambda: (closed.append(True), original_close()) + _install_opener(monkeypatch, _Opener(error=error)) + + with pytest.raises(device_connect.DeviceConnectError): + device_connect.connect(TOKEN, control_url=CONTROL_URL, compute_url=COMPUTE_URL) + + assert closed == [True] + + +def test_incomplete_read_is_not_an_oserror_or_a_valueerror(): + """Pins the premise the drain guard is built on, so a wrong comment cannot return. + + ``http.client.IncompleteRead.__mro__`` is + ``(IncompleteRead, HTTPException, Exception, BaseException, object)``. It is *only* an + ``HTTPException``; an earlier version of this module claimed it also subclassed + ``ValueError`` and guarded the drain with ``(OSError, ValueError)`` on that basis. + """ + + assert not issubclass(http.client.IncompleteRead, OSError) + assert not issubclass(http.client.IncompleteRead, ValueError) + assert issubclass(http.client.IncompleteRead, http.client.HTTPException) + + +@pytest.mark.parametrize("failure", [ + http.client.IncompleteRead(b'{"detail":"pri'), + http.client.LineTooLong("chunk size"), + ConnectionResetError("peer went away mid-drain"), +]) +def test_a_truncated_error_body_still_reports_the_status(monkeypatch, failure): + """A flaky drain must not replace the status copy with a traceback. + + ``exc.read()`` is called from *inside* the ``except HTTPError`` block, so the sibling + ``except http.client.HTTPException`` clause of the same ``try`` can never catch what it + raises -- the drain needs its own guard, and that guard has to name ``HTTPException`` + because ``IncompleteRead`` is neither an ``OSError`` nor a ``ValueError``. + """ + + error = _http_error(401) + + def _boom(*args, **kwargs): + raise failure + + error.read = _boom + error.close = _boom + _install_opener(monkeypatch, _Opener(error=error)) + + with pytest.raises(device_connect.DeviceConnectError) as caught: + device_connect.connect(TOKEN, control_url=CONTROL_URL, compute_url=COMPUTE_URL) + + # The customer gets the 401 copy they would have got from a well-behaved body. + assert caught.value.status == 401 + assert "Generate a new one" in str(caught.value) + assert TOKEN not in str(caught.value) + + +def test_the_cli_reports_a_truncated_error_body_instead_of_a_traceback( + monkeypatch, capsys +): + """End-to-end proof of the same path: exit 1 and clean copy, no stack trace.""" + + error = _http_error(401) + + def _boom(*args, **kwargs): + raise http.client.IncompleteRead(b'{"detail":"pri') + + error.read = _boom + error.close = _boom + _install_opener(monkeypatch, _Opener(error=error)) + + code = connect_cli.main( + ["--token", TOKEN, "--control-url", CONTROL_URL, "--compute-url", COMPUTE_URL] + ) + + assert code == 1 + err = capsys.readouterr().err + assert "Traceback" not in err + assert "IncompleteRead" not in err + assert "Generate a new one" in err + + +def test_network_failure_reports_an_outage_without_leaking_the_detail(monkeypatch): + _install_opener( + monkeypatch, _Opener(error=urllib.error.URLError("proxy.internal refused")) + ) + + with pytest.raises(device_connect.DeviceConnectError) as caught: + device_connect.connect(TOKEN, control_url=CONTROL_URL, compute_url=COMPUTE_URL) + + assert caught.value.status == 503 + assert "proxy.internal" not in str(caught.value) + + +@pytest.mark.parametrize("credential", [ + ["tok_abc"], # str(["tok_abc"]) == "['tok_abc']" -- truthy, and not a token + {"value": "tok_abc"}, + 12345, + True, +]) +def test_a_non_string_refresh_credential_is_refused(monkeypatch, tmp_path, credential): + """``str(value or "")`` is a coercion, not a validation. + + A JSON array or object arrives as a Python ``list``/``dict`` and ``str()`` renders its + ``repr`` -- truthy and non-empty. The old check accepted it, ``save_bootstrap`` wrote + the literal text ``['tok_abc']``, ``configured()`` read it back as a usable session and + the CLI reported success, while the next refresh submitted that junk. By then the + single-use connect token was spent, so the customer could not simply retry. + """ + + body = dict(REGISTRATION) + body["refresh_credential"] = credential + _install_opener(monkeypatch, _Opener(body=body)) + + with pytest.raises(device_connect.DeviceConnectError) as caught: + device_connect.connect(TOKEN, control_url=CONTROL_URL, compute_url=COMPUTE_URL) + + assert caught.value.status == 502 + assert "generate a new one" in str(caught.value).lower() + assert not (tmp_path / "cloud_session.json").exists() + assert cloud_session.configured() is False + + +def test_the_writer_refuses_a_non_string_credential_too(tmp_path): + """Defence at the boundary, not only at the one call site above.""" + + body = dict(REGISTRATION) + body["refresh_credential"] = ["tok_abc"] + + with pytest.raises(cloud_session.CloudSessionError): + cloud_session.save_bootstrap( + body, control_url=CONTROL_URL, compute_url=COMPUTE_URL + ) + + assert not (tmp_path / "cloud_session.json").exists() + + +def test_non_string_identity_fields_are_dropped_not_stringified(monkeypatch, tmp_path): + """A ``repr`` in the session file is junk the dashboard would display verbatim.""" + + body = dict(REGISTRATION) + body["installation_id"] = ["instl_alpha"] + body["device_id"] = {"id": "devc_alpha"} + body["refresh_expires_at"] = ["2026-08-21T00:00:00Z"] + _install_opener(monkeypatch, _Opener(body=body)) + + device_connect.connect(TOKEN, control_url=CONTROL_URL, compute_url=COMPUTE_URL) + + saved = json.loads((tmp_path / "cloud_session.json").read_text(encoding="utf-8")) + assert saved["installation_id"] == "" + assert saved["device_id"] == "" + assert saved["refresh_expires_at"] == "" + assert "[" not in saved["installation_id"] + # The valid fields still land. + assert saved["refresh_credential"] == "rotating-refresh-credential" + assert saved["organization_id"] == "org_alpha" + + +@pytest.mark.parametrize("credential", [None, "", " "]) +def test_a_success_without_a_refresh_credential_writes_nothing( + monkeypatch, tmp_path, credential +): + """A 200 spends the token even when the body is unusable, so "try again" is wrong. + + The control plane consumes the single-use connect token as it writes a 200. Re-running + the same command therefore cannot succeed -- it deterministically produces a 401 -- so + the copy has to send the customer to the portal for a fresh token, exactly like the + truncated-reply path does. + """ + + body = dict(REGISTRATION) + if credential is None: + body.pop("refresh_credential") + else: + body["refresh_credential"] = credential + _install_opener(monkeypatch, _Opener(body=body)) + + with pytest.raises(device_connect.DeviceConnectError) as caught: + device_connect.connect(TOKEN, control_url=CONTROL_URL, compute_url=COMPUTE_URL) + + message = str(caught.value) + assert caught.value.status == 502 + assert "generate a new one" in message.lower() + # The old copy sent the customer back into a guaranteed 401. + assert "Try again," not in message + assert TOKEN not in message + assert not (tmp_path / "cloud_session.json").exists() + + +@pytest.mark.parametrize("payload", [b"not json", b'"a string"', b"[]"]) +def test_a_non_object_response_is_rejected(monkeypatch, payload): + class _Raw(_Opener): + def open(self, request, timeout): + return _Response(payload) + + _install_opener(monkeypatch, _Raw()) + + with pytest.raises(device_connect.DeviceConnectError, match="invalid connect"): + device_connect.connect(TOKEN, control_url=CONTROL_URL, compute_url=COMPUTE_URL) + + +@pytest.mark.parametrize("payload", [ + # ``id=`` is not cosmetic on the oversized case: pytest puts the parameter in the test + # id, and a 64 KiB id overflows the 32767-character limit on ``PYTEST_CURRENT_TEST``. + pytest.param(b"not json", id="unparseable"), + pytest.param(b'"a string"', id="json-string"), + pytest.param(b"[]", id="json-array"), + pytest.param(b"{" + b" " * (64 * 1024 + 8), id="oversized"), +]) +def test_an_unusable_2xx_body_says_the_token_was_spent(monkeypatch, tmp_path, payload): + """Every branch past ``opener.open()`` follows a 2xx, which consumes the token. + + ``urllib`` raises ``HTTPError`` for any status >= 400 and ``_NoRedirect`` turns a 3xx + into one too, so reaching the body checks at all means the control plane answered + successfully -- and the single-use token is gone. Reporting only "invalid connect + response" left the customer retrying a consumed token into a guaranteed 401. + """ + + class _Raw(_Opener): + def open(self, request, timeout): + return _Response(payload) + + _install_opener(monkeypatch, _Raw()) + + with pytest.raises(device_connect.DeviceConnectError) as caught: + device_connect.connect(TOKEN, control_url=CONTROL_URL, compute_url=COMPUTE_URL) + + message = str(caught.value) + assert caught.value.status == 502 + assert "no session was saved" in message + assert "generate a new one" in message.lower() + assert TOKEN not in message + assert not (tmp_path / "cloud_session.json").exists() + + +@pytest.mark.parametrize("failure", [ + TimeoutError("read timed out"), + ConnectionResetError("peer reset mid-body"), + OSError("device disappeared"), +]) +def test_a_body_read_failure_after_success_headers_is_not_a_plain_retry( + monkeypatch, tmp_path, failure +): + """``OSError`` means opposite things either side of ``opener.open()``. + + Once ``open`` returns, urllib has already parsed a success status line, so the control + plane answered and the single-use token is spent. While the open and the body read + shared one ``try``, a socket that timed out or reset *mid-body* inherited the + connection-phase copy -- "temporarily unreachable ... try again" -- and sent the + customer back with a token that could not work. + """ + + class _DiesMidBody: + def __enter__(self): + return self + + def __exit__(self, *exc) -> bool: + return False + + def read(self, size: int = -1) -> bytes: + raise failure + + class _Opened(_Opener): + def open(self, request, timeout): + self.calls.append({"body": request.data}) + return _DiesMidBody() + + _install_opener(monkeypatch, _Opened()) + + with pytest.raises(device_connect.DeviceConnectError) as caught: + device_connect.connect(TOKEN, control_url=CONTROL_URL, compute_url=COMPUTE_URL) + + message = str(caught.value) + assert caught.value.status == 502 + assert "may already have been used" in message + # The connection-phase copy must not be what a post-response fault reports. + assert "temporarily unreachable" not in message + assert TOKEN not in message + assert not (tmp_path / "cloud_session.json").exists() + + +def test_a_bare_reset_from_the_open_phase_is_also_ambiguous(monkeypatch): + """A bare ``ConnectionResetError`` reaching us unwrapped came from ``getresponse()``. + + urllib wraps connect-and-send failures in ``URLError``; only ``getresponse()`` raises + through. So an unwrapped reset means the POST was already written, and the token may be + spent -- see ``test_a_failure_before_the_request_goes_out_still_says_retry`` for the + ``URLError`` half that still says "try again". + """ + + _install_opener( + monkeypatch, _Opener(error=ConnectionResetError("reset awaiting the reply")) + ) + + with pytest.raises(device_connect.DeviceConnectError) as caught: + device_connect.connect(TOKEN, control_url=CONTROL_URL, compute_url=COMPUTE_URL) + + assert caught.value.status == 502 + assert "may already have been used" in str(caught.value) + + +def test_a_lock_failure_racing_the_save_says_the_token_was_spent(monkeypatch): + """``_refresh_lock`` wraps a filesystem fault in ``CloudSessionError``. + + That means it does *not* reach the ``OSError`` clause that carries the spent-token + warning, so the bare lock message was forwarded on its own and the customer was left + retrying a token the control plane had already consumed. + """ + + _install_opener(monkeypatch, _Opener(body=REGISTRATION)) + + def _locked(*args, **kwargs): + raise cloud_session.CloudSessionError( + "The cloud session refresh lock is unavailable or unsafe.", status=409 + ) + + monkeypatch.setattr(cloud_session, "save_bootstrap", _locked) + + with pytest.raises(device_connect.DeviceConnectError) as caught: + device_connect.connect(TOKEN, control_url=CONTROL_URL, compute_url=COMPUTE_URL) + + message = str(caught.value) + assert caught.value.status == 409 + # The underlying cause survives... + assert "refresh lock is unavailable or unsafe" in message + # ...and so does the fact that retrying this token cannot work. + assert "has now been used" in message + assert "generate a new one" in message.lower() + assert TOKEN not in message + + +# ------------------------------------------- failures *after* the token has been spent +# +# Once ``opener.open`` has returned, the control plane has answered and the single-use +# connect token is gone. Every fault from that point on has to arrive as a +# ``DeviceConnectError`` that says so -- a traceback here is the worst possible outcome, +# because the customer cannot tell whether to retry or fetch a new token. + + +class _TruncatedBody: + """A 200 whose body stops mid-stream, the way a dropped chunked reply does.""" + + def __enter__(self): + return self + + def __exit__(self, *exc) -> bool: + return False + + def read(self, size: int = -1) -> bytes: + # What ``HTTPResponse._read_chunked`` raises for a truncated body. Its only base + # is ``HTTPException`` -- it is neither an ``OSError``, a ``ValueError``, nor a + # ``URLError``. See ``test_incomplete_read_is_not_an_oserror_or_a_valueerror``. + raise http.client.IncompleteRead(b'{"organization_id":"org_al') + + +@pytest.mark.parametrize("failure", [ + http.client.IncompleteRead(b'{"organization_id":"org_al'), + http.client.LineTooLong("header line"), + http.client.BadStatusLine("garbage"), +]) +def test_a_truncated_reply_reports_the_token_state_not_a_traceback( + monkeypatch, tmp_path, failure +): + """``IncompleteRead`` is an ``HTTPException``, so the transport clause never saw it.""" + + class _Broken(_Opener): + def open(self, request, timeout): + self.calls.append({"body": request.data}) + if isinstance(failure, http.client.IncompleteRead): + return _TruncatedBody() + raise failure + + opener = _Broken() + _install_opener(monkeypatch, opener) + + with pytest.raises(device_connect.DeviceConnectError) as caught: + device_connect.connect(TOKEN, control_url=CONTROL_URL, compute_url=COMPUTE_URL) + + message = str(caught.value) + assert caught.value.status == 502 + # The customer must be told the token may be gone rather than told to just retry. + assert "may already have been used" in message + assert TOKEN not in message + assert not (tmp_path / "cloud_session.json").exists() + + +def test_a_reset_with_no_reply_is_ambiguous_not_a_clean_miss(monkeypatch): + """This assertion is the reverse of what it once was, and the reversal is the point. + + The old copy claimed ``RemoteDisconnected`` meant "the peer closed without answering at + all. Nothing was consumed." That was an overclaim: ``RemoteDisconnected`` is raised when + ``getresponse()`` reads zero bytes for the status line, which happens *after* + ``h.request(...)`` has written the POST in full. The control plane may well have + received and processed it, and a processed connect spends the token -- so telling the + customer to retry sent them into a 401 reading "the token I just generated is invalid". + + ``_TRUNCATED_REPLY`` is the honest copy under ambiguity: check the portal, then decide. + """ + + _install_opener( + monkeypatch, _Opener(error=http.client.RemoteDisconnected("closed early")) + ) + + with pytest.raises(device_connect.DeviceConnectError) as caught: + device_connect.connect(TOKEN, control_url=CONTROL_URL, compute_url=COMPUTE_URL) + + assert caught.value.status == 502 + assert "may already have been used" in str(caught.value) + assert "temporarily unreachable" not in str(caught.value) + + +@pytest.mark.parametrize("reason", [ + ConnectionRefusedError("connection refused"), + OSError("dns failure"), + TimeoutError("connect timed out"), +]) +def test_a_failure_before_the_request_goes_out_still_says_retry(monkeypatch, reason): + """The other half of the same split, and the reason it is not blanket-ambiguous. + + ``AbstractHTTPHandler.do_open`` wraps everything from establishing the connection and + from ``h.request(...)`` in ``URLError``, and lets ``h.getresponse()`` raise unwrapped. + So a ``URLError`` is the one honest "nothing was consumed" signal -- DNS, refused, TLS, + a write that never completed -- and those customers must not be told to fetch a new + token for what is just flaky wifi. + """ + + _install_opener(monkeypatch, _Opener(error=urllib.error.URLError(reason))) + + with pytest.raises(device_connect.DeviceConnectError) as caught: + device_connect.connect(TOKEN, control_url=CONTROL_URL, compute_url=COMPUTE_URL) + + assert caught.value.status == 503 + assert "temporarily unreachable" in str(caught.value) + assert "may already have been used" not in str(caught.value) + + +@pytest.mark.parametrize("failure", [ + CloudUrlUnresolved("cloud service URL could not be resolved"), + ValueError("cloud service URL must not target private/reserved IP ranges"), +]) +def test_endpoint_validation_failing_after_redemption_is_not_a_traceback( + monkeypatch, tmp_path, failure +): + """``save_bootstrap`` re-resolves both endpoints *after* the POST spent the token. + + ``CloudUrlUnresolved`` is a ``ValueError``, so neither the ``CloudSessionError`` nor + the ``OSError`` handler in ``connect()`` covers it: a resolver that dies mid-connect, + or a host that starts resolving to a rejected address, escaped as a raw traceback. + """ + + opener = _Opener(body=REGISTRATION) + _install_opener(monkeypatch, opener) + + def _rejects(value): + raise failure + + # Only the save path: the pre-POST checks in ``connect()`` already passed. + monkeypatch.setattr(cloud_session, "validate_cloud_base_url", _rejects) + + with pytest.raises(device_connect.DeviceConnectError) as caught: + device_connect.connect(TOKEN, control_url=CONTROL_URL, compute_url=COMPUTE_URL) + + message = str(caught.value) + assert caught.value.status == 409 + assert opener.calls, "the token was spent; the copy has to admit it" + assert "has been used" in message + assert TOKEN not in message + assert not (tmp_path / "cloud_session.json").exists() + + +def test_the_cli_reports_a_post_redemption_fault_instead_of_a_traceback( + monkeypatch, capsys +): + _install_opener(monkeypatch, _Opener(body=REGISTRATION)) + + def _rejects(value): + raise CloudUrlUnresolved("cloud service URL could not be resolved") + + monkeypatch.setattr(cloud_session, "validate_cloud_base_url", _rejects) + + code = connect_cli.main([ + "--token", TOKEN, "--control-url", CONTROL_URL, "--compute-url", COMPUTE_URL, + ]) + + err = capsys.readouterr().err + assert code == 1 + assert "Traceback" not in err + assert "has been used" in err + + +# ------------------------------------------------------ pre-flight on session storage +# +# The POST is the point of no return: the control plane consumes the single-use connect +# token as it answers. ``save_bootstrap`` runs afterwards, so before the pre-flight a +# state directory that had lost its permissions -- or a ``cloud_session.json`` replaced +# by a link -- spent the customer's token and then failed, leaving them with nothing to +# retry with. Every test here asserts the same thing: no request was put on the wire. + + +def _probe_files(root: Path): + """Temporary files the writability probe must never leave behind.""" + + return sorted(path.name for path in Path(root).iterdir() if ".preflight." in path.name) + + +def test_an_unsafe_session_path_fails_before_the_token_is_spent(monkeypatch, tmp_path): + """The reported regression: a session leaf that is not a plain private file. + + A directory stands in for the symlink/hard-link family because it trips exactly the + same ``private_file_stat`` rejection on every platform, including a Windows runner + without the privilege to create a symlink at all. + """ + + opener = _Opener(body=REGISTRATION) + _install_opener(monkeypatch, opener) + # The identity file already exists, which is precisely why its earlier success is no + # evidence that the directory is still writable now. + device_connect.client_identity() + session_path = tmp_path / "cloud_session.json" + session_path.mkdir() + + with pytest.raises(device_connect.DeviceConnectError) as caught: + device_connect.connect(TOKEN, control_url=CONTROL_URL, compute_url=COMPUTE_URL) + + # The one assertion that matters: the token was never presented, so it is still live. + assert opener.calls == [] + assert caught.value.status == 409 + # The customer has to be told which path to fix, by name. + assert str(session_path) in str(caught.value) + assert "has not been used" in str(caught.value) + assert session_path.is_dir() + assert _probe_files(tmp_path) == [] + + +def test_an_unsafe_refresh_lock_also_fails_before_the_token_is_spent(monkeypatch, tmp_path): + """``save_bootstrap`` takes the refresh lock first, so an unusable lock spends it too.""" + + opener = _Opener(body=REGISTRATION) + _install_opener(monkeypatch, opener) + lock_path = tmp_path / ".cloud_session.refresh.lock" + lock_path.mkdir() + + with pytest.raises(device_connect.DeviceConnectError) as caught: + device_connect.connect(TOKEN, control_url=CONTROL_URL, compute_url=COMPUTE_URL) + + assert opener.calls == [] + assert str(lock_path) in str(caught.value) + assert not (tmp_path / "cloud_session.json").exists() + + +def test_a_hard_linked_session_is_refused_before_the_token_is_spent(monkeypatch, tmp_path): + """A second pathname to the session would keep resolving to the rotated credential.""" + + opener = _Opener(body=REGISTRATION) + _install_opener(monkeypatch, opener) + session_path = tmp_path / "cloud_session.json" + session_path.write_text("{}", encoding="utf-8") + try: + os.link(str(session_path), str(tmp_path / "session_alias.json")) + except (OSError, AttributeError, NotImplementedError) as exc: # pragma: no cover + pytest.skip("hard links are unavailable on this filesystem: %s" % exc) + + with pytest.raises(device_connect.DeviceConnectError) as caught: + device_connect.connect(TOKEN, control_url=CONTROL_URL, compute_url=COMPUTE_URL) + + assert opener.calls == [] + assert str(session_path) in str(caught.value) + # Inspection only: a pre-existing file is never opened, truncated or replaced. + assert session_path.read_text(encoding="utf-8") == "{}" + + +def test_an_uncreatable_state_directory_fails_before_the_token_is_spent( + monkeypatch, tmp_path +): + """``ENGRAPHIS_STATE_DIR`` under a regular file: the directory can never be made.""" + + blocker = tmp_path / "blocker" + blocker.write_text("not a directory", encoding="utf-8") + monkeypatch.setenv("ENGRAPHIS_STATE_DIR", str(blocker / "state")) + opener = _Opener(body=REGISTRATION) + _install_opener(monkeypatch, opener) + + with pytest.raises(device_connect.DeviceConnectError) as caught: + device_connect.connect(TOKEN, control_url=CONTROL_URL, compute_url=COMPUTE_URL) + + assert opener.calls == [] + assert str(blocker / "state") in str(caught.value) + assert blocker.read_text(encoding="utf-8") == "not a directory" + + +def test_an_unwritable_state_directory_fails_before_the_token_is_spent( + monkeypatch, tmp_path +): + """A read-only mount, which is the failure the report describes. + + Simulated at the probe rather than with ``chmod``: POSIX modes are advisory for root + and Windows has no equivalent, so a mode-based test would silently stop failing. The + probe is the only thing standing between the caller and ``atomic_private_text``. + """ + + opener = _Opener(body=REGISTRATION) + _install_opener(monkeypatch, opener) + device_connect.client_identity() + + def _read_only(*args, **kwargs): + raise PermissionError(13, "read-only file system") + + monkeypatch.setattr(cloud_session.tempfile, "mkstemp", _read_only) + + with pytest.raises(device_connect.DeviceConnectError) as caught: + device_connect.connect(TOKEN, control_url=CONTROL_URL, compute_url=COMPUTE_URL) + + assert opener.calls == [] + assert caught.value.status == 409 + assert str(tmp_path) in str(caught.value) + assert "not writable" in str(caught.value) + assert not (tmp_path / "cloud_session.json").exists() + + +def test_the_preflight_creates_nothing_and_leaves_nothing_behind(monkeypatch, tmp_path): + """A pre-flight that wrote the session would defeat the "nothing on failure" rule.""" + + session_path = cloud_session.preflight_save() + + assert session_path == tmp_path / "cloud_session.json" + assert not session_path.exists() + assert _probe_files(tmp_path) == [] + + # Idempotent, and equally inert once a real session exists. + _install_opener(monkeypatch, _Opener(body=REGISTRATION)) + device_connect.connect(TOKEN, control_url=CONTROL_URL, compute_url=COMPUTE_URL) + saved = session_path.read_bytes() + + assert cloud_session.preflight_save() == session_path + assert session_path.read_bytes() == saved + assert _probe_files(tmp_path) == [] + + +def test_an_unopenable_refresh_lock_fails_before_the_token_is_spent(monkeypatch, tmp_path): + """``private_file_stat`` is an ``lstat``; it does not prove the lock can be opened. + + ``_refresh_lock`` opens it ``O_RDWR``, so a lock file left behind by another UID passes + the stat, the preflight passes, the token is redeemed -- and then ``save_bootstrap`` + fails opening a file whose inaccessibility was detectable before the POST. + """ + + lock_path = tmp_path / ".cloud_session.refresh.lock" + lock_path.write_bytes(b"") + + opener = _Opener(body=REGISTRATION) + _install_opener(monkeypatch, opener) + + real_open = cloud_session.os.open + + def _denied(path, flags, *args): + if str(path).endswith(".cloud_session.refresh.lock"): + raise PermissionError("owned by another uid") + return real_open(path, flags, *args) + + monkeypatch.setattr(cloud_session.os, "open", _denied) + + with pytest.raises(device_connect.DeviceConnectError) as caught: + device_connect.connect(TOKEN, control_url=CONTROL_URL, compute_url=COMPUTE_URL) + + # Detected before the POST, so the token is untouched. + assert opener.calls == [] + assert caught.value.status == 409 + assert "refresh lock" in str(caught.value) + assert not (tmp_path / "cloud_session.json").exists() + + +def test_a_probe_that_cannot_be_removed_fails_the_preflight(monkeypatch, tmp_path): + """Creating a file and removing one are separate rights. + + A directory ACL granting add-file but denying delete lets ``mkstemp`` succeed and + ``unlink`` fail. ``atomic_private_text`` finishes with ``os.replace`` over the session + leaf, which needs exactly the right that just failed -- so swallowing the ``unlink`` + error let the preflight pass on a directory the real save cannot use, redeeming the + single-use token before failing, and littered a probe file on every attempt. + """ + + opener = _Opener(body=REGISTRATION) + _install_opener(monkeypatch, opener) + + real_unlink = cloud_session.os.unlink + + def _no_delete(target): + if ".preflight." in str(target): + raise PermissionError("delete denied by ACL") + return real_unlink(target) + + monkeypatch.setattr(cloud_session.os, "unlink", _no_delete) + + with pytest.raises(device_connect.DeviceConnectError) as caught: + device_connect.connect(TOKEN, control_url=CONTROL_URL, compute_url=COMPUTE_URL) + + # The token was never put on the wire, so it is still usable once the ACL is fixed. + assert opener.calls == [] + assert caught.value.status == 409 + assert "replaced" in str(caught.value) + assert str(tmp_path) in str(caught.value) + assert not (tmp_path / "cloud_session.json").exists() + + +# ------------------------------------------------------------- pre-flight token hygiene + + +@pytest.mark.parametrize("token", [ + "", + " ", + "engr_ct_", + "engr_ct_short", + "ct_missing_prefix_but_long_enough_to_pass", + "engr_ct_has spaces in the middle", + "engr_ct_" + "x" * 600, + "engraphis connect --token engr_ct_pasted_whole_command", +]) +def test_malformed_tokens_are_rejected_before_any_network_call(monkeypatch, token): + """A bad paste must cost nothing: no request, no rate budget, no consumed token.""" + + class _Poison: + def open(self, request, timeout): + raise AssertionError("a malformed token must not reach the network") + + _install_opener(monkeypatch, _Poison()) + + with pytest.raises(device_connect.DeviceConnectError) as caught: + device_connect.connect(token, control_url=CONTROL_URL) + + assert caught.value.status == 400 + + +def test_a_rejected_token_is_not_quoted_back(monkeypatch): + """Even an invalid token is a credential: it must stay out of the error text.""" + + secret = "engr_ct_" + "s" * 4 # too short, but still secret-shaped + with pytest.raises(device_connect.DeviceConnectError) as caught: + device_connect.connect(secret, control_url=CONTROL_URL) + + assert secret not in str(caught.value) + assert "ssss" not in str(caught.value) + + +def test_surrounding_whitespace_from_a_paste_is_tolerated(monkeypatch): + opener = _Opener(body=REGISTRATION) + _install_opener(monkeypatch, opener) + + device_connect.connect( + " %s\n" % TOKEN, control_url=CONTROL_URL, compute_url=COMPUTE_URL + ) + + body = json.loads(opener.calls[0]["body"].decode("utf-8")) + assert body["connect_token"] == TOKEN + + +# ------------------------------------------------------------------ credential leakage + + +def test_the_token_never_reaches_stdout_stderr_or_disk(monkeypatch, tmp_path, capsys): + """The one invariant that has to hold on the success path too. + + A token echoed into a terminal lands in scrollback, CI logs and screenshots; a token + persisted next to the session outlives the single use it was minted for. + """ + + opener = _Opener(body=REGISTRATION) + _install_opener(monkeypatch, opener) + + exit_code = connect_cli.main([ + "--token", TOKEN, "--control-url", CONTROL_URL, "--compute-url", COMPUTE_URL, + ]) + assert exit_code == 0 + + captured = capsys.readouterr() + assert TOKEN not in captured.out + assert TOKEN not in captured.err + written = _state_files(tmp_path) + assert written, "connect must actually write the session" + for path in written: + assert TOKEN.encode("utf-8") not in path.read_bytes(), "token leaked into %s" % path + # It did go where it belongs. + assert TOKEN.encode("utf-8") in opener.calls[0]["body"] + + +def test_the_session_secrets_are_not_printed(monkeypatch, capsys): + _install_opener(monkeypatch, _Opener(body=REGISTRATION)) + + connect_cli.main([ + "--token", TOKEN, "--control-url", CONTROL_URL, "--compute-url", COMPUTE_URL, + ]) + + captured = capsys.readouterr() + assert REGISTRATION["refresh_credential"] not in captured.out + assert REGISTRATION["access_token"] not in captured.out + assert "org_alpha" in captured.out # the useful, non-secret parts are shown + + +def test_the_json_summary_carries_no_secrets(monkeypatch, capsys): + _install_opener(monkeypatch, _Opener(body=REGISTRATION)) + + connect_cli.main([ + "--token", TOKEN, "--control-url", CONTROL_URL, "--compute-url", COMPUTE_URL, + "--json", + ]) + + summary = json.loads(capsys.readouterr().out) + assert "refresh_credential" not in summary + assert "access_token" not in summary + assert summary["plan"] == "team" + assert summary["control_url"] == CONTROL_URL + + +@pytest.mark.parametrize("as_json", [True, False]) +def test_an_unusable_session_prints_no_success_output_at_all(monkeypatch, capsys, as_json): + """The usability check must run *before* success is announced, not after. + + ``cloud_session.configured()`` reads the session back and can raise (an invalid + ``ENGRAPHIS_CLOUD_TOKEN_SUBJECT``, or the file changing under the read). While the + check ran last, a complete ``--json`` success object was already on stdout when the + command then wrote an error and exited 1 -- so a consumer parsing stdout accepted a + connect that had failed, and a human got two contradictory answers. + """ + + _install_opener(monkeypatch, _Opener(body=REGISTRATION)) + + def _unusable(*args, **kwargs): + raise cloud_session.CloudSessionError( + "The saved cloud session names an unknown token subject.", status=409 + ) + + monkeypatch.setattr(cloud_session, "configured", _unusable) + + argv = ["--token", TOKEN, "--control-url", CONTROL_URL, "--compute-url", COMPUTE_URL] + if as_json: + argv.append("--json") + + assert connect_cli.main(argv) == 1 + + captured = capsys.readouterr() + # Nothing on stdout at all: no JSON object to mis-parse, no "Connected" to believe. + assert captured.out.strip() == "" + assert "not usable" in captured.err + assert "Connected this device" not in captured.out + assert TOKEN not in captured.out + assert TOKEN not in captured.err + + +def test_summarize_drops_every_secret_field(): + summary = device_connect.summarize(REGISTRATION) + assert set(summary) & {"refresh_credential", "access_token"} == set() + assert summary["organization_id"] == "org_alpha" + + +# ------------------------------------------------------------------------- CLI surface + + +def test_cli_reports_the_connection_and_the_next_step(monkeypatch, capsys): + _install_opener(monkeypatch, _Opener(body=REGISTRATION)) + + assert connect_cli.main([ + "--token", TOKEN, "--control-url", CONTROL_URL, "--compute-url", COMPUTE_URL, + ]) == 0 + + out = capsys.readouterr().out + assert "Connected this device to Engraphis Cloud." in out + assert "team" in out + assert "cloud_session.json" in out + assert "engraphis-dashboard" in out + + +def test_cli_says_so_when_compute_is_not_configured(monkeypatch, capsys): + """Connecting against a custom control plane must not silently half-configure.""" + + _install_opener(monkeypatch, _Opener(body=REGISTRATION)) + + assert connect_cli.main(["--token", TOKEN, "--control-url", CONTROL_URL]) == 0 + + out = capsys.readouterr().out + assert "hosted compute is not configured" in out + assert "ENGRAPHIS_CLOUD_COMPUTE_URL" in out + assert cloud_session.configured(require_compute=False) is True + + +def test_compute_url_is_taken_from_the_environment(monkeypatch): + monkeypatch.setenv("ENGRAPHIS_CLOUD_COMPUTE_URL", COMPUTE_URL) + _install_opener(monkeypatch, _Opener(body=REGISTRATION)) + + summary = device_connect.connect(TOKEN, control_url=CONTROL_URL) + + assert summary["compute_url"] == COMPUTE_URL + assert cloud_session.configured() is True + + +def test_control_url_defaults_to_the_shipped_manifest(monkeypatch): + from engraphis.commercial import manifest + + assert device_connect.default_control_url() == manifest()["control_plane"] + # And only the shipped control plane gets a guessed compute endpoint. + assert device_connect.default_compute_url(manifest()["control_plane"]) == \ + device_connect.DEFAULT_COMPUTE_URL + assert device_connect.default_compute_url(CONTROL_URL) == "" + + +def test_the_manifest_outranks_the_compute_constant(monkeypatch): + """A published ``compute_plane`` wins, so the manifest stays the endpoint authority. + + The constant is only the fallback for today's manifest, which declares no + ``compute_plane``; reading the key alone would resolve ``""`` and save a session + ``configured()`` rejects. + """ + + from engraphis.commercial import manifest + + shipped = manifest()["control_plane"] + monkeypatch.setattr( + "engraphis.commercial.manifest", + lambda: {"control_plane": shipped, "compute_plane": COMPUTE_URL}, + ) + + assert device_connect.default_compute_url(shipped) == COMPUTE_URL + # A self-hosted control plane still gets no guess from either source. + assert device_connect.default_compute_url(CONTROL_URL) == "" + + +@pytest.mark.parametrize("spelling", [ + "https://api.engraphis.test", + "HTTPS://API.ENGRAPHIS.TEST", + "https://API.engraphis.test", + "https://api.engraphis.test:443", + "https://api.engraphis.test/", + "HTTPS://API.ENGRAPHIS.TEST:443/", +]) +def test_equivalent_spellings_of_the_shipped_control_url_still_get_compute( + monkeypatch, spelling +): + """``validate_cloud_base_url`` normalises the scheme but not the netloc. + + It returns ``urlunsplit((scheme, parts.netloc, ...))``, so ``HTTPS://API.EXAMPLE`` keeps + its upper-case host and ``:443`` is never dropped. A raw string comparison against the + shipped URL therefore missed, ``default_compute_url`` returned "", and a customer who + typed the production endpoint with a different case or an explicit port connected + successfully and then found every hosted feature off, because + ``cloud_session.configured()`` needs a compute endpoint. + """ + + monkeypatch.setattr( + "engraphis.commercial.manifest", + lambda: { + "control_plane": "https://api.engraphis.test", + "compute_plane": COMPUTE_URL, + }, + ) + + assert device_connect.default_compute_url(spelling) == COMPUTE_URL + + +@pytest.mark.parametrize("other", [ + "https://api.engraphis.test:8443", # a different port really is a different endpoint + "http://api.engraphis.test", # and so is a different scheme + "https://api.engraphis.test/v2", # and a different path + "https://evil.example.test", +]) +def test_a_genuinely_different_control_url_still_gets_no_compute(monkeypatch, other): + """The normalisation must not become a wildcard: only equivalents may match.""" + + monkeypatch.setattr( + "engraphis.commercial.manifest", + lambda: { + "control_plane": "https://api.engraphis.test", + "compute_plane": COMPUTE_URL, + }, + ) + + assert device_connect.default_compute_url(other) == "" + + +def test_an_unresolvable_home_directory_is_an_error_not_a_traceback(monkeypatch, capsys): + """``Path.home()`` raises ``RuntimeError`` on a service account with no home. + + ``_read_identity`` swallowed it and returned ``{}``, but the *next* ``_identity_path()`` + call sits outside that guard and raised it again. The CLI catches only + ``DeviceConnectError``, so a command whose whole selling point is being non-interactive + and scriptable printed a traceback. + """ + + monkeypatch.delenv("ENGRAPHIS_STATE_DIR", raising=False) + + def _no_home(): + raise RuntimeError("Could not determine home directory") + + monkeypatch.setattr(device_connect.Path, "home", staticmethod(_no_home)) + + opener = _Opener(body=REGISTRATION) + _install_opener(monkeypatch, opener) + + code = connect_cli.main( + ["--token", TOKEN, "--control-url", CONTROL_URL, "--compute-url", COMPUTE_URL] + ) + + assert code == 1 + err = capsys.readouterr().err + assert "Traceback" not in err + assert "ENGRAPHIS_STATE_DIR" in err + # Nothing was sent, so the token is still good once the variable is set. + assert opener.calls == [] + + +@pytest.mark.parametrize("bad", ["nan", "inf", "-inf", "0", "-5", "1e9", "10000000000"]) +def test_a_non_finite_timeout_is_refused_before_any_request(monkeypatch, bad): + """``argparse``'s ``type=float`` accepts ``nan``/``inf``; the socket layer does not. + + Those reach ``urllib``'s deadline arithmetic and raise ``ValueError``/``OverflowError`` + that ``post_connect`` does not catch, so the CLI printed a traceback instead of its + structured error. Nothing may be sent, either. + """ + + class _Exploding: + @staticmethod + def open(request, timeout=None): # pragma: no cover - must never run + raise AssertionError("a request was started with an unusable timeout") + + _install_opener(monkeypatch, _Exploding()) + + with pytest.raises(device_connect.DeviceConnectError) as caught: + device_connect.connect(TOKEN, control_url=CONTROL_URL, timeout=float(bad)) + + assert caught.value.status == 400 + assert "timeout" in str(caught.value) + + +def test_a_huge_finite_timeout_is_refused_before_the_socket_overflows(): + """``math.isfinite`` is not the same test as "a socket can use this". + + ``socket.settimeout`` computes an absolute deadline, so a large but finite value raises + ``OverflowError`` ("timeout doesn't fit into C timeval" / "timestamp out of range for + platform time_t") -- an exception ``post_connect`` does not catch, so ``--timeout + 10000000000`` printed a traceback. The bound is checked against a real socket here so + the constant cannot drift into the range that overflows. + """ + + sock = socket.socket() + try: + sock.settimeout(device_connect._MAX_TIMEOUT_SECONDS) + assert sock.gettimeout() == device_connect._MAX_TIMEOUT_SECONDS + finally: + sock.close() + + with pytest.raises(device_connect.DeviceConnectError) as caught: + device_connect._validated_timeout(device_connect._MAX_TIMEOUT_SECONDS + 1) + + assert caught.value.status == 400 + assert "timeout" in str(caught.value) + + +def test_the_cli_reports_an_oversized_timeout_instead_of_a_traceback(monkeypatch, capsys): + class _Exploding: + @staticmethod + def open(request, timeout=None): # pragma: no cover - must never run + raise AssertionError("a request was started with an unusable timeout") + + _install_opener(monkeypatch, _Exploding()) + + code = connect_cli.main( + ["--token", TOKEN, "--control-url", CONTROL_URL, "--timeout", "10000000000"] + ) + + assert code == 1 + err = capsys.readouterr().err + assert "timeout" in err + assert "Traceback" not in err + + +def test_the_cli_reports_a_bad_timeout_instead_of_a_traceback(monkeypatch, capsys): + class _Exploding: + @staticmethod + def open(request, timeout=None): # pragma: no cover - must never run + raise AssertionError("a request was started with an unusable timeout") + + _install_opener(monkeypatch, _Exploding()) + + code = connect_cli.main( + ["--token", TOKEN, "--control-url", CONTROL_URL, "--timeout", "nan"] + ) + + assert code == 1 + err = capsys.readouterr().err + assert "timeout" in err + assert "Traceback" not in err + + +def test_a_storage_fault_racing_the_save_is_not_a_traceback(monkeypatch): + """``UnsafeStateFile`` is an ``OSError``, not a ``CloudSessionError``. + + The pre-flight closes the common case, but the state directory can still change + between the pre-flight and the write. The token is spent by then, so the customer + must be told that plainly rather than shown a stack trace. + """ + + _install_opener(monkeypatch, _Opener(body=REGISTRATION)) + + def _boom(*args, **kwargs): + raise UnsafeStateFile("cloud_session.json is not a plain private file") + + monkeypatch.setattr(cloud_session, "save_bootstrap", _boom) + + with pytest.raises(device_connect.DeviceConnectError) as caught: + device_connect.connect(TOKEN, control_url=CONTROL_URL) + + message = str(caught.value) + assert caught.value.status == 409 + # The token is gone; the copy must not promise it can be reused. + assert "has been used" in message + assert TOKEN not in message + + +def test_control_url_env_override_wins(monkeypatch): + monkeypatch.setenv("ENGRAPHIS_CLOUD_CONTROL_URL", CONTROL_URL) + assert device_connect.default_control_url() == CONTROL_URL + + +def test_an_unconfigured_control_plane_is_a_clear_error(monkeypatch): + monkeypatch.setattr(device_connect, "default_control_url", lambda: "") + + with pytest.raises(device_connect.DeviceConnectError) as caught: + device_connect.connect(TOKEN) + + assert "ENGRAPHIS_CLOUD_CONTROL_URL" in str(caught.value) + assert caught.value.status == 400 + + +def test_token_dash_reads_stdin(monkeypatch, capsys): + _install_opener(monkeypatch, _Opener(body=REGISTRATION)) + + class _Stdin: + @staticmethod + def isatty() -> bool: + return False + + @staticmethod + def readline() -> str: + return TOKEN + "\n" + + monkeypatch.setattr(connect_cli.sys, "stdin", _Stdin) + + assert connect_cli.main([ + "--token", "-", "--control-url", CONTROL_URL, "--compute-url", COMPUTE_URL, + ]) == 0 + assert TOKEN not in capsys.readouterr().out + + +def test_token_dash_on_a_terminal_explains_itself_instead_of_hanging(monkeypatch, capsys): + class _Tty: + @staticmethod + def isatty() -> bool: + return True + + monkeypatch.setattr(connect_cli.sys, "stdin", _Tty) + + assert connect_cli.main(["--token", "-", "--control-url", CONTROL_URL]) == 1 + assert "reads the token from stdin" in capsys.readouterr().err + + +def test_connect_is_reachable_as_the_documented_engraphis_verb(monkeypatch, capsys): + """The portal shows `engraphis connect --token ...`; that string has to run.""" + + from scripts import entry + + _install_opener(monkeypatch, _Opener(body=REGISTRATION)) + assert entry.COMMANDS["connect"] == "scripts.connect:main" + + assert entry.main([ + "connect", "--token", TOKEN, "--control-url", CONTROL_URL, + "--compute-url", COMPUTE_URL, + ]) == 0 + assert "Connected this device to Engraphis Cloud." in capsys.readouterr().out + + +def test_the_dispatcher_restores_argv_and_rejects_unknown_verbs(capsys): + from scripts import entry + + saved = list(entry.sys.argv) + assert entry.main(["definitely-not-a-command"]) == 2 + assert entry.sys.argv == saved + assert "unknown command" in capsys.readouterr().err + + +def test_every_dispatched_verb_resolves_to_a_real_callable(): + from importlib import import_module + + from scripts import entry + + for verb, target in entry.COMMANDS.items(): + module_name, _, attribute = target.partition(":") + assert callable(getattr(import_module(module_name), attribute)), verb diff --git a/tests/test_licensing_launch.py b/tests/test_licensing_launch.py index 9f07b57..da1e370 100644 --- a/tests/test_licensing_launch.py +++ b/tests/test_licensing_launch.py @@ -14,6 +14,7 @@ import json import socket import urllib.error +import http.client import pytest @@ -158,7 +159,7 @@ def open(self, request, timeout=None): [ urllib.error.URLError(ConnectionRefusedError("connection refused")), urllib.error.URLError(socket.gaierror("name resolution failed")), - TimeoutError("timed out"), + urllib.error.URLError(TimeoutError("timed out before send")), ], ) def test_transport_failures_report_a_retryable_outage(monkeypatch, error) -> None: @@ -170,6 +171,42 @@ def test_transport_failures_report_a_retryable_outage(monkeypatch, error) -> Non cloud_session._post_refresh("https://control.example.test", "r", "ws", "member") +@pytest.mark.parametrize("error", [ + TimeoutError("timed out waiting for status"), + http.client.RemoteDisconnected("closed waiting for status"), + ConnectionResetError("reset waiting for status"), + OSError("TLS connection failed while reading status"), +]) +def test_post_send_refresh_transport_failures_require_reconnect(monkeypatch, error) -> None: + """Unwrapped getresponse failures are ambiguous after the refresh POST was written.""" + + monkeypatch.setattr( + cloud_session, "build_pinned_https_opener", _opener_raising(error) + ) + + with pytest.raises(CloudSessionError, match="Connect this installation again") as caught: + cloud_session._post_refresh("https://control.example.test", "r", "ws", "member") + + assert caught.value.status == 409 + + +@pytest.mark.parametrize("error", [ + http.client.BadStatusLine("garbled status"), + http.client.LineTooLong("status line too long"), +]) +def test_malformed_refresh_status_after_post_requires_reconnect(monkeypatch, error) -> None: + """The POST may have spent the credential before ``getresponse`` rejects its status.""" + + monkeypatch.setattr( + cloud_session, "build_pinned_https_opener", _opener_raising(error) + ) + + with pytest.raises(CloudSessionError, match="Connect this installation again") as caught: + cloud_session._post_refresh("https://control.example.test", "r", "ws", "member") + + assert caught.value.status == 409 + + # ── (a)/(6) billing and authorization copy must be actionable ───────────────── def test_lapsed_subscription_is_not_reported_as_an_outage(monkeypatch, _upgrade_url): """402 is the control plane's "no active paid entitlement".