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 fa9691f..e3f65de 100644 --- a/README.md +++ b/README.md @@ -97,6 +97,29 @@ 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. + +`dataplicity auth logout` clears saved credentials before contacting the server. +For a Core user login it revokes only that sign-in and its rotated credentials; +other apps/devices remain signed in. Use `dataplicity auth logout --everywhere` +to explicitly revoke all sign-ins on the account. That action exits unsuccessfully +if server revocation is unavailable or cannot be confirmed, even though local +credentials have been cleared. Neither action renews credentials to sign out. A delayed +reply cannot overwrite a replacement login. The CLI reports if server sign-out +could not be confirmed. Organisation API keys are only removed locally. +Server revocation requires the matching backend session-authority release; +an already-consumed refresh is accepted only as revocation proof. + Login with email/password: ``` @@ -113,6 +136,12 @@ SSO login: dataplicity auth sso --email you@example.com ``` +On remote terminals, use `--no-open-browser` to open the printed sign-in link +on your workstation, then paste the final code/state callback when prompted. +The CLI keeps its PKCE verifier locally and exchanges the code at the configured +HTTPS identity origin. Browser callbacks contain no access or refresh tokens. +This login flow requires the coordinated native-code backend/frontend release. + Use an organisation API key: ``` diff --git a/dataplicity_cli/account_logout.py b/dataplicity_cli/account_logout.py new file mode 100644 index 0000000..835ec12 --- /dev/null +++ b/dataplicity_cli/account_logout.py @@ -0,0 +1,47 @@ +"""Explicit Core sign-in revocation, independent of the detached local owner.""" +import time + +import requests + +from .api import _EXPLICIT_API_AUTH +from .api_recipient import recipient_url +from .config import LogoutProof + + +def revoke_account(proof: LogoutProof, *, everywhere: bool = False) -> str: + """Never renew, follow redirects, borrow cookies, or write credentials. + + The paired backend accepts a consumed refresh only as revocation proof. + A network/server failure is ambiguous and must not start another mutation. + """ + if not proof.access and not proof.refresh: + return 'unavailable' + try: + path = '/api/auth/logout/everywhere/' if everywhere else '/api/auth/logout/' + url = recipient_url(proof.base_url, path) + except ValueError: + return 'unavailable' + deadline = time.monotonic() + 3.0 + session = requests.Session() + try: + for access in ((proof.access, None) if proof.access and proof.refresh else (proof.access,)): + remaining = deadline - time.monotonic() + if remaining <= 0: + break + session.cookies.clear() + response = session.post(url, + headers={'Authorization': 'Bearer ' + access} if access else {}, + json={} if access else {'refresh': proof.refresh}, + auth=_EXPLICIT_API_AUTH, allow_redirects=False, timeout=remaining, stream=True) + try: + if 200 <= response.status_code < 300: + return 'revoked' + if response.status_code not in (401, 403): + break + finally: + response.close() + except requests.RequestException: + pass + finally: + session.close() + return 'failed' 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 a04f2c1..6c0a06a 100644 --- a/dataplicity_cli/cli.py +++ b/dataplicity_cli/cli.py @@ -7,6 +7,7 @@ import os import queue import re +import secrets import shutil import shlex import socket @@ -22,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 @@ -34,11 +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=( @@ -102,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: @@ -135,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) @@ -184,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: @@ -543,65 +555,20 @@ def _extract_sso_tokens(payload: Any) -> Tuple[Optional[str], Optional[str]]: return access, refresh -def _decode_json_payload(raw: str) -> Optional[Dict[str, Any]]: - try: - parsed = json.loads(raw) - except json.JSONDecodeError: - return None - if isinstance(parsed, dict): - return parsed - return None - - -def _extract_sso_payload_from_query(query: Dict[str, List[str]]) -> Optional[Dict[str, Any]]: - payload: Dict[str, Any] = {} - for key in ("access", "refresh", "code", "state", "token"): - values = query.get(key) - if values: - payload[key] = values[0] - for key in ("payload", "json", "data"): - values = query.get(key) - if not values: - continue - nested = _decode_json_payload(values[0]) - if nested: - payload.update(nested) - break - return payload if payload else None - - -def _extract_sso_payload_from_url(url: str) -> Optional[Dict[str, Any]]: - parsed = urlparse(url) - query_payload = _extract_sso_payload_from_query(parse_qs(parsed.query, keep_blank_values=True)) - fragment_payload = _extract_sso_payload_from_query(parse_qs(parsed.fragment, keep_blank_values=True)) - if query_payload and fragment_payload: - merged = dict(query_payload) - merged.update(fragment_payload) - return merged - return query_payload or fragment_payload - - -def _parse_sso_user_artifact(raw: str) -> Optional[Dict[str, Any]]: - text = raw.strip() - if not text: - return None - payload = _decode_json_payload(text) - if payload: - return payload - if text.startswith("http://") or text.startswith("https://"): - return _extract_sso_payload_from_url(text) - return _extract_sso_payload_from_query(parse_qs(text, keep_blank_values=True)) - - class _SsoCallbackListener: def __init__(self) -> None: - self._queue: "queue.Queue[Dict[str, Any]]" = queue.Queue() + self._queue: "queue.Queue[Dict[str, Any]]" = queue.Queue(maxsize=1) self._server: Optional[ThreadingHTTPServer] = None self._thread: Optional[threading.Thread] = None self.callback_url: Optional[str] = None + self.login: Optional[NativeLogin] = None + self._nonce = secrets.token_urlsafe(32) def start(self) -> bool: payload_queue = self._queue + listener = self + capture_lock = threading.Lock() + captured = False class Handler(BaseHTTPRequestHandler): def log_message(self, format: str, *args: Any) -> None: # pragma: no cover @@ -754,47 +721,57 @@ def _write_html(self, status: int, message: str) -> None: self.send_response(status) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(body))) + self.send_header("Cache-Control", "no-store") + self.send_header("Referrer-Policy", "no-referrer") + self.send_header("Content-Security-Policy", "default-src 'none'; style-src 'unsafe-inline'; img-src data:; sandbox") self.end_headers() self.wfile.write(body) - def _capture_payload(self) -> Optional[Dict[str, Any]]: - parsed = urlparse(self.path) - query_payload = _extract_sso_payload_from_query(parse_qs(parsed.query, keep_blank_values=True)) - if query_payload: - return query_payload - if self.command != "POST": - return None - length = int(self.headers.get("Content-Length") or "0") - if length <= 0: - return None - raw = self.rfile.read(length).decode("utf-8", "ignore") - if "application/json" in (self.headers.get("Content-Type") or ""): - return _decode_json_payload(raw) - form_payload = _extract_sso_payload_from_query(parse_qs(raw, keep_blank_values=True)) - return form_payload - def do_GET(self) -> None: # noqa: N802 - payload = self._capture_payload() - if payload: - payload_queue.put(payload) - self._write_html(200, "Sign-in complete.") + nonlocal captured + try: + parsed = urlparse(self.path) + except ValueError: + self._write_html(400, "Invalid authentication callback.") + return + path = "/" + listener._nonce + if (len(self.path) > 4096 or parsed.scheme or parsed.netloc or parsed.fragment + or self.headers.get_all("Host") != [f"127.0.0.1:{self.server.server_port}"]): + self._write_html(400, "Invalid authentication callback.") + return + if parsed.path == path + "/complete" and not parsed.query and captured: + self._write_html(200, "Sign-in handoff received.") return - self._write_html(200, "Waiting for authentication callback.") + values = parse_qs(parsed.query, keep_blank_values=True) + if (parsed.path != path or listener.login is None + or any(len(value) != 1 for value in values.values())): + self._write_html(400, "Invalid authentication callback.") + return + payload = {key: value[0] for key, value in values.items()} + if listener.login.accept_payload(payload) is None: + self._write_html(400, "Invalid authentication callback.") + return + with capture_lock: + if captured: + self._write_html(410, "Authentication callback already received.") + return + captured = True + payload_queue.put_nowait(payload) + self.send_response(303) + self.send_header("Location", path + "/complete") + self.send_header("Cache-Control", "no-store") + self.send_header("Referrer-Policy", "no-referrer") + self.end_headers() def do_POST(self) -> None: # noqa: N802 - payload = self._capture_payload() - if payload: - payload_queue.put(payload) - self._write_html(200, "Sign-in complete.") - return - self._write_html(400, "No authentication payload was provided.") + self._write_html(405, "Use the browser authentication callback.") try: self._server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) except OSError: return False host, port = self._server.server_address[:2] - self.callback_url = f"http://{host}:{port}/callback" + self.callback_url = f"http://{host}:{port}/{self._nonce}" self._thread = threading.Thread(target=self._server.serve_forever, kwargs={"poll_interval": 0.2}, daemon=True) self._thread.start() return True @@ -817,13 +794,37 @@ 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 +def _anonymous_identity_api(state: AppContext, origin: str) -> ApiClient: + # Never borrow or rotate an existing account's credentials to complete a + # different account's primary login. The callback cannot supply this origin. + return ApiClient(Config(base_url=identity_origin(origin), install_id=state.config.install_id)) + + +def _exchange_native_callback(state: AppContext, listener: _SsoCallbackListener, payload: Any) -> bool: + if listener.login is None: + return False + try: + if identity_origin(state.config.base_url) != listener.login.origin: + return False + except ValueError: + return False + fields = listener.login.exchange_fields(payload) + if fields is None: + return False + response = _anonymous_identity_api(state, listener.login.origin).post( + "/api/auth/native/token/", json_data=fields, + ) + access, refresh = _extract_sso_tokens(response.data) + return bool(response.ok and isinstance(access, str) and access + and isinstance(refresh, str) and refresh and _apply_tokens_or_none(state, response.data)) + + def _attempt_sso_auto_complete( state: AppContext, listener: Optional[_SsoCallbackListener], @@ -835,8 +836,8 @@ def _attempt_sso_auto_complete( deadline = time.monotonic() + max(timeout_seconds, 1) while time.monotonic() < deadline: payload = listener.wait_for_payload(timeout_seconds=1.0) - if payload and _apply_tokens_or_none(state, payload): - return True + if payload: + return _exchange_native_callback(state, listener, payload) return False @@ -934,7 +935,7 @@ def _second_factor_login_hint(*, has_totp: bool, browser_login_url: Optional[str def _looks_like_cli_login_url(url: str) -> bool: parsed = urlparse(url) query = parse_qs(parsed.query, keep_blank_values=True) - return any(key in query for key in ("cli", "cli_state", "cli_callback", "cli_callback_url")) + return any(key in query for key in ("native_request",)) def _browser_login_url_from_bootstrap(data: Any) -> Optional[str]: @@ -944,6 +945,12 @@ def _browser_login_url_from_bootstrap(data: Any) -> Optional[str]: url = _first_text(data.get("redirect_url") or data.get("login_url") or data.get("browser_login_url")) if not url: return None + try: + parsed = urlparse(url) + if parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password: + return None + except ValueError: + return None if status in {"sso_redirect", "cli_browser_login"}: return url if status == "password" and _looks_like_cli_login_url(url): @@ -955,13 +962,16 @@ def _bootstrap_cli_login( state: AppContext, email: str, *, - callback_url: Optional[str] = None, + listener: Optional[_SsoCallbackListener] = None, ) -> ApiResponse: payload: Dict[str, Any] = {"email": email} - if callback_url: - payload["callback_url"] = callback_url - payload["cli_callback"] = callback_url - return state.api.post("/api/auth/bootstrap/", json_data=payload) + origin = state.config.base_url + if listener is not None and listener.callback_url: + if listener.login is None: + listener.login = NativeLogin(origin, listener.callback_url) + payload.update(listener.login.discovery_fields()) + origin = listener.login.origin + return _anonymous_identity_api(state, origin).post("/api/auth/bootstrap/", json_data=payload) def _start_cli_callback_listener() -> Optional[_SsoCallbackListener]: @@ -1015,7 +1025,7 @@ def _attempt_password_browser_login( timeout_seconds = _coerce_timeout_seconds(timeout) listener = _start_cli_callback_listener() callback_url = listener.callback_url if listener else None - response = _bootstrap_cli_login(state, email, callback_url=callback_url) + response = _bootstrap_cli_login(state, email, listener=listener) if not response.ok: if listener: listener.stop() @@ -1846,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, @@ -1868,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, @@ -2124,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}) @@ -2166,90 +2179,49 @@ def _attempt_sso_login( open_browser: bool = True, timeout: int = 180, ) -> bool: - """Complete browser SSO when the account supports it. - - Returns False if SSO is not enabled so callers can fall back. Other - failures raise ``typer.Exit`` after printing an error. - """ + """Authenticate with a PKCE-bound browser code; keep manual handoffs proof-only.""" state = _ctx(ctx) timeout_seconds = _coerce_timeout_seconds(timeout) - listener: Optional[_SsoCallbackListener] = None - bootstrap_payload: Dict[str, Any] = {"email": email} - if open_browser: - listener = _SsoCallbackListener() - if listener.start() and listener.callback_url: - bootstrap_payload["callback_url"] = listener.callback_url - if not state.json_output: - state.console.print(f"Listening for SSO callback on [blue]{listener.callback_url}[/blue]") - else: - listener = None - - response = state.api.post("/api/auth/bootstrap/", json_data=bootstrap_payload) - if not response.ok: - if listener: - listener.stop() - message = _friendly_response_message("Unable to start SSO.", response.data, response.text) - _fail_login(state, message) - - if not isinstance(response.data, dict) or response.data.get("status") != "sso_redirect": - if listener: - listener.stop() - return False - - redirect_url = response.data.get("redirect_url") - if not redirect_url: - if listener: - listener.stop() - _fail_login(state, "SSO redirect URL missing.") - + listener = _start_cli_callback_listener() try: + response = _bootstrap_cli_login(state, email, listener=listener) + if not response.ok: + _fail_login(state, "Unable to start SSO. Restart sign-in.") + if not isinstance(response.data, dict) or response.data.get("status") != "sso_redirect": + return False + if listener is None or listener.login is None: + _fail_login(state, "Unable to start a private browser callback listener.") + redirect_url = _browser_login_url_from_bootstrap(response.data) + if redirect_url is None: + _fail_login(state, "SSO redirect URL missing or invalid.") + completed = False if open_browser: webbrowser.open(redirect_url) - if not state.json_output: - state.console.print("Waiting for browser sign-in to complete...") - if _attempt_sso_auto_complete(state, listener, timeout_seconds=timeout_seconds): - state.config.last_email = email - state.config.preferred_login_method = "sso" - state.config.save(state.config_path) + if not state.json_output: + state.console.print("Waiting for browser sign-in to complete...") + completed = _attempt_sso_auto_complete(state, listener, timeout_seconds=timeout_seconds) + if not completed: if state.json_output: - _print_json({"ok": True, "detail": "SSO login complete"}) - else: - state.console.print("[green]SSO login complete.[/green]") - return True + _print_json({"ok": False, "detail": "Browser sign-in did not complete. Restart sign-in, or use interactive SSO without --json for a manual code handoff."}) + raise typer.Exit(code=2) + state.console.print("Open this sign-in URL in your browser:") + state.console.print(f"[blue]{redirect_url}[/blue]") + raw = typer.prompt("Final callback URL or code/state query", hide_input=True) + payload = listener.login.parse_artifact(raw) + if payload is None or not _exchange_native_callback(state, listener, payload): + _fail_login(state, "The sign-in code was not accepted. Restart sign-in.") + state.config.last_email = email + state.config.preferred_login_method = "sso" + state.config.save(state.config_path) + if state.json_output: + _print_json({"ok": True, "detail": "SSO login complete"}) + else: + state.console.print("[green]SSO login complete.[/green]") + return True finally: - if listener: + if listener is not None: listener.stop() - sso_complete_url = state.api._build_url("/api/auth/sso/complete/") - if state.json_output: - _print_json( - { - "ok": False, - "detail": "Automatic SSO completion timed out. Complete sign-in in your browser and run `dataplicity auth sso` again.", - "sso_complete_url": sso_complete_url, - } - ) - raise typer.Exit(code=2) - - state.console.print("[yellow]Automatic callback did not complete in time.[/yellow]") - state.console.print("Complete SSO in your browser, then open:") - state.console.print(f"[blue]{sso_complete_url}[/blue]") - state.console.print("Paste either the final browser URL, query string, or JSON payload below.") - - raw = typer.prompt("SSO response") - payload = _parse_sso_user_artifact(raw) - if payload is None: - _show_error(state.console, "Could not parse SSO response.") - raise typer.Exit(code=1) - if not _apply_tokens_or_none(state, payload): - _show_error(state.console, "No access token found in payload.") - raise typer.Exit(code=1) - state.config.last_email = email - state.config.preferred_login_method = "sso" - state.config.save(state.config_path) - state.console.print("[green]SSO login complete.[/green]") - return True - @auth_app.command("login") def auth_login( @@ -2273,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 ...`." @@ -2300,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) @@ -2333,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) @@ -2357,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: @@ -2404,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: @@ -2415,20 +2387,42 @@ def auth_api_key_set(ctx: typer.Context) -> None: @auth_app.command("logout") -def auth_logout(ctx: typer.Context) -> None: - """Clear saved authentication credentials. +def auth_logout( + ctx: typer.Context, + everywhere: bool = typer.Option( + False, "--everywhere", help="Revoke every Core user sign-in on this account." + ), +) -> None: + """Clear saved credentials and revoke this Core user sign-in. + + Other apps/devices remain signed in unless --everywhere is supplied. + Organisation API keys are removed locally without revoking the key. Examples: dataplicity auth logout + dataplicity auth logout --everywhere """ state = _ctx(ctx) - state.config.clear_tokens() - state.config.clear_api_key() - state.config.save(state.config_path) + from .account_logout import revoke_account + proof = state.config.detach_for_logout(state.config.capture_owner()) + result = revoke_account(proof, everywhere=everywhere) if state.json_output: - _print_json({"ok": True, "detail": "Logged out"}) + detail = 'Logged out' if not everywhere else ( + 'Logged out everywhere' if result == 'revoked' else 'Logged out locally' + ) + _print_json({"ok": not everywhere or result == 'revoked', "detail": detail, + "scope": "account" if everywhere else "sign_in", "server_session_revocation": result}) + elif result == 'failed': + scope = 'Account-wide sign-out' if everywhere else 'Server sign-out' + state.console.print(f'Saved credentials cleared. {scope} could not be confirmed.') + elif everywhere and result == 'unavailable': + state.console.print('Saved credentials cleared. Sign out everywhere requires a Core user sign-in.') + elif everywhere: + state.console.print("[green]Logged out everywhere.[/green]") else: state.console.print("[green]Logged out.[/green]") + if everywhere and result != 'revoked': + raise typer.Exit(1) @auth_app.command("status") diff --git a/dataplicity_cli/config.py b/dataplicity_cli/config.py index f66998c..2540d90 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,312 @@ 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 + + +@dataclass(frozen=True) +class LogoutProof: + base_url: str = field(repr=False) + access: Optional[str] = field(default=None, repr=False) + refresh: Optional[str] = field(default=None, repr=False) + + +_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 detach_for_logout(self, owner): + """Capture the latest owned proof and clear it in one file transaction. + + Network work must happen after this returns, without a config reference. + A waiting rotation or primary login cannot publish into the old owner. + """ + with self._owned_record(owner) as record: + jwt = record.auth_method == 'jwt' + proof = LogoutProof(record.base_url, + record.access_token if jwt and valid_api_credential(record.access_token) else None, + record.refresh_token if jwt and valid_api_credential(record.refresh_token) else None) + record.clear_tokens() + record.clear_api_key() + self._persist_owned(record) + return proof + + 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/native_auth.py b/dataplicity_cli/native_auth.py new file mode 100644 index 0000000..5ed96af --- /dev/null +++ b/dataplicity_cli/native_auth.py @@ -0,0 +1,102 @@ +"""PKCE login state is local; callbacks carry only a code and matching state.""" + +from __future__ import annotations + +import base64 +from dataclasses import dataclass, field +import hashlib +import json +import re +import secrets +from typing import Any, Optional +from urllib.parse import parse_qs, urlsplit + +_CODE = re.compile(r"[A-Za-z0-9_-]{43}\Z") + + +def identity_origin(raw: str) -> str: + try: + if not isinstance(raw, str): + raise ValueError() + parts = urlsplit(raw) + if (parts.scheme != "https" or not parts.hostname or parts.username or parts.password + or parts.path not in ("", "/") or parts.query or parts.fragment): + raise ValueError() + parts.port + except (TypeError, ValueError): + raise ValueError("Native sign-in requires a configured HTTPS identity origin.") from None + return raw.rstrip("/") + + +@dataclass(frozen=True) +class NativeLogin: + origin: str + callback_url: str + verifier: str = field(default_factory=lambda: secrets.token_urlsafe(32), repr=False) + state: str = field(default_factory=lambda: secrets.token_urlsafe(32), repr=False) + + def __post_init__(self): + object.__setattr__(self, "origin", identity_origin(self.origin)) + try: + callback = urlsplit(self.callback_url) + if (callback.scheme != "http" or callback.hostname != "127.0.0.1" + or callback.port is None or not 1024 <= callback.port <= 65535 + or callback.username or callback.password or callback.query or callback.fragment + or not re.fullmatch(r"/[A-Za-z0-9/_-]{1,160}", callback.path)): + raise ValueError() + except (TypeError, ValueError): + raise ValueError("Native sign-in requires a private loopback callback.") from None + + def discovery_fields(self): + challenge = base64.urlsafe_b64encode(hashlib.sha256(self.verifier.encode("ascii")).digest()) + return { + "native_client_id": "cli", "native_redirect_uri": self.callback_url, + "native_state": self.state, "native_code_challenge": challenge.decode().rstrip("="), + "native_code_challenge_method": "S256", + } + + def accept_payload(self, payload: Any) -> Optional[str]: + if not isinstance(payload, dict) or set(payload) != {"code", "state"}: + return None + code, state = payload["code"], payload["state"] + if (not isinstance(code, str) or not _CODE.fullmatch(code) + or not isinstance(state, str) or not state.isascii() + or not secrets.compare_digest(state, self.state)): + return None + return code + + def parse_artifact(self, raw: str): + """Allow a manual code/state handoff, never a token URL or token JSON.""" + if not isinstance(raw, str) or len(raw) > 4096: + return None + raw = raw.strip() + if raw.startswith("{"): + try: + payload = json.loads(raw) + except ValueError: + return None + else: + try: + parts = urlsplit(raw) + except ValueError: + return None + if parts.fragment: + return None + if parts.scheme or parts.netloc: + if f"{parts.scheme}://{parts.netloc}{parts.path}" != self.callback_url: + return None + raw = parts.query + values = parse_qs(raw, keep_blank_values=True) + if any(len(value) != 1 for value in values.values()): + return None + payload = {key: value[0] for key, value in values.items()} + return payload if self.accept_payload(payload) else None + + def exchange_fields(self, payload): + code = self.accept_payload(payload) + if code is None: + return None + return { + "grant_type": "authorization_code", "client_id": "cli", + "redirect_uri": self.callback_url, "code": code, "code_verifier": self.verifier, + } 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_account_logout.py b/tests/test_account_logout.py new file mode 100644 index 0000000..468a15c --- /dev/null +++ b/tests/test_account_logout.py @@ -0,0 +1,257 @@ +"""Sign-out detaches before I/O and can never publish into a replacement owner.""" +import json +import re +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +import threading +from unittest.mock import Mock, patch + +import pytest +import requests +from typer.testing import CliRunner + +from dataplicity_cli.account_logout import revoke_account +from dataplicity_cli.cli import app +from dataplicity_cli.config import Config, LogoutProof +from dataplicity_cli.config_store import ConfigChangedError + + +def response(status): + result = requests.Response() + result.status_code = status + result._content = b'{}' + result.close = Mock() + return result + + +def signed_in(path): + config = Config(base_url='https://gateway.example.test', auth_method='jwt', + access_token='owned-access', refresh_token='owned-refresh') + config.save(path) + return config + + +def test_detachment_captures_latest_owned_rotation_and_invalidates_its_pending_work(tmp_path): + path = tmp_path/'cli.json' + config = signed_in(path) + owner = config.capture_owner() + other = Config.load(path) + ticket = other.claim_refresh(other.capture_owner()) + other.commit_rotation(ticket, 'rotated-access', 'rotated-refresh') + pending = other.claim_refresh(other.capture_owner()) + proof = config.detach_for_logout(owner) + assert (proof.access, proof.refresh) == ('rotated-access', 'rotated-refresh') + saved = Config.load(path) + assert saved.auth_method is None and saved.access_token is None and saved.refresh_token is None + assert saved.refresh_attempt is None and saved.capture_owner() != owner + with pytest.raises(ConfigChangedError): + other.commit_rotation(pending, 'late-access', 'late-refresh') + assert Config.load(path).access_token is None + assert 'rotated' not in repr(proof) + + +def test_stale_logout_cannot_detach_a_replacement_primary_even_with_identical_credentials(tmp_path): + path = tmp_path/'cli.json' + old = signed_in(path) + owner = old.capture_owner() + replacement = Config.load(path) + replacement.install_user_tokens('owned-access', 'owned-refresh') + replacement.save(path) + with pytest.raises(ConfigChangedError): + old.detach_for_logout(owner) + assert Config.load(path).capture_owner() == replacement.capture_owner() + + +@pytest.mark.parametrize('everywhere', [False, True]) +def test_cli_clears_before_server_io_and_late_success_preserves_a_new_primary(tmp_path, everywhere): + path = tmp_path/'cli.json' + signed_in(path) + def send(session, url, **kwargs): + detached = Config.load(path) + assert detached.auth_method is None + assert detached.access_token is None and detached.refresh_token is None + suffix = 'everywhere/' if everywhere else '' + assert url == 'https://gateway.example.test/api/auth/logout/' + suffix + assert kwargs['headers'] == {'Authorization': 'Bearer owned-access'} + assert kwargs['json'] == {} + assert not kwargs['allow_redirects'] + detached.install_user_tokens('replacement-access', 'replacement-refresh') + detached.save(path) + return response(200) + with patch.object(requests.Session, 'post', send): + result = CliRunner().invoke(app, ['--config', str(path), '--json', 'auth', 'logout', + *(['--everywhere'] if everywhere else [])]) + assert result.exit_code == 0, result.output + assert json.loads(result.output)['server_session_revocation'] == 'revoked' + assert json.loads(result.output)['scope'] == ('account' if everywhere else 'sign_in') + assert Config.load(path).access_token == 'replacement-access' + assert 'owned-access' not in result.output and 'owned-refresh' not in result.output + + +@pytest.mark.parametrize('everywhere', [False, True]) +def test_explicit_access_rejection_uses_only_the_captured_refresh_and_no_ambient_auth(everywhere): + session = requests.Session() + session.cookies.set('unrelated', 'private-cookie') + session.auth = ('ambient', 'password') + # Explicit auth must beat session Basic and netrc, including refresh-only. + responses = [response(401), response(200)] + session.send = Mock(side_effect=responses) + with patch('dataplicity_cli.account_logout.requests.Session', return_value=session), \ + patch('requests.sessions.get_netrc_auth', return_value=('netrc', 'password')): + assert revoke_account(LogoutProof('https://gateway.example.test', 'owned-access', 'owned-refresh'), everywhere=everywhere) == 'revoked' + first, second = [call.args[0] for call in session.send.call_args_list] + assert first.headers['Authorization'] == 'Bearer owned-access' + assert 'Authorization' not in second.headers + assert json.loads(first.body) == {} + assert json.loads(second.body) == {'refresh': 'owned-refresh'} + assert 'Cookie' not in first.headers and 'Cookie' not in second.headers + suffix = 'everywhere/' if everywhere else '' + assert first.url == second.url == 'https://gateway.example.test/api/auth/logout/' + suffix + assert session.trust_env + for call in session.send.call_args_list: + assert call.kwargs['allow_redirects'] is False + assert call.kwargs['stream'] is True + assert 0 < call.kwargs['timeout'] <= 3 + for item in responses: + item.close.assert_called_once() + + +@pytest.mark.parametrize('status', [302, 307, 400, 429, 500, 503]) +@pytest.mark.parametrize('everywhere', [False, True]) +def test_ambiguous_or_non_auth_failure_never_starts_another_mutation(status, everywhere): + with patch.object(requests.Session, 'post', return_value=response(status)) as send: + assert revoke_account(LogoutProof('https://gateway.example.test', 'access', 'refresh'), everywhere=everywhere) == 'failed' + assert send.call_count == 1 + + +def test_connection_failure_never_renews_or_retries_and_cli_reports_local_completion(tmp_path): + path = tmp_path/'cli.json' + signed_in(path) + with patch.object(requests.Session, 'post', side_effect=requests.Timeout('private-credential')) as send: + result = CliRunner().invoke(app, ['--config', str(path), 'auth', 'logout']) + assert result.exit_code == 0 + assert 'credentials cleared' in result.output + assert 'could not be confirmed' in result.output + assert 'private-credential' not in result.output + assert Config.load(path).access_token is None + assert send.call_count == 1 + + +@pytest.mark.parametrize('method', ['api_key', None]) +def test_non_user_realms_clear_locally_without_sending_core_proof(tmp_path, method): + path = tmp_path/'cli.json' + config = Config(auth_method=method, api_key='organisation-secret', + access_token='inactive-user-access', refresh_token='inactive-user-refresh') + config.save(path) + with patch.object(requests.Session, 'post') as send: + result = CliRunner().invoke(app, ['--config', str(path), '--json', 'auth', 'logout']) + assert result.exit_code == 0 + assert json.loads(result.output)['server_session_revocation'] == 'unavailable' + saved = Config.load(path) + assert saved.api_key is None and saved.access_token is None and saved.auth_method is None + send.assert_not_called() + + +@pytest.mark.parametrize('base', ['https://user:private@gateway.example.test', 'http://foreign.example.test', 'file:///private']) +def test_invalid_recipient_never_receives_the_detached_proof(base): + proof = LogoutProof(base, 'private-access', 'private-refresh') + with patch.object(requests.Session, 'post') as send: + assert revoke_account(proof) == 'unavailable' + send.assert_not_called() + assert 'private' not in repr(proof) + + +def test_refresh_only_and_exhausted_budget_are_bounded_without_renewal(): + with patch.object(requests.Session, 'post', return_value=response(403)) as send: + assert revoke_account(LogoutProof('https://gateway.example.test', refresh='owned-refresh')) == 'failed' + assert send.call_args.kwargs['headers'] == {} + assert send.call_args.kwargs['json'] == {'refresh': 'owned-refresh'} + with patch('dataplicity_cli.account_logout.time.monotonic', side_effect=[0, 4]), \ + patch.object(requests.Session, 'post') as send: + assert revoke_account(LogoutProof('https://gateway.example.test', 'access')) == 'failed' + send.assert_not_called() + + +@pytest.mark.parametrize('everywhere', [False, True]) +def test_actual_http_fallback_never_adopts_server_cookies_or_a_new_saved_account(tmp_path, everywhere): + path = tmp_path/'cli.json' + calls = [] + + class Handler(BaseHTTPRequestHandler): + def do_POST(self): + body = json.loads(self.rfile.read(int(self.headers['Content-Length']))) + calls.append((self.path, self.headers.get('Authorization'), self.headers.get('Cookie'), body)) + if len(calls) == 1: + current = Config.load(path) + assert current.access_token is None and current.refresh_token is None + current.install_user_tokens('replacement-access', 'replacement-refresh') + current.save(path) + self.send_response(401 if len(calls) == 1 else 200) + self.send_header('Set-Cookie', 'ambient=must-not-be-forwarded; Path=/') + self.send_header('Content-Length', '0') + self.end_headers() + + def log_message(self, *_args): + pass + + server = ThreadingHTTPServer(('127.0.0.1', 0), Handler) + worker = threading.Thread(target=server.serve_forever, daemon=True) + worker.start() + try: + config = Config(base_url=f'http://127.0.0.1:{server.server_port}', auth_method='jwt', + access_token='owned-access', refresh_token='owned-refresh') + config.save(path) + with patch('requests.sessions.get_netrc_auth', return_value=('ambient', 'password')): + result = CliRunner().invoke(app, ['--config', str(path), '--json', 'auth', 'logout', + *(['--everywhere'] if everywhere else [])]) + assert result.exit_code == 0, result.output + assert json.loads(result.output)['server_session_revocation'] == 'revoked' + endpoint = '/api/auth/logout/everywhere/' if everywhere else '/api/auth/logout/' + assert calls == [ + (endpoint, 'Bearer owned-access', None, {}), + (endpoint, None, None, {'refresh': 'owned-refresh'}), + ] + assert Config.load(path).access_token == 'replacement-access' + finally: + server.shutdown() + server.server_close() + worker.join(timeout=5) + + +@pytest.mark.parametrize('json_output', [False, True]) +@pytest.mark.parametrize('user_login', [False, True]) +def test_everywhere_failure_is_explicit_after_local_detachment(tmp_path, json_output, user_login): + path = tmp_path/'cli.json' + if user_login: + signed_in(path) + else: + Config(auth_method='api_key', api_key='organisation-secret').save(path) + with patch.object(requests.Session, 'post', side_effect=requests.Timeout('private-proof')) as send: + result = CliRunner().invoke(app, ['--config', str(path), + *(['--json'] if json_output else []), 'auth', 'logout', '--everywhere']) + assert result.exit_code == 1, result.output + assert Config.load(path).auth_method is None + if json_output: + output = json.loads(result.output) + assert output['ok'] is False and output['scope'] == 'account' + assert output['server_session_revocation'] == ('failed' if user_login else 'unavailable') + else: + assert ('could not be confirmed' if user_login else 'requires a Core user sign-in') in result.output + assert 'private-proof' not in result.output and 'organisation-secret' not in result.output + assert send.call_count == (1 if user_login else 0) + + +def test_successful_everywhere_command_names_its_scope(tmp_path): + path = tmp_path/'cli.json' + signed_in(path) + with patch.object(requests.Session, 'post', return_value=response(200)): + result = CliRunner().invoke(app, ['--config', str(path), 'auth', 'logout', '--everywhere']) + assert result.exit_code == 0 and 'Logged out everywhere.' in result.output + + +def test_logout_help_describes_explicit_global_scope(): + result = CliRunner().invoke(app, ['auth', 'logout', '--help']) + assert result.exit_code == 0 + # Rich splits option text with ANSI styles on CI's coloured terminal. + plain = re.sub(r'\x1b\[[0-9;]*m', '', result.output).replace('│', ' ') + assert '--everywhere' in plain + assert 'Other apps/devices remain signed in unless' in ' '.join(plain.split()) 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_auth_login_mfa.py b/tests/test_auth_login_mfa.py index 98fdd71..d8c5acc 100644 --- a/tests/test_auth_login_mfa.py +++ b/tests/test_auth_login_mfa.py @@ -38,11 +38,7 @@ }, } -BROWSER_LOGIN_URL = ( - "https://www.dataplicity.com/login/?cli=1" - "&cli_callback=http://127.0.0.1:43111/callback" - "&cli_state=one-time-state" -) +BROWSER_LOGIN_URL = "https://www.dataplicity.com/login/?native_request=one-time-request" def _mfa_text(payload: dict) -> str: @@ -52,7 +48,9 @@ def _mfa_text(payload: dict) -> str: class _FakeLoopbackListener: def __init__(self, payload: dict | None = None) -> None: self.callback_url = "http://127.0.0.1:43111/callback" - self.payload = payload or {"access": "browser-access", "refresh": "browser-refresh"} + self.payload = payload + self.login = None + self.delivered = False self.stopped = False def start(self) -> bool: @@ -60,8 +58,10 @@ def start(self) -> bool: def wait_for_payload(self, timeout_seconds: float) -> dict | None: _ = timeout_seconds - payload, self.payload = self.payload, None - return payload + if self.delivered: + return None + self.delivered = True + return self.payload or {"code": "c"*43, "state": self.login.state} def stop(self) -> None: self.stopped = True @@ -112,11 +112,14 @@ def test_login_opens_gateway_cli_browser_login_url(self) -> None: bootstraps: list[dict] = [] def fake_post(path, json_data=None, data=None): + if path == "/api/auth/native/token/": + return ApiResponse(True, 200, {"access": "browser-access", "refresh": "browser-refresh"}, "") _ = data if path == "/api/auth/bootstrap/": bootstraps.append(json_data or {}) - if json_data and json_data.get("callback_url"): - self.assertEqual(json_data.get("cli_callback"), json_data.get("callback_url")) + if json_data and json_data.get("native_redirect_uri"): + self.assertEqual(json_data["native_code_challenge_method"], "S256") + self.assertNotIn("cli_callback", json_data) return ApiResponse( True, 200, @@ -140,7 +143,7 @@ def fake_post(path, json_data=None, data=None): mock_open.assert_called_once_with(BROWSER_LOGIN_URL) self.assertEqual(saved["access_token"], "browser-access") self.assertEqual(saved["preferred_login_method"], "email-password") - self.assertTrue(any(item.get("callback_url") for item in bootstraps)) + self.assertTrue(any(item.get("native_redirect_uri") for item in bootstraps)) self.assertTrue(listener.stopped) def test_login_falls_back_to_totp_when_browser_login_url_is_missing(self) -> None: @@ -256,9 +259,11 @@ def fake_post(path, json_data=None, data=None): def test_explicit_webauthn_type_opens_browser_when_login_url_exists(self) -> None: def fake_post(path, json_data=None, data=None): + if path == "/api/auth/native/token/": + return ApiResponse(True, 200, {"access": "browser-access", "refresh": "browser-refresh"}, "") _ = data if path == "/api/auth/bootstrap/": - if json_data and json_data.get("callback_url"): + if json_data and json_data.get("native_redirect_uri"): return ApiResponse( True, 200, 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_native_auth.py b/tests/test_native_auth.py new file mode 100644 index 0000000..d027b1e --- /dev/null +++ b/tests/test_native_auth.py @@ -0,0 +1,82 @@ +import base64 +from dataclasses import FrozenInstanceError +import hashlib +import json +from urllib.parse import urlencode + +import pytest + +from dataplicity_cli.native_auth import NativeLogin, identity_origin + + +def pending(): + return NativeLogin('https://identity.example/', 'http://127.0.0.1:49152/private-callback') + + +def test_pkce_verifier_stays_local_and_the_issuer_is_immutable(): + login = pending() + fields = login.discovery_fields() + assert fields['native_client_id'] == 'cli' + assert fields['native_code_challenge_method'] == 'S256' + assert fields['native_code_challenge'] == base64.urlsafe_b64encode( + hashlib.sha256(login.verifier.encode()).digest()).decode().rstrip('=') + assert login.verifier not in json.dumps(fields) + assert login.verifier not in repr(login) + with pytest.raises(FrozenInstanceError): + login.origin = 'https://foreign.example' + payload = {'code': 'c'*43, 'state': login.state} + assert login.exchange_fields(payload) == { + 'grant_type': 'authorization_code', 'client_id': 'cli', + 'redirect_uri': login.callback_url, 'code': 'c'*43, 'code_verifier': login.verifier, + } + assert login.origin == 'https://identity.example' + + +@pytest.mark.parametrize('value', [ + 'http://identity.example', 'https://user:password@identity.example', + 'https://identity.example/path', 'https://identity.example?issuer=other', + 'https://identity.example#fragment', 'https://identity.example:broken', 'https://', None, +]) +def test_native_issuer_must_be_a_configured_https_origin(value): + with pytest.raises(ValueError, match='HTTPS identity origin'): + identity_origin(value) + + +@pytest.mark.parametrize('callback', [ + 'http://localhost:49152/callback', 'http://evil.example:49152/callback', + 'https://127.0.0.1:49152/callback', 'http://127.0.0.1:80/callback', + 'http://127.0.0.1/callback', 'http://127.0.0.1:broken/callback', + 'http://user@127.0.0.1:49152/callback', 'http://127.0.0.1:49152/callback?state=x', + 'http://127.0.0.1:49152/callback#refresh=x', 'http://127.0.0.1:49152/../callback', +]) +def test_no_dns_or_privileged_port_callback_fallback(callback): + with pytest.raises(ValueError, match='private loopback'): + NativeLogin('https://identity.example', callback) + + +@pytest.mark.parametrize('changes', [ + {'state': 'another-login'}, {'state': '非ASCII'}, {'state': True}, + {'code': 'too-short'}, {'code': True}, {'access': 'injected-token'}, + {'refresh': 'injected-refresh'}, {'issuer': 'https://foreign.example'}, +]) +def test_foreign_state_or_additional_token_and_issuer_fields_never_form_a_grant(changes): + login = pending() + payload = {'code': 'c'*43, 'state': login.state, **changes} + assert login.accept_payload(payload) is None + assert login.exchange_fields(payload) is None + + +def test_manual_handoff_accepts_only_closed_code_state_proofs_for_its_callback(): + login = pending() + payload = {'code': 'c'*43, 'state': login.state} + query = urlencode(payload) + for value in (query, login.callback_url+'?'+query, json.dumps(payload)): + assert login.parse_artifact(value) == payload + for value in ( + 'access=stolen&refresh=stolen', '{"access":"stolen","refresh":"stolen"}', + query+'&code='+'d'*43, login.callback_url+'?'+query+'#access=stolen', + 'https://foreign.example/callback?'+query, '{broken-json', '[', 'http://[', + 'x'*4097, None, + ): + assert login.parse_artifact(value) is None + assert login.accept_payload(None) is None 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 6f30c7f..8789cd4 100644 --- a/tests/test_sso_auth_flow.py +++ b/tests/test_sso_auth_flow.py @@ -1,128 +1,161 @@ -from __future__ import annotations - -import unittest +import json from types import SimpleNamespace -from urllib.request import urlopen from unittest.mock import Mock, patch +from urllib.parse import urlencode, urlsplit + +import pytest +import requests +from rich.console import Console +import typer +from dataplicity_cli.api import ApiClient, ApiResponse +from dataplicity_cli.config import Config +from dataplicity_cli.native_auth import NativeLogin from dataplicity_cli.cli import ( - _SsoCallbackListener, - _attempt_sso_auto_complete, - _coerce_timeout_seconds, - _extract_sso_payload_from_url, - _extract_sso_payload_from_query, - _extract_sso_tokens, - _parse_sso_user_artifact, - auth_sso, + _SsoCallbackListener, _attempt_sso_auto_complete, _attempt_sso_login, + _coerce_timeout_seconds, _exchange_native_callback, _extract_sso_tokens, ) -class _FakeSsoListener: - def __init__(self, payload: dict | None) -> None: - self.payload = payload - - def wait_for_payload(self, timeout_seconds: float) -> dict | None: - _ = timeout_seconds - payload, self.payload = self.payload, None - return payload - - -class SsoAuthFlowTest(unittest.TestCase): - def test_extract_sso_tokens_supports_nested_tokens(self) -> None: - access, refresh = _extract_sso_tokens({"tokens": {"access": "a1", "refresh": "r1"}}) - self.assertEqual(access, "a1") - self.assertEqual(refresh, "r1") - - def test_extract_sso_payload_from_query_reads_payload_json(self) -> None: - payload = _extract_sso_payload_from_query({"payload": ['{"access":"a2","refresh":"r2"}']}) - self.assertIsNotNone(payload) - self.assertEqual(payload.get("access"), "a2") - self.assertEqual(payload.get("refresh"), "r2") - - def test_extract_sso_payload_from_url_reads_query_and_fragment(self) -> None: - payload = _extract_sso_payload_from_url("https://dataplicity.com/cb?code=abc#state=xyz") - self.assertEqual(payload, {"code": "abc", "state": "xyz"}) - - def test_callback_listener_captures_query_payload(self) -> None: - listener = _SsoCallbackListener() - started = listener.start() - self.assertTrue(started) - self.assertIsNotNone(listener.callback_url) - try: - response = urlopen(f"{listener.callback_url}?access=abc&refresh=def") - self.assertEqual(response.status, 200) - payload = listener.wait_for_payload(timeout_seconds=1.0) - self.assertEqual(payload, {"access": "abc", "refresh": "def"}) - finally: - listener.stop() - - def test_auto_complete_uses_loopback_payload_without_backend_polling(self) -> None: - state = SimpleNamespace(api=Mock()) - listener = _FakeSsoListener({"access": "abc", "refresh": "def"}) - - with patch("dataplicity_cli.cli._apply_tokens_or_none", return_value=True) as apply_tokens: - completed = _attempt_sso_auto_complete(state, listener, timeout_seconds=1) - - self.assertTrue(completed) - apply_tokens.assert_called_once_with(state, {"access": "abc", "refresh": "def"}) - state.api.get.assert_not_called() - state.api.post.assert_not_called() - - def test_auto_complete_without_listener_returns_immediately(self) -> None: - state = SimpleNamespace(api=Mock()) - - completed = _attempt_sso_auto_complete(state, None, timeout_seconds=180) - - self.assertFalse(completed) - state.api.get.assert_not_called() - state.api.post.assert_not_called() - - def test_sso_bootstrap_registers_loopback_callback_without_rewriting_redirect(self) -> None: - listener = Mock() - listener.start.return_value = True - listener.callback_url = "http://127.0.0.1:1234/callback" - redirect_url = "https://example.com/sso?redirect_uri=https%3A%2F%2Fexample.com%2Fcomplete" - state = SimpleNamespace( - api=Mock(), - config=SimpleNamespace(last_email=None, preferred_login_method=None, save=Mock()), - config_path="config.json", - json_output=True, - ) - state.api.post.return_value = SimpleNamespace( - ok=True, - data={"status": "sso_redirect", "redirect_url": redirect_url}, - text="", - ) - - with ( - patch("dataplicity_cli.cli._ctx", return_value=state), - patch("dataplicity_cli.cli._resolve_email_for_auth", return_value="user@example.com"), - patch("dataplicity_cli.cli._SsoCallbackListener", return_value=listener), - patch("dataplicity_cli.cli._attempt_sso_auto_complete", return_value=True), - patch("dataplicity_cli.cli.webbrowser.open") as open_browser, - patch("dataplicity_cli.cli._print_json"), +def state_for(tmp_path): + return SimpleNamespace(api=Mock(), config=Config(base_url='https://identity.example', + auth_method='jwt', access_token='old-access', refresh_token='old-refresh'), + config_path=tmp_path/'config.json', json_output=False, console=Console()) + + +def listener_for(state): + login = NativeLogin(state.config.base_url, 'http://127.0.0.1:49152/private') + return SimpleNamespace(login=login, + wait_for_payload=Mock(return_value={'code': 'c'*43, 'state': login.state})) + + +def test_callback_accepts_one_matching_proof_and_never_raw_tokens(): + listener = _SsoCallbackListener() + assert listener.start() + listener.login = NativeLogin('https://identity.example', listener.callback_url) + proof = {'code': 'c'*43, 'state': listener.login.state} + try: + for payload, headers in ( + ({'access': 'injected-access', 'refresh': 'injected-refresh'}, {}), + ({**proof, 'state': 'another-login'}, {}), ({**proof, 'state': '非ASCII'}, {}), + ({**proof, 'issuer': 'https://evil.example'}, {}), (proof, {'Host': 'evil.example'}), ): - auth_sso(Mock(), email="user@example.com", open_browser=True, timeout=1) - - state.api.post.assert_called_once_with( - "/api/auth/bootstrap/", - json_data={"email": "user@example.com", "callback_url": listener.callback_url}, - ) - open_browser.assert_called_once_with(redirect_url) - listener.stop.assert_called_once_with() - - def test_coerce_timeout_seconds_handles_invalid_values(self) -> None: - self.assertEqual(_coerce_timeout_seconds(30), 30) - self.assertEqual(_coerce_timeout_seconds("45"), 45) - self.assertEqual(_coerce_timeout_seconds(0), 1) - self.assertEqual(_coerce_timeout_seconds(object()), 180) - - def test_parse_sso_user_artifact_supports_url_and_query(self) -> None: - url_payload = _parse_sso_user_artifact("https://dataplicity.com/callback?code=abc&state=xyz") - self.assertEqual(url_payload, {"code": "abc", "state": "xyz"}) - query_payload = _parse_sso_user_artifact("access=tok123&refresh=ref456") - self.assertEqual(query_payload, {"access": "tok123", "refresh": "ref456"}) - - -if __name__ == "__main__": - unittest.main() + response = requests.get(listener.callback_url, params=payload, headers=headers, timeout=3) + assert response.status_code == 400 + assert requests.post(listener.callback_url, json=proof, timeout=3).status_code == 405 + assert requests.get(listener.callback_url+'?'+urlencode(proof)+'&code='+'d'*43, timeout=3).status_code == 400 + response = requests.get(listener.callback_url, params=proof, timeout=3) + assert response.status_code == 200 and response.history[0].status_code == 303 + assert response.url.endswith('/complete') and not urlsplit(response.url).query + assert response.headers['Cache-Control'] == 'no-store' + assert response.headers['Referrer-Policy'] == 'no-referrer' + assert listener.login.verifier not in response.text and 'injected-access' not in response.text + assert listener.wait_for_payload(1) == proof + assert requests.get(listener.callback_url, params=proof, timeout=3).status_code == 410 + assert listener.wait_for_payload(0.01) is None + finally: + listener.stop() + + +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)) + return SimpleNamespace(status_code=200, text='', json=lambda: {'access':'new-access', 'refresh':'new-refresh'}) + with patch('dataplicity_cli.api.requests.Session.request', side_effect=request), patch.object(ApiClient, '_attempt_refresh', side_effect=AssertionError('Borrowed old refresh')): + assert _exchange_native_callback(state, listener, listener.wait_for_payload()) + assert len(captured) == 1 + method, url, kwargs = captured[0] + assert method == 'POST' and url == 'https://identity.example/api/auth/native/token/' + assert 'Authorization' not in kwargs['headers'] and kwargs['allow_redirects'] is False + assert kwargs['json']['code_verifier'] == listener.login.verifier + assert kwargs['json']['redirect_uri'] == listener.login.callback_url + 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() + 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']) +def test_changed_or_invalid_context_does_not_send_a_grant_or_replace_credentials(tmp_path, change): + state = state_for(tmp_path) + listener = listener_for(state) + proof = listener.wait_for_payload() + if change == 'origin': state.config.base_url = 'https://foreign.example' + if change == 'invalid_origin': state.config.base_url = 'http://foreign.example' + if change == 'wrong_state': proof['state'] = 'another-login' + if change == 'token_payload': proof = {'access':'injected', 'refresh':'injected'} + if change == 'missing_context': listener.login = None + with patch.object(ApiClient, 'post') as posted: + assert not _exchange_native_callback(state, listener, proof) + posted.assert_not_called() + assert state.config.access_token == 'old-access' and not state.config_path.exists() + + +@pytest.mark.parametrize('ok,payload', [(False, {}), (True, {'access':'new'}), + (True, {'access':True, 'refresh':'new'}), (True, None)]) +def test_failed_or_malformed_exchange_preserves_credentials(tmp_path, ok, payload): + state = state_for(tmp_path) + listener = listener_for(state) + with patch.object(ApiClient, 'post', return_value=ApiResponse(ok, 400 if not ok else 200, payload, '')): + assert not _exchange_native_callback(state, listener, listener.wait_for_payload()) + assert state.config.access_token == 'old-access' and not state.config_path.exists() + + +def test_auto_complete_exchanges_proof_without_polling_session_completion(tmp_path): + state = state_for(tmp_path) + listener = listener_for(state) + with patch('dataplicity_cli.cli._exchange_native_callback', return_value=True) as exchange: + assert _attempt_sso_auto_complete(state, listener, timeout_seconds=1) + exchange.assert_called_once() + assert not _attempt_sso_auto_complete(state, None, timeout_seconds=180) + state.api.get.assert_not_called() + + +@pytest.mark.parametrize('automatic', [True, False]) +def test_sso_browser_and_manual_handoffs_use_pkce_and_never_token_paste(tmp_path, automatic): + state = state_for(tmp_path) + listener = SimpleNamespace(callback_url='http://127.0.0.1:49152/callback', login=None, + start=Mock(return_value=True), stop=Mock()) + url = 'https://identity.example/sso?next=bound-relay' + def post(path, **kwargs): + if path == '/api/auth/bootstrap/': + assert kwargs['json_data'] == {'email':'user@example.invalid', **listener.login.discovery_fields()} + return ApiResponse(True, 200, {'status':'sso_redirect', 'redirect_url':url}, '') + assert path == '/api/auth/native/token/' + return ApiResponse(True, 200, {'access':'new-access', 'refresh':'new-refresh'}, '') + with patch('dataplicity_cli.cli._ctx', return_value=state), patch('dataplicity_cli.cli._SsoCallbackListener', return_value=listener), patch.object(ApiClient, 'post', side_effect=post), patch('dataplicity_cli.cli._attempt_sso_auto_complete', return_value=True), patch('dataplicity_cli.cli.typer.prompt', side_effect=lambda *a,**k: urlencode({'code':'c'*43, 'state':listener.login.state})) as prompt, patch('dataplicity_cli.cli.webbrowser.open') as opened: + assert _attempt_sso_login(Mock(), 'user@example.invalid', open_browser=automatic) + if automatic: + opened.assert_called_once_with(url) + prompt.assert_not_called() + else: + opened.assert_not_called() + assert prompt.call_args.kwargs['hide_input'] is True + assert state.config.access_token == 'new-access' + listener.stop.assert_called_once() + + +def test_json_headless_sso_fails_without_raw_token_paste(tmp_path): + state = state_for(tmp_path) + state.json_output = True + listener = SimpleNamespace(callback_url='http://127.0.0.1:49152/callback', login=None, + start=Mock(return_value=True), stop=Mock()) + with patch('dataplicity_cli.cli._ctx', return_value=state), patch('dataplicity_cli.cli._SsoCallbackListener', return_value=listener), patch.object(ApiClient, 'post', return_value=ApiResponse(True, 200, {'status':'sso_redirect', 'redirect_url':'https://identity.example/sso'}, '')), patch('dataplicity_cli.cli.typer.prompt') as prompt: + with pytest.raises(typer.Exit) as stopped: + _attempt_sso_login(Mock(), 'user@example.invalid', open_browser=False) + assert stopped.value.exit_code == 2 + prompt.assert_not_called() + assert state.config.access_token == 'old-access' + + +def test_token_json_extraction_and_timeout_inputs(): + assert _extract_sso_tokens({'tokens': {'access':'a', 'refresh':'r'}}) == ('a', 'r') + assert _coerce_timeout_seconds(0) == 1 + assert _coerce_timeout_seconds('45') == 45 + assert _coerce_timeout_seconds(object()) == 180 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):