diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index bd09403..c99ea90 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -169,7 +169,7 @@ jobs: macos-pkg-smoke: name: macOS existing PKG smoke - runs-on: macos-14 + runs-on: macos-15-intel steps: - uses: actions/checkout@v4 - uses: actions/setup-python@v5 diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 3a87896..ee5edf0 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -26,7 +26,7 @@ jobs: matrix: include: - name: macos-universal2 - os: macos-14 + os: macos-15-intel macos_arch: universal2 - name: windows-x64 os: windows-latest diff --git a/README.md b/README.md index f03421f..d0f1266 100644 --- a/README.md +++ b/README.md @@ -97,6 +97,18 @@ Set the API base URL if you are not using the default: dataplicity config set --base-url https://gateway.dataplicity.com ``` +Use HTTPS for remote API servers; HTTP is supported for loopback development. +Changing the saved server clears its previous authentication. A one-command +`--base-url` override uses no credentials saved for a different server; sign in +to the selected server before using protected commands. An override alone does +not overwrite the original configuration file. + +Commands retain the sign-in they started with. If another process signs in or +out while a command is waiting, the older command stops using that session and +asks you to start again. Configuration updates use a private atomic file and a +short process lock; a delayed renewal cannot overwrite a newer sign-in. A lost +renewal response can be retried with the saved recovery identifier. + Login with email/password: ``` diff --git a/dataplicity_cli/api.py b/dataplicity_cli/api.py index 4cb3884..6bafc3e 100644 --- a/dataplicity_cli/api.py +++ b/dataplicity_cli/api.py @@ -3,6 +3,7 @@ import asyncio import base64 import json +import math import threading import time from dataclasses import dataclass @@ -11,8 +12,9 @@ import requests from requests.structures import CaseInsensitiveDict -from .config import Config +from .config import Config, ConfigError, SessionOwner, valid_api_credential from .client_identity import identity_headers +from .api_recipient import INVALID_RECIPIENT, recipient_identity, recipient_url # Refresh before the access JWT expires so long-running commands (tunnels, # terminals, watches) do not wait for a 401 mid-flight. @@ -21,6 +23,15 @@ ACCESS_TOKEN_KEEPALIVE_INTERVAL_SECONDS = 120.0 +class _ExplicitApiAuth(requests.auth.AuthBase): + """Keep netrc/session Basic auth from replacing explicitly selected proof.""" + def __call__(self, request): + return request + + +_EXPLICIT_API_AUTH = _ExplicitApiAuth() + + @dataclass class ApiResponse: ok: bool @@ -29,6 +40,12 @@ class ApiResponse: text: str +@dataclass(frozen=True) +class _RefreshOutcome: + ok: bool + owner: Optional[SessionOwner] + + class ApiClient: def __init__( self, @@ -37,6 +54,8 @@ def __init__( on_config_reload: Optional[Callable[[], None]] = None, ) -> None: self.config = config + self._owner = config.capture_owner() + self._configured_base_url = config.base_url self.session = requests.Session() self.session.headers.update(identity_headers(self.config.install_id)) self._on_token_update = on_token_update @@ -44,17 +63,31 @@ def __init__( self._refresh_lock = threading.Lock() def _build_url(self, path: str) -> str: - if path.startswith("http://") or path.startswith("https://"): - return path - base = (self.config.base_url or "").rstrip("/") - return f"{base}/{path.lstrip('/')}" - - def _auth_headers(self) -> Dict[str, str]: - if self.config.auth_method == "jwt" and self.config.access_token: - return {"Authorization": f"Bearer {self.config.access_token}"} - if self.config.auth_method == "api_key" and self.config.api_key: - return {"Authorization": f"ApiKey {self.config.api_key}"} - return {} + if recipient_identity(self.config.base_url) != recipient_identity(self._configured_base_url): + raise ValueError(INVALID_RECIPIENT) + return recipient_url(self._configured_base_url, path) + + def _assert_actor(self, owner): + if owner != self._owner: + raise ConfigError("The active account changed.") + self.config.assert_owner(owner) + + def _auth_headers(self, owner=None) -> Dict[str, str]: + if owner is not None: + self._assert_actor(owner) + self.config.reload_owned(owner) + with self.config._mutex: + if owner is not None: + self._assert_actor(owner) + if self.config.auth_method == "jwt" and self.config.access_token: + if not valid_api_credential(self.config.access_token): + raise ConfigError("Invalid authentication credentials. Sign in again.") + return {"Authorization": f"Bearer {self.config.access_token}"} + if self.config.auth_method == "api_key" and self.config.api_key: + if not valid_api_credential(self.config.api_key): + raise ConfigError("Invalid authentication credentials. Sign in again.") + return {"Authorization": f"ApiKey {self.config.api_key}"} + return {} @staticmethod def _extract_error_message(response: requests.Response) -> str: @@ -92,18 +125,21 @@ def _looks_like_invalid_token_message(message: str) -> bool: @staticmethod def _jwt_expiry(token: Optional[str]) -> Optional[float]: - if not token or token.count(".") < 2: + if not isinstance(token, str) or token.count(".") < 2: return None try: payload_b64 = token.split(".")[1] payload_b64 += "=" * (-len(payload_b64) % 4) payload = json.loads(base64.urlsafe_b64decode(payload_b64.encode("ascii"))) + if not isinstance(payload, dict): + return None + exp = payload.get("exp") + if isinstance(exp, bool) or not isinstance(exp, (int, float)): + return None + expiry = float(exp) + return expiry if math.isfinite(expiry) else None except Exception: return None - exp = payload.get("exp") - if isinstance(exp, bool) or not isinstance(exp, (int, float)): - return None - return float(exp) def access_token_seconds_remaining(self) -> Optional[float]: exp = self._jwt_expiry(self.config.access_token) @@ -114,88 +150,136 @@ def access_token_seconds_remaining(self) -> Optional[float]: def _response_indicates_invalid_jwt(self, response: requests.Response) -> bool: if response.status_code not in {400, 401, 403}: return False + try: + payload = response.json() + except Exception: + payload = None + if isinstance(payload, dict): + code = payload.get("code") or payload.get("error") + if isinstance(payload.get("detail"), dict): + code = code or payload["detail"].get("code") + if isinstance(code, str) and code in {"session_expired", "token_not_valid", "token_expired", "invalid_token", "token_blacklisted"}: + return True return self._looks_like_invalid_token_message(self._extract_error_message(response)) - def _invalidate_jwt_session(self) -> None: - self.config.clear_tokens() + def _notify_update(self, owner): if self._on_token_update is not None: try: self._on_token_update() + except ConfigError: + raise except Exception: + # Optional notifications cannot break a committed owned update. pass + self.config.assert_owner(owner) - def _reload_tokens(self) -> bool: - if self._on_config_reload is None: - return False - before = (self.config.access_token, self.config.refresh_token, self.config.auth_method) + def _invalidate_jwt_session(self, owner=None, expected_refresh=None): + owner = owner or self._owner + self._assert_actor(owner) + current = self.config.invalidate_owned(owner, expected_refresh) + self._notify_update(current) + return current + + def _reload_tokens(self, owner=None) -> bool: + owner = owner or self._owner try: - self._on_config_reload() + self._assert_actor(owner) + before = self.config._credential_state() + self.config.reload_owned(owner) + if self._on_config_reload is not None and before == self.config._credential_state(): + self._on_config_reload() + self._assert_actor(owner) + return before != self.config._credential_state() except Exception: return False - after = (self.config.access_token, self.config.refresh_token, self.config.auth_method) - return after != before - def _attempt_refresh(self) -> str: - """Try one refresh call. Returns 'ok', 'invalid', or 'failed'.""" - if not self.config.refresh_token: - return "failed" - url = self._build_url("/api/token/refresh/") + def _perform_refresh(self, owner): + ticket = None try: - resp = self.session.post(url, json={"refresh": self.config.refresh_token}, timeout=10, allow_redirects=False) - except Exception: - return "failed" + self._assert_actor(owner) + url = self._build_url("/api/token/refresh/") + ticket = self.config.claim_refresh(owner) + if ticket is None: + return "no_refresh", None + # Recovery ID is committed before HTTP and shared only by this + # session/input. File locks are released before any network I/O. + resp = self.session.post(url, json={"refresh": ticket.refresh, "request_id": ticket.request_id}, + timeout=10, allow_redirects=False, auth=_EXPLICIT_API_AUTH) + self._assert_actor(owner) + except ConfigError: + return "changed", ticket + except (ValueError, requests.RequestException): + try: + self._assert_actor(owner) + except ConfigError: + return "changed", ticket + return "failed", ticket if resp.status_code >= 400: - return "invalid" if self._response_indicates_invalid_jwt(resp) else "failed" + return ("invalid" if self._response_indicates_invalid_jwt(resp) else "failed"), ticket if not 200 <= resp.status_code < 300: - return "failed" + return "failed", ticket try: payload = resp.json() except Exception: - return "failed" - access = payload.get("access") - if not access: - return "failed" - self.config.access_token = access - refresh = payload.get("refresh") - if refresh: - self.config.refresh_token = refresh - self.config.auth_method = "jwt" - if self._on_token_update is not None: + return "failed", ticket + if not isinstance(payload, dict): + return "failed", ticket + access, refresh = payload.get("access"), payload.get("refresh") + if not valid_api_credential(access) or (refresh is not None and not valid_api_credential(refresh)): + return "failed", ticket + try: + self._assert_actor(owner) + self.config.commit_rotation(ticket, access, refresh) + self._notify_update(owner) + return "ok", ticket + except ConfigError: + return "changed", ticket + + def _attempt_refresh(self) -> str: + """Compatibility helper: one owned attempt, without session invalidation.""" + state, _ = self._perform_refresh(self._owner) + return "failed" if state in ("changed", "no_refresh") else state + + def _renew(self, owner): + with self._refresh_lock: try: - self._on_token_update() - except Exception: - pass - return "ok" + self._assert_actor(owner) + state, ticket = self._perform_refresh(owner) + if state == "ok": + return _RefreshOutcome(True, owner) + if state == "changed": + return _RefreshOutcome(False, None) + if state == "invalid" and self._reload_tokens(owner): + state, ticket = self._perform_refresh(owner) + if state == "ok": + return _RefreshOutcome(True, owner) + self._assert_actor(owner) + if state in ("invalid", "no_refresh"): + current = self._invalidate_jwt_session(owner, ticket.refresh if ticket else None) + # A later owned rotation may win between the invalid reply + # and conditional revocation; keep and retry that successor. + return _RefreshOutcome(current == owner, current) + return _RefreshOutcome(False, owner) + except ConfigError: + return _RefreshOutcome(False, None) def _refresh_access_token(self) -> bool: - with self._refresh_lock: - result = self._attempt_refresh() - if result == "ok": - return True - # Another CLI process may have rotated the refresh token on disk. - if result == "invalid" and self._reload_tokens(): - result = self._attempt_refresh() - if result == "ok": - return True - if result == "invalid": - self._invalidate_jwt_session() - return False + return self._renew(self._owner).ok def ensure_fresh_access_token(self, *, skew_seconds: float = ACCESS_TOKEN_REFRESH_SKEW_SECONDS) -> bool: - """Refresh the JWT access token when it is expired or near expiry. - - Long-running commands rely on this so heartbeats and other control - calls do not wait until the server returns 401. - """ - if self.config.auth_method != "jwt" or not self.config.refresh_token: - return False - remaining = self.access_token_seconds_remaining() - if remaining is None: - # Opaque/unparseable tokens: keep reactive 401 refresh only. + try: + self._assert_actor(self._owner) + self.config.reload_owned(self._owner) + if self.config.auth_method != "jwt" or not self.config.refresh_token: + return False + remaining = self.access_token_seconds_remaining() + if remaining is None: + return False + if remaining > skew_seconds: + return True + return self._refresh_access_token() + except ConfigError: return False - if remaining > skew_seconds: - return True - return self._refresh_access_token() def refresh_session(self) -> bool: if self.config.auth_method != "jwt": @@ -233,58 +317,74 @@ def request( ) -> ApiResponse: supplied_headers = CaseInsensitiveDict(headers or {}) owns_session_auth = "Authorization" not in supplied_headers - if allow_refresh and owns_session_auth and self.config.auth_method == "jwt" and self.config.refresh_token: - self.ensure_fresh_access_token() - - url = self._build_url(path) - req_headers = CaseInsensitiveDict({ - **identity_headers(self.config.install_id), - "Accept": "application/json", - **self._auth_headers(), - }) - req_headers.update(supplied_headers) + owner = self._owner if owns_session_auth else None + publication_owner = owner + changed = ApiResponse(False, 0, None, "The saved account changed. Start this command again.") try: - resp = self.session.request( - method.upper(), - url, - params=params, - json=json_data, - data=data, - headers=req_headers, - timeout=timeout, - allow_redirects=False, - ) - except requests.RequestException as exc: - return ApiResponse(False, 0, None, str(exc)) - - should_retry_with_refresh = ( - allow_refresh - and owns_session_auth - and self.config.auth_method == "jwt" - and (resp.status_code == 401 or self._response_indicates_invalid_jwt(resp)) - ) - if should_retry_with_refresh: - if self._refresh_access_token(): - return self.request( - method, - path, - params=params, - json_data=json_data, - data=data, - headers=headers, - timeout=timeout, - allow_refresh=False, - ) - if (owns_session_auth and self.config.auth_method == "jwt" - and req_headers.get("Authorization") == f"Bearer {self.config.access_token}" - and self._response_indicates_invalid_jwt(resp)): - self._invalidate_jwt_session() - - text = resp.text or "" + url = self._build_url(path) + except ValueError: + return ApiResponse(False, 0, None, INVALID_RECIPIENT) try: - payload = resp.json() - except Exception: - payload = None + if owner is not None: + self._assert_actor(owner) + self.config.reload_owned(owner) + if allow_refresh and self.config.auth_method == "jwt" and self.config.refresh_token: + self.ensure_fresh_access_token() + self._assert_actor(owner) + req_headers = CaseInsensitiveDict({**identity_headers(self.config.install_id), "Accept": "application/json", + **self._auth_headers(owner)}) + req_headers.update(supplied_headers) + # Verify the recipient again after admission/renewal callbacks. + self._build_url(path) + resp = self.session.request(method.upper(), url, params=params, json=json_data, data=data, + headers=req_headers, timeout=timeout, allow_redirects=False, auth=_EXPLICIT_API_AUTH) + if owner is not None: + self._assert_actor(owner) + except ConfigError: + return changed + except ValueError: + return ApiResponse(False, 0, None, INVALID_RECIPIENT) + except requests.RequestException: + if owner is not None: + try: + self._assert_actor(owner) + except ConfigError: + return changed + return ApiResponse(False, 0, None, "The API request failed. Check your connection or sign in again.") + + auth_failure = resp.status_code == 401 or self._response_indicates_invalid_jwt(resp) + tried_refresh = False + try: + if owner is not None and allow_refresh and self.config.auth_method == "jwt" and auth_failure: + self.config.reload_owned(owner) + current_auth = self._auth_headers(owner).get("Authorization") + if current_auth and current_auth != req_headers.get("Authorization"): + return self.request(method, path, params=params, json_data=json_data, data=data, + headers=headers, timeout=timeout, allow_refresh=False) + tried_refresh = True + renewed = self._renew(owner) + if renewed.ok: + self._assert_actor(owner) + return self.request(method, path, params=params, json_data=json_data, data=data, + headers=headers, timeout=timeout, allow_refresh=False) + if renewed.owner is None: + return changed + publication_owner = renewed.owner + if (owner is not None and not tried_refresh and self.config.auth_method == "jwt" + and req_headers.get("Authorization") == self._auth_headers(owner).get("Authorization") + and self._response_indicates_invalid_jwt(resp)): + publication_owner = self._invalidate_jwt_session(owner) + if publication_owner is not None: + self.config.assert_owner(publication_owner) + text = resp.text or "" + try: + payload = resp.json() + except Exception: + payload = None + if publication_owner is not None: + self.config.assert_owner(publication_owner) + except ConfigError: + return changed ok = 200 <= resp.status_code < 300 if not ok and not text: text = f"HTTP {resp.status_code}" diff --git a/dataplicity_cli/api_recipient.py b/dataplicity_cli/api_recipient.py new file mode 100644 index 0000000..eb513b2 --- /dev/null +++ b/dataplicity_cli/api_recipient.py @@ -0,0 +1,60 @@ +"""Canonical recipients for credentialed CLI API transports.""" +from ipaddress import ip_address +from typing import Tuple +from urllib.parse import unquote, urlsplit + +import requests + + +INVALID_RECIPIENT = "The API destination is invalid or outside the configured recipient." + + +def canonical_url(value: str) -> str: + try: + if not isinstance(value, str) or not value or any(ord(char) < 32 or ord(char) == 127 for char in value): + raise ValueError + parsed = urlsplit(value) + if parsed.username is not None or parsed.password is not None or parsed.fragment or not parsed.hostname: + raise ValueError + loopback = parsed.hostname.lower() == "localhost" + if not loopback: + try: + loopback = ip_address(parsed.hostname).is_loopback + except ValueError: + pass + if parsed.scheme != "https" and not (parsed.scheme == "http" and loopback): + raise ValueError + # Use the same URL normalization as the actual requests transport. + prepared = requests.Request("GET", value).prepare().url + path = unquote(urlsplit(prepared).path) + if "\\" in path or any(part in (".", "..") for part in path.split("/")): + raise ValueError + return prepared + except (ValueError, requests.RequestException): + raise ValueError(INVALID_RECIPIENT) from None + + +def recipient_identity(value: str) -> Tuple[str, str, int, str]: + parsed = urlsplit(canonical_url(value)) + if parsed.query: + raise ValueError(INVALID_RECIPIENT) + return parsed.scheme, parsed.hostname.lower(), parsed.port or (443 if parsed.scheme == "https" else 80), unquote(parsed.path).rstrip("/") + + +def recipient_url(base: str, path: str) -> str: + try: + identity = recipient_identity(base) + if not isinstance(path, str) or path.startswith("//"): + raise ValueError + parsed = urlsplit(path) + candidate = path if parsed.scheme or parsed.netloc else f"{base.rstrip('/')}/{path.lstrip('/')}" + target = canonical_url(candidate) + parts = urlsplit(target) + origin = (parts.scheme, parts.hostname.lower(), parts.port or (443 if parts.scheme == "https" else 80)) + target_path = unquote(parts.path) + prefix = identity[3] + if origin != identity[:3] or (prefix and target_path != prefix and not target_path.startswith(prefix + "/")): + raise ValueError + return target + except (ValueError, requests.RequestException): + raise ValueError(INVALID_RECIPIENT) from None diff --git a/dataplicity_cli/cli.py b/dataplicity_cli/cli.py index 152c3a7..0223959 100644 --- a/dataplicity_cli/cli.py +++ b/dataplicity_cli/cli.py @@ -23,6 +23,7 @@ from urllib.parse import parse_qs, urlparse import typer +from typer.core import TyperGroup from rich.console import Console from rich.live import Live from rich.markup import escape @@ -35,12 +36,23 @@ from . import __version__ from .api import ApiClient, ApiResponse from .client_identity import identity_headers -from .config import Config, default_config_path +from .config import Config, ConfigError, default_config_path from .native_auth import NativeLogin, identity_origin from .m2m import M2MClient +class _SafeConfigGroup(TyperGroup): + def invoke(self, ctx): + try: + return super().invoke(ctx) + except ConfigError as exc: + typer.echo(f"Error: {exc}", err=True) + raise typer.Exit(code=1) from None + + app = typer.Typer( + cls=_SafeConfigGroup, + pretty_exceptions_show_locals=False, add_completion=True, invoke_without_command=True, help=( @@ -104,12 +116,14 @@ def _ctx(ctx: typer.Context) -> AppContext: @asynccontextmanager -async def _jwt_keepalive(state: AppContext) -> AsyncIterator[None]: - """Keep SSO/JWT access tokens fresh for the lifetime of a long command.""" - if state.config.auth_method != "jwt" or not state.config.refresh_token: +async def _jwt_keepalive(state: AppContext, api: Optional[ApiClient] = None) -> AsyncIterator[None]: + """Renew only the credential realm actually running the long command.""" + selected = api if api is not None else state.api + selected_config = selected.config if api is not None else state.config + if selected_config.auth_method != "jwt" or not selected_config.refresh_token: yield return - task = asyncio.create_task(state.api.maintain_fresh_access_token()) + task = asyncio.create_task(selected.maintain_fresh_access_token()) try: yield finally: @@ -137,10 +151,6 @@ def _tunnel_control(state: AppContext, organisation: Optional[str], token: Optio raise TunnelError("Select an organisation with --org .") organisation = str(rows[0].get("hash_id") or "") api = state.api - if token: - # Machine credential failure must never refresh or invalidate a human - # session saved in the same CLI configuration. - api = ApiClient(Config(base_url=state.config.base_url, install_id=state.config.install_id)) return TunnelAPI(api, organisation, token=token) @@ -186,7 +196,7 @@ async def refresh() -> None: live.update(session.stats.render(session.control.organisation, state.console.width, state.console.height), refresh=True) await asyncio.sleep(1) try: - async with _jwt_keepalive(state): + async with _jwt_keepalive(state, session.control.api if session is not None else None): with live if live is not None else nullcontext(): refresh_task = asyncio.create_task(refresh()) if live is not None else None try: @@ -784,10 +794,9 @@ def _apply_tokens_or_none(state: AppContext, payload: Any) -> bool: access, refresh = _extract_sso_tokens(payload) if not access: return False - state.config.access_token = access - state.config.refresh_token = refresh - state.config.auth_method = "jwt" + state.config.install_user_tokens(access, refresh) state.config.save(state.config_path) + state.api = ApiClient(state.config) return True @@ -1847,6 +1856,23 @@ def _normalize_timeline_type_filters(raw_value: Optional[str]) -> List[str]: return list(dict.fromkeys(cleaned)) +def _set_api_recipient(config: Config, base_url: str) -> None: + from .api_recipient import recipient_identity + selected = base_url.rstrip("/") + try: + selected_identity = recipient_identity(selected) + except ValueError as exc: + raise typer.BadParameter(str(exc), param_hint="--base-url") from None + try: + current_identity = recipient_identity(config.base_url) + except ValueError: + current_identity = None + if selected_identity != current_identity: + config.clear_tokens() + config.clear_api_key() + config.base_url = selected + + @app.callback() def main( ctx: typer.Context, @@ -1869,24 +1895,10 @@ def main( with suppress(OSError): config.save(path) if base_url: - config.base_url = base_url.rstrip("/") + _set_api_recipient(config, base_url) console = Console(theme=THEME) - def _persist_tokens() -> None: - config.save(path) - - def _reload_tokens() -> None: - # Pick up refresh-token rotations written by another CLI process so a - # long-running command (e.g. tunnel publish) can keep refreshing. - fresh = Config.load(path) - config.auth_method = fresh.auth_method - config.access_token = fresh.access_token - config.refresh_token = fresh.refresh_token - config.api_key = fresh.api_key - - api_client = ApiClient(config, on_token_update=_persist_tokens, on_config_reload=_reload_tokens) - if config.auth_method == "jwt" and config.refresh_token: - api_client.ensure_fresh_access_token() + api_client = ApiClient(config) ctx.obj = AppContext( config=config, config_path=path, @@ -2125,7 +2137,7 @@ def config_set( """ state = _ctx(ctx) if base_url: - state.config.base_url = base_url.rstrip("/") + _set_api_recipient(state.config, base_url) state.config.save(state.config_path) if state.json_output: _print_json({"ok": True, "base_url": state.config.base_url}) @@ -2233,7 +2245,8 @@ def auth_login( _print_json({"ok": False, "detail": "Password is required in --json mode. Pass `--password`."}) raise typer.Exit(code=2) password = typer.prompt("Password", hide_input=True) - bootstrap = state.api.post("/api/auth/bootstrap/", json_data={"email": email}) + primary_api = ApiClient(Config(base_url=state.config.base_url, install_id=state.config.install_id)) + bootstrap = primary_api.post("/api/auth/bootstrap/", json_data={"email": email}) if bootstrap.ok and isinstance(bootstrap.data, dict) and bootstrap.data.get("status") == "sso_redirect": if state.json_output: message = "SSO is required for this account. Use `dataplicity auth sso --email ...`." @@ -2260,7 +2273,7 @@ def auth_login( _fail_login(state, _webauthn_unsupported_message(), extra={"error_code": "unsupported_mfa_type"}) payload = _password_login_payload(email, password, mfa_code=mfa_code, mfa_type=mfa_type) - response = state.api.post("/api/token/", json_data=payload) + response = primary_api.post("/api/token/", json_data=payload) if not response.ok and _extract_error_code(response.data) == "mfa_required": mfa = _extract_mfa_payload(response.data) available = _mfa_available_types(mfa) @@ -2293,7 +2306,7 @@ def auth_login( if has_totp: mfa_code = typer.prompt("Authenticator app code") payload = _password_login_payload(email, password, mfa_code=mfa_code, mfa_type="TOTP") - response = state.api.post("/api/token/", json_data=payload) + response = primary_api.post("/api/token/", json_data=payload) else: _fail_login(state, _webauthn_unsupported_message(), extra=extra) @@ -2317,12 +2330,11 @@ def auth_login( if not access: _fail_login(state, "Login succeeded but no access token was returned.") - state.config.access_token = access - state.config.refresh_token = refresh - state.config.auth_method = "jwt" + state.config.install_user_tokens(access, refresh) state.config.last_email = email state.config.preferred_login_method = "email-password" state.config.save(state.config_path) + state.api = ApiClient(state.config) if state.json_output: _print_json({"ok": True, "detail": "Logged in"}) else: @@ -2364,10 +2376,10 @@ def auth_api_key_set(ctx: typer.Context) -> None: """ state = _ctx(ctx) api_key = typer.prompt("API key", hide_input=True) - state.config.api_key = api_key.strip() - state.config.auth_method = "api_key" + state.config.install_api_key(api_key.strip()) state.config.preferred_login_method = "api-key" state.config.save(state.config_path) + state.api = ApiClient(state.config) if state.json_output: _print_json({"ok": True, "detail": "API key saved"}) else: diff --git a/dataplicity_cli/config.py b/dataplicity_cli/config.py index f66998c..9d2511b 100644 --- a/dataplicity_cli/config.py +++ b/dataplicity_cli/config.py @@ -1,11 +1,18 @@ from __future__ import annotations import json +import copy +import hashlib +import threading import os import uuid -from dataclasses import dataclass +from dataclasses import dataclass, field from pathlib import Path from typing import Any, Dict, Optional +from contextlib import contextmanager + +from .api_recipient import recipient_identity +from .config_store import ConfigError, ConfigChangedError, atomic_write, locked_config try: from platformdirs import user_config_dir @@ -36,71 +43,289 @@ def default_config_path() -> Path: return _config_root() / "cli.json" +@dataclass(frozen=True) +class SessionOwner: + identity: str + path: Optional[Path] + method: Optional[str] + recipient: tuple + + +@dataclass(frozen=True) +class RefreshTicket: + owner: SessionOwner + access: Optional[str] + refresh: str = field(repr=False) + request_id: str + + +_AUTH_FIELDS = {"base_url", "auth_method", "access_token", "refresh_token", "api_key"} +_SECRET_FIELDS = {"auth_method", "access_token", "refresh_token", "api_key"} + + +def valid_api_credential(value): + return isinstance(value, str) and bool(value) and value.isascii() and all(33 <= ord(char) <= 126 for char in value) + + +def _valid_uuid(value): + try: + return isinstance(value, str) and str(uuid.UUID(value)) == value + except ValueError: + return False + + @dataclass class Config: base_url: str = DEFAULT_BASE_URL - auth_method: Optional[str] = None # "jwt" | "api_key" - access_token: Optional[str] = None - refresh_token: Optional[str] = None - api_key: Optional[str] = None + auth_method: Optional[str] = None + access_token: Optional[str] = field(default=None, repr=False) + refresh_token: Optional[str] = field(default=None, repr=False) + api_key: Optional[str] = field(default=None, repr=False) last_email: Optional[str] = None - preferred_login_method: Optional[str] = None # "email-password" | "sso" | "api-key" + preferred_login_method: Optional[str] = None install_id: Optional[str] = None + local_session_id: str = field(default_factory=lambda: str(uuid.uuid4())) + refresh_attempt: Optional[dict] = field(default=None, repr=False) + _path: Optional[Path] = field(default=None, init=False, repr=False, compare=False) + _baseline: Optional[dict] = field(default=None, init=False, repr=False, compare=False) + _baseline_id: Optional[str] = field(default=None, init=False, repr=False, compare=False) + _observed: tuple = field(default=(), init=False, repr=False, compare=False) + _extra: dict = field(default_factory=dict, init=False, repr=False, compare=False) + _mutex: Any = field(default_factory=threading.RLock, init=False, repr=False, compare=False) + + def __post_init__(self): + self._observed = self._credential_state() + + def _recipient(self): + try: + return recipient_identity(self.base_url) + except ValueError: + return ("invalid",) + + def _credential_state(self): + return self._recipient(), self.auth_method, self.access_token, self.refresh_token, self.api_key + + def _observe(self): + current = self._credential_state() + if current != self._observed or not _valid_uuid(self.local_session_id): + self.local_session_id = str(uuid.uuid4()) + self.refresh_attempt = None + self._observed = current @classmethod - def load(cls, path: Path) -> "Config": - if not path.exists(): - return cls() + def _read(cls, path): + try: + raw_bytes = path.read_bytes() + except FileNotFoundError: + return None + except OSError: + raise ConfigError("Unable to read the configuration.") from None try: - raw = json.loads(path.read_text(encoding="utf-8")) - except Exception: - return cls() - base_url = raw.get("base_url") or DEFAULT_BASE_URL - if base_url in LEGACY_DEFAULT_BASE_URLS: - # Seamlessly move older installs onto the public gateway host. - base_url = DEFAULT_BASE_URL - return cls( - base_url=base_url, - auth_method=raw.get("auth_method"), - access_token=raw.get("access_token"), - refresh_token=raw.get("refresh_token"), - api_key=raw.get("api_key"), - last_email=raw.get("last_email"), - preferred_login_method=raw.get("preferred_login_method"), - install_id=(str(raw.get("install_id") or "").strip() or None), - ) + raw = json.loads(raw_bytes) + if not isinstance(raw, dict): + raw = {} + except (ValueError, UnicodeError): + raw = {} + base = raw.get("base_url") or DEFAULT_BASE_URL + if base in LEGACY_DEFAULT_BASE_URLS: + base = DEFAULT_BASE_URL + identity = raw.get("local_session_id") + if not _valid_uuid(identity): + # Legacy readers only share an owner while the exact file is + # unchanged. Never infer rotation/account lineage from JWT claims. + identity = str(uuid.uuid5(uuid.NAMESPACE_URL, "dataplicity-cli-legacy:" + hashlib.sha256(raw_bytes).hexdigest())) + config = cls(base_url=base, auth_method=raw.get("auth_method"), + access_token=raw.get("access_token"), refresh_token=raw.get("refresh_token"), api_key=raw.get("api_key"), + last_email=raw.get("last_email"), preferred_login_method=raw.get("preferred_login_method"), + install_id=str(raw.get("install_id") or "").strip() or None, + local_session_id=identity, refresh_attempt=raw.get("refresh_attempt")) + known = set(config.to_dict()) + config._extra = {key: value for key, value in raw.items() if key not in known} + config._mark_saved(path) + return config + + @classmethod + def load(cls, path: Path) -> "Config": + path = Path(path).resolve() + config = cls._read(path) + if config is None: + config = cls() + config._path = path + config._baseline = config.to_dict() + return config def to_dict(self) -> Dict[str, Any]: - return { - "base_url": self.base_url, - "auth_method": self.auth_method, - "access_token": self.access_token, - "refresh_token": self.refresh_token, - "api_key": self.api_key, - "last_email": self.last_email, - "preferred_login_method": self.preferred_login_method, - "install_id": self.install_id, - } + return copy.deepcopy({**self._extra, "base_url": self.base_url, "auth_method": self.auth_method, + "access_token": self.access_token, "refresh_token": self.refresh_token, "api_key": self.api_key, + "last_email": self.last_email, "preferred_login_method": self.preferred_login_method, + "install_id": self.install_id, "local_session_id": self.local_session_id, + "refresh_attempt": self.refresh_attempt}) + + def _mark_saved(self, path): + self._path = path + self._baseline = self.to_dict() + self._baseline_id = self.local_session_id + self._observed = self._credential_state() + + def _adopt(self, record): + for key in ("base_url", "auth_method", "access_token", "refresh_token", "api_key", "last_email", + "preferred_login_method", "install_id", "local_session_id", "refresh_attempt"): + setattr(self, key, copy.deepcopy(getattr(record, key))) + self._extra = copy.deepcopy(record._extra) + self._path = record._path + self._baseline = copy.deepcopy(record._baseline) + self._baseline_id = record._baseline_id + self._observed = self._credential_state() def save(self, path: Path) -> None: - path.parent.mkdir(parents=True, exist_ok=True) - payload = json.dumps(self.to_dict(), indent=2, sort_keys=True).encode("utf-8") - fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) - try: - os.write(fd, payload) - finally: - os.close(fd) + path = Path(path).resolve() + with self._mutex, locked_config(path): + self._observe() + current = self._read(path) + expected = self._baseline_id if self._path == path else None + if (current.local_session_id if current else None) != expected: + raise ConfigChangedError() + local = self.to_dict() + if current is None: + payload = local + else: + baseline = self._baseline or {} + payload = current.to_dict() + replacing = self.local_session_id != expected + changed = {key for key, value in local.items() if value != baseline.get(key)} + if replacing: + for key in _AUTH_FIELDS | {"local_session_id", "refresh_attempt"}: + payload[key] = local[key] + for key in changed - (_AUTH_FIELDS | {"local_session_id", "refresh_attempt"} if replacing else _SECRET_FIELDS | {"local_session_id"}): + if not replacing and payload.get(key) not in (baseline.get(key), local[key]): + raise ConfigChangedError() + payload[key] = local[key] + atomic_write(path, payload) + self._adopt(self._read(path)) + + def capture_owner(self): + with self._mutex: + self._observe() + return SessionOwner(self.local_session_id, self._path, self.auth_method, self._recipient()) + + def _check_local(self, owner): + if self.capture_owner() != owner: + raise ConfigChangedError() + + @contextmanager + def _owned_record(self, owner): + with self._mutex: + self._check_local(owner) + if self._path is None: + yield self + else: + with locked_config(self._path): + record = self._read(self._path) + if record is None and self._baseline_id is None: + yield self + elif record is None or record.capture_owner() != owner: + raise ConfigChangedError() + else: + yield record + + def assert_owner(self, owner): + with self._owned_record(owner): + pass + + def _persist_owned(self, record): + if self._path is not None: + atomic_write(self._path, record.to_dict()) + record._mark_saved(self._path) + record._observed = record._credential_state() + self._adopt(record) + + def reload_owned(self, owner): + with self._owned_record(owner) as record: + before = self._credential_state() + self._adopt(record) + return before != self._credential_state() + + def claim_refresh(self, owner): + with self._owned_record(owner) as record: + if record.auth_method != "jwt" or not isinstance(record.refresh_token, str) or not record.refresh_token: + return None + if not valid_api_credential(record.refresh_token): + raise ConfigError("Invalid authentication credentials. Sign in again.") + digest = hashlib.sha256(record.refresh_token.encode()).hexdigest() + attempt = record.refresh_attempt + if not isinstance(attempt, dict) or attempt.get("input_sha256") != digest or not _valid_uuid(attempt.get("request_id")): + attempt = {"input_sha256": digest, "request_id": str(uuid.uuid4())} + record.refresh_attempt = attempt + self._persist_owned(record) + else: + self._adopt(record) + return RefreshTicket(owner, record.access_token, record.refresh_token, attempt["request_id"]) + + def commit_rotation(self, ticket, access, refresh): + if not valid_api_credential(access) or (refresh is not None and not valid_api_credential(refresh)): + raise ConfigError("Session renewal returned invalid credentials. Sign in again.") + with self._owned_record(ticket.owner) as record: + if record.refresh_token == ticket.refresh: + record.access_token = access + record.refresh_token = refresh if refresh is not None else ticket.refresh + record.auth_method = "jwt" + record.refresh_attempt = None + self._persist_owned(record) + else: + # A later legitimate rotation already won; never overwrite it + # or clear its independently pending recovery receipt. + self._adopt(record) + + def invalidate_owned(self, owner, expected_refresh=None): + with self._owned_record(owner) as record: + if expected_refresh is not None and record.refresh_token != expected_refresh: + self._adopt(record) + return self.capture_owner() + record.clear_tokens() + self._persist_owned(record) + return self.capture_owner() + + def _new_session(self): + self.local_session_id = str(uuid.uuid4()) + self.refresh_attempt = None + self._observed = self._credential_state() + + def install_user_tokens(self, access, refresh): + if not valid_api_credential(access) or (refresh is not None and not valid_api_credential(refresh)): + raise ConfigError("Sign-in returned invalid credentials. Restart sign-in.") + with self._mutex: + self.auth_method = "jwt" + self.access_token = access + self.refresh_token = refresh + self.api_key = None + self._new_session() + + def install_api_key(self, key): + if not valid_api_credential(key): + raise ConfigError("The API key is invalid. Enter it again.") + with self._mutex: + self.auth_method = "api_key" + self.api_key = key + self.access_token = None + self.refresh_token = None + self._new_session() def clear_tokens(self) -> None: - self.access_token = None - self.refresh_token = None - if self.auth_method == "jwt": - self.auth_method = None + with self._mutex: + self.access_token = None + self.refresh_token = None + if self.auth_method == "jwt": + self.auth_method = None + self._new_session() def clear_api_key(self) -> None: - self.api_key = None - if self.auth_method == "api_key": - self.auth_method = None + with self._mutex: + changed = self.api_key is not None or self.auth_method == "api_key" + self.api_key = None + if self.auth_method == "api_key": + self.auth_method = None + if changed: + self._new_session() def ensure_install_id(self) -> bool: current = (self.install_id or "").strip() diff --git a/dataplicity_cli/config_store.py b/dataplicity_cli/config_store.py new file mode 100644 index 0000000..04f8112 --- /dev/null +++ b/dataplicity_cli/config_store.py @@ -0,0 +1,93 @@ +"""Short, cross-process transactions and atomic private configuration files.""" +import json +import os +import stat +import sys +import tempfile +import threading +import time +from contextlib import contextmanager +from pathlib import Path + + +class ConfigError(OSError): + pass + + +class ConfigChangedError(ConfigError): + def __init__(self): + super().__init__("The saved account or configuration changed. Start this command again.") + + +WINDOWS = sys.platform == "win32" +_thread_lock = threading.RLock() + + +def _lock(fd, *, release=False): + if WINDOWS: + import msvcrt + os.lseek(fd, 0, os.SEEK_SET) + msvcrt.locking(fd, msvcrt.LK_UNLCK if release else msvcrt.LK_NBLCK, 1) + else: + import fcntl + fcntl.flock(fd, fcntl.LOCK_UN if release else fcntl.LOCK_EX | fcntl.LOCK_NB) + + +@contextmanager +def locked_config(path: Path, timeout=5.0): + # Keep the lock inode after release; unlinking it lets waiters lock different + # files. No API/provider work may run inside this context. + with _thread_lock: + path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + lock_path = path.with_name(path.name + ".lock") + if lock_path.is_symlink(): + raise ConfigError("The configuration lock is unavailable.") + fd = os.open(lock_path, os.O_CREAT | os.O_RDWR | getattr(os, "O_NOFOLLOW", 0), 0o600) + acquired = False + try: + if not stat.S_ISREG(os.fstat(fd).st_mode): + raise ConfigError("The configuration lock is unavailable.") + if WINDOWS and os.fstat(fd).st_size == 0: + os.write(fd, b"0") + deadline = time.monotonic() + timeout + while not acquired: + try: + _lock(fd) + acquired = True + except OSError: + if time.monotonic() >= deadline: + raise ConfigError("The configuration is busy. Retry this command.") from None + time.sleep(.01) + yield + finally: + try: + if acquired: + _lock(fd, release=True) + finally: + os.close(fd) + + +def atomic_write(path: Path, payload): + """Caller holds locked_config. Replace only a complete, synced private file.""" + temporary = None + try: + fd, temporary = tempfile.mkstemp(prefix="." + path.name + ".", suffix=".tmp", dir=path.parent) + with os.fdopen(fd, "wb") as stream: + stream.write((json.dumps(payload, indent=2, sort_keys=True) + "\n").encode()) + stream.flush() + os.fsync(stream.fileno()) + os.replace(temporary, path) + if not WINDOWS: + directory = os.open(path.parent, os.O_RDONLY) + try: + os.fsync(directory) + finally: + os.close(directory) + except OSError: + raise ConfigError("Unable to save the configuration. Retry before continuing.") from None + finally: + if temporary is not None: + try: + os.unlink(temporary) + except FileNotFoundError: + pass diff --git a/dataplicity_cli/tunnels.py b/dataplicity_cli/tunnels.py index 979b163..6be115d 100644 --- a/dataplicity_cli/tunnels.py +++ b/dataplicity_cli/tunnels.py @@ -10,6 +10,7 @@ from urllib.parse import urlsplit, urlunsplit from .api import ApiClient +from .config import Config from .client_identity import identity_headers from .remote_access import bridge_tcp_channel, run_port_forward, _close_stream_writer from .tunnel_transport import TunnelM2MClient @@ -78,7 +79,9 @@ def __init__(self, api: ApiClient, organisation: str, token: Optional[str] = Non self.organisation = organisation self.base = f"/api/organisations/{organisation}/development-tunnels/" if token: - self.base = publisher_origin(api.config.base_url) + self.base + # A publisher has its own recipient, cookie jar and authentication + # state. It cannot reload or invalidate the saved human account. + self.api = ApiClient(Config(base_url=publisher_origin(api.config.base_url), install_id=api.config.install_id)) self.headers = {"Authorization": f"TunnelPublisher {token}"} if token else None self._request_lock = threading.Lock() self._call_lock: Optional[asyncio.Lock] = None diff --git a/docs/gateway-credential-recipients.md b/docs/gateway-credential-recipients.md new file mode 100644 index 0000000..2e9dbf2 --- /dev/null +++ b/docs/gateway-credential-recipients.md @@ -0,0 +1,87 @@ +# CLI credential recipients + +The user API client is bound to the server and optional path prefix selected at +construction. Relative requests and absolute URLs within that recipient remain +supported. Scheme, canonical host, effective port and path boundary must match; +userinfo, fragments, path traversal, sibling prefixes and remote HTTP are +rejected before renewal or HTTP dispatch. Explicit loopback HTTP remains +available for development. Requests do not follow redirects with credentials. + +Changing the configured recipient requires new authentication. A command-line +override clears credentials in its selected in-memory configuration; changing +the saved server persists that clearing. Equivalent host case/default HTTPS port +does not discard the session. Token reload cannot import credentials from a +different saved recipient. An existing API client cannot silently change server +when its shared configuration object is changed. + +Tunnel publishers construct an independent client for the selected Runtime +recipient. That client has no saved user JWT, organisation key, cookies or token +persistence/reload callbacks. Its scoped publisher proof cannot refresh or clear +the saved human account. Background renewal follows the client running the +command. CLI startup and local configuration/auth-status operations perform no +automatic user renewal; user API requests retain proactive renewal and existing +long-running user commands retain keepalive. Primary password/MFA authentication +uses an anonymous client, as the native code flow already does. Installing user +JWTs removes an old organisation key, and installing a key removes old user JWTs. + +An explicit no-op requests authenticator prevents netrc or Session Basic auth +from replacing the chosen API proof. Environment proxy and CA support remain +enabled. Transport exceptions return a safe message because requests can include +credential headers or private URLs in exception text. Structured +`session_expired` is an authentication failure only at the existing authentication +status codes; ordinary permission failures do not acquire refresh authority. + +## Session ownership and persistence + +Each saved session has a local opaque identity. Explicit sign-in, key installation +and sign-out replace that identity, even when the credential bytes are unchanged. +Legitimate renewal retains it. A client captures its owner at construction; +admission, renewal, retry, persistence and response publication require that same +owner. An old request cannot pick up a replacement account, replay its mutation +under that account, clear it after a delayed denial, or return old account data +after replacement. Primary sign-in constructs a new client after committing its +new session. Local identity is concurrency metadata, never server authority. + +Configuration transactions use an OS process lock and replace a complete, +synced private file. They release locks before HTTP. Writes compare the observed +session against the latest file, preserve concurrent legitimate renewal and +independent preference edits, and refuse conflicting changes. A persistent lock +inode prevents waiters locking different files; contention has a bounded wait. +Legacy files acquire an identity from the exact unchanged file rather than from +unverified JWT claims. A changed legacy file requires restarting the command. +Unknown configuration metadata is preserved. + +Before renewal, the client persists one recovery UUID bound to that session and +the refresh input digest. Competing processes and a process restarted after a +lost response reuse it. Only the current owner and expected refresh input can +commit a result or revoke credentials. A delayed older reply cannot roll back a +newer rotation or erase its recovery receipt. Transport failure preserves the +input and receipt; the server's replay policy decides whether recovery remains +valid. Malformed credentials are refused before installation or transmission. + +## Validation and limits + +Tests exercise real requests URL/header preparation with synthetic responses, +foreign/sibling recipients and encoded traversal, netrc/session-auth interference, +recipient changes during admission/renewal, publisher startup/keepalive, public +primary sign-in and saved-server changes. The initial unchanged-source recipient +baseline produced 39 failures with six passing tests and six passing subtests. +Session regression tests additionally exercise real threads and spawned +processes, competing sign-ins, late success/denial/error replies, conditional +revocation, lost responses across restart, atomic-write failure, bounded lock +contention and preference conflicts. The unchanged-source ownership baseline +reproduced 13 failures with two passing tests. Native Windows locking still +requires the Windows CI run; a local primitive test alone is insufficient. + +Run the complete unit suite with its configured 100% coverage gate, the three +qualification contract files selected by CI, compile/help smoke checks, and a +local wheel build. Local tests redirect the default CLI configuration to a +temporary directory after coverage starts; they do not use real saved credentials. +Qualification contract tests include an actual loopback TCP bridge. Their success +is not evidence that an original installed agent passed a staging campaign. + +Actual server logout remains separate required work, including the choice +between per-session and account-wide revocation. Native platform build/signing, +installed-client journeys and final +unchanged-agent staging checks remain held. No production or public package, +installer, Homebrew, WinGet or client-feed publication is authorised here. diff --git a/tests/test_api_client_full.py b/tests/test_api_client_full.py index 423ad9f..c2613e4 100644 --- a/tests/test_api_client_full.py +++ b/tests/test_api_client_full.py @@ -11,7 +11,7 @@ from dataplicity_cli.api import ApiClient, ApiResponse from dataplicity_cli.client_identity import identity_headers -from dataplicity_cli.config import Config +from dataplicity_cli.config import Config, RefreshTicket class _FakeResponse: @@ -36,7 +36,9 @@ class ApiClientFullTest(unittest.TestCase): def test_build_url_supports_absolute_and_relative(self) -> None: client = ApiClient(Config(base_url="https://example.test/")) self.assertEqual(client._build_url("/v1/status"), "https://example.test/v1/status") - self.assertEqual(client._build_url("https://other.test/x"), "https://other.test/x") + self.assertEqual(client._build_url("https://example.test/x"), "https://example.test/x") + with self.assertRaises(ValueError): + client._build_url("https://other.test/x") def test_auth_headers_for_jwt_api_key_and_none(self) -> None: jwt_client = ApiClient(Config(auth_method="jwt", access_token="tok")) @@ -119,6 +121,9 @@ def test_refresh_session_only_when_jwt(self) -> None: self.assertFalse(client.refresh_session()) def test_jwt_expiry_parsing(self) -> None: + for payload in ([], None, True, "payload", {"exp": float("inf")}, {"exp": 10 ** 400}): + with self.subTest(payload=payload): + self.assertIsNone(ApiClient._jwt_expiry(_unsigned_jwt(payload))) self.assertIsNone(ApiClient._jwt_expiry(None)) self.assertIsNone(ApiClient._jwt_expiry("not-a-jwt")) self.assertIsNone(ApiClient._jwt_expiry("a.!!!.b")) @@ -151,7 +156,7 @@ def test_refresh_reloads_rotated_token_from_disk(self) -> None: reloaded = {"refresh": "rotated", "access": "new"} def reload() -> None: - cfg.refresh_token = reloaded["refresh"] + cfg.commit_rotation(RefreshTicket(cfg.capture_owner(), cfg.access_token, "stale", "00000000-0000-4000-8000-000000000001"), cfg.access_token, reloaded["refresh"]) client = ApiClient(cfg, on_config_reload=reload) client.session.post = Mock( @@ -212,7 +217,7 @@ def test_request_exception_returns_failed_response(self) -> None: client = ApiClient(Config(base_url="https://example.test")) client.session.request = Mock(side_effect=requests.RequestException("timeout")) resp = client.get("/status") - self.assertEqual(resp, ApiResponse(False, 0, None, "timeout")) + self.assertEqual(resp, ApiResponse(False, 0, None, "The API request failed. Check your connection or sign in again.")) def test_request_retries_once_after_refresh_success(self) -> None: cfg = Config(base_url="https://example.test", auth_method="jwt", access_token="old", refresh_token="ref") @@ -220,12 +225,12 @@ def test_request_retries_once_after_refresh_success(self) -> None: first = _FakeResponse(401, {"detail": "Token has expired"}, text="unauthorized") second = _FakeResponse(200, {"ok": True}, text='{"ok": true}') client.session.request = Mock(side_effect=[first, second]) - client._refresh_access_token = Mock(return_value=True) # type: ignore[assignment] + client.session.post = Mock(return_value=_FakeResponse(200, {"access": "renewed", "refresh": "rotated"})) resp = client.request("GET", "/x") self.assertTrue(resp.ok) self.assertEqual(resp.status_code, 200) - client._refresh_access_token.assert_called_once() + client.session.post.assert_called_once() self.assertEqual(client.session.request.call_count, 2) def test_request_handles_error_text_fallback_and_invalidation(self) -> None: diff --git a/tests/test_api_credentials.py b/tests/test_api_credentials.py index 7c85b49..dcd9494 100644 --- a/tests/test_api_credentials.py +++ b/tests/test_api_credentials.py @@ -16,8 +16,11 @@ def client(self): def test_publisher_uses_runtime_without_human_tokens_or_refresh(self): api = self.client() - TunnelAPI(api, "org", token="machine-secret").request("GET", "bootstrap/") - call = api.session.request.call_args + publisher = TunnelAPI(api, "org", token="machine-secret") + publisher.api.session.request = Mock(return_value=Mock(status_code=200, text="{}", json=Mock(return_value={}))) + publisher.request("GET", "bootstrap/") + call = publisher.api.session.request.call_args + api.session.request.assert_not_called() self.assertEqual(call.args[1], "https://runtime.dataplicity.com/api/organisations/org/development-tunnels/bootstrap/") self.assertEqual(call.kwargs["headers"]["Authorization"], "TunnelPublisher machine-secret") self.assertFalse(call.kwargs["allow_redirects"]) @@ -33,9 +36,10 @@ def test_human_calls_keep_gateway_and_their_refresh_lifecycle(self): def test_rejected_publisher_does_not_erase_saved_human_tokens(self): api = self.client() - api.session.request.return_value = Mock(status_code=401, text="invalid token", json=Mock(return_value={"detail": "invalid token"})) + publisher = TunnelAPI(api, "org", token="revoked-machine") + publisher.api.session.request = Mock(return_value=Mock(status_code=401, text="invalid token", json=Mock(return_value={"detail": "invalid token"}))) with self.assertRaises(TunnelError): - TunnelAPI(api, "org", token="revoked-machine").request("GET", "bootstrap/") + publisher.request("GET", "bootstrap/") self.assertEqual(api.config.access_token, "human-access") self.assertEqual(api.config.refresh_token, "human-refresh") api.ensure_fresh_access_token.assert_not_called() diff --git a/tests/test_api_recipient_boundaries.py b/tests/test_api_recipient_boundaries.py new file mode 100644 index 0000000..c0122c1 --- /dev/null +++ b/tests/test_api_recipient_boundaries.py @@ -0,0 +1,280 @@ +"""Credential recipients are checked before admission, refresh or HTTP I/O.""" +import json +import asyncio +import os +import tempfile +import unittest +from pathlib import Path +from unittest.mock import Mock, patch + +import requests +from typer.testing import CliRunner + +from dataplicity_cli.api import ApiClient +from dataplicity_cli.cli import app +from dataplicity_cli.config import Config +from dataplicity_cli.tunnels import TunnelAPI + + +def response(status=200, payload=None): + result = requests.Response() + result.status_code = status + result._content = json.dumps({} if payload is None else payload).encode() + return result + + +class ApiRecipientTests(unittest.TestCase): + def client(self, base="https://gateway.example.test/tenant", method="jwt"): + config = Config(base_url=base, auth_method=method, access_token="human-access", + refresh_token="human-refresh", api_key="organisation-secret", install_id="local-install") + client = ApiClient(config) + client.session.request = Mock(return_value=response()) + client.session.post = Mock(return_value=response()) + client.ensure_fresh_access_token = Mock() + return client + + def test_rejects_foreign_or_escaping_destinations_before_renewal_or_send(self): + targets = [ + "https://foreign.example.test/tenant/api/devices/", + "https://gateway.example.test.attacker.test/tenant/api/", + "https://gateway.example.test:444/tenant/api/", + "http://gateway.example.test/tenant/api/", + "https://gateway.example.test/outside/api/", + "https://gateway.example.test/tenant-sibling/api/", + "https://user:password@gateway.example.test/tenant/api/", + "https://gateway.example.test/tenant/../outside/", + "https://gateway.example.test/tenant/%2e%2e/outside/", + "//foreign.example.test/api/", + "file:///etc/passwd", "https://[invalid", "https://gateway.example.test:invalid/api/", + ] + for auth_method in ("jwt", "api_key"): + for target in targets: + with self.subTest(auth_method=auth_method, target=target): + client = self.client(method=auth_method) + result = client.request("POST", target, json_data={"command": "must not run"}) + self.assertFalse(result.ok) + self.assertEqual(result.status_code, 0) + self.assertIsNone(result.data) + self.assertNotIn("human-access", result.text) + self.assertNotIn("organisation-secret", result.text) + client.ensure_fresh_access_token.assert_not_called() + client.session.request.assert_not_called() + + def test_keeps_same_recipient_relative_and_absolute_calls(self): + for path in ("/api/devices/", "https://gateway.example.test:443/tenant/api/devices/"): + with self.subTest(path=path): + client = self.client() + self.assertTrue(client.get(path).ok) + call = client.session.request.call_args + self.assertEqual(call.kwargs["headers"]["Authorization"], "Bearer human-access") + self.assertFalse(call.kwargs["allow_redirects"]) + client.ensure_fresh_access_token.assert_called_once() + + def test_rejects_invalid_configured_recipient_without_sending_or_renewing(self): + for base in ("", "http://gateway.example.test", "https://name:password@gateway.example.test", + "https://gateway.example.test/?credential=secret", "https://gateway.example.test/#fragment"): + with self.subTest(base=base): + client = self.client(base) + self.assertFalse(client.get("/api/devices/").ok) + self.assertEqual(client._attempt_refresh(), "failed") + client.ensure_fresh_access_token.assert_not_called() + client.session.request.assert_not_called() + + def test_explicit_loopback_development_remains_supported(self): + for base in ("http://localhost:8000", "http://127.0.0.1:8000", "http://[::1]:8000"): + with self.subTest(base=base): + self.assertTrue(self.client(base).get("/api/devices/").ok) + + def test_netrc_and_session_basic_auth_cannot_replace_the_selected_api_credential(self): + for auth_method, expected in (("jwt", "Bearer human-access"), ("api_key", "ApiKey organisation-secret")): + for ambient in ("netrc", "session-auth"): + with self.subTest(auth_method=auth_method, ambient=ambient): + client = self.client(method=auth_method) + client.session.request = requests.Session.request.__get__(client.session) + if ambient == "session-auth": + client.session.auth = ("ambient-user", "ambient-password") + client.session.send = Mock(return_value=response()) + with patch("requests.sessions.get_netrc_auth", return_value=("netrc-user", "netrc-password")): + self.assertTrue(client.get("/api/devices/").ok) + prepared = client.session.send.call_args.args[0] + self.assertEqual(prepared.headers["Authorization"], expected) + # Corporate proxy/CA configuration remains available. + self.assertTrue(client.session.trust_env) + + def test_structured_session_failure_is_exact_and_status_bound(self): + client = self.client() + for code in ("session_expired", "token_not_valid", "token_expired", "invalid_token", "token_blacklisted"): + self.assertTrue(client._response_indicates_invalid_jwt(response(403, {"code": code}))) + for status, payload in ((403, {"code": "permission_denied"}), (403, {"code": "device_inactive"}), + (404, {"code": "session_expired"}), (500, {"code": "session_expired"})): + self.assertFalse(client._response_indicates_invalid_jwt(response(status, payload))) + for payload in ({"error": "session_expired"}, {"detail": {"code": "session_expired"}}): + self.assertTrue(client._response_indicates_invalid_jwt(response(401, payload))) + for text, invalid in ((b"temporary upstream failure", False), (b"Token has expired", True)): + wire = requests.Response(); wire.status_code = 401; wire._content = text + self.assertEqual(client._response_indicates_invalid_jwt(wire), invalid) + + def test_tunnel_publisher_has_an_independent_client_and_no_human_state(self): + parent = self.client("https://gateway.dataplicity.com") + parent.session.cookies.set("human-cookie", "private", domain=".dataplicity.com") + parent.session.auth = ("human", "password") + control = TunnelAPI(parent, "org", token="machine-secret") + self.assertIsNot(control.api, parent) + self.assertIsNot(control.api.session, parent.session) + self.assertEqual(control.api.config.base_url, "https://runtime.dataplicity.com") + self.assertIsNone(control.api.config.auth_method) + self.assertIsNone(control.api.config.access_token) + self.assertIsNone(control.api.config.refresh_token) + self.assertIsNone(control.api.config.api_key) + self.assertEqual(list(control.api.session.cookies), []) + self.assertIsNone(control.api.session.auth) + control.api.session.request = Mock(return_value=response()) + control.request("GET", "bootstrap/") + call = control.api.session.request.call_args + self.assertEqual(call.args[1], "https://runtime.dataplicity.com/api/organisations/org/development-tunnels/bootstrap/") + self.assertEqual(call.kwargs["headers"]["Authorization"], "TunnelPublisher machine-secret") + parent.session.request.assert_not_called() + self.assertEqual(parent.config.access_token, "human-access") + + def test_request_cannot_change_recipient_or_authentication_method_during_admission(self): + for attribute, value in (("base_url", "https://foreign.example.test"), ("auth_method", "api_key")): + with self.subTest(attribute=attribute): + client = self.client() + client.ensure_fresh_access_token.side_effect = lambda: setattr(client.config, attribute, value) + self.assertFalse(client.get("/api/devices/").ok) + client.session.request.assert_not_called() + + def test_old_issuer_refresh_cannot_overwrite_a_reconfigured_account(self): + client = self.client() + def moved(*args, **kwargs): + client.config.base_url = "https://foreign.example.test" + client.config.access_token = "new-recipient-access" + client.config.refresh_token = "new-recipient-refresh" + return response(200, {"access": "old-recipient-access", "refresh": "old-recipient-refresh"}) + client.session.post.side_effect = moved + self.assertEqual(client._attempt_refresh(), "failed") + self.assertEqual(client.config.access_token, "new-recipient-access") + self.assertEqual(client.config.refresh_token, "new-recipient-refresh") + + def test_refresh_keeps_netrc_out_and_rejects_malformed_payloads_without_installing_tokens(self): + for payload in (None, [], "unexpected", {"access": 123}, {"access": "a", "refresh": 123}, {"access": "a", "refresh": ""}): + with self.subTest(payload=payload): + client = self.client() + wire = requests.Response(); wire.status_code = 200 + wire._content = json.dumps(payload).encode() + client.session.post.return_value = wire + self.assertEqual(client._attempt_refresh(), "failed") + self.assertEqual(client.config.access_token, "human-access") + client = self.client() + client.session.post = requests.Session.post.__get__(client.session) + client.session.request = requests.Session.request.__get__(client.session) + client.session.send = Mock(return_value=response(200, {"access": "renewed"})) + with patch("requests.sessions.get_netrc_auth", return_value=("ambient", "password")): + self.assertEqual(client._attempt_refresh(), "ok") + self.assertNotIn("Authorization", client.session.send.call_args.args[0].headers) + + def test_transport_failure_does_not_reflect_a_credential_or_url(self): + client = self.client() + client.session.request.side_effect = requests.RequestException("header Bearer secret-value https://private.example.invalid/") + result = client.get("/api/devices/") + self.assertFalse(result.ok) + self.assertNotIn("secret-value", result.text) + self.assertNotIn("private.example.invalid", result.text) + + +class OverrideRecipientTests(unittest.TestCase): + def test_changing_cli_recipient_requires_new_authentication_and_keeps_original_file(self): + for target, retain in (("https://foreign.example.test", False), ("https://gateway.example.test/other", False), + ("https://GATEWAY.example.test:443/tenant/", True)): + with self.subTest(target=target), tempfile.TemporaryDirectory() as folder: + path = Path(folder)/"cli.json" + config = Config(base_url="https://gateway.example.test/tenant", auth_method="jwt", + access_token="human-access", refresh_token="human-refresh", api_key="organisation-secret", install_id="install") + config.save(path) + result = CliRunner().invoke(app, ["--config", str(path), "--base-url", target, "--json", "auth", "status"]) + self.assertEqual(result.exit_code, 0, result.output) + payload = json.loads(result.stdout) + self.assertEqual(payload["has_access_token"], retain) + self.assertEqual(payload["has_refresh_token"], retain) + self.assertEqual(payload["has_api_key"], retain) + self.assertEqual(Config.load(path).access_token, "human-access") + + def test_persisted_recipient_change_removes_previous_credentials(self): + with tempfile.TemporaryDirectory() as folder: + path = Path(folder)/"cli.json" + Config(auth_method="api_key", api_key="old-key", access_token="old-access", refresh_token="old-refresh", install_id="install").save(path) + result = CliRunner().invoke(app, ["--config", str(path), "--json", "config", "set", "--base-url", "https://new.example.test"]) + self.assertEqual(result.exit_code, 0, result.output) + saved = Config.load(path) + self.assertEqual(saved.base_url, "https://new.example.test") + self.assertIsNone(saved.access_token) + self.assertIsNone(saved.refresh_token) + self.assertIsNone(saved.api_key) + + def test_reload_never_adopts_another_issuer_credentials(self): + for target in ("https://new.example.test", "https://[invalid"): + with self.subTest(target=target), tempfile.TemporaryDirectory() as folder: + path = Path(folder)/"cli.json" + Config(auth_method="jwt", access_token="old-access", refresh_token="old-refresh", install_id="install").save(path) + config = Config.load(path) + client = ApiClient(config) + client.session.post = Mock(side_effect=AssertionError("foreign account adopted")) + replacement = Config.load(path) + replacement.base_url = target + replacement.access_token = "foreign-access" + replacement.refresh_token = "foreign-refresh" + replacement.save(path) + self.assertFalse(client.refresh_session()) + client.session.post.assert_not_called() + self.assertEqual(Config.load(path).access_token, "foreign-access") + + def test_password_signin_uses_no_existing_user_authority_and_discards_an_old_key(self): + with tempfile.TemporaryDirectory() as folder: + path = Path(folder)/"cli.json" + Config(auth_method="jwt", access_token="old-access", refresh_token="old-refresh", api_key="old-org-key", install_id="install").save(path) + sent = [] + def request(method, url, **kwargs): + sent.append(url) + self.assertNotIn("Authorization", kwargs["headers"]) + return response(200, {"status": "password_required"} if url.endswith('/bootstrap/') else {"access": "new-access", "refresh": "new-refresh"}) + with patch("requests.Session.request", side_effect=request), patch.object(ApiClient, "_attempt_refresh", side_effect=AssertionError("borrowed old user")): + result = CliRunner().invoke(app, ["--config", str(path), "--json", "auth", "login", "--email", "new@example.test", "--password", "new-password"]) + self.assertEqual(result.exit_code, 0, result.output) + self.assertEqual(len(sent), 2) + saved = Config.load(path) + self.assertEqual(saved.access_token, "new-access") + self.assertIsNone(saved.api_key) + + def test_key_installation_drops_saved_user_tokens_without_renewing_them(self): + with tempfile.TemporaryDirectory() as folder: + path = Path(folder)/"cli.json" + Config(auth_method="jwt", access_token="old-access", refresh_token="old-refresh", install_id="install").save(path) + with patch.object(ApiClient, "ensure_fresh_access_token", side_effect=AssertionError("borrowed old user")): + result = CliRunner().invoke(app, ["--config", str(path), "--json", "auth", "api-key"], input="organisation-key\n") + self.assertEqual(result.exit_code, 0, result.output) + saved = Config.load(path) + self.assertEqual(saved.api_key, "organisation-key") + self.assertIsNone(saved.access_token) + self.assertIsNone(saved.refresh_token) + + def test_publisher_startup_and_keepalive_do_not_touch_the_saved_human(self): + with tempfile.TemporaryDirectory() as folder: + path = Path(folder)/"cli.json" + Config(auth_method="jwt", access_token="old-access", refresh_token="old-refresh", install_id="install").save(path) + renewed = [] + async def keepalive(client): + renewed.append(client.config.auth_method) + await asyncio.sleep(0) + async def publish(*args, **kwargs): await asyncio.sleep(0) + with patch.object(ApiClient, "ensure_fresh_access_token", side_effect=AssertionError("startup borrowed saved user")), \ + patch.object(ApiClient, "maintain_fresh_access_token", new=keepalive), \ + patch("dataplicity_cli.tunnels.TunnelSession.publish", new=publish), \ + patch.dict(os.environ, {"AUDIT_TUNNEL_TOKEN": "machine-secret"}): + args = ["--config", str(path), "--json", "tunnel", "publish", "test", "--org", "org", "--port", "3000"] + result = CliRunner().invoke(app, [*args, "--token-env", "AUDIT_TUNNEL_TOKEN"]) + self.assertEqual(result.exit_code, 0, result.output) + self.assertEqual(renewed, []) + self.assertEqual(Config.load(path).access_token, "old-access") + result = CliRunner().invoke(app, args) + self.assertEqual(result.exit_code, 0, result.output) + self.assertEqual(renewed, ["jwt"]) diff --git a/tests/test_cli_session_ownership.py b/tests/test_cli_session_ownership.py new file mode 100644 index 0000000..9ce9ff3 --- /dev/null +++ b/tests/test_cli_session_ownership.py @@ -0,0 +1,402 @@ +"""Real files, threads and processes must not transfer an old action to a new login.""" +import json +import multiprocessing +import tempfile +import unittest +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from threading import Event +from unittest.mock import Mock, patch + +import requests + +from dataplicity_cli.api import ApiClient +from dataplicity_cli.config import Config +from dataplicity_cli.config_store import ConfigError, ConfigChangedError, locked_config + + +def http(status=200, data=None): + response = requests.Response() + response.status_code = status + response._content = json.dumps({} if data is None else data).encode() + return response + + +def install(config, name): + config.install_user_tokens(name + "-access", name + "-refresh") + + +def primary_writer(path, ready, go, results, name): + config = Config.load(Path(path)) + ready.put(name) + if not go.wait(10): + results.put((name, "timeout")); return + install(config, name) + try: + config.save(Path(path)) + results.put((name, "saved")) + except OSError: + results.put((name, "superseded")) + + +def claim_writer(path, ready, go, results): + config = Config.load(Path(path)); owner = config.capture_owner() + ready.put(True) + if go.wait(10): + results.put(config.claim_refresh(owner).request_id) + + +def lock_holder(path, ready, release): + with locked_config(Path(path)): + ready.set() + release.wait(10) + + +class SessionOwnershipTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.path = Path(self.temp.name)/"cli.json" + config = Config(install_id="test-install") + install(config, "a") + config.save(self.path) + + def client(self): + config = Config.load(self.path) + def reload(): + fresh = Config.load(self.path) + config.auth_method, config.access_token, config.refresh_token = fresh.auth_method, fresh.access_token, fresh.refresh_token + return ApiClient(config, on_token_update=lambda: config.save(self.path), on_config_reload=reload) + + def replace(self, name): + current = Config.load(self.path) + install(current, name) + current.save(self.path) + + def test_stale_settings_and_primary_writes_cannot_overwrite_a_replacement_login(self): + for change in ("settings", "primary"): + with self.subTest(change=change): + old = Config.load(self.path) + self.replace("b") + if change == "settings": old.last_email = "old@example.test" + else: install(old, "old-primary") + with self.assertRaises(OSError): old.save(self.path) + self.assertEqual(Config.load(self.path).access_token, "b-access") + + def test_an_empty_logout_invalidates_an_anonymous_pending_primary(self): + anonymous = Config.load(self.path) + anonymous.clear_tokens(); anonymous.save(self.path) + pending = Config.load(self.path) + logout = Config.load(self.path) + logout.clear_tokens(); logout.save(self.path) + install(pending, "pending") + with self.assertRaises(OSError): pending.save(self.path) + self.assertIsNone(Config.load(self.path).access_token) + + def test_refresh_response_cannot_restore_an_old_account_over_a_new_login(self): + client, entered, resume = self.client(), Event(), Event() + def post(*args, **kwargs): + entered.set(); self.assertTrue(resume.wait(10)) + return http(200, {"access": "a-renewed", "refresh": "a-rotated"}) + client.session.post = post + with ThreadPoolExecutor(max_workers=1) as pool: + pending = pool.submit(client.refresh_session) + try: + self.assertTrue(entered.wait(10)); self.replace("b") + finally: resume.set() + self.assertFalse(pending.result(10)) + self.assertEqual(Config.load(self.path).access_token, "b-access") + + def test_late_success_and_failure_do_not_publish_old_account_data_or_clear_new_login(self): + for status in (200, 401, 403, 500): + with self.subTest(status=status): + self.replace("a") + client, entered, resume = self.client(), Event(), Event() + def request(*args, **kwargs): + entered.set(); self.assertTrue(resume.wait(10)) + return http(status, {"private": "account-a", "code": "session_expired"}) + client.session.request = Mock(side_effect=request) + client.session.post = Mock(return_value=http(200, {"access": "must-not-use", "refresh": "must-not-use"})) + with ThreadPoolExecutor(max_workers=1) as pool: + pending = pool.submit(client.request, "POST", "/api/action/", json_data={"command": "account-a"}) + try: + self.assertTrue(entered.wait(10)); self.replace("b") + finally: resume.set() + result = pending.result(10) + self.assertFalse(result.ok); self.assertEqual(result.status_code, 0); self.assertIsNone(result.data) + client.session.request.assert_called_once(); client.session.post.assert_not_called() + self.assertEqual(Config.load(self.path).access_token, "b-access") + + def test_invalid_old_refresh_never_reloads_another_user_and_replays_its_mutation(self): + client = self.client() + def request(*args, **kwargs): + self.replace("b") + return http(401, {"code": "token_not_valid"}) + client.session.request = Mock(side_effect=request) + client.session.post = Mock(side_effect=[http(401, {"detail": "Token is invalid or expired"}), + http(200, {"access": "b-rotated", "refresh": "b-rotated-refresh"})]) + result = client.request("POST", "/api/action/", json_data={"command": "account-a"}) + self.assertFalse(result.ok); self.assertEqual(result.status_code, 0) + client.session.request.assert_called_once(); client.session.post.assert_not_called() + self.assertEqual(Config.load(self.path).access_token, "b-access") + + def test_owned_renewal_preserves_concurrent_preferences(self): + client, entered, resume = self.client(), Event(), Event() + def post(*args, **kwargs): + entered.set(); self.assertTrue(resume.wait(10)) + return http(200, {"access": "a-renewed", "refresh": "a-rotated"}) + client.session.post = post + with ThreadPoolExecutor(max_workers=1) as pool: + pending = pool.submit(client.refresh_session) + try: + self.assertTrue(entered.wait(10)) + preferences = Config.load(self.path) + preferences.last_email = "latest@example.test" + preferences.save(self.path) + finally: resume.set() + self.assertTrue(pending.result(10)) + saved = Config.load(self.path) + self.assertEqual(saved.access_token, "a-renewed") + self.assertEqual(saved.last_email, "latest@example.test") + + def test_competing_primary_processes_have_one_winner_without_partial_json(self): + context = multiprocessing.get_context("spawn") + ready, results, go = context.Queue(), context.Queue(), context.Event() + processes = [context.Process(target=primary_writer, args=(str(self.path), ready, go, results, name)) for name in ("b", "c")] + try: + for process in processes: process.start() + self.assertEqual({ready.get(timeout=10), ready.get(timeout=10)}, {"b", "c"}) + go.set() + outcomes = [results.get(timeout=10), results.get(timeout=10)] + self.assertEqual(sorted(value for _, value in outcomes), ["saved", "superseded"]) + winner = next(name for name, value in outcomes if value == "saved") + self.assertEqual(Config.load(self.path).access_token, winner + "-access") + finally: + go.set() + for process in processes: + process.join(10) + if process.is_alive(): process.terminate(); process.join(5) + ready.close(); results.close() + + def test_failed_atomic_replace_keeps_the_previous_complete_private_config(self): + config = Config.load(self.path); install(config, "b") + with patch("os.replace", side_effect=OSError("disk failure")): + with self.assertRaises(OSError): config.save(self.path) + self.assertEqual(Config.load(self.path).access_token, "a-access") + self.assertFalse(list(self.path.parent.glob("*.tmp"))) + + def test_lost_refresh_response_recovers_with_the_same_owned_request_id(self): + client = self.client(); bodies = [] + def lost(*args, **kwargs): + bodies.append(kwargs["json"]) + raise requests.ConnectionError("response lost") + client.session.post = lost + self.assertFalse(client.refresh_session()) + self.assertRegex(bodies[0]["request_id"], r"^[0-9a-f-]{36}$") + restored = self.client() + restored.session.post = Mock(return_value=http(200, {"access": "a-renewed", "refresh": "a-rotated"})) + self.assertTrue(restored.refresh_session()) + self.assertEqual(restored.session.post.call_args.kwargs["json"], bodies[0]) + self.assertEqual(Config.load(self.path).access_token, "a-renewed") + + def test_competing_refresh_processes_share_one_input_recovery_id(self): + context = multiprocessing.get_context("spawn") + ready, results, go = context.Queue(), context.Queue(), context.Event() + processes = [context.Process(target=claim_writer, args=(str(self.path), ready, go, results)) for _ in range(2)] + try: + for process in processes: process.start() + ready.get(timeout=10); ready.get(timeout=10); go.set() + ids = [results.get(timeout=10), results.get(timeout=10)] + self.assertEqual(ids[0], ids[1]) + self.assertEqual(Config.load(self.path).refresh_attempt["request_id"], ids[0]) + finally: + go.set() + for process in processes: + process.join(10) + if process.is_alive(): process.terminate(); process.join(5) + ready.close(); results.close() + + def test_busy_lock_is_bounded_and_cannot_overwrite_the_config(self): + context = multiprocessing.get_context("spawn") + ready, release = context.Event(), context.Event() + process = context.Process(target=lock_holder, args=(str(self.path), ready, release)) + process.start() + try: + self.assertTrue(ready.wait(10)) + pending = Config.load(self.path); install(pending, "b") + with patch("dataplicity_cli.config.locked_config", side_effect=lambda path: locked_config(path, timeout=.03)): + with self.assertRaisesRegex(ConfigError, "busy"): pending.save(self.path) + self.assertEqual(Config.load(self.path).access_token, "a-access") + finally: + release.set(); process.join(10) + if process.is_alive(): process.terminate(); process.join(5) + pending.save(self.path) + self.assertEqual(Config.load(self.path).access_token, "b-access") + + def test_independent_preferences_merge_but_conflicting_changes_are_rejected(self): + first, second = Config.load(self.path), Config.load(self.path) + first.last_email = "first@example.test"; first.save(self.path) + second.preferred_login_method = "sso"; second.save(self.path) + self.assertEqual(Config.load(self.path).last_email, "first@example.test") + stale = Config.load(self.path) + second.last_email = "second@example.test"; second.save(self.path) + stale.last_email = "stale@example.test" + with self.assertRaises(ConfigChangedError): stale.save(self.path) + + def test_an_old_rotation_cannot_erase_a_newer_rotation_or_its_pending_receipt(self): + old = Config.load(self.path); owner = old.capture_owner(); ticket = old.claim_refresh(owner) + winner = Config.load(self.path); winning = winner.claim_refresh(winner.capture_owner()) + winner.commit_rotation(winning, "a-2-access", "a-2-refresh") + following = Config.load(self.path); next_ticket = following.claim_refresh(following.capture_owner()) + old.commit_rotation(ticket, "a-2-access", "a-2-refresh") + saved = Config.load(self.path) + self.assertEqual(saved.refresh_token, "a-2-refresh") + self.assertEqual(saved.refresh_attempt["request_id"], next_ticket.request_id) + self.assertEqual(old.access_token, "a-2-access") + + def test_invalid_old_parent_cannot_revoke_a_successor_committed_before_invalidation(self): + client, entered, resume = self.client(), Event(), Event() + original = Config.invalidate_owned + def delayed(config, *args, **kwargs): + entered.set(); self.assertTrue(resume.wait(10)) + return original(config, *args, **kwargs) + client.session.post = Mock(return_value=http(401, {"code": "token_not_valid"})) + with ThreadPoolExecutor(max_workers=1) as pool, patch.object(Config, "invalidate_owned", delayed): + pending = pool.submit(client.refresh_session) + try: + self.assertTrue(entered.wait(10)) + winner = Config.load(self.path); ticket = winner.claim_refresh(winner.capture_owner()) + winner.commit_rotation(ticket, "successor-access", "successor-refresh") + finally: resume.set() + self.assertTrue(pending.result(10)) + self.assertEqual(Config.load(self.path).refresh_token, "successor-refresh") + + def test_raw_memory_mutation_cannot_change_the_actor_of_an_existing_client(self): + client = self.client(); install(client.config, "b") + client.session.request = Mock() + self.assertFalse(client.get("/api/private/").ok) + client.session.request.assert_not_called() + self.assertEqual(Config.load(self.path).access_token, "a-access") + + def test_missing_file_context_cannot_follow_a_later_created_login(self): + missing = self.path.with_name("missing.json") + client = ApiClient(Config.load(missing)); client.session.request = Mock(return_value=http()) + self.assertTrue(client.get("/api/public/").ok) + other = Config.load(missing); install(other, "b"); other.save(missing) + self.assertFalse(client.get("/api/private/").ok) + self.assertEqual(client.session.request.call_count, 1) + + def test_foreign_owner_cannot_revoke_and_changed_header_context_cannot_emit_a_proof(self): + client = self.client() + with self.assertRaises(ConfigError): client._invalidate_jwt_session(Config().capture_owner()) + owner = client.config.capture_owner() + with patch.object(client.config, "reload_owned", side_effect=lambda _: install(client.config, "b")): + with self.assertRaises(ConfigError): client._auth_headers(owner) + self.assertEqual(Config.load(self.path).access_token, "a-access") + + def test_a_late_network_error_does_not_publish_or_persist_the_replacement_account(self): + for refresh in (False, True): + with self.subTest(refresh=refresh): + self.replace("a"); client = self.client() + def changed(*args, **kwargs): + self.replace("b"); raise requests.ConnectionError("old secret") + if refresh: + client.session.post = changed + self.assertFalse(client.refresh_session()) + else: + client.session.request = changed + result = client.get("/api/private/") + self.assertFalse(result.ok); self.assertIsNone(result.data) + self.assertNotIn("old secret", result.text) + self.assertEqual(Config.load(self.path).access_token, "b-access") + + def test_a_response_body_decoded_after_replacement_is_not_published(self): + for refresh in (False, True): + with self.subTest(refresh=refresh): + self.replace("a"); client = self.client() + def decode(): + self.replace("b") + return {"access": "old-a-result", "refresh": "old-a-result", "private": "old account"} + wire = http(); wire.json = decode + if refresh: + client.session.post = Mock(return_value=wire) + self.assertFalse(client.refresh_session()) + else: + client.session.request = Mock(return_value=wire) + result = client.get("/api/private/") + self.assertFalse(result.ok); self.assertIsNone(result.data) + self.assertEqual(Config.load(self.path).access_token, "b-access") + + def test_current_failure_and_transient_recovery_preserve_their_distinct_contracts(self): + client = self.client(); client.session.request = Mock(return_value=http(403, {"code": "token_not_valid"})) + result = client.request("GET", "/api/private/", allow_refresh=False) + self.assertEqual(result.status_code, 403) + self.assertIsNone(Config.load(self.path).access_token) + self.replace("a"); client = self.client() + client.session.request = Mock(return_value=http(401, {"code": "token_not_valid"})) + client.session.post = Mock(side_effect=requests.ConnectionError("response lost")) + result = client.get("/api/private/") + self.assertEqual(result.status_code, 401) + self.assertEqual(Config.load(self.path).refresh_token, "a-refresh") + self.assertIsNotNone(Config.load(self.path).refresh_attempt) + + def test_an_owned_rotation_can_recover_a_late_401_without_another_refresh(self): + client = self.client() + def request(*args, **kwargs): + if kwargs["headers"]["Authorization"] == "Bearer a-access": + winner = Config.load(self.path); ticket = winner.claim_refresh(winner.capture_owner()) + winner.commit_rotation(ticket, "a-2-access", "a-2-refresh") + return http(401, {"code": "token_not_valid"}) + return http(200, {"owned": True}) + client.session.request = Mock(side_effect=request); client.session.post = Mock() + self.assertTrue(client.get("/api/private/").ok) + self.assertEqual(client.session.request.call_count, 2) + client.session.post.assert_not_called() + + def test_stale_mutation_cannot_replay_after_renewal_or_callback_changes_its_owner(self): + client = self.client() + client.session.request = Mock(return_value=http(401, {"code": "token_not_valid"})) + def post(*args, **kwargs): + self.replace("b"); return http(200, {"access": "a-2", "refresh": "a-2"}) + client.session.post = post + self.assertEqual(client.post("/api/action/", json_data={"old": True}).status_code, 0) + client.session.request.assert_called_once() + self.assertFalse(client.ensure_fresh_access_token()) + self.assertEqual(Config.load(self.path).access_token, "b-access") + self.replace("a"); config = Config.load(self.path) + def callback(): raise ConfigChangedError() + client = ApiClient(config, on_token_update=callback) + client.session.post = Mock(return_value=http(200, {"access": "a-2", "refresh": "a-2"})) + self.assertFalse(client.refresh_session()) + + def test_explicit_machine_proof_cannot_change_recipient_during_header_preparation(self): + client = self.client() + def moved(*args): + client.config.base_url = 'https://different.example.test' + return {} + client._auth_headers = moved + client.session.request = Mock() + result = client.request('GET', '/api/action/', headers={'Authorization':'TunnelPublisher machine'}) + self.assertEqual(result.status_code, 0) + client.session.request.assert_not_called() + + def test_malformed_credentials_cannot_be_installed_or_forwarded(self): + for value in ({'not':'a-token'}, 'bad\nvalue', 'space value', 'λ', ''): + with self.subTest(value=value): + config = Config.load(self.path) + with self.assertRaises(ConfigError): config.install_user_tokens(value, 'refresh') + with self.assertRaises(ConfigError): config.install_user_tokens('access', value) + with self.assertRaises(ConfigError): config.install_api_key(value) + ticket = config.claim_refresh(config.capture_owner()) + with self.assertRaises(ConfigError): config.commit_rotation(ticket, value, 'refresh') + self.assertEqual(Config.load(self.path).access_token, 'a-access') + if value: + for mode in ('jwt', 'api_key'): + bad = Config(auth_method=mode, access_token=value, api_key=value) + client = ApiClient(bad); client.session.request = Mock() + self.assertFalse(client.get('/api/private/').ok) + client.session.request.assert_not_called() + config = Config(auth_method='jwt', access_token='access', refresh_token='bad\nrefresh') + client = ApiClient(config); client.session.post = Mock() + self.assertFalse(client.refresh_session()) + client.session.post.assert_not_called() diff --git a/tests/test_config_atomic_store.py b/tests/test_config_atomic_store.py new file mode 100644 index 0000000..4739cad --- /dev/null +++ b/tests/test_config_atomic_store.py @@ -0,0 +1,81 @@ +import json +import os +import sys +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import patch + +import pytest + +from dataplicity_cli.config import Config +from dataplicity_cli.config_store import ConfigError, ConfigChangedError, atomic_write, locked_config + + +def test_lock_file_symlink_and_nonregular_file_are_rejected(tmp_path): + path = tmp_path/'cli.json' + target = tmp_path/'private'; target.write_bytes(b'untouched') + if os.name != 'nt': + path.with_name(path.name+'.lock').symlink_to(target) + with pytest.raises(ConfigError, match='unavailable'): + with locked_config(path): pytest.fail('symlink lock admitted') + assert target.read_bytes() == b'untouched' + path.with_name(path.name+'.lock').unlink() + else: + with patch.object(Path, 'is_symlink', return_value=True), pytest.raises(ConfigError): + with locked_config(path): pytest.fail('symlink lock admitted') + with patch('dataplicity_cli.config_store.stat.S_ISREG', return_value=False), pytest.raises(ConfigError): + with locked_config(path): pytest.fail('nonregular lock admitted') + + +def test_windows_lock_uses_one_persistent_byte_and_unlocks_the_same_offset(tmp_path): + calls = [] + def locking(fd, operation, size): calls.append((operation, size, os.lseek(fd, 0, os.SEEK_CUR))) + fake = SimpleNamespace(locking=locking, LK_NBLCK=2, LK_UNLCK=0) + path = tmp_path/'cli.json' + with patch.dict(sys.modules, {'msvcrt': fake}), patch('dataplicity_cli.config_store.WINDOWS', True): + with locked_config(path): atomic_write(path, {'complete': True}) + with locked_config(path): pass + assert calls == [(2,1,0),(0,1,0),(2,1,0),(0,1,0)] + assert path.with_name(path.name+'.lock').read_bytes() == b'0' + assert json.loads(path.read_text()) == {'complete': True} + + +def test_config_private_modes_and_unknown_metadata_survive_updates(tmp_path): + path = tmp_path/'cli.json' + original = {'auth_method':'jwt','access_token':'access','refresh_token':'refresh', + 'local_session_id':'invalid-uuid','extension':{'opaque':['exact',None,{'permission':'customer-value'}]}} + path.write_text(json.dumps(original)) + config = Config.load(path) + assert config.local_session_id == Config.load(path).local_session_id + config.last_email = 'new@example.test'; config.save(path) + assert json.loads(path.read_text())['extension'] == original['extension'] + if os.name != 'nt': assert path.stat().st_mode & 0o777 == 0o600 + path.write_text('[]') + assert Config.load(path).access_token is None + + +def test_read_failure_is_safe_and_never_replaces_the_file(tmp_path): + path = tmp_path/'cli.json'; path.write_text('original') + with patch.object(Path, 'read_bytes', side_effect=PermissionError('private-token-value')): + with pytest.raises(ConfigError) as raised: Config.load(path) + assert 'private-token-value' not in str(raised.value) + assert path.read_text() == 'original' + + +def test_cli_reports_configuration_failure_without_a_traceback_or_private_values(tmp_path): + from typer.testing import CliRunner + from dataplicity_cli.cli import app + with patch.object(Path, 'read_bytes', side_effect=PermissionError('private-token-value')): + result = CliRunner().invoke(app, ['--config', str(tmp_path/'cli.json'), 'auth', 'status']) + assert result.exit_code == 1 + assert 'Unable to read the configuration' in result.output + assert 'private-token-value' not in result.output + assert 'Traceback' not in result.output + + +def test_unwitnessed_overwrite_and_file_removal_cannot_restore_old_authentication(tmp_path): + path = tmp_path/'cli.json'; Config(access_token='original').save(path) + with pytest.raises(ConfigChangedError): Config(access_token='unwitnessed').save(path) + old = Config.load(path); path.unlink() + with pytest.raises(ConfigChangedError): old.save(path) + assert not path.exists() diff --git a/tests/test_release_qualification.py b/tests/test_release_qualification.py index 3baa339..b4ec85f 100644 --- a/tests/test_release_qualification.py +++ b/tests/test_release_qualification.py @@ -76,9 +76,15 @@ def test_entire_release_and_distribution_write_dependency_closure_is_guarded(): assert 'Verify Windows MSI signature' in blocks['build'] -def test_all_original_build_sign_upload_and_default_release_bytes_are_preserved(): +def test_original_build_sign_upload_and_release_bytes_except_reviewed_runner_migration(): source = WORKFLOW.read_text() assert source.count(INPUT) == 1 and source.count(GUARD) == 1 - original = source.replace(INPUT, '', 1).replace(GUARD, '', 1).encode() + # GitHub retired macOS 14 on 2026-11-02. Permit only this exact build-runner + # replacement; retain the original digest over every other workflow byte. + runner = ' - name: macos-universal2\n os: macos-15-intel\n' + assert source.count(runner) == 1 + assert 'macos-14' not in source + original = source.replace(INPUT, '', 1).replace(GUARD, '', 1).replace( + runner, runner.replace('macos-15-intel', 'macos-14'), 1).encode() digest = hashlib.sha1(b'blob ' + str(len(original)).encode() + b'\0' + original).hexdigest() assert digest == 'b88e00dcb2bb4beb7597c2896f726efa808b75c6' diff --git a/tests/test_sso_auth_flow.py b/tests/test_sso_auth_flow.py index 419583b..8789cd4 100644 --- a/tests/test_sso_auth_flow.py +++ b/tests/test_sso_auth_flow.py @@ -60,6 +60,7 @@ def test_callback_accepts_one_matching_proof_and_never_raw_tokens(): def test_exchange_uses_pinned_anonymous_issuer_without_refreshing_old_login(tmp_path): state = state_for(tmp_path) listener = listener_for(state) + previous_api = state.api captured = [] def request(method, url, **kwargs): captured.append((method, url, kwargs)) @@ -75,8 +76,9 @@ def request(method, url, **kwargs): saved = json.loads(state.config_path.read_text()) assert saved['access_token'] == 'new-access' and saved['refresh_token'] == 'new-refresh' assert listener.login.verifier not in state.config_path.read_text() - state.api.get.assert_not_called() - state.api.post.assert_not_called() + previous_api.get.assert_not_called() + assert state.api is not previous_api + previous_api.post.assert_not_called() @pytest.mark.parametrize('change', ['origin', 'invalid_origin', 'wrong_state', 'token_payload', 'missing_context']) diff --git a/tests/test_tunnels.py b/tests/test_tunnels.py index 4374fd0..862b40d 100644 --- a/tests/test_tunnels.py +++ b/tests/test_tunnels.py @@ -33,7 +33,7 @@ def test_names_and_secure_owner_urls(self): websocket_url(url) def test_auth_separation_and_errors_never_reflect_secrets(self): - api = Mock(config=SimpleNamespace(auth_method="jwt", access_token="jwt", base_url="https://gateway.dataplicity.com")) + api = Mock(config=SimpleNamespace(auth_method="jwt", access_token="jwt", base_url="https://gateway.dataplicity.com", install_id="install")) control = TunnelAPI(api, "org-1") api.request.return_value = ApiResponse(True, 200, {"results": []}, "") self.assertEqual(control.request("GET", ""), {"results": []}) @@ -53,11 +53,12 @@ def test_auth_separation_and_errors_never_reflect_secrets(self): with self.assertRaises(TunnelError): TunnelAPI(api, "../org") machine = TunnelAPI(api, "org-1", token="machine") - api.request.return_value = ApiResponse(True, 201, {"session_id": "s"}, "") + machine.api.request = Mock(return_value=ApiResponse(True, 201, {"session_id": "s"}, "")) machine.request("POST", "publish/", payload={"name": "api"}) - self.assertEqual(api.request.call_args.kwargs["headers"], {"Authorization": "TunnelPublisher machine"}) - self.assertFalse(api.request.call_args.kwargs["allow_refresh"]) - self.assertEqual(api.request.call_args.args[1], "https://runtime.dataplicity.com/api/organisations/org-1/development-tunnels/publish/") + self.assertEqual(machine.api.request.call_args.kwargs["headers"], {"Authorization": "TunnelPublisher machine"}) + self.assertFalse(machine.api.request.call_args.kwargs["allow_refresh"]) + self.assertEqual(machine.api.config.base_url, "https://runtime.dataplicity.com") + self.assertEqual(machine.api.request.call_args.args[1], "/api/organisations/org-1/development-tunnels/publish/") class LifecycleTests(unittest.IsolatedAsyncioTestCase):