scriptoriumadmin/strix
1
0
Fork 0
mirror of https://github.com/usestrix/strix.git synced 2026-09-08 22:21:05 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Make cloud workspace switching session-safe

Browse source
This commit is contained in:
bearsyankees 2026-08-28 15:23:22 -04:00
parent 6d83e96b6f
commit 192e041a35
15 changed files with 775 additions and 125 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
AGENTS.md
View file

@ -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 <id>`. 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 <resource> <verb>` 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 <id>`. 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 <resource> <verb>` 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`.

25
README.md
View file

@ -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 <resource> <verb>`:
@ -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 <domain-id> # 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`:

17
skills/managed-pentesting-with-strix/SKILL.md
View file

@ -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 <name-or-id>`) 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 <name-or-id>`) 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 <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 <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

6
strix/interface/cloud/__init__.py
View file

@ -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 <command> [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")

35
strix/interface/cloud/http.py
View file

@ -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)

4
strix/interface/cloud/payment_proxy.py
View file

@ -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",

7
strix/interface/cloud/runner.py
View file

@ -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",

161
strix/interface/cloud/session.py Normal file
View file

@ -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.")

99
strix/interface/cloud/workspaces.py
View file

@ -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

48
strix/interface/completions.py
View file

@ -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")

220
strix/interface/platform_cli.py
View file

@ -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

46
strix/interface/platform_identity.py Normal file
View file

@ -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

27
tests/test_cloud_cli.py
View file

@ -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"}]})

30
tests/test_cloud_cli_runtime.py
View file

@ -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

171
tests/test_cloud_session.py Normal file
View file

@ -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