mirror of
https://github.com/usestrix/strix.git
synced 2026-10-08 03:08:08 +00:00
feat(cli): add the strix cloud command surface for the managed platform
This commit is contained in:
Alex Schapiro
parent
44fb7ac969
commit
56008a6c50
13 changed files with 2008 additions and 102 deletions
12
AGENTS.md
12
AGENTS.md
|
|
@ -38,12 +38,14 @@ Target-specific workflows built on the same engine:
|
|||
|
||||
- **Managed cloud (app.strix.ai):** no Docker, no LLM key, no local install; adds team dashboards, scheduling, PR reviews, and downloadable PDF/DOCX reports (Enterprise plan). Best in sandboxed/CI environments and for teams. Use it when local infra isn't available.
|
||||
```bash
|
||||
# token from `strix login` (device sign-in, stored in ~/.strix/platform-auth.json)
|
||||
# or from Settings → API Access; register the target as an asset, then:
|
||||
curl -sS https://app.strix.ai/api/v1/scans -H "Authorization: Bearer $STRIX_API_TOKEN" \
|
||||
-H "Content-Type: application/json" -d '{"engagement_type":"live_test","domain_ids":["<uuid>"]}'
|
||||
strix cloud login --scopes scans:read scans:write billing:read # device sign-in, no prompts
|
||||
strix cloud domains add --domain example.com --asset-type web_app
|
||||
strix cloud scans start --engagement-type live_test --domain-ids <uuid> --wait
|
||||
strix cloud vulns list --severity critical
|
||||
strix cloud billing topup --credits 20 # buy credits when a scan returns exit code 5
|
||||
```
|
||||
- API docs: https://docs.app.strix.ai (OpenAPI: https://docs.app.strix.ai/openapi.json).
|
||||
- Every REST operation has a `strix cloud <resource> <verb>` command. Run `strix cloud` to list them. Output is JSON when stdout is not a terminal (or with `--json`), and there are no prompts without a TTY. 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.
|
||||
- The REST API works directly too: https://docs.app.strix.ai (OpenAPI: https://docs.app.strix.ai/openapi.json).
|
||||
|
||||
- CLI docs index for LLMs: https://docs.strix.ai/llms.txt (full: https://docs.strix.ai/llms-full.txt).
|
||||
- Only scan targets the user is authorized to test.
|
||||
|
|
|
|||
38
README.md
38
README.md
|
|
@ -320,19 +320,41 @@ strix auth status # show the active sign-in
|
|||
strix auth logout # forget the sign-in
|
||||
```
|
||||
|
||||
#### Sign in to the managed platform
|
||||
#### Use the managed platform: `strix cloud`
|
||||
|
||||
To use the managed platform ([app.strix.ai](https://app.strix.ai)) from the terminal, run the device sign-in. It creates your account and workspace on first use and stores a personal API token in `~/.strix/platform-auth.json`:
|
||||
The `strix cloud` commands drive the managed platform ([app.strix.ai](https://app.strix.ai)) from the terminal. Sign in once with the device flow. The sign-in creates your account and workspace on first use and stores a personal API token in `~/.strix/platform-auth.json`:
|
||||
|
||||
```bash
|
||||
strix login # opens the browser, then pick a workspace and scopes
|
||||
strix login --scopes scans:read scans:write # request specific token scopes
|
||||
strix login --workspace "My Team" # select a workspace by name or ID
|
||||
strix login status # show the active sign-in
|
||||
strix login logout # forget the sign-in
|
||||
strix cloud login # opens the browser, then pick a workspace and scopes
|
||||
strix cloud login --scopes scans:read scans:write # request specific token scopes
|
||||
strix cloud login --workspace "My Team" # select a workspace by name or ID
|
||||
strix cloud whoami # show the active sign-in
|
||||
strix cloud logout # forget the sign-in
|
||||
```
|
||||
|
||||
The token authorizes the [REST API](https://docs.app.strix.ai) — scans, vulnerabilities, credits, and top-ups — with no dashboard visit.
|
||||
Every operation of the [REST API](https://docs.app.strix.ai) has a matching command in the form `strix cloud <resource> <verb>`:
|
||||
|
||||
```bash
|
||||
strix cloud # list all resources
|
||||
strix cloud scans # 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 <uuid> --wait
|
||||
strix cloud vulns list --severity critical
|
||||
strix cloud credits # credit balance
|
||||
strix cloud billing topup --credits 20 # buy credits (agent payment, HTTP 402)
|
||||
```
|
||||
|
||||
The commands are agent friendly. Output is JSON when stdout is not a terminal or when you pass `--json`. There are no prompts when stdin is not a terminal. Exit codes: `0` success, `1` error, `2` invalid usage, `4` authentication required, `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`:
|
||||
|
||||
```bash
|
||||
strix cloud scans start --data '{"engagement_type":"code_review"}' # literal JSON
|
||||
strix cloud scans start --data @request.json # read a file
|
||||
cat request.json | strix cloud scans start --data - # read standard input
|
||||
```
|
||||
|
||||
The platform enforces plan and role limits. Report downloads need the Enterprise plan, schedules need the Pro plan, and billing writes need an admin token. Those commands return the platform message and exit code `4`.
|
||||
|
||||
#### Connect your own MCP servers
|
||||
|
||||
|
|
|
|||
|
|
@ -36,13 +36,14 @@ npx skills use usestrix/strix@penetration-testing-with-strix | claude
|
|||
Both use the same engine and produce the same validated findings and SARIF, so agents can pick per situation or combine them:
|
||||
|
||||
- **Open-source CLI (self-hosted)** — runs locally in a Docker sandbox with your own LLM key. Free, fully local, air-gap capable. Best for local dev loops and full control.
|
||||
- **Managed cloud** — runs on Strix's infrastructure via the [app.strix.ai REST API](https://docs.app.strix.ai). No Docker, no LLM key, no local install; adds team dashboards, scheduling, PR reviews, and downloadable PDF/DOCX reports (Enterprise plan). Best in sandboxed/CI environments and for teams. Get an API token with `strix login` (browser device sign-in, account created on first use) or in the dashboard under **Settings → API Access**. The `managed-pentesting-with-strix` skill has the full flow.
|
||||
- **Managed cloud** — runs on Strix's infrastructure. Drive it with the `strix cloud` CLI (every REST operation has a `strix cloud <resource> <verb>` command) or the [app.strix.ai REST API](https://docs.app.strix.ai) directly. No Docker, no LLM key; adds team dashboards, scheduling, PR reviews, and downloadable PDF/DOCX reports (Enterprise plan). Best in sandboxed/CI environments and for teams. Sign in with `strix cloud login` (browser device sign-in, account created on first use) or create a token in the dashboard under **Settings → API Access**. The `managed-pentesting-with-strix` skill has the full flow.
|
||||
|
||||
## Agent-Friendly Interfaces
|
||||
|
||||
Everything an agent needs is machine-readable:
|
||||
|
||||
- **Headless CLI** — `strix -n` runs without the TUI and exits with `0` (clean), `1` (error), or `2` (vulnerabilities found).
|
||||
- **Cloud CLI** — `strix cloud` prints JSON when stdout is not a terminal (or with `--json`), never prompts without a TTY, and exits with `0` (success), `1` (error), `2` (usage), `4` (authentication required), or `5` (payment required). Credit top-ups pay the Stripe machine-payment challenge with an agent wallet (`strix cloud billing topup --credits N --yes`).
|
||||
- **REST API** — the managed platform exposes a documented [OpenAPI](https://docs.app.strix.ai/openapi.json) at `https://app.strix.ai/api/v1` (scans, vulnerabilities, assets, PR reviews, schedules, webhooks) with bearer tokens and scopes.
|
||||
- **Structured results** — every run writes `vulnerabilities.json`, `vulnerabilities.csv`, `findings.sarif` (SARIF 2.1.0), and per-finding Markdown under `strix_runs/<run-name>/`; the cloud exposes the same as JSON plus SARIF export.
|
||||
- **Budget controls** — `--max-budget` and `--max-turns` give agents hard cost/time caps.
|
||||
|
|
|
|||
|
|
@ -1,23 +1,46 @@
|
|||
---
|
||||
name: managed-pentesting-with-strix
|
||||
description: Run a managed pentest of a web app or API through the app.strix.ai REST API — no local Docker, LLM key, or install needed. Create an API token, register domain/repository assets, launch and poll scans, triage vulnerabilities, export SARIF, download PDF/DOCX pentest reports for SOC 2 and other compliance evidence (Enterprise plan), start PR reviews, and set up schedules and webhooks. Use when the user wants continuous or scheduled pentesting-as-a-service, an auditor-ready pentest report, scans tracked in a team dashboard, or security testing from a sandboxed agent/CI environment with no infrastructure.
|
||||
description: Run a managed pentest of a web app or API on the app.strix.ai platform with the `strix cloud` CLI or the REST API — no local Docker, LLM key, or install needed. Sign in with a browser device flow, register domain/repository assets, launch and poll scans, triage vulnerabilities, export SARIF, download PDF/DOCX pentest reports for SOC 2 and other compliance evidence (Enterprise plan), start PR reviews, buy credits with an agent payment, and set up schedules and webhooks. Use when the user wants continuous or scheduled pentesting-as-a-service, an auditor-ready pentest report, scans tracked in a team dashboard, or security testing from a sandboxed agent/CI environment with no infrastructure.
|
||||
license: Apache-2.0
|
||||
metadata:
|
||||
author: usestrix
|
||||
homepage: https://docs.app.strix.ai
|
||||
---
|
||||
|
||||
# Strix Cloud API (managed, no local infra)
|
||||
# Strix Cloud (managed, no local infra)
|
||||
|
||||
Use this when you want Strix's autonomous pentesting **without running Docker or an LLM yourself** — the scan runs on Strix's infrastructure and results are tracked in a team dashboard. This is the right choice in sandboxed/hosted agent and CI environments, for teams, and for scheduled/continuous testing (downloadable PDF/DOCX reports are an Enterprise-plan feature). For fully local, free, air-gapped, or BYO-LLM runs, use the open-source CLI in the **penetration-testing-with-strix** skill instead — both share the same engine and SARIF output, so you can mix them.
|
||||
|
||||
Full reference: **[docs.app.strix.ai](https://docs.app.strix.ai)** · OpenAPI: `https://docs.app.strix.ai/openapi.json`
|
||||
There are two equivalent interfaces. Prefer the CLI:
|
||||
|
||||
## Setup
|
||||
- **`strix cloud` CLI** — every REST operation has a command in the form `strix cloud <resource> <verb>`. Install with `curl -sSL https://strix.ai/install | bash`. Run `strix cloud` to list all resources and `strix cloud <resource>` to list its verbs.
|
||||
- **REST API** — base URL `https://app.strix.ai/api/v1`, `Authorization: Bearer <token>` on every request. Full reference: **[docs.app.strix.ai](https://docs.app.strix.ai)** · OpenAPI: `https://docs.app.strix.ai/openapi.json`.
|
||||
|
||||
- **Base URL:** `https://app.strix.ai/api/v1`
|
||||
- **Auth:** every request sends `Authorization: Bearer <token>`. Tokens are **org-scoped**.
|
||||
- **Get a token (no dashboard needed):** run `strix login` from the open-source CLI. It starts an OAuth device flow — the user approves the sign-in in the browser, the account and workspace are created on first use, and a personal API token is stored in `~/.strix/platform-auth.json` (read `api_token` from that file). In an interactive terminal the CLI then offers a workspace picker and scope presets (Recommended, Full access, Minimal, Custom). For non-interactive use pass `--scopes SCOPE ...` and `--workspace <name-or-id>` to skip the prompts. Alternatively the user creates a token in the dashboard at **Settings → API Access** (app.strix.ai). Never hardcode, log, or commit the token. Store it in an env var or the CI secret store.
|
||||
The CLI is agent friendly. Output is JSON when stdout is not a terminal, or when you pass `--json`. There are no interactive prompts when stdin is not a terminal. Exit codes: `0` success, `1` error, `2` invalid usage, `4` authentication required, `5` payment required.
|
||||
|
||||
Write commands take request fields as flags. Every write command also accepts one JSON object with `--data`, which is the way to send fields that have no flag:
|
||||
|
||||
```bash
|
||||
strix cloud scans start --data '{"engagement_type":"code_review"}' # literal JSON
|
||||
strix cloud scans start --data @request.json # read a file
|
||||
cat request.json | strix cloud scans start --data - # read standard input
|
||||
```
|
||||
|
||||
The platform enforces plan and role limits, and the CLI passes the platform message through. Report downloads need the Enterprise plan. Schedules need the Pro plan. Billing writes need an admin token. A blocked command exits with code `4`.
|
||||
|
||||
## Setup: sign in
|
||||
|
||||
Run the device sign-in. It creates the user's account and workspace on first use and stores a personal API token in `~/.strix/platform-auth.json`:
|
||||
|
||||
```bash
|
||||
strix cloud login --scopes scans:read scans:write billing:read vulnerabilities:read assets:read assets:write
|
||||
```
|
||||
|
||||
The user approves the sign-in in the browser. With `--scopes` (and optionally `--workspace <name-or-id>`) there are no prompts, so the command works from a non-interactive agent shell. In an interactive terminal without flags, the CLI offers a workspace picker and scope presets (Recommended, Full access, Minimal, Custom).
|
||||
|
||||
- `strix cloud whoami` shows the active sign-in. `strix cloud logout` removes it.
|
||||
- Every other `strix cloud` command uses the stored token automatically. To use a different token (for example one created in the dashboard at **Settings → API Access**), pass `--token <token>` or set `STRIX_API_TOKEN`.
|
||||
- Never hardcode, log, or commit the token. Store it in an env var or the CI secret store.
|
||||
- **Scopes (least-privilege):** assign only what the integration needs and rotate regularly:
|
||||
|
||||
| Scope | Grants |
|
||||
|
|
@ -31,30 +54,32 @@ Full reference: **[docs.app.strix.ai](https://docs.app.strix.ai)** · OpenAPI: `
|
|||
| `tokens:write` | create/revoke API tokens |
|
||||
| `billing:read` / `billing:write` | read credit balance & auto top-up settings · buy credits (admin) |
|
||||
|
||||
```bash
|
||||
export STRIX_API_TOKEN="<token>"
|
||||
BASE=https://app.strix.ai/api/v1
|
||||
auth=(-H "Authorization: Bearer $STRIX_API_TOKEN")
|
||||
```
|
||||
|
||||
All examples use `jq` to parse JSON. Handle HTTP errors: `401` bad/expired token, `402` out of credits, `403` scope/plan-tier limit, `422` validation error.
|
||||
HTTP errors map to messages and exit codes: `401` bad/expired token (exit `4`), `402` out of credits (exit `5`), `403` scope/plan-tier limit (exit `4`), `422` validation error (exit `1`).
|
||||
|
||||
## 0. Credits & top-ups
|
||||
|
||||
Scans consume org credits. Check the balance before a scan (`billing:read`):
|
||||
|
||||
```bash
|
||||
curl -sS "$BASE/billing/credits" "${auth[@]}" | jq '{credits_available, price_per_credit_cents}'
|
||||
strix cloud credits
|
||||
```
|
||||
|
||||
When the balance is too low, buy credits with `POST /billing/topup` (`billing:write`, admin token). The first request returns **HTTP 402 with a machine-payment challenge** (Stripe Machine Payments Protocol). Pay the challenge with a configured agent wallet — for example the [Link Agent Wallet](https://link.com/agents) — and retry with the payment credential. The response returns the receipt and the new balance. If no wallet is available, ask the user to top up in the dashboard instead.
|
||||
When the balance is too low, buy credits with `strix cloud billing topup` (`billing:write`, admin token). The server answers the first request with **HTTP 402 and a machine-payment challenge** (Stripe Machine Payments Protocol). The CLI pays the challenge with the `mppx` client when Node.js is available — the user approves the spend in their agent wallet, for example the [Link Agent Wallet](https://link.com/agents). The response returns the receipt (`credits_granted`, `duplicate`, `reference`) and the new balance.
|
||||
|
||||
```bash
|
||||
curl -sS "$BASE/billing/topup" "${auth[@]}" -H "Content-Type: application/json" \
|
||||
-d '{"credits": 20}' # → 402 + payment challenge, then retry with payment
|
||||
strix cloud billing topup --credits 20 --yes # --yes skips the confirmation prompt
|
||||
strix cloud billing topup --credits 20 --no-pay # print the 402 challenge without paying
|
||||
```
|
||||
|
||||
Automatic top-ups (admin): `GET /billing/auto-topup` shows the setting, `PUT /billing/auto-topup` enables it with `{"enabled": true, "topup_credits": 20, "monthly_cap_credits": 200}` — an omitted `monthly_cap_credits` keeps the stored cap, an explicit `null` removes it.
|
||||
If no wallet is available, print the challenge with `--no-pay` and ask the user to top up in the dashboard instead.
|
||||
|
||||
Automatic top-ups (admin): `strix cloud billing auto-topup` shows the setting. Enable it with:
|
||||
|
||||
```bash
|
||||
strix cloud billing auto-topup update --enabled --topup-credits 20 --monthly-cap-credits 200
|
||||
```
|
||||
|
||||
An omitted `--monthly-cap-credits` keeps the stored cap. Pass `--no-monthly-cap` to remove the cap.
|
||||
|
||||
## 1. Register the target as an asset
|
||||
|
||||
|
|
@ -62,67 +87,53 @@ Scans run against **registered assets**, not raw URLs. Register once, then reuse
|
|||
|
||||
```bash
|
||||
# Domain (black-box / live target). Requires domain verification before external scanning.
|
||||
# asset_type must be one of: web_app | api | attack_surface.
|
||||
curl -sS "$BASE/domains" "${auth[@]}" -H "Content-Type: application/json" \
|
||||
-d '{"domain":"staging.example.com","asset_type":"web_app"}' | jq '{id:.domain.id, status, reachable, verification}'
|
||||
# --asset-type must be one of: web_app | api | attack_surface.
|
||||
strix cloud domains add --domain staging.example.com --asset-type web_app
|
||||
|
||||
# Repository (white-box / code review). `full_name` is "owner/name".
|
||||
# Send one repository object, or a bare JSON array for several — not an object
|
||||
# wrapping a "repositories" key (that is rejected with 400).
|
||||
curl -sS "$BASE/repositories" "${auth[@]}" -H "Content-Type: application/json" \
|
||||
-d '[{"full_name":"org/app","provider":"github"}]' | jq '.repositories[] | {id, full_name}'
|
||||
strix cloud repos add --data '{"full_name":"org/app","provider":"github"}'
|
||||
```
|
||||
|
||||
Look up existing assets instead of re-adding: `GET /domains`, `GET /repositories` (both `assets:read`, paginated with `?page=&limit=`).
|
||||
Look up existing assets instead of re-adding: `strix cloud domains list`, `strix cloud repos list` (both `assets:read`).
|
||||
|
||||
## 2. Launch a scan
|
||||
|
||||
`POST /scans` (`scans:write`). Provide at least one target via `domain_ids`, `repository_ids`, or `internal_targets` (internal infra needs a network connector — see docs).
|
||||
`strix cloud scans start` (`scans:write`). Provide at least one target with `--domain-ids`, `--repository-ids`, or `--internal-targets` (internal infra needs a network connector — see docs).
|
||||
|
||||
```bash
|
||||
scan_id=$(curl -sS "$BASE/scans" "${auth[@]}" -H "Content-Type: application/json" -d '{
|
||||
"engagement_type": "live_test",
|
||||
"domain_ids": ["<domain-uuid>"],
|
||||
"focus": "IDOR, auth bypass, SSRF",
|
||||
"context": "Staging. Test account creds are configured as a test user.",
|
||||
"notify_on_completion": true
|
||||
}' | jq -r .scan_id)
|
||||
echo "$scan_id"
|
||||
strix cloud scans start \
|
||||
--engagement-type live_test \
|
||||
--domain-ids <domain-uuid> \
|
||||
--focus "IDOR, auth bypass, SSRF" \
|
||||
--context "Staging. Test account creds are configured as a test user." \
|
||||
--notify-on-completion
|
||||
```
|
||||
|
||||
Useful `CreateScanRequest` fields:
|
||||
Useful flags (each maps to a `CreateScanRequest` field):
|
||||
|
||||
| Field | Purpose |
|
||||
| Flag | Purpose |
|
||||
|---|---|
|
||||
| `engagement_type` | `live_test` (default), `code_review`, `internal_infra`, `compliance_pentest` |
|
||||
| `domain_ids` / `repository_ids` / `internal_targets` | targets (at least one) |
|
||||
| `domain_paths` / `repository_branches` | narrow to specific paths / branches |
|
||||
| `credentials` | authenticated scanning, incl. `mfa_method` (`totp`/`email_otp`/…) + `totp_secret` |
|
||||
| `headers` | extra HTTP headers (API keys, for example) for the target |
|
||||
| `focus` / `concerns` / `context` | steer the agents |
|
||||
| `upload_ids` | attach uploaded source/docs archives for white-box context |
|
||||
| `notify_on_completion` / `notification_emails` | email when done |
|
||||
| `--engagement-type` | `live_test` (default), `code_review`, `internal_infra`, `compliance_pentest` |
|
||||
| `--domain-ids` / `--repository-ids` / `--internal-targets` | targets (at least one) |
|
||||
| `--domain-paths` / `--repository-branches` | narrow to specific paths / branches (JSON maps) |
|
||||
| `--credentials` | authenticated scanning, incl. `mfa_method` (`totp`/`email_otp`/…) + `totp_secret` (JSON list) |
|
||||
| `--headers` | extra HTTP headers (API keys, for example) for the target (JSON map) |
|
||||
| `--focus` / `--concerns` / `--context` | steer the agents |
|
||||
| `--upload-ids` | attach uploaded source/docs archives for white-box context |
|
||||
| `--notify-on-completion` / `--notification-emails` | email when done |
|
||||
|
||||
Response is `{ scan_id, title, status }` with `status` = `pending`.
|
||||
The response is `{ scan_id, title, status }` with `status` = `pending`.
|
||||
|
||||
## 3. Poll to completion
|
||||
## 3. Wait for completion
|
||||
|
||||
`GET /scans/{scanId}` (`scans:read`). Status flow: `pending → running → completed` (or `failed` / `cancelled`). Poll on an interval — scans take minutes to hours. Do not block.
|
||||
|
||||
```bash
|
||||
while :; do
|
||||
s=$(curl -sS "$BASE/scans/$scan_id" "${auth[@]}" | jq -r .status)
|
||||
echo "status=$s"; [[ "$s" =~ ^(completed|failed|cancelled)$ ]] && break
|
||||
sleep 60
|
||||
done
|
||||
```
|
||||
Pass `--wait` to `scans start` to poll until the scan reaches a final state, or poll yourself with `strix cloud scans get <scan-id>` (`scans:read`). Status flow: `pending → running → completed` (or `failed` / `cancelled`). Scans take minutes to hours — poll on an interval, do not block.
|
||||
|
||||
## 4. Read findings
|
||||
|
||||
The scan-detail response includes `executive_summary`, `methodology`, `recommendations`, a `findings` severity roll-up, and a `vulnerabilities[]` array. Each vulnerability carries `title, severity, status, cvss, cwe, endpoint, method, impact, technical_analysis, poc_description, poc_script_code`, and (for code findings) `code_file`/`code_diff`/`code_before`/`code_after`.
|
||||
|
||||
```bash
|
||||
curl -sS "$BASE/scans/$scan_id" "${auth[@]}" \
|
||||
strix cloud scans get <scan-id> --json \
|
||||
| jq '["critical","high","medium","low","info"] as $order
|
||||
| .vulnerabilities
|
||||
| sort_by(.severity as $s | $order | index($s))
|
||||
|
|
@ -131,37 +142,35 @@ curl -sS "$BASE/scans/$scan_id" "${auth[@]}" \
|
|||
|
||||
Cloud severities are `critical | high | medium | low` and statuses are `open | in_progress | fixed | ignored`. Sort by an explicit severity order rather than `sort_by(.severity)`, which sorts alphabetically (critical, high, low, medium).
|
||||
|
||||
Org-wide triage across scans: `GET /vulnerabilities` (`vulnerabilities:read`; filter by severity/status). Update triage state with the vulnerabilities `:write` endpoints. To remediate, hand off to the **fix-security-vulnerabilities-with-strix** skill.
|
||||
Org-wide triage across scans: `strix cloud vulns list --severity critical` (`vulnerabilities:read`, and it also filters by `--status`, `--scan-id`, and more). Update triage state with `strix cloud vulns update <id> --status fixed`. To remediate, hand off to the **fix-security-vulnerabilities-with-strix** skill.
|
||||
|
||||
## 5. Export & report
|
||||
|
||||
```bash
|
||||
# SARIF 2.1.0 for GitHub code scanning / ASPM ingestion
|
||||
curl -sS "$BASE/scans/$scan_id/sarif" "${auth[@]}" -o findings.sarif
|
||||
strix cloud scans sarif <scan-id> --output findings.sarif
|
||||
|
||||
# Report. The format and file type are query params (`Accept` is ignored):
|
||||
# format=technical (default) | retest | attestation | executive_summary
|
||||
# type=pdf (default) | docx
|
||||
# Any report download requires the Enterprise plan; formats beyond `technical`,
|
||||
# Report. Formats: technical (default) | retest | attestation | executive_summary
|
||||
# Types: pdf (default) | docx
|
||||
# Any report download requires the Enterprise plan. Formats beyond `technical`,
|
||||
# DOCX, and white-label branding are Enterprise-only too. Scan must be completed.
|
||||
curl -sS "$BASE/scans/$scan_id/report?format=technical&type=pdf" "${auth[@]}" -o strix-report.pdf
|
||||
strix cloud scans report <scan-id> --format technical --type pdf --output strix-report.pdf
|
||||
```
|
||||
|
||||
## 6. PR reviews
|
||||
|
||||
Trigger an automated security review of a pull request (`pr_reviews:write`); results appear as PR comments and in the dashboard:
|
||||
Trigger an automated security review of a pull request (`pr_reviews:write`). The results appear as PR comments and in the dashboard:
|
||||
|
||||
```bash
|
||||
curl -sS "$BASE/pr-reviews/start" "${auth[@]}" -H "Content-Type: application/json" \
|
||||
-d '{"repository_full_name":"org/app","pr_number":123}'
|
||||
strix cloud pr-reviews start --repository-full-name org/app --pr-number 123
|
||||
```
|
||||
|
||||
List/inspect via `GET /pr-reviews` and `GET /pr-reviews/{id}`. Repo-level PR-review behavior is configured with the repository-settings endpoint.
|
||||
List/inspect with `strix cloud pr-reviews list` and `strix cloud pr-reviews get <id>`. Repo-level PR-review behavior is configured with `strix cloud pr-reviews settings`.
|
||||
|
||||
## 7. Continuous testing (schedules & webhooks)
|
||||
|
||||
- **Schedules** (`schedules:write`, Pro plan): create recurring scans and trigger them on demand — the managed equivalent of a cron-driven CLI loop.
|
||||
- **Webhooks** (`webhooks:write`): subscribe to pentest/vulnerability lifecycle events such as `scan.completed` and `vulnerability.created` to push results into Slack, ticketing, or your own pipeline instead of polling.
|
||||
- **Schedules** (`schedules:write`, Pro plan): `strix cloud schedules create` makes recurring scans, and `strix cloud schedules trigger <id>` runs one on demand — the managed equivalent of a cron-driven CLI loop.
|
||||
- **Webhooks** (`webhooks:write`): `strix cloud webhooks create` subscribes to pentest/vulnerability lifecycle events such as `scan.completed` and `vulnerability.created` to push results into Slack, ticketing, or your own pipeline instead of polling.
|
||||
|
||||
See the schedules and webhooks sections at [docs.app.strix.ai](https://docs.app.strix.ai) for payloads.
|
||||
|
||||
|
|
|
|||
72
strix/interface/cloud/__init__.py
Normal file
72
strix/interface/cloud/__init__.py
Normal file
|
|
@ -0,0 +1,72 @@
|
|||
"""`strix cloud` — the managed Strix platform (app.strix.ai) from the terminal.
|
||||
|
||||
Every command maps to one operation of the public REST API. Output is JSON
|
||||
when stdout is not a terminal, so agents can parse every result. Exit codes:
|
||||
0 success, 1 error, 2 invalid usage, 4 authentication required, 5 payment
|
||||
required.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from rich.console import Console
|
||||
|
||||
from strix.interface.cloud.runner import resolve, run
|
||||
from strix.interface.cloud.spec import DEFAULT_VERBS, GROUP_HELP, SPEC
|
||||
from strix.interface.platform_cli import run_login
|
||||
|
||||
|
||||
_USAGE_HEADER = """[bold]Usage:[/] strix cloud <command> [arguments]
|
||||
|
||||
[bold]Session commands:[/]
|
||||
login Sign in to the managed platform and store an API token
|
||||
logout Remove the stored API token
|
||||
whoami Show the stored account, workspace, and token state
|
||||
credits Show the credit balance of the workspace
|
||||
|
||||
[bold]Resource commands:[/]"""
|
||||
|
||||
_USAGE_FOOTER = """
|
||||
Run [bold]strix cloud <command>[/] without a verb to list its verbs.
|
||||
Every command accepts [bold]--json[/] and [bold]--token[/]. Write commands
|
||||
accept [bold]--data[/] with a JSON object of extra request fields.
|
||||
API reference: https://docs.app.strix.ai"""
|
||||
|
||||
|
||||
def run_cloud(argv: list[str]) -> int:
|
||||
"""Entry point for ``strix cloud …``. Returns a process exit code."""
|
||||
console = Console()
|
||||
if not argv or argv[0] in ("-h", "--help", "help"):
|
||||
_print_usage(console)
|
||||
return 0
|
||||
|
||||
group, rest = argv[0], argv[1:]
|
||||
if group in ("login", "logout", "whoami"):
|
||||
session_argv = {"login": rest, "logout": ["logout"], "whoami": ["status"]}
|
||||
return run_login(session_argv[group])
|
||||
if group == "credits":
|
||||
group, rest = "billing", ["credits", *rest]
|
||||
|
||||
if group not in SPEC:
|
||||
console.print(f"[red]Unknown command:[/] {group}")
|
||||
_print_usage(console)
|
||||
return 2
|
||||
resolved = resolve(group, rest)
|
||||
if resolved is None:
|
||||
_print_verbs(console, group)
|
||||
return 0 if not rest or rest[0] in ("-h", "--help", "help") else 2
|
||||
cmd, remaining = resolved
|
||||
verb_label = " ".join(rest[: len(rest) - len(remaining)]) or DEFAULT_VERBS.get(group, "")
|
||||
return run(group, verb_label, cmd, remaining)
|
||||
|
||||
|
||||
def _print_usage(console: Console) -> None:
|
||||
console.print(_USAGE_HEADER)
|
||||
for group in SPEC:
|
||||
console.print(f" {group:<14}{GROUP_HELP.get(group, '')}")
|
||||
console.print(_USAGE_FOOTER)
|
||||
|
||||
|
||||
def _print_verbs(console: Console, group: str) -> None:
|
||||
console.print(f"[bold]strix cloud {group}[/] verbs:")
|
||||
for verb, cmd in SPEC[group].items():
|
||||
console.print(f" {verb:<28}{cmd.help}")
|
||||
114
strix/interface/cloud/http.py
Normal file
114
strix/interface/cloud/http.py
Normal file
|
|
@ -0,0 +1,114 @@
|
|||
"""HTTP client for the managed Strix platform API (app.strix.ai)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from typing import Any, cast
|
||||
|
||||
import requests
|
||||
|
||||
from strix.config import load_settings
|
||||
from strix.interface.platform_cli import read_record
|
||||
|
||||
|
||||
_DEFAULT_TIMEOUT_S = 120
|
||||
_app_url_override: str | None = None
|
||||
_timeout_s: float = _DEFAULT_TIMEOUT_S
|
||||
|
||||
EXIT_OK = 0
|
||||
EXIT_ERROR = 1
|
||||
EXIT_AUTH = 4
|
||||
EXIT_PAYMENT = 5
|
||||
|
||||
|
||||
class CloudError(Exception):
|
||||
"""A failed cloud command. Carries the process exit code."""
|
||||
|
||||
def __init__(self, message: str, *, exit_code: int = EXIT_ERROR, payload: Any = None) -> None:
|
||||
super().__init__(message)
|
||||
self.exit_code = exit_code
|
||||
self.payload = payload
|
||||
|
||||
|
||||
def configure(*, base_url: str | None = None, timeout: float | None = None) -> None:
|
||||
"""Set the platform URL and the request timeout for this process."""
|
||||
global _app_url_override, _timeout_s # noqa: PLW0603
|
||||
if base_url:
|
||||
_app_url_override = base_url.rstrip("/")
|
||||
if timeout:
|
||||
_timeout_s = timeout
|
||||
|
||||
|
||||
def app_url() -> str:
|
||||
if _app_url_override:
|
||||
return _app_url_override
|
||||
return load_settings().viewer.app_url.rstrip("/")
|
||||
|
||||
|
||||
def api_token(override: str | None = None) -> str:
|
||||
token = override or os.environ.get("STRIX_API_TOKEN")
|
||||
if not token:
|
||||
record = read_record()
|
||||
if record is not None:
|
||||
stored = record.get("api_token")
|
||||
if isinstance(stored, str):
|
||||
token = stored
|
||||
if not token or not token.strip():
|
||||
raise CloudError(
|
||||
"not signed in. Run `strix cloud login`, or set STRIX_API_TOKEN.",
|
||||
exit_code=EXIT_AUTH,
|
||||
)
|
||||
return token.strip()
|
||||
|
||||
|
||||
def request(
|
||||
method: str,
|
||||
path: str,
|
||||
*,
|
||||
token: str | None = None,
|
||||
query: dict[str, Any] | None = None,
|
||||
body: dict[str, Any] | None = None,
|
||||
) -> requests.Response:
|
||||
url = f"{app_url()}/api/v1{path}"
|
||||
headers = {"Authorization": f"Bearer {api_token(token)}"}
|
||||
try:
|
||||
response = requests.request(
|
||||
method,
|
||||
url,
|
||||
headers=headers,
|
||||
params={k: v for k, v in (query or {}).items() if v is not None} or None,
|
||||
json=body,
|
||||
timeout=_timeout_s,
|
||||
)
|
||||
except requests.RequestException as exc:
|
||||
raise CloudError(f"could not reach {app_url()}: {exc}") from exc
|
||||
return response
|
||||
|
||||
|
||||
def parsed(response: requests.Response) -> Any:
|
||||
content_type = response.headers.get("content-type", "")
|
||||
if "application/json" in content_type:
|
||||
try:
|
||||
return response.json()
|
||||
except ValueError:
|
||||
return response.text
|
||||
return response.text
|
||||
|
||||
|
||||
def check(response: requests.Response) -> Any:
|
||||
data = parsed(response)
|
||||
if response.ok:
|
||||
return data
|
||||
detail = ""
|
||||
if isinstance(data, dict):
|
||||
raw = cast("dict[str, Any]", data)
|
||||
detail = str(raw.get("detail") or raw.get("error") or "")
|
||||
message = detail or f"HTTP {response.status_code}"
|
||||
if response.status_code in (401, 403):
|
||||
raise CloudError(message, exit_code=EXIT_AUTH, payload=data)
|
||||
if response.status_code == 402:
|
||||
hint = detail or (
|
||||
"not enough credits. Run `strix cloud billing topup --credits N` to buy credits."
|
||||
)
|
||||
raise CloudError(hint, exit_code=EXIT_PAYMENT, payload=data)
|
||||
raise CloudError(message, exit_code=EXIT_ERROR, payload=data)
|
||||
94
strix/interface/cloud/render.py
Normal file
94
strix/interface/cloud/render.py
Normal file
|
|
@ -0,0 +1,94 @@
|
|||
"""Output rendering for `strix cloud` commands."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import sys
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
from rich.table import Table
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from rich.console import Console
|
||||
|
||||
|
||||
_MAX_TABLE_COLUMNS = 8
|
||||
_MAX_CELL_LENGTH = 60
|
||||
|
||||
_PREFERRED_KEYS = (
|
||||
"id",
|
||||
"name",
|
||||
"title",
|
||||
"domain",
|
||||
"status",
|
||||
"state",
|
||||
"severity",
|
||||
"role",
|
||||
"email",
|
||||
"url",
|
||||
"scan_type",
|
||||
"engagement_type",
|
||||
"created_at",
|
||||
"updated_at",
|
||||
)
|
||||
|
||||
|
||||
def json_mode(*, flag: bool) -> bool:
|
||||
"""JSON output is on when the flag is set or when stdout is not a terminal."""
|
||||
return flag or not sys.stdout.isatty()
|
||||
|
||||
|
||||
def emit(console: Console, data: Any, *, as_json: bool) -> None:
|
||||
if as_json:
|
||||
sys.stdout.write(json.dumps(data, indent=2, default=str) + "\n")
|
||||
return
|
||||
rows = _list_of_dicts(data)
|
||||
if rows is not None:
|
||||
_print_table(console, rows)
|
||||
return
|
||||
if isinstance(data, str):
|
||||
console.print(data)
|
||||
return
|
||||
console.print_json(json.dumps(data, default=str))
|
||||
|
||||
|
||||
def _list_of_dicts(data: Any) -> list[dict[str, Any]] | None:
|
||||
if isinstance(data, dict) and len(data) >= 1:
|
||||
lists = [v for v in data.values() if isinstance(v, list)]
|
||||
scalars = [v for v in data.values() if not isinstance(v, list | dict)]
|
||||
if len(lists) == 1 and not scalars:
|
||||
data = lists[0]
|
||||
if not isinstance(data, list) or not data:
|
||||
return None
|
||||
if not all(isinstance(item, dict) for item in data):
|
||||
return None
|
||||
return data
|
||||
|
||||
|
||||
def _print_table(console: Console, rows: list[dict[str, Any]]) -> None:
|
||||
columns: list[str] = [key for key in _PREFERRED_KEYS if any(key in row for row in rows)]
|
||||
for row in rows:
|
||||
for key in row:
|
||||
if (
|
||||
key not in columns
|
||||
and len(columns) < _MAX_TABLE_COLUMNS
|
||||
and not isinstance(row[key], dict | list)
|
||||
):
|
||||
columns.append(key)
|
||||
table = Table(show_lines=False)
|
||||
for column in columns[:_MAX_TABLE_COLUMNS]:
|
||||
table.add_column(column)
|
||||
for row in rows:
|
||||
table.add_row(*[_cell(row.get(column)) for column in columns[:_MAX_TABLE_COLUMNS]])
|
||||
console.print(table)
|
||||
console.print(f"[dim]{len(rows)} item(s). Use --json for the full records.[/]")
|
||||
|
||||
|
||||
def _cell(value: Any) -> str:
|
||||
if value is None:
|
||||
return ""
|
||||
text = str(value)
|
||||
if len(text) > _MAX_CELL_LENGTH:
|
||||
return text[: _MAX_CELL_LENGTH - 1] + "…"
|
||||
return text
|
||||
342
strix/interface/cloud/runner.py
Normal file
342
strix/interface/cloud/runner.py
Normal file
|
|
@ -0,0 +1,342 @@
|
|||
"""Generic command runner for `strix cloud`.
|
||||
|
||||
The runner turns one entry of the command table into an argument parser,
|
||||
sends the HTTP request, renders the result, and returns the exit code.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Any, cast
|
||||
|
||||
from rich.console import Console
|
||||
|
||||
from strix.interface.cloud import http
|
||||
from strix.interface.cloud.render import emit, json_mode
|
||||
from strix.interface.cloud.spec import DEFAULT_VERBS, SPEC, Cmd, P
|
||||
|
||||
|
||||
_PLACEHOLDER = re.compile(r"\{([^{}]+)\}")
|
||||
_CAMEL_BOUNDARY = re.compile(r"(?<=[a-z0-9])(?=[A-Z])")
|
||||
_WAIT_POLL_S = 15
|
||||
_TERMINAL_STATUSES = frozenset(
|
||||
{
|
||||
"completed",
|
||||
"failed",
|
||||
"cancelled",
|
||||
"canceled",
|
||||
"stopped",
|
||||
"error",
|
||||
"expired",
|
||||
"succeeded",
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def _dest(name: str) -> str:
|
||||
return _CAMEL_BOUNDARY.sub("_", name).lower()
|
||||
|
||||
|
||||
def _metavar(name: str) -> str:
|
||||
return _CAMEL_BOUNDARY.sub("_", name).upper()
|
||||
|
||||
|
||||
def resolve(group: str, tokens: list[str]) -> tuple[Cmd, list[str]] | None:
|
||||
"""Find the command for a verb. Two-word verbs match before one-word verbs."""
|
||||
commands = SPEC.get(group)
|
||||
if commands is None:
|
||||
return None
|
||||
if len(tokens) >= 2:
|
||||
two = f"{tokens[0]} {tokens[1]}"
|
||||
if two in commands:
|
||||
return commands[two], tokens[2:]
|
||||
if tokens and tokens[0] in commands:
|
||||
return commands[tokens[0]], tokens[1:]
|
||||
default = DEFAULT_VERBS.get(group)
|
||||
if default is not None and (not tokens or tokens[0].startswith("-")):
|
||||
return commands[default], tokens
|
||||
return None
|
||||
|
||||
|
||||
def run(group: str, verb_label: str, cmd: Cmd, argv: list[str]) -> int:
|
||||
console = Console()
|
||||
parser = _build_parser(group, verb_label, cmd)
|
||||
try:
|
||||
args = parser.parse_args(argv)
|
||||
except SystemExit as exc:
|
||||
return exc.code if isinstance(exc.code, int) else 2
|
||||
|
||||
path = cmd.path
|
||||
for name in _PLACEHOLDER.findall(cmd.path):
|
||||
path = path.replace("{" + name + "}", str(getattr(args, _dest(name))))
|
||||
|
||||
as_json = json_mode(flag=bool(getattr(args, "json", False)))
|
||||
token = getattr(args, "token", None)
|
||||
http.configure(base_url=getattr(args, "app_url", None), timeout=getattr(args, "timeout", None))
|
||||
|
||||
try:
|
||||
query = _collect(args, cmd.query)
|
||||
body = _collect(args, cmd.body)
|
||||
data = getattr(args, "data", None)
|
||||
if data:
|
||||
body.update(_load_data(data))
|
||||
if getattr(args, "no_monthly_cap", False):
|
||||
body["monthly_cap_credits"] = None
|
||||
return _execute(console, cmd, args, path, query, body, as_json=as_json, token=token)
|
||||
except http.CloudError as exc:
|
||||
_emit_error(console, exc, as_json=as_json)
|
||||
return exc.exit_code
|
||||
|
||||
|
||||
def _execute(
|
||||
console: Console,
|
||||
cmd: Cmd,
|
||||
args: argparse.Namespace,
|
||||
path: str,
|
||||
query: dict[str, Any],
|
||||
body: dict[str, Any],
|
||||
*,
|
||||
as_json: bool,
|
||||
token: str | None,
|
||||
) -> int:
|
||||
if cmd.path == "/billing/topup":
|
||||
return _topup(console, args, body, as_json=as_json, token=token)
|
||||
response = http.request(
|
||||
cmd.method,
|
||||
path,
|
||||
token=token,
|
||||
query=query or None,
|
||||
body=body if cmd.method in ("POST", "PUT", "PATCH") else None,
|
||||
)
|
||||
if cmd.binary:
|
||||
return _emit_binary(console, response, getattr(args, "output", None))
|
||||
result = http.check(response)
|
||||
if getattr(args, "wait", False):
|
||||
if cmd.wait_self:
|
||||
result = _poll(console, path, token=token, as_json=as_json)
|
||||
elif cmd.wait_path:
|
||||
result = _wait(console, cmd, result, token=token, as_json=as_json)
|
||||
emit(console, result, as_json=as_json)
|
||||
return http.EXIT_OK
|
||||
|
||||
|
||||
def _load_data(value: str) -> dict[str, Any]:
|
||||
"""Read a JSON object from a literal string, a `@file` path, or `-` for stdin."""
|
||||
if value == "-":
|
||||
text = sys.stdin.read()
|
||||
elif value.startswith("@"):
|
||||
path = Path(value[1:]).expanduser()
|
||||
try:
|
||||
text = path.read_text(encoding="utf-8")
|
||||
except OSError as exc:
|
||||
raise http.CloudError(f"could not read {path}: {exc}") from exc
|
||||
else:
|
||||
text = value
|
||||
try:
|
||||
parsed_value = json.loads(text)
|
||||
except ValueError as exc:
|
||||
raise http.CloudError("--data must be a JSON object.") from exc
|
||||
if not isinstance(parsed_value, dict):
|
||||
raise http.CloudError("--data must be a JSON object.")
|
||||
return cast("dict[str, Any]", parsed_value)
|
||||
|
||||
|
||||
def _build_parser(group: str, verb_label: str, cmd: Cmd) -> argparse.ArgumentParser:
|
||||
parser = argparse.ArgumentParser(prog=f"strix cloud {group} {verb_label}", description=cmd.help)
|
||||
for name in _PLACEHOLDER.findall(cmd.path):
|
||||
parser.add_argument(_dest(name), metavar=_metavar(name))
|
||||
for param in cmd.query + cmd.body:
|
||||
_add_option(parser, param)
|
||||
parser.add_argument("--json", action="store_true", help="Print the raw JSON response.")
|
||||
parser.add_argument("--token", default=None, help="API token override.")
|
||||
parser.add_argument("--app-url", default=None, metavar="URL", help="Platform URL override.")
|
||||
parser.add_argument(
|
||||
"--timeout",
|
||||
default=None,
|
||||
type=float,
|
||||
metavar="SECONDS",
|
||||
help="Request timeout in seconds.",
|
||||
)
|
||||
if cmd.method in ("POST", "PUT", "PATCH"):
|
||||
parser.add_argument(
|
||||
"--data",
|
||||
default=None,
|
||||
metavar="JSON",
|
||||
help="JSON object with extra request fields. Use @file to read a file, or - for stdin.",
|
||||
)
|
||||
if cmd.path == "/billing/auto-topup" and cmd.method == "PUT":
|
||||
parser.add_argument(
|
||||
"--no-monthly-cap",
|
||||
action="store_true",
|
||||
help="Remove the monthly cap. Omit this flag to keep the stored cap.",
|
||||
)
|
||||
if cmd.binary:
|
||||
parser.add_argument("--output", default=None, metavar="FILE", help="Write to this file.")
|
||||
if cmd.wait_path or cmd.wait_self:
|
||||
parser.add_argument(
|
||||
"--wait", action="store_true", help="Wait until the operation reaches a final state."
|
||||
)
|
||||
if cmd.path == "/billing/topup":
|
||||
parser.add_argument(
|
||||
"--yes", action="store_true", help="Do not ask for confirmation before payment."
|
||||
)
|
||||
parser.add_argument(
|
||||
"--no-pay",
|
||||
action="store_true",
|
||||
help="Print the payment challenge instead of paying it.",
|
||||
)
|
||||
return parser
|
||||
|
||||
|
||||
def _add_option(parser: argparse.ArgumentParser, param: P) -> None:
|
||||
flag = "--" + (param.flag or param.name.replace("_", "-"))
|
||||
if param.kind == "bool":
|
||||
parser.add_argument(
|
||||
flag,
|
||||
dest=param.name,
|
||||
action=argparse.BooleanOptionalAction,
|
||||
default=None,
|
||||
required=param.required,
|
||||
help=param.help,
|
||||
)
|
||||
elif param.kind == "list":
|
||||
parser.add_argument(
|
||||
flag,
|
||||
dest=param.name,
|
||||
nargs="+",
|
||||
default=None,
|
||||
required=param.required,
|
||||
help=param.help,
|
||||
)
|
||||
elif param.kind in ("int", "float"):
|
||||
parser.add_argument(
|
||||
flag,
|
||||
dest=param.name,
|
||||
type=int if param.kind == "int" else float,
|
||||
default=None,
|
||||
required=param.required,
|
||||
help=param.help,
|
||||
)
|
||||
else:
|
||||
parser.add_argument(
|
||||
flag, dest=param.name, default=None, required=param.required, help=param.help
|
||||
)
|
||||
|
||||
|
||||
def _collect(args: argparse.Namespace, params: tuple[P, ...]) -> dict[str, Any]:
|
||||
values: dict[str, Any] = {}
|
||||
for param in params:
|
||||
value = getattr(args, param.name, None)
|
||||
if value is None:
|
||||
continue
|
||||
if param.kind == "json" and isinstance(value, str):
|
||||
try:
|
||||
value = json.loads(value)
|
||||
except ValueError as exc:
|
||||
raise http.CloudError(f"--{param.name.replace('_', '-')} must be JSON") from exc
|
||||
values[param.name] = value
|
||||
return values
|
||||
|
||||
|
||||
def _emit_binary(console: Console, response: Any, output: str | None) -> int:
|
||||
if not response.ok:
|
||||
http.check(response)
|
||||
if output:
|
||||
Path(output).write_bytes(response.content)
|
||||
console.print(f"Saved to [bold]{output}[/]")
|
||||
return http.EXIT_OK
|
||||
sys.stdout.buffer.write(response.content)
|
||||
return http.EXIT_OK
|
||||
|
||||
|
||||
def _emit_error(console: Console, exc: http.CloudError, *, as_json: bool) -> None:
|
||||
if as_json:
|
||||
payload = {"error": str(exc)}
|
||||
if exc.payload is not None:
|
||||
payload["detail"] = exc.payload
|
||||
sys.stdout.write(json.dumps(payload, indent=2, default=str) + "\n")
|
||||
return
|
||||
console.print(f"[red]Error:[/] {exc}")
|
||||
|
||||
|
||||
def _created_id(created: Any) -> str | None:
|
||||
"""Read the identifier of a created item. The API names it `id` or `<resource>_id`."""
|
||||
if not isinstance(created, dict):
|
||||
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):
|
||||
return value
|
||||
return None
|
||||
|
||||
|
||||
def _wait(console: Console, cmd: Cmd, created: Any, *, token: str | None, as_json: bool) -> Any:
|
||||
item_id = _created_id(created)
|
||||
if not item_id or not cmd.wait_path:
|
||||
return created
|
||||
path = cmd.wait_path.replace("{id}", str(item_id))
|
||||
if not as_json:
|
||||
console.print(f"[dim]Waiting for {item_id} to reach a final state…[/]")
|
||||
return _poll(console, path, token=token, as_json=as_json)
|
||||
|
||||
|
||||
def _poll(console: Console, path: str, *, token: str | None, as_json: bool) -> Any:
|
||||
"""Poll a GET path until its status is final. Returns the last response."""
|
||||
while True:
|
||||
time.sleep(_WAIT_POLL_S)
|
||||
current = http.check(http.request("GET", path, token=token))
|
||||
status = str(current.get("status", "")) if isinstance(current, dict) else ""
|
||||
if status.lower() in _TERMINAL_STATUSES:
|
||||
return current
|
||||
if not as_json:
|
||||
console.print(f"[dim] status: {status or 'unknown'}[/]")
|
||||
|
||||
|
||||
def _topup(
|
||||
console: Console,
|
||||
args: argparse.Namespace,
|
||||
body: dict[str, Any],
|
||||
*,
|
||||
as_json: bool,
|
||||
token: str | None,
|
||||
) -> int:
|
||||
response = http.request("POST", "/billing/topup", token=token, body=body)
|
||||
if response.status_code != 402:
|
||||
emit(console, http.check(response), as_json=as_json)
|
||||
return http.EXIT_OK
|
||||
|
||||
challenge = http.parsed(response)
|
||||
if getattr(args, "no_pay", False):
|
||||
emit(console, challenge, as_json=as_json)
|
||||
return http.EXIT_PAYMENT
|
||||
|
||||
npx = shutil.which("npx")
|
||||
if npx is None:
|
||||
emit(console, challenge, as_json=as_json)
|
||||
console.print(
|
||||
"[yellow]Payment required.[/] Install Node.js and run the command again, "
|
||||
"or pay the challenge above with an MPP wallet client."
|
||||
)
|
||||
return http.EXIT_PAYMENT
|
||||
|
||||
credit_count = body.get("credits")
|
||||
if not getattr(args, "yes", False) and sys.stdin.isatty():
|
||||
answer = console.input(f"Buy {credit_count} credit(s) now? [y/N]: ").strip().lower()
|
||||
if answer not in ("y", "yes"):
|
||||
console.print("[yellow]Payment cancelled.[/]")
|
||||
return http.EXIT_PAYMENT
|
||||
|
||||
url = f"{http.app_url()}/api/v1/billing/topup"
|
||||
auth_header = f"Authorization: Bearer {http.api_token(token)}"
|
||||
result = subprocess.run( # noqa: S603
|
||||
[npx, "--yes", "mppx", url, "-J", json.dumps(body), "-H", auth_header],
|
||||
check=False,
|
||||
)
|
||||
return http.EXIT_OK if result.returncode == 0 else http.EXIT_PAYMENT
|
||||
890
strix/interface/cloud/spec.py
Normal file
890
strix/interface/cloud/spec.py
Normal file
|
|
@ -0,0 +1,890 @@
|
|||
"""Declarative command table for `strix cloud`.
|
||||
|
||||
Each command maps one CLI verb to one managed API operation. The runner
|
||||
builds the argument parser and the HTTP request from this table, so the
|
||||
CLI surface stays aligned with the OpenAPI specification.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class P:
|
||||
"""One command parameter.
|
||||
|
||||
``kind`` is one of ``str``, ``int``, ``float``, ``bool``, ``list``, or ``json``.
|
||||
"""
|
||||
|
||||
name: str
|
||||
kind: str = "str"
|
||||
required: bool = False
|
||||
help: str = ""
|
||||
# Command-line name when the field name collides with a common option.
|
||||
flag: str | None = None
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Cmd:
|
||||
method: str
|
||||
path: str
|
||||
help: str
|
||||
query: tuple[P, ...] = ()
|
||||
body: tuple[P, ...] = ()
|
||||
binary: bool = False
|
||||
wait_path: str | None = None
|
||||
# When true, `--wait` polls GET on this same path until the status is final.
|
||||
wait_self: bool = False
|
||||
|
||||
|
||||
def _q(*names: str) -> tuple[P, ...]:
|
||||
return tuple(P(name) for name in names)
|
||||
|
||||
|
||||
_SCAN_START_BODY = (
|
||||
P("engagement_type", help="Test category, for example live_test or code_review."),
|
||||
P("domain_ids", "list", help="Domain asset IDs to test."),
|
||||
P("domain_paths", "json", help="JSON map of domain ID to start paths."),
|
||||
P("repository_ids", "list", help="Repository asset IDs to test."),
|
||||
P("repository_branches", "json", help="JSON map of repository ID to branch."),
|
||||
P("credentials", "json", help="JSON list of credential objects."),
|
||||
P("headers", "json", help="JSON map of extra HTTP headers for the target."),
|
||||
P("concerns", "list", help="Vulnerability classes to focus on."),
|
||||
P("focus", help="Free-form focus instructions for the agents."),
|
||||
P("context", help="Extra context about the target."),
|
||||
P("upload_ids", "list", help="Upload IDs to attach to the scan."),
|
||||
P("connector_id", help="Network connector ID for internal targets."),
|
||||
P("internal_targets", "list", help="Internal IP addresses or ranges."),
|
||||
P("org_knowledge_enabled", "bool", help="Use the organization knowledge base."),
|
||||
P("notify_on_completion", "bool", help="Send an email when the scan completes."),
|
||||
P("notification_emails", "list", help="Extra notification email addresses."),
|
||||
P("scan_tier", help="Scan tier, for example lite, pro, or max."),
|
||||
P("model_config_id", help="Model configuration ID to run with."),
|
||||
P("max_budget_usd", "float", help="Budget limit for the scan in USD."),
|
||||
)
|
||||
|
||||
_TEST_USER_ADD_BODY = (
|
||||
P("label", required=True, help="Display label for the test user."),
|
||||
P("username", required=True, help="Sign-in username or email address."),
|
||||
P("password", help="Sign-in password."),
|
||||
P("notes", help="Free-form notes for the agents."),
|
||||
P("login_url", help="URL of the sign-in page."),
|
||||
P("mfa_method", help="MFA method, for example totp or email."),
|
||||
P("totp_secret", help="TOTP secret for MFA sign-in."),
|
||||
P("mfa_email", help="Email address that receives MFA codes."),
|
||||
P("scope_domain_ids", "list", help="Domain IDs where this user applies."),
|
||||
)
|
||||
|
||||
_TEST_USER_BODY = (
|
||||
P("label", help="Display label for the test user."),
|
||||
P("username", help="Sign-in username or email address."),
|
||||
P("password", help="Sign-in password."),
|
||||
P("notes", help="Free-form notes for the agents."),
|
||||
P("login_url", help="URL of the sign-in page."),
|
||||
P("mfa_method", help="MFA method, for example totp or email."),
|
||||
P("totp_secret", help="TOTP secret for MFA sign-in."),
|
||||
P("mfa_email", help="Email address that receives MFA codes."),
|
||||
P("scope_domain_ids", "list", help="Domain IDs where this user applies."),
|
||||
)
|
||||
|
||||
_GIT_TOKEN_BODY = (
|
||||
P("token", required=True, help="Provider access token.", flag="provider-token"),
|
||||
P("instance_url", help="Base URL for a self-hosted instance."),
|
||||
P("workspace", help="Bitbucket workspace name."),
|
||||
)
|
||||
|
||||
|
||||
SPEC: dict[str, dict[str, Cmd]] = {
|
||||
"scans": {
|
||||
"list": Cmd(
|
||||
"GET",
|
||||
"/scans",
|
||||
"List scans.",
|
||||
query=_q(
|
||||
"status",
|
||||
"scan_type",
|
||||
"date_from",
|
||||
"date_to",
|
||||
"domain_id",
|
||||
"repository_id",
|
||||
"search",
|
||||
"include_retests",
|
||||
),
|
||||
),
|
||||
"start": Cmd(
|
||||
"POST",
|
||||
"/scans",
|
||||
"Start a scan.",
|
||||
body=_SCAN_START_BODY,
|
||||
wait_path="/scans/{id}",
|
||||
),
|
||||
"get": Cmd("GET", "/scans/{scanId}", "Get one scan."),
|
||||
"delete": Cmd("DELETE", "/scans/{scanId}", "Delete a scan."),
|
||||
"agents": Cmd("GET", "/scans/{scanId}/agents", "List the agents of a scan."),
|
||||
"cancel": Cmd("POST", "/scans/{scanId}/cancel", "Cancel a running scan."),
|
||||
"message": Cmd(
|
||||
"POST",
|
||||
"/scans/{scanId}/message",
|
||||
"Send a message to the scan agents.",
|
||||
body=(
|
||||
P("message", required=True, help="Message text for the agents."),
|
||||
P("cancel_current", "bool", help="Stop the current task first."),
|
||||
P("agent_id", help="Target one agent instead of the root agent."),
|
||||
),
|
||||
),
|
||||
"report": Cmd(
|
||||
"GET",
|
||||
"/scans/{scanId}/report",
|
||||
"Download the scan report.",
|
||||
query=_q("format", "type"),
|
||||
binary=True,
|
||||
),
|
||||
"rerun": Cmd("POST", "/scans/{scanId}/rerun", "Run the scan again.", wait_path=None),
|
||||
"retest-all": Cmd(
|
||||
"POST",
|
||||
"/scans/{scanId}/retest-all",
|
||||
"Retest all open findings of a scan.",
|
||||
body=(
|
||||
P("scope", help="Retest scope."),
|
||||
P("upload_ids", "list", help="Upload IDs with updated code."),
|
||||
),
|
||||
),
|
||||
"retests": Cmd("GET", "/scans/{scanId}/retests", "List the retests of a scan."),
|
||||
"sarif": Cmd(
|
||||
"GET",
|
||||
"/scans/{scanId}/sarif",
|
||||
"Download the scan findings as SARIF.",
|
||||
query=_q("repository"),
|
||||
binary=True,
|
||||
),
|
||||
"sarif-upload": Cmd(
|
||||
"POST",
|
||||
"/scans/{scanId}/sarif",
|
||||
"Upload the scan findings to GitHub code scanning.",
|
||||
body=(
|
||||
P("repository", help="Repository full name."),
|
||||
P("ref", help="Git ref for the upload."),
|
||||
P("commit_sha", help="Commit SHA for the upload."),
|
||||
P("checkout_uri", help="Checkout URI for the upload."),
|
||||
P("github_api_base_url", help="GitHub API base URL."),
|
||||
),
|
||||
),
|
||||
"template": Cmd("GET", "/scans/{scanId}/template", "Get the scan configuration template."),
|
||||
"trace": Cmd(
|
||||
"GET",
|
||||
"/scans/{scanId}/trace",
|
||||
"List trace events for one agent of a scan.",
|
||||
query=(
|
||||
P("agent_id", required=True, help="Agent ID to read the trace for."),
|
||||
P("cursor"),
|
||||
P("limit", "int"),
|
||||
P("tool_name"),
|
||||
),
|
||||
),
|
||||
"trace-event": Cmd(
|
||||
"GET", "/scans/{scanId}/trace/{eventId}", "Get one trace event of a scan."
|
||||
),
|
||||
},
|
||||
"vulns": {
|
||||
"list": Cmd(
|
||||
"GET",
|
||||
"/vulnerabilities",
|
||||
"List vulnerabilities.",
|
||||
query=_q(
|
||||
"scan_id",
|
||||
"severity",
|
||||
"status",
|
||||
"search",
|
||||
"from",
|
||||
"to",
|
||||
"domain_id",
|
||||
"repository_id",
|
||||
"finding_type",
|
||||
"dependency_relation",
|
||||
"reachability",
|
||||
"sort_by",
|
||||
),
|
||||
),
|
||||
"get": Cmd("GET", "/vulnerabilities/{vulnerabilityId}", "Get one vulnerability."),
|
||||
"history": Cmd(
|
||||
"GET",
|
||||
"/vulnerabilities/{vulnerabilityId}/history",
|
||||
"Get the change history of a vulnerability.",
|
||||
),
|
||||
"update": Cmd(
|
||||
"PATCH",
|
||||
"/vulnerabilities/{vulnerabilityId}",
|
||||
"Update the status or severity of a vulnerability.",
|
||||
body=(
|
||||
P("status", help="New status, for example triaged or false_positive."),
|
||||
P("note", help="Note that explains the change."),
|
||||
P("severity", help="New severity."),
|
||||
P("severity_reason", help="Reason for the severity change."),
|
||||
),
|
||||
),
|
||||
"retest": Cmd(
|
||||
"POST",
|
||||
"/vulnerabilities/{vulnerabilityId}/retest",
|
||||
"Retest one vulnerability.",
|
||||
body=(P("upload_ids", "list", help="Upload IDs with updated code."),),
|
||||
),
|
||||
"fix-pr": Cmd(
|
||||
"POST",
|
||||
"/vulnerabilities/{vulnerabilityId}/create-fix-pr",
|
||||
"Create a fix pull request for a vulnerability.",
|
||||
),
|
||||
"push": Cmd(
|
||||
"POST",
|
||||
"/vulnerabilities/{vulnerabilityId}/push",
|
||||
"Push one vulnerability to an issue tracker.",
|
||||
body=(
|
||||
P("provider", required=True, help="Tracker provider, for example jira or linear."),
|
||||
P("collection_id", help="Tracker project or collection ID."),
|
||||
),
|
||||
),
|
||||
"push-bulk": Cmd(
|
||||
"POST",
|
||||
"/vulnerabilities/bulk-push",
|
||||
"Push many vulnerabilities to an issue tracker.",
|
||||
body=(
|
||||
P("provider", required=True, help="Tracker provider, for example jira or linear."),
|
||||
P("vulnerability_ids", "list", required=True, help="Vulnerability IDs to push."),
|
||||
P("collection_id", help="Tracker project or collection ID."),
|
||||
),
|
||||
),
|
||||
},
|
||||
"domains": {
|
||||
"list": Cmd(
|
||||
"GET",
|
||||
"/domains",
|
||||
"List domain assets.",
|
||||
query=_q("limit", "search", "verified", "business_unit", "tags", "sort_by"),
|
||||
),
|
||||
"add": Cmd(
|
||||
"POST",
|
||||
"/domains",
|
||||
"Add a domain asset.",
|
||||
body=(
|
||||
P("domain", required=True, help="Domain name or URL."),
|
||||
P("asset_type", required=True, help="Asset type, for example web_app or api."),
|
||||
P("context", help="Extra context about the asset."),
|
||||
P("tags", "list", help="Tags for the asset."),
|
||||
P("business_unit", help="Business unit for the asset."),
|
||||
),
|
||||
),
|
||||
"update": Cmd(
|
||||
"PATCH",
|
||||
"/domains/{domainId}",
|
||||
"Update a domain asset.",
|
||||
body=(
|
||||
P("context", help="Extra context about the asset."),
|
||||
P("tags", "list", help="Tags for the asset."),
|
||||
P("business_unit", help="Business unit for the asset."),
|
||||
),
|
||||
),
|
||||
"remove": Cmd("DELETE", "/domains/{domainId}", "Remove a domain asset."),
|
||||
"verify": Cmd("POST", "/domains/{domainId}/verify", "Verify domain ownership."),
|
||||
"auto-verify": Cmd(
|
||||
"POST",
|
||||
"/domains/{domainId}/auto-verify",
|
||||
"Verify domain ownership through a DNS provider.",
|
||||
body=(P("provider", required=True, help="DNS provider name."),),
|
||||
),
|
||||
"test-users list": Cmd(
|
||||
"GET", "/domains/{domainId}/test-users", "List the test users of a domain."
|
||||
),
|
||||
"test-users add": Cmd(
|
||||
"POST",
|
||||
"/domains/{domainId}/test-users",
|
||||
"Add a test user to a domain.",
|
||||
body=_TEST_USER_ADD_BODY,
|
||||
),
|
||||
"test-users update": Cmd(
|
||||
"PATCH",
|
||||
"/domains/{domainId}/test-users/{userId}",
|
||||
"Update a test user.",
|
||||
body=_TEST_USER_BODY,
|
||||
),
|
||||
"test-users remove": Cmd(
|
||||
"DELETE", "/domains/{domainId}/test-users/{userId}", "Remove a test user."
|
||||
),
|
||||
"test-users provision-inbox": Cmd(
|
||||
"POST",
|
||||
"/domains/{domainId}/test-users/provision-inbox",
|
||||
"Create a test user with a managed email inbox.",
|
||||
body=(P("label", help="Display label for the test user."),),
|
||||
),
|
||||
"test-users inbox": Cmd(
|
||||
"GET",
|
||||
"/domains/{domainId}/test-users/{userId}/inbox",
|
||||
"List the inbox messages of a test user.",
|
||||
query=(P("limit", "int"),),
|
||||
),
|
||||
"test-users inbox-message": Cmd(
|
||||
"GET",
|
||||
"/domains/{domainId}/test-users/{userId}/inbox/{messageId}",
|
||||
"Get one inbox message of a test user.",
|
||||
),
|
||||
"test-users verify": Cmd(
|
||||
"POST",
|
||||
"/domains/{domainId}/test-users/{userId}/verify",
|
||||
"Verify that the test user credentials work.",
|
||||
query=(P("force"),),
|
||||
wait_self=True,
|
||||
),
|
||||
"test-users verify-status": Cmd(
|
||||
"GET",
|
||||
"/domains/{domainId}/test-users/{userId}/verify",
|
||||
"Get the verification status of a test user.",
|
||||
),
|
||||
},
|
||||
"repos": {
|
||||
"list": Cmd(
|
||||
"GET",
|
||||
"/repositories",
|
||||
"List repository assets.",
|
||||
query=_q("limit", "search", "business_unit", "tags", "sort_by"),
|
||||
),
|
||||
"add": Cmd("POST", "/repositories", "Add a repository asset. Use --data for the fields."),
|
||||
"update": Cmd(
|
||||
"PATCH",
|
||||
"/repositories/{repositoryId}",
|
||||
"Update a repository asset.",
|
||||
body=(
|
||||
P("pr_review_enabled", "bool", help="Turn PR reviews on or off."),
|
||||
P("pr_review_approvals_enabled", "bool", help="Let reviews approve clean PRs."),
|
||||
P("pr_review_non_blocking", "bool", help="Make review verdicts non-blocking."),
|
||||
P("pr_review_on_push", "bool", help="Review new pushes to open PRs."),
|
||||
P("tags", "list", help="Tags for the asset."),
|
||||
P("business_unit", help="Business unit for the asset."),
|
||||
),
|
||||
),
|
||||
"remove": Cmd("DELETE", "/repositories/{repositoryId}", "Remove a repository asset."),
|
||||
"supply-chain scan": Cmd(
|
||||
"POST",
|
||||
"/repositories/{repositoryId}/supply-chain/scan",
|
||||
"Start a supply-chain scan for a repository.",
|
||||
),
|
||||
"supply-chain summary": Cmd(
|
||||
"GET",
|
||||
"/repositories/{repositoryId}/supply-chain/summary",
|
||||
"Get the supply-chain summary of a repository.",
|
||||
query=_q("job_id", "snapshot_id"),
|
||||
),
|
||||
"supply-chain findings": Cmd(
|
||||
"GET",
|
||||
"/repositories/{repositoryId}/supply-chain/findings",
|
||||
"List the supply-chain findings of a repository.",
|
||||
query=_q("job_id", "snapshot_id", "component_id"),
|
||||
),
|
||||
"supply-chain components": Cmd(
|
||||
"GET",
|
||||
"/repositories/{repositoryId}/supply-chain/components",
|
||||
"List the dependency components of a repository.",
|
||||
query=_q(
|
||||
"job_id",
|
||||
"snapshot_id",
|
||||
"component_id",
|
||||
"ecosystem",
|
||||
"status",
|
||||
"relationship",
|
||||
"source_file",
|
||||
"q",
|
||||
"changed",
|
||||
"limit",
|
||||
"offset",
|
||||
),
|
||||
),
|
||||
"supply-chain sbom": Cmd(
|
||||
"GET",
|
||||
"/repositories/{repositoryId}/supply-chain/sbom",
|
||||
"Download the SBOM of a repository.",
|
||||
query=_q("job_id", "snapshot_id", "format"),
|
||||
binary=True,
|
||||
),
|
||||
"supply-chain policy": Cmd(
|
||||
"PATCH",
|
||||
"/repositories/{repositoryId}/supply-chain/policy",
|
||||
"Update the supply-chain policy of a repository.",
|
||||
body=(
|
||||
P("supply_chain_enabled", "bool", help="Turn supply-chain scans on or off."),
|
||||
P("supply_chain_pr_checks_enabled", "bool", help="Run checks on pull requests."),
|
||||
P("supply_chain_policy_mode", help="Policy mode for new findings."),
|
||||
),
|
||||
),
|
||||
},
|
||||
"supply-chain": {
|
||||
"summary": Cmd(
|
||||
"GET", "/supply-chain/summary", "Get the organization supply-chain summary."
|
||||
),
|
||||
},
|
||||
"schedules": {
|
||||
"list": Cmd("GET", "/schedules", "List scan schedules."),
|
||||
"create": Cmd("POST", "/schedules", "Create a scan schedule. Use --data for the fields."),
|
||||
"get": Cmd("GET", "/schedules/{scheduleId}", "Get one schedule."),
|
||||
"update": Cmd(
|
||||
"PATCH",
|
||||
"/schedules/{scheduleId}",
|
||||
"Update a schedule. Use --data for fields that have no option.",
|
||||
body=(
|
||||
P("action", help="Lifecycle action, for example pause or resume."),
|
||||
P("cron_expression", help="Cron expression for the schedule."),
|
||||
P("timezone", help="Time zone for the cron expression."),
|
||||
P("name", help="Display name of the schedule."),
|
||||
P("max_budget_usd", "int", help="Budget limit per run in USD."),
|
||||
P("scan_tier", help="Scan tier for scheduled runs."),
|
||||
),
|
||||
),
|
||||
"delete": Cmd("DELETE", "/schedules/{scheduleId}", "Delete a schedule."),
|
||||
"template": Cmd(
|
||||
"GET", "/schedules/{scheduleId}/template", "Get the schedule configuration template."
|
||||
),
|
||||
"trigger": Cmd("POST", "/schedules/{scheduleId}/trigger", "Run a schedule now."),
|
||||
},
|
||||
"pr-reviews": {
|
||||
"list": Cmd(
|
||||
"GET",
|
||||
"/pr-reviews",
|
||||
"List PR reviews.",
|
||||
query=_q(
|
||||
"search",
|
||||
"status",
|
||||
"group",
|
||||
"pr_state",
|
||||
"repository_full_name",
|
||||
"date_from",
|
||||
"date_to",
|
||||
"sort_by",
|
||||
"sort_order",
|
||||
"include_counts",
|
||||
),
|
||||
),
|
||||
"get": Cmd("GET", "/pr-reviews/{prReviewId}", "Get one PR review."),
|
||||
"findings": Cmd(
|
||||
"GET",
|
||||
"/pr-reviews/findings",
|
||||
"List PR review findings.",
|
||||
query=_q("severity", "pr_state", "search", "repository_full_name", "include_stats"),
|
||||
),
|
||||
"start": Cmd(
|
||||
"POST",
|
||||
"/pr-reviews/start",
|
||||
"Start a PR review.",
|
||||
body=(
|
||||
P("repository_full_name", required=True, help="Repository full name."),
|
||||
P("pr_number", "int", required=True, help="Pull request number."),
|
||||
),
|
||||
),
|
||||
"settings": Cmd("GET", "/pr-reviews/settings", "Get the PR review settings."),
|
||||
"settings update": Cmd(
|
||||
"PATCH",
|
||||
"/pr-reviews/settings",
|
||||
"Update the PR review settings. Use --data for fields that have no option.",
|
||||
body=(
|
||||
P("review_on_push", "bool", help="Review new pushes to open PRs."),
|
||||
P("block_on_findings", "bool", help="Block PRs that have findings."),
|
||||
P("blocking_severities", "list", help="Severities that block a PR."),
|
||||
P("approve_clean_prs", "bool", help="Approve PRs without findings."),
|
||||
P("target_branches", "list", help="Branches that get reviews."),
|
||||
),
|
||||
),
|
||||
},
|
||||
"billing": {
|
||||
"credits": Cmd("GET", "/billing/credits", "Get the credit balance of the workspace."),
|
||||
"topup": Cmd(
|
||||
"POST",
|
||||
"/billing/topup",
|
||||
"Buy credits with an agent payment (HTTP 402 flow).",
|
||||
body=(P("credits", "int", required=True, help="Number of credits to buy."),),
|
||||
),
|
||||
"auto-topup": Cmd("GET", "/billing/auto-topup", "Get the automatic top-up settings."),
|
||||
"auto-topup update": Cmd(
|
||||
"PUT",
|
||||
"/billing/auto-topup",
|
||||
"Update the automatic top-up settings.",
|
||||
body=(
|
||||
P("enabled", "bool", required=True, help="Turn automatic top-up on or off."),
|
||||
P("topup_credits", "int", required=True, help="Credits to buy on each top-up."),
|
||||
P("monthly_cap_credits", "int", help="Monthly credit cap for automatic top-ups."),
|
||||
),
|
||||
),
|
||||
},
|
||||
"chat": {
|
||||
"list": Cmd("GET", "/chat", "List chat sessions."),
|
||||
"start": Cmd(
|
||||
"POST",
|
||||
"/chat",
|
||||
"Start a chat session.",
|
||||
body=(
|
||||
P("message", required=True, help="First message of the session."),
|
||||
P("repos", "list", help="Repository full names for context."),
|
||||
P("domain_ids", "list", help="Domain asset IDs for context."),
|
||||
),
|
||||
),
|
||||
"get": Cmd("GET", "/chat/{chatId}", "Get one chat session."),
|
||||
"send": Cmd(
|
||||
"POST",
|
||||
"/chat/{chatId}/message",
|
||||
"Send a message in a chat session.",
|
||||
body=(
|
||||
P("message", required=True, help="Message text."),
|
||||
P("cancel_current", "bool", help="Stop the current task first."),
|
||||
P("stop_agent", "bool", help="Stop the agent."),
|
||||
P("repos", "list", help="Repository full names for context."),
|
||||
P("agent_id", help="Target one agent."),
|
||||
),
|
||||
),
|
||||
"findings": Cmd("GET", "/chat/{chatId}/findings", "List the findings of a chat session."),
|
||||
"finding": Cmd(
|
||||
"GET", "/chat/{chatId}/findings/{findingId}", "Get one finding of a chat session."
|
||||
),
|
||||
"finding file": Cmd(
|
||||
"POST",
|
||||
"/chat/{chatId}/findings/{findingId}/file",
|
||||
"File a chat finding into the organization issue list.",
|
||||
),
|
||||
"files": Cmd("GET", "/chat/{chatId}/files", "List the files of a chat session."),
|
||||
"files download": Cmd(
|
||||
"GET",
|
||||
"/chat/{chatId}/files/download",
|
||||
"Download one file of a chat session.",
|
||||
query=(P("path", required=True, help="File path inside the session."),),
|
||||
binary=True,
|
||||
),
|
||||
"files archive": Cmd(
|
||||
"GET",
|
||||
"/chat/{chatId}/files/archive",
|
||||
"Download all files of a chat session as an archive.",
|
||||
binary=True,
|
||||
),
|
||||
"credentials": Cmd(
|
||||
"GET",
|
||||
"/chat/{chatId}/credentials",
|
||||
"Get the credentials of a chat session.",
|
||||
query=_q("scan_ids"),
|
||||
),
|
||||
"credentials set": Cmd(
|
||||
"POST",
|
||||
"/chat/{chatId}/credentials",
|
||||
"Set the credentials of a chat session.",
|
||||
body=(
|
||||
P("test_user_ids", "list", help="Test user IDs to attach."),
|
||||
P("credentials", "json", help="JSON list of credential objects."),
|
||||
P("scan_ids", "list", help="Scan IDs that use the credentials."),
|
||||
),
|
||||
),
|
||||
"credentials clear": Cmd(
|
||||
"DELETE", "/chat/{chatId}/credentials", "Remove the credentials of a chat session."
|
||||
),
|
||||
"domains set": Cmd(
|
||||
"PUT",
|
||||
"/chat/{chatId}/domains",
|
||||
"Set the domains of a chat session.",
|
||||
body=(P("domain_ids", "list", help="Domain asset IDs."),),
|
||||
),
|
||||
"terminal": Cmd(
|
||||
"POST",
|
||||
"/chat/{chatId}/terminal",
|
||||
"Run a command in the chat session sandbox.",
|
||||
body=(
|
||||
P("command", required=True, help="Shell command to run."),
|
||||
P("cwd", help="Working directory for the command."),
|
||||
),
|
||||
),
|
||||
"share": Cmd("POST", "/chat/{chatId}/share", "Create a share link for a chat session."),
|
||||
},
|
||||
"knowledge": {
|
||||
"list": Cmd(
|
||||
"GET",
|
||||
"/knowledge",
|
||||
"List knowledge documents.",
|
||||
query=_q("source_type", "search", "limit"),
|
||||
),
|
||||
"add": Cmd(
|
||||
"POST",
|
||||
"/knowledge",
|
||||
"Add a knowledge document.",
|
||||
body=(
|
||||
P("title", required=True, help="Document title."),
|
||||
P("content", required=True, help="Document content."),
|
||||
P("tags", "list", help="Tags for the document."),
|
||||
P("metadata", "json", help="JSON metadata for the document."),
|
||||
),
|
||||
),
|
||||
"update": Cmd(
|
||||
"PATCH",
|
||||
"/knowledge/{documentId}",
|
||||
"Update a knowledge document.",
|
||||
body=(
|
||||
P("title", help="Document title."),
|
||||
P("content", help="Document content."),
|
||||
P("tags", "list", help="Tags for the document."),
|
||||
P("metadata", "json", help="JSON metadata for the document."),
|
||||
),
|
||||
),
|
||||
"delete": Cmd("DELETE", "/knowledge/{documentId}", "Delete a knowledge document."),
|
||||
"query": Cmd(
|
||||
"GET",
|
||||
"/knowledge/query",
|
||||
"Query the knowledge base.",
|
||||
query=(P("q", help="Query text."), P("limit", "int")),
|
||||
),
|
||||
"policies": Cmd("GET", "/knowledge/policies", "List knowledge policies."),
|
||||
"policies add": Cmd(
|
||||
"POST",
|
||||
"/knowledge/policies",
|
||||
"Add a knowledge policy.",
|
||||
body=(
|
||||
P("key", required=True, help="Policy key."),
|
||||
P("content", help="Policy content."),
|
||||
P("enabled", "bool", help="Turn the policy on or off."),
|
||||
),
|
||||
),
|
||||
"policies delete": Cmd(
|
||||
"DELETE", "/knowledge/policies/{policyKey}", "Delete a knowledge policy."
|
||||
),
|
||||
"repos": Cmd("GET", "/knowledge/repos", "List repositories with knowledge entries."),
|
||||
"repos entries": Cmd(
|
||||
"GET", "/knowledge/repos/{repo}/entries", "List the knowledge entries of a repository."
|
||||
),
|
||||
"repos profile": Cmd(
|
||||
"PATCH",
|
||||
"/knowledge/repos/{repo}/profile",
|
||||
"Update the knowledge profile of a repository. Use --data for the fields.",
|
||||
),
|
||||
"settings": Cmd("GET", "/knowledge/settings", "Get the knowledge settings."),
|
||||
"settings update": Cmd(
|
||||
"PATCH",
|
||||
"/knowledge/settings",
|
||||
"Update the knowledge settings.",
|
||||
body=(P("org_knowledge_enabled", "bool", help="Use organization knowledge in scans."),),
|
||||
),
|
||||
},
|
||||
"org": {
|
||||
"get": Cmd("GET", "/organization", "Get the organization."),
|
||||
"update": Cmd(
|
||||
"PATCH",
|
||||
"/organization",
|
||||
"Update the organization.",
|
||||
body=(P("name", required=True, help="Organization name."),),
|
||||
),
|
||||
"members": Cmd("GET", "/organization/members", "List the organization members."),
|
||||
"members invite": Cmd(
|
||||
"POST",
|
||||
"/organization/members",
|
||||
"Invite a member to the organization.",
|
||||
body=(
|
||||
P("email", required=True, help="Email address of the new member."),
|
||||
P("role", help="Member role, for example admin, analyst, or viewer."),
|
||||
P("scopes", "list", help="RBAC scopes for the member."),
|
||||
),
|
||||
),
|
||||
"members update": Cmd(
|
||||
"PATCH",
|
||||
"/organization/members/{membershipId}",
|
||||
"Update a member of the organization.",
|
||||
body=(
|
||||
P("role", required=True, help="Member role."),
|
||||
P("scopes", "list", help="RBAC scopes for the member."),
|
||||
),
|
||||
),
|
||||
"members remove": Cmd("DELETE", "/organization/members/{membershipId}", "Remove a member."),
|
||||
"invitations": Cmd("GET", "/organization/invitations", "List open invitations."),
|
||||
"invitations revoke": Cmd(
|
||||
"DELETE", "/organization/invitations/{invitationId}", "Revoke an invitation."
|
||||
),
|
||||
},
|
||||
"integrations": {
|
||||
"list": Cmd("GET", "/integrations", "List the connected integrations."),
|
||||
"connect": Cmd(
|
||||
"POST",
|
||||
"/integrations/{provider}/connect",
|
||||
"Connect a Git provider. The provider is gitlab or bitbucket.",
|
||||
body=_GIT_TOKEN_BODY,
|
||||
),
|
||||
"validate": Cmd(
|
||||
"POST",
|
||||
"/integrations/{provider}/validate",
|
||||
"Validate a Git provider token. The provider is gitlab or bitbucket.",
|
||||
body=_GIT_TOKEN_BODY,
|
||||
),
|
||||
"disconnect": Cmd("DELETE", "/integrations/{provider}", "Disconnect an integration."),
|
||||
},
|
||||
"connectors": {
|
||||
"list": Cmd("GET", "/connectors", "List network connectors."),
|
||||
"create": Cmd(
|
||||
"POST",
|
||||
"/connectors",
|
||||
"Create a network connector.",
|
||||
body=(P("name", required=True, help="Connector name."),),
|
||||
),
|
||||
"get": Cmd(
|
||||
"GET",
|
||||
"/connectors/{connectorId}",
|
||||
"Get one network connector.",
|
||||
query=_q("include_command"),
|
||||
),
|
||||
"status": Cmd(
|
||||
"GET", "/connectors/{connectorId}/status", "Get the status of a network connector."
|
||||
),
|
||||
"delete": Cmd("DELETE", "/connectors/{connectorId}", "Delete a network connector."),
|
||||
},
|
||||
"webhooks": {
|
||||
"list": Cmd("GET", "/webhooks", "List webhooks."),
|
||||
"create": Cmd(
|
||||
"POST",
|
||||
"/webhooks",
|
||||
"Create a webhook.",
|
||||
body=(
|
||||
P("url", required=True, help="Delivery URL."),
|
||||
P("events", "list", required=True, help="Event names to deliver."),
|
||||
P("business_unit", help="Business unit filter."),
|
||||
P("is_active", "bool", help="Turn the webhook on or off."),
|
||||
),
|
||||
),
|
||||
"get": Cmd("GET", "/webhooks/{webhookId}", "Get one webhook."),
|
||||
"update": Cmd(
|
||||
"PATCH",
|
||||
"/webhooks/{webhookId}",
|
||||
"Update a webhook.",
|
||||
body=(
|
||||
P("url", help="Delivery URL."),
|
||||
P("events", "list", help="Event names to deliver."),
|
||||
P("business_unit", help="Business unit filter."),
|
||||
P("is_active", "bool", help="Turn the webhook on or off."),
|
||||
P("rotate_secret", "bool", help="Create a new signing secret."),
|
||||
),
|
||||
),
|
||||
"delete": Cmd("DELETE", "/webhooks/{webhookId}", "Delete a webhook."),
|
||||
"deliveries": Cmd(
|
||||
"GET", "/webhooks/{webhookId}/deliveries", "List the deliveries of a webhook."
|
||||
),
|
||||
},
|
||||
"analytics": {
|
||||
"overview": Cmd(
|
||||
"GET",
|
||||
"/analytics/overview",
|
||||
"Get the analytics overview.",
|
||||
query=_q("range", "from", "to"),
|
||||
),
|
||||
"stats": Cmd("GET", "/analytics/stats", "Get the analytics statistics."),
|
||||
"scan-frequency": Cmd(
|
||||
"GET", "/analytics/scan-frequency", "Get the scan frequency data.", query=_q("tz")
|
||||
),
|
||||
},
|
||||
"audit": {
|
||||
"list": Cmd(
|
||||
"GET",
|
||||
"/audit",
|
||||
"List audit log entries.",
|
||||
query=_q(
|
||||
"action", "resource_type", "actor_id", "date_from", "date_to", "format", "all"
|
||||
),
|
||||
),
|
||||
},
|
||||
"costs": {
|
||||
"overview": Cmd(
|
||||
"GET", "/llm-costs", "Show the LLM cost overview.", query=_q("range", "from", "to")
|
||||
),
|
||||
"run": Cmd("GET", "/llm-costs/runs/{runType}/{runId}", "Get the LLM costs of one run."),
|
||||
},
|
||||
"llm-settings": {
|
||||
"get": Cmd("GET", "/llm-settings", "Get the LLM settings."),
|
||||
"update": Cmd(
|
||||
"PUT",
|
||||
"/llm-settings",
|
||||
"Update the LLM settings.",
|
||||
body=(
|
||||
P("modelConfigs", "json", required=True, help="JSON list of model configurations."),
|
||||
P("assignments", "json", required=True, help="JSON map of model assignments."),
|
||||
),
|
||||
),
|
||||
},
|
||||
"settings": {
|
||||
"notifications": Cmd("GET", "/settings/notifications", "Get the notification settings."),
|
||||
"notifications update": Cmd(
|
||||
"PATCH",
|
||||
"/settings/notifications",
|
||||
"Update the notification settings.",
|
||||
body=(
|
||||
P("sla_reminders_enabled", "bool", help="Turn SLA reminders on or off."),
|
||||
P("sla_reminder_email", "bool", help="Send SLA reminders by email."),
|
||||
P("sla_reminder_slack", "bool", help="Send SLA reminders to Slack."),
|
||||
P("sla_warning_days", "int", help="Days before an SLA warning."),
|
||||
),
|
||||
),
|
||||
},
|
||||
"license": {
|
||||
"show": Cmd("GET", "/license", "Get the license information."),
|
||||
},
|
||||
"tokens": {
|
||||
"list": Cmd("GET", "/tokens", "List API tokens.", query=_q("type")),
|
||||
"create": Cmd(
|
||||
"POST",
|
||||
"/tokens",
|
||||
"Create an API token.",
|
||||
body=(
|
||||
P("type", required=True, help="Token type, personal or service."),
|
||||
P("name", required=True, help="Token name."),
|
||||
P("scopes", "list", help="API scopes for the token."),
|
||||
P("expires_in_days", "int", help="Days until the token expires."),
|
||||
),
|
||||
),
|
||||
"revoke": Cmd("DELETE", "/tokens/{tokenId}", "Revoke an API token."),
|
||||
},
|
||||
"uploads": {
|
||||
"request": Cmd(
|
||||
"POST",
|
||||
"/uploads/request",
|
||||
"Request an upload URL.",
|
||||
body=(
|
||||
P("file_name", required=True, help="File name."),
|
||||
P("file_size", "int", required=True, help="File size in bytes."),
|
||||
P("category", help="Upload category."),
|
||||
),
|
||||
),
|
||||
"complete": Cmd(
|
||||
"POST",
|
||||
"/uploads/complete",
|
||||
"Mark an upload as complete.",
|
||||
body=(P("upload_id", required=True, help="Upload ID."),),
|
||||
),
|
||||
"delete": Cmd("DELETE", "/uploads/{uploadId}", "Delete an upload."),
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
# Default verbs let a bare group name run its most common read command.
|
||||
DEFAULT_VERBS: dict[str, str] = {
|
||||
"costs": "overview",
|
||||
"audit": "list",
|
||||
"license": "show",
|
||||
"supply-chain": "summary",
|
||||
}
|
||||
|
||||
|
||||
GROUP_HELP: dict[str, str] = {
|
||||
"scans": "Start, watch, and manage scans",
|
||||
"vulns": "Triage and remediate vulnerabilities",
|
||||
"domains": "Manage domain assets and test users",
|
||||
"repos": "Manage repository assets and supply-chain scans",
|
||||
"supply-chain": "Organization supply-chain summary",
|
||||
"schedules": "Manage scan schedules",
|
||||
"pr-reviews": "Manage pull request reviews",
|
||||
"billing": "Credits, top-ups, and automatic top-up",
|
||||
"chat": "Interactive pentest chat sessions",
|
||||
"knowledge": "Manage the knowledge base",
|
||||
"org": "Manage the organization and its members",
|
||||
"integrations": "Connect Git providers",
|
||||
"connectors": "Manage network connectors",
|
||||
"webhooks": "Manage webhooks",
|
||||
"analytics": "Read analytics data",
|
||||
"audit": "Read the audit log",
|
||||
"costs": "Read LLM cost data",
|
||||
"llm-settings": "Manage LLM model settings",
|
||||
"settings": "Manage notification settings",
|
||||
"license": "Read license information",
|
||||
"tokens": "Manage API tokens",
|
||||
"uploads": "Upload files for scans",
|
||||
}
|
||||
|
|
@ -431,12 +431,12 @@ def main() -> None:
|
|||
|
||||
sys.exit(run_auth(sys.argv[2:]))
|
||||
|
||||
# `strix login …` manages managed-platform sign-in (app.strix.ai) and
|
||||
# exits; it needs no target, Docker, or scan setup.
|
||||
if len(sys.argv) > 1 and sys.argv[1] == "login":
|
||||
from strix.interface.platform_cli import run_login
|
||||
# `strix cloud …` drives the managed platform (app.strix.ai) and exits;
|
||||
# it needs no target, Docker, or scan setup.
|
||||
if len(sys.argv) > 1 and sys.argv[1] == "cloud":
|
||||
from strix.interface.cloud import run_cloud
|
||||
|
||||
sys.exit(run_login(sys.argv[2:]))
|
||||
sys.exit(run_cloud(sys.argv[2:]))
|
||||
|
||||
from strix.llm.warmup import start_import_warmup
|
||||
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
"""`strix login` — managed platform sign-in (app.strix.ai).
|
||||
"""`strix cloud login` — managed platform sign-in (app.strix.ai).
|
||||
|
||||
Signing in runs an OAuth 2.0 device authorization flow in the browser, creates
|
||||
the Strix account and workspace when they do not exist yet, and stores a
|
||||
|
|
@ -35,8 +35,9 @@ _MAX_POLL_INTERVAL_S = 60
|
|||
_MAX_EXPIRES_IN_S = 30 * 60
|
||||
|
||||
_LOGIN_USAGE = (
|
||||
"Usage:\n strix login [--no-browser] [--scopes SCOPE ...] [--workspace WORKSPACE]\n"
|
||||
" strix login status\n strix login logout"
|
||||
"Usage:\n"
|
||||
" strix cloud login [--no-browser] [--scopes SCOPE ...] [--workspace WORKSPACE]\n"
|
||||
" strix cloud whoami\n strix cloud logout"
|
||||
)
|
||||
|
||||
_ROLE_RANK = {"viewer": 0, "analyst": 1, "admin": 2}
|
||||
|
|
@ -78,7 +79,7 @@ def logout() -> bool:
|
|||
|
||||
|
||||
def run_login(argv: list[str]) -> int:
|
||||
"""Entry point for ``strix login …``. Returns a process exit code."""
|
||||
"""Entry point for ``strix cloud login``. Returns a process exit code."""
|
||||
console = Console()
|
||||
subcommand = argv[0] if argv else None
|
||||
|
||||
|
|
@ -93,7 +94,7 @@ def run_login(argv: list[str]) -> int:
|
|||
|
||||
|
||||
def _login(console: Console, argv: list[str]) -> int:
|
||||
parser = argparse.ArgumentParser(prog="strix login", add_help=True)
|
||||
parser = argparse.ArgumentParser(prog="strix cloud login", add_help=True)
|
||||
parser.add_argument(
|
||||
"--no-browser",
|
||||
action="store_true",
|
||||
|
|
@ -123,7 +124,7 @@ def _login(console: Console, argv: list[str]) -> int:
|
|||
try:
|
||||
args = parser.parse_args(argv)
|
||||
except SystemExit as exc: # argparse already printed the message
|
||||
return int(exc.code or 2)
|
||||
return exc.code if isinstance(exc.code, int) else 2
|
||||
|
||||
console.print()
|
||||
host = urlparse(_app_url()).netloc or _app_url()
|
||||
|
|
@ -152,7 +153,8 @@ def _login(console: Console, argv: list[str]) -> int:
|
|||
except OSError as exc:
|
||||
console.print(f"[red]Sign-in succeeded, but the token could not be stored:[/] {exc}")
|
||||
console.print(
|
||||
f"[dim]Check that {AUTH_PATH.parent} is writable, then run `strix login` again.[/]"
|
||||
f"[dim]Check that {AUTH_PATH.parent} is writable, "
|
||||
"then run `strix cloud login` again.[/]"
|
||||
)
|
||||
return 1
|
||||
_print_success(console, record)
|
||||
|
|
@ -237,7 +239,7 @@ def _run_device_flow(
|
|||
break
|
||||
interval += delta
|
||||
|
||||
raise PlatformAuthError("the sign-in request expired. Run `strix login` again.")
|
||||
raise PlatformAuthError("the sign-in request expired. Run `strix cloud login` again.")
|
||||
|
||||
|
||||
def _handle_poll_error(poll: requests.Response) -> int | None:
|
||||
|
|
@ -461,15 +463,15 @@ def _print_success(console: Console, record: dict[str, Any]) -> None:
|
|||
console.print(f" Token: stored in [dim]{AUTH_PATH}[/]")
|
||||
console.print()
|
||||
console.print(
|
||||
"[dim]The managed API is ready. "
|
||||
"See https://docs.app.strix.ai for scans, credits, and top-ups.[/]"
|
||||
"[dim]The managed platform is ready. Run `strix cloud` to list the commands. "
|
||||
"See https://docs.app.strix.ai for the API reference.[/]"
|
||||
)
|
||||
|
||||
|
||||
def _status(console: Console) -> int:
|
||||
record = read_record()
|
||||
if record is None:
|
||||
console.print("[yellow]Not signed in.[/] Run [bold]strix login[/] to sign in.")
|
||||
console.print("[yellow]Not signed in.[/] Run [bold]strix cloud login[/] to sign in.")
|
||||
return 1
|
||||
email = record.get("email", "unknown")
|
||||
organization = record.get("organization_name") or record.get("organization_id", "")
|
||||
|
|
|
|||
|
|
@ -228,7 +228,10 @@ def test_resume_still_requires_targets_or_a_workspace(
|
|||
|
||||
assert "has no targets_info" in capsys.readouterr().err
|
||||
|
||||
def test_resume_non_object_run_json_exits(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
|
||||
|
||||
def test_resume_non_object_run_json_exits(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
|
||||
) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
run_dir = tmp_path / "strix_runs" / "pentest_abcd"
|
||||
run_dir.mkdir(parents=True)
|
||||
|
|
|
|||
355
tests/test_cloud_cli.py
Normal file
355
tests/test_cloud_cli.py
Normal file
|
|
@ -0,0 +1,355 @@
|
|||
"""Tests for the `strix cloud` CLI: routing, request building, and output."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import json
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
import pytest
|
||||
import requests
|
||||
|
||||
from strix.interface import cloud, platform_cli
|
||||
from strix.interface.cloud import http, render, runner
|
||||
from strix.interface.cloud.spec import SPEC
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
class FakeResponse:
|
||||
def __init__(
|
||||
self,
|
||||
status_code: int = 200,
|
||||
payload: Any = None,
|
||||
text: str = "",
|
||||
content: bytes = b"",
|
||||
) -> None:
|
||||
self.status_code = status_code
|
||||
self._payload = payload
|
||||
self.text = text if payload is None else json.dumps(payload)
|
||||
self.content = content
|
||||
self.ok = 200 <= status_code < 400
|
||||
self.headers = {"content-type": "application/json" if payload is not None else "text/plain"}
|
||||
|
||||
def json(self) -> Any:
|
||||
if self._payload is None:
|
||||
raise ValueError("no JSON")
|
||||
return self._payload
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _token_env(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setenv("STRIX_API_TOKEN", "test-token")
|
||||
|
||||
|
||||
def test_help_returns_zero() -> None:
|
||||
assert cloud.run_cloud([]) == 0
|
||||
assert cloud.run_cloud(["--help"]) == 0
|
||||
|
||||
|
||||
def test_unknown_group_returns_usage_error() -> None:
|
||||
assert cloud.run_cloud(["bogus"]) == 2
|
||||
|
||||
|
||||
def test_unknown_verb_returns_usage_error() -> None:
|
||||
assert cloud.run_cloud(["scans", "bogus"]) == 2
|
||||
|
||||
|
||||
def test_group_without_verb_lists_verbs() -> None:
|
||||
assert cloud.run_cloud(["scans"]) == 0
|
||||
|
||||
|
||||
def test_resolve_prefers_two_word_verbs() -> None:
|
||||
resolved = runner.resolve("billing", ["auto-topup", "update", "--enabled"])
|
||||
assert resolved is not None
|
||||
cmd, remaining = resolved
|
||||
assert cmd.path == "/billing/auto-topup"
|
||||
assert cmd.method == "PUT"
|
||||
assert remaining == ["--enabled"]
|
||||
|
||||
|
||||
def test_resolve_default_verb() -> None:
|
||||
resolved = runner.resolve("audit", [])
|
||||
assert resolved is not None
|
||||
cmd, remaining = resolved
|
||||
assert cmd.method == "GET"
|
||||
assert remaining == []
|
||||
|
||||
|
||||
def test_dest_converts_camel_case() -> None:
|
||||
assert runner._dest("scanId") == "scan_id"
|
||||
assert runner._dest("chatId") == "chat_id"
|
||||
assert runner._metavar("findingId") == "FINDING_ID"
|
||||
|
||||
|
||||
def test_placeholder_substitution(monkeypatch: pytest.MonkeyPatch, capsys: Any) -> None:
|
||||
seen: dict[str, Any] = {}
|
||||
|
||||
def fake_request(method: str, path: str, **kwargs: Any) -> FakeResponse:
|
||||
seen.update(method=method, path=path, query=kwargs.get("query"))
|
||||
return FakeResponse(payload={"id": "abc"})
|
||||
|
||||
monkeypatch.setattr(http, "request", fake_request)
|
||||
code = cloud.run_cloud(["scans", "get", "abc-123", "--json"])
|
||||
assert code == 0
|
||||
assert seen["method"] == "GET"
|
||||
assert seen["path"] == "/scans/abc-123"
|
||||
assert json.loads(capsys.readouterr().out) == {"id": "abc"}
|
||||
|
||||
|
||||
def test_query_and_body_collection(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
seen: dict[str, Any] = {}
|
||||
|
||||
def fake_request(_method: str, _path: str, **kwargs: Any) -> FakeResponse:
|
||||
seen.update(query=kwargs.get("query"), body=kwargs.get("body"))
|
||||
return FakeResponse(payload={"ok": True})
|
||||
|
||||
monkeypatch.setattr(http, "request", fake_request)
|
||||
assert cloud.run_cloud(["scans", "list", "--status", "running", "--json"]) == 0
|
||||
assert seen["query"] == {"status": "running"}
|
||||
|
||||
assert (
|
||||
cloud.run_cloud(
|
||||
[
|
||||
"scans",
|
||||
"start",
|
||||
"--engagement-type",
|
||||
"live_test",
|
||||
"--domain-ids",
|
||||
"d1",
|
||||
"d2",
|
||||
"--json",
|
||||
]
|
||||
)
|
||||
== 0
|
||||
)
|
||||
assert seen["body"] == {"engagement_type": "live_test", "domain_ids": ["d1", "d2"]}
|
||||
|
||||
|
||||
def test_data_merges_extra_fields(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
seen: dict[str, Any] = {}
|
||||
|
||||
def fake_request(_method: str, _path: str, **kwargs: Any) -> FakeResponse:
|
||||
seen["body"] = kwargs.get("body")
|
||||
return FakeResponse(payload={"ok": True})
|
||||
|
||||
monkeypatch.setattr(http, "request", fake_request)
|
||||
code = cloud.run_cloud(
|
||||
["scans", "start", "--data", '{"engagement_type": "code_review"}', "--json"]
|
||||
)
|
||||
assert code == 0
|
||||
assert seen["body"] == {"engagement_type": "code_review"}
|
||||
|
||||
|
||||
def test_data_reads_a_file(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
||||
seen: dict[str, Any] = {}
|
||||
|
||||
def fake_request(_method: str, _path: str, **kwargs: Any) -> FakeResponse:
|
||||
seen["body"] = kwargs.get("body")
|
||||
return FakeResponse(payload={"ok": True})
|
||||
|
||||
monkeypatch.setattr(http, "request", fake_request)
|
||||
request_file = tmp_path / "request.json"
|
||||
request_file.write_text('{"focus": "IDOR"}', encoding="utf-8")
|
||||
assert cloud.run_cloud(["scans", "start", "--data", f"@{request_file}", "--json"]) == 0
|
||||
assert seen["body"] == {"focus": "IDOR"}
|
||||
|
||||
|
||||
def test_data_reads_stdin(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
seen: dict[str, Any] = {}
|
||||
|
||||
def fake_request(_method: str, _path: str, **kwargs: Any) -> FakeResponse:
|
||||
seen["body"] = kwargs.get("body")
|
||||
return FakeResponse(payload={"ok": True})
|
||||
|
||||
monkeypatch.setattr(http, "request", fake_request)
|
||||
monkeypatch.setattr("sys.stdin", io.StringIO('{"context": "staging"}'))
|
||||
assert cloud.run_cloud(["scans", "start", "--data", "-", "--json"]) == 0
|
||||
assert seen["body"] == {"context": "staging"}
|
||||
|
||||
|
||||
def test_data_reports_a_missing_file(tmp_path: Path) -> None:
|
||||
assert cloud.run_cloud(["scans", "start", "--data", f"@{tmp_path / 'nope.json'}"]) == 1
|
||||
|
||||
|
||||
def test_auto_topup_removes_the_monthly_cap(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
seen: dict[str, Any] = {}
|
||||
|
||||
def fake_request(_method: str, _path: str, **kwargs: Any) -> FakeResponse:
|
||||
seen["body"] = kwargs.get("body")
|
||||
return FakeResponse(payload={"ok": True})
|
||||
|
||||
monkeypatch.setattr(http, "request", fake_request)
|
||||
code = cloud.run_cloud(
|
||||
[
|
||||
"billing",
|
||||
"auto-topup",
|
||||
"update",
|
||||
"--enabled",
|
||||
"--topup-credits",
|
||||
"20",
|
||||
"--no-monthly-cap",
|
||||
"--json",
|
||||
]
|
||||
)
|
||||
assert code == 0
|
||||
assert seen["body"] == {
|
||||
"enabled": True,
|
||||
"topup_credits": 20,
|
||||
"monthly_cap_credits": None,
|
||||
}
|
||||
|
||||
|
||||
def test_costs_default_verb_is_the_overview(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
seen: dict[str, Any] = {}
|
||||
|
||||
def fake_request(_method: str, path: str, **_kwargs: Any) -> FakeResponse:
|
||||
seen["path"] = path
|
||||
return FakeResponse(payload={"total_cost": 1})
|
||||
|
||||
monkeypatch.setattr(http, "request", fake_request)
|
||||
assert cloud.run_cloud(["costs", "--json"]) == 0
|
||||
assert seen["path"] == "/llm-costs"
|
||||
|
||||
|
||||
def test_binary_download_writes_a_file(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
||||
monkeypatch.setattr(
|
||||
http, "request", lambda *_a, **_k: FakeResponse(status_code=200, content=b"%PDF-1.7")
|
||||
)
|
||||
target = tmp_path / "report.pdf"
|
||||
assert cloud.run_cloud(["scans", "report", "scan-1", "--output", str(target)]) == 0
|
||||
assert target.read_bytes() == b"%PDF-1.7"
|
||||
|
||||
|
||||
def test_wait_polls_until_the_status_is_final(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
statuses = iter(["running", "completed"])
|
||||
|
||||
def fake_request(method: str, _path: str, **_kwargs: Any) -> FakeResponse:
|
||||
if method == "POST":
|
||||
return FakeResponse(payload={"id": "scan-1", "status": "pending"})
|
||||
return FakeResponse(payload={"id": "scan-1", "status": next(statuses)})
|
||||
|
||||
monkeypatch.setattr(http, "request", fake_request)
|
||||
monkeypatch.setattr(runner, "_WAIT_POLL_S", 0)
|
||||
assert cloud.run_cloud(["scans", "start", "--domain-ids", "d1", "--wait", "--json"]) == 0
|
||||
assert next(statuses, None) is None
|
||||
|
||||
|
||||
def test_insufficient_credits_exits_with_payment_code(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(
|
||||
http, "request", lambda *_a, **_k: FakeResponse(status_code=402, payload={})
|
||||
)
|
||||
assert cloud.run_cloud(["scans", "start", "--domain-ids", "d1"]) == http.EXIT_PAYMENT
|
||||
|
||||
|
||||
def test_data_rejects_non_object() -> None:
|
||||
assert cloud.run_cloud(["scans", "start", "--data", "[1,2]"]) == 1
|
||||
assert cloud.run_cloud(["scans", "start", "--data", "not json"]) == 1
|
||||
|
||||
|
||||
def test_missing_token_exits_with_auth_code(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
monkeypatch.delenv("STRIX_API_TOKEN", raising=False)
|
||||
monkeypatch.setattr(platform_cli, "AUTH_PATH", tmp_path / "platform-auth.json")
|
||||
assert cloud.run_cloud(["credits"]) == http.EXIT_AUTH
|
||||
|
||||
|
||||
def test_http_error_exit_codes(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
for status, expected in ((401, http.EXIT_AUTH), (403, http.EXIT_AUTH), (500, http.EXIT_ERROR)):
|
||||
monkeypatch.setattr(
|
||||
http,
|
||||
"request",
|
||||
lambda *_a, _s=status, **_k: FakeResponse(status_code=_s, payload={"error": "x"}),
|
||||
)
|
||||
assert cloud.run_cloud(["scans", "list"]) == expected
|
||||
|
||||
|
||||
def test_credits_alias_routes_to_billing(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
seen: dict[str, Any] = {}
|
||||
|
||||
def fake_request(_method: str, path: str, **_kwargs: Any) -> FakeResponse:
|
||||
seen["path"] = path
|
||||
return FakeResponse(payload={"balance": 3})
|
||||
|
||||
monkeypatch.setattr(http, "request", fake_request)
|
||||
assert cloud.run_cloud(["credits", "--json"]) == 0
|
||||
assert seen["path"] == "/billing/credits"
|
||||
|
||||
|
||||
def test_topup_no_pay_prints_challenge(monkeypatch: pytest.MonkeyPatch, capsys: Any) -> None:
|
||||
challenge = {"payment_requirements": [{"amount": 500}]}
|
||||
monkeypatch.setattr(
|
||||
http, "request", lambda *_a, **_k: FakeResponse(status_code=402, payload=challenge)
|
||||
)
|
||||
code = cloud.run_cloud(["billing", "topup", "--credits", "5", "--no-pay", "--json"])
|
||||
assert code == http.EXIT_PAYMENT
|
||||
assert json.loads(capsys.readouterr().out) == challenge
|
||||
|
||||
|
||||
def test_topup_success_without_payment(monkeypatch: pytest.MonkeyPatch, capsys: Any) -> None:
|
||||
receipt = {"credits_granted": 5, "duplicate": False, "balance": 5}
|
||||
monkeypatch.setattr(
|
||||
http, "request", lambda *_a, **_k: FakeResponse(status_code=200, payload=receipt)
|
||||
)
|
||||
code = cloud.run_cloud(["billing", "topup", "--credits", "5", "--json"])
|
||||
assert code == 0
|
||||
assert json.loads(capsys.readouterr().out) == receipt
|
||||
|
||||
|
||||
def test_render_json_mode_when_not_a_tty() -> None:
|
||||
assert render.json_mode(flag=True) is True
|
||||
# Under pytest, stdout is captured and is not a terminal.
|
||||
assert render.json_mode(flag=False) is True
|
||||
|
||||
|
||||
def test_render_list_extraction() -> None:
|
||||
rows = render._list_of_dicts({"scans": [{"id": "a"}, {"id": "b"}]})
|
||||
assert rows == [{"id": "a"}, {"id": "b"}]
|
||||
assert render._list_of_dicts({"scans": [], "total": 1}) is None
|
||||
assert render._list_of_dicts([{"id": "a"}, "x"]) is None
|
||||
|
||||
|
||||
def test_spec_paths_are_well_formed() -> None:
|
||||
for group, commands in SPEC.items():
|
||||
for verb, cmd in commands.items():
|
||||
assert cmd.path.startswith("/"), f"{group} {verb}"
|
||||
assert cmd.method in ("GET", "POST", "PUT", "PATCH", "DELETE"), f"{group} {verb}"
|
||||
assert cmd.help, f"{group} {verb} has no help text"
|
||||
for param in cmd.query + cmd.body:
|
||||
assert param.kind in ("str", "int", "float", "bool", "list", "json"), (
|
||||
f"{group} {verb} {param.name}"
|
||||
)
|
||||
|
||||
|
||||
def test_every_command_builds_a_parser() -> None:
|
||||
for group, commands in SPEC.items():
|
||||
for verb, cmd in commands.items():
|
||||
parser = runner._build_parser(group, verb, cmd)
|
||||
assert parser.prog == f"strix cloud {group} {verb}"
|
||||
|
||||
|
||||
def test_app_url_and_timeout_overrides(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
seen: dict[str, Any] = {}
|
||||
|
||||
def fake_request(_method: str, url: str, **kwargs: Any) -> FakeResponse:
|
||||
seen["url"] = url
|
||||
seen["timeout"] = kwargs.get("timeout")
|
||||
return FakeResponse(status_code=200, payload={"balance": 1})
|
||||
|
||||
monkeypatch.setattr(http, "api_token", lambda _override=None: "t")
|
||||
monkeypatch.setattr(requests, "request", fake_request)
|
||||
code = cloud.run_cloud(
|
||||
["credits", "--app-url", "https://example.test/", "--timeout", "7", "--json"]
|
||||
)
|
||||
assert code == 0
|
||||
assert seen["url"] == "https://example.test/api/v1/billing/credits"
|
||||
assert seen["timeout"] == 7
|
||||
|
||||
|
||||
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
|
||||
Loading…
Add table
Reference in a new issue