From 192e041a353c6860d80aaec2d44d15fe99d265c0 Mon Sep 17 00:00:00 2001 From: bearsyankees Date: Fri, 28 Aug 2026 15:23:22 -0400 Subject: [PATCH] Make cloud workspace switching session-safe --- AGENTS.md | 4 +- README.md | 25 +- skills/managed-pentesting-with-strix/SKILL.md | 17 +- strix/interface/cloud/__init__.py | 6 +- strix/interface/cloud/http.py | 35 ++- strix/interface/cloud/payment_proxy.py | 4 + strix/interface/cloud/runner.py | 7 + strix/interface/cloud/session.py | 161 +++++++++++++ strix/interface/cloud/workspaces.py | 99 ++++---- strix/interface/completions.py | 48 +++- strix/interface/platform_cli.py | 220 +++++++++++++++--- strix/interface/platform_identity.py | 46 ++++ tests/test_cloud_cli.py | 27 +-- tests/test_cloud_cli_runtime.py | 30 ++- tests/test_cloud_session.py | 171 ++++++++++++++ 15 files changed, 775 insertions(+), 125 deletions(-) create mode 100644 strix/interface/cloud/session.py create mode 100644 strix/interface/platform_identity.py create mode 100644 tests/test_cloud_session.py diff --git a/AGENTS.md b/AGENTS.md index 5114676d..e9039195 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -47,8 +47,8 @@ Target-specific workflows built on the same engine: strix cloud vulns list --severity critical strix cloud billing topup --credits 20 --yes # explicit approval after exit code 5 ``` - - Account setup runs from the CLI too: `strix cloud workspaces list|create|use` (`workspace` is an alias and `use` accepts a displayed number, name, or ID), `strix cloud org members invite`, `strix cloud billing subscribe --plan strix_cloud`, `strix cloud billing portal`, `strix cloud integrations install github`, and `strix cloud domains verify `. The last four end at a person: the command prints a link or a DNS record for the user to open or add, and it never completes the payment, installation, or DNS change for them. - - Every REST operation has a `strix cloud ` command. Run `strix cloud` to list them. Output is JSON when stdout is not a terminal (or with `--json`), and there are no prompts without a TTY. Binary downloads are the exception: redirect raw bytes intentionally, or combine `--output FILE --json` for structured download metadata. Exit codes: `0` success, `1` error, `2` usage, `4` auth or plan limit, `5` payment required. `--token` or `STRIX_API_TOKEN` overrides the stored sign-in. `--data` adds extra request fields as JSON, and accepts `@file` or `-` for standard input. + - Account setup runs from the CLI too: `strix cloud workspaces list|create|use` (`workspace` is an alias and `use` accepts a displayed number, name, or ID), `strix cloud session scopes|scopes set`, `strix cloud org members invite`, `strix cloud billing subscribe --plan strix_cloud`, `strix cloud billing portal`, `strix cloud integrations install github`, and `strix cloud domains verify `. Workspace switching preserves the server-side profile and can never widen past the login ceiling; ordinary switches do not reprompt. The last four end at a person: the command prints a link or a DNS record for the user to open or add, and it never completes the payment, installation, or DNS change for them. + - Every REST operation has a `strix cloud ` command. Run `strix cloud` to list them. Output is JSON when stdout is not a terminal (or with `--json`), and there are no prompts without a TTY. Binary downloads are the exception: redirect raw bytes intentionally, or combine `--output FILE --json` for structured download metadata. Exit codes: `0` success, `1` error, `2` usage, `4` auth or plan limit, `5` payment required. `--token` or `STRIX_API_TOKEN` is a stateless override and never replaces stored auth; set `--workspace-id`/`STRIX_WORKSPACE_ID` for an override CLI session. `--data` adds extra request fields as JSON, and accepts `@file` or `-` for standard input. - Local source uploads require `uploads:write`. For an agent/CI handoff, review `scans start --source . --dry-run --show-files --json`, capture `source.archive_sha256`, then rerun with the same `--source`, `--exclude`, and `--include-*` selection flags plus `--approve-sha256 HASH`. A changed snapshot is rejected. `--yes` approves only the snapshot built in that invocation, so reserve it for a deliberate human or one-shot approval rather than a digest-bound two-step handoff. - Git ignores, hidden files, `.git`, symlinks, dependency/build output, secret-like filenames, and nested archives are excluded by default; `.strixignore` and `--exclude` narrow the manifest further (a trailing `/` excludes a directory subtree). Limits: 20,000 files, 25 MiB/file, 250 MiB expanded, 50 MiB compressed. Source-only infers `code_review`; source plus a domain infers `live_test`. - The temporary local archive is always removed. A staged upload is deleted after a definitive rejection, but retained when a network error, `5xx`, malformed success response, or interruption leaves the scan launch ambiguous. JSON reports its `upload_id` with `launch_outcome_unknown: true`, or with `cleanup_unknown: true` when automatic deletion cannot be confirmed. Check `scans list` before retrying; if no scan is linked, run `uploads delete UPLOAD_ID`. diff --git a/README.md b/README.md index 169f0a14..6cb6aaac 100644 --- a/README.md +++ b/README.md @@ -325,15 +325,18 @@ strix auth logout # forget the sign-in The `strix cloud` commands drive the managed platform ([app.strix.ai](https://app.strix.ai)) from the terminal. Sign in once with the device flow. The sign-in creates your account and workspace on first use and stores a personal API token in `~/.strix/platform-auth.json`: ```bash -strix cloud login # browser approval, then workspace + scope picker +strix cloud login # browser approval, then workspace + scope profile strix cloud login --workspace "My Team" # select a workspace by name or ID -strix cloud whoami # show the active sign-in and scopes -strix cloud logout # forget the sign-in +strix cloud whoami # fast local account/workspace status +strix cloud session # verify remote session + consent ceiling +strix cloud logout # revoke remotely, then remove locally ``` The default **Recommended** scope preset supports normal scan work, local source uploads, -and workspace switching. For strict least privilege, pass an explicit list such as -`--scopes scans:read scans:write uploads:write billing:read`. +workspace switching, and user-approved credit top-ups. It excludes credential creation; +request `tokens:write` explicitly (or choose Full) when needed. For strict least privilege, pass an explicit list such as +`--scopes scans:read scans:write uploads:write billing:read`. Named automation +profiles are also available with `--scope-profile minimal|recommended|full`. Every operation of the [REST API](https://docs.app.strix.ai) has a matching command in the form `strix cloud `: @@ -357,6 +360,8 @@ Workspaces and account setup also work from the terminal: strix cloud workspaces list # numbered list; `workspace` is also accepted strix cloud workspaces create --name "My Team" # admin + organizations:write strix cloud workspaces use 2 # switch by list number, exact name, or ID +strix cloud session scopes # granted scopes + login ceiling +strix cloud session scopes set minimal # narrow without another browser sign-in strix cloud billing subscribe --plan strix_cloud # opens the hosted checkout page strix cloud billing portal # opens the billing portal strix cloud integrations install github # opens the app installation page @@ -365,7 +370,15 @@ strix cloud domains verify # prints the DNS record to add The last four commands end at a person. Strix creates the link, opens the browser for an interactive terminal, and always prints the URL. The user enters the card, approves the installation, or adds the DNS record. Pass `--no-browser` to print the URL only. -The commands work for humans and agents: terminal output favors names, branches, lifecycle states, and numbered selectors, while redirected output (or `--json`) preserves complete machine-readable records and IDs. Human lists retain the selectors needed by follow-up commands but omit internal organization/user IDs; a selector too long for the compact table is repeated losslessly in a copyable block. Paginated lists print the next `--page` or `--offset`, and detail views preserve useful prose within a safe terminal bound; use `--json` for the complete record. Token lists label credentials as active, expired, or revoked. Binary downloads are the exception: intentionally redirect their raw bytes, or use `--output FILE --json` to write the file and receive structured download metadata. There are no prompts when stdin is not a terminal. Exit codes: `0` success, `1` error, `2` invalid usage, `4` authentication or plan limit, `5` payment required. Set the token with `--token` or `STRIX_API_TOKEN` to skip the stored sign-in. +The commands work for humans and agents: terminal output favors names, branches, lifecycle states, and numbered selectors, while redirected output (or `--json`) preserves complete machine-readable records and IDs. Human lists retain the selectors needed by follow-up commands but omit internal organization/user IDs; a selector too long for the compact table is repeated losslessly in a copyable block. Paginated lists print the next `--page` or `--offset`, and detail views preserve useful prose within a safe terminal bound; use `--json` for the complete record. Token lists distinguish API keys from named CLI device sessions. Binary downloads are the exception: intentionally redirect their raw bytes, or use `--output FILE --json` to write the file and receive structured download metadata. There are no prompts when stdin is not a terminal. Exit codes: `0` success, `1` error, `2` invalid usage, `4` authentication or plan limit, `5` payment required. `--token` and `STRIX_API_TOKEN` are stateless per-command overrides and never replace the stored sign-in; pair a CLI-session override with `--workspace-id` or `STRIX_WORKSPACE_ID`. + +A browser sign-in creates one reusable credential per CLI installation. Logging in again on the +same installation replaces its secret instead of accumulating keys. Workspace switches keep that +credential and expiry, preserve the server-side scope preference, cap access by the target role, +and can never exceed the login consent ceiling. Each process pins its starting workspace, so a +concurrent switch fails safely instead of sending a stale command to another organization. +`strix cloud logout` revokes the server session before deleting the local token; use +`--local-only` only when you deliberately cannot reach the server. Write commands take request fields as flags, and every write command also accepts one JSON object with `--data`: diff --git a/skills/managed-pentesting-with-strix/SKILL.md b/skills/managed-pentesting-with-strix/SKILL.md index 48800ae4..14ff6f52 100644 --- a/skills/managed-pentesting-with-strix/SKILL.md +++ b/skills/managed-pentesting-with-strix/SKILL.md @@ -44,12 +44,15 @@ Run the device sign-in. It creates the user's account and workspace on first use strix cloud login # Non-interactive least-privilege example: strix cloud login --scopes scans:read scans:write uploads:write billing:read vulnerabilities:read assets:read assets:write +# Or use a stable named profile: +strix cloud login --scope-profile recommended ``` -The user approves the sign-in in the browser. With `--scopes` (and optionally `--workspace `) there are no terminal prompts, so the command works from a non-interactive agent shell. In an interactive terminal without flags, the CLI offers a workspace picker and scope presets (Recommended, Full access, Minimal, Custom). Recommended covers ordinary scans, source uploads, and workspace switching; use explicit scopes for a narrower automation token. +The user approves the sign-in in the browser. With `--scopes` (and optionally `--workspace `) there are no terminal prompts, so the command works from a non-interactive agent shell. In an interactive terminal without flags, the CLI offers a workspace picker and scope presets (Recommended, Full access, Minimal, Custom). Recommended covers ordinary scans, source uploads, workspace switching, and user-approved credit top-ups; it excludes `tokens:write`, which must be requested explicitly when credential management is required. Use explicit scopes for a narrower automation token. -- `strix cloud whoami` shows the active sign-in; add `--json` when another agent will parse it. `strix cloud logout` removes it. -- Every other `strix cloud` command uses the stored token automatically. To use a different token (for example one created in the dashboard at **Settings → API Access**), pass `--token ` or set `STRIX_API_TOKEN`. +- `strix cloud whoami` is the fast local status. `strix cloud session --json` verifies the remote device session; `strix cloud session scopes` shows both effective access and the immutable login ceiling. +- `strix cloud logout` revokes the remote session before removing the local token. On a network or server failure it keeps the token so the user can retry; `--local-only` deliberately skips revocation. +- Every other `strix cloud` command uses the stored token automatically. `--token ` or `STRIX_API_TOKEN` is a stateless per-command override and never overwrites the stored account. For an override that is itself a CLI session, also pass `--workspace-id` or set `STRIX_WORKSPACE_ID`. - Never hardcode, log, or commit the token. Store it in an env var or the CI secret store. - **Scopes (least-privilege):** assign only what the integration needs and rotate regularly: @@ -62,9 +65,9 @@ The user approves the sign-in in the browser. With `--scopes` (and optionally `- | `pr_reviews:write` | trigger PR security reviews | | `webhooks:read` / `:write` | manage webhook subscriptions | | `uploads:write` | upload local source or documents for a scan | - | `organizations:read` | list workspaces | + | `organizations:read` | read organization details (listing/switching the signed-in user's workspaces needs no API scope) | | `organizations:write` | create/update workspaces (admin) | - | `tokens:write` | switch workspaces · create/revoke API tokens | + | `tokens:write` | create/revoke ordinary API tokens (not needed to manage the current CLI session) | | `knowledge:read` / `:write` | read/update organization knowledge | | `audit:read` | read/export the Enterprise audit log | | `billing:read` / `billing:write` | read credit balance & auto top-up settings · buy credits (admin) | @@ -125,10 +128,12 @@ strix cloud workspaces list # numbered name/role/current list strix cloud workspaces create --name "My Team" # admin + organizations:write strix cloud workspaces use 2 # displayed number, exact name, or ID strix cloud workspace use "My Team" # singular `workspace` alias also works +strix cloud session scopes # effective scopes + consent ceiling +strix cloud session scopes set minimal # narrow the session strix cloud org members invite --email dev@example.com --role analyst ``` -`workspaces use` retargets the current personal token to a workspace the user already belongs to and stores the updated workspace metadata; the bearer secret and expiry stay unchanged. The role in the target workspace limits the scopes. Add `--scopes` to request a smaller set. +`workspaces use` retargets the current personal token to a workspace the user already belongs to and stores the updated workspace metadata; the bearer secret and expiry stay unchanged. It does not reprompt during ordinary switches: the server preserves the chosen profile, enforces the immutable login ceiling, and caps effective scopes by the target role. Use `--scope-profile` or `--scopes` to narrow within that ceiling; broader consent requires `strix cloud login` again. The CLI pins each process to the workspace it started in, so concurrent shells fail with a recoverable conflict instead of silently crossing organizations. ### Handoffs a person must finish diff --git a/strix/interface/cloud/__init__.py b/strix/interface/cloud/__init__.py index 4db4f13b..37250e9e 100644 --- a/strix/interface/cloud/__init__.py +++ b/strix/interface/cloud/__init__.py @@ -17,6 +17,7 @@ from rich.markup import escape import strix.interface.cloud.http as http # noqa: PLR0402 from strix.interface.cloud.render import json_mode from strix.interface.cloud.runner import resolve, run +from strix.interface.cloud.session import run_session from strix.interface.cloud.spec import DEFAULT_VERBS, GROUP_HELP, SPEC from strix.interface.cloud.workspaces import run_workspace_use from strix.interface.platform_cli import run_login @@ -29,6 +30,7 @@ _USAGE_HEADER = """[bold]Usage:[/] strix cloud [arguments] login Sign in to the managed platform and store an API token logout Remove the stored API token whoami Show the stored account, workspace, and token state + session Inspect or narrow the remote CLI session credits Show the credit balance of the workspace [bold]Resource commands:[/]""" @@ -83,6 +85,8 @@ def _run_cloud(argv: list[str]) -> int: # noqa: PLR0911, PLR0912 group = "workspaces" if group in ("login", "logout", "whoami"): return _run_session(console, group, rest) + if group == "session": + return run_session(rest) if group == "credits": group, rest = "billing", ["credits", *rest] if group == "workspaces" and rest and rest[0] == "use": @@ -159,7 +163,7 @@ def _print_verbs( def _print_usage_json() -> None: payload = { "command": "strix cloud", - "session_commands": ["login", "logout", "whoami", "credits"], + "session_commands": ["login", "logout", "whoami", "session", "credits"], "resource_commands": [{"name": group, "help": GROUP_HELP.get(group, "")} for group in SPEC], } sys.stdout.write(json.dumps(payload, indent=2) + "\n") diff --git a/strix/interface/cloud/http.py b/strix/interface/cloud/http.py index 36858c70..3d11688e 100644 --- a/strix/interface/cloud/http.py +++ b/strix/interface/cloud/http.py @@ -24,6 +24,7 @@ _SUPABASE_STORAGE_HOST = re.compile(r"^[a-z0-9-]+\.supabase\.co$") _STORAGE_PATH_PREFIX = "/storage/v1/" _app_url_override: str | None = None _token_override_active = False +_workspace_id_override: str | None = None _timeout_s: float = _DEFAULT_TIMEOUT_S EXIT_OK = 0 @@ -51,11 +52,26 @@ def configure( base_url: str | None = None, timeout: float | None = None, token_override: bool = False, + workspace_id: str | None = None, ) -> None: """Set the platform URL and the request timeout for this process.""" global _app_url_override, _timeout_s, _token_override_active # noqa: PLW0603 + global _workspace_id_override # noqa: PLW0603 _app_url_override = base_url.rstrip("/") if base_url else None _token_override_active = token_override + explicit_workspace = workspace_id or os.environ.get("STRIX_WORKSPACE_ID") + if explicit_workspace: + _workspace_id_override = explicit_workspace.strip() + elif not token_override and not os.environ.get("STRIX_API_TOKEN"): + record = read_record() + stored_workspace = record.get("organization_id") if record is not None else None + _workspace_id_override = ( + stored_workspace.strip() + if isinstance(stored_workspace, str) and stored_workspace.strip() + else None + ) + else: + _workspace_id_override = None if timeout is not None: if not math.isfinite(timeout) or timeout <= 0: raise CloudError( @@ -140,7 +156,15 @@ def request( idempotency_key: str | None = None, ) -> requests.Response: url = f"{app_url()}/api/v1{path}" - headers = {"Authorization": f"Bearer {api_token(token)}"} + headers = { + "Authorization": f"Bearer {api_token(token)}", + } + bypass = os.environ.get("STRIX_VERCEL_PROTECTION_BYPASS", "").strip() + if bypass: + headers["x-vercel-protection-bypass"] = bypass + workspace_id = _expected_workspace_id(token_override=token is not None) + if workspace_id: + headers["X-Strix-Workspace"] = workspace_id if idempotency_key is not None: headers["Idempotency-Key"] = idempotency_key try: @@ -164,6 +188,15 @@ def request( return response +def _expected_workspace_id(*, token_override: bool) -> str | None: + """Pin every request in this process to the workspace selected at startup.""" + if _workspace_id_override: + return _workspace_id_override + if token_override or _token_override_active or os.environ.get("STRIX_API_TOKEN"): + return None + return None + + def upload_file(signed_url: str, upload_token: str, path: Path) -> None: """Stream a file to a platform-issued storage URL.""" _validate_upload_url(signed_url) diff --git a/strix/interface/cloud/payment_proxy.py b/strix/interface/cloud/payment_proxy.py index 36e31598..54b52a05 100644 --- a/strix/interface/cloud/payment_proxy.py +++ b/strix/interface/cloud/payment_proxy.py @@ -8,6 +8,7 @@ requests (challenge and paid retry) to the fixed billing endpoint. from __future__ import annotations +import os import secrets import threading from contextlib import contextmanager, suppress @@ -163,6 +164,9 @@ def _make_handler(state: _BridgeState) -> type[BaseHTTPRequestHandler]: headers = _forward_request_headers(self) headers["X-Strix-Authorization"] = state.authorization + bypass = os.environ.get("STRIX_VERCEL_PROTECTION_BYPASS", "").strip() + if bypass: + headers["x-vercel-protection-bypass"] = bypass try: response = requests.request( "POST", diff --git a/strix/interface/cloud/runner.py b/strix/interface/cloud/runner.py index 2c991f1d..db7169fc 100644 --- a/strix/interface/cloud/runner.py +++ b/strix/interface/cloud/runner.py @@ -198,6 +198,7 @@ def run(group: str, verb_label: str, cmd: Cmd, argv: list[str]) -> int: base_url=getattr(args, "app_url", None), timeout=getattr(args, "timeout", None), token_override=bool(token), + workspace_id=getattr(args, "workspace_id", None), ) query = _collect(args, cmd.query) body = _collect(args, cmd.body) @@ -597,6 +598,12 @@ def _build_parser(group: str, verb_label: str, cmd: Cmd) -> argparse.ArgumentPar json_help = "Print JSON results, or download metadata when exporting with --output." parser.add_argument("--json", action="store_true", help=json_help) parser.add_argument("--token", default=None, help="API token override.") + parser.add_argument( + "--workspace-id", + default=None, + metavar="ORG_ID", + help="Expected workspace for an override CLI token (or STRIX_WORKSPACE_ID).", + ) parser.add_argument("--app-url", default=None, metavar="URL", help="Platform URL override.") parser.add_argument( "--timeout", diff --git a/strix/interface/cloud/session.py b/strix/interface/cloud/session.py new file mode 100644 index 00000000..5089bdce --- /dev/null +++ b/strix/interface/cloud/session.py @@ -0,0 +1,161 @@ +"""Inspect and safely narrow a managed Strix CLI session.""" + +from __future__ import annotations + +import os +from typing import TYPE_CHECKING, Any, cast + +from rich.console import Console +from rich.markup import escape + +import strix.interface.cloud.http as http # noqa: PLR0402 +from strix.interface.cloud.arguments import CloudArgumentParser +from strix.interface.cloud.render import emit, json_mode +from strix.interface.platform_cli import read_record, save_record +from strix.interface.terminal_text import sanitize_terminal_text + + +if TYPE_CHECKING: + import argparse + + +def run_session(argv: list[str]) -> int: + console = Console() + normalized = ["show", *argv] if not argv or argv[0].startswith("-") else list(argv) + if normalized[0] == "help": + normalized = ["--help", *normalized[1:]] + if normalized[0] in {"-h", "--help"}: + _print_help(console) + return 0 + verb = normalized.pop(0) + if verb == "scopes" and normalized and normalized[0] == "set": + normalized.pop(0) + return _run_scopes_set(console, normalized) + if verb not in {"show", "scopes"}: + console.print(f"[red]Unknown session command:[/] {escape(sanitize_terminal_text(verb))}") + _print_help(console) + return http.EXIT_USAGE + return _run_show(console, normalized, scopes_only=verb == "scopes") + + +def _common(parser: argparse.ArgumentParser) -> None: + parser.add_argument("--json", action="store_true", help="Print the raw JSON response.") + parser.add_argument("--show-scopes", action="store_true", help="Print every granted scope.") + parser.add_argument("--token", default=None, help="API token override.") + parser.add_argument("--workspace-id", default=None, metavar="ORG_ID") + parser.add_argument("--app-url", default=None, metavar="URL") + parser.add_argument("--timeout", default=None, type=float, metavar="SECONDS") + + +def _configure(args: argparse.Namespace) -> bool: + external = args.token is not None or bool(os.environ.get("STRIX_API_TOKEN", "").strip()) + http.configure( + base_url=args.app_url, + timeout=args.timeout, + token_override=bool(args.token), + workspace_id=args.workspace_id, + ) + return external + + +def _run_show(console: Console, argv: list[str], *, scopes_only: bool) -> int: + parser = CloudArgumentParser(prog=f"strix cloud session {'scopes' if scopes_only else 'show'}") + _common(parser) + as_json = json_mode(flag="--json" in argv) + try: + args = parser.parse_args(argv) + _configure(args) + payload = http.check(http.request("GET", "/cli/session", token=args.token)) + except SystemExit as exc: + return int(exc.code or 0) + except http.CloudError as exc: + return _error(console, exc, as_json=as_json) + if not isinstance(payload, dict): + return _error(console, http.CloudError("invalid CLI session response"), as_json=as_json) + record = cast("dict[str, Any]", payload) + if as_json: + emit(console, record, as_json=True) + return http.EXIT_OK + scopes = _string_list(record.get("scopes")) + ceiling = _string_list(record.get("scope_ceiling")) + profile = str(record.get("scope_profile") or "custom").title() + if not scopes_only: + device_name = escape(str(record.get("device_name") or "this device")) + console.print(f"[green]Active CLI session[/] on [bold]{device_name}[/]") + console.print(f" Workspace: {escape(str(record.get('organization_id') or 'unknown'))}") + console.print(f" Access: {profile} · {len(scopes)} scopes granted · {len(ceiling)} maximum") + if args.show_scopes or scopes_only: + console.print(f" Granted: [dim]{escape(' '.join(scopes))}[/]") + console.print(f" Ceiling: [dim]{escape(' '.join(ceiling))}[/]") + return http.EXIT_OK + + +def _run_scopes_set(console: Console, argv: list[str]) -> int: + parser = CloudArgumentParser( + prog="strix cloud session scopes set", + description="Change scopes within the access approved at browser sign-in.", + ) + mode = parser.add_mutually_exclusive_group(required=True) + mode.add_argument("profile", nargs="?", choices=("minimal", "recommended", "full")) + mode.add_argument("--scopes", nargs="+", metavar="SCOPE") + _common(parser) + as_json = json_mode(flag="--json" in argv) + try: + args = parser.parse_args(argv) + external = _configure(args) + body = ( + {"scope_profile": args.profile} + if args.profile + else {"scope_profile": "custom", "scopes": args.scopes} + ) + payload = http.check(http.request("PATCH", "/cli/session", token=args.token, body=body)) + except SystemExit as exc: + return int(exc.code or 0) + except http.CloudError as exc: + return _error(console, exc, as_json=as_json) + if not isinstance(payload, dict): + return _error(console, http.CloudError("invalid CLI session response"), as_json=as_json) + result = cast("dict[str, Any]", payload) + if not external: + stored = read_record() + if stored is not None: + stored.update( + { + key: result[key] + for key in ("scopes", "requested_scopes", "scope_ceiling", "scope_profile") + if key in result + } + ) + save_record(stored) + if as_json: + emit(console, result, as_json=True) + else: + scopes = _string_list(result.get("scopes")) + profile = str(result.get("scope_profile") or "custom").title() + console.print(f"[green]✓ CLI access updated.[/] {profile} · {len(scopes)} scopes granted") + if args.show_scopes: + console.print(f" Scopes: [dim]{escape(' '.join(scopes))}[/]") + return http.EXIT_OK + + +def _string_list(value: Any) -> list[str]: + if not isinstance(value, list): + return [] + items = cast("list[Any]", cast("Any", value)) + return [str(item) for item in items] + + +def _error(console: Console, error: http.CloudError, *, as_json: bool) -> int: + if as_json: + emit(console, {"error": str(error)}, as_json=True) + else: + console.print(f"[red]Error:[/] {escape(sanitize_terminal_text(error))}") + return error.exit_code + + +def _print_help(console: Console) -> None: + console.print("[bold]strix cloud session[/] commands:") + console.print(" show Show the remote CLI session (default).") + console.print(" scopes Show granted scopes and consent ceiling.") + console.print(" scopes set PROFILE Use minimal, recommended, or full.") + console.print(" scopes set --scopes SCOPE… Use a custom set within the ceiling.") diff --git a/strix/interface/cloud/workspaces.py b/strix/interface/cloud/workspaces.py index 62e97136..40b558b9 100644 --- a/strix/interface/cloud/workspaces.py +++ b/strix/interface/cloud/workspaces.py @@ -18,6 +18,7 @@ import strix.interface.cloud.http as http # noqa: PLR0402 from strix.interface.cloud.arguments import CloudArgumentParser from strix.interface.cloud.render import emit, json_mode from strix.interface.platform_cli import AUTH_PATH, read_record, save_record +from strix.interface.platform_identity import read_or_create_identity from strix.interface.terminal_text import sanitize_terminal_text @@ -37,17 +38,32 @@ def run_workspace_use(argv: list[str]) -> int: metavar="WORKSPACE", help="Workspace number from `workspaces list`, ID, or exact name.", ) - parser.add_argument( + scope_mode = parser.add_mutually_exclusive_group() + scope_mode.add_argument( "--scopes", nargs="+", metavar="SCOPE", default=None, help=( - "API scopes after switching. Without this option, preserve the stored request scopes." + "Use a custom scope set within the login-approved ceiling. " + "Without this option, preserve the server-side scope preference." ), ) + scope_mode.add_argument( + "--scope-profile", + choices=("minimal", "recommended", "full"), + default=None, + help="Change to a profile within the authority approved at login.", + ) + parser.add_argument("--show-scopes", action="store_true", help="Print every granted scope.") parser.add_argument("--json", action="store_true", help="Print the raw JSON response.") parser.add_argument("--token", default=None, help="API token override.") + parser.add_argument( + "--workspace-id", + default=None, + metavar="ORG_ID", + help="Expected workspace for an override CLI token.", + ) parser.add_argument("--app-url", default=None, metavar="URL", help="Platform URL override.") parser.add_argument( "--timeout", default=None, type=float, metavar="SECONDS", help="Request timeout in seconds." @@ -67,6 +83,7 @@ def run_workspace_use(argv: list[str]) -> int: base_url=args.app_url, timeout=args.timeout, token_override=bool(args.token), + workspace_id=args.workspace_id, ) return _use(console, args, as_json=as_json) except http.CloudError as exc: @@ -74,7 +91,9 @@ def run_workspace_use(argv: list[str]) -> int: return exc.exit_code -def _use(console: Console, args: argparse.Namespace, *, as_json: bool) -> int: +def _use( # noqa: PLR0912, PLR0915 + console: Console, args: argparse.Namespace, *, as_json: bool +) -> int: workspace = _find_workspace(args.workspace, token=args.token) stored_record: dict[str, Any] = read_record() or {} # An override token may belong to a different account. Never mix its new @@ -84,15 +103,14 @@ def _use(console: Console, args: argparse.Namespace, *, as_json: bool) -> int: body: dict[str, Any] = {} if args.scopes: body["scopes"] = args.scopes - elif not external_token: - stored_scopes = stored_record.get("requested_scopes", stored_record.get("scopes")) - stored_scope_items = cast("list[Any]", cast("Any", stored_scopes)) - if ( - isinstance(stored_scopes, list) - and stored_scope_items - and all(isinstance(scope, str) for scope in stored_scope_items) - ): - body["scopes"] = [scope for scope in stored_scope_items if isinstance(scope, str)] + body["scope_profile"] = "custom" + elif args.scope_profile: + body["scope_profile"] = args.scope_profile + if not external_token: + try: + body.update(read_or_create_identity()) + except (OSError, ValueError) as exc: + raise http.CloudError(f"could not load the CLI device identity: {exc}") from exc switched = _switch_workspace_token( str(workspace["id"]), token=args.token, @@ -119,38 +137,33 @@ def _use(console: Console, args: argparse.Namespace, *, as_json: bool) -> int: "organization_name": switched_record.get( "organization_name", workspace.get("name", "") ), - "expires_at": switched_record.get("expires_at"), + "expires_at": switched_record.get("expires_at") or stored_record.get("expires_at"), "scopes": validated_scopes, - "requested_scopes": ( - list(args.scopes) - if args.scopes - else ( - stored_record.get( - "requested_scopes", - stored_record.get("scopes", []), - ) - if not external_token - else validated_scopes - ) - ), + "requested_scopes": switched_record.get("requested_scopes", validated_scopes), + "scope_ceiling": switched_record.get("scope_ceiling", []), + "scope_profile": switched_record.get("scope_profile", "custom"), + "token_id": switched_record.get("token_id"), + "credential_source": switched_record.get("credential_source", "api"), + "device_name": switched_record.get("device_name"), "app_url": http.app_url(), } ) if switched_record.get("email"): record["email"] = switched_record["email"] - try: - save_record(record) - except OSError as exc: - raise http.CloudError( - "the platform switched the token, but the local workspace metadata could not be " - f"stored in {AUTH_PATH}: {exc}. The bearer is still valid; fix the file and safely " - "rerun the same workspace use command.", - payload={ - "workspace_switched": True, - "local_record_updated": False, - "retry_safe": True, - }, - ) from exc + if not external_token: + try: + save_record(record) + except OSError as exc: + raise http.CloudError( + "the platform switched the token, but the local workspace metadata could not be " + f"stored in {AUTH_PATH}: {exc}. The bearer is still valid; fix the file and safely " + "rerun the same workspace use command.", + payload={ + "workspace_switched": True, + "local_record_updated": False, + "retry_safe": True, + }, + ) from exc result = { "workspace_id": record["organization_id"], @@ -166,10 +179,16 @@ def _use(console: Console, args: argparse.Namespace, *, as_json: bool) -> int: if isinstance(scopes, list) and scopes: scope_items = cast("list[Any]", cast("Any", scopes)) scope_names = [scope for scope in scope_items if isinstance(scope, str)] - if scope_names: + if scope_names and args.show_scopes: rendered_scopes = escape(sanitize_terminal_text(" ".join(scope_names))) console.print(f" Scopes: [dim]{rendered_scopes}[/]") - console.print(f" Token: stored in [dim]{escape(sanitize_terminal_text(AUTH_PATH))}[/]") + elif scope_names: + profile = str(record.get("scope_profile") or "custom").title() + console.print(f" Access: [dim]{profile} · {len(scope_names)} scopes granted[/]") + if external_token: + console.print(" Token: [dim]override used for this command only; not stored[/]") + else: + console.print(f" Token: stored in [dim]{escape(sanitize_terminal_text(AUTH_PATH))}[/]") return http.EXIT_OK diff --git a/strix/interface/completions.py b/strix/interface/completions.py index d1ebdfce..f188fc3e 100644 --- a/strix/interface/completions.py +++ b/strix/interface/completions.py @@ -11,10 +11,18 @@ from strix.interface.terminal_text import has_terminal_control, sanitize_termina _ROOT_COMMANDS = ("cloud", "auth", "view", "completions", "completion") -_SESSION_COMMANDS = ("login", "logout", "whoami", "credits") -_COMMON_FLAGS = ("--json", "--token", "--app-url", "--timeout", "-h", "--help") -_COMMON_VALUE_FLAGS = frozenset({"--token", "--app-url", "--timeout"}) -_WORKSPACE_USE_FLAGS = (*_COMMON_FLAGS, "--scopes") +_SESSION_COMMANDS = ("login", "logout", "whoami", "session", "credits") +_COMMON_FLAGS = ( + "--json", + "--token", + "--workspace-id", + "--app-url", + "--timeout", + "-h", + "--help", +) +_COMMON_VALUE_FLAGS = frozenset({"--token", "--workspace-id", "--app-url", "--timeout"}) +_WORKSPACE_USE_FLAGS = (*_COMMON_FLAGS, "--scopes", "--scope-profile", "--show-scopes") def run_completions(argv: list[str]) -> int: @@ -115,10 +123,26 @@ def _cloud_candidates(prior: list[str], current: str) -> list[str]: # noqa: PLR def _session_candidates(group: str, prior: list[str], current: str) -> list[str]: + if group == "session": + if not prior: + return _matching(("show", "scopes", "help", *_COMMON_FLAGS, "--show-scopes"), current) + if prior[:1] == ["scopes"] and len(prior) == 1: + return _matching(("set", *_COMMON_FLAGS, "--show-scopes"), current) + if prior[:2] == ["scopes", "set"]: + return _matching( + ("minimal", "recommended", "full", "--scopes", *_COMMON_FLAGS, "--show-scopes"), + current, + ) + return _flag_candidates( + (*_COMMON_FLAGS, "--show-scopes"), + prior, + current, + value_flags=_COMMON_VALUE_FLAGS | {"--scopes"}, + ) flags = _session_flags(group) value_flags: frozenset[str] = frozenset() if group == "login": - value_flags = frozenset({"--scopes", "--workspace"}) + value_flags = frozenset({"--scopes", "--scope-profile", "--workspace", "--device-name"}) elif group == "credits": value_flags = _COMMON_VALUE_FLAGS return _flag_candidates(flags, prior, current, value_flags=value_flags) @@ -126,11 +150,19 @@ def _session_candidates(group: str, prior: list[str], current: str) -> list[str] def _session_flags(group: str) -> tuple[str, ...]: if group == "login": - return ("--no-browser", "--scopes", "--workspace", "-h", "--help") + return ( + "--no-browser", + "--scopes", + "--scope-profile", + "--workspace", + "--device-name", + "-h", + "--help", + ) if group == "whoami": - return ("--json", "-h", "--help") + return ("--json", "--show-scopes", "-h", "--help") if group == "logout": - return ("--json", "-h", "--help") + return ("--json", "--local-only", "-h", "--help") if group == "credits": return _COMMON_FLAGS return ("-h", "--help") diff --git a/strix/interface/platform_cli.py b/strix/interface/platform_cli.py index 8c05d17e..1835d367 100644 --- a/strix/interface/platform_cli.py +++ b/strix/interface/platform_cli.py @@ -11,6 +11,7 @@ from __future__ import annotations import argparse import contextlib import json +import os import sys import time import webbrowser @@ -25,6 +26,7 @@ from rich.panel import Panel from rich.text import Text from strix.config import load_settings +from strix.interface.platform_identity import read_or_create_identity from strix.interface.terminal_text import sanitize_terminal_text from strix.interface.url_safety import is_safe_web_url from strix.utils.secret_files import write_secret_text @@ -107,7 +109,8 @@ def _login(console: Console, argv: list[str]) -> int: action="store_true", help="Do not open the browser. Print the verification URL instead.", ) - parser.add_argument( + scope_mode = parser.add_mutually_exclusive_group() + scope_mode.add_argument( "--scopes", nargs="+", metavar="SCOPE", @@ -118,6 +121,18 @@ def _login(console: Console, argv: list[str]) -> int: "Without this option, an interactive picker opens after the browser step." ), ) + scope_mode.add_argument( + "--scope-profile", + choices=("minimal", "recommended", "full"), + default=None, + help="Scope profile to approve. Defaults to an interactive choice in a TTY.", + ) + parser.add_argument( + "--device-name", + default=None, + metavar="NAME", + help="Privacy-safe label shown for this CLI session in the dashboard.", + ) parser.add_argument( "--workspace", metavar="WORKSPACE", @@ -128,6 +143,7 @@ def _login(console: Console, argv: list[str]) -> int: "more than one workspace." ), ) + previous_record = read_record() try: args = parser.parse_args(argv) except SystemExit as exc: # argparse already printed the message @@ -146,7 +162,9 @@ def _login(console: Console, argv: list[str]) -> int: console, open_browser=not args.no_browser, scopes=args.scopes, + scope_profile=args.scope_profile, workspace=args.workspace, + device_name=args.device_name, ) except PlatformAuthError as exc: console.print(f"[red]Sign-in failed:[/] {_terminal_markup(exc)}") @@ -166,23 +184,33 @@ def _login(console: Console, argv: list[str]) -> int: "then run `strix cloud login` again.[/]" ) return 1 + _revoke_replaced_legacy_session(previous_record, record) _print_success(console, record) return 0 -def _run_device_flow( +def _run_device_flow( # noqa: PLR0912, PLR0915 console: Console, *, open_browser: bool, scopes: list[str] | None = None, + scope_profile: str | None = None, workspace: str | None = None, + device_name: str | None = None, ) -> dict[str, Any]: app_url = _app_url() - interactive = workspace is not None or (sys.stdin.isatty() and scopes is None) + interactive = workspace is not None or ( + sys.stdin.isatty() and scopes is None and scope_profile is None + ) + try: + identity = read_or_create_identity(device_name=device_name) + except (OSError, ValueError) as exc: + raise PlatformAuthError(f"could not prepare the CLI device identity: {exc}") from exc try: response = requests.post( f"{app_url}/api/v1/cli/login", + headers=_preview_headers(), timeout=_HTTP_TIMEOUT_S, allow_redirects=False, ) @@ -230,11 +258,13 @@ def _run_device_flow( console.print("[dim]Waiting for browser confirmation…[/]") - poll_body: dict[str, Any] = {"device_code": device_code} + poll_body: dict[str, Any] = {"device_code": device_code, **identity} if interactive: poll_body["interactive"] = True elif scopes: poll_body["scopes"] = scopes + elif scope_profile: + poll_body["scope_profile"] = scope_profile deadline = time.monotonic() + expires_in while time.monotonic() < deadline: @@ -245,6 +275,7 @@ def _run_device_flow( try: poll = requests.post( f"{app_url}/api/v1/cli/login/poll", + headers=_preview_headers(), json=poll_body, timeout=_HTTP_TIMEOUT_S, allow_redirects=False, @@ -252,7 +283,14 @@ def _run_device_flow( except requests.RequestException: continue if 200 <= poll.status_code < 300: - return _finish_login(console, app_url, poll, scopes=scopes, workspace=workspace) + return _finish_login( + console, + app_url, + poll, + scopes=scopes, + scope_profile=scope_profile, + workspace=workspace, + ) delta = _handle_poll_error(poll) if delta is None: break @@ -283,24 +321,30 @@ def _finish_login( poll: requests.Response, *, scopes: list[str] | None, + scope_profile: str | None, workspace: str | None, ) -> dict[str, Any]: result = _json_object(poll) if result.get("selection_required"): - return _complete_selection(console, app_url, result, scopes=scopes, workspace=workspace) - return _bind_login_record(_require_api_token(result), app_url, scopes) + return _complete_selection( + console, + app_url, + result, + scopes=scopes, + scope_profile=scope_profile, + workspace=workspace, + ) + return _bind_login_record(_require_api_token(result), app_url) def _signed_in_record( response: requests.Response, *, app_url: str, - requested_scopes: list[str] | None, ) -> dict[str, Any]: return _bind_login_record( _require_api_token(_json_object(response)), app_url, - requested_scopes, ) @@ -311,9 +355,7 @@ def _require_api_token(record: dict[str, Any]) -> dict[str, Any]: return record -def _bind_login_record( - record: dict[str, Any], app_url: str, requested_scopes: list[str] | None -) -> dict[str, Any]: +def _bind_login_record(record: dict[str, Any], app_url: str) -> dict[str, Any]: """Bind a stored credential to its issuer and preserve its scope preference.""" parsed = urlsplit(app_url) if ( @@ -332,7 +374,7 @@ def _bind_login_record( bound["app_url"] = urlunsplit( (parsed.scheme.lower(), parsed.netloc.lower(), parsed.path.rstrip("/"), "", "") ) - preference: Any = requested_scopes if requested_scopes is not None else record.get("scopes") + preference: Any = record.get("requested_scopes", record.get("scopes")) preference_items = cast("list[Any]", preference) if isinstance(preference, list) and all(isinstance(scope, str) for scope in preference_items): bound["requested_scopes"] = list(dict.fromkeys(cast("list[str]", preference_items))) @@ -345,6 +387,7 @@ def _complete_selection( selection: dict[str, Any], *, scopes: list[str] | None, + scope_profile: str | None, workspace: str | None, ) -> dict[str, Any]: organizations = _dict_items(selection.get("organizations")) @@ -356,8 +399,9 @@ def _complete_selection( chosen_org = _choose_workspace(console, organizations, workspace) role = str(chosen_org.get("role") or "admin") chosen_scopes = scopes - if chosen_scopes is None and sys.stdin.isatty(): - chosen_scopes = _choose_scopes(console, catalog, role) + chosen_profile = scope_profile + if chosen_scopes is None and chosen_profile is None and sys.stdin.isatty(): + chosen_profile, chosen_scopes = _choose_scopes(console, catalog, role) body: dict[str, Any] = { "selection_token": selection_token, @@ -365,9 +409,13 @@ def _complete_selection( } if chosen_scopes is not None: body["scopes"] = chosen_scopes + body["scope_profile"] = "custom" + elif chosen_profile is not None: + body["scope_profile"] = chosen_profile try: response = requests.post( f"{app_url}/api/v1/cli/login/complete", + headers=_preview_headers(), json=body, timeout=_HTTP_TIMEOUT_S, allow_redirects=False, @@ -379,7 +427,6 @@ def _complete_selection( return _signed_in_record( response, app_url=app_url, - requested_scopes=chosen_scopes, ) @@ -433,20 +480,22 @@ def _choose_workspace( console.print("[yellow]Enter a number from the list.[/]") -def _choose_scopes(console: Console, catalog: list[dict[str, Any]], role: str) -> list[str] | None: - """Prompt for token scopes. Returns None to accept the server defaults.""" +def _choose_scopes( + console: Console, catalog: list[dict[str, Any]], role: str +) -> tuple[str, list[str] | None]: + """Prompt for a named scope profile or a custom scope list.""" rank = _ROLE_RANK.get(role, 2) allowed = [ item for item in catalog if _ROLE_RANK.get(str(item.get("min_role", "viewer")), 0) <= rank ] if not allowed: - return None + return "recommended", None console.print() console.print("[bold]Select token scopes:[/]") console.print( " [cyan]1[/]. Recommended [dim](scans, findings, schedules, assets, uploads, " - "workspace switching, billing)[/]" + "workspace switching, billing/top-ups; no token creation)[/]" ) console.print(" [cyan]2[/]. Full access [dim](every scope your role allows)[/]") console.print(" [cyan]3[/]. Minimal [dim](scan read/write and billing read)[/]") @@ -454,15 +503,13 @@ def _choose_scopes(console: Console, catalog: list[dict[str, Any]], role: str) - while True: answer = console.input("Scopes [1-4] (1): ").strip() or "1" if answer == "1": - return None + return "recommended", None if answer == "2": - return [str(item["scope"]) for item in allowed if item.get("scope")] + return "full", None if answer == "3": - return [ - str(item["scope"]) for item in allowed if item.get("scope") and item.get("minimum") - ] + return "minimal", None if answer == "4": - return _choose_custom_scopes(console, allowed) + return "custom", _choose_custom_scopes(console, allowed) console.print("[yellow]Enter a number from 1 to 4.[/]") @@ -534,6 +581,67 @@ def _error_detail(response: requests.Response) -> str: return f"HTTP {response.status_code}" +def _preview_headers() -> dict[str, str]: + bypass = os.environ.get("STRIX_VERCEL_PROTECTION_BYPASS", "").strip() + return {"x-vercel-protection-bypass": bypass} if bypass else {} + + +def _session_headers(record: dict[str, Any]) -> dict[str, str]: + headers = {"Authorization": f"Bearer {record['api_token']}"} + workspace_id = record.get("organization_id") + if isinstance(workspace_id, str) and workspace_id: + headers["X-Strix-Workspace"] = workspace_id + headers.update(_preview_headers()) + return headers + + +def _revoke_stored_session(record: dict[str, Any]) -> tuple[bool, str | None]: + """Revoke one server session; return (definitively_inactive, error).""" + app_url = record.get("app_url") + if not isinstance(app_url, str) or not app_url: + return False, ( + "the stored sign-in has no trusted platform URL; use --local-only to remove it" + ) + try: + response = requests.delete( + f"{app_url.rstrip('/')}/api/v1/cli/session", + headers=_session_headers(record), + timeout=_HTTP_TIMEOUT_S, + allow_redirects=False, + ) + except requests.RequestException as exc: + return False, f"could not revoke the remote CLI session: {exc}" + if response.status_code in {200, 204, 401}: + return True, None + return False, f"could not revoke the remote CLI session: {_error_detail(response)}" + + +def _print_logout_failure(console: Console, message: str, *, as_json: bool) -> int: + if as_json: + sys.stdout.write(json.dumps({"error": message, "removed": False}) + "\n") + else: + console.print(f"[red]Sign-out failed:[/] {_terminal_markup(message)}") + console.print("[dim]The local token was kept so you can safely retry.[/]") + return 1 + + +def _revoke_replaced_legacy_session( + previous: dict[str, Any] | None, current: dict[str, Any] +) -> None: + """Best-effort cleanup when the first device-aware login replaces a legacy token.""" + if not previous or previous.get("api_token") == current.get("api_token"): + return + if previous.get("app_url") != current.get("app_url"): + return + with contextlib.suppress(KeyError, requests.RequestException): + requests.delete( + f"{previous['app_url']}/api/v1/cli/session", + headers=_session_headers(previous), + timeout=_HTTP_TIMEOUT_S, + allow_redirects=False, + ) + + def _print_success(console: Console, record: dict[str, Any]) -> None: email = record.get("email", "") organization = record.get("organization_name") or record.get("organization_id", "") @@ -545,9 +653,7 @@ def _print_success(console: Console, record: dict[str, Any]) -> None: console.print(f" Workspace: [bold]{_terminal_markup(organization)}[/]") scopes = record.get("scopes") if isinstance(scopes, list) and scopes: - scope_items = cast("list[Any]", cast("Any", scopes)) - rendered_scopes = _terminal_markup(" ".join(str(scope) for scope in scope_items)) - console.print(f" Scopes: [dim]{rendered_scopes}[/]") + console.print(f" Access: [dim]{_terminal_markup(_scope_summary(record))}[/]") console.print(f" Token: stored in [dim]{_terminal_markup(AUTH_PATH)}[/]") console.print() console.print( @@ -556,12 +662,13 @@ def _print_success(console: Console, record: dict[str, Any]) -> None: ) -def _status(console: Console, argv: list[str]) -> int: +def _status(console: Console, argv: list[str]) -> int: # noqa: PLR0912 parser = _SessionArgumentParser( prog="strix cloud whoami", description="Show the stored managed-platform account, workspace, scopes, and expiry.", ) parser.add_argument("--json", action="store_true", help="Print the session as JSON.") + parser.add_argument("--show-scopes", action="store_true", help="Print every granted scope.") as_json = "--json" in argv or not sys.stdout.isatty() try: args = parser.parse_args(argv) @@ -607,18 +714,34 @@ def _status(console: Console, argv: list[str]) -> int: scopes = record.get("scopes") if isinstance(scopes, list) and scopes: scope_items = cast("list[Any]", cast("Any", scopes)) - console.print( - f" Scopes: {_terminal_markup(' '.join(str(scope) for scope in scope_items))}" - ) + if args.show_scopes: + console.print( + f" Scopes: {_terminal_markup(' '.join(str(scope) for scope in scope_items))}" + ) + else: + console.print(f" Access: {_terminal_markup(_scope_summary(record))}") return 0 -def _logout(console: Console, argv: list[str]) -> int: # noqa: PLR0911 +def _scope_summary(record: dict[str, Any]) -> str: + scopes = record.get("scopes") + scope_items = cast("list[Any]", cast("Any", scopes)) if isinstance(scopes, list) else [] + count = len(scope_items) + profile = str(record.get("scope_profile") or "custom").replace("_", " ").title() + return f"{profile} · {count} scope{'s' if count != 1 else ''} granted" + + +def _logout(console: Console, argv: list[str]) -> int: # noqa: PLR0911, PLR0912 parser = _SessionArgumentParser( prog="strix cloud logout", - description="Remove the managed-platform API token stored on this machine.", + description="Revoke this CLI session and remove its token from this machine.", ) parser.add_argument("--json", action="store_true", help="Print the result as JSON.") + parser.add_argument( + "--local-only", + action="store_true", + help="Remove only the local token, leaving the remote session active.", + ) as_json = "--json" in argv or not sys.stdout.isatty() try: args = parser.parse_args(argv) @@ -637,6 +760,13 @@ def _logout(console: Console, argv: list[str]) -> int: # noqa: PLR0911 return 0 console.print("[yellow]Not signed in.[/]") return 0 + record = read_record() + remotely_revoked = False + if record is not None and not args.local_only: + remotely_revoked, revoke_error = _revoke_stored_session(record) + if revoke_error: + return _print_logout_failure(console, revoke_error, as_json=as_json) + if not logout(): if as_json: sys.stdout.write( @@ -656,7 +786,23 @@ def _logout(console: Console, argv: list[str]) -> int: # noqa: PLR0911 ) return 1 if as_json: - sys.stdout.write(json.dumps({"signed_in": False, "removed": True}) + "\n") + sys.stdout.write( + json.dumps( + { + "signed_in": False, + "removed": True, + "remotely_revoked": remotely_revoked, + "local_only": bool(args.local_only), + } + ) + + "\n" + ) return 0 - console.print("[green]Signed out.[/] The stored API token was removed from this machine.") + if args.local_only: + console.print( + "[yellow]Local sign-out only.[/] The remote CLI session is still active; " + "revoke it from API Access if needed." + ) + else: + console.print("[green]Signed out.[/] The CLI session was revoked and removed locally.") return 0 diff --git a/strix/interface/platform_identity.py b/strix/interface/platform_identity.py new file mode 100644 index 00000000..508e7e39 --- /dev/null +++ b/strix/interface/platform_identity.py @@ -0,0 +1,46 @@ +"""Stable, privacy-safe identity for this Strix CLI installation.""" + +from __future__ import annotations + +import json +import platform +from pathlib import Path +from typing import Any, cast +from uuid import uuid4 + +from strix.utils.secret_files import write_secret_text + + +IDENTITY_PATH = Path.home() / ".strix" / "cli-identity.json" + + +def _default_device_name(instance_id: str) -> str: + system = {"Darwin": "macOS", "Windows": "Windows", "Linux": "Linux"}.get( + platform.system(), "Computer" + ) + return f"{system} CLI · {instance_id[:8]}" + + +def read_or_create_identity(*, device_name: str | None = None) -> dict[str, str]: + """Return one installation ID, optionally updating its user-facing label.""" + record: dict[str, Any] = {} + try: + raw = json.loads(IDENTITY_PATH.read_text(encoding="utf-8")) + if isinstance(raw, dict): + record = cast("dict[str, Any]", raw) + except (OSError, json.JSONDecodeError): + pass + + instance_id = record.get("client_instance_id") + if not isinstance(instance_id, str) or len(instance_id) < 8: + instance_id = str(uuid4()) + label = device_name.strip() if device_name is not None else record.get("device_name") + if not isinstance(label, str) or not label.strip(): + label = _default_device_name(instance_id) + label = " ".join(label.split()) + if not 1 <= len(label) <= 80: + raise ValueError("device name must be 1-80 printable characters") + + identity = {"client_instance_id": instance_id, "device_name": label} + write_secret_text(IDENTITY_PATH, json.dumps(identity, indent=2)) + return identity diff --git a/tests/test_cloud_cli.py b/tests/test_cloud_cli.py index e3aef155..6e5c2613 100644 --- a/tests/test_cloud_cli.py +++ b/tests/test_cloud_cli.py @@ -1671,12 +1671,15 @@ def test_workspaces_use_switches_stored_token( ) monkeypatch.setattr(http, "request", fake_request) + monkeypatch.setattr( + workspaces, + "read_or_create_identity", + lambda: {"client_instance_id": "client-test", "device_name": "Test CLI"}, + ) code = cloud.run_cloud(["workspaces", "use", "team one", "--json"]) assert code == 0 assert calls == [("GET", "/workspaces"), ("POST", "/workspaces/org_1/token")] - assert token_body == { - "scopes": ["scans:read", "scans:write", "organizations:read", "tokens:write"] - } + assert token_body == {"client_instance_id": "client-test", "device_name": "Test CLI"} record = platform_cli.read_record() assert record is not None assert record["api_token"] == "old" @@ -1727,11 +1730,9 @@ def test_workspace_use_explicit_token_starts_with_fresh_account_state( assert switch_body is None record = platform_cli.read_record() assert record is not None - assert record["api_token"] == "account-b-token" - assert record["organization_id"] == "org_b" - assert record["requested_scopes"] == ["scans:read", "organizations:read"] - assert "email" not in record - assert "account-a@example.test" not in auth_path.read_text(encoding="utf-8") + assert record["api_token"] == "account-a-token" + assert record["organization_id"] == "org_a" + assert record["email"] == "account-a@example.test" def test_workspace_use_environment_token_starts_with_fresh_account_state( @@ -1775,11 +1776,9 @@ def test_workspace_use_environment_token_starts_with_fresh_account_state( assert switch_body is None record = platform_cli.read_record() assert record is not None - assert record["api_token"] == "account-b-token" - assert record["organization_id"] == "org_b" - assert record["email"] == "account-b@example.test" - assert record["requested_scopes"] == ["scans:read", "organizations:read"] - assert "account-a@example.test" not in auth_path.read_text(encoding="utf-8") + assert record["api_token"] == "account-a-token" + assert record["organization_id"] == "org_a" + assert record["email"] == "account-a@example.test" def test_workspaces_use_reports_unknown_workspace(monkeypatch: pytest.MonkeyPatch) -> None: @@ -1796,6 +1795,8 @@ def test_workspaces_use_reports_unknown_workspace(monkeypatch: pytest.MonkeyPatc def test_workspaces_use_reports_auth_storage_failure( monkeypatch: pytest.MonkeyPatch, capsys: Any ) -> None: + monkeypatch.delenv("STRIX_API_TOKEN", raising=False) + def fake_request(method: str, path: str, **_kwargs: Any) -> FakeResponse: if method == "GET": return FakeResponse(payload={"workspaces": [{"id": "org_1", "name": "Team One"}]}) diff --git a/tests/test_cloud_cli_runtime.py b/tests/test_cloud_cli_runtime.py index 2aa5f39f..2d6ccc93 100644 --- a/tests/test_cloud_cli_runtime.py +++ b/tests/test_cloud_cli_runtime.py @@ -387,7 +387,7 @@ def test_session_human_output_neutralizes_server_control_sequences() -> None: rendered = output.getvalue() assert "\x1b" not in rendered assert "\x07" not in rendered - assert rendered.count("\\x1b]52;c;copied\\x07\\x1b[2J") == 3 + assert rendered.count("\\x1b]52;c;copied\\x07\\x1b[2J") == 2 def test_device_login_rejects_non_http_verification_url( @@ -775,7 +775,7 @@ def test_session_help_is_specific_and_human_whoami_shows_scopes( who_help = capsys.readouterr().out assert "strix cloud whoami" in who_help assert "--no-browser" not in who_help - assert cloud.run_cloud(["whoami"]) == 0 + assert cloud.run_cloud(["whoami", "--show-scopes"]) == 0 assert "scans:read organizations:read" in capsys.readouterr().out @@ -791,14 +791,25 @@ def test_non_tty_whoami_and_logout_emit_json( "email": "agent@example.test", "organization_name": "Demo", "scopes": ["scans:read"], + "app_url": "https://app.example.test", } ) assert cloud.run_cloud(["whoami"]) == 0 assert json.loads(capsys.readouterr().out)["email"] == "agent@example.test" + monkeypatch.setattr( + platform_cli.requests, + "delete", + lambda *_args, **_kwargs: type("Response", (), {"status_code": 200})(), + ) assert cloud.run_cloud(["logout"]) == 0 - assert json.loads(capsys.readouterr().out) == {"signed_in": False, "removed": True} + assert json.loads(capsys.readouterr().out) == { + "signed_in": False, + "removed": True, + "remotely_revoked": True, + "local_only": False, + } assert not platform_cli.AUTH_PATH.exists() @@ -806,14 +817,11 @@ def test_scope_picker_labels_match_the_server_presets() -> None: output = io.StringIO() console = Console(file=output, width=120) console.input = lambda *_args, **_kwargs: "1" # type: ignore[method-assign] - assert ( - platform_cli._choose_scopes( - console, - [{"scope": "scans:read", "min_role": "viewer", "minimum": True}], - "admin", - ) - is None - ) + assert platform_cli._choose_scopes( + console, + [{"scope": "scans:read", "min_role": "viewer", "minimum": True}], + "admin", + ) == ("recommended", None) rendered = output.getvalue() assert "uploads" in rendered assert "workspace switching" in rendered diff --git a/tests/test_cloud_session.py b/tests/test_cloud_session.py new file mode 100644 index 00000000..5a4d1ea1 --- /dev/null +++ b/tests/test_cloud_session.py @@ -0,0 +1,171 @@ +"""CLI-session lifecycle, scope, and workspace-race behavior.""" + +from __future__ import annotations + +import json +from typing import TYPE_CHECKING, Any + +import pytest + +from strix.interface import cloud, platform_cli, platform_identity +from strix.interface.cloud import http + + +if TYPE_CHECKING: + from pathlib import Path + + +class Response: + def __init__(self, payload: Any = None, status_code: int = 200) -> None: + self._payload = payload + self.status_code = status_code + self.ok = 200 <= status_code < 400 + self.text = json.dumps(payload) if payload is not None else "" + self.headers = {"content-type": "application/json"} + + def json(self) -> Any: + return self._payload + + +@pytest.fixture +def auth_path(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + path = tmp_path / "platform-auth.json" + monkeypatch.setattr(platform_cli, "AUTH_PATH", path) + monkeypatch.delenv("STRIX_API_TOKEN", raising=False) + monkeypatch.delenv("STRIX_WORKSPACE_ID", raising=False) + monkeypatch.delenv("STRIX_VERCEL_PROTECTION_BYPASS", raising=False) + return path + + +def test_http_workspace_pin_is_captured_once( + auth_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + platform_cli.save_record( + { + "api_token": "secret", + "organization_id": "org_start", + "app_url": "https://app.example.test", + } + ) + assert auth_path.exists() + sent: list[dict[str, str]] = [] + + def fake_request(*_args: Any, **kwargs: Any) -> Response: + sent.append(dict(kwargs["headers"])) + return Response({}) + + monkeypatch.setattr(http.requests, "request", fake_request) + http.configure() + platform_cli.save_record( + { + "api_token": "secret", + "organization_id": "org_changed_elsewhere", + "app_url": "https://app.example.test", + } + ) + http.request("GET", "/scans") + assert sent[0]["X-Strix-Workspace"] == "org_start" + + +def test_preview_bypass_is_opt_in_and_shared_by_login_and_api_requests( + auth_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + assert auth_path.parent.exists() + platform_cli.save_record( + { + "api_token": "secret", + "organization_id": "org_1", + "app_url": "https://preview.example.test", + } + ) + sent: list[dict[str, str]] = [] + monkeypatch.setenv("STRIX_VERCEL_PROTECTION_BYPASS", "preview-secret") + monkeypatch.setattr( + http.requests, + "request", + lambda *_args, **kwargs: sent.append(dict(kwargs["headers"])) or Response({}), + ) + + http.configure() + http.request("GET", "/scans") + + assert sent[0]["x-vercel-protection-bypass"] == "preview-secret" + assert platform_cli._preview_headers() == {"x-vercel-protection-bypass": "preview-secret"} + + +def test_session_scope_update_persists_only_for_stored_session( + auth_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: Any +) -> None: + platform_cli.save_record( + { + "api_token": "secret", + "organization_id": "org_1", + "app_url": "https://app.example.test", + "scopes": ["scans:read"], + } + ) + monkeypatch.setattr( + http, + "request", + lambda *_args, **_kwargs: Response( + { + "scopes": ["scans:read", "scans:write", "billing:read"], + "requested_scopes": ["scans:read", "scans:write", "billing:read"], + "scope_ceiling": ["scans:read", "scans:write", "billing:read"], + "scope_profile": "minimal", + } + ), + ) + assert cloud.run_cloud(["session", "scopes", "set", "minimal", "--json"]) == 0 + stored = platform_cli.read_record() + assert stored is not None + assert stored["scope_profile"] == "minimal" + assert json.loads(capsys.readouterr().out)["scope_profile"] == "minimal" + + before = auth_path.read_text(encoding="utf-8") + assert ( + cloud.run_cloud(["session", "scopes", "set", "minimal", "--token", "override", "--json"]) + == 0 + ) + assert auth_path.read_text(encoding="utf-8") == before + + +def test_logout_keeps_local_token_when_remote_outcome_is_not_definitive( + auth_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: Any +) -> None: + platform_cli.save_record( + { + "api_token": "secret", + "organization_id": "org_1", + "app_url": "https://app.example.test", + } + ) + monkeypatch.setattr( + platform_cli.requests, + "delete", + lambda *_args, **_kwargs: Response({"detail": "unavailable"}, 503), + ) + assert cloud.run_cloud(["logout", "--json"]) == 1 + assert auth_path.exists() + assert json.loads(capsys.readouterr().out)["removed"] is False + + +def test_local_only_logout_is_explicit_and_recoverable(auth_path: Path, capsys: Any) -> None: + platform_cli.save_record({"api_token": "secret"}) + assert cloud.run_cloud(["logout", "--local-only", "--json"]) == 0 + payload = json.loads(capsys.readouterr().out) + assert payload["local_only"] is True + assert payload["remotely_revoked"] is False + assert not auth_path.exists() + + +def test_cli_device_identity_is_stable_and_privacy_safe( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + path = tmp_path / "cli-identity.json" + monkeypatch.setattr(platform_identity, "IDENTITY_PATH", path) + first = platform_identity.read_or_create_identity() + second = platform_identity.read_or_create_identity(device_name=" Build laptop ") + assert second["client_instance_id"] == first["client_instance_id"] + assert second["device_name"] == "Build laptop" + assert path.stat().st_mode & 0o777 == 0o600