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

feat(cloud): upload local source for managed scans

Browse source
This commit is contained in:
bearsyankees 2026-08-27 18:11:07 -04:00
parent bb32a7d4b0
commit 7095d1dcae
12 changed files with 906 additions and 39 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
AGENTS.md
View file

@ -41,11 +41,14 @@ Target-specific workflows built on the same engine:
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 scans start --source . --dry-run --show-files --json # review local upload
strix cloud scans start --source . --yes --engagement-type code_review --wait
strix cloud vulns list --severity critical
strix cloud billing topup --credits 20 # buy credits when a scan returns exit code 5
```
- Account setup runs from the CLI too: `strix cloud workspaces list|create|use`, `strix cloud org members invite`, `strix cloud billing subscribe --plan strix_cloud`, `strix cloud billing portal`, `strix cloud integrations install github`, and `strix cloud domains verify <id>`. The last four end at a person: the command prints a link or a DNS record for the user to open or add, and it never completes the payment, the installation, or the DNS change for them.
- Every REST operation has a `strix cloud <resource> <verb>` command. Run `strix cloud` to list them. Output is JSON when stdout is not a terminal (or with `--json`), and there are no prompts without a TTY. Exit codes: `0` success, `1` error, `2` usage, `4` auth or plan limit, `5` payment required. `--token` or `STRIX_API_TOKEN` overrides the stored sign-in. `--data` adds extra request fields as JSON, and accepts `@file` or `-` for standard input.
- Local source uploads require `uploads:write`. Review with `scans start --source . --dry-run --show-files --json`, then approve with `--yes`. Git ignores, hidden files, `.git`, symlinks, dependency/build output, secret-like filenames, and nested archives are excluded by default; `.strixignore` and `--exclude` narrow the manifest further.
- 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). Managed API docs for LLMs: https://docs.app.strix.ai/llms.txt.

16
docs/cloud/overview.mdx
View file

@ -35,6 +35,22 @@ Skip the setup. Run Strix in the cloud at [app.strix.ai](https://app.strix.ai).
2. Connect your repository or enter a target URL
3. Launch your first scan
## Scan Local Source
Send a local working tree to the managed white-box scanner without connecting a source-control provider:
```bash
# Review the exact file manifest first. Nothing is uploaded.
strix cloud scans start --source . --dry-run --show-files --json
# Approve the reviewed selection, upload it, and wait for the scan.
strix cloud scans start --source . --yes --engagement-type code_review --wait
```
In a Git repository, Strix includes tracked files and untracked files that are not ignored. Hidden files, `.git`, symlinks, dependencies and build output, secret-like filenames, and nested archives are excluded by default. Use `.strixignore` or repeat `--exclude GLOB` for project-specific exclusions. `--include-hidden`, `--include-sensitive`, and `--include-archives` are explicit opt-ins.
The CLI limits individual files, total expanded bytes, archive bytes, and file count. Non-interactive environments must pass `--yes`, so an agent cannot upload a workspace without an explicit approval flag.
<Card title="Try Strix Cloud" icon="rocket" href="https://app.strix.ai">
Run your first pentest in minutes.
</Card>

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

@ -1,6 +1,6 @@
---
name: managed-pentesting-with-strix
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.
description: Run a managed pentest of a web app, API, repository, or local workspace on the app.strix.ai platform with the `strix cloud` CLI or REST API — no local Docker or LLM key needed. Safely review and upload local source, register assets, launch and poll scans, triage vulnerabilities, export SARIF, download compliance reports, start PR reviews, buy credits, and set up schedules or webhooks. Use for managed, continuous, scheduled, team-tracked, or sandboxed-agent security testing.
license: Apache-2.0
metadata:
author: usestrix
@ -33,7 +33,7 @@ The platform enforces plan and role limits, and the CLI passes the platform mess
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
strix cloud login --scopes scans:read scans:write uploads: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).
@ -168,6 +168,19 @@ Useful flags (each maps to a `CreateScanRequest` field):
The response is `{ scan_id, title, status }` with `status` = `pending`.
### Scan a local workspace in the cloud
Review the exact local upload before sending it. An agent or CI process must never skip this review merely because it can pass `--yes`:
```bash
strix cloud scans start --source . --dry-run --show-files --json
strix cloud scans start --source . --yes --engagement-type code_review --wait
```
The default selection is privacy-conscious: in a Git worktree it includes tracked files plus untracked files that are not ignored; it honors `.gitignore`, excludes every hidden path component, always excludes `.git`, symlinks, dependencies/build output, secret-like filenames, and nested archives, and enforces file-count, per-file, expanded-size, and compressed-size limits. Add project exclusions to `.strixignore` (one exclude glob per line) or repeat `--exclude GLOB`.
Only use `--include-hidden`, `--include-sensitive`, or `--include-archives` after the dry-run manifest shows that the scan needs them. Hidden and sensitive files are separate opt-ins: for example, including `.env` requires both `--include-hidden` and `--include-sensitive`. Non-interactive uploads require `--yes`; interactive terminals show one summary and confirmation prompt. The CLI deletes its temporary archive after the request and best-effort deletes the uploaded object if scan creation fails.
## 3. Wait for completion
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.

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

@ -3,7 +3,7 @@
from __future__ import annotations
import os
from typing import Any, cast
from typing import TYPE_CHECKING, Any, cast
import requests
@ -11,6 +11,10 @@ from strix.config import load_settings
from strix.interface.platform_cli import read_record
if TYPE_CHECKING:
from pathlib import Path
_DEFAULT_TIMEOUT_S = 120
_app_url_override: str | None = None
_timeout_s: float = _DEFAULT_TIMEOUT_S
@ -85,6 +89,33 @@ def request(
return response
def upload_file(signed_url: str, upload_token: str, path: Path) -> None:
"""Stream a file to a platform-issued storage URL."""
try:
with path.open("rb") as stream:
response = requests.put(
signed_url,
data=stream,
headers={
"Authorization": f"Bearer {upload_token}",
"Content-Type": "application/zip",
},
timeout=_timeout_s,
)
except (OSError, requests.RequestException) as exc:
raise CloudError(f"source upload failed: {exc}") from exc
if not response.ok:
detail = ""
try:
payload = response.json()
if isinstance(payload, dict):
fields = cast("dict[str, Any]", payload)
detail = str(fields.get("message") or fields.get("error") or "")
except ValueError:
pass
raise CloudError(detail or f"source upload failed (HTTP {response.status_code})")
def parsed(response: requests.Response) -> Any:
content_type = response.headers.get("content-type", "")
if "application/json" in content_type:

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

@ -188,9 +188,7 @@ def _print_cards(
def _print_detail(console: Console, data: dict[str, Any]) -> None:
"""Render one API record as a readable field/value view."""
keys = [key for key in _PREFERRED_KEYS if key in data and key not in _INTERNAL_COLUMNS]
keys.extend(
key for key in data if key not in keys and key not in _INTERNAL_COLUMNS
)
keys.extend(key for key in data if key not in keys and key not in _INTERNAL_COLUMNS)
table = Table(show_header=False, show_edge=False, box=None, padding=(0, 2))
table.add_column("field", style="bold cyan", no_wrap=True)
table.add_column("value", overflow="fold")

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

@ -15,6 +15,7 @@ import subprocess
import sys
import time
import webbrowser
from contextlib import suppress
from pathlib import Path
from typing import Any, cast
from urllib.parse import quote
@ -23,6 +24,7 @@ from rich.console import Console
from strix.interface.cloud import http
from strix.interface.cloud.render import emit, json_mode
from strix.interface.cloud.source_upload import SourceBundle, prepare_source, remove_bundle
from strix.interface.cloud.spec import DEFAULT_VERBS, SPEC, Cmd, P
@ -99,7 +101,7 @@ def run(group: str, verb_label: str, cmd: Cmd, argv: list[str]) -> int:
return exc.exit_code
def _execute(
def _execute( # noqa: PLR0912
console: Console,
cmd: Cmd,
args: argparse.Namespace,
@ -112,21 +114,63 @@ def _execute(
) -> 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,
)
source_bundle: SourceBundle | None = None
source_upload_id: str | None = None
try:
if cmd.path == "/scans" and cmd.method == "POST":
source_bundle = _prepare_scan_source(console, args, as_json=as_json)
if source_bundle is not None and getattr(args, "dry_run", False):
emit(
console,
{
"source": source_bundle.summary(
show_files=getattr(args, "show_files", False)
)
},
as_json=as_json,
)
return http.EXIT_OK
if source_bundle is not None:
source_upload_id = _upload_scan_source(source_bundle, token=token)
existing = body.get("upload_ids")
body["upload_ids"] = [
*(existing if isinstance(existing, list) else []),
source_upload_id,
]
response = http.request(
cmd.method,
path,
token=token,
query=query or None,
body=body if cmd.method in ("POST", "PUT", "PATCH") else None,
)
except BaseException:
if source_upload_id is not None:
_delete_upload(source_upload_id, token=token)
raise
finally:
if source_bundle is not None:
remove_bundle(source_bundle)
if cmd.binary:
return _emit_binary(console, response, getattr(args, "output", None))
result = http.check(response)
try:
result = http.check(response)
except BaseException:
if source_upload_id is not None:
_delete_upload(source_upload_id, token=token)
raise
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)
if source_bundle is not None:
result = {
"source": source_bundle.summary(show_files=getattr(args, "show_files", False)),
"upload_id": source_upload_id,
"scan": result,
}
if cmd.link:
return _handoff_link(console, cmd, args, result, as_json=as_json)
workspace_list = cmd.method == "GET" and cmd.path == "/workspaces"
@ -136,11 +180,7 @@ def _execute(
as_json=as_json,
row_numbers=workspace_list,
omit_columns=frozenset({"id"}) if workspace_list else frozenset(),
hint=(
"Switch with `strix cloud workspaces use NUMBER`."
if workspace_list
else None
),
hint=("Switch with `strix cloud workspaces use NUMBER`." if workspace_list else None),
)
return http.EXIT_OK
@ -149,7 +189,8 @@ def _handoff_link(
console: Console, cmd: Cmd, args: argparse.Namespace, result: Any, *, as_json: bool
) -> int:
"""Print a hosted URL a person must open, and open the browser when interactive."""
url = result.get(cmd.link) if isinstance(result, dict) else None
fields = cast("dict[str, Any]", result) if isinstance(result, dict) else {}
url = fields.get(cmd.link) if cmd.link else None
if not isinstance(url, str) or not url:
emit(console, result, as_json=as_json)
return http.EXIT_OK
@ -243,9 +284,151 @@ def _build_parser(group: str, verb_label: str, cmd: Cmd) -> argparse.ArgumentPar
"Defaults to MPPX_STRIPE_PAYMENT_METHOD."
),
)
if cmd.path == "/scans" and cmd.method == "POST":
parser.add_argument(
"--source",
default=None,
metavar="DIRECTORY",
help="Package a local directory, upload it, and attach it to this scan.",
)
parser.add_argument(
"--dry-run",
action="store_true",
help="Build and print the source manifest without uploading or starting a scan.",
)
parser.add_argument(
"--yes",
action="store_true",
help="Approve the displayed source upload without an interactive prompt.",
)
parser.add_argument(
"--show-files",
action="store_true",
help="Include every selected relative path in the source manifest.",
)
parser.add_argument(
"--exclude",
action="append",
default=[],
metavar="GLOB",
help="Exclude a path glob from the upload. May be repeated.",
)
parser.add_argument(
"--include-hidden",
action="store_true",
help="Include hidden files except .git and secret-like filenames.",
)
parser.add_argument(
"--include-sensitive",
action="store_true",
help="Include files with secret-like names. Use only after reviewing --dry-run.",
)
parser.add_argument(
"--include-archives",
action="store_true",
help="Include nested archives. Use only when they are required source inputs.",
)
return parser
def _prepare_scan_source(
console: Console, args: argparse.Namespace, *, as_json: bool
) -> SourceBundle | None:
source = getattr(args, "source", None)
source_flags = (
"dry_run",
"show_files",
"include_hidden",
"include_sensitive",
"include_archives",
)
if source is None:
if any(getattr(args, name, False) for name in source_flags) or getattr(args, "exclude", []):
raise http.CloudError("source upload options require --source DIRECTORY.")
return None
bundle = prepare_source(
source,
include_hidden=bool(getattr(args, "include_hidden", False)),
include_sensitive=bool(getattr(args, "include_sensitive", False)),
include_archives=bool(getattr(args, "include_archives", False)),
exclude=cast("list[str]", getattr(args, "exclude", [])),
)
if getattr(args, "dry_run", False):
return bundle
if getattr(args, "yes", False):
return bundle
if as_json or not (sys.stdin.isatty() and sys.stdout.isatty()):
remove_bundle(bundle)
raise http.CloudError(
"source upload requires explicit approval in non-interactive mode. "
"Review with --dry-run --show-files, then rerun with --yes."
)
console.print(
"[bold]Local source upload[/]\n"
f" {len(bundle.manifest.files):,} file(s), "
f"{_format_bytes(bundle.manifest.total_bytes)} "
f"({_format_bytes(bundle.archive_bytes)} compressed)\n"
f" {sum(bundle.manifest.excluded.values()):,} path(s) excluded\n"
" Only the selected files will be sent to Strix Cloud."
)
answer = console.input("Upload this source and start the scan? [y/N]: ").strip().lower()
if answer not in ("y", "yes"):
remove_bundle(bundle)
raise http.CloudError("source upload cancelled.")
return bundle
def _upload_scan_source(bundle: SourceBundle, *, token: str | None) -> str:
file_name = f"strix-source-{bundle.archive_sha256[:12]}.zip"
requested = http.check(
http.request(
"POST",
"/uploads/request",
token=token,
body={
"file_name": file_name,
"file_size": bundle.archive_bytes,
"category": "repository",
},
)
)
if not isinstance(requested, dict):
raise http.CloudError("the platform returned an invalid source upload response.")
fields = cast("dict[str, Any]", requested)
upload_id = fields.get("upload_id")
signed_url = fields.get("signed_url")
upload_token = fields.get("token")
if not all(isinstance(value, str) and value for value in (upload_id, signed_url, upload_token)):
raise http.CloudError("the platform did not return complete source upload credentials.")
try:
http.upload_file(cast("str", signed_url), cast("str", upload_token), bundle.archive_path)
http.check(
http.request(
"POST",
"/uploads/complete",
token=token,
body={"upload_id": upload_id},
)
)
except BaseException:
_delete_upload(cast("str", upload_id), token=token)
raise
return cast("str", upload_id)
def _delete_upload(upload_id: str, *, token: str | None) -> None:
with suppress(http.CloudError):
http.request("DELETE", f"/uploads/{quote(upload_id, safe='')}", token=token)
def _format_bytes(value: int) -> str:
if value < 1024:
return f"{value} B"
if value < 1024 * 1024:
return f"{value / 1024:.1f} KB"
return f"{value / (1024 * 1024):.1f} MB"
def _add_option(parser: argparse.ArgumentParser, param: P) -> None:
flag = "--" + (param.flag or param.name.replace("_", "-"))
if param.kind == "bool":
@ -342,10 +525,11 @@ def _poll(console: Console, path: str, *, token: str | None, as_json: bool) -> A
"""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 ""
current: Any = http.check(http.request("GET", path, token=token))
fields = cast("dict[str, Any]", current) if isinstance(current, dict) else {}
status = str(fields.get("status", ""))
if status.lower() in _TERMINAL_STATUSES:
return current
return fields if isinstance(current, dict) else current
if not as_json:
console.print(f"[dim] status: {status or 'unknown'}[/]")

382
strix/interface/cloud/source_upload.py Normal file
View file

@ -0,0 +1,382 @@
"""Privacy-conscious local source packaging for managed scans."""
from __future__ import annotations
import fnmatch
import hashlib
import os
import shutil
import stat
import subprocess # nosec B404
import tempfile
import zipfile
from collections import Counter
from dataclasses import dataclass
from pathlib import Path, PurePosixPath
from strix.interface.cloud import http
MAX_FILES = 20_000
MAX_FILE_BYTES = 25 * 1024 * 1024
MAX_TOTAL_BYTES = 250 * 1024 * 1024
MAX_ARCHIVE_BYTES = 50 * 1024 * 1024
_ALWAYS_EXCLUDED_DIRS = frozenset(
{
".git",
".hg",
".svn",
"node_modules",
"vendor",
"venv",
"__pycache__",
"dist",
"build",
"coverage",
"target",
}
)
_SENSITIVE_NAMES = frozenset(
{
"id_rsa",
"id_dsa",
"id_ecdsa",
"id_ed25519",
"credentials.json",
"service-account.json",
"service_account.json",
".env",
".npmrc",
".pypirc",
".netrc",
}
)
_SENSITIVE_PATTERNS = (
"*.pem",
"*.key",
"*.p12",
"*.pfx",
"*.keystore",
"*.jks",
"secrets.*",
"secret.*",
".env.*",
)
_ARCHIVE_SUFFIXES = (
".zip",
".tar",
".tgz",
".tar.gz",
".tar.bz2",
".tar.xz",
".7z",
".rar",
".gz",
".bz2",
".xz",
)
@dataclass(frozen=True)
class SelectedFile:
path: Path
archive_name: str
size: int
@dataclass(frozen=True)
class SourceManifest:
source: Path
files: tuple[SelectedFile, ...]
excluded: Counter[str]
include_hidden: bool
include_sensitive: bool
include_archives: bool
@property
def total_bytes(self) -> int:
return sum(item.size for item in self.files)
def as_dict(
self,
*,
show_files: bool,
archive_bytes: int | None = None,
archive_sha256: str | None = None,
) -> dict[str, object]:
result: dict[str, object] = {
"source": str(self.source),
"file_count": len(self.files),
"uncompressed_bytes": self.total_bytes,
"excluded_count": sum(self.excluded.values()),
"excluded_by_reason": dict(sorted(self.excluded.items())),
"include_hidden": self.include_hidden,
"include_sensitive": self.include_sensitive,
"include_archives": self.include_archives,
}
if archive_bytes is not None:
result["archive_bytes"] = archive_bytes
if archive_sha256 is not None:
result["archive_sha256"] = archive_sha256
if show_files:
result["files"] = [item.archive_name for item in self.files]
return result
@dataclass(frozen=True)
class SourceBundle:
manifest: SourceManifest
archive_path: Path
archive_bytes: int
archive_sha256: str
def summary(self, *, show_files: bool) -> dict[str, object]:
return self.manifest.as_dict(
show_files=show_files,
archive_bytes=self.archive_bytes,
archive_sha256=self.archive_sha256,
)
def prepare_source(
value: str,
*,
include_hidden: bool,
include_sensitive: bool,
include_archives: bool,
exclude: list[str],
) -> SourceBundle:
"""Select safe source files and build a bounded temporary ZIP archive."""
source = Path(value).expanduser().resolve()
if not source.is_dir():
raise http.CloudError(f"--source must be a directory: {source}")
manifest = select_source(
source,
include_hidden=include_hidden,
include_sensitive=include_sensitive,
include_archives=include_archives,
exclude=exclude,
)
if not manifest.files:
raise http.CloudError("no files remain after applying source upload exclusions.")
with tempfile.NamedTemporaryFile(prefix="strix-source-", suffix=".zip", delete=False) as handle:
archive_path = Path(handle.name)
try:
_write_archive(archive_path, manifest.files)
except BaseException:
archive_path.unlink(missing_ok=True)
raise
archive_bytes = archive_path.stat().st_size
if archive_bytes > MAX_ARCHIVE_BYTES:
archive_path.unlink(missing_ok=True)
raise http.CloudError(
"source archive is larger than the 50 MB upload limit; narrow --source or "
"add --exclude patterns."
)
digest = _sha256(archive_path)
return SourceBundle(manifest, archive_path, archive_bytes, digest)
def select_source(
source: Path,
*,
include_hidden: bool = False,
include_sensitive: bool = False,
include_archives: bool = False,
exclude: list[str] | None = None,
) -> SourceManifest:
excluded: Counter[str] = Counter()
selected: list[SelectedFile] = []
patterns = [*_load_ignore_patterns(source), *(exclude or [])]
total_bytes = 0
for relative in _candidate_paths(source):
archive_name = relative.as_posix()
reason = _exclusion_reason(
relative,
include_hidden=include_hidden,
include_sensitive=include_sensitive,
include_archives=include_archives,
patterns=patterns,
)
if reason:
excluded[reason] += 1
continue
path = source / relative
try:
info = path.lstat()
except OSError:
excluded["unreadable"] += 1
continue
if not stat.S_ISREG(info.st_mode):
excluded["symlink_or_non_file"] += 1
continue
if info.st_size > MAX_FILE_BYTES:
raise http.CloudError(
f"{archive_name} is larger than the 25 MB per-file limit; exclude it explicitly."
)
selected.append(SelectedFile(path, archive_name, info.st_size))
total_bytes += info.st_size
if len(selected) > MAX_FILES:
raise http.CloudError(
f"source contains more than {MAX_FILES:,} files; narrow --source or add exclusions."
)
if total_bytes > MAX_TOTAL_BYTES:
raise http.CloudError(
"selected source is larger than the 250 MB expanded-size limit; narrow --source "
"or add --exclude patterns."
)
selected.sort(key=lambda item: item.archive_name)
return SourceManifest(
source,
tuple(selected),
excluded,
include_hidden,
include_sensitive,
include_archives,
)
def remove_bundle(bundle: SourceBundle) -> None:
bundle.archive_path.unlink(missing_ok=True)
def _candidate_paths(source: Path) -> list[Path]:
git_root = _git_root(source)
if git_root is not None:
git = shutil.which("git")
if git is None:
return [path.relative_to(source) for path in source.rglob("*")]
relative_source = source.relative_to(git_root)
command = [
git,
"-C",
str(git_root),
"ls-files",
"-z",
"--cached",
"--others",
"--exclude-standard",
"--",
]
if relative_source != Path():
command.append(relative_source.as_posix())
result = subprocess.run( # noqa: S603 # nosec B603
command, check=False, capture_output=True
)
if result.returncode == 0:
paths: list[Path] = []
for raw in result.stdout.split(b"\0"):
if not raw:
continue
repo_relative = Path(os.fsdecode(raw))
try:
paths.append(repo_relative.relative_to(relative_source))
except ValueError:
continue
return paths
return [path.relative_to(source) for path in source.rglob("*")]
def _git_root(source: Path) -> Path | None:
git = shutil.which("git")
if git is None:
return None
result = subprocess.run( # noqa: S603 # nosec B603
[git, "-C", str(source), "rev-parse", "--show-toplevel"],
check=False,
capture_output=True,
text=True,
)
if result.returncode != 0:
return None
try:
return Path(result.stdout.strip()).resolve()
except OSError:
return None
def _exclusion_reason( # noqa: PLR0911
relative: Path,
*,
include_hidden: bool,
include_sensitive: bool,
include_archives: bool,
patterns: list[str],
) -> str | None:
parts = relative.parts
if any(part == ".git" for part in parts):
return "git_metadata"
if any(part in _ALWAYS_EXCLUDED_DIRS for part in parts[:-1]):
return "dependency_or_build_output"
if not include_hidden and any(part.startswith(".") for part in parts):
return "hidden"
posix = PurePosixPath(relative.as_posix())
if any(
posix.match(pattern) or fnmatch.fnmatch(relative.as_posix(), pattern)
for pattern in patterns
):
return "user_pattern"
name = relative.name.lower()
if not include_sensitive and (
name in _SENSITIVE_NAMES
or any(fnmatch.fnmatch(name, pattern) for pattern in _SENSITIVE_PATTERNS)
):
return "sensitive_filename"
if not include_archives and name.endswith(_ARCHIVE_SUFFIXES):
return "nested_archive"
return None
def _write_archive(destination: Path, files: tuple[SelectedFile, ...]) -> None:
with zipfile.ZipFile(
destination, "w", compression=zipfile.ZIP_DEFLATED, compresslevel=6
) as archive:
for item in files:
flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
try:
descriptor = os.open(item.path, flags)
except OSError as exc:
raise http.CloudError(f"could not safely read {item.archive_name}: {exc}") from exc
with os.fdopen(descriptor, "rb") as source_file:
current = os.fstat(source_file.fileno())
if not stat.S_ISREG(current.st_mode) or current.st_size != item.size:
raise http.CloudError(
f"{item.archive_name} changed while the source archive was being built; "
"retry."
)
info = zipfile.ZipInfo(item.archive_name)
info.compress_type = zipfile.ZIP_DEFLATED
info.external_attr = 0o100644 << 16
with archive.open(info, "w", force_zip64=True) as target:
shutil.copyfileobj(source_file, target, length=1024 * 1024)
def _sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as stream:
for chunk in iter(lambda: stream.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def _load_ignore_patterns(source: Path) -> list[str]:
path = source / ".strixignore"
try:
lines = path.read_text(encoding="utf-8").splitlines()
except FileNotFoundError:
return []
except OSError as exc:
raise http.CloudError(f"could not read {path}: {exc}") from exc
patterns: list[str] = []
for line_number, raw in enumerate(lines, start=1):
value = raw.strip()
if not value or value.startswith("#"):
continue
if value.startswith("!"):
raise http.CloudError(
f"{path}:{line_number}: negated patterns are not supported; use exclude-only globs."
)
patterns.append(value)
return patterns

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

@ -36,8 +36,7 @@ def run_workspace_use(argv: list[str]) -> int:
metavar="SCOPE",
default=None,
help=(
"API scopes for the new token. Without this option, preserve the stored token's "
"scopes."
"API scopes for the new token. Without this option, preserve the stored token's scopes."
),
)
parser.add_argument("--json", action="store_true", help="Print the raw JSON response.")
@ -71,8 +70,10 @@ def _use(console: Console, args: argparse.Namespace, *, as_json: bool) -> int:
body["scopes"] = args.scopes
elif args.token is None:
stored_scopes = record.get("scopes")
if isinstance(stored_scopes, list) and stored_scopes and all(
isinstance(scope, str) for scope in stored_scopes
if (
isinstance(stored_scopes, list)
and stored_scopes
and all(isinstance(scope, str) for scope in stored_scopes)
):
body["scopes"] = stored_scopes
minted = http.check(
@ -147,7 +148,6 @@ def _find_workspace(selector: str, *, token: str | None) -> dict[str, Any]:
f"multiple workspaces are named {wanted!r}. Use its list number: {numbers}"
)
names = ", ".join(
f"{index}: {workspace.get('name')}"
for index, workspace in enumerate(workspaces, start=1)
f"{index}: {workspace.get('name')}" for index, workspace in enumerate(workspaces, start=1)
)
raise http.CloudError(f"no workspace matches {wanted!r}. Your workspaces: {names}")

13
strix/interface/completions.py
View file

@ -108,6 +108,19 @@ def _command_flags(cmd: Cmd) -> tuple[str, ...]:
flags.append("--wait")
if cmd.path == "/billing/topup":
flags.extend(("--yes", "--no-pay", "--payment-method"))
if cmd.path == "/scans" and cmd.method == "POST":
flags.extend(
(
"--source",
"--dry-run",
"--yes",
"--show-files",
"--exclude",
"--include-hidden",
"--include-sensitive",
"--include-archives",
)
)
if cmd.path == "/billing/auto-topup" and cmd.method == "PUT":
flags.append("--no-monthly-cap")
return tuple(dict.fromkeys(flags))

12
tests/test_cloud_cli.py
View file

@ -621,9 +621,7 @@ def test_workspaces_use_switches_stored_token(
code = cloud.run_cloud(["workspaces", "use", "team one", "--json"])
assert code == 0
assert calls == [("GET", "/workspaces"), ("POST", "/workspaces/org_1/token")]
assert token_body == {
"scopes": ["scans:read", "organizations:read", "tokens:write"]
}
assert token_body == {"scopes": ["scans:read", "organizations:read", "tokens:write"]}
record = platform_cli.read_record()
assert record is not None
assert record["api_token"] == "new-token"
@ -757,15 +755,11 @@ def test_human_get_prioritizes_details_and_hides_internal_identity_fields(
assert "lossless machine-readable" in output
def test_workspace_use_accepts_list_number(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
def test_workspace_use_accepts_list_number(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
auth_path = tmp_path / "platform-auth.json"
monkeypatch.setattr(platform_cli, "AUTH_PATH", auth_path)
monkeypatch.setattr(workspaces, "AUTH_PATH", auth_path)
platform_cli.save_record(
{"api_token": "old", "scopes": ["organizations:read", "tokens:write"]}
)
platform_cli.save_record({"api_token": "old", "scopes": ["organizations:read", "tokens:write"]})
called_paths: list[str] = []
def fake_request(_method: str, path: str, **_kwargs: Any) -> FakeResponse:

230
tests/test_cloud_source_upload.py Normal file
View file

@ -0,0 +1,230 @@
"""Local-source packaging and scan upload tests."""
from __future__ import annotations
import json
import shutil
import subprocess
import zipfile
from typing import TYPE_CHECKING, Any
import pytest
from strix.interface import cloud
from strix.interface.cloud import http, source_upload
if TYPE_CHECKING:
from pathlib import Path
class FakeResponse:
def __init__(self, payload: Any, status_code: int = 200) -> None:
self.status_code = status_code
self._payload = payload
self.text = json.dumps(payload)
self.content = b""
self.ok = 200 <= status_code < 400
self.headers = {"content-type": "application/json"}
def json(self) -> Any:
return self._payload
@pytest.fixture(autouse=True)
def _token_env(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("STRIX_API_TOKEN", "test-token")
def _git_source(tmp_path: Path) -> Path:
git = shutil.which("git")
assert git is not None
subprocess.run([git, "init", "-q", str(tmp_path)], check=True) # noqa: S603
(tmp_path / "app.py").write_text("print('hello')\n", encoding="utf-8")
(tmp_path / "README.md").write_text("hello\n", encoding="utf-8")
(tmp_path / ".gitignore").write_text("ignored.log\n", encoding="utf-8")
(tmp_path / "ignored.log").write_text("ignored\n", encoding="utf-8")
subprocess.run( # noqa: S603
[git, "-C", str(tmp_path), "add", "app.py", ".gitignore"], check=True
)
return tmp_path
def test_source_defaults_are_private_and_git_aware(tmp_path: Path) -> None:
source = _git_source(tmp_path)
(source / ".hidden.py").write_text("hidden\n", encoding="utf-8")
(source / ".env").write_text("TOKEN=secret\n", encoding="utf-8")
(source / "private.pem").write_text("secret\n", encoding="utf-8")
(source / "fixture.zip").write_bytes(b"not really a zip")
(source / "node_modules").mkdir()
(source / "node_modules" / "dep.js").write_text("dep\n", encoding="utf-8")
(source / "linked.py").symlink_to(source / "app.py")
bundle = source_upload.prepare_source(
str(source),
include_hidden=False,
include_sensitive=False,
include_archives=False,
exclude=[],
)
try:
names = [item.archive_name for item in bundle.manifest.files]
assert names == ["README.md", "app.py"]
assert bundle.manifest.total_bytes > 0
assert bundle.archive_bytes <= source_upload.MAX_ARCHIVE_BYTES
assert bundle.manifest.excluded["hidden"] == 3
assert bundle.manifest.excluded["sensitive_filename"] == 1
assert bundle.manifest.excluded["nested_archive"] == 1
assert bundle.manifest.excluded["dependency_or_build_output"] == 1
assert bundle.manifest.excluded["symlink_or_non_file"] == 1
with zipfile.ZipFile(bundle.archive_path) as archive:
assert archive.namelist() == names
finally:
source_upload.remove_bundle(bundle)
def test_hidden_and_sensitive_files_need_separate_opt_ins(tmp_path: Path) -> None:
source = _git_source(tmp_path)
(source / ".env").write_text("TOKEN=secret\n", encoding="utf-8")
(source / ".github").mkdir()
(source / ".github" / "workflow.yml").write_text("name: test\n", encoding="utf-8")
hidden = source_upload.select_source(source, include_hidden=True)
hidden_names = {item.archive_name for item in hidden.files}
assert ".github/workflow.yml" in hidden_names
assert ".env" not in hidden_names
sensitive = source_upload.select_source(source, include_hidden=True, include_sensitive=True)
assert ".env" in {item.archive_name for item in sensitive.files}
assert all(
not name.startswith(".git/") for name in (item.archive_name for item in sensitive.files)
)
def test_strixignore_and_cli_excludes_are_applied(tmp_path: Path) -> None:
(tmp_path / "keep.py").write_text("keep\n", encoding="utf-8")
(tmp_path / "generated.py").write_text("generated\n", encoding="utf-8")
(tmp_path / "test_app.py").write_text("test\n", encoding="utf-8")
(tmp_path / ".strixignore").write_text("generated.py\n", encoding="utf-8")
manifest = source_upload.select_source(tmp_path, exclude=["test_*.py"])
assert [item.archive_name for item in manifest.files] == ["keep.py"]
assert manifest.excluded["user_pattern"] == 2
def test_source_limits_expanded_bytes_before_compression(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(source_upload, "MAX_TOTAL_BYTES", 5)
(tmp_path / "large.py").write_bytes(b"a" * 6)
with pytest.raises(http.CloudError, match="expanded-size limit"):
source_upload.select_source(tmp_path)
def test_source_dry_run_never_calls_the_api(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: Any
) -> None:
(tmp_path / "app.py").write_text("print('safe')\n", encoding="utf-8")
def fail_request(*_args: Any, **_kwargs: Any) -> Any:
raise AssertionError("dry-run must not make an API request")
monkeypatch.setattr(http, "request", fail_request)
assert (
cloud.run_cloud(
["scans", "start", "--source", str(tmp_path), "--dry-run", "--show-files", "--json"]
)
== 0
)
payload = json.loads(capsys.readouterr().out)
assert payload["source"]["files"] == ["app.py"]
assert payload["source"]["archive_sha256"]
def test_noninteractive_source_upload_requires_yes(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: Any
) -> None:
(tmp_path / "app.py").write_text("print('safe')\n", encoding="utf-8")
monkeypatch.setattr(
http,
"request",
lambda *_args, **_kwargs: pytest.fail("approval must happen before any API request"),
)
assert cloud.run_cloud(["scans", "start", "--source", str(tmp_path), "--json"]) == 1
assert "requires explicit approval" in capsys.readouterr().out
def test_source_upload_is_completed_and_attached_to_scan(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: Any
) -> None:
(tmp_path / "app.py").write_text("print('safe')\n", encoding="utf-8")
calls: list[tuple[str, str, dict[str, Any]]] = []
uploaded_path: Path | None = None
def fake_request(method: str, path: str, **kwargs: Any) -> FakeResponse:
calls.append((method, path, kwargs))
if path == "/uploads/request":
return FakeResponse(
{
"upload_id": "upload-1",
"signed_url": "https://storage.test/object",
"token": "signed",
}
)
if path == "/uploads/complete":
return FakeResponse({"id": "upload-1"})
if path == "/scans":
return FakeResponse({"scan_id": "scan-1", "status": "pending"})
raise AssertionError(path)
def fake_upload(_url: str, _token: str, path: Path) -> None:
nonlocal uploaded_path
uploaded_path = path
assert path.exists()
monkeypatch.setattr(http, "request", fake_request)
monkeypatch.setattr(http, "upload_file", fake_upload)
assert (
cloud.run_cloud(
["scans", "start", "--source", str(tmp_path), "--yes", "--show-files", "--json"]
)
== 0
)
payload = json.loads(capsys.readouterr().out)
assert payload["upload_id"] == "upload-1"
assert payload["scan"]["scan_id"] == "scan-1"
assert payload["source"]["files"] == ["app.py"]
scan_call = next(call for call in calls if call[1] == "/scans")
assert scan_call[2]["body"]["upload_ids"] == ["upload-1"]
assert uploaded_path is not None and not uploaded_path.exists()
def test_failed_scan_deletes_completed_source_upload(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
(tmp_path / "app.py").write_text("print('safe')\n", encoding="utf-8")
paths: list[tuple[str, str]] = []
def fake_request(method: str, path: str, **_kwargs: Any) -> FakeResponse:
paths.append((method, path))
if path == "/uploads/request":
return FakeResponse(
{
"upload_id": "upload-1",
"signed_url": "https://storage.test/object",
"token": "signed",
}
)
if path == "/uploads/complete":
return FakeResponse({"id": "upload-1"})
if path == "/scans":
return FakeResponse({"detail": "not enough credits"}, status_code=402)
if path == "/uploads/upload-1":
return FakeResponse({"ok": True})
raise AssertionError(path)
monkeypatch.setattr(http, "request", fake_request)
monkeypatch.setattr(http, "upload_file", lambda *_args, **_kwargs: None)
assert cloud.run_cloud(["scans", "start", "--source", str(tmp_path), "--yes", "--json"]) == 5
assert ("DELETE", "/uploads/upload-1") in paths

3
tests/test_completions.py
View file

@ -28,6 +28,9 @@ def test_cloud_leaf_flag_candidates_come_from_command_spec() -> None:
assert "--domain-ids" in candidates
assert "--json" in candidates
assert "--wait" in candidates
assert "--source" in candidates
assert "--dry-run" in candidates
assert "--include-hidden" in candidates
def test_boolean_completion_includes_positive_and_negative_flags() -> None: