mirror of
https://github.com/usestrix/strix.git
synced 2026-09-07 08:25:56 +00:00
fix(scope): reconcile web targets as hosts
This commit is contained in:
oyasumi
parent
00593da239
commit
011ca8ff47
28 changed files with 706 additions and 561 deletions
|
|
@ -191,12 +191,17 @@ strix view my-run-name
|
|||
strix --target ./app-directory
|
||||
|
||||
# Security review of a GitHub repository
|
||||
strix --target https://github.com/org/repo
|
||||
strix --target https://github.com/org/repo.git
|
||||
|
||||
# Black-box web application assessment
|
||||
strix --target https://your-app.com
|
||||
```
|
||||
|
||||
Web targets are host-level: paths and queries passed to `--target` are removed,
|
||||
and repeated URLs on the same host collapse to one target. Put exact starting
|
||||
endpoints in `--instruction`. Free-form text entered in the interactive start
|
||||
screen remains the task and is not split into inferred targets.
|
||||
|
||||
### API Testing (OpenAPI / Swagger / Postman)
|
||||
|
||||
Point Strix at an API contract and it tests every declared endpoint instead of
|
||||
|
|
@ -226,7 +231,7 @@ strix --target "postman://<collection-uuid>?env=<environment-uuid>"
|
|||
strix --target https://your-app.com --instruction "Perform authenticated testing using credentials: user:pass"
|
||||
|
||||
# Multi-target testing (source code + deployed app)
|
||||
strix -t https://github.com/org/app -t https://your-app.com
|
||||
strix -t https://github.com/org/app.git -t https://your-app.com
|
||||
|
||||
# Targets from a file, one target per non-empty, non-comment line
|
||||
strix --target-list ./targets.txt
|
||||
|
|
|
|||
|
|
@ -55,13 +55,13 @@ Strix accepts multiple target types:
|
|||
strix --target ./app-directory
|
||||
|
||||
# GitHub repository
|
||||
strix --target https://github.com/org/repo
|
||||
strix --target https://github.com/org/repo.git
|
||||
|
||||
# Live web application
|
||||
strix --target https://your-app.com
|
||||
|
||||
# Multiple targets (white-box testing)
|
||||
strix -t https://github.com/org/repo -t https://your-app.com
|
||||
strix -t https://github.com/org/repo.git -t https://your-app.com
|
||||
|
||||
# Targets from a file, one target per non-empty, non-comment line
|
||||
strix --target-list ./targets.txt
|
||||
|
|
|
|||
|
|
@ -6,13 +6,17 @@ description: "Command-line options for Strix"
|
|||
## Basic Usage
|
||||
|
||||
```bash
|
||||
strix (--target <target> | --target-list <path>) [options]
|
||||
strix [(--target <target> | --target-list <path>)] [options]
|
||||
```
|
||||
|
||||
## Options
|
||||
|
||||
<ParamField path="--target, -t" type="string">
|
||||
Target to test. Accepts URLs, repositories, local directories, domains, IP addresses, API spec files (OpenAPI/Swagger `.json`/`.yaml`, a Postman collection export), or a live Postman collection by id (`postman://<collection-uuid>`). Can be specified multiple times. Fresh runs require at least one target source: `--target` or `--target-list`.
|
||||
Target to test. Accepts URLs, repositories, local directories, domains, IP addresses, API spec files (OpenAPI/Swagger `.json`/`.yaml`, a Postman collection export), or a live Postman collection by id (`postman://<collection-uuid>`). Can be specified multiple times. Fresh headless runs require at least one target source: `--target` or `--target-list`.
|
||||
|
||||
Web URL targets are canonicalized to their hostname, and repeated URLs on the same host become one target. Put endpoint paths, query strings, and other starting-point details in `--instruction`; hosts explicitly named there are also in prompt-level scope. Free-form text entered on the interactive start screen is kept entirely as the task and is not split into inferred targets.
|
||||
|
||||
HTTP repository URLs ending in `.git` are recognized automatically. For a repository URL without `.git`, prefix it with `git+` (for example, `git+https://github.com/org/repo`). This explicit syntax prevents ordinary web paths from being mistaken for repositories.
|
||||
|
||||
When the target is an API spec, Strix copies it into the agent's workspace and authorizes the base URLs it declares (including those resolved from a Postman environment) as in-scope hosts - so the agent reads the contract and tests the full declared surface instead of discovering endpoints by crawling. Pair the spec with the deployed base URL (e.g. `--target ./openapi.yaml --target https://api.example.com`) so the agent has a reachable host to attack.
|
||||
|
||||
|
|
@ -139,7 +143,7 @@ strix --target https://example.com --max-budget 25 --max-turns 300
|
|||
strix -n --target ./ --scan-mode quick --scope-mode diff --diff-base origin/main
|
||||
|
||||
# Multi-target white-box testing
|
||||
strix -t https://github.com/org/app -t https://staging.example.com
|
||||
strix -t https://github.com/org/app.git -t https://staging.example.com
|
||||
|
||||
# API spec + live target (OpenAPI/Swagger file or Postman collection)
|
||||
strix -t ./openapi.yaml -t https://api.example.com
|
||||
|
|
|
|||
|
|
@ -4,6 +4,10 @@ description: "Guide Strix with custom testing instructions"
|
|||
---
|
||||
|
||||
Use instructions to provide context, credentials, or focus areas for your scan.
|
||||
Configured web targets identify hosts, not endpoints. Put exact paths and query
|
||||
strings in the instruction so they remain part of the task. Hosts explicitly
|
||||
named in the instruction and their descendant subdomains are in prompt-level
|
||||
scope.
|
||||
|
||||
## Inline Instructions
|
||||
|
||||
|
|
|
|||
|
|
@ -64,7 +64,7 @@ strix -n -t ./ --scan-mode standard --max-budget 10
|
|||
strix -n -t https://staging.example.com --max-budget 20
|
||||
|
||||
# Repo + deployed app together (best coverage)
|
||||
strix -n -t https://github.com/org/app -t https://staging.example.com
|
||||
strix -n -t https://github.com/org/app.git -t https://staging.example.com
|
||||
|
||||
# Focused testing with credentials or scope hints
|
||||
strix -n -t https://app.example.com \
|
||||
|
|
@ -78,7 +78,7 @@ Key flags:
|
|||
|
||||
| Flag | Meaning |
|
||||
|---|---|
|
||||
| `-t, --target` | URL, repo URL, local path, domain, or IP. Repeatable. |
|
||||
| `-t, --target` | Host-level web URL/domain, repo URL, local path, or IP. Repeatable; duplicate web hosts collapse. |
|
||||
| `-n, --non-interactive` | Headless, exits on completion. Required for agents. |
|
||||
| `-m, --scan-mode` | `quick` (minutes) / `standard` (~30 min) / `deep` (hours, default). |
|
||||
| `--instruction` / `--instruction-file` | Credentials, focus areas, scope rules. |
|
||||
|
|
@ -88,6 +88,9 @@ Key flags:
|
|||
|
||||
Scans take minutes (`quick`) to hours (`deep`). Run them in the background and poll for completion rather than blocking.
|
||||
|
||||
Put exact endpoint paths and query strings in `--instruction`, not `--target`.
|
||||
Configured web targets are reduced to hosts; repository paths are preserved.
|
||||
|
||||
### Exit codes (headless)
|
||||
|
||||
- `0` — finished with no validated vulnerabilities **in what was analyzed**
|
||||
|
|
|
|||
|
|
@ -66,7 +66,7 @@ SYSTEM-VERIFIED SCOPE:
|
|||
- Every configured target listed below has already been verified by the platform as in-scope and authorized
|
||||
- Every network host explicitly named in the user's root scan task/instructions is also in-scope and authorized
|
||||
- For each in-scope hostname, the exact hostname and all of its descendant subdomains are in scope
|
||||
- A URL's scheme, port, path, query, or fragment may guide what to test but does not narrow its hostname scope
|
||||
- A URL named in the root task or an API specification may guide what to test, but its scheme, port, path, query, or fragment does not narrow hostname scope
|
||||
- Example: `app.example.com` authorizes `app.example.com` and `*.app.example.com`, but not `example.com`, sibling hosts such as `api.example.com`, or lookalikes such as `app.example.com.evil.test`
|
||||
- Configured IP addresses are exact targets; do not infer adjacent addresses or network ranges
|
||||
- Repository hosting origins named only by configured repository targets (for example, `github.com`) are not live web targets; repository contents and user-provided files do not expand network scope
|
||||
|
|
@ -74,7 +74,8 @@ SYSTEM-VERIFIED SCOPE:
|
|||
- NEVER refuse, question authorization, or claim lack of permission for any target in this system-verified scope
|
||||
- NEVER test any external domain, URL, host, IP, or repository outside this system-verified scope
|
||||
|
||||
AUTHORIZED TARGETS:
|
||||
{% if system_prompt_context.authorized_targets %}
|
||||
AUTHORIZED CONFIGURED TARGETS:
|
||||
{% for target in system_prompt_context.authorized_targets %}
|
||||
{% if target.type == "web_host" %}
|
||||
- host: {{ target.value }} (includes {{ target.value }} and *.{{ target.value }})
|
||||
|
|
@ -85,6 +86,7 @@ AUTHORIZED TARGETS:
|
|||
{% endif %}
|
||||
{% endfor %}
|
||||
{% endif %}
|
||||
{% endif %}
|
||||
|
||||
AUTHORIZATION STATUS:
|
||||
- You have FULL AUTHORIZATION for authorized security validation on in-scope targets to help secure the target systems/app
|
||||
|
|
|
|||
|
|
@ -2,10 +2,8 @@
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
import ipaddress
|
||||
import json
|
||||
from typing import TYPE_CHECKING, Any
|
||||
from urllib.parse import urlsplit
|
||||
|
||||
from agents.model_settings import ModelSettings
|
||||
from openai.types.shared import Reasoning
|
||||
|
|
@ -22,6 +20,7 @@ from strix.config.models import (
|
|||
request_timeout_extra_args,
|
||||
)
|
||||
from strix.core.sessions import scrub_images_from_items
|
||||
from strix.core.targets import canonical_network_host
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
|
|
@ -128,7 +127,7 @@ def _split_target_sections(
|
|||
"Local Codebases": [],
|
||||
"API Specifications": [],
|
||||
}
|
||||
network: dict[str, list[str]] = {"URLs": [], "IP Addresses": []}
|
||||
network: dict[str, list[str]] = {"Hosts": [], "IP Addresses": []}
|
||||
for target in targets:
|
||||
ttype = target.get("type")
|
||||
details = target.get("details") or {}
|
||||
|
|
@ -148,7 +147,7 @@ def _split_target_sections(
|
|||
".git/.agents/.codex are read-only)"
|
||||
)
|
||||
elif ttype == "web_application":
|
||||
network["URLs"].append(f"- {details.get('target_url', '')}")
|
||||
network["Hosts"].append(f"- {details.get('target_host', '')}")
|
||||
elif ttype == "ip_address":
|
||||
network["IP Addresses"].append(f"- {details.get('target_ip', '')}")
|
||||
elif ttype == "api_spec":
|
||||
|
|
@ -208,16 +207,9 @@ def build_root_task(scan_config: dict[str, Any]) -> str:
|
|||
)
|
||||
|
||||
|
||||
def _network_scope_target(value: str) -> tuple[str, str]:
|
||||
"""Reduce a web URL to its hostname-level prompt scope."""
|
||||
hostname = (urlsplit(value).hostname or "").rstrip(".").lower()
|
||||
if not hostname:
|
||||
return "web_application", value
|
||||
try:
|
||||
ipaddress.ip_address(hostname)
|
||||
except ValueError:
|
||||
return "web_host", hostname
|
||||
return "ip_address", hostname
|
||||
def _scope_target_from_url(value: str) -> tuple[str, str]:
|
||||
"""Extract host-level prompt scope from an API-spec base URL."""
|
||||
return canonical_network_host(value)
|
||||
|
||||
|
||||
def build_scope_context(scan_config: dict[str, Any]) -> dict[str, Any]:
|
||||
|
|
@ -235,7 +227,7 @@ def build_scope_context(scan_config: dict[str, Any]) -> dict[str, Any]:
|
|||
value_keys = {
|
||||
"repository": "target_repo",
|
||||
"local_code": "target_path",
|
||||
"web_application": "target_url",
|
||||
"web_application": "target_host",
|
||||
"ip_address": "target_ip",
|
||||
"api_spec": "target_spec",
|
||||
}
|
||||
|
|
@ -248,7 +240,7 @@ def build_scope_context(scan_config: dict[str, Any]) -> dict[str, Any]:
|
|||
workspace_subdir = details.get("workspace_subdir")
|
||||
workspace_path = f"/workspace/{workspace_subdir}" if workspace_subdir else ""
|
||||
if ttype == "web_application":
|
||||
scope_type, scope_value = _network_scope_target(str(value or ""))
|
||||
scope_type, scope_value = canonical_network_host(str(value or ""))
|
||||
add_authorized(scope_type, scope_value)
|
||||
else:
|
||||
add_authorized(str(ttype), str(value or ""), workspace_path)
|
||||
|
|
@ -257,7 +249,7 @@ def build_scope_context(scan_config: dict[str, Any]) -> dict[str, Any]:
|
|||
# so the agent can exercise every endpoint without expanding scope.
|
||||
if ttype == "api_spec":
|
||||
for base_url in details.get("base_urls") or []:
|
||||
scope_type, scope_value = _network_scope_target(str(base_url))
|
||||
scope_type, scope_value = _scope_target_from_url(str(base_url))
|
||||
add_authorized(scope_type, scope_value)
|
||||
|
||||
return {
|
||||
|
|
|
|||
31
strix/core/targets.py
Normal file
31
strix/core/targets.py
Normal file
|
|
@ -0,0 +1,31 @@
|
|||
"""Canonical host identities shared by target ingestion and prompt scope."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import ipaddress
|
||||
import re
|
||||
from urllib.parse import urlsplit
|
||||
|
||||
|
||||
_DNS_LABEL = re.compile(r"^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$")
|
||||
|
||||
|
||||
def canonical_network_host(value: str) -> tuple[str, str]:
|
||||
"""Return ``(web_host|ip_address, canonical value)`` for a network input."""
|
||||
parsed = urlsplit(value if "://" in value else f"//{value}")
|
||||
hostname = (parsed.hostname or "").rstrip(".").lower()
|
||||
if not hostname:
|
||||
raise ValueError(f"Network target '{value}' does not contain a valid host")
|
||||
|
||||
try:
|
||||
return "ip_address", str(ipaddress.ip_address(hostname))
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
try:
|
||||
hostname = hostname.encode("idna").decode("ascii")
|
||||
except UnicodeError as exc:
|
||||
raise ValueError(f"Network target '{value}' contains an invalid host") from exc
|
||||
if len(hostname) > 253 or any(not _DNS_LABEL.fullmatch(label) for label in hostname.split(".")):
|
||||
raise ValueError(f"Network target '{value}' contains an invalid host")
|
||||
return "web_host", hostname
|
||||
|
|
@ -9,12 +9,20 @@ from pathlib import Path
|
|||
from strix.config import apply_config_override
|
||||
from strix.config.settings import DEFAULT_MAX_TURNS
|
||||
from strix.core.paths import run_dir_for, runtime_state_dir
|
||||
from strix.interface.scan_setup import attach_workspace_mount, build_targets_info
|
||||
from strix.interface.scan_setup import (
|
||||
HOST_GATEWAY_HOSTNAME,
|
||||
attach_workspace_mount,
|
||||
build_targets_info,
|
||||
)
|
||||
from strix.interface.update_check import self_update
|
||||
from strix.interface.utils import (
|
||||
canonicalize_targets_info,
|
||||
check_mountable_dir,
|
||||
collect_local_sources,
|
||||
dedupe_targets,
|
||||
resolve_workspace_files,
|
||||
restore_staged_api_specs,
|
||||
rewrite_localhost_targets,
|
||||
validate_config_file,
|
||||
)
|
||||
|
||||
|
|
@ -60,7 +68,7 @@ Examples:
|
|||
strix --target https://example.com
|
||||
|
||||
# GitHub repository analysis
|
||||
strix --target https://github.com/user/repo
|
||||
strix --target https://github.com/user/repo.git
|
||||
strix --target git@github.com:user/repo.git
|
||||
|
||||
# Local code analysis
|
||||
|
|
@ -81,7 +89,7 @@ Examples:
|
|||
strix --target 192.168.1.42
|
||||
|
||||
# Multiple targets (e.g., white-box testing with source and deployed app)
|
||||
strix --target https://github.com/user/repo --target https://example.com
|
||||
strix --target https://github.com/user/repo.git --target https://example.com
|
||||
strix --target ./my-project --target https://staging.example.com --target https://prod.example.com
|
||||
|
||||
# Targets from a file, one target per non-empty, non-comment line
|
||||
|
|
@ -123,9 +131,10 @@ Examples:
|
|||
help="Target to test: URL, repository, local directory path, domain name, IP address, "
|
||||
"an API spec file (OpenAPI/Swagger .json/.yaml or a Postman collection export), or a "
|
||||
"Postman collection by id (postman://<collection-uuid>[?env=<environment-uuid>], needs "
|
||||
"POSTMAN_API_KEY). Local directories are mounted into the sandbox writable. "
|
||||
"POSTMAN_API_KEY). Web URLs are reduced to their host; put endpoint paths and queries "
|
||||
"in --instruction. Local directories are mounted into the sandbox writable. "
|
||||
"Can be specified multiple times for multi-target scans. "
|
||||
"Fresh runs require --target or --target-list.",
|
||||
"Fresh headless runs require --target or --target-list.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--target-list",
|
||||
|
|
@ -320,8 +329,7 @@ Examples:
|
|||
"(or use --resume <run_name> to continue a prior scan)"
|
||||
)
|
||||
# Interactive launch with no target: open the normal TUI on its
|
||||
# start screen, where the user gives a target or a bare prompt
|
||||
# before the scan starts.
|
||||
# start screen, where the user gives the task before the scan starts.
|
||||
args.needs_setup = True
|
||||
return args
|
||||
|
||||
|
|
@ -335,7 +343,7 @@ Examples:
|
|||
|
||||
def _load_resume_state(args: argparse.Namespace, parser: argparse.ArgumentParser) -> None:
|
||||
"""Populate ``args.targets_info`` and friends from a prior run's run.json."""
|
||||
from strix.report.writer import read_run_record
|
||||
from strix.report.writer import read_run_record, write_run_record
|
||||
|
||||
run_dir = run_dir_for(args.resume)
|
||||
state_path = run_dir / "run.json"
|
||||
|
|
@ -346,10 +354,19 @@ def _load_resume_state(args: argparse.Namespace, parser: argparse.ArgumentParser
|
|||
)
|
||||
try:
|
||||
state = read_run_record(run_dir)
|
||||
except RuntimeError as exc:
|
||||
except (RuntimeError, TypeError) as exc:
|
||||
parser.error(f"--resume {args.resume}: run.json unreadable: {exc}")
|
||||
|
||||
args.targets_info = state.get("targets_info") or []
|
||||
persisted_targets = state.get("targets_info") or []
|
||||
try:
|
||||
args.targets_info = canonicalize_targets_info(persisted_targets)
|
||||
rewrite_localhost_targets(args.targets_info, HOST_GATEWAY_HOSTNAME)
|
||||
args.targets_info = dedupe_targets(args.targets_info)
|
||||
except (TypeError, ValueError) as exc:
|
||||
parser.error(f"--resume {args.resume}: invalid persisted target: {exc}")
|
||||
if args.targets_info != persisted_targets:
|
||||
state["targets_info"] = args.targets_info
|
||||
write_run_record(run_dir, state)
|
||||
# A target-less run has no targets_info at all. It is driven by its
|
||||
# instruction, over a mounted working directory or over nothing when the
|
||||
# mount was declined, so either of those is enough to resume it.
|
||||
|
|
@ -384,6 +401,10 @@ def _load_resume_state(args: argparse.Namespace, parser: argparse.ArgumentParser
|
|||
if not getattr(args, "user_instruction", None):
|
||||
args.user_instruction = state.get("user_instruction") or None
|
||||
args.local_sources = collect_local_sources(args.targets_info)
|
||||
try:
|
||||
args.local_sources.extend(restore_staged_api_specs(args.targets_info, str(args.resume)))
|
||||
except ValueError as exc:
|
||||
parser.error(f"--resume {args.resume}: could not restore API specification: {exc}")
|
||||
# Remount the workspace the run was started with. The user already confirmed
|
||||
# this directory, so the target mount guard does not apply to it; it only has
|
||||
# to still be there.
|
||||
|
|
|
|||
|
|
@ -16,11 +16,13 @@ from typing import TYPE_CHECKING, Any
|
|||
|
||||
from strix.config import Settings, codex, load_settings
|
||||
from strix.core.paths import run_dir_for
|
||||
from strix.core.targets import canonical_network_host
|
||||
from strix.interface.utils import (
|
||||
assign_workspace_subdirs,
|
||||
canonicalize_targets_info,
|
||||
clone_repository,
|
||||
collect_local_sources,
|
||||
dedupe_local_targets,
|
||||
dedupe_targets,
|
||||
derive_local_base_name,
|
||||
generate_run_name,
|
||||
infer_target_type,
|
||||
|
|
@ -117,6 +119,10 @@ def build_targets_info(args: argparse.Namespace) -> None:
|
|||
|
||||
if target_type == "local_code":
|
||||
display_target = target_dict.get("target_path", target)
|
||||
elif target_type == "web_application":
|
||||
display_target = target_dict["target_host"]
|
||||
elif target_type == "ip_address":
|
||||
display_target = target_dict["target_ip"]
|
||||
else:
|
||||
display_target = target
|
||||
|
||||
|
|
@ -127,10 +133,10 @@ def build_targets_info(args: argparse.Namespace) -> None:
|
|||
{"type": target_type, "details": target_dict, "original": display_target}
|
||||
)
|
||||
|
||||
args.targets_info = dedupe_local_targets(args.targets_info)
|
||||
|
||||
assign_workspace_subdirs(args.targets_info)
|
||||
args.targets_info = canonicalize_targets_info(args.targets_info)
|
||||
rewrite_localhost_targets(args.targets_info, HOST_GATEWAY_HOSTNAME)
|
||||
args.targets_info = dedupe_targets(args.targets_info)
|
||||
assign_workspace_subdirs(args.targets_info)
|
||||
|
||||
|
||||
def _resolve_api_spec(target: str, details: dict[str, Any]) -> None:
|
||||
|
|
@ -154,8 +160,12 @@ def _resolve_api_spec(target: str, details: dict[str, Any]) -> None:
|
|||
raw = load_spec(str(details["target_spec"]))
|
||||
extra_variables = None
|
||||
base_urls = spec_base_urls(raw, extra_variables=extra_variables)
|
||||
for base_url in base_urls:
|
||||
canonical_network_host(base_url)
|
||||
except SpecParseError as exc:
|
||||
raise ValueError(f"Invalid API spec '{target}': {exc}") from None
|
||||
except ValueError as exc:
|
||||
raise ValueError(f"Invalid API spec '{target}': {exc}") from None
|
||||
|
||||
details["spec_title"] = spec_title(raw)
|
||||
details["base_urls"] = base_urls
|
||||
|
|
@ -164,9 +174,8 @@ def _resolve_api_spec(target: str, details: dict[str, Any]) -> None:
|
|||
def prepare_run(args: argparse.Namespace) -> None:
|
||||
"""Resolve the run name, clone repos, compute diff-scope, and persist state.
|
||||
|
||||
Shared by the CLI startup path and the interactive TUI setup phase (once the
|
||||
user has supplied a target via ``/target``). Mutates *args* in place and
|
||||
raises :class:`ValueError` on any preparation failure.
|
||||
Shared by the CLI startup path and the interactive TUI setup phase. Mutates
|
||||
*args* in place and raises :class:`ValueError` on any preparation failure.
|
||||
"""
|
||||
args.run_name = args.resume or generate_run_name(args.targets_info)
|
||||
|
||||
|
|
|
|||
|
|
@ -263,7 +263,6 @@ class TuiController:
|
|||
|
||||
async def handle(self, command: str, payload: dict[str, Any]) -> dict[str, Any]:
|
||||
handlers = {
|
||||
"setup.add_target": self._add_target,
|
||||
"setup.set_instruction": self._set_instruction,
|
||||
"setup.start": self._start,
|
||||
"setup.confirm_mount": self._confirm_mount,
|
||||
|
|
@ -279,13 +278,6 @@ class TuiController:
|
|||
self.notify_changed()
|
||||
return result
|
||||
|
||||
async def _add_target(self, payload: dict[str, Any]) -> dict[str, Any]:
|
||||
self._require_setup_mutable()
|
||||
target = self._required_string(payload, "target")
|
||||
if target not in self.targets:
|
||||
self.targets.append(target)
|
||||
return {"target": target, "total": len(self.targets)}
|
||||
|
||||
async def _set_instruction(self, payload: dict[str, Any]) -> dict[str, Any]:
|
||||
self._require_setup_mutable()
|
||||
instruction = payload.get("instruction", "")
|
||||
|
|
|
|||
|
|
@ -245,7 +245,7 @@ func TestSetupUsesDedicatedStartScreen(t *testing.T) {
|
|||
SetupMode: true,
|
||||
ScanState: "setup",
|
||||
Model: "gpt-5.4",
|
||||
Targets: []string{"/workspace/source", "https://example.com"},
|
||||
Targets: []string{"/workspace/source", "example.com"},
|
||||
Instruction: "focus on access control",
|
||||
ScanMode: "quick",
|
||||
MaxBudgetUSD: floatPointer(12.5),
|
||||
|
|
@ -260,7 +260,7 @@ func TestSetupUsesDedicatedStartScreen(t *testing.T) {
|
|||
for _, want := range []string{
|
||||
"gpt-5.4",
|
||||
"/workspace/source",
|
||||
"https://example.com",
|
||||
"example.com",
|
||||
} {
|
||||
if !strings.Contains(view, want) {
|
||||
t.Fatalf("start screen is missing %q: %s", want, view)
|
||||
|
|
@ -342,9 +342,9 @@ func TestLeadingSlashIsPromptTextNotACommand(t *testing.T) {
|
|||
if !contains(types, "setup.start") {
|
||||
t.Fatalf("a slash-leading prompt did not launch a scan: %v", types)
|
||||
}
|
||||
// The path is read as a target and the sentence as the instruction.
|
||||
if !contains(types, "setup.add_target") || !contains(types, "setup.set_instruction") {
|
||||
t.Fatalf("slash-leading prompt was not split into target and instruction: %v", types)
|
||||
// The entire value remains prompt text; no token is promoted to a target.
|
||||
if contains(types, "setup.add_target") || !contains(types, "setup.set_instruction") {
|
||||
t.Fatalf("slash-leading prompt was not preserved as instruction text: %v", types)
|
||||
}
|
||||
for _, line := range result.setupLog {
|
||||
if strings.Contains(ansi.Strip(line), "Unknown command") {
|
||||
|
|
|
|||
|
|
@ -2,8 +2,6 @@ package app
|
|||
|
||||
import (
|
||||
"fmt"
|
||||
"net"
|
||||
"regexp"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
|
|
@ -26,30 +24,16 @@ func (m Model) submit(value string) (tea.Model, tea.Cmd) {
|
|||
return m, send(m.client, "agent.send_message", map[string]any{"agent_id": m.snapshot.Agents[m.selectedAgent].ID, "message": value})
|
||||
}
|
||||
|
||||
// submitSetupPrompt handles free text the way a coding agent's prompt does:
|
||||
// anything that looks like a target is added, the rest becomes the scan
|
||||
// instruction, and the prompt alone is enough to launch. With no target, the
|
||||
// backend scans the current working directory.
|
||||
// submitSetupPrompt treats free text as the task verbatim. Target-looking words
|
||||
// are not promoted into configured targets: hosts named in the task receive
|
||||
// prompt-level scope, while configured targets come only from explicit flags.
|
||||
func (m *Model) submitSetupPrompt(value string) (tea.Model, tea.Cmd) {
|
||||
var commands []tea.Cmd
|
||||
fields := strings.Fields(value)
|
||||
targets := 0
|
||||
for _, field := range fields {
|
||||
token := strings.Trim(field, ",;")
|
||||
if !looksLikeTarget(token) || m.hasTarget(token) {
|
||||
continue
|
||||
}
|
||||
targets++
|
||||
commands = append(commands, send(m.client, "setup.add_target", map[string]any{"target": token}))
|
||||
}
|
||||
if len(fields) > targets {
|
||||
commands = append(commands, send(m.client, "setup.set_instruction", map[string]any{"instruction": value}))
|
||||
}
|
||||
commands := []tea.Cmd{send(m.client, "setup.set_instruction", map[string]any{"instruction": value})}
|
||||
// With a target, verify the model connection before the scan commits to it.
|
||||
// A bare prompt launches optimistically, like a coding agent, and mounts the
|
||||
// working directory - the backend asks about that from the live view, so the
|
||||
// prompt is held here in case it is declined.
|
||||
verify := targets > 0 || len(m.snapshot.Targets) > 0
|
||||
verify := len(m.snapshot.Targets) > 0
|
||||
payload := map[string]any{"verify": verify}
|
||||
if verify {
|
||||
m.setupMsg("Verifying model connection...", render.Col(amber))
|
||||
|
|
@ -74,54 +58,6 @@ func (m *Model) answerMountConfirmation(approved bool) tea.Cmd {
|
|||
return send(m.client, "setup.confirm_mount", map[string]any{"approved": approved})
|
||||
}
|
||||
|
||||
func (m Model) hasTarget(candidate string) bool {
|
||||
for _, target := range m.snapshot.Targets {
|
||||
if target == candidate {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// looksLikeTarget reports whether a whitespace-delimited token names something
|
||||
// scannable: a URL, repo, filesystem path, domain, or IP address.
|
||||
func looksLikeTarget(token string) bool {
|
||||
if token == "" {
|
||||
return false
|
||||
}
|
||||
if strings.Contains(token, "://") || strings.HasSuffix(token, ".git") {
|
||||
return true
|
||||
}
|
||||
if strings.HasPrefix(token, "/") || strings.HasPrefix(token, "./") || strings.HasPrefix(token, "~/") || strings.HasPrefix(token, "../") {
|
||||
return true
|
||||
}
|
||||
if ip := net.ParseIP(token); ip != nil {
|
||||
return true
|
||||
}
|
||||
host := token
|
||||
if at := strings.LastIndex(host, "@"); at >= 0 {
|
||||
host = host[at+1:]
|
||||
}
|
||||
host = strings.SplitN(host, "/", 2)[0]
|
||||
host = strings.SplitN(host, ":", 2)[0]
|
||||
if !domainPattern.MatchString(host) {
|
||||
return false
|
||||
}
|
||||
tld := host[strings.LastIndex(host, ".")+1:]
|
||||
return len(tld) >= 2 && !isNumeric(tld)
|
||||
}
|
||||
|
||||
var domainPattern = regexp.MustCompile(`^([a-zA-Z0-9]([a-zA-Z0-9-]*[a-zA-Z0-9])?\.)+[a-zA-Z0-9]{2,}$`)
|
||||
|
||||
func isNumeric(value string) bool {
|
||||
for _, char := range value {
|
||||
if char < '0' || char > '9' {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// statusVisible mirrors #agent_status_display: shown only when an agent is
|
||||
// selected during a scan; hidden (display:none) in setup mode.
|
||||
func (m Model) statusVisible() bool {
|
||||
|
|
|
|||
|
|
@ -258,34 +258,39 @@ func TestMountConfirmationAnswers(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
// A prompt that names a target adds it and launches.
|
||||
func TestSetupPromptWithTargetLaunches(t *testing.T) {
|
||||
// URLs in free-form setup text remain task text rather than becoming targets.
|
||||
func TestSetupPromptWithURLsLaunchesAsInstructionOnly(t *testing.T) {
|
||||
connection := &recordingConn{}
|
||||
model := New(&Client{conn: connection})
|
||||
model.snapshot = protocol.Snapshot{SetupMode: true}
|
||||
|
||||
_, cmd := model.submit("https://juice-shop.example.com hit the coupon endpoint")
|
||||
prompt := "test https://example.com/search?q=x and https://example.com/blog/"
|
||||
updated, cmd := model.submit(prompt)
|
||||
model = updated.(Model)
|
||||
envelopes := drainCommands(t, cmd, connection)
|
||||
types := commandTypes(envelopes)
|
||||
|
||||
for _, want := range []string{"setup.add_target", "setup.set_instruction", "setup.start"} {
|
||||
for _, want := range []string{"setup.set_instruction", "setup.start"} {
|
||||
if !contains(types, want) {
|
||||
t.Fatalf("missing %s in %v", want, types)
|
||||
}
|
||||
}
|
||||
// A named target keeps the upfront model check.
|
||||
if verify, found := startVerify(t, envelopes); !found || !verify {
|
||||
t.Fatalf("targeted prompt should launch with verify=true, got verify=%v found=%v", verify, found)
|
||||
if contains(types, "setup.add_target") {
|
||||
t.Fatalf("free-form URLs were promoted into targets: %v", types)
|
||||
}
|
||||
if verify, found := startVerify(t, envelopes); !found || verify {
|
||||
t.Fatalf("instruction-only prompt should launch with verify=false, got verify=%v found=%v", verify, found)
|
||||
}
|
||||
if mount, found := startPayloadFlag(t, envelopes, "mount_working_dir"); !found || !mount {
|
||||
t.Fatalf("instruction-only prompt did not request mount choice: mount=%v found=%v", mount, found)
|
||||
}
|
||||
// The target and instruction must reach the backend before setup.start
|
||||
// closes the setup guard.
|
||||
start := firstIndex(types, "setup.start")
|
||||
if target := lastIndex(types, "setup.add_target"); start < target {
|
||||
t.Fatalf("setup.start (%d) must come after setup.add_target (%d): %v", start, target, types)
|
||||
}
|
||||
if instr := lastIndex(types, "setup.set_instruction"); start < instr {
|
||||
t.Fatalf("setup.start (%d) must come after setup.set_instruction (%d): %v", start, instr, types)
|
||||
}
|
||||
if model.pendingPrompt != prompt {
|
||||
t.Fatalf("prompt was not preserved verbatim: %q", model.pendingPrompt)
|
||||
}
|
||||
}
|
||||
|
||||
// The prompt's buttons are buttons: clicking Cancel has to answer the backend,
|
||||
|
|
|
|||
|
|
@ -18,7 +18,6 @@ from strix.core.agents import AgentCoordinator
|
|||
from strix.core.hooks import BudgetExceededError
|
||||
from strix.core.runner import run_strix_scan
|
||||
from strix.interface.scan_setup import (
|
||||
build_targets_info,
|
||||
preflight_model_connection,
|
||||
prepare_run,
|
||||
telemetry_start,
|
||||
|
|
@ -115,12 +114,6 @@ class GoTuiRuntime:
|
|||
candidate.max_turns = self.controller.max_turns
|
||||
candidate.scope_mode = self.controller.scope_mode
|
||||
candidate.diff_base = self.controller.diff_base
|
||||
existing_targets = [
|
||||
str(target["original"])
|
||||
for target in candidate.targets_info
|
||||
if isinstance(target, dict) and target.get("original")
|
||||
]
|
||||
targets_changed = self.controller.targets != existing_targets
|
||||
model = (load_settings().llm.model or "").strip()
|
||||
# A bare prompt launches optimistically: it skips the network preflight
|
||||
# and lets any model error surface once the agent starts, like a coding
|
||||
|
|
@ -134,12 +127,6 @@ class GoTuiRuntime:
|
|||
# A confirmed target-less launch mounts the working directory for the
|
||||
# agent to work in, without making it a scan target.
|
||||
candidate.workspace_mount = self.controller.workspace_mount
|
||||
if targets_changed:
|
||||
# Rebuild the full typed set so path canonicalization and local
|
||||
# deduplication match the CLI.
|
||||
candidate.target = list(self.controller.targets)
|
||||
candidate.target_list = []
|
||||
build_targets_info(candidate)
|
||||
prepare_run(candidate)
|
||||
telemetry_start(candidate)
|
||||
|
||||
|
|
|
|||
|
|
@ -14,7 +14,6 @@ from typing import Any
|
|||
from urllib.parse import parse_qs, urlparse
|
||||
|
||||
import docker
|
||||
import requests
|
||||
from docker.errors import DockerException, ImageNotFound
|
||||
from rich.console import Console
|
||||
from rich.panel import Panel
|
||||
|
|
@ -22,6 +21,8 @@ from rich.text import Text
|
|||
|
||||
from strix.config import load_settings
|
||||
from strix.core.inputs import build_scope_target_labels
|
||||
from strix.core.paths import run_dir_for
|
||||
from strix.core.targets import canonical_network_host
|
||||
from strix.utils.api_spec import detect_spec_format
|
||||
|
||||
|
||||
|
|
@ -34,7 +35,9 @@ def build_target_summary_text(targets_info: list[dict[str, Any]]) -> Text:
|
|||
target_text = Text()
|
||||
target_text.append("Target", style="dim")
|
||||
target_text.append(" ")
|
||||
if len(labels) == 1:
|
||||
if not labels:
|
||||
target_text.append("task-defined scope", style="bold white")
|
||||
elif len(labels) == 1:
|
||||
target_text.append(labels[0], style="bold white")
|
||||
else:
|
||||
target_text.append(f"{len(labels)} targets", style="bold white")
|
||||
|
|
@ -497,12 +500,7 @@ def _derive_target_label_for_run_name(targets_info: list[dict[str, Any]] | None)
|
|||
original = first.get("original", "") or ""
|
||||
|
||||
if target_type == "web_application":
|
||||
url = details.get("target_url", original)
|
||||
try:
|
||||
parsed = urlparse(url)
|
||||
return str(parsed.netloc or parsed.path or url)
|
||||
except Exception:
|
||||
return str(url)
|
||||
return str(details.get("target_host", original) or original)
|
||||
|
||||
if target_type == "repository":
|
||||
repo = details.get("target_repo", original)
|
||||
|
|
@ -1137,15 +1135,31 @@ def resolve_diff_scope_context(
|
|||
)
|
||||
|
||||
|
||||
def _validated_repository_target(value: str) -> str:
|
||||
"""Validate repository transport text without discarding its path."""
|
||||
if any(char.isspace() or ord(char) < 0x20 or ord(char) == 0x7F for char in value):
|
||||
raise ValueError("Repository targets cannot contain whitespace or control characters")
|
||||
if value.startswith("git@"):
|
||||
host, separator, _path = value[4:].partition(":")
|
||||
if not separator:
|
||||
raise ValueError("SSH repository targets must include a host and path")
|
||||
canonical_network_host(host)
|
||||
elif urlparse(value).scheme in {"git", "http", "https"}:
|
||||
canonical_network_host(value)
|
||||
return value
|
||||
|
||||
|
||||
def _is_http_git_repo(url: str) -> bool:
|
||||
check_url = f"{url.rstrip('/')}/info/refs?service=git-upload-pack"
|
||||
try:
|
||||
with requests.get(check_url, headers={"User-Agent": "git/2.43.0"}, timeout=10) as resp:
|
||||
if resp.status_code >= 400:
|
||||
return resp.status_code == 401
|
||||
return "x-git-upload-pack-advertisement" in resp.headers.get("Content-Type", "")
|
||||
except (requests.RequestException, ValueError):
|
||||
return False
|
||||
"""Classify deterministic repository URL forms without probing the target."""
|
||||
return urlparse(url).path.rstrip("/").endswith(".git")
|
||||
|
||||
|
||||
def _canonical_network_target(value: str) -> tuple[str, dict[str, str]]:
|
||||
"""Reduce a web input to one canonical host or exact IP target."""
|
||||
scope_type, canonical = canonical_network_host(value)
|
||||
if scope_type == "ip_address":
|
||||
return scope_type, {"target_ip": canonical}
|
||||
return "web_application", {"target_host": canonical}
|
||||
|
||||
|
||||
def infer_target_type(target: str) -> tuple[str, dict[str, str]]: # noqa: PLR0911
|
||||
|
|
@ -1155,10 +1169,14 @@ def infer_target_type(target: str) -> tuple[str, dict[str, str]]: # noqa: PLR09
|
|||
target = target.strip()
|
||||
|
||||
if target.startswith("git@"):
|
||||
return "repository", {"target_repo": target}
|
||||
return "repository", {"target_repo": _validated_repository_target(target)}
|
||||
|
||||
if target.startswith("git://"):
|
||||
return "repository", {"target_repo": target}
|
||||
return "repository", {"target_repo": _validated_repository_target(target)}
|
||||
|
||||
if target.startswith(("git+http://", "git+https://")):
|
||||
repository = target.removeprefix("git+")
|
||||
return "repository", {"target_repo": _validated_repository_target(repository)}
|
||||
|
||||
parsed = urlparse(target)
|
||||
if parsed.scheme == "postman":
|
||||
|
|
@ -1180,16 +1198,9 @@ def infer_target_type(target: str) -> tuple[str, dict[str, str]]: # noqa: PLR09
|
|||
return "api_spec", details
|
||||
|
||||
if parsed.scheme in ("http", "https"):
|
||||
if parsed.username or parsed.password:
|
||||
return "repository", {"target_repo": target}
|
||||
if parsed.path.rstrip("/").endswith(".git"):
|
||||
return "repository", {"target_repo": target}
|
||||
if parsed.query or parsed.fragment:
|
||||
return "web_application", {"target_url": target}
|
||||
path_segments = [s for s in parsed.path.split("/") if s]
|
||||
if len(path_segments) >= 2 and _is_http_git_repo(target):
|
||||
return "repository", {"target_repo": target}
|
||||
return "web_application", {"target_url": target}
|
||||
if _is_http_git_repo(target):
|
||||
return "repository", {"target_repo": _validated_repository_target(target)}
|
||||
return _canonical_network_target(target)
|
||||
|
||||
try:
|
||||
ip_obj = ipaddress.ip_address(target)
|
||||
|
|
@ -1215,26 +1226,25 @@ def infer_target_type(target: str) -> tuple[str, dict[str, str]]: # noqa: PLR09
|
|||
raise ValueError(f"Invalid path: {target} - {e!s}") from e
|
||||
|
||||
if target.endswith(".git"):
|
||||
return "repository", {"target_repo": target}
|
||||
return "repository", {"target_repo": _validated_repository_target(target)}
|
||||
|
||||
if "/" in target:
|
||||
host_part, _, path_part = target.partition("/")
|
||||
if "." in host_part and not host_part.startswith(".") and path_part:
|
||||
full_url = f"https://{target}"
|
||||
if _is_http_git_repo(full_url):
|
||||
return "repository", {"target_repo": full_url}
|
||||
return "web_application", {"target_url": full_url}
|
||||
return "repository", {"target_repo": _validated_repository_target(full_url)}
|
||||
return _canonical_network_target(full_url)
|
||||
|
||||
if "." in target and "/" not in target and not target.startswith("."):
|
||||
parts = target.split(".")
|
||||
if len(parts) >= 2 and all(p and p.strip() for p in parts):
|
||||
return "web_application", {"target_url": f"https://{target}"}
|
||||
return _canonical_network_target(target)
|
||||
|
||||
raise ValueError(
|
||||
f"Invalid target: {target}\n"
|
||||
"Target must be one of:\n"
|
||||
"- A valid URL (http:// or https://)\n"
|
||||
"- A Git repository URL (https://host/org/repo or git@host:org/repo.git)\n"
|
||||
"- A Git repository URL (https://host/org/repo.git, "
|
||||
"git+https://host/org/repo, or git@host:org/repo.git)\n"
|
||||
"- A local directory path\n"
|
||||
"- An API spec file (OpenAPI/Swagger .json/.yaml or a Postman collection)\n"
|
||||
"- A Postman collection by id (postman://<collection-uid>[?env=<environment-uid>], "
|
||||
|
|
@ -1457,17 +1467,66 @@ def check_mountable_dir(path: Path) -> None:
|
|||
)
|
||||
|
||||
|
||||
def dedupe_local_targets(targets_info: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||||
def canonicalize_targets_info(targets_info: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||||
"""Migrate target records to the canonical host-level web schema."""
|
||||
canonical: list[dict[str, Any]] = []
|
||||
for target in targets_info:
|
||||
if not isinstance(target, dict):
|
||||
raise TypeError("target records must be objects")
|
||||
raw_details = target.get("details") or {}
|
||||
if not isinstance(raw_details, dict):
|
||||
raise TypeError("target details must be an object")
|
||||
details = dict(raw_details)
|
||||
target_type = target.get("type")
|
||||
if target_type == "repository":
|
||||
repository = _validated_repository_target(
|
||||
str(details.get("target_repo") or target.get("original") or "")
|
||||
)
|
||||
details["target_repo"] = repository
|
||||
canonical.append({**target, "details": details, "original": repository})
|
||||
continue
|
||||
if target_type not in {"web_application", "ip_address"}:
|
||||
canonical.append({**target, "details": details})
|
||||
continue
|
||||
value = str(
|
||||
details.get("target_host")
|
||||
or details.get("target_ip")
|
||||
or details.get("target_url")
|
||||
or target.get("original")
|
||||
or ""
|
||||
)
|
||||
canonical_type, network_details = _canonical_network_target(value)
|
||||
details.pop("target_url", None)
|
||||
details.pop("target_host", None)
|
||||
details.pop("target_ip", None)
|
||||
details.update(network_details)
|
||||
normalized = {**target, "type": canonical_type, "details": details}
|
||||
normalized["original"] = next(iter(network_details.values()))
|
||||
canonical.append(normalized)
|
||||
return canonical
|
||||
|
||||
|
||||
def dedupe_targets(targets_info: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||||
"""Deduplicate canonical host, IP, and local-directory targets."""
|
||||
result: list[dict[str, Any]] = []
|
||||
seen_paths: set[str] = set()
|
||||
seen: set[tuple[str, str]] = set()
|
||||
for target in targets_info:
|
||||
details = target.get("details") or {}
|
||||
path = details.get("target_path")
|
||||
if target.get("type") != "local_code" or not path:
|
||||
if not isinstance(details, dict):
|
||||
raise TypeError("target details must be an object")
|
||||
target_type = str(target.get("type") or "")
|
||||
identity_keys = {
|
||||
"web_application": "target_host",
|
||||
"ip_address": "target_ip",
|
||||
"local_code": "target_path",
|
||||
}
|
||||
identity = str(details.get(identity_keys.get(target_type, "")) or "")
|
||||
if not identity:
|
||||
result.append(target)
|
||||
continue
|
||||
if path not in seen_paths:
|
||||
seen_paths.add(path)
|
||||
key = (target_type, identity)
|
||||
if key not in seen:
|
||||
seen.add(key)
|
||||
result.append(target)
|
||||
return result
|
||||
|
||||
|
|
@ -1491,26 +1550,23 @@ def _is_localhost_host(host: str) -> bool:
|
|||
|
||||
|
||||
def rewrite_localhost_targets(targets_info: list[dict[str, Any]], host_gateway: str) -> None:
|
||||
from yarl import URL
|
||||
|
||||
for target_info in targets_info:
|
||||
target_type = target_info.get("type")
|
||||
details = target_info.get("details", {})
|
||||
|
||||
if target_type == "web_application":
|
||||
target_url = details.get("target_url", "")
|
||||
try:
|
||||
url = URL(target_url)
|
||||
except (ValueError, TypeError):
|
||||
continue
|
||||
|
||||
if url.host and _is_localhost_host(url.host):
|
||||
details["target_url"] = str(url.with_host(host_gateway))
|
||||
target_host = str(details.get("target_host") or "")
|
||||
if target_host and _is_localhost_host(target_host):
|
||||
details["target_host"] = host_gateway
|
||||
target_info["original"] = host_gateway
|
||||
|
||||
elif target_type == "ip_address":
|
||||
target_ip = details.get("target_ip", "")
|
||||
if target_ip and _is_localhost_host(target_ip):
|
||||
details["target_ip"] = host_gateway
|
||||
target_info["type"] = "web_application"
|
||||
details.pop("target_ip", None)
|
||||
details["target_host"] = host_gateway
|
||||
target_info["original"] = host_gateway
|
||||
|
||||
|
||||
#: API spec targets are copied into one workspace directory rather than mounted
|
||||
|
|
@ -1518,6 +1574,20 @@ def rewrite_localhost_targets(targets_info: list[dict[str, Any]], host_gateway:
|
|||
API_SPEC_WORKSPACE_SUBDIR = "api-specs"
|
||||
|
||||
|
||||
def _api_spec_staging_dir(run_name: str) -> Path:
|
||||
return run_dir_for(run_name) / ".state" / API_SPEC_WORKSPACE_SUBDIR
|
||||
|
||||
|
||||
def _api_spec_source(staging: Path) -> list[dict[str, Any]]:
|
||||
return [
|
||||
{
|
||||
"source_path": str(staging),
|
||||
"workspace_subdir": API_SPEC_WORKSPACE_SUBDIR,
|
||||
"protect_metadata": False,
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
def write_fetched_collection(collection: dict[str, Any], collection_uid: str) -> str:
|
||||
"""Write a collection fetched from the Postman API to a local file.
|
||||
|
||||
|
|
@ -1543,7 +1613,7 @@ def stage_api_specs(targets_info: list[dict[str, Any]], run_name: str) -> list[d
|
|||
if not specs:
|
||||
return []
|
||||
|
||||
staging = Path(tempfile.gettempdir()) / "strix_api_specs" / run_name
|
||||
staging = _api_spec_staging_dir(run_name)
|
||||
staging.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
used: set[str] = set()
|
||||
|
|
@ -1560,13 +1630,26 @@ def stage_api_specs(targets_info: list[dict[str, Any]], run_name: str) -> list[d
|
|||
shutil.copy2(source, staging / name)
|
||||
details["workspace_path"] = f"/workspace/{API_SPEC_WORKSPACE_SUBDIR}/{name}"
|
||||
|
||||
return [
|
||||
{
|
||||
"source_path": str(staging),
|
||||
"workspace_subdir": API_SPEC_WORKSPACE_SUBDIR,
|
||||
"protect_metadata": False,
|
||||
}
|
||||
]
|
||||
return _api_spec_source(staging)
|
||||
|
||||
|
||||
def restore_staged_api_specs(
|
||||
targets_info: list[dict[str, Any]], run_name: str
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Restore only API specs previously staged inside this run directory."""
|
||||
specs = [target for target in targets_info if target.get("type") == "api_spec"]
|
||||
if not specs:
|
||||
return []
|
||||
staging = _api_spec_staging_dir(run_name)
|
||||
for target in specs:
|
||||
workspace_path = str((target.get("details") or {}).get("workspace_path") or "")
|
||||
prefix = f"/workspace/{API_SPEC_WORKSPACE_SUBDIR}/"
|
||||
if not workspace_path.startswith(prefix):
|
||||
raise ValueError("persisted API specification has an invalid workspace path")
|
||||
name = workspace_path.removeprefix(prefix)
|
||||
if not name or "/" in name or not (staging / name).is_file():
|
||||
raise ValueError(f"staged API specification '{name or 'unknown'}' is missing")
|
||||
return _api_spec_source(staging)
|
||||
|
||||
|
||||
def clone_repository(repo_url: str, run_name: str, dest_name: str | None = None) -> str:
|
||||
|
|
|
|||
|
|
@ -60,7 +60,9 @@ export function RunDetails({
|
|||
// Configuration (launch inputs)
|
||||
const targets = arr(raw.targets_info).map((t) => {
|
||||
const o = rec(t);
|
||||
const display = str(o.original) ?? str(rec(o.details).target_url) ?? "unknown target";
|
||||
const details = rec(o.details);
|
||||
const display =
|
||||
str(o.original) ?? str(details.target_host) ?? str(details.target_url) ?? "unknown target";
|
||||
const type = str(o.type);
|
||||
return { display, type: type ? humanize(type) : null };
|
||||
});
|
||||
|
|
|
|||
File diff suppressed because one or more lines are too long
|
|
@ -6,7 +6,7 @@
|
|||
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
|
||||
<meta name="color-scheme" content="dark" />
|
||||
<title>Strix Results</title>
|
||||
<script type="module" crossorigin src="./assets/index-DBJ-RJqo.js"></script>
|
||||
<script type="module" crossorigin src="./assets/index-C2fFWAll.js"></script>
|
||||
<link rel="stylesheet" crossorigin href="./assets/index-DKbLYAbP.css">
|
||||
</head>
|
||||
<body>
|
||||
|
|
|
|||
|
|
@ -104,7 +104,10 @@ def test_build_targets_info_rejects_unparseable_spec(tmp_path: Path) -> None:
|
|||
build_targets_info(args)
|
||||
|
||||
|
||||
def test_stage_api_specs_copies_spec_into_workspace_dir(tmp_path: Path) -> None:
|
||||
def test_stage_api_specs_copies_spec_into_workspace_dir(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
targets = _resolved_targets(_write_spec(tmp_path / "src"))
|
||||
(source,) = stage_api_specs(targets, "stage-run")
|
||||
|
||||
|
|
@ -114,7 +117,10 @@ def test_stage_api_specs_copies_spec_into_workspace_dir(tmp_path: Path) -> None:
|
|||
assert targets[0]["details"]["workspace_path"] == "/workspace/api-specs/openapi.json"
|
||||
|
||||
|
||||
def test_stage_api_specs_disambiguates_same_filename(tmp_path: Path) -> None:
|
||||
def test_stage_api_specs_disambiguates_same_filename(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
targets = _resolved_targets(
|
||||
_write_spec(tmp_path / "a"),
|
||||
_write_spec(tmp_path / "b"),
|
||||
|
|
|
|||
|
|
@ -10,6 +10,8 @@ from typing import TYPE_CHECKING, Any
|
|||
|
||||
import pytest
|
||||
|
||||
from strix.interface.utils import stage_api_specs
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
|
|
@ -40,8 +42,8 @@ def test_parse_arguments_accepts_target_list_file(
|
|||
args = cli_main.parse_arguments()
|
||||
|
||||
assert [target["original"] for target in args.targets_info] == [
|
||||
"https://test1.com/",
|
||||
"http://test2.com:5789/",
|
||||
"test1.com",
|
||||
"test2.com",
|
||||
]
|
||||
assert [target["type"] for target in args.targets_info] == [
|
||||
"web_application",
|
||||
|
|
@ -64,8 +66,64 @@ def test_parse_arguments_combines_target_and_target_list(
|
|||
args = cli_main.parse_arguments()
|
||||
|
||||
assert [target["original"] for target in args.targets_info] == [
|
||||
"https://test1.com/",
|
||||
"http://test2.com:5789/",
|
||||
"test1.com",
|
||||
"test2.com",
|
||||
]
|
||||
|
||||
|
||||
def test_parse_arguments_collapses_endpoint_targets_by_host(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
target_list = tmp_path / "targets.txt"
|
||||
target_list.write_text("https://EXAMPLE.com/blog/\n", encoding="utf-8")
|
||||
_stub_settings(monkeypatch)
|
||||
monkeypatch.setattr(
|
||||
sys,
|
||||
"argv",
|
||||
[
|
||||
"strix",
|
||||
"-t",
|
||||
"https://example.com/search?q=test",
|
||||
"--target-list",
|
||||
str(target_list),
|
||||
],
|
||||
)
|
||||
|
||||
args = cli_main.parse_arguments()
|
||||
|
||||
assert args.targets_info == [
|
||||
{
|
||||
"type": "web_application",
|
||||
"details": {"target_host": "example.com"},
|
||||
"original": "example.com",
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
def test_parse_arguments_collapses_loopback_aliases_to_runtime_host(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
_stub_settings(monkeypatch)
|
||||
monkeypatch.setattr(
|
||||
sys,
|
||||
"argv",
|
||||
[
|
||||
"strix",
|
||||
"-t",
|
||||
"http://127.0.0.1/api",
|
||||
"-t",
|
||||
"http://localhost/admin",
|
||||
],
|
||||
)
|
||||
|
||||
args = cli_main.parse_arguments()
|
||||
|
||||
assert args.targets_info == [
|
||||
{
|
||||
"type": "web_application",
|
||||
"details": {"target_host": "host.docker.internal"},
|
||||
"original": "host.docker.internal",
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
|
|
@ -227,3 +285,180 @@ def test_resume_still_requires_targets_or_a_workspace(
|
|||
cli_main.parse_arguments()
|
||||
|
||||
assert "has no targets_info" in capsys.readouterr().err
|
||||
|
||||
|
||||
def test_resume_migrates_and_deduplicates_legacy_endpoint_targets(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
_write_run_record(
|
||||
tmp_path / "strix_runs",
|
||||
"legacy_abcd",
|
||||
{
|
||||
"run_name": "legacy_abcd",
|
||||
"targets_info": [
|
||||
{
|
||||
"type": "web_application",
|
||||
"details": {"target_url": "https://Example.com/search?q=test"},
|
||||
"original": "https://Example.com/search?q=test",
|
||||
},
|
||||
{
|
||||
"type": "web_application",
|
||||
"details": {"target_url": "https://example.com/blog/"},
|
||||
"original": "https://example.com/blog/",
|
||||
},
|
||||
],
|
||||
"user_instruction": "Test both endpoints.",
|
||||
},
|
||||
)
|
||||
monkeypatch.setattr(sys, "argv", ["strix", "--resume", "legacy_abcd"])
|
||||
|
||||
args = cli_main.parse_arguments()
|
||||
|
||||
assert args.targets_info == [
|
||||
{
|
||||
"type": "web_application",
|
||||
"details": {"target_host": "example.com"},
|
||||
"original": "example.com",
|
||||
}
|
||||
]
|
||||
persisted = json.loads((tmp_path / "strix_runs" / "legacy_abcd" / "run.json").read_text())
|
||||
assert persisted["targets_info"] == args.targets_info
|
||||
|
||||
|
||||
def test_resume_rejects_malformed_target_records(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
|
||||
) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
_write_run_record(
|
||||
tmp_path / "strix_runs",
|
||||
"malformed_abcd",
|
||||
{
|
||||
"run_name": "malformed_abcd",
|
||||
"targets_info": ["not-an-object"],
|
||||
"user_instruction": "test example.com",
|
||||
},
|
||||
)
|
||||
monkeypatch.setattr(sys, "argv", ["strix", "--resume", "malformed_abcd"])
|
||||
|
||||
with pytest.raises(SystemExit):
|
||||
cli_main.parse_arguments()
|
||||
|
||||
assert "invalid persisted target" in capsys.readouterr().err
|
||||
|
||||
|
||||
def test_resume_rejects_malformed_target_details(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
|
||||
) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
_write_run_record(
|
||||
tmp_path / "strix_runs",
|
||||
"malformed_details",
|
||||
{
|
||||
"run_name": "malformed_details",
|
||||
"targets_info": [{"type": "repository", "details": "not-an-object"}],
|
||||
"user_instruction": "test example.com",
|
||||
},
|
||||
)
|
||||
monkeypatch.setattr(sys, "argv", ["strix", "--resume", "malformed_details"])
|
||||
|
||||
with pytest.raises(SystemExit):
|
||||
cli_main.parse_arguments()
|
||||
|
||||
assert "invalid persisted target" in capsys.readouterr().err
|
||||
|
||||
|
||||
def test_resume_revalidates_repository_target_text(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
|
||||
) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
_write_run_record(
|
||||
tmp_path / "strix_runs",
|
||||
"malformed_repo",
|
||||
{
|
||||
"run_name": "malformed_repo",
|
||||
"targets_info": [
|
||||
{
|
||||
"type": "repository",
|
||||
"details": {"target_repo": "git@github.com:acme/service.git\nforged"},
|
||||
"original": "git@github.com:acme/service.git\nforged",
|
||||
}
|
||||
],
|
||||
"user_instruction": "review the repository",
|
||||
},
|
||||
)
|
||||
monkeypatch.setattr(sys, "argv", ["strix", "--resume", "malformed_repo"])
|
||||
|
||||
with pytest.raises(SystemExit):
|
||||
cli_main.parse_arguments()
|
||||
|
||||
assert "invalid persisted target" in capsys.readouterr().err
|
||||
|
||||
|
||||
def test_resume_migrates_loopback_ip_to_runtime_host(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
_write_run_record(
|
||||
tmp_path / "strix_runs",
|
||||
"loopback_abcd",
|
||||
{
|
||||
"run_name": "loopback_abcd",
|
||||
"targets_info": [
|
||||
{
|
||||
"type": "ip_address",
|
||||
"details": {"target_ip": "127.0.0.1"},
|
||||
"original": "127.0.0.1",
|
||||
}
|
||||
],
|
||||
"user_instruction": "test the local service",
|
||||
},
|
||||
)
|
||||
monkeypatch.setattr(sys, "argv", ["strix", "--resume", "loopback_abcd"])
|
||||
|
||||
args = cli_main.parse_arguments()
|
||||
|
||||
assert args.targets_info == [
|
||||
{
|
||||
"type": "web_application",
|
||||
"details": {"target_host": "host.docker.internal"},
|
||||
"original": "host.docker.internal",
|
||||
}
|
||||
]
|
||||
persisted = json.loads((tmp_path / "strix_runs" / "loopback_abcd" / "run.json").read_text())
|
||||
assert persisted["targets_info"] == args.targets_info
|
||||
|
||||
|
||||
def test_resume_restores_only_run_local_staged_api_spec(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
spec = tmp_path / "openapi.json"
|
||||
spec.write_text("{}", encoding="utf-8")
|
||||
monkeypatch.chdir(tmp_path)
|
||||
targets = [
|
||||
{
|
||||
"type": "api_spec",
|
||||
"details": {"target_spec": str(spec), "spec_format": "openapi"},
|
||||
"original": str(spec),
|
||||
}
|
||||
]
|
||||
stage_api_specs(targets, "api_abcd")
|
||||
_write_run_record(
|
||||
tmp_path / "strix_runs",
|
||||
"api_abcd",
|
||||
{
|
||||
"run_name": "api_abcd",
|
||||
"targets_info": targets,
|
||||
},
|
||||
)
|
||||
monkeypatch.setattr(sys, "argv", ["strix", "--resume", "api_abcd"])
|
||||
|
||||
args = cli_main.parse_arguments()
|
||||
|
||||
assert args.local_sources == [
|
||||
{
|
||||
"source_path": str(tmp_path / "strix_runs" / "api_abcd" / ".state" / "api-specs"),
|
||||
"workspace_subdir": "api-specs",
|
||||
"protect_metadata": False,
|
||||
}
|
||||
]
|
||||
|
|
|
|||
|
|
@ -298,9 +298,15 @@ async def test_setup_preflights_model_before_starting(
|
|||
) -> None:
|
||||
runtime_args = args()
|
||||
runtime_args.instruction = "CLI instruction"
|
||||
runtime_args.targets_info = [
|
||||
{
|
||||
"type": "web_application",
|
||||
"details": {"target_host": "example.com"},
|
||||
"original": "example.com",
|
||||
}
|
||||
]
|
||||
runtime = GoTuiRuntime(runtime_args)
|
||||
assert runtime.controller.instruction == "CLI instruction"
|
||||
runtime.controller.targets = ["https://example.com", "/workspace/mounted"]
|
||||
runtime.controller.scan_mode = "quick"
|
||||
runtime.controller.instruction = ""
|
||||
runtime.controller.max_budget_usd = 8.5
|
||||
|
|
@ -320,22 +326,6 @@ async def test_setup_preflights_model_before_starting(
|
|||
)
|
||||
monkeypatch.setattr(go_tui, "preflight_model_connection", preflight)
|
||||
|
||||
def build(candidate: argparse.Namespace, **_: object) -> None:
|
||||
calls.append("targets")
|
||||
assert candidate.target == ["https://example.com", "/workspace/mounted"]
|
||||
candidate.targets_info = [
|
||||
{
|
||||
"type": "web",
|
||||
"details": {"target_url": "https://example.com"},
|
||||
"original": "https://example.com",
|
||||
},
|
||||
{
|
||||
"type": "local_code",
|
||||
"details": {"target_path": "/workspace/mounted"},
|
||||
"original": "/workspace/mounted",
|
||||
},
|
||||
]
|
||||
|
||||
def prepare(candidate: argparse.Namespace) -> None:
|
||||
calls.append("prepare")
|
||||
assert candidate.max_budget_usd == 8.5
|
||||
|
|
@ -343,7 +333,6 @@ async def test_setup_preflights_model_before_starting(
|
|||
assert candidate.scope_mode == "diff"
|
||||
assert candidate.diff_base == "origin/main"
|
||||
|
||||
monkeypatch.setattr(go_tui, "build_targets_info", build)
|
||||
monkeypatch.setattr(go_tui, "prepare_run", prepare)
|
||||
monkeypatch.setattr(go_tui, "telemetry_start", lambda _args: calls.append("telemetry"))
|
||||
monkeypatch.setattr(runtime, "init_run_state", lambda: calls.append("state"))
|
||||
|
|
@ -351,7 +340,7 @@ async def test_setup_preflights_model_before_starting(
|
|||
|
||||
await runtime.start_from_setup()
|
||||
|
||||
assert calls == ["preflight", "targets", "prepare", "telemetry", "state", "scan"]
|
||||
assert calls == ["preflight", "prepare", "telemetry", "state", "scan"]
|
||||
assert runtime.args.scan_mode == "quick"
|
||||
assert runtime.args.instruction == ""
|
||||
assert runtime.args.max_budget_usd == 8.5
|
||||
|
|
@ -365,7 +354,6 @@ async def test_optimistic_setup_skips_model_preflight(
|
|||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
runtime = GoTuiRuntime(args())
|
||||
runtime.controller.targets = [str(Path.cwd())]
|
||||
calls: list[str] = []
|
||||
|
||||
async def preflight(_model: str) -> None:
|
||||
|
|
@ -377,7 +365,6 @@ async def test_optimistic_setup_skips_model_preflight(
|
|||
lambda: SimpleNamespace(llm=SimpleNamespace(model="openrouter/test-model")),
|
||||
)
|
||||
monkeypatch.setattr(go_tui, "preflight_model_connection", preflight)
|
||||
monkeypatch.setattr(go_tui, "build_targets_info", lambda _args, **_kw: calls.append("targets"))
|
||||
monkeypatch.setattr(go_tui, "prepare_run", lambda _args: calls.append("prepare"))
|
||||
monkeypatch.setattr(go_tui, "telemetry_start", lambda _args: calls.append("telemetry"))
|
||||
monkeypatch.setattr(runtime, "init_run_state", lambda: calls.append("state"))
|
||||
|
|
@ -388,7 +375,7 @@ async def test_optimistic_setup_skips_model_preflight(
|
|||
# No preflight: the scan launches straight through and any model error
|
||||
# surfaces once the agent runs.
|
||||
assert "preflight" not in calls
|
||||
assert calls == ["targets", "prepare", "telemetry", "state", "scan"]
|
||||
assert calls == ["prepare", "telemetry", "state", "scan"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
|
@ -409,11 +396,6 @@ async def test_confirmed_target_less_launch_mounts_workspace_without_targets(
|
|||
lambda: SimpleNamespace(llm=SimpleNamespace(model="openrouter/test-model")),
|
||||
)
|
||||
monkeypatch.setattr(go_tui, "preflight_model_connection", preflight)
|
||||
monkeypatch.setattr(
|
||||
go_tui,
|
||||
"build_targets_info",
|
||||
lambda _args, **_kw: pytest.fail("a target-less launch must not build targets"),
|
||||
)
|
||||
monkeypatch.setattr(go_tui, "prepare_run", prepared.append)
|
||||
monkeypatch.setattr(go_tui, "telemetry_start", lambda _args: None)
|
||||
monkeypatch.setattr(runtime, "init_run_state", lambda: None)
|
||||
|
|
@ -434,9 +416,9 @@ async def test_setup_preserves_prepared_cli_targets(
|
|||
runtime_args.target_list = []
|
||||
runtime_args.targets_info = [
|
||||
{
|
||||
"type": "web",
|
||||
"details": {"url": "https://example.com"},
|
||||
"original": "https://example.com",
|
||||
"type": "web_application",
|
||||
"details": {"target_host": "example.com"},
|
||||
"original": "example.com",
|
||||
}
|
||||
]
|
||||
runtime = GoTuiRuntime(runtime_args)
|
||||
|
|
@ -451,11 +433,6 @@ async def test_setup_preserves_prepared_cli_targets(
|
|||
lambda: SimpleNamespace(llm=SimpleNamespace(model="openrouter/test-model")),
|
||||
)
|
||||
monkeypatch.setattr(go_tui, "preflight_model_connection", preflight)
|
||||
monkeypatch.setattr(
|
||||
go_tui,
|
||||
"build_targets_info",
|
||||
lambda _args, **_kw: pytest.fail("prepared targets should not be rebuilt"),
|
||||
)
|
||||
monkeypatch.setattr(go_tui, "prepare_run", lambda _args: calls.append("prepare"))
|
||||
monkeypatch.setattr(go_tui, "telemetry_start", lambda _args: calls.append("telemetry"))
|
||||
monkeypatch.setattr(runtime, "init_run_state", lambda: calls.append("state"))
|
||||
|
|
@ -463,245 +440,11 @@ async def test_setup_preserves_prepared_cli_targets(
|
|||
|
||||
await runtime.start_from_setup()
|
||||
|
||||
assert runtime.controller.targets == ["https://example.com"]
|
||||
assert runtime.args.targets_info[0]["type"] == "web"
|
||||
assert runtime.controller.targets == ["example.com"]
|
||||
assert runtime.args.targets_info[0]["type"] == "web_application"
|
||||
assert calls == ["preflight", "prepare", "telemetry", "state", "scan"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_setup_target_change_preserves_local_targets(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
runtime_args = args()
|
||||
runtime_args.target = []
|
||||
runtime_args.target_list = ["targets.txt"]
|
||||
runtime_args.targets_info = [
|
||||
{
|
||||
"type": "local_code",
|
||||
"details": {"target_path": "/workspace/source"},
|
||||
"original": "/workspace/source",
|
||||
}
|
||||
]
|
||||
runtime = GoTuiRuntime(runtime_args)
|
||||
runtime.controller.targets.append("https://example.com")
|
||||
|
||||
async def preflight(_model: str) -> None:
|
||||
return None
|
||||
|
||||
def build(target_args: argparse.Namespace, **_: object) -> None:
|
||||
assert target_args.target == ["/workspace/source", "https://example.com"]
|
||||
target_args.targets_info = [
|
||||
{
|
||||
"type": "web",
|
||||
"details": {"url": "https://example.com"},
|
||||
"original": "https://example.com",
|
||||
},
|
||||
{
|
||||
"type": "local_code",
|
||||
"details": {"target_path": "/workspace/source"},
|
||||
"original": "/workspace/source",
|
||||
},
|
||||
]
|
||||
|
||||
monkeypatch.setattr(
|
||||
go_tui,
|
||||
"load_settings",
|
||||
lambda: SimpleNamespace(llm=SimpleNamespace(model="openrouter/test-model")),
|
||||
)
|
||||
monkeypatch.setattr(go_tui, "preflight_model_connection", preflight)
|
||||
monkeypatch.setattr(go_tui, "build_targets_info", build)
|
||||
monkeypatch.setattr(go_tui, "prepare_run", lambda _args: None)
|
||||
monkeypatch.setattr(go_tui, "telemetry_start", lambda _args: None)
|
||||
monkeypatch.setattr(runtime, "init_run_state", lambda: None)
|
||||
monkeypatch.setattr(runtime, "start_scan", lambda: None)
|
||||
|
||||
await runtime.start_from_setup()
|
||||
|
||||
assert runtime.args.target_list == []
|
||||
assert runtime.args.targets_info[0]["type"] == "web"
|
||||
assert runtime.args.targets_info[1]["type"] == "local_code"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_setup_same_basename_uses_combined_workspace_names_on_retry(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
existing_repo = "https://example.com/first/app.git"
|
||||
added_repo = "https://example.com/second/app.git"
|
||||
runtime_args = args()
|
||||
runtime_args.target = []
|
||||
runtime_args.target_list = ["targets.txt"]
|
||||
runtime_args.targets_info = [
|
||||
{
|
||||
"type": "repository",
|
||||
"details": {
|
||||
"target_repo": existing_repo,
|
||||
"workspace_subdir": "app",
|
||||
"cloned_repo_path": "/clones/app",
|
||||
},
|
||||
"original": existing_repo,
|
||||
}
|
||||
]
|
||||
runtime = GoTuiRuntime(runtime_args)
|
||||
runtime.controller.targets.append(added_repo)
|
||||
prepare_attempts = 0
|
||||
started: list[str] = []
|
||||
|
||||
async def preflight(_model: str) -> None:
|
||||
return None
|
||||
|
||||
def build(target_args: argparse.Namespace, **_: object) -> None:
|
||||
assert target_args.target == [existing_repo, added_repo]
|
||||
target_args.targets_info = [
|
||||
{
|
||||
"type": "repository",
|
||||
"details": {
|
||||
"target_repo": existing_repo,
|
||||
"workspace_subdir": "app",
|
||||
},
|
||||
"original": existing_repo,
|
||||
},
|
||||
{
|
||||
"type": "repository",
|
||||
"details": {
|
||||
"target_repo": added_repo,
|
||||
"workspace_subdir": "app-2",
|
||||
},
|
||||
"original": added_repo,
|
||||
},
|
||||
]
|
||||
|
||||
def prepare(candidate: argparse.Namespace) -> None:
|
||||
nonlocal prepare_attempts
|
||||
prepare_attempts += 1
|
||||
assert [target["details"]["workspace_subdir"] for target in candidate.targets_info] == [
|
||||
"app",
|
||||
"app-2",
|
||||
]
|
||||
if prepare_attempts == 1:
|
||||
candidate.targets_info[0]["details"]["target_repo"] = "/mutated"
|
||||
candidate.targets_info[1]["details"]["workspace_subdir"] = "mutated"
|
||||
raise ValueError("retry setup")
|
||||
|
||||
monkeypatch.setattr(
|
||||
go_tui,
|
||||
"load_settings",
|
||||
lambda: SimpleNamespace(llm=SimpleNamespace(model="openrouter/test-model")),
|
||||
)
|
||||
monkeypatch.setattr(go_tui, "preflight_model_connection", preflight)
|
||||
monkeypatch.setattr(go_tui, "build_targets_info", build)
|
||||
monkeypatch.setattr(go_tui, "prepare_run", prepare)
|
||||
monkeypatch.setattr(go_tui, "telemetry_start", lambda _args: None)
|
||||
monkeypatch.setattr(runtime, "init_run_state", lambda: started.append("state"))
|
||||
monkeypatch.setattr(runtime, "start_scan", lambda: started.append("scan"))
|
||||
|
||||
with pytest.raises(ValueError, match="retry setup"):
|
||||
await runtime.start_from_setup()
|
||||
|
||||
assert runtime.args.targets_info[0]["details"] == {
|
||||
"target_repo": existing_repo,
|
||||
"workspace_subdir": "app",
|
||||
"cloned_repo_path": "/clones/app",
|
||||
}
|
||||
assert started == []
|
||||
|
||||
await runtime.start_from_setup()
|
||||
|
||||
assert prepare_attempts == 2
|
||||
assert runtime.args.target_list == []
|
||||
assert [target["details"]["workspace_subdir"] for target in runtime.args.targets_info] == [
|
||||
"app",
|
||||
"app-2",
|
||||
]
|
||||
assert runtime.args.targets_info[0]["details"]["target_repo"] == existing_repo
|
||||
assert started == ["state", "scan"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_setup_target_rebuild_restores_all_target_fields_on_failure(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
runtime_args = args()
|
||||
runtime_args.target = None
|
||||
runtime_args.target_list = ["targets.txt"]
|
||||
runtime_args.targets_info = [
|
||||
{
|
||||
"type": "local_code",
|
||||
"details": {"target_path": "/workspace/source"},
|
||||
"original": "/workspace/source",
|
||||
}
|
||||
]
|
||||
original_targets_info = json.loads(json.dumps(runtime_args.targets_info))
|
||||
runtime = GoTuiRuntime(runtime_args)
|
||||
runtime.controller.targets.append("https://example.com")
|
||||
|
||||
async def preflight(_model: str) -> None:
|
||||
return None
|
||||
|
||||
def fail_rebuild(target_args: argparse.Namespace, **_: object) -> None:
|
||||
target_args.target = ["mutated"]
|
||||
target_args.target_list = ["mutated.txt"]
|
||||
target_args.targets_info = [{"original": "partial"}]
|
||||
raise ValueError("bad target")
|
||||
|
||||
monkeypatch.setattr(
|
||||
go_tui,
|
||||
"load_settings",
|
||||
lambda: SimpleNamespace(llm=SimpleNamespace(model="openrouter/test-model")),
|
||||
)
|
||||
monkeypatch.setattr(go_tui, "preflight_model_connection", preflight)
|
||||
monkeypatch.setattr(go_tui, "build_targets_info", fail_rebuild)
|
||||
|
||||
with pytest.raises(ValueError, match="bad target"):
|
||||
await runtime.start_from_setup()
|
||||
|
||||
assert runtime.args.target is None
|
||||
assert runtime.args.target_list == ["targets.txt"]
|
||||
assert runtime.args.targets_info == original_targets_info
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_setup_rebuild_canonicalizes_relative_local_target(
|
||||
tmp_path: Any,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
source = tmp_path / "source"
|
||||
source.mkdir()
|
||||
monkeypatch.chdir(tmp_path)
|
||||
runtime_args = args()
|
||||
runtime_args.target = []
|
||||
runtime_args.target_list = []
|
||||
runtime = GoTuiRuntime(runtime_args)
|
||||
runtime.controller.targets = ["source"]
|
||||
prepared = False
|
||||
|
||||
async def preflight(_model: str) -> None:
|
||||
return None
|
||||
|
||||
def prepare(candidate: argparse.Namespace) -> None:
|
||||
nonlocal prepared
|
||||
prepared = True
|
||||
assert len(candidate.targets_info) == 1
|
||||
assert candidate.targets_info[0]["details"]["target_path"] == str(source.resolve())
|
||||
|
||||
monkeypatch.setattr(
|
||||
go_tui,
|
||||
"load_settings",
|
||||
lambda: SimpleNamespace(llm=SimpleNamespace(model="openrouter/test-model")),
|
||||
)
|
||||
monkeypatch.setattr(go_tui, "preflight_model_connection", preflight)
|
||||
monkeypatch.setattr(go_tui, "prepare_run", prepare)
|
||||
monkeypatch.setattr(go_tui, "telemetry_start", lambda _args: None)
|
||||
monkeypatch.setattr(runtime, "init_run_state", lambda: None)
|
||||
monkeypatch.setattr(runtime, "start_scan", lambda: None)
|
||||
|
||||
await runtime.start_from_setup()
|
||||
|
||||
assert prepared is True
|
||||
assert runtime.args.targets_info[0]["original"] == str(source.resolve())
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.asyncio
|
||||
async def test_setup_prepare_system_exit_is_recoverable_and_transactional(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
|
|
@ -713,9 +456,9 @@ async def test_setup_prepare_system_exit_is_recoverable_and_transactional(
|
|||
runtime_args.target_list = []
|
||||
runtime_args.targets_info = [
|
||||
{
|
||||
"type": "web",
|
||||
"details": {"url": "https://example.com"},
|
||||
"original": "https://example.com",
|
||||
"type": "web_application",
|
||||
"details": {"target_host": "example.com"},
|
||||
"original": "example.com",
|
||||
}
|
||||
]
|
||||
original_args = json.loads(json.dumps(vars(runtime_args)))
|
||||
|
|
@ -730,7 +473,7 @@ async def test_setup_prepare_system_exit_is_recoverable_and_transactional(
|
|||
def fail_prepare(candidate: argparse.Namespace) -> None:
|
||||
assert candidate is not runtime.args
|
||||
candidate.run_name = "mutated-run"
|
||||
candidate.targets_info[0]["details"]["url"] = "https://mutated.example"
|
||||
candidate.targets_info[0]["details"]["target_host"] = "mutated.example"
|
||||
raise ValueError("invalid diff scope")
|
||||
|
||||
def telemetry(_candidate: argparse.Namespace) -> None:
|
||||
|
|
@ -788,7 +531,7 @@ async def test_setup_preflight_failure_does_not_start_scan(
|
|||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
runtime = GoTuiRuntime(args())
|
||||
runtime.controller.targets = ["https://example.com"]
|
||||
runtime.controller.targets = ["example.com"]
|
||||
started = False
|
||||
|
||||
async def preflight(_model: str) -> None:
|
||||
|
|
@ -804,7 +547,6 @@ async def test_setup_preflight_failure_does_not_start_scan(
|
|||
lambda: SimpleNamespace(llm=SimpleNamespace(model="openrouter/test-model")),
|
||||
)
|
||||
monkeypatch.setattr(go_tui, "preflight_model_connection", preflight)
|
||||
monkeypatch.setattr(go_tui, "build_targets_info", mark_started)
|
||||
monkeypatch.setattr(runtime, "init_run_state", mark_started)
|
||||
monkeypatch.setattr(runtime, "start_scan", mark_started)
|
||||
|
||||
|
|
|
|||
|
|
@ -200,14 +200,15 @@ def test_build_root_task_web_target_injected_as_context() -> None:
|
|||
"""The prompt leads and the configured target remains visible below it."""
|
||||
config = {
|
||||
"targets": [
|
||||
{"type": "web_application", "details": {"target_url": "https://app.example.com"}},
|
||||
{"type": "web_application", "details": {"target_host": "app.example.com"}},
|
||||
],
|
||||
"user_instructions": "Focus on auth.",
|
||||
}
|
||||
task = build_root_task(config)
|
||||
|
||||
assert task.startswith("Focus on auth.")
|
||||
assert "https://app.example.com" in task
|
||||
assert "Hosts:" in task
|
||||
assert "app.example.com" in task
|
||||
assert "configured targets and supporting material" in task
|
||||
assert "Special instructions:" not in task
|
||||
assert "SYSTEM-VERIFIED" not in task
|
||||
|
|
@ -228,7 +229,7 @@ def test_build_root_task_workspace_mount_is_not_a_target() -> None:
|
|||
assert "No scan target was set" in task
|
||||
assert task.startswith("Find IDOR in the checkout flow.")
|
||||
# It must not be presented as an asset to test.
|
||||
for label in ("Local Codebases:", "Repositories:", "URLs:", "IP Addresses:"):
|
||||
for label in ("Local Codebases:", "Repositories:", "Hosts:", "IP Addresses:"):
|
||||
assert label not in task
|
||||
|
||||
|
||||
|
|
@ -240,6 +241,7 @@ def test_build_scope_context_authorizes_nothing_without_targets() -> None:
|
|||
|
||||
assert scope["authorized_targets"] == []
|
||||
assert scope["user_instruction_hosts_expand_scope"] is True
|
||||
assert build_target_summary_text([]).plain == "Target task-defined scope"
|
||||
|
||||
|
||||
def test_scope_prompt_authorizes_flag_and_instruction_hosts_with_subdomains() -> None:
|
||||
|
|
@ -247,14 +249,14 @@ def test_scope_prompt_authorizes_flag_and_instruction_hosts_with_subdomains() ->
|
|||
"targets": [
|
||||
{
|
||||
"type": "web_application",
|
||||
"details": {"target_url": "https://app.example.com/search?q=test"},
|
||||
},
|
||||
{
|
||||
"type": "web_application",
|
||||
"details": {"target_url": "https://app.example.com/blog/"},
|
||||
},
|
||||
"details": {"target_host": "app.example.com"},
|
||||
"original": "app.example.com",
|
||||
}
|
||||
],
|
||||
"user_instructions": "Also test https://api.example.net/v1.",
|
||||
"user_instructions": (
|
||||
"Test https://app.example.com/search?q=test and "
|
||||
"https://app.example.com/blog/. Also test https://api.example.net/v1."
|
||||
),
|
||||
}
|
||||
context = build_scope_context(config)
|
||||
|
||||
|
|
@ -269,6 +271,7 @@ def test_scope_prompt_authorizes_flag_and_instruction_hosts_with_subdomains() ->
|
|||
assert prompt.count("host: app.example.com") == 1
|
||||
assert "https://app.example.com/search?q=test" not in prompt
|
||||
assert "https://app.example.com/search?q=test" in task
|
||||
assert "https://app.example.com/blog/" in task
|
||||
assert "https://api.example.net/v1" in task
|
||||
assert "Every network host explicitly named in the user's root scan task" in prompt
|
||||
assert "exact hostname and all of its descendant subdomains" in prompt
|
||||
|
|
@ -288,8 +291,8 @@ def test_scope_prompt_keeps_web_ip_targets_exact() -> None:
|
|||
{
|
||||
"targets": [
|
||||
{
|
||||
"type": "web_application",
|
||||
"details": {"target_url": "https://192.0.2.10:8443/admin"},
|
||||
"type": "ip_address",
|
||||
"details": {"target_ip": "192.0.2.10"},
|
||||
}
|
||||
]
|
||||
}
|
||||
|
|
|
|||
|
|
@ -12,7 +12,7 @@ from strix.interface.scan_setup import attach_workspace_mount
|
|||
from strix.interface.utils import (
|
||||
check_mountable_dir,
|
||||
collect_local_sources,
|
||||
dedupe_local_targets,
|
||||
dedupe_targets,
|
||||
infer_target_type,
|
||||
read_target_list_file,
|
||||
)
|
||||
|
|
@ -181,6 +181,77 @@ def test_infer_target_type_applies_the_mount_policy() -> None:
|
|||
infer_target_type("/etc")
|
||||
|
||||
|
||||
def test_infer_web_target_reduces_endpoint_to_host() -> None:
|
||||
assert infer_target_type("https://Example.COM:8443/search?q=test#results") == (
|
||||
"web_application",
|
||||
{"target_host": "example.com"},
|
||||
)
|
||||
|
||||
|
||||
def test_infer_multi_segment_web_path_is_not_probed_as_repository() -> None:
|
||||
assert infer_target_type("https://app.example.com/api/v1/users") == (
|
||||
"web_application",
|
||||
{"target_host": "app.example.com"},
|
||||
)
|
||||
|
||||
|
||||
def test_infer_web_ip_target_becomes_exact_ip() -> None:
|
||||
assert infer_target_type("https://192.0.2.10:8443/admin") == (
|
||||
"ip_address",
|
||||
{"target_ip": "192.0.2.10"},
|
||||
)
|
||||
|
||||
|
||||
def test_infer_repository_keeps_its_path() -> None:
|
||||
target = "https://github.com/acme/service.git"
|
||||
assert infer_target_type(target) == ("repository", {"target_repo": target})
|
||||
|
||||
|
||||
def test_infer_http_repository_without_explicit_git_syntax_is_a_web_target() -> None:
|
||||
target = "https://github.com/acme/service"
|
||||
assert infer_target_type(target) == (
|
||||
"web_application",
|
||||
{"target_host": "github.com"},
|
||||
)
|
||||
|
||||
|
||||
def test_infer_explicit_git_https_repository_keeps_self_hosted_path() -> None:
|
||||
target = "https://git.example.com/acme/service"
|
||||
assert infer_target_type(f"git+{target}") == ("repository", {"target_repo": target})
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"target",
|
||||
[
|
||||
"https://github.com/login/oauth",
|
||||
"https://github.com/about/security",
|
||||
"https://github.com/acme/service/issues/1",
|
||||
],
|
||||
)
|
||||
def test_infer_git_provider_web_pages_remain_web_targets(target: str) -> None:
|
||||
assert infer_target_type(target) == (
|
||||
"web_application",
|
||||
{"target_host": "github.com"},
|
||||
)
|
||||
|
||||
|
||||
def test_infer_web_basic_auth_url_remains_web_target() -> None:
|
||||
assert infer_target_type("https://admin:secret@app.example.com/dashboard") == (
|
||||
"web_application",
|
||||
{"target_host": "app.example.com"},
|
||||
)
|
||||
|
||||
|
||||
def test_infer_web_target_rejects_invalid_hostname() -> None:
|
||||
with pytest.raises(ValueError, match="invalid host"):
|
||||
infer_target_type("https://example.com bad-scope-text")
|
||||
|
||||
|
||||
def test_infer_repository_rejects_control_characters() -> None:
|
||||
with pytest.raises(ValueError, match="control characters"):
|
||||
infer_target_type("git@github.com:acme/service.git\nforged")
|
||||
|
||||
|
||||
def test_read_target_list_file_strips_blank_lines(tmp_path: Path) -> None:
|
||||
target_list = tmp_path / "targets.txt"
|
||||
target_list.write_text(
|
||||
|
|
@ -237,13 +308,22 @@ def test_read_target_list_file_rejects_empty_path(empty: str) -> None:
|
|||
def test_dedupe_keeps_distinct_targets_in_order() -> None:
|
||||
targets = [
|
||||
_local_target("/a"),
|
||||
{"type": "web_application", "details": {"target_url": "https://x"}},
|
||||
{"type": "web_application", "details": {"target_host": "x.example"}},
|
||||
_local_target("/b"),
|
||||
]
|
||||
assert dedupe_local_targets(targets) == targets
|
||||
assert dedupe_targets(targets) == targets
|
||||
|
||||
|
||||
def test_dedupe_collapses_the_same_path() -> None:
|
||||
assert dedupe_local_targets([_local_target("/repo"), _local_target("/repo")]) == [
|
||||
assert dedupe_targets([_local_target("/repo"), _local_target("/repo")]) == [
|
||||
_local_target("/repo")
|
||||
]
|
||||
|
||||
|
||||
def test_dedupe_collapses_the_same_web_host() -> None:
|
||||
target = {
|
||||
"type": "web_application",
|
||||
"details": {"target_host": "example.com"},
|
||||
"original": "example.com",
|
||||
}
|
||||
assert dedupe_targets([target, target.copy()]) == [target]
|
||||
|
|
|
|||
|
|
@ -127,7 +127,7 @@ async def test_root_prompt_options_flow_into_root_agent(
|
|||
kwargs = captured["kwargs"]
|
||||
instructions_override = kwargs["instructions_override"]
|
||||
assert "SYSTEM-VERIFIED SCOPE" in instructions_override
|
||||
assert "AUTHORIZED TARGETS" in instructions_override
|
||||
assert "AUTHORIZED CONFIGURED TARGETS" in instructions_override
|
||||
assert "host: example.com (includes example.com and *.example.com)" in instructions_override
|
||||
assert "exact hostname and all of its descendant subdomains" in instructions_override
|
||||
assert "CUSTOM SCAN PROMPT" in instructions_override
|
||||
|
|
|
|||
|
|
@ -24,7 +24,7 @@ def test_parse_repo_full_name_handles_common_forms() -> None:
|
|||
def test_repository_context_none_for_non_repository_targets() -> None:
|
||||
state = ReportState(run_name="t")
|
||||
state.run_record["targets_info"] = [
|
||||
{"type": "web_application", "details": {"target_url": "https://example.com"}}
|
||||
{"type": "web_application", "details": {"target_host": "example.com"}}
|
||||
]
|
||||
assert state._sarif_repository_context() is None
|
||||
|
||||
|
|
|
|||
|
|
@ -29,6 +29,18 @@ def args() -> argparse.Namespace:
|
|||
)
|
||||
|
||||
|
||||
def args_with_target(host: str = "example.com") -> argparse.Namespace:
|
||||
setup_args = args()
|
||||
setup_args.targets_info = [
|
||||
{
|
||||
"type": "web_application",
|
||||
"details": {"target_host": host},
|
||||
"original": host,
|
||||
}
|
||||
]
|
||||
return setup_args
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def isolated_config(tmp_path: Path) -> None:
|
||||
for key in (
|
||||
|
|
@ -47,11 +59,10 @@ def isolated_config(tmp_path: Path) -> None:
|
|||
|
||||
@pytest.mark.asyncio
|
||||
async def test_setup_state_is_serializable() -> None:
|
||||
controller = TuiController(args())
|
||||
await controller.handle("setup.add_target", {"target": "https://example.com"})
|
||||
controller = TuiController(args_with_target())
|
||||
await controller.handle("setup.set_instruction", {"instruction": "focus on auth"})
|
||||
snapshot = controller.snapshot()
|
||||
assert snapshot["targets"] == ["https://example.com"]
|
||||
assert snapshot["targets"] == ["example.com"]
|
||||
assert snapshot["instruction"] == "focus on auth"
|
||||
assert snapshot["scan_state"] == "setup"
|
||||
assert snapshot["scan_mode"] == "deep"
|
||||
|
|
@ -82,19 +93,15 @@ async def test_setup_controls_reject_changes_after_start() -> None:
|
|||
controller.scan_started = True
|
||||
|
||||
with pytest.raises(RuntimeError, match="can no longer be changed"):
|
||||
await controller.handle("setup.add_target", {"target": "https://example.com"})
|
||||
await controller.handle("setup.set_instruction", {"instruction": "new task"})
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_large_target_list_reports_truncated_snapshot_count() -> None:
|
||||
controller = TuiController(args())
|
||||
|
||||
for index in range(20):
|
||||
await controller.handle("setup.add_target", {"target": f"https://target-{index}.example"})
|
||||
added = await controller.handle("setup.add_target", {"target": "https://last.example"})
|
||||
controller.targets = [f"target-{index}.example" for index in range(21)]
|
||||
snapshot = controller.snapshot()
|
||||
|
||||
assert added == {"target": "https://last.example", "total": 21}
|
||||
assert snapshot["target_count"] == 21
|
||||
# The snapshot only carries a bounded prefix of the list.
|
||||
assert len(snapshot["targets"]) == 16
|
||||
|
|
@ -113,13 +120,13 @@ def test_state_populates_model_warning_for_non_frontier_model() -> None:
|
|||
def test_setup_restores_prepared_cli_targets() -> None:
|
||||
setup_args = args()
|
||||
setup_args.targets_info = [
|
||||
{"type": "web", "details": {}, "original": "https://example.com"},
|
||||
{"type": "web_application", "details": {}, "original": "example.com"},
|
||||
{"type": "local_code", "details": {}, "original": "/workspace/source"},
|
||||
]
|
||||
|
||||
controller = TuiController(setup_args)
|
||||
|
||||
assert controller.snapshot()["targets"] == ["https://example.com", "/workspace/source"]
|
||||
assert controller.snapshot()["targets"] == ["example.com", "/workspace/source"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
|
|
@ -130,8 +137,7 @@ async def test_start_validates_model_before_callback() -> None:
|
|||
nonlocal started
|
||||
started = True
|
||||
|
||||
controller = TuiController(args(), on_start=start)
|
||||
await controller.handle("setup.add_target", {"target": "https://example.com"})
|
||||
controller = TuiController(args_with_target(), on_start=start)
|
||||
with pytest.raises(ValueError, match="No model configured"):
|
||||
await controller.handle("setup.start", {})
|
||||
assert started is False
|
||||
|
|
@ -147,8 +153,7 @@ async def test_start_launches_with_a_configured_model() -> None:
|
|||
|
||||
os.environ["STRIX_LLM"] = "anthropic/claude-sonnet-4"
|
||||
loader._cached = None
|
||||
controller = TuiController(args(), on_start=start)
|
||||
await controller.handle("setup.add_target", {"target": "https://example.com"})
|
||||
controller = TuiController(args_with_target(), on_start=start)
|
||||
|
||||
result = await controller.handle("setup.start", {})
|
||||
|
||||
|
|
@ -305,8 +310,7 @@ async def test_start_forwards_verify_flag_by_default() -> None:
|
|||
os.environ["STRIX_LLM"] = "anthropic/claude-sonnet-4"
|
||||
os.environ["ANTHROPIC_API_KEY"] = "test-key"
|
||||
loader._cached = None
|
||||
controller = TuiController(args(), on_start=start)
|
||||
await controller.handle("setup.add_target", {"target": "https://example.com"})
|
||||
controller = TuiController(args_with_target(), on_start=start)
|
||||
|
||||
# A named target keeps the upfront model check.
|
||||
await controller.handle("setup.start", {})
|
||||
|
|
@ -326,8 +330,7 @@ async def test_start_rejects_concurrent_and_repeated_submissions() -> None:
|
|||
os.environ["STRIX_LLM"] = "anthropic/claude-sonnet-4"
|
||||
os.environ["ANTHROPIC_API_KEY"] = "test-key"
|
||||
loader._cached = None
|
||||
controller = TuiController(args(), on_start=start)
|
||||
await controller.handle("setup.add_target", {"target": "https://example.com"})
|
||||
controller = TuiController(args_with_target(), on_start=start)
|
||||
|
||||
first_start = asyncio.create_task(controller.handle("setup.start", {}))
|
||||
await entered.wait()
|
||||
|
|
|
|||
|
|
@ -187,17 +187,17 @@ async def test_server_command_round_trip_over_inherited_socket() -> None:
|
|||
child,
|
||||
{
|
||||
"version": 3,
|
||||
"type": "setup.add_target",
|
||||
"type": "setup.set_instruction",
|
||||
"request_id": "test-1",
|
||||
"payload": {"target": "example.com"},
|
||||
"payload": {"instruction": "test example.com"},
|
||||
},
|
||||
)
|
||||
result = await receive_until(child, "command_result", request_id="test-1")
|
||||
assert result["payload"]["ok"] is True
|
||||
assert result["payload"]["command"] == "setup.add_target"
|
||||
assert result["payload"]["command"] == "setup.set_instruction"
|
||||
state = await receive_until(child, "state")
|
||||
assert state["payload"]["revision"] >= 1
|
||||
assert state["payload"]["state"]["targets"] == ["example.com"]
|
||||
assert state["payload"]["state"]["instruction"] == "test example.com"
|
||||
finally:
|
||||
child.close()
|
||||
await server.close()
|
||||
|
|
@ -284,9 +284,9 @@ async def test_invalid_version_error_is_correlated_and_next_command_succeeds() -
|
|||
child,
|
||||
{
|
||||
"version": 2,
|
||||
"type": "setup.add_target",
|
||||
"type": "setup.set_instruction",
|
||||
"request_id": "bad-version",
|
||||
"payload": {"target": "ignored.example"},
|
||||
"payload": {"instruction": "ignored"},
|
||||
},
|
||||
)
|
||||
rejected = await receive_until(child, "command_result", request_id="bad-version")
|
||||
|
|
@ -296,9 +296,9 @@ async def test_invalid_version_error_is_correlated_and_next_command_succeeds() -
|
|||
child,
|
||||
{
|
||||
"version": 3,
|
||||
"type": "setup.add_target",
|
||||
"type": "setup.set_instruction",
|
||||
"request_id": "after-error",
|
||||
"payload": {"target": "example.com"},
|
||||
"payload": {"instruction": "test example.com"},
|
||||
},
|
||||
)
|
||||
accepted = await receive_until(child, "command_result", request_id="after-error")
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue