diff --git a/AGENTS.md b/AGENTS.md index 451d0e91..5114676d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -51,7 +51,7 @@ Target-specific workflows built on the same engine: - 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. - 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`, 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`. + - 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`. - Non-Enterprise scans consume the scope estimate (a default-tier source-only review currently starts at 60 credits); Enterprise scans are plan-included. A rejected launch does not consume credits. - Human output is compact and numbered; non-TTY output and `--json` retain full records. Enable tab completion with `source <(strix completions zsh)` (or `bash`), or `strix completions fish | source`. - The REST API works directly too: https://docs.app.strix.ai (OpenAPI: https://docs.app.strix.ai/openapi.json). diff --git a/README.md b/README.md index 7a3500c9..169f0a14 100644 --- a/README.md +++ b/README.md @@ -339,7 +339,8 @@ Every operation of the [REST API](https://docs.app.strix.ai) has a matching comm ```bash strix cloud # list all resources -strix cloud scans # list the verbs of a resource +strix cloud scans # run the safe default (`scans list`) +strix cloud scans help # list the verbs of a resource strix cloud domains add --domain example.com --asset-type web_app strix cloud scans start --engagement-type live_test --domain-ids --wait strix cloud scans start --source . --dry-run --show-files --json # review + capture source.archive_sha256 @@ -354,7 +355,7 @@ Workspaces and account setup also work from the terminal: ```bash strix cloud workspaces list # numbered list; `workspace` is also accepted -strix cloud workspaces create --name "My Team" +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 billing subscribe --plan strix_cloud # opens the hosted checkout page strix cloud billing portal # opens the billing portal @@ -364,7 +365,7 @@ 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 asset, knowledge, test-user, and token lists retain the record IDs needed by follow-up commands but omit internal organization/user IDs; 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 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. Write commands take request fields as flags, and every write command also accepts one JSON object with `--data`: @@ -389,8 +390,8 @@ the service independently validates the archive. Source alone infers a code revi domain infers a live test. You can always pass `--engagement-type` explicitly. Strix removes the temporary local archive after every invocation. It deletes a staged remote -upload after a definitive scan rejection. If a network error, `5xx` response, or interruption -makes the launch outcome ambiguous, it retains the upload and reports its `upload_id` with +upload after a definitive scan rejection. If a network error, `5xx` response, malformed +success response, or interruption makes the launch outcome ambiguous, it retains the upload and reports its `upload_id` with `launch_outcome_unknown: true`; if automatic deletion cannot be confirmed, it reports the ID with `cleanup_unknown: true`. Check `strix cloud scans list` before retrying. If no scan is linked to the retained upload, delete it with `strix cloud uploads delete UPLOAD_ID`. diff --git a/skills/managed-pentesting-with-strix/SKILL.md b/skills/managed-pentesting-with-strix/SKILL.md index d5e54427..48800ae4 100644 --- a/skills/managed-pentesting-with-strix/SKILL.md +++ b/skills/managed-pentesting-with-strix/SKILL.md @@ -13,12 +13,12 @@ Use this when you want Strix's autonomous pentesting **without running Docker or There are two equivalent interfaces. Prefer the CLI: -- **`strix cloud` CLI** — every REST operation has a command in the form `strix cloud `. Install with `curl -sSL https://strix.ai/install | bash`. Run `strix cloud` to list all resources and `strix cloud ` to list its verbs. +- **`strix cloud` CLI** — every REST operation has a command in the form `strix cloud `. Install with `curl -sSL https://strix.ai/install | bash`. Run `strix cloud` to list all resources and `strix cloud help` (or `-h`) to list a resource's verbs; a bare resource with a safe read operation runs its documented default. - **REST API** — base URL `https://app.strix.ai/api/v1`, `Authorization: Bearer ` on every request. Full reference: **[docs.app.strix.ai](https://docs.app.strix.ai)** · agent index: `https://docs.app.strix.ai/llms.txt` · OpenAPI: `https://docs.app.strix.ai/openapi.json`. -The CLI is equally usable by agents and people. Output is complete JSON when stdout is not a terminal, or when you pass `--json`; terminal tables favor names, branches, lifecycle states, and numbered selectors. Human asset, knowledge, test-user, and token lists retain the record IDs needed by follow-up commands but omit internal organization/user IDs; token lists label credentials as active, expired, or revoked. Binary downloads are the exception: redirect raw bytes intentionally, or use `--output FILE --json` to write the file and receive structured metadata. There are no interactive prompts when stdin is not a terminal. Exit codes: `0` success, `1` request/runtime error, `2` invalid usage, `4` authentication or plan limit, `5` payment required. +The CLI is equally usable by agents and people. Output is complete JSON when stdout is not a terminal, or when you pass `--json`; terminal tables favor names, branches, lifecycle states, and numbered selectors. 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: redirect raw bytes intentionally, or use `--output FILE --json` to write the file and receive structured metadata. There are no interactive prompts when stdin is not a terminal. Exit codes: `0` success, `1` request/runtime error, `2` invalid usage, `4` authentication or plan limit, `5` payment required. -Every resource group has a useful default list/read action, and `-h` or `help` always shows its verbs. Native tab completion includes resources, verbs, flags, workspace commands, and local paths: +Every resource group with a safe read operation has a useful default action, and `-h` or `help` always shows its verbs. Native tab completion includes resources, verbs, flags, workspace commands, and local paths: ```bash source <(strix completions zsh) # current zsh session @@ -62,8 +62,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 and switch workspaces | - | `tokens:write` | create/revoke API tokens | + | `organizations:read` | list workspaces | + | `organizations:write` | create/update workspaces (admin) | + | `tokens:write` | switch workspaces · create/revoke API tokens | | `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) | @@ -121,13 +122,13 @@ Manage workspaces with a personal token from `strix cloud login`: ```bash strix cloud workspaces list # numbered name/role/current list -strix cloud workspaces create --name "My Team" +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 org members invite --email dev@example.com --role analyst ``` -`workspaces use` rotates the current personal token to a workspace the user already belongs to and stores the updated workspace metadata; the bearer secret stays 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. The role in the target workspace limits the scopes. Add `--scopes` to request a smaller set. ### Handoffs a person must finish @@ -199,7 +200,10 @@ Useful flags (each maps to a `CreateScanRequest` field): | `--upload-ids` | attach uploaded source/docs archives for white-box context | | `--notify-on-completion` / `--notification-emails` | email when done | -The response is `{ scan_id, title, status }` with `status` = `pending`. +Without `--source`, the response is `{ scan_id, title, status }` with `status` = `pending`. +Local-source success wraps that platform response as +`{ source, upload_id, scan: { scan_id, title, status } }`, so automation can retain the exact +approved manifest and staged-upload identifier alongside the created scan. ### Scan a local workspace in the cloud diff --git a/strix/interface/cloud/http.py b/strix/interface/cloud/http.py index 3f9e5c68..36858c70 100644 --- a/strix/interface/cloud/http.py +++ b/strix/interface/cloud/http.py @@ -300,7 +300,13 @@ def check(response: requests.Response) -> Any: "the server returned a non-JSON response. Check STRIX_APP_URL and preview " "access, then retry." ) - return data + try: + return response.json() + except ValueError as exc: + raise CloudError( + "the server returned malformed JSON. Check STRIX_APP_URL and preview " + "access, then retry." + ) from exc detail = "" error_code = "" if isinstance(data, dict): diff --git a/strix/interface/cloud/render.py b/strix/interface/cloud/render.py index 36fa0887..69a497f3 100644 --- a/strix/interface/cloud/render.py +++ b/strix/interface/cloud/render.py @@ -22,10 +22,27 @@ if TYPE_CHECKING: _MAX_TABLE_COLUMNS = 8 _MAX_CELL_LENGTH = 60 +_MAX_DETAIL_CELL_LENGTH = 2000 _MAX_DETAIL_FIELDS = 36 _MAX_NESTED_PREVIEW = 5 _NARROW_TABLE_WIDTH = 120 _CAMEL_BOUNDARY = re.compile(r"(?<=[a-z0-9])(?=[A-Z])") +_COPYABLE_SELECTOR_COLUMNS = frozenset( + { + "event_id", + "id", + "installation_id", + "parent_id", + "path", + "policy_key", + "repo_key", + "retest_scan_id", + "scan_id", + "test_user_id", + "vulnerability_id", + } +) +_SELECTOR_NO_WRAP_MAX = 40 _INTERNAL_COLUMNS = frozenset( { "organization_id", @@ -69,6 +86,7 @@ _PREFERRED_KEYS = ( "display_number", "status", "state", + "workspace_state", "severity", "cve", "cvss", @@ -101,13 +119,80 @@ _PREFERRED_KEYS = ( "id", ) +_DETAIL_PRIORITY_KEYS = ( + "id", + "display_number", + "title", + "name", + "status", + "state", + "severity", + "finding_type", + "cve", + "cwe", + "cvss", + "filed_at", + "target", + "location_meta", + "urls", + "repositories", + "internal_targets", + "endpoint", + "method", + "url", + "events", + "business_unit", + "is_active", + "secret_prefix", + "last_success_at", + "last_failure_at", + "run_id", + "sandbox_attached", + "description", + "impact", + "technical_analysis", + "evidence", + "assumptions", + "remediation_steps", + "fix_pr_eligible", + "fix_pr_reason", + "fix_pr_url", + "poc_description", + "poc_script_code", + "code_file", + "code_locations", + "code_diff", + "code_before", + "code_after", + "dependency_metadata", + "fix_effort", + "executive_summary", + "methodology", + "recommendations", + "auth_status", + "auth_failure_code", + "auth_detail", + "scan_scope", + "findings", + "duration", + "created_at", + "updated_at", +) + _LIST_ENVELOPE_KEYS = frozenset( { "items", "data", "scans", + "agents", + "chats", "vulnerabilities", "findings", + "files", + "messages", + "runs", + "steps", + "components", "domains", "repositories", "repos", @@ -171,7 +256,7 @@ _VIEW_COLUMNS: dict[str, tuple[str, ...]] = { "title", "severity", "status", - "target", + "location", "cvss", "finding_type", "id", @@ -209,11 +294,10 @@ _VIEW_COLUMNS: dict[str, tuple[str, ...]] = { "GET /repositories": ( "full_name", "provider", - "default_branch", "pr_review_enabled", "last_scan_at", - "tags", "business_unit", + "tags", "id", ), "GET /knowledge": ( @@ -236,6 +320,7 @@ _VIEW_COLUMNS: dict[str, tuple[str, ...]] = { "updated_at", "id", ), + "GET /knowledge/repos": ("repo_key", "docs_count", "last_updated_at"), "GET /knowledge/policies": ( "policy_key", "policy_type", @@ -248,9 +333,9 @@ _VIEW_COLUMNS: dict[str, tuple[str, ...]] = { "GET /domains/{domainId}/test-users": ( "label", "username", - "mfa_method", - "mfa_email", - "has_password", + "password", + "mfa", + "verification", "login_url", "updated_at", "id", @@ -260,21 +345,225 @@ _VIEW_COLUMNS: dict[str, tuple[str, ...]] = { "type", "status", "scopes", - "secret_prefix", + "access", "expires_at", "last_used_at", "id", ), - "GET /webhooks": ("url", "events", "is_active", "last_delivery_at", "created_at", "id"), - "GET /webhooks/{webhookId}/deliveries": ( - "event", + "GET /chat": ("title", "status", "last_message_at", "created_at", "id"), + "GET /chat/{chatId}/files": ("path", "size"), + "GET /chat/{chatId}/findings": ( + "title", + "severity", "status", - "response_status", - "attempts", + "location", + "cvss", + "filed_at", "created_at", - "delivered_at", "id", ), + "GET /domains/{domainId}/test-users/{userId}/inbox": ( + "from", + "subject", + "detected_code", + "timestamp", + "preview", + "id", + ), + "GET /scans/{scanId}/agents": ( + "name", + "status", + "task", + "finding_count", + "parent_id", + "created_at", + "id", + ), + "GET /scans/{scanId}/trace": ( + "timestamp", + "kind", + "tool_name", + "status", + "summary", + "event_id", + ), + "GET /scans/{scanId}/retests": ( + "title", + "severity", + "issue_status", + "retest_status", + "created_at", + "vulnerability_id", + "retest_scan_id", + ), + "GET /pr-reviews/findings": ( + "repository", + "pull_request", + "pr_state", + "title", + "severity", + "status", + "created_at", + "id", + ), + "GET /vulnerabilities/{vulnerabilityId}/history": ( + "created_at", + "previous_status", + "new_status", + "snooze", + "previous_severity", + "new_severity", + "note", + "reason", + ), + "GET /repositories/{repositoryId}/supply-chain/findings": ( + "title", + "package", + "severity", + "status", + "fixed_version", + "manifest_path", + "direct", + "id", + ), + "GET /repositories/{repositoryId}/supply-chain/components": ( + "name", + "version", + "ecosystem", + "relationship", + "status", + "highest_open_severity", + "manifest_path", + "id", + ), + "GET /schedules": ( + "name", + "target", + "cron_expression", + "timezone", + "state", + "last_run_status", + "next_run_at", + "id", + ), + "GET /connectors": ("name", "last_status", "last_status_checked_at", "created_at", "id"), + "GET /organization/members": ( + "email", + "firstName", + "lastName", + "role", + "access", + "status", + "joinedAt", + "id", + ), + "GET /organization/invitations": ( + "email", + "role", + "access", + "state", + "expiresAt", + "createdAt", + "id", + ), + "GET /webhooks": ( + "url", + "events", + "is_active", + "business_unit", + "last_success_at", + "last_failure_at", + "created_at", + "id", + ), + "GET /webhooks/{webhookId}/deliveries": ( + "event_type", + "status", + "response_status", + "last_error", + "attempts", + "sent_at", + "next_attempt_at", + "id", + ), + "GET /audit": ( + "action", + "resource_type", + "resource_id", + "actor_email", + "ip_address", + "created_at", + ), + "supply_chain_totals": ( + "repositories", + "components", + "findings", + "open_issues", + "malicious", + "suspicious", + "vulnerable", + ), + "supply_chain_repositories": ( + "repository", + "components", + "findings", + "severity", + "risks", + "latest_scan", + "policy", + "id", + ), + "chat_credentials_attached": ( + "label", + "username", + "login_url", + "mfa_method", + "password", + "totp", + "test_user_id", + ), + "chat_credentials_test_users": ( + "label", + "username", + "domain", + "login_url", + "mfa_method", + "password", + "totp", + "id", + ), + "chat_credentials_scans": ( + "scan_title", + "username", + "login_url", + "mfa_method", + "password", + "totp", + "scan_id", + ), +} + +_VIEW_LIST_KEYS: dict[str, str] = { + "GET /chat": "chats", + "GET /chat/{chatId}/files": "files", + "GET /chat/{chatId}/findings": "findings", + "GET /domains/{domainId}/test-users/{userId}/inbox": "messages", + "GET /scans/{scanId}/agents": "agents", + "GET /scans/{scanId}/trace": "steps", + "GET /scans/{scanId}/retests": "runs", + "GET /pr-reviews/findings": "items", + "GET /vulnerabilities/{vulnerabilityId}/history": "items", + "GET /knowledge/repos/{repo}/entries": "docs", + "GET /repositories/{repositoryId}/supply-chain/findings": "findings", + "GET /repositories/{repositoryId}/supply-chain/components": "components", + "GET /schedules": "schedules", + "GET /organization/members": "members", + "GET /organization/invitations": "invitations", +} + +_DETAIL_ENVELOPE_KEYS: dict[str, str] = { + "GET /chat/{chatId}": "chat", + "GET /webhooks/{webhookId}": "webhook", } @@ -291,7 +580,7 @@ def json_mode(*, flag: bool) -> bool: return flag or not sys.stdout.isatty() -def emit( # noqa: PLR0911, PLR0912 +def emit( # noqa: PLR0911, PLR0912, PLR0915 console: Console, data: Any, *, @@ -307,6 +596,7 @@ def emit( # noqa: PLR0911, PLR0912 return if warning: console.print(f"[bold yellow]Save this now:[/] {escape(sanitize_terminal_text(warning))}") + hint = _combine_hints(hint, _pagination_hint(data)) if view == "source_manifest" and _is_record(data): _print_source_manifest(console, data) return @@ -316,6 +606,18 @@ def emit( # noqa: PLR0911, PLR0912 if view in {"GET /analytics/overview", "GET /analytics/stats"} and _is_record(data): _print_analytics(console, data) return + if view == "GET /supply-chain/summary" and _is_record(data): + _print_supply_chain_summary(console, data) + return + if view == "GET /chat/{chatId}/credentials" and _is_record(data): + _print_chat_credentials(console, data) + return + detail_key = _DETAIL_ENVELOPE_KEYS.get(view or "") + if detail_key and _is_record(data): + detail = data.get(detail_key) + if _is_record(detail): + _print_detail(console, _detail_envelope_record(detail, view)) + return if view == "GET /integrations": integration_rows = _integration_rows(data) if integration_rows is not None: @@ -348,19 +650,19 @@ def emit( # noqa: PLR0911, PLR0912 scan_rows, row_numbers=False, omit_columns=omit_columns, - hint="Inspect one scan with `strix cloud scans get ID`.", + hint=_combine_hints("Inspect one scan with `strix cloud scans get ID`.", hint), view=view, ) return if view == "GET /vulnerabilities": - vulnerability_rows = _list_of_dicts(data) + vulnerability_rows = _finding_location_rows(data) if vulnerability_rows is not None: _print_table( console, vulnerability_rows, row_numbers=False, omit_columns=omit_columns | frozenset({"scan_id"}), - hint="Inspect one finding with `strix cloud vulns get ID`.", + hint=_combine_hints("Inspect one finding with `strix cloud vulns get ID`.", hint), view=view, ) return @@ -372,10 +674,25 @@ def emit( # noqa: PLR0911, PLR0912 review_rows, row_numbers=False, omit_columns=omit_columns, - hint="Use `strix cloud pr-reviews get ID` for one review.", + hint=_combine_hints( + "Use `strix cloud pr-reviews get ID` for one review.", + _view_hint(data, view), + hint, + ), view=view, ) return + view_rows = _rows_for_view(data, view) + if view_rows is not None: + _print_table( + console, + view_rows, + row_numbers=row_numbers, + omit_columns=omit_columns, + hint=_combine_hints(_view_hint(data, view), hint), + view=view, + ) + return rows = _list_of_dicts(data) if rows is not None: _print_table( @@ -430,6 +747,404 @@ def _list_of_dicts(data: Any) -> list[dict[str, Any]] | None: return records +def _records_at_key(data: Any, key: str) -> list[dict[str, Any]] | None: + """Extract one deliberate collection even when an envelope has other lists.""" + if _is_list(data): + return _list_of_dicts(data) + if not _is_record(data): + return None + value = data.get(key) + if not _is_list(value) or not all(_is_record(item) for item in value): + return None + return list(value) + + +def _detail_envelope_record(data: dict[str, Any], view: str | None) -> dict[str, Any]: + record = dict(data) + if view == "GET /webhooks/{webhookId}" and not record.get("business_unit"): + record["business_unit"] = "all organization" + if view == "GET /chat/{chatId}": + record["sandbox_attached"] = bool(record.pop("sandbox_api_url", False)) + return record + + +def _rows_for_view(data: Any, view: str | None) -> list[dict[str, Any]] | None: + """Shape non-standard list envelopes into compact, actionable rows.""" + if view == "GET /domains/{domainId}/test-users": + return _test_user_rows(data) + key = _VIEW_LIST_KEYS.get(view or "") + if key is None: + return None + records = _records_at_key(data, key) + if records is None: + return None + transforms = { + "GET /scans/{scanId}/trace": _trace_rows, + "GET /chat/{chatId}/findings": _finding_location_rows, + "GET /pr-reviews/findings": _pr_finding_rows, + "GET /vulnerabilities/{vulnerabilityId}/history": _vulnerability_history_rows, + "GET /repositories/{repositoryId}/supply-chain/findings": (_supply_chain_finding_rows), + "GET /schedules": _schedule_rows, + "GET /organization/members": _access_rows, + "GET /organization/invitations": _access_rows, + } + transform = transforms.get(view or "") + return transform(records) if transform else records + + +def _view_hint(data: Any, view: str | None) -> str | None: + hint: str | None = None + if view == "GET /scans/{scanId}/trace" and _is_record(data): + hint = _trace_view_hint(data) + elif view == "GET /scans/{scanId}/retests" and _is_record(data): + total = data.get("total") + completed = data.get("completed") + running = data.get("running") + if all(isinstance(value, int) for value in (total, completed, running)): + hint = f"{completed}/{total} retest(s) complete; {running} running." + elif view == "GET /knowledge/repos/{repo}/entries" and _is_record(data): + hint = _knowledge_repo_hint(data) + elif view == "GET /pr-reviews" and _is_record(data): + hint = _metric_hint( + "Review counts", + data.get("counts"), + ("all", "open", "attention", "merged_open", "passed", "running"), + ) + elif view == "GET /pr-reviews/findings" and _is_record(data): + hint = _metric_hint( + "Impact", + data.get("stats"), + ("prs_reviewed", "issues_found", "critical_high_found", "merges_blocked"), + ) + elif view == "GET /domains/{domainId}/test-users/{userId}/inbox" and _is_record(data): + address = data.get("address") + if isinstance(address, str) and address.strip(): + hint = f"Inbox: {sanitize_terminal_text(address.strip())}." + elif view == "GET /schedules": + hint = "Inspect one schedule with `strix cloud schedules get ID`." + return hint + + +def _trace_view_hint(data: dict[str, Any]) -> str | None: + scan_id = data.get("scan_id") + suffix = f" {scan_id}" if scan_id else " SCAN_ID" + parts = [ + f"Inspect a complete event with `strix cloud scans trace-event{suffix} EVENT_ID`; " + "use --json for full tool arguments and results." + ] + cursor = data.get("cursor") + if data.get("has_more") and isinstance(cursor, str) and cursor: + parts.append( + f"Continue the same trace command with `--cursor {cursor}`; keep its " + "--agent-id, --tool-name, and --limit options." + ) + note = data.get("note") + if isinstance(note, str) and note.strip(): + parts.append(note.strip()) + return _combine_hints(*parts) + + +def _knowledge_repo_hint(data: dict[str, Any]) -> str | None: + parts: list[str] = [] + profile = data.get("profile") + if _is_record(profile): + title = sanitize_terminal_text(str(profile.get("title") or "present")) + parts.append(f"Repository profile: {title}.") + policies = data.get("policies") + if _is_list(policies): + noun = "policy" if len(policies) == 1 else "policies" + parts.append(f"{len(policies)} {noun} apply.") + if parts: + parts.append("Use --json to view the profile and policy metadata.") + return _combine_hints(*parts) + + +def _metric_hint(label: str, value: Any, keys: tuple[str, ...]) -> str | None: + if not _is_record(value): + return None + metrics = [ + f"{_human_label(key)} {value[key]}" for key in keys if isinstance(value.get(key), int) + ] + return f"{label}: {', '.join(metrics)}." if metrics else None + + +def _combine_hints(*hints: str | None) -> str | None: + combined = " ".join(hint.strip() for hint in hints if hint and hint.strip()) + return combined or None + + +def _pagination_hint(data: Any) -> str | None: + """Explain how to continue a paginated human list without hiding API metadata.""" + if not _is_record(data): + return None + candidates = [data.get(key) for key in ("meta", "pagination")] + for pagination in candidates: + if not _is_record(pagination): + continue + message = _pagination_message(pagination) + if message: + return message + for wrapper in ("data", "result"): + nested = data.get(wrapper) + if _is_record(nested): + hint = _pagination_hint(nested) + if hint: + return hint + return None + + +def _pagination_message(pagination: dict[str, Any]) -> str | None: + page = pagination.get("page") + total_pages = pagination.get("total_pages") + total = pagination.get("total_items", pagination.get("total")) + has_next = pagination.get("has_next") + if isinstance(page, int) and isinstance(total_pages, int): + return _page_pagination_message(page, total_pages, total, has_next=has_next) + + offset = pagination.get("offset") + limit = pagination.get("limit") + if not isinstance(offset, int) or not isinstance(limit, int) or not isinstance(total, int): + return None + if total <= 0: + return "0 total." + if offset >= total: + last_offset = max(0, ((total - 1) // max(1, limit)) * max(1, limit)) + return f"No items at offset {offset}; {total} total. Retry with `--offset {last_offset}`." + shown_through = min(offset + limit, total) + message = f"Showing {offset + 1}-{shown_through} of {total}." + if offset + limit < total: + message += f" Continue with `--offset {offset + limit}`." + return message + + +def _page_pagination_message(page: int, total_pages: int, total: Any, *, has_next: Any) -> str: + if total == 0: + return "0 total." + last_page = max(1, total_pages) + if page > last_page: + total_note = f"; {total} total" if isinstance(total, int) else "" + return f"No items on page {page}{total_note}. Retry with `--page {last_page}`." + parts = [f"Page {page}/{last_page}"] + if isinstance(total, int): + parts.append(f"{total} total") + message = " · ".join(parts) + "." + if (has_next is True or page < total_pages) and page >= 0: + message += f" Continue with `--page {page + 1}`." + return message + + +def _test_user_rows(data: Any) -> list[dict[str, Any]] | None: + records = _records_at_key(data, "items") + if records is None: + return None + checks = data.get("auth_checks") if _is_record(data) else None + rows: list[dict[str, Any]] = [] + for record in records: + row = dict(record) + row["password"] = "set" if record.get("has_password") else "not set" + method = str(record.get("mfa_method") or "none") + if method == "totp": + row["mfa"] = "totp (secret set)" if record.get("has_totp_secret") else "totp (missing)" + elif method in {"email_otp", "magic_link"}: + address = str(record.get("mfa_email") or "address missing") + row["mfa"] = f"{method}: {address}" + elif record.get("has_totp_secret"): + row["mfa"] = "none (TOTP secret stored)" + else: + row["mfa"] = "none" + + check: Any = None + if _is_record(checks): + check = checks.get(str(record.get("id") or "")) + elif _is_list(checks): + check = next( + ( + candidate + for candidate in checks + if _is_record(candidate) and candidate.get("test_user_id") == record.get("id") + ), + None, + ) + if _is_record(check): + status = str(check.get("status") or "unknown") + failure = str(check.get("failure_code") or "").strip() + row["verification"] = f"{status}: {failure}" if failure else status + else: + row["verification"] = "not checked" + rows.append(row) + return rows + + +def _trace_rows(records: list[dict[str, Any]]) -> list[dict[str, Any]]: + rows: list[dict[str, Any]] = [] + for record in records: + row = dict(record) + finding = record.get("finding") + if _is_record(finding): + title = str(finding.get("title") or "finding") + severity = str(finding.get("severity") or "").strip() + row["summary"] = f"{severity}: {title}" if severity else title + elif record.get("result") is not None: + row["summary"] = _trace_payload_shape("result", record["result"]) + elif record.get("args") is not None: + row["summary"] = _trace_payload_shape("arguments", record["args"]) + elif record.get("content") is not None: + row["summary"] = _trace_payload_shape("message", record["content"]) + rows.append(row) + return rows + + +def _trace_payload_shape(label: str, value: Any) -> str: + """Describe trace payload structure without leaking credentials or response bodies.""" + if _is_record(value): + return f"{label}: {len(value)} field(s)" + if _is_list(value): + return f"{label}: {len(value)} item(s)" + if value is None: + return f"{label}: empty" + text = str(value) + kind = "text" if isinstance(value, str) else type(value).__name__ + return f"{label}: {kind} ({len(text)} character(s))" + + +def _pr_finding_rows(records: list[dict[str, Any]]) -> list[dict[str, Any]]: + rows: list[dict[str, Any]] = [] + for record in records: + row = dict(record) + row["repository"] = record.get("repository_full_name") + number = record.get("pr_number") + title = str(record.get("pr_title") or "").strip() + row["pull_request"] = " ".join( + part for part in (f"#{number}" if number is not None else "", title) if part + ) + rows.append(row) + return rows + + +def _finding_location_rows(data: Any) -> list[dict[str, Any]] | None: + records = _list_of_dicts(data) + if records is None: + return None + rows: list[dict[str, Any]] = [] + for record in records: + row = dict(record) + endpoint = str(record.get("endpoint") or "").strip() + method = str(record.get("method") or "").strip().upper() + target = str(record.get("target") or "").strip() + row["location"] = f"{method} {endpoint}".strip() if endpoint else target + rows.append(row) + return rows + + +def _supply_chain_finding_rows(records: list[dict[str, Any]]) -> list[dict[str, Any]]: + rows: list[dict[str, Any]] = [] + for record in records: + row = dict(record) + name = str(record.get("package_name") or "").strip() + version = str(record.get("package_version") or "").strip() + row["package"] = f"{name}@{version}" if name and version else name or version + rows.append(row) + return rows + + +def _schedule_rows(records: list[dict[str, Any]]) -> list[dict[str, Any]]: + rows: list[dict[str, Any]] = [] + for record in records: + row = dict(record) + row["cron_expression"] = record.get("cron") or record.get("cron_expression") + row["state"] = "paused" if record.get("isPaused") else "active" + targets: list[str] = [] + if record.get("supply_chain") is True: + targets.append("supply chain") + for key, singular in (("domain_ids", "domain"), ("repository_ids", "repo")): + values = record.get(key) + if _is_list(values) and values: + noun = singular if len(values) == 1 else f"{singular}s" + targets.append(f"{len(values)} {noun}") + internal_targets = record.get("internal_targets") + if _is_list(internal_targets) and internal_targets: + first = sanitize_terminal_text(str(internal_targets[0])) + suffix = f" (+{len(internal_targets) - 1} more)" if len(internal_targets) > 1 else "" + targets.append(f"{first}{suffix}") + if record.get("connector_id"): + targets.append("network connector") + row["target"] = " · ".join(targets) if targets else "no targets" + rows.append(row) + return rows + + +def _access_rows(records: list[dict[str, Any]]) -> list[dict[str, Any]]: + rows: list[dict[str, Any]] = [] + for record in records: + row = dict(record) + row["access"] = _scope_summary(record.get("scopes")) + rows.append(row) + return rows + + +def _scope_summary(scopes: Any) -> str: + if not _is_list(scopes) or not scopes: + return "all assets" + labels: list[str] = [] + for scope in scopes[:2]: + if _is_record(scope): + scope_type = str(scope.get("type") or "scope").strip() + value = str(scope.get("value") or "").strip() + labels.append(f"{scope_type}:{value}" if value else scope_type) + else: + labels.append(str(scope)) + suffix = f" (+{len(scopes) - len(labels)} more)" if len(scopes) > len(labels) else "" + return ", ".join(labels) + suffix + + +def _credential_summary_rows(data: dict[str, Any], key: str) -> list[dict[str, Any]]: + records = _records_at_key(data, key) or [] + rows: list[dict[str, Any]] = [] + for record in records: + row = dict(record) + row["password"] = "set" if record.get("has_password") else "not set" + row["totp"] = "set" if record.get("has_totp_secret") else "not set" + rows.append(row) + return rows + + +def _print_chat_credentials(console: Console, data: dict[str, Any]) -> None: + """Render attached and attachable credential metadata without exposing secrets.""" + sections = ( + ( + "Attached credentials", + _credential_summary_rows(data, "credentials"), + "chat_credentials_attached", + None, + ), + ( + "Available saved test users", + _credential_summary_rows(data, "available_test_users"), + "chat_credentials_test_users", + "Attach one with `strix cloud chat credentials set CHAT_ID --test-user-ids ID`.", + ), + ( + "Credentials from requested scans", + _credential_summary_rows(data, "available_scan_credentials"), + "chat_credentials_scans", + ( + "Discover these with `strix cloud chat credentials CHAT_ID --scan-ids SCAN_ID`; " + "attach them with `strix cloud chat credentials set CHAT_ID --scan-ids SCAN_ID`." + ), + ), + ) + for title, rows, section_view, section_hint in sections: + console.print(f"[bold]{title}[/]") + _print_table( + console, + rows, + view=section_view, + hint=section_hint, + show_json_hint=False, + ) + console.print("[dim]Use --json for the complete credential metadata.[/]") + + def _integration_rows(data: Any) -> list[dict[str, Any]] | None: """Flatten the two integration collections into one compact human view.""" if not _is_record(data): @@ -459,6 +1174,7 @@ def _token_rows(data: Any) -> list[dict[str, Any]] | None: rows: list[dict[str, Any]] = [] for record in records: row = dict(record) + row["access"] = _scope_summary(record.get("rbac_scopes")) if record.get("revoked_at"): row["status"] = "revoked" elif _timestamp_has_passed(record.get("expires_at")): @@ -494,24 +1210,7 @@ def _scan_rows(data: Any) -> list[dict[str, Any]] | None: if not row.get("id") and isinstance(row.get("scan_id"), str): row["id"] = row["scan_id"] - targets: list[str] = [] - urls = record.get("urls") - if _is_list(urls): - targets.extend(url.strip() for url in urls if isinstance(url, str) and url.strip()) - repositories = record.get("repositories") - if _is_list(repositories): - for repository in repositories: - if not _is_record(repository): - continue - identifier = str( - repository.get("full_name") - or repository.get("name") - or repository.get("url") - or "" - ).strip() - branch = str(repository.get("branch") or "").strip() - if identifier: - targets.append(f"{identifier} @ {branch}" if branch else identifier) + targets = _scan_targets(record) if targets: visible_targets = targets[:2] summary = " | ".join(visible_targets) @@ -526,6 +1225,34 @@ def _scan_rows(data: Any) -> list[dict[str, Any]] | None: return rows +def _scan_targets(record: dict[str, Any]) -> list[str]: + targets: list[str] = [] + urls = record.get("urls") + if _is_list(urls): + targets.extend(url.strip() for url in urls if isinstance(url, str) and url.strip()) + repositories = record.get("repositories") + if _is_list(repositories): + for repository in repositories: + if not _is_record(repository): + continue + identifier = str( + repository.get("full_name") or repository.get("name") or repository.get("url") or "" + ).strip() + branch = str(repository.get("branch") or "").strip() + if identifier: + targets.append(f"{identifier} @ {branch}" if branch else identifier) + internal_targets = record.get("internal_targets") + if _is_list(internal_targets): + targets.extend( + target.strip() + for target in internal_targets + if isinstance(target, str) and target.strip() + ) + if record.get("has_code_upload") is True: + targets.append("uploaded source") + return targets + + def _pr_review_rows(data: Any) -> list[dict[str, Any]] | None: """Collapse related PR fields into an eight-column, action-oriented human view.""" records = _list_of_dicts(data) @@ -537,14 +1264,22 @@ def _pr_review_rows(data: Any) -> list[dict[str, Any]] | None: number = record.get("pr_number") title = str(record.get("pr_title") or "").strip() row["repository"] = record.get("repository_full_name") or record.get("repository") - row["pull_request"] = " ".join( + pull_request = " ".join( part for part in (f"#{number}" if number is not None else "", title) if part ) + pr_state = str(record.get("pr_state") or "").strip() + row["pull_request"] = f"{pull_request} [{pr_state}]" if pr_state else pull_request head = str(record.get("head_branch") or "").strip() base = str(record.get("base_branch") or "").strip() row["branches"] = f"{head} → {base}" if head and base else head or base - total = record.get("findings_count") - opened = record.get("open_findings_count") + findings = record.get("findings") + total = findings.get("total") if _is_record(findings) else None + unresolved = findings.get("unresolved") if _is_record(findings) else None + opened = unresolved.get("total") if _is_record(unresolved) else None + if not isinstance(total, int): + total = record.get("findings_count") + if not isinstance(opened, int): + opened = record.get("open_findings_count") if isinstance(total, int) and isinstance(opened, int): row["findings"] = f"{opened} open / {total} total" elif isinstance(total, int): @@ -553,6 +1288,23 @@ def _pr_review_rows(data: Any) -> list[dict[str, Any]] | None: return rows +def _vulnerability_history_rows(records: list[dict[str, Any]]) -> list[dict[str, Any]]: + rows: list[dict[str, Any]] = [] + for record in records: + row = dict(record) + previous = record.get("previous_snoozed_until") + current = record.get("new_snoozed_until") + if previous != current: + if previous and current: + row["snooze"] = f"{previous} → {current}" + elif current: + row["snooze"] = f"set until {current}" + else: + row["snooze"] = f"cleared (was {previous})" + rows.append(row) + return rows + + def _print_table( console: Console, rows: list[dict[str, Any]], @@ -561,6 +1313,7 @@ def _print_table( omit_columns: frozenset[str] = frozenset(), hint: str | None = None, view: str | None = None, + show_json_hint: bool = True, ) -> None: if not rows: console.print("[dim]No items.[/]") @@ -573,22 +1326,20 @@ def _print_table( omit_columns = omit_columns | (_INTERNAL_COLUMNS - visible_internal) | view_omissions preferred = _VIEW_COLUMNS.get(view or "", _PREFERRED_KEYS) columns: list[str] = [ - key for key in preferred if key not in omit_columns and any(key in row for row in rows) + key + for key in preferred + if key not in omit_columns and any(_meaningful(row.get(key)) for row in rows) ] - for row in rows: - for key in row: - if ( - key not in columns - and key not in omit_columns - and len(columns) < _MAX_TABLE_COLUMNS - and not isinstance(row[key], dict | list) - ): - columns.append(key) + if view not in _VIEW_COLUMNS: + _append_fallback_columns(rows, columns, omit_columns) columns = columns[:_MAX_TABLE_COLUMNS] if console.width < _NARROW_TABLE_WIDTH: _print_cards(console, rows, columns, row_numbers=row_numbers) - _print_long_identifiers(console, rows) - console.print(f"[dim]{len(rows)} item(s). Use --json for the full records.[/]") + _print_copyable_selectors(console, rows, columns) + footer = f"{len(rows)} item(s)." + if show_json_hint: + footer += " Use --json for the full records." + console.print(f"[dim]{footer}[/]") if hint: console.print(f"[dim]{escape(sanitize_terminal_text(hint))}[/]") return @@ -596,15 +1347,21 @@ def _print_table( if row_numbers: table.add_column("#", justify="right", style="cyan", no_wrap=True) for column in columns: - table.add_column(escape(_human_label(column))) + table.add_column( + escape(_human_label(column)), + no_wrap=_selector_can_no_wrap(column, rows), + ) for index, row in enumerate(rows, start=1): cells = [escape(_cell(row.get(column))) for column in columns] if row_numbers: cells.insert(0, str(index)) table.add_row(*cells) console.print(table) - _print_long_identifiers(console, rows) - console.print(f"[dim]{len(rows)} item(s). Use --json for the full records.[/]") + _print_copyable_selectors(console, rows, columns) + footer = f"{len(rows)} item(s)." + if show_json_hint: + footer += " Use --json for the full records." + console.print(f"[dim]{footer}[/]") if hint: console.print(f"[dim]{escape(sanitize_terminal_text(hint))}[/]") @@ -633,17 +1390,55 @@ def _print_cards( console.print(continuation + part, soft_wrap=True) -def _print_long_identifiers(console: Console, rows: list[dict[str, Any]]) -> None: - """Print opaque IDs losslessly when the compact cell view shortens them.""" - identifiers = [ - (index, row, str(row["id"])) +def _append_fallback_columns( + rows: list[dict[str, Any]], + columns: list[str], + omit_columns: frozenset[str], +) -> None: + for row in rows: + for key in row: + if ( + key not in columns + and key not in omit_columns + and len(columns) < _MAX_TABLE_COLUMNS + and not isinstance(row[key], dict | list) + ): + columns.append(key) + + +def _meaningful(value: Any) -> bool: + return value is not None and value not in ("", [], {}) + + +def _is_selector_column(column: str) -> bool: + return column in _COPYABLE_SELECTOR_COLUMNS + + +def _selector_can_no_wrap(column: str, rows: list[dict[str, Any]]) -> bool: + return _is_selector_column(column) and all( + len(str(row[column])) <= _SELECTOR_NO_WRAP_MAX + for row in rows + if row.get(column) is not None + ) + + +def _print_copyable_selectors( + console: Console, + rows: list[dict[str, Any]], + columns: list[str], +) -> None: + """Print command selectors losslessly when their compact cell is shortened.""" + selector_columns = [column for column in columns if _is_selector_column(column)] + selectors = [ + (index, row, column, str(row[column])) for index, row in enumerate(rows, start=1) - if row.get("id") is not None and len(str(row["id"])) > _MAX_CELL_LENGTH + for column in selector_columns + if row.get(column) is not None and len(str(row[column])) > _SELECTOR_NO_WRAP_MAX ] - if not identifiers: + if not selectors: return - console.print("[dim]Copyable IDs:[/]") - for index, row, identifier in identifiers: + console.print("[dim]Copyable selectors:[/]") + for index, row, column, value in selectors: label = next( ( str(row[key]) @@ -653,7 +1448,8 @@ def _print_long_identifiers(console: Console, rows: list[dict[str, Any]]) -> Non f"item {index}", ) console.print( - f" {index}. {sanitize_terminal_text(label)}: {sanitize_terminal_text(identifier)}", + f" {index}. {sanitize_terminal_text(label)} ({_human_label(column)}): " + f"{sanitize_terminal_text(value)}", markup=False, soft_wrap=True, ) @@ -661,7 +1457,12 @@ def _print_long_identifiers(console: Console, rows: list[dict[str, Any]]) -> Non def _print_detail(console: Console, data: dict[str, Any]) -> None: """Render one API record as a readable field/value view.""" - keys = [key for key in _PREFERRED_KEYS if key in data and key not in _INTERNAL_COLUMNS] + keys = [key for key in _DETAIL_PRIORITY_KEYS if key in data and key not in _INTERNAL_COLUMNS] + keys.extend( + key + for key in _PREFERRED_KEYS + if key in data and key not in keys and key not in _INTERNAL_COLUMNS + ) keys.extend(key for key in data if key not in keys and key not in _INTERNAL_COLUMNS) table = Table(show_header=False, show_edge=False, box=None, padding=(0, 2)) table.add_column("field", style="bold cyan", no_wrap=True) @@ -674,7 +1475,9 @@ def _print_detail(console: Console, data: dict[str, Any]) -> None: if _is_lossless_detail(key, value): lossless_fields.append((key, value)) continue - rendered = _nested_summary(value) if _is_record(value) or _is_list(value) else _cell(value) + rendered = ( + _nested_summary(value) if _is_record(value) or _is_list(value) else _detail_cell(value) + ) table.add_row(escape(_human_label(key)), escape(rendered)) if table.row_count: console.print(table) @@ -751,6 +1554,91 @@ def _print_source_manifest(console: Console, data: dict[str, Any]) -> None: console.print(f" {escape(sanitize_terminal_text(path))}", soft_wrap=True) +def _print_supply_chain_summary(console: Console, data: dict[str, Any]) -> None: + """Render organization totals and one actionable row per repository.""" + totals = data.get("totals") + console.print("[bold]Supply-chain totals[/]") + _print_table( + console, + [dict(totals)] if _is_record(totals) else [], + view="supply_chain_totals", + show_json_hint=False, + ) + + console.print("[bold]Repositories[/]") + _print_table( + console, + _supply_chain_repository_rows(data), + view="supply_chain_repositories", + hint="Inspect one repository with `strix cloud repos supply-chain summary ID`.", + show_json_hint=False, + ) + console.print("[dim]Use --json for complete totals and repository records.[/]") + + +def _supply_chain_repository_rows(data: dict[str, Any]) -> list[dict[str, Any]]: + entries = data.get("repositories") + if not _is_list(entries): + return [] + rows: list[dict[str, Any]] = [] + for entry in entries: + if not _is_record(entry): + continue + repository = entry.get("repository") + summary = entry.get("summary") + if not _is_record(repository) or not _is_record(summary): + continue + row: dict[str, Any] = { + "repository": repository.get("full_name") or repository.get("name"), + "components": summary.get("component_count", 0), + "findings": summary.get("finding_count", 0), + "severity": _supply_chain_severity_summary(summary.get("severity_counts")), + "risks": _supply_chain_risk_summary(summary), + "latest_scan": _supply_chain_scan_summary(entry.get("latest_supply_chain_scan")), + "policy": _supply_chain_policy_summary(summary.get("policy")), + "id": repository.get("id"), + } + rows.append(row) + return rows + + +def _supply_chain_risk_summary(summary: dict[str, Any]) -> str: + return " · ".join( + ( + f"{summary.get('malicious_count', 0)} malicious", + f"{summary.get('suspicious_count', 0)} suspicious", + f"{summary.get('vulnerable_count', 0)} vulnerable", + ) + ) + + +def _supply_chain_severity_summary(value: Any) -> str: + if not _is_record(value): + return "none" + ordered = ("critical", "high", "medium", "low", "info", "unknown") + counts = [f"{key} {value[key]}" for key in ordered if isinstance(value.get(key), int)] + return " · ".join(counts) if counts else "none" + + +def _supply_chain_scan_summary(value: Any) -> str: + if not _is_record(value): + return "not run" + status = str(value.get("status") or "unknown") + created_at = str(value.get("created_at") or "").strip() + return f"{status} · {created_at}" if created_at else status + + +def _supply_chain_policy_summary(value: Any) -> str: + if not _is_record(value): + return "unknown" + if value.get("enabled") is False: + return "disabled" + mode = str(value.get("mode") or "monitor") + if value.get("pr_checks_enabled") is False: + return f"{mode} · PR checks off" + return mode + + def _print_analytics(console: Console, data: dict[str, Any]) -> None: rows = list(_flatten_summary(data)) table = Table(show_header=False, show_edge=False, box=None, padding=(0, 2)) @@ -856,3 +1744,16 @@ def _cell(value: Any) -> str: if len(text) > _MAX_CELL_LENGTH: return text[: _MAX_CELL_LENGTH - 1] + "…" return text + + +def _detail_cell(value: Any) -> str: + """Keep prose useful in a detail view while bounding hostile responses.""" + if value is None: + return "" + if isinstance(value, bool): + return "yes" if value else "no" + text = "\n".join(sanitize_terminal_text(line) for line in str(value).split("\n")) + if len(text) > _MAX_DETAIL_CELL_LENGTH: + suffix = "… [truncated; use --json]" + return text[: _MAX_DETAIL_CELL_LENGTH - len(suffix)] + suffix + return text diff --git a/strix/interface/cloud/runner.py b/strix/interface/cloud/runner.py index 1d813f87..2c991f1d 100644 --- a/strix/interface/cloud/runner.py +++ b/strix/interface/cloud/runner.py @@ -377,7 +377,10 @@ def _execute( # noqa: PLR0912, PLR0915 return http.EXIT_OK source_workflow.mark_launch_started() - scan_request_started = cmd.idempotent + # Every wait-path mutation creates a scan, even when the endpoint has + # not yet adopted idempotency keys (for example vulnerability retests). + # Once sent, transport and malformed-success failures are ambiguous. + scan_request_started = cmd.idempotent or cmd.wait_path is not None response = _request_with_idempotency( cmd, path, @@ -414,7 +417,7 @@ def _execute( # noqa: PLR0912, PLR0915 json_metadata=binary_json_metadata, ) try: - result = http.check(response) + result = _validated_operation_result(http.check(response), cmd) except BaseException as exc: source_workflow.handle_response_failure( exc, @@ -1044,11 +1047,20 @@ def _created_id(created: Any) -> str | None: return None fields = cast("dict[str, Any]", created) for key, value in fields.items(): - if (key == "id" or key.endswith("_id")) and isinstance(value, str): + if (key == "id" or key.endswith("_id")) and isinstance(value, str) and value.strip(): return value return None +def _validated_operation_result(result: Any, cmd: Cmd) -> Any: + """Reject malformed success bodies for mutations that create a scan.""" + if cmd.wait_path and _created_id(result) is None: + raise http.CloudError( + "the platform returned a successful operation response without an operation ID." + ) + return result + + def _wait( console: Console, cmd: Cmd, @@ -1059,8 +1071,12 @@ def _wait( wait_timeout: float, ) -> Any: item_id = _created_id(created) - if not item_id or not cmd.wait_path: + if not cmd.wait_path: return created + if not item_id: + raise http.CloudError( + "cannot wait because the platform response did not include an operation ID." + ) path = cmd.wait_path.replace("{id}", str(item_id)) if not as_json: console.print( diff --git a/strix/interface/cloud/source_scan.py b/strix/interface/cloud/source_scan.py index a3698c96..7893bf5a 100644 --- a/strix/interface/cloud/source_scan.py +++ b/strix/interface/cloud/source_scan.py @@ -251,7 +251,7 @@ def _upload_scan_source(bundle: SourceBundle, *, token: str | None) -> str: raise error try: http.upload_file(cast("str", signed_url), cast("str", upload_token), bundle.archive_path) - http.check( + completed = http.check( http.request( "POST", "/uploads/complete", @@ -259,6 +259,7 @@ def _upload_scan_source(bundle: SourceBundle, *, token: str | None) -> str: body={"upload_id": upload_id}, ) ) + _validate_completed_upload(completed, expected_id=cast("str", upload_id)) except BaseException as error: try: _delete_upload(cast("str", upload_id), token=token) @@ -273,6 +274,12 @@ def _upload_scan_source(bundle: SourceBundle, *, token: str | None) -> str: return cast("str", upload_id) +def _validate_completed_upload(completed: Any, *, expected_id: str) -> None: + fields = cast("dict[str, Any]", completed) if isinstance(completed, dict) else {} + if fields.get("id") != expected_id: + raise http.CloudError("the platform returned an invalid source upload completion response.") + + def _delete_upload(upload_id: str, *, token: str | None) -> None: response = http.request("DELETE", f"/uploads/{quote(upload_id, safe='')}", token=token) if response.status_code == 404 or 200 <= response.status_code < 300: diff --git a/strix/interface/cloud/spec.py b/strix/interface/cloud/spec.py index 2e1aa16f..5b64a120 100644 --- a/strix/interface/cloud/spec.py +++ b/strix/interface/cloud/spec.py @@ -300,8 +300,7 @@ SPEC: dict[str, dict[str, Cmd]] = { P( "status", help=( - "New status: open, in_progress, snoozed, fixed, ignored, or " - "not_affected." + "New status: open, in_progress, snoozed, fixed, ignored, or not_affected." ), ), P("note", help="Note that explains the change."), @@ -1096,10 +1095,27 @@ SPEC: dict[str, dict[str, Cmd]] = { # Default verbs let a bare group name run its most common read command. DEFAULT_VERBS: dict[str, str] = { + "scans": "list", + "vulns": "list", + "domains": "list", + "repos": "list", "workspaces": "list", + "schedules": "list", + "pr-reviews": "list", + "billing": "credits", + "chat": "list", + "knowledge": "list", + "org": "get", + "integrations": "list", + "connectors": "list", + "webhooks": "list", + "analytics": "overview", "costs": "overview", "audit": "list", + "llm-settings": "get", + "settings": "notifications", "license": "show", + "tokens": "list", "supply-chain": "summary", } diff --git a/tests/test_cloud_cli.py b/tests/test_cloud_cli.py index 47f1a51e..e3aef155 100644 --- a/tests/test_cloud_cli.py +++ b/tests/test_cloud_cli.py @@ -13,6 +13,7 @@ from typing import Any import pytest import requests +from rich.console import Console from strix.interface import cloud, platform_cli from strix.interface.cloud import billing, http, payment_proxy, render, runner, workspaces @@ -81,8 +82,21 @@ def test_successful_html_response_is_reported_without_dumping_html( assert " None: - assert cloud.run_cloud(["scans"]) == 0 +def test_successful_malformed_json_response_is_rejected( + monkeypatch: pytest.MonkeyPatch, capsys: Any +) -> None: + response = FakeResponse(text="accepted") + response.headers = {"content-type": "application/json"} + monkeypatch.setattr(http, "request", lambda *_a, **_k: response) + + assert cloud.run_cloud(["workspaces", "list", "--json"]) == 1 + output = capsys.readouterr().out + assert "malformed JSON" in output + assert "accepted" not in output + + +def test_group_without_safe_read_default_lists_verbs() -> None: + assert cloud.run_cloud(["uploads"]) == 0 def test_resolve_prefers_two_word_verbs() -> None: @@ -102,6 +116,41 @@ def test_resolve_default_verb() -> None: assert remaining == [] +@pytest.mark.parametrize( + ("group", "verb"), + [ + ("scans", "list"), + ("vulns", "list"), + ("domains", "list"), + ("repos", "list"), + ("schedules", "list"), + ("pr-reviews", "list"), + ("billing", "credits"), + ("chat", "list"), + ("knowledge", "list"), + ("org", "get"), + ("integrations", "list"), + ("connectors", "list"), + ("webhooks", "list"), + ("analytics", "overview"), + ("audit", "list"), + ("costs", "overview"), + ("llm-settings", "get"), + ("settings", "notifications"), + ("license", "show"), + ("tokens", "list"), + ("supply-chain", "summary"), + ("workspaces", "list"), + ], +) +def test_read_groups_have_safe_defaults(group: str, verb: str) -> None: + resolved = runner.resolve(group, []) + assert resolved is not None + command, remaining = resolved + assert command is runner.SPEC[group][verb] + assert remaining == [] + + def test_dest_converts_camel_case() -> None: assert runner._dest("scanId") == "scan_id" assert runner._dest("chatId") == "chat_id" @@ -141,8 +190,10 @@ def test_placeholder_substitution_percent_encodes_path_segments( def test_query_and_body_collection(monkeypatch: pytest.MonkeyPatch) -> None: seen: dict[str, Any] = {} - def fake_request(_method: str, _path: str, **kwargs: Any) -> FakeResponse: + def fake_request(method: str, path: str, **kwargs: Any) -> FakeResponse: seen.update(query=kwargs.get("query"), body=kwargs.get("body")) + if method == "POST" and path == "/scans": + return FakeResponse(payload={"scan_id": "scan-1", "status": "pending"}) return FakeResponse(payload={"ok": True}) monkeypatch.setattr(http, "request", fake_request) @@ -172,7 +223,7 @@ def test_data_merges_extra_fields(monkeypatch: pytest.MonkeyPatch) -> None: def fake_request(_method: str, _path: str, **kwargs: Any) -> FakeResponse: seen["body"] = kwargs.get("body") - return FakeResponse(payload={"ok": True}) + return FakeResponse(payload={"scan_id": "scan-1", "status": "pending"}) monkeypatch.setattr(http, "request", fake_request) code = cloud.run_cloud( @@ -264,7 +315,7 @@ def test_data_reads_a_file(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> N def fake_request(_method: str, _path: str, **kwargs: Any) -> FakeResponse: seen["body"] = kwargs.get("body") - return FakeResponse(payload={"ok": True}) + return FakeResponse(payload={"scan_id": "scan-1", "status": "pending"}) monkeypatch.setattr(http, "request", fake_request) request_file = tmp_path / "request.json" @@ -278,7 +329,7 @@ def test_data_reads_stdin(monkeypatch: pytest.MonkeyPatch) -> None: def fake_request(_method: str, _path: str, **kwargs: Any) -> FakeResponse: seen["body"] = kwargs.get("body") - return FakeResponse(payload={"ok": True}) + return FakeResponse(payload={"scan_id": "scan-1", "status": "pending"}) monkeypatch.setattr(http, "request", fake_request) monkeypatch.setattr("sys.stdin", io.StringIO('{"context": "staging"}')) @@ -439,6 +490,42 @@ def test_ambiguous_scan_request_warns_before_retry( assert "scans list" in payload["error"] +@pytest.mark.parametrize("response_payload", [{}, "accepted"]) +def test_malformed_scan_success_is_reported_as_ambiguous( + response_payload: Any, monkeypatch: pytest.MonkeyPatch, capsys: Any +) -> None: + monkeypatch.setattr( + http, + "request", + lambda *_args, **_kwargs: FakeResponse(payload=response_payload), + ) + + assert cloud.run_cloud(["scans", "start", "--domain-ids", "d1", "--json"]) == 1 + payload = json.loads(capsys.readouterr().out) + assert payload["launch_outcome_unknown"] is True + assert payload["retry_safe"] is True + assert payload["idempotency_key"] + assert "scans list" in payload["error"] + + +@pytest.mark.parametrize( + "command", + [ + ["scans", "rerun", "scan-1", "--wait", "--json"], + ["vulns", "retest", "vuln-1", "--wait", "--json"], + ], +) +def test_waitable_scan_mutation_requires_an_operation_id( + command: list[str], monkeypatch: pytest.MonkeyPatch, capsys: Any +) -> None: + monkeypatch.setattr(http, "request", lambda *_a, **_k: FakeResponse(payload={})) + + assert cloud.run_cloud(command) == 1 + payload = json.loads(capsys.readouterr().out) + assert payload["launch_outcome_unknown"] is True + assert "successful operation response without an operation ID" in payload["error"] + + def test_insufficient_credits_exits_with_payment_code(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr( http, "request", lambda *_a, **_k: FakeResponse(status_code=402, payload={}) @@ -1351,6 +1438,8 @@ def test_created_id_reads_resource_id() -> None: assert runner._created_id({"scan_id": "abc", "status": "pending"}) == "abc" assert runner._created_id({"id": "xyz"}) == "xyz" assert runner._created_id({"status": "pending"}) is None + assert runner._created_id({"scan_id": ""}) is None + assert runner._created_id({"id": " "}) is None def test_billing_subscribe_prints_checkout_url( @@ -1893,10 +1982,81 @@ def test_pr_review_human_list_prioritizes_actionable_fields( "pr_title": "Improve cloud CLI", "head_branch": "feature", "base_branch": "main", - "verdict": "pass", + "pr_state": "merged", + "verdict": "request_changes", "status": "posted", - "findings_count": 0, - "open_findings_count": 0, + "findings_count": 99, + "open_findings_count": 88, + "findings": { + "total": 7, + "critical": 1, + "high": 2, + "medium": 3, + "low": 1, + "unresolved": {"total": 2}, + "snoozed": 1, + "fixed": 4, + }, + } + ], + "meta": {"total": 1}, + "counts": { + "all": 12, + "open": 3, + "attention": 2, + "merged_open": 1, + "passed": 6, + "running": 1, + }, + } + ), + ) + + assert cloud.run_cloud(["pr-reviews", "list", "--include-counts"]) == 0 + output = capsys.readouterr().out + for value in ( + "usestrix/strix", + "1177", + "Improve cloud CLI", + "merged", + "feature", + "main", + "posted", + "request_changes", + "2 open / 7 total", + "Review counts", + "attention 2", + "passed 6", + "review-id", + ): + assert value in output + for value in ("org-id", "user-id", "installation_id"): + assert value not in output + + +@pytest.mark.parametrize("pr_state", ("open", "merged", "closed")) +def test_pr_review_human_list_shows_pull_request_state( + monkeypatch: pytest.MonkeyPatch, + capsys: Any, + pr_state: str, +) -> None: + monkeypatch.setattr(render.sys.stdout, "isatty", lambda: True) + monkeypatch.setattr( + http, + "request", + lambda *_a, **_k: FakeResponse( + payload={ + "items": [ + { + "id": f"{pr_state}-review-id", + "repository_full_name": "usestrix/strix", + "pr_number": 1177, + "pr_title": "Renderer test", + "head_branch": "feature", + "base_branch": "main", + "pr_state": pr_state, + "status": "posted", + "findings": {"total": 0, "unresolved": {"total": 0}}, } ], "meta": {"total": 1}, @@ -1905,26 +2065,94 @@ def test_pr_review_human_list_prioritizes_actionable_fields( ) assert cloud.run_cloud(["pr-reviews", "list"]) == 0 + assert f"[{pr_state}]" in capsys.readouterr().out + + +@pytest.mark.parametrize( + ("record", "expected_targets"), + [ + ( + { + "id": "internal-scan-id", + "title": "Private network review", + "engagement_type": "internal_infra", + "scan_type": "blackbox", + "status": "running", + "internal_targets": ["10.24.0.0/16", "db.internal"], + "findings": {"total": 0}, + }, + ("10.24.0.0/16", "db.internal"), + ), + ( + { + "id": "upload-scan-id", + "title": "Local source review", + "engagement_type": "code_review", + "scan_type": "whitebox", + "status": "pending", + "has_code_upload": True, + "findings": {"total": 0}, + }, + ("uploaded source",), + ), + ], +) +def test_scan_human_list_identifies_internal_and_uploaded_targets( + monkeypatch: pytest.MonkeyPatch, + capsys: Any, + record: dict[str, Any], + expected_targets: tuple[str, ...], +) -> None: + monkeypatch.setattr(render.sys.stdout, "isatty", lambda: True) + monkeypatch.setattr( + http, + "request", + lambda *_a, **_k: FakeResponse( + payload={ + "items": [record], + "meta": { + "page": 1, + "limit": 20, + "total_items": 1, + "total_pages": 1, + "has_next": False, + }, + } + ), + ) + + assert cloud.run_cloud(["scans", "list"]) == 0 output = capsys.readouterr().out - for value in ( - "usestrix/strix", - "1177", - "Improve cloud CLI", - "feature", - "main", - "posted", - "pass", - "0 open / 0 total", - "review-id", - ): - assert value in output - for value in ("org-id", "user-id", "installation_id"): - assert value not in output + for target in expected_targets: + assert target in output @pytest.mark.parametrize( ("command", "payload", "visible", "hidden"), [ + ( + ["vulns", "list"], + { + "items": [ + { + "id": "vuln-id", + "scan_id": "scan-secret", + "display_number": 17, + "title": "Missing authorization", + "severity": "high", + "status": "open", + "target": None, + "method": "get", + "endpoint": "/api/admin", + "cvss": 8.2, + "finding_type": "dynamic", + } + ], + "meta": {"total_items": 1}, + }, + ("17", "Missing authorization", "GET /api/admin", "vuln-id"), + ("scan-secret",), + ), ( ["domains", "list"], { @@ -1956,7 +2184,6 @@ def test_pr_review_human_list_prioritizes_actionable_fields( "organization_id": "org-secret", "full_name": "usestrix/strix", "provider": "github", - "default_branch": "main", "pr_review_enabled": True, "tags": ["core"], "business_unit": "product", @@ -1966,7 +2193,7 @@ def test_pr_review_human_list_prioritizes_actionable_fields( ], "meta": {"total_items": 1}, }, - ("usestrix/strix", "github", "main", "yes", "repo-id"), + ("usestrix/strix", "github", "yes", "repo-id"), ("org-secret", "user-secret", "added by"), ), ( @@ -1997,7 +2224,7 @@ def test_pr_review_human_list_prioritizes_actionable_fields( "manual", "dashboard/notes/auth.md", "doc-id-that-is-deliberately-long-enough-to-require-a-lossless-copyable-value", - "Copyable IDs", + "Copyable selectors", ), ("org-secret", "Long private content"), ), @@ -2067,6 +2294,7 @@ def test_token_human_list_shows_lifecycle_status( "name": "Active CI", "type": "service", "scopes": ["scans:read"], + "rbac_scopes": [], "secret_prefix": "strix_svc_a", "expires_at": "2099-01-01T00:00:00Z", "last_used_at": None, @@ -2078,6 +2306,7 @@ def test_token_human_list_shows_lifecycle_status( "name": "Old CI", "type": "service", "scopes": ["scans:read"], + "rbac_scopes": [{"type": "tag", "value": "staging"}], "secret_prefix": "strix_svc_r", "expires_at": None, "last_used_at": None, @@ -2089,6 +2318,7 @@ def test_token_human_list_shows_lifecycle_status( "name": "Expired CI", "type": "service", "scopes": ["scans:read"], + "rbac_scopes": [{"type": "business_unit", "value": "payments"}], "secret_prefix": "strix_svc_e", "expires_at": "2000-01-01T00:00:00Z", "last_used_at": None, @@ -2101,11 +2331,1148 @@ def test_token_human_list_shows_lifecycle_status( assert cloud.run_cloud(["tokens", "list"]) == 0 output = capsys.readouterr().out - for value in ("Active CI", "active", "Old CI", "revoked", "Expired CI", "expired"): + for value in ( + "Active CI", + "active", + "all assets", + "Old CI", + "revoked", + "tag:staging", + "Expired CI", + "expired", + "business_unit:payments", + "scans:read", + ): assert value in output assert "org-secret" not in output +@pytest.mark.parametrize( + ("command", "payload", "visible", "hidden"), + [ + ( + ["chat", "list"], + { + "chats": [ + { + "id": "chat-id", + "title": "Investigate auth", + "status": "running", + "created_at": "2026-08-27T10:00:00Z", + "last_message_at": "2026-08-27T11:00:00Z", + } + ] + }, + ("Investigate auth", "running", "chat-id"), + (), + ), + ( + ["chat", "files", "chat-id"], + { + "files": [ + { + "path": "/workspace/" + "nested/" * 12 + "report.md", + "size": 42, + } + ] + }, + ( + "/workspace/" + "nested/" * 12 + "report.md", + "Copyable selectors", + "42", + ), + (), + ), + ( + ["chat", "findings", "chat-id"], + { + "findings": [ + { + "id": "finding-id", + "chat_id": "chat-secret", + "filed_by": "user-secret", + "title": "Broken access control", + "severity": "high", + "status": "open", + "target": None, + "method": "post", + "endpoint": "/admin/users", + "cvss": 8.1, + "filed_at": "2026-08-27T11:00:00Z", + "created_at": "2026-08-27T10:00:00Z", + } + ] + }, + ("Broken access control", "high", "POST /admin/users", "finding-id"), + ("chat-secret", "user-secret"), + ), + ( + ["scans", "agents", "scan-id"], + { + "scan_id": "scan-secret", + "agents": [ + { + "id": "agent-id", + "name": "Authorization tester", + "status": "completed", + "task": "Test object ownership", + "parent_id": None, + "created_at": "2026-08-27T10:00:00Z", + "finding_count": 2, + } + ], + }, + ("Authorization tester", "completed", "Test object ownership", "agent-id"), + ("scan-secret",), + ), + ( + ["scans", "retests", "scan-id"], + { + "runs": [ + { + "vulnerability_id": "vuln-id", + "title": "IDOR", + "severity": "high", + "issue_status": "open", + "retest_scan_id": "retest-id", + "retest_status": "running", + "created_at": "2026-08-27T10:00:00Z", + } + ], + "total": 1, + "completed": 0, + "running": 1, + }, + ("IDOR", "high", "vuln-id", "retest-id", "0/1 retest(s) complete"), + (), + ), + ( + ["pr-reviews", "findings", "--include-stats"], + { + "items": [ + { + "id": "pr-finding-id", + "pr_review_id": "review-secret", + "provider": "github", + "repository_full_name": "usestrix/strix", + "pr_number": 1177, + "pr_title": "Improve cloud CLI", + "pr_state": "open", + "title": "Unsafe redirect", + "severity": "medium", + "status": "open", + "created_at": "2026-08-27T10:00:00Z", + } + ], + "meta": {"total_items": 1}, + "stats": { + "prs_reviewed": 9, + "issues_found": 1, + "critical_high_found": 1, + "merges_blocked": 2, + }, + }, + ( + "usestrix/strix", + "1177", + "Improve cloud CLI", + "Unsafe redirect", + "pr-finding-id", + "Impact", + "prs reviewed 9", + "merges blocked 2", + ), + ("review-secret",), + ), + ( + ["vulns", "history", "vuln-id"], + [ + { + "id": "history-secret", + "vulnerability_id": "vuln-secret", + "previous_status": "snoozed", + "new_status": "snoozed", + "previous_severity": "high", + "new_severity": "medium", + "previous_snoozed_until": "2026-09-01T00:00:00Z", + "new_snoozed_until": "2026-09-15T00:00:00Z", + "changed_by": "user-secret", + "note": "Extended pending vendor fix", + "reason": "Vendor ETA changed", + "created_at": "2026-08-27T10:00:00Z", + }, + { + "id": "history-clear-secret", + "vulnerability_id": "vuln-secret", + "previous_status": "snoozed", + "new_status": "snoozed", + "previous_severity": "medium", + "new_severity": "medium", + "previous_snoozed_until": "2026-09-15T00:00:00Z", + "new_snoozed_until": None, + "changed_by": "user-secret", + "note": "Snooze removed", + "reason": "Fix available", + "created_at": "2026-08-28T10:00:00Z", + }, + ], + ( + "snoozed", + "high", + "medium", + "2026-09-01T00:00:00Z", + "2026-09-15T00:00:00Z", + "cleared", + "Extended pending vendor fix", + "Vendor ETA changed", + "Snooze removed", + "Fix available", + ), + ("history-secret", "history-clear-secret", "vuln-secret", "user-secret"), + ), + ( + ["repos", "supply-chain", "findings", "repo-id"], + { + "snapshot": {"id": "snapshot-secret"}, + "findings": [ + { + "id": "dependency-id", + "repository_id": "repo-secret", + "title": "Vulnerable package", + "package_name": "lodash", + "package_version": "1.0.0", + "severity": "high", + "status": "open", + "fixed_version": "4.17.21", + "manifest_path": "package-lock.json", + "direct": True, + } + ], + }, + ("Vulnerable package", "lodash@1.0.0", "4.17.21", "dependency-id"), + ("snapshot-secret", "repo-secret"), + ), + ( + ["repos", "supply-chain", "components", "repo-id"], + { + "snapshot": {"id": "snapshot-secret"}, + "components": [ + { + "id": "component-id", + "snapshot_id": "snapshot-secret", + "name": "requests", + "version": "2.0.0", + "ecosystem": "pypi", + "relationship": "direct", + "status": "active", + "highest_open_severity": "critical", + "manifest_path": "requirements.txt", + } + ], + "meta": {"total": 3, "limit": 1, "offset": 0}, + }, + ("requests", "2.0.0", "pypi", "critical", "component-id", "--offset 1"), + ("snapshot-secret",), + ), + ( + ["domains", "test-users", "inbox", "domain-id", "test-user-id"], + { + "address": "inbox@security-mail.strix.ai", + "messages": [ + { + "id": "message-id", + "from": "login@example.com", + "subject": "Your code", + "preview": "Code 123456", + "timestamp": "2026-08-27T10:00:00Z", + "detected_code": "123456", + } + ], + }, + ( + "login@example.com", + "Your code", + "123456", + "message-id", + "Inbox: inbox@security-mail.strix.ai", + ), + (), + ), + ( + ["knowledge", "repos", "entries", "usestrix/strix"], + { + "organization_id": "org-secret", + "repo_key": "usestrix/strix", + "profile": {"id": "profile-secret", "title": "Profile"}, + "docs": [ + { + "id": "doc-id", + "title": "Auth notes", + "source_type": "system", + "source_id": "repos/usestrix__strix/auth.md", + "tags": [], + "updated_at": "2026-08-27T10:00:00Z", + } + ], + "insights": [], + "policies": [{"id": "policy-secret", "policy_key": "no-prod"}], + "stats": {"docs_count": 1}, + }, + ( + "Auth notes", + "system", + "repos/usestrix__strix/auth.md", + "doc-id", + "Repository profile: Profile", + "1 policy apply", + ), + ("org-secret", "profile-secret", "policy-secret", "nested field"), + ), + ], +) +def test_nonstandard_human_list_envelopes_are_actionable( + monkeypatch: pytest.MonkeyPatch, + capsys: Any, + command: list[str], + payload: Any, + visible: tuple[str, ...], + hidden: tuple[str, ...], +) -> None: + monkeypatch.setattr(render.sys.stdout, "isatty", lambda: True) + monkeypatch.setattr(http, "request", lambda *_a, **_k: FakeResponse(payload=payload)) + + assert cloud.run_cloud(command) == 0 + output = capsys.readouterr().out + for value in visible: + assert value in output + for value in hidden: + assert value not in output + + +def test_chat_credentials_human_view_separates_attached_and_available_sources( + monkeypatch: pytest.MonkeyPatch, capsys: Any +) -> None: + payload = { + "credentials": [ + { + "label": "Attached admin", + "username": "admin@example.com", + "login_url": "https://example.com/login", + "mfa_method": "totp", + "has_password": True, + "has_totp_secret": True, + "test_user_id": "attached-test-user-id", + } + ], + "available_test_users": [ + { + "id": "available-test-user-id", + "label": "Saved analyst", + "username": "analyst@example.com", + "domain": "example.com", + "login_url": "https://example.com/login", + "mfa_method": "email_otp", + "has_password": False, + "has_totp_secret": False, + } + ], + "available_scan_credentials": [ + { + "scan_id": "source-scan-id", + "scan_title": "August staging pentest", + "username": "scan-user@example.com", + "login_url": "https://staging.example.com/login", + "mfa_method": "none", + "has_password": True, + "has_totp_secret": False, + } + ], + } + monkeypatch.setattr(render.sys.stdout, "isatty", lambda: True) + monkeypatch.setattr(http, "request", lambda *_a, **_k: FakeResponse(payload=payload)) + + command = ["chat", "credentials", "chat-id", "--scan-ids", "source-scan-id"] + assert cloud.run_cloud(command) == 0 + output = capsys.readouterr().out + for value in ( + "Attached credentials", + "Attached admin", + "attached-test-user-id", + "Available saved test users", + "Saved analyst", + "available-test-user-id", + "Credentials from requested scans", + "August staging pentest", + "source-scan-id", + "password: set", + "password: not set", + "--test-user-ids ID", + "--scan-ids SCAN_ID", + ): + assert value in output + + assert cloud.run_cloud([*command, "--json"]) == 0 + assert json.loads(capsys.readouterr().out) == payload + + +@pytest.mark.parametrize( + ("command", "payload", "visible", "hidden"), + [ + ( + ["schedules", "list"], + { + "schedules": [ + { + "id": "schedule-id", + "organization_id": "org-secret", + "created_by": "user-secret", + "name": "Weekly staging", + "cron": "0 9 * * 1", + "timezone": "America/New_York", + "isPaused": True, + "supply_chain": True, + "domain_ids": [], + "repository_ids": ["repo-id"], + "internal_targets": ["10.24.0.0/16"], + "connector_id": "connector-secret", + "last_run_status": "ok", + "next_run_at": "2026-08-31T13:00:00Z", + "run_count": 4, + } + ] + }, + ( + "Weekly staging", + "supply chain", + "1 repo", + "10.24.0.0/16", + "network connector", + "0 9 * * 1", + "paused", + "ok", + "schedule-id", + "schedules get ID", + ), + ("org-secret", "user-secret", "connector-secret"), + ), + ( + ["connectors", "list"], + [ + { + "id": "connector-id", + "name": "Private network", + "last_status": "healthy", + "last_status_checked_at": "2026-08-27T10:00:00Z", + "created_at": "2026-08-20T10:00:00Z", + "unexpected": "hidden", + } + ], + ("Private network", "healthy", "connector-id"), + ("unexpected", "hidden"), + ), + ( + ["org", "members"], + { + "members": [ + { + "id": "membership-id", + "userId": "user-secret", + "email": "analyst@example.com", + "firstName": "Ada", + "lastName": "Lovelace", + "role": "analyst", + "scopes": [ + {"type": "tag", "value": "production"}, + {"type": "business_unit", "value": "payments"}, + ], + "status": "active", + "joinedAt": "2026-08-20T10:00:00Z", + } + ] + }, + ( + "analyst@example.com", + "Ada", + "Lovelace", + "tag:production", + "business_unit:payments", + "active", + "membership-id", + ), + ("user-secret",), + ), + ( + ["org", "invitations"], + { + "invitations": [ + { + "id": "invitation-id", + "email": "invitee@example.com", + "role": "analyst", + "scopes": [], + "state": "pending", + "expiresAt": "2026-09-01T10:00:00Z", + "createdAt": "2026-08-27T10:00:00Z", + } + ] + }, + ("invitee@example.com", "analyst", "all assets", "pending", "invitation-id"), + (), + ), + ( + ["webhooks", "list"], + { + "webhooks": [ + { + "id": "webhook-id", + "organization_id": "org-secret", + "url": "https://example.com/hook", + "events": ["scan.completed"], + "business_unit": "product", + "is_active": True, + "last_success_at": "2026-08-27T10:00:00Z", + "last_failure_at": None, + "created_at": "2026-08-20T10:00:00Z", + } + ] + }, + ("https://example.com/hook", "scan.completed", "product", "webhook-id"), + ("org-secret", "last delivery"), + ), + ( + ["webhooks", "deliveries", "webhook-id"], + { + "items": [ + { + "id": "delivery-id", + "subscription_id": "subscription-secret", + "organization_id": "org-secret", + "event_type": "scan.completed", + "status": "delivered", + "response_status": 200, + "last_error": "temporary timeout", + "attempts": 1, + "sent_at": "2026-08-27T10:01:00Z", + "next_attempt_at": None, + "created_at": "2026-08-27T10:00:00Z", + } + ], + "meta": {"total_items": 1}, + }, + ("scan.completed", "delivered", "200", "temporary timeout", "delivery-id"), + ("subscription-secret", "org-secret"), + ), + ( + ["knowledge", "repos"], + { + "repos": [ + { + "repo_key": "usestrix/strix", + "docs_count": 4, + "last_updated_at": "2026-08-27T10:00:00Z", + "future_internal_field": "hidden", + } + ] + }, + ("usestrix/strix", "4", "2026-08-27"), + ("future_internal_field", "hidden"), + ), + ( + ["audit", "list"], + { + "data": [ + { + "id": "audit-row-secret", + "organization_id": "org-secret", + "actor_id": "actor-secret", + "actor_email": "ada@example.com", + "action": "scan.started", + "resource_type": "scan", + "resource_id": "scan-id", + "metadata": {"private": "details"}, + "ip_address": "192.0.2.1", + "created_at": "2026-08-27T10:00:00Z", + } + ], + "pagination": { + "page": 1, + "limit": 20, + "total": 41, + "total_pages": 3, + }, + }, + ("scan.started", "scan", "scan-id", "ada@example.com", "192.0.2.1", "--page 2"), + ("audit-row-secret", "org-secret", "actor-secret", "private"), + ), + ], +) +def test_named_human_list_views_match_api_fields( + monkeypatch: pytest.MonkeyPatch, + capsys: Any, + command: list[str], + payload: Any, + visible: tuple[str, ...], + hidden: tuple[str, ...], +) -> None: + monkeypatch.setattr(render.sys.stdout, "isatty", lambda: True) + monkeypatch.setattr(http, "request", lambda *_a, **_k: FakeResponse(payload=payload)) + + assert cloud.run_cloud(command) == 0 + output = capsys.readouterr().out + for value in visible: + assert value in output + for value in hidden: + assert value not in output + + +def test_supply_chain_org_summary_human_view_shows_totals_and_repository_risk( + monkeypatch: pytest.MonkeyPatch, capsys: Any +) -> None: + monkeypatch.setattr(render.sys.stdout, "isatty", lambda: True) + monkeypatch.setattr( + http, + "request", + lambda *_a, **_k: FakeResponse( + payload={ + "totals": { + "repositories": 2, + "components": 145, + "findings": 9, + "open_issues": 4, + "malicious": 1, + "suspicious": 2, + "vulnerable": 6, + "ecosystems": {"npm": 100, "pypi": 45}, + "severities": {"critical": 1, "high": 3, "medium": 5}, + }, + "repositories": [ + { + "repository": { + "id": "repo-id", + "organization_id": "org-secret", + "full_name": "usestrix/strix", + "provider": "github", + }, + "summary": { + "component_count": 100, + "finding_count": 7, + "malicious_count": 1, + "suspicious_count": 2, + "vulnerable_count": 4, + "severity_counts": {"critical": 1, "high": 2, "medium": 4}, + "policy": { + "enabled": True, + "pr_checks_enabled": False, + "mode": "block", + }, + }, + "latest_supply_chain_scan": { + "id": "scan-secret", + "status": "completed", + "created_at": "2026-08-27T10:00:00Z", + }, + }, + { + "repository": { + "id": "repo-id-2", + "organization_id": "org-secret", + "full_name": "usestrix/sdk", + "provider": "github", + }, + "summary": { + "component_count": 45, + "finding_count": 2, + "malicious_count": 0, + "suspicious_count": 0, + "vulnerable_count": 2, + "severity_counts": {"high": 1, "medium": 1}, + "policy": {"enabled": False}, + }, + "latest_supply_chain_scan": None, + }, + ], + } + ), + ) + + assert cloud.run_cloud(["supply-chain", "summary"]) == 0 + output = capsys.readouterr().out + for value in ( + "Supply-chain totals", + "145", + "open issues", + "usestrix/strix", + "critical 1", + "1 malicious", + "completed", + "block", + "PR checks off", + "usestrix/sdk", + "not run", + "disabled", + "repo-id", + ): + assert value in output + for value in ("org-secret", "scan-secret", "ecosystems"): + assert value not in output + assert "Use --json for complete totals and repository records" in output + + +@pytest.mark.parametrize( + ("command", "payload", "visible", "hidden"), + [ + ( + ["webhooks", "get", "webhook-id"], + { + "webhook": { + "id": "webhook-id", + "organization_id": "org-secret", + "url": "https://example.com/hook", + "events": ["scan.completed", "scan.failed"], + "business_unit": None, + "secret_prefix": "whsec_1234", + "is_active": True, + "last_success_at": "2026-08-27T10:00:00Z", + "last_failure_at": "2026-08-26T10:00:00Z", + "created_by": "user-secret", + "created_at": "2026-08-20T10:00:00Z", + "updated_at": "2026-08-27T10:00:00Z", + } + }, + ( + "webhook-id", + "https://example.com/hook", + "scan.completed", + "scan.failed", + "all organization", + "whsec_1234", + "yes", + "2026-08-26T10:00:00Z", + ), + ("org-secret", "user-secret", "nested field"), + ), + ( + ["chat", "get", "chat-id"], + { + "chat": { + "workspace_state": "running", + "id": "chat-id", + "title": "Investigate auth", + "status": "active", + "run_id": "run-id", + "sandbox_api_url": True, + "created_at": "2026-08-20T10:00:00Z", + "updated_at": "2026-08-27T10:00:00Z", + "last_message_at": "2026-08-27T09:59:00Z", + } + }, + ( + "chat-id", + "Investigate auth", + "active", + "workspace state", + "running", + "run-id", + "sandbox attached", + "yes", + "2026-08-27T09:59:00Z", + ), + ("sandbox api url", "nested field"), + ), + ], +) +def test_wrapped_detail_human_views_are_unwrapped_and_actionable( + monkeypatch: pytest.MonkeyPatch, + capsys: Any, + command: list[str], + payload: dict[str, Any], + visible: tuple[str, ...], + hidden: tuple[str, ...], +) -> None: + monkeypatch.setattr(render.sys.stdout, "isatty", lambda: True) + monkeypatch.setattr(http, "request", lambda *_a, **_k: FakeResponse(payload=payload)) + + assert cloud.run_cloud(command) == 0 + output = capsys.readouterr().out + for value in visible: + assert value in output + for value in hidden: + assert value not in output + + +def test_trace_human_view_summarizes_events_and_preserves_selector( + monkeypatch: pytest.MonkeyPatch, capsys: Any +) -> None: + monkeypatch.setattr(render.sys.stdout, "isatty", lambda: True) + seen_query: dict[str, Any] = {} + + def fake_trace_request(_method: str, _path: str, **kwargs: Any) -> FakeResponse: + seen_query.update(kwargs.get("query") or {}) + return FakeResponse( + payload={ + "scan_id": "scan-id", + "agent_id": "agent-id", + "steps": [ + { + "timestamp": "2026-08-27T10:00:00Z", + "kind": "tool_call", + "event_id": "event-id", + "tool_name": "browser", + "args": { + "url": "https://example.com", + "password": "trace-password-must-not-render", + "headers": {"Authorization": "Bearer trace-token-must-not-render"}, + "sk_live_secret-as-dictionary-key": True, + }, + "truncated": True, + }, + { + "timestamp": "2026-08-27T10:01:00Z", + "kind": "finding", + "event_id": "finding-event-id", + "finding": {"title": "IDOR", "severity": "high"}, + }, + { + "timestamp": "2026-08-27T10:02:00Z", + "kind": "tool_result", + "event_id": "result-event-id", + "tool_name": "browser", + "status": "completed", + "result": "result-token-must-not-render", + }, + ], + "cursor": "next-secret", + "has_more": True, + "note": "Older trace events remain available.", + } + ) + + monkeypatch.setattr( + http, + "request", + fake_trace_request, + ) + + assert ( + cloud.run_cloud( + [ + "scans", + "trace", + "scan-id", + "--agent-id", + "agent-id", + "--tool-name", + "browser", + "--limit", + "25", + ] + ) + == 0 + ) + output = capsys.readouterr().out + for value in ( + "tool_call", + "browser", + "arguments: 4 field(s)", + "high: IDOR", + "event-id", + "tool_result", + "result: text (", + ): + assert value in output + assert seen_query == {"agent_id": "agent-id", "tool_name": "browser", "limit": 25} + for secret in ( + "trace-password-must-not-render", + "trace-token-must-not-render", + "result-token-must-not-render", + "password", + "sk_live_secret-as-dictionary-key", + ): + assert secret not in output + assert "scans trace-event scan-id EVENT_ID" in output + normalized_output = " ".join(output.replace("`", "").split()) + assert "same trace command with --cursor next-secret" in normalized_output + assert "keep its --agent-id, --tool-name, and --limit options" in normalized_output + assert "Older trace events remain available." in normalized_output + + +def test_paginated_human_list_shows_total_and_continuation_command( + monkeypatch: pytest.MonkeyPatch, capsys: Any +) -> None: + monkeypatch.setattr(render.sys.stdout, "isatty", lambda: True) + monkeypatch.setattr( + http, + "request", + lambda *_a, **_k: FakeResponse( + payload={ + "items": [ + { + "id": "domain-id", + "domain": "staging.example.com", + "asset_type": "web_app", + } + ], + "meta": { + "page": 1, + "limit": 20, + "total_items": 51, + "total_pages": 3, + "has_next": True, + }, + } + ), + ) + + assert cloud.run_cloud(["domains", "list"]) == 0 + output = capsys.readouterr().out + assert "Page 1/3" in output + assert "51 total" in output + assert "--page 2" in output + + +def test_empty_paginated_human_list_does_not_claim_page_one_of_zero() -> None: + stream = io.StringIO() + console = Console(file=stream, width=120, color_system=None, force_terminal=False) + + render.emit( + console, + { + "items": [], + "meta": { + "page": 1, + "limit": 25, + "total_items": 0, + "total_pages": 0, + "has_next": False, + }, + }, + as_json=False, + view="GET /domains", + ) + + output = stream.getvalue() + assert "0 total." in output + assert "Page 1/0" not in output + + +def test_offset_pagination_explains_an_out_of_range_page() -> None: + stream = io.StringIO() + console = Console(file=stream, width=120, color_system=None, force_terminal=False) + + render.emit( + console, + {"components": [], "meta": {"total": 3, "limit": 2, "offset": 4}}, + as_json=False, + view="GET /repositories/{repositoryId}/supply-chain/components", + ) + + output = stream.getvalue() + assert "No items at offset 4; 3 total." in output + assert "--offset 2" in output + assert "Showing 3-3" not in output + + +def test_page_pagination_explains_an_out_of_range_page() -> None: + stream = io.StringIO() + console = Console(file=stream, width=120, color_system=None, force_terminal=False) + + render.emit( + console, + { + "items": [], + "meta": { + "page": 4, + "limit": 20, + "total_items": 51, + "total_pages": 3, + "has_next": False, + }, + }, + as_json=False, + view="GET /domains", + ) + + output = stream.getvalue() + assert "No items on page 4; 51 total." in output + assert "--page 3" in output + assert "Page 4/3" not in output + + +def test_human_detail_preserves_long_prose_beyond_table_cell_limit( + monkeypatch: pytest.MonkeyPatch, capsys: Any +) -> None: + monkeypatch.setattr(render.sys.stdout, "isatty", lambda: True) + description = ( + " ".join(["authorization context"] * 12) + " final-description-marker\nsecond-line-marker" + ) + assert len(description) > 60 + monkeypatch.setattr( + http, + "request", + lambda *_a, **_k: FakeResponse( + payload={ + "id": "vuln-id", + "title": "Cross-tenant access", + "description": description, + "remediation_steps": "Validate tenant ownership before every object lookup.", + } + ), + ) + + assert cloud.run_cloud(["vulns", "get", "vuln-id"]) == 0 + output = capsys.readouterr().out + assert "final-description-marker" in output + assert "second-line-marker" in output + assert "\\x0a" not in output + assert "Validate tenant ownership" in output + + +def test_human_detail_bounds_extreme_scalar_values() -> None: + assert render._detail_cell("x" * 2500).endswith("… [truncated; use --json]") + assert len(render._detail_cell("x" * 2500)) == 2000 + assert render._detail_cell("first\nsecond") == "first\nsecond" + + +def test_large_vulnerability_detail_prioritizes_evidence_and_remediation() -> None: + stream = io.StringIO() + console = Console(file=stream, width=240, color_system=None, force_terminal=False) + payload: dict[str, Any] = {f"future_field_{index}": f"value-{index}" for index in range(45)} + payload.update( + { + "id": "vuln-id", + "title": "Cross-tenant access", + "status": "open", + "severity": "high", + "description": "A caller can read another tenant's object.", + "technical_analysis": "The object lookup omits the tenant predicate.", + "evidence": "GET /objects/other-tenant returned HTTP 200.", + "remediation_steps": "Bind every object lookup to the authenticated tenant.", + "cwe": ["CWE-639"], + "location_meta": {"path": "src/routes/objects.ts", "line": 42}, + "fix_pr_eligible": True, + "fix_pr_reason": "A repository and exact code location are available.", + "fix_pr_url": "https://github.com/example/app/pull/42", + "filed_at": "2026-08-28T12:00:00Z", + "dependency_metadata": {"package": "example", "installed_version": "1.0.0"}, + } + ) + + render.emit(console, payload, as_json=False, view="GET /vulnerabilities/{vulnerabilityId}") + + output = stream.getvalue() + for value in ( + "Cross-tenant access", + "The object lookup omits the tenant predicate.", + "GET /objects/other-tenant returned HTTP 200.", + "Bind every object lookup to the authenticated tenant.", + "CWE-639", + "src/routes/objects.ts", + "A repository and exact code location are available.", + "https://github.com/example/app/pull/42", + "2026-08-28T12:00:00Z", + "package: example", + ): + assert value in output + assert "additional field(s) omitted" in output + + +def test_test_user_human_view_joins_latest_verification( + monkeypatch: pytest.MonkeyPatch, capsys: Any +) -> None: + monkeypatch.setattr(render.sys.stdout, "isatty", lambda: True) + monkeypatch.setattr( + http, + "request", + lambda *_a, **_k: FakeResponse( + payload={ + "items": [ + { + "id": "test-user-id", + "label": "Admin", + "username": "admin@example.com", + "mfa_method": "totp", + "has_password": True, + "has_totp_secret": True, + "login_url": "https://example.com/login", + "updated_at": "2026-08-27T10:00:00Z", + } + ], + "auth_checks": { + "test-user-id": { + "status": "failed", + "failure_code": "invalid_credentials", + } + }, + } + ), + ) + + assert cloud.run_cloud(["domains", "test-users", "list", "domain-id"]) == 0 + output = capsys.readouterr().out + for value in ("password: set", "totp (secret set)", "failed: invalid_credentials"): + assert value in output + + +def test_explicit_human_view_is_an_allowlist_and_preserves_uuid() -> None: + stream = io.StringIO() + console = Console(file=stream, width=120, color_system=None, force_terminal=False) + uuid = "4d3a33cc-5c96-4e91-921c-682093efe780" + + render.emit( + console, + { + "items": [ + { + "id": uuid, + "domain": "staging.example.com", + "asset_type": "web_app", + "verified": False, + "unknown_internal_scalar": "must-not-render", + } + ], + "meta": {"total_items": 1}, + }, + as_json=False, + view="GET /domains", + ) + + output = stream.getvalue() + assert uuid in output + assert "must-not-render" not in output + + +def test_wide_knowledge_table_keeps_title_readable_with_long_identifiers() -> None: + stream = io.StringIO() + console = Console(file=stream, width=120, color_system=None, force_terminal=False) + document_id = "document-selector-" + "x" * 80 + source_id = "repos/usestrix__strix/" + "nested/" * 12 + "authentication.md" + + render.emit( + console, + { + "organization_id": "org-secret", + "docs": [ + { + "id": document_id, + "title": "Authentication guidance", + "source_type": "system", + "source_id": source_id, + "tags": ["auth"], + "updated_at": "2026-08-27T10:00:00Z", + } + ], + "total": 1, + }, + as_json=False, + view="GET /knowledge", + ) + + output = stream.getvalue() + assert "Authentication guidance" in output + assert document_id in output + assert "Copyable selectors" in output + + def test_human_get_prioritizes_details_and_hides_internal_identity_fields( monkeypatch: pytest.MonkeyPatch, capsys: Any ) -> None: diff --git a/tests/test_cloud_cli_runtime.py b/tests/test_cloud_cli_runtime.py index c62365de..2aa5f39f 100644 --- a/tests/test_cloud_cli_runtime.py +++ b/tests/test_cloud_cli_runtime.py @@ -940,10 +940,10 @@ def test_device_flow_accepts_external_authkit_url_and_binds_token_origin( def test_missing_verb_json_is_structured(capsys: Any) -> None: - assert cloud.run_cloud(["scans", "--json"]) == 0 + assert cloud.run_cloud(["uploads", "--json"]) == 0 payload = json.loads(capsys.readouterr().out) - assert payload["command"] == "strix cloud scans" - assert any(item["name"] == "start" for item in payload["verbs"]) + assert payload["command"] == "strix cloud uploads" + assert any(item["name"] == "request" for item in payload["verbs"]) @pytest.mark.parametrize( @@ -981,8 +981,8 @@ def test_non_tty_dispatcher_always_emits_structured_json( assert cloud.run_cloud([]) == 0 assert json.loads(capsys.readouterr().out)["command"] == "strix cloud" - assert cloud.run_cloud(["scans"]) == 0 - assert json.loads(capsys.readouterr().out)["command"] == "strix cloud scans" + assert cloud.run_cloud(["uploads"]) == 0 + assert json.loads(capsys.readouterr().out)["command"] == "strix cloud uploads" assert cloud.run_cloud(["does-not-exist"]) == http.EXIT_USAGE assert json.loads(capsys.readouterr().out) == {"error": "unknown command: does-not-exist"} diff --git a/tests/test_cloud_source_upload.py b/tests/test_cloud_source_upload.py index 341fe05b..24004cd8 100644 --- a/tests/test_cloud_source_upload.py +++ b/tests/test_cloud_source_upload.py @@ -32,6 +32,11 @@ class FakeResponse: return self._payload +class MalformedJsonResponse(FakeResponse): + def json(self) -> Any: + raise ValueError("malformed JSON") + + @pytest.fixture(autouse=True) def _token_env(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setenv("STRIX_API_TOKEN", "test-token") @@ -467,7 +472,10 @@ def test_failed_scan_deletes_completed_source_upload( assert ("DELETE", "/uploads/upload-1") in paths -@pytest.mark.parametrize("failure", ["network", "server", "malformed_success"]) +@pytest.mark.parametrize( + "failure", + ["network", "server", "malformed_success", "malformed_json_success", "wrong_shape_success"], +) def test_ambiguous_scan_launch_retains_completed_source_upload( failure: str, tmp_path: Path, @@ -494,6 +502,10 @@ def test_ambiguous_scan_launch_retains_completed_source_upload( raise http.CloudError("connection closed before a response") if failure == "server": return FakeResponse({"detail": "temporary failure"}, status_code=500) + if failure == "malformed_json_success": + return MalformedJsonResponse("accepted") + if failure == "wrong_shape_success": + return FakeResponse({}) response = FakeResponse("accepted") response.headers = {"content-type": "text/html"} return response @@ -516,6 +528,37 @@ def test_ambiguous_scan_launch_retains_completed_source_upload( assert ("DELETE", "/uploads/upload-ambiguous") not in paths +def test_mismatched_upload_completion_response_is_cleaned_before_launch( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + (tmp_path / "app.py").write_text("print('safe')\n", encoding="utf-8") + paths: list[tuple[str, str]] = [] + + def fake_request(method: str, path: str, **_kwargs: Any) -> FakeResponse: + paths.append((method, path)) + if path == "/uploads/request": + return FakeResponse( + { + "upload_id": "upload-expected", + "signed_url": "https://storage.test/object", + "token": "signed", + } + ) + if path == "/uploads/complete": + return FakeResponse({"id": "upload-different"}) + if path == "/uploads/upload-expected": + return FakeResponse({"ok": True}) + if path == "/scans": + pytest.fail("a scan must not launch before upload completion is confirmed") + raise AssertionError(path) + + monkeypatch.setattr(http, "request", fake_request) + monkeypatch.setattr(http, "upload_file", lambda *_args, **_kwargs: None) + + assert cloud.run_cloud(["scans", "start", "--source", str(tmp_path), "--yes"]) == 1 + assert ("DELETE", "/uploads/upload-expected") in paths + + def test_interrupted_scan_launch_retains_completed_source_upload( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: Any ) -> None: