Compare commits
9 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 639ce5b31a | |||
| 5ee34481fe | |||
| f342808d2b | |||
| f554523378 | |||
| 69e82f0258 | |||
| 52ca641679 | |||
| 60d68d85f3 | |||
| 777005a42b | |||
| 8cdf0683a3 |
@@ -11,7 +11,7 @@ repos:
|
||||
|
||||
# MyPy for static type checking
|
||||
- repo: https://github.com/pre-commit/mirrors-mypy
|
||||
rev: v1.16.0
|
||||
rev: v1.17.1
|
||||
hooks:
|
||||
- id: mypy
|
||||
additional_dependencies: [
|
||||
|
||||
@@ -8,7 +8,7 @@
|
||||
|
||||
# Strix
|
||||
|
||||
### Open-source AI hackers to find and fix your app’s vulnerabilities.
|
||||
### The open-source AI pentesting tool. Autonomous AI hackers that find and fix your app’s vulnerabilities.
|
||||
|
||||
<br/>
|
||||
|
||||
@@ -40,15 +40,15 @@
|
||||
|
||||
## Strix Overview
|
||||
|
||||
Strix are autonomous AI agents that act just like real hackers - they run your code dynamically, find vulnerabilities, and validate them through actual proof-of-concepts. Built for developers and security teams who need fast, accurate security testing without the overhead of manual pentesting or the false positives of static analysis tools.
|
||||
Strix are autonomous AI penetration testing agents that act just like real hackers - they run your code dynamically, find vulnerabilities, and validate them through actual proof-of-concepts. Built for developers and security teams who need fast, accurate security testing without the overhead of manual pentesting or the false positives of static analysis tools.
|
||||
|
||||
**Key Capabilities:**
|
||||
|
||||
- **Full hacker toolkit** out of the box
|
||||
- **Teams of agents** that collaborate and scale
|
||||
- **Real validation** with PoCs, not false positives
|
||||
- **Developer‑first** CLI with actionable reports
|
||||
- **Auto‑fix & reporting** to accelerate remediation
|
||||
- **Full pentesting toolkit** - reconnaissance, exploitation, and validation out of the box
|
||||
- **Multi-agent orchestration** - teams of AI pentesters that collaborate and scale
|
||||
- **Real exploit validation** - working PoCs, not false positives like legacy vulnerability scanners
|
||||
- **Developer‑first CLI** - actionable findings with remediation guidance
|
||||
- **Auto‑fix & reporting** - generate patches and compliance-ready pentest reports
|
||||
|
||||
|
||||
<br>
|
||||
@@ -95,13 +95,13 @@ strix --target ./app-directory
|
||||
|
||||
## ☁️ Strix Platform
|
||||
|
||||
Try the Strix full-stack security platform at **[app.strix.ai](https://app.strix.ai)** — sign up for free, connect your repos and domains, and launch a pentest in minutes.
|
||||
Try the Strix full-stack penetration testing platform at **[app.strix.ai](https://app.strix.ai)** - sign up for free, connect your repos and domains, and launch a pentest in minutes.
|
||||
|
||||
- **Validated findings with PoCs** and reproduction steps
|
||||
- **One-click autofix** as ready-to-merge pull requests
|
||||
- **Continuous monitoring** across code, cloud, and infrastructure
|
||||
- **Integrations** with GitHub, Slack, Jira, Linear, and CI/CD pipelines
|
||||
- **Continuous learning** that builds on past findings and remediations
|
||||
- **Validated findings with PoCs** - every vulnerability includes a working proof-of-concept exploit and reproduction steps
|
||||
- **One-click autofix** - AI-generated security patches as ready-to-merge pull requests
|
||||
- **Continuous pentesting** - always-on vulnerability scanning that keeps pace with your deployments
|
||||
- **DevSecOps integrations** - GitHub, GitLab, Bitbucket, Slack, Jira, Linear, and CI/CD pipelines
|
||||
- **Continuous learning** - AI that builds on past findings, adapts to your codebase, and reduces false positives over time
|
||||
|
||||
[**Start your first pentest →**](https://app.strix.ai)
|
||||
|
||||
@@ -109,37 +109,38 @@ Try the Strix full-stack security platform at **[app.strix.ai](https://app.strix
|
||||
|
||||
## ✨ Features
|
||||
|
||||
### Agentic Security Tools
|
||||
### Agentic Pentesting Tools
|
||||
|
||||
Strix agents come equipped with a comprehensive security testing toolkit:
|
||||
Strix agents come equipped with a comprehensive offensive security toolkit - the same tools used by professional penetration testers and ethical hackers:
|
||||
|
||||
- **Full HTTP Proxy** - Full request/response manipulation and analysis
|
||||
- **Browser Automation** - Multi-tab browser for testing of XSS, CSRF, auth flows
|
||||
- **Terminal Environments** - Interactive shells for command execution and testing
|
||||
- **Python Runtime** - Custom exploit development and validation
|
||||
- **Reconnaissance** - Automated OSINT and attack surface mapping
|
||||
- **Code Analysis** - Static and dynamic analysis capabilities
|
||||
- **Knowledge Management** - Structured findings and attack documentation
|
||||
- **HTTP Interception Proxy** - Full request/response manipulation and analysis with Caido
|
||||
- **Browser Exploitation** - Automated browser for testing XSS, CSRF, clickjacking, and auth bypass flows
|
||||
- **Shell & Command Execution** - Interactive terminal for exploit development and post-exploitation
|
||||
- **Custom Exploit Runtime** - Python sandbox for writing and validating proof-of-concept exploits
|
||||
- **Reconnaissance & OSINT** - Automated attack surface mapping, subdomain enumeration, and fingerprinting
|
||||
- **Static & Dynamic Code Analysis** - SAST + DAST capabilities for comprehensive application security testing
|
||||
- **Vulnerability Knowledge Base** - Structured findings with CVSS scoring and OWASP classification
|
||||
|
||||
### Comprehensive Vulnerability Detection
|
||||
### Comprehensive Vulnerability Scanner
|
||||
|
||||
Strix can identify and validate a wide range of security vulnerabilities:
|
||||
Strix identifies, validates, and exploits a wide range of security vulnerabilities across the OWASP Top 10 and beyond:
|
||||
|
||||
- **Access Control** - IDOR, privilege escalation, auth bypass
|
||||
- **Injection Attacks** - SQL, NoSQL, command injection
|
||||
- **Server-Side** - SSRF, XXE, deserialization flaws
|
||||
- **Client-Side** - XSS, prototype pollution, DOM vulnerabilities
|
||||
- **Business Logic** - Race conditions, workflow manipulation
|
||||
- **Authentication** - JWT vulnerabilities, session management
|
||||
- **Infrastructure** - Misconfigurations, exposed services
|
||||
- **Broken Access Control** - IDOR, privilege escalation, auth bypass
|
||||
- **Injection Attacks** - SQL injection, NoSQL injection, OS command injection, SSTI
|
||||
- **Server-Side Vulnerabilities** - SSRF, XXE, insecure deserialization, RCE
|
||||
- **Client-Side Attacks** - XSS (stored/reflected/DOM), prototype pollution, CSRF
|
||||
- **Business Logic Flaws** - Race conditions, payment manipulation, workflow bypass
|
||||
- **Authentication & Session** - JWT attacks, session fixation, credential stuffing vectors
|
||||
- **Infrastructure & Cloud** - Misconfigurations, exposed services, cloud security issues
|
||||
- **API Security** - Broken authentication, mass assignment, rate limiting bypass
|
||||
|
||||
### Graph of Agents
|
||||
### Graph of Agents (Multi-Agent Pentesting)
|
||||
|
||||
Advanced multi-agent orchestration for comprehensive security testing:
|
||||
Advanced multi-agent orchestration for comprehensive automated penetration testing:
|
||||
|
||||
- **Distributed Workflows** - Specialized agents for different attacks and assets
|
||||
- **Scalable Testing** - Parallel execution for fast comprehensive coverage
|
||||
- **Dynamic Coordination** - Agents collaborate and share discoveries
|
||||
- **Distributed Pentesting** - Specialized AI agents for recon, exploitation, and post-exploitation
|
||||
- **Scalable Security Testing** - Parallel execution across multiple targets for fast, comprehensive coverage
|
||||
- **Dynamic Coordination** - Agents share discoveries, chain vulnerabilities, and collaborate like a red team
|
||||
|
||||
---
|
||||
|
||||
@@ -182,7 +183,7 @@ strix -n --target ./ --scan-mode quick --scope-mode diff --diff-base origin/main
|
||||
|
||||
### Headless Mode
|
||||
|
||||
Run Strix programmatically without interactive UI using the `-n/--non-interactive` flag—perfect for servers and automated jobs. The CLI prints real-time vulnerability findings, and the final report before exiting. Exits with non-zero code when vulnerabilities are found.
|
||||
Run Strix programmatically without interactive UI using the `-n/--non-interactive` flag - perfect for servers and automated jobs. The CLI prints real-time vulnerability findings, and the final report before exiting. Exits with non-zero code when vulnerabilities are found.
|
||||
|
||||
```bash
|
||||
strix -n --target https://your-app.com
|
||||
@@ -239,19 +240,19 @@ export STRIX_REASONING_EFFORT="high" # control thinking effort (default: high,
|
||||
|
||||
**Recommended models for best results:**
|
||||
|
||||
- [OpenAI GPT-5.4](https://openai.com/api/) — `openai/gpt-5.4`
|
||||
- [Anthropic Claude Sonnet 4.6](https://claude.com/platform/api) — `anthropic/claude-sonnet-4-6`
|
||||
- [Google Gemini 3 Pro Preview](https://cloud.google.com/vertex-ai) — `vertex_ai/gemini-3-pro-preview`
|
||||
- [OpenAI GPT-5.4](https://openai.com/api/) - `openai/gpt-5.4`
|
||||
- [Anthropic Claude Sonnet 4.6](https://claude.com/platform/api) - `anthropic/claude-sonnet-4-6`
|
||||
- [Google Gemini 3 Pro Preview](https://cloud.google.com/vertex-ai) - `vertex_ai/gemini-3-pro-preview`
|
||||
|
||||
See the [LLM Providers documentation](https://docs.strix.ai/llm-providers/overview) for all supported providers including Vertex AI, Bedrock, Azure, and local models.
|
||||
|
||||
## Enterprise
|
||||
## Enterprise Pentesting
|
||||
|
||||
Get the same Strix experience with [enterprise-grade](https://strix.ai/demo) controls: SSO (SAML/OIDC), custom compliance reports, dedicated support & SLA, custom deployment options (VPC/self-hosted), BYOK model support, and tailored agents optimized for your environment. [Learn more](https://strix.ai/demo).
|
||||
Get the same Strix experience with [enterprise-grade](https://strix.ai/demo) controls: SSO (SAML/OIDC), custom compliance-ready penetration testing reports (SOC 2, ISO 27001, PCI DSS), dedicated support & SLA, custom deployment options (VPC/self-hosted), BYOK model support, and tailored AI pentesting agents optimized for your environment. [Learn more](https://strix.ai/demo).
|
||||
|
||||
## Documentation
|
||||
|
||||
Full documentation is available at **[docs.strix.ai](https://docs.strix.ai)** — including detailed guides for usage, CI/CD integrations, skills, and advanced configuration.
|
||||
Full documentation is available at **[docs.strix.ai](https://docs.strix.ai)** - including detailed guides for usage, CI/CD integrations, skills, and advanced configuration.
|
||||
|
||||
## Contributing
|
||||
|
||||
|
||||
+20
-23
@@ -136,30 +136,27 @@ def child_initial_input(
|
||||
task: str,
|
||||
parent_history: list[Any],
|
||||
) -> list[dict[str, Any]]:
|
||||
initial_input: list[dict[str, Any]] = []
|
||||
"""Build the initial input for a child agent as a single user message.
|
||||
|
||||
Collapsing the inherited-context block, the identity line, and the task into
|
||||
one ``{"role": "user"}`` message keeps providers that require strictly
|
||||
alternating roles (e.g. Perplexity, llama.cpp) from rejecting consecutive
|
||||
user messages.
|
||||
"""
|
||||
parts: list[str] = []
|
||||
if parent_history:
|
||||
rendered = json.dumps(parent_history, ensure_ascii=False, default=str)
|
||||
initial_input.append(
|
||||
{
|
||||
"role": "user",
|
||||
"content": (
|
||||
"== Inherited context from parent (background only) ==\n"
|
||||
f"{rendered}\n"
|
||||
"== End of inherited context ==\n"
|
||||
"Use the above as background only; do not continue the "
|
||||
"parent's work. Your task follows."
|
||||
),
|
||||
},
|
||||
parts.append(
|
||||
"== Inherited context from parent (background only) ==\n"
|
||||
f"{rendered}\n"
|
||||
"== End of inherited context ==\n"
|
||||
"Use the above as background only; do not continue the "
|
||||
"parent's work. Your task follows.",
|
||||
)
|
||||
initial_input.append(
|
||||
{
|
||||
"role": "user",
|
||||
"content": (
|
||||
f"You are agent {name} ({child_id}); your parent is {parent_id}. "
|
||||
"Maintain your own identity. Call agent_finish when your task "
|
||||
"is complete."
|
||||
),
|
||||
}
|
||||
parts.append(
|
||||
f"You are agent {name} ({child_id}); your parent is {parent_id}. "
|
||||
"Maintain your own identity. Call agent_finish when your task "
|
||||
"is complete.",
|
||||
)
|
||||
initial_input.append({"role": "user", "content": task})
|
||||
return initial_input
|
||||
parts.append(task)
|
||||
return [{"role": "user", "content": "\n\n".join(parts)}]
|
||||
|
||||
@@ -11,6 +11,7 @@ from typing import TYPE_CHECKING, Any
|
||||
|
||||
from agents import RunConfig
|
||||
from agents.sandbox import SandboxRunConfig
|
||||
from openai import RateLimitError
|
||||
|
||||
from strix.agents.factory import build_strix_agent, make_child_factory
|
||||
from strix.config import load_settings
|
||||
@@ -308,6 +309,19 @@ async def run_strix_scan(
|
||||
with contextlib.suppress(Exception):
|
||||
await coordinator.set_status(root_id, "stopped")
|
||||
return None
|
||||
except RateLimitError as exc:
|
||||
logger.warning(
|
||||
"Scan %s stopped: persistent rate limit from the LLM provider (%s). "
|
||||
"Resume with 'strix --resume %s' once the limit clears.",
|
||||
scan_id,
|
||||
exc,
|
||||
scan_id,
|
||||
)
|
||||
if root_id is not None:
|
||||
await coordinator.cancel_descendants(root_id)
|
||||
with contextlib.suppress(Exception):
|
||||
await coordinator.set_status(root_id, "stopped")
|
||||
return None
|
||||
except BaseException:
|
||||
logger.exception("Strix scan %s failed", scan_id)
|
||||
if root_id is not None:
|
||||
|
||||
@@ -820,10 +820,10 @@ def main() -> None:
|
||||
asyncio.run(run_tui(args))
|
||||
except KeyboardInterrupt:
|
||||
exit_reason = "interrupted"
|
||||
except Exception as e:
|
||||
except Exception:
|
||||
exit_reason = "error"
|
||||
posthog.error("unhandled_exception", str(e))
|
||||
scarf.error("unhandled_exception", str(e))
|
||||
posthog.error("unhandled_exception")
|
||||
scarf.error("unhandled_exception")
|
||||
raise
|
||||
finally:
|
||||
report_state = get_global_report_state()
|
||||
|
||||
+18
-16
@@ -3,6 +3,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import csv
|
||||
import io
|
||||
import json
|
||||
import logging
|
||||
import tempfile
|
||||
@@ -58,9 +59,9 @@ def write_vulnerabilities(
|
||||
new_reports = [r for r in vulnerability_reports if r["id"] not in saved_vuln_ids]
|
||||
|
||||
for report in new_reports:
|
||||
(vuln_dir / f"{report['id']}.md").write_text(
|
||||
_atomic_write_text(
|
||||
vuln_dir / f"{report['id']}.md",
|
||||
render_vulnerability_md(report),
|
||||
encoding="utf-8",
|
||||
)
|
||||
saved_vuln_ids.add(report["id"])
|
||||
|
||||
@@ -69,20 +70,21 @@ def write_vulnerabilities(
|
||||
key=lambda r: (_SEVERITY_ORDER.get(r["severity"], 5), r["timestamp"]),
|
||||
)
|
||||
csv_path = run_dir / "vulnerabilities.csv"
|
||||
with csv_path.open("w", encoding="utf-8", newline="") as f:
|
||||
fieldnames = ["id", "title", "severity", "timestamp", "file"]
|
||||
writer = csv.DictWriter(f, fieldnames=fieldnames)
|
||||
writer.writeheader()
|
||||
for report in sorted_reports:
|
||||
writer.writerow(
|
||||
{
|
||||
"id": report["id"],
|
||||
"title": report["title"],
|
||||
"severity": report["severity"].upper(),
|
||||
"timestamp": report["timestamp"],
|
||||
"file": f"vulnerabilities/{report['id']}.md",
|
||||
},
|
||||
)
|
||||
csv_buf = io.StringIO()
|
||||
fieldnames = ["id", "title", "severity", "timestamp", "file"]
|
||||
csv_writer = csv.DictWriter(csv_buf, fieldnames=fieldnames, lineterminator="\r\n")
|
||||
csv_writer.writeheader()
|
||||
for report in sorted_reports:
|
||||
csv_writer.writerow(
|
||||
{
|
||||
"id": report["id"],
|
||||
"title": report["title"],
|
||||
"severity": report["severity"].upper(),
|
||||
"timestamp": report["timestamp"],
|
||||
"file": f"vulnerabilities/{report['id']}.md",
|
||||
},
|
||||
)
|
||||
_atomic_write_text(csv_path, csv_buf.getvalue())
|
||||
|
||||
_atomic_write_text(
|
||||
run_dir / "vulnerabilities.json",
|
||||
|
||||
@@ -123,8 +123,6 @@ def end(report_state: "ReportState", exit_reason: str = "completed") -> None:
|
||||
)
|
||||
|
||||
|
||||
def error(error_type: str, error_msg: str | None = None) -> None:
|
||||
def error(error_type: str) -> None:
|
||||
props = {**base_props(), "error_type": error_type}
|
||||
if error_msg:
|
||||
props["error_msg"] = error_msg
|
||||
_send("error", props)
|
||||
|
||||
@@ -129,12 +129,10 @@ def end(report_state: ReportState, exit_reason: str = "completed") -> None:
|
||||
)
|
||||
|
||||
|
||||
def error(error_type: str, error_msg: str | None = None) -> None:
|
||||
def error(error_type: str) -> None:
|
||||
props: dict[str, Any] = {
|
||||
**base_props(),
|
||||
"session": SESSION_ID,
|
||||
"error_type": error_type,
|
||||
}
|
||||
if error_msg:
|
||||
props["error_msg"] = error_msg
|
||||
_send("error", props)
|
||||
|
||||
@@ -0,0 +1,178 @@
|
||||
"""Tests for strix.config.loader: JSON overrides, alias resolution, persistence."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import pytest
|
||||
from pydantic import AliasChoices, Field
|
||||
from pydantic.fields import FieldInfo
|
||||
|
||||
from strix.config import loader
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
_LLM_ENV_KEYS = [
|
||||
"STRIX_LLM",
|
||||
"LLM_API_KEY",
|
||||
"OPENAI_API_KEY",
|
||||
"LLM_API_BASE",
|
||||
"OPENAI_API_BASE",
|
||||
"OPENAI_BASE_URL",
|
||||
"LITELLM_BASE_URL",
|
||||
"OLLAMA_API_BASE",
|
||||
"STRIX_REASONING_EFFORT",
|
||||
"LLM_TIMEOUT",
|
||||
"PERPLEXITY_API_KEY",
|
||||
# RuntimeSettings
|
||||
"STRIX_IMAGE",
|
||||
"STRIX_RUNTIME_BACKEND",
|
||||
"STRIX_MAX_LOCAL_COPY_MB",
|
||||
# TelemetrySettings
|
||||
"STRIX_TELEMETRY",
|
||||
]
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _reset_loader_state(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Reset module globals and clear known env vars for deterministic runs."""
|
||||
for key in _LLM_ENV_KEYS:
|
||||
monkeypatch.delenv(key, raising=False)
|
||||
monkeypatch.setattr(loader, "_cached", None)
|
||||
monkeypatch.setattr(loader, "_override", None)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# _read_json_overrides
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def test_read_json_overrides_missing_file(tmp_path: Path) -> None:
|
||||
assert loader._read_json_overrides(tmp_path / "nope.json") == {}
|
||||
|
||||
|
||||
def test_read_json_overrides_corrupt_json(tmp_path: Path) -> None:
|
||||
path = tmp_path / "cli-config.json"
|
||||
path.write_text("{not valid json", encoding="utf-8")
|
||||
assert loader._read_json_overrides(path) == {}
|
||||
|
||||
|
||||
def test_read_json_overrides_non_dict_env(tmp_path: Path) -> None:
|
||||
path = tmp_path / "cli-config.json"
|
||||
path.write_text(json.dumps({"env": ["not", "a", "dict"]}), encoding="utf-8")
|
||||
assert loader._read_json_overrides(path) == {}
|
||||
|
||||
|
||||
def test_read_json_overrides_maps_to_nested_settings(tmp_path: Path) -> None:
|
||||
path = tmp_path / "cli-config.json"
|
||||
path.write_text(
|
||||
json.dumps({"env": {"STRIX_LLM": "my-model", "PERPLEXITY_API_KEY": "pk"}}),
|
||||
encoding="utf-8",
|
||||
)
|
||||
assert loader._read_json_overrides(path) == {
|
||||
"llm": {"model": "my-model"},
|
||||
"integrations": {"perplexity_api_key": "pk"},
|
||||
}
|
||||
|
||||
|
||||
def test_read_json_overrides_skips_keys_already_in_environ(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
monkeypatch.setenv("STRIX_LLM", "from-env")
|
||||
path = tmp_path / "cli-config.json"
|
||||
path.write_text(json.dumps({"env": {"STRIX_LLM": "from-file"}}), encoding="utf-8")
|
||||
# env wins -> the JSON value is not surfaced as an init kwarg.
|
||||
assert loader._read_json_overrides(path) == {}
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# _aliases_for
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def test_aliases_for_simple_alias() -> None:
|
||||
finfo = FieldInfo(alias="SIMPLE_ALIAS")
|
||||
assert loader._aliases_for(finfo) == ["SIMPLE_ALIAS"]
|
||||
|
||||
|
||||
def test_aliases_for_alias_choices() -> None:
|
||||
finfo: FieldInfo = Field( # type: ignore[assignment]
|
||||
default=None,
|
||||
validation_alias=AliasChoices("FIRST", "SECOND"),
|
||||
)
|
||||
assert loader._aliases_for(finfo) == ["FIRST", "SECOND"]
|
||||
|
||||
|
||||
def test_aliases_for_string_validation_alias() -> None:
|
||||
finfo: FieldInfo = Field(default=None, validation_alias="STR_ALIAS") # type: ignore[assignment]
|
||||
assert loader._aliases_for(finfo) == ["STR_ALIAS"]
|
||||
|
||||
|
||||
def test_aliases_for_no_alias() -> None:
|
||||
assert loader._aliases_for(FieldInfo()) == []
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# apply_config_override + load_settings round-trip
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def test_apply_override_and_load_settings_round_trip(tmp_path: Path) -> None:
|
||||
path = tmp_path / "cli-config.json"
|
||||
path.write_text(
|
||||
json.dumps({"env": {"STRIX_LLM": "round-trip-model", "PERPLEXITY_API_KEY": "pk"}}),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
loader.apply_config_override(path)
|
||||
settings = loader.load_settings()
|
||||
|
||||
assert settings.llm.model == "round-trip-model"
|
||||
assert settings.integrations.perplexity_api_key == "pk"
|
||||
# Second call is memoized -> same object.
|
||||
assert loader.load_settings() is settings
|
||||
|
||||
|
||||
def test_apply_config_override_invalidates_cache(tmp_path: Path) -> None:
|
||||
first = tmp_path / "first.json"
|
||||
first.write_text(json.dumps({"env": {"STRIX_LLM": "first-model"}}), encoding="utf-8")
|
||||
second = tmp_path / "second.json"
|
||||
second.write_text(json.dumps({"env": {"STRIX_LLM": "second-model"}}), encoding="utf-8")
|
||||
|
||||
loader.apply_config_override(first)
|
||||
assert loader.load_settings().llm.model == "first-model"
|
||||
|
||||
loader.apply_config_override(second)
|
||||
assert loader.load_settings().llm.model == "second-model"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# persist_current
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
|
||||
def test_persist_current_writes_env_block(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setenv("STRIX_LLM", "persisted-model")
|
||||
target = tmp_path / "sub" / "cli-config.json"
|
||||
loader.apply_config_override(target)
|
||||
|
||||
loader.persist_current()
|
||||
|
||||
assert target.exists()
|
||||
assert json.loads(target.read_text(encoding="utf-8")) == {
|
||||
"env": {"STRIX_LLM": "persisted-model"}
|
||||
}
|
||||
|
||||
|
||||
def test_persist_current_sets_0600_mode(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setenv("STRIX_LLM", "persisted-model")
|
||||
target = tmp_path / "cli-config.json"
|
||||
loader.apply_config_override(target)
|
||||
|
||||
loader.persist_current()
|
||||
|
||||
assert target.stat().st_mode & 0o777 == 0o600
|
||||
@@ -0,0 +1,114 @@
|
||||
"""Tests for pure input builders in strix.core.inputs."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from itertools import pairwise
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from strix.core.inputs import build_root_task, child_initial_input
|
||||
|
||||
|
||||
def _child_kwargs(parent_history: list[Any]) -> dict[str, Any]:
|
||||
return {
|
||||
"name": "scout",
|
||||
"child_id": "agent-2",
|
||||
"parent_id": "agent-1",
|
||||
"task": "Audit the login flow.",
|
||||
"parent_history": parent_history,
|
||||
}
|
||||
|
||||
|
||||
def test_child_initial_input_single_message_without_history() -> None:
|
||||
result = child_initial_input(**_child_kwargs([]))
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0]["role"] == "user"
|
||||
content = result[0]["content"]
|
||||
assert "agent scout (agent-2)" in content
|
||||
assert "Audit the login flow." in content
|
||||
assert "Inherited context" not in content
|
||||
|
||||
|
||||
def test_child_initial_input_single_message_with_history() -> None:
|
||||
history = [{"role": "assistant", "content": "previous work"}]
|
||||
result = child_initial_input(**_child_kwargs(history))
|
||||
|
||||
assert len(result) == 1
|
||||
assert result[0]["role"] == "user"
|
||||
content = result[0]["content"]
|
||||
assert "Inherited context from parent" in content
|
||||
assert "previous work" in content
|
||||
assert "agent scout (agent-2)" in content
|
||||
assert "Audit the login flow." in content
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"parent_history",
|
||||
[[], [{"role": "assistant", "content": "previous work"}]],
|
||||
)
|
||||
def test_child_initial_input_no_consecutive_same_role(parent_history: list[Any]) -> None:
|
||||
result = child_initial_input(**_child_kwargs(parent_history))
|
||||
|
||||
roles = [msg["role"] for msg in result]
|
||||
assert all(prev != nxt for prev, nxt in pairwise(roles))
|
||||
|
||||
|
||||
def test_build_root_task_empty_config() -> None:
|
||||
assert build_root_task({}) == ""
|
||||
|
||||
|
||||
def test_build_root_task_repository_target() -> None:
|
||||
config = {
|
||||
"targets": [
|
||||
{
|
||||
"type": "repository",
|
||||
"details": {
|
||||
"target_repo": "https://example.com/repo.git",
|
||||
"cloned_repo_path": "/workspace/repo",
|
||||
"workspace_subdir": "repo",
|
||||
},
|
||||
},
|
||||
],
|
||||
}
|
||||
task = build_root_task(config)
|
||||
|
||||
assert "Repositories:" in task
|
||||
assert "/workspace/repo" in task
|
||||
assert "https://example.com/repo.git" in task
|
||||
|
||||
|
||||
def test_build_root_task_web_application_with_instructions() -> None:
|
||||
config = {
|
||||
"targets": [
|
||||
{"type": "web_application", "details": {"target_url": "https://app.example.com"}},
|
||||
],
|
||||
"user_instructions": "Focus on auth.",
|
||||
}
|
||||
task = build_root_task(config)
|
||||
|
||||
assert "URLs:" in task
|
||||
assert "https://app.example.com" in task
|
||||
assert "Special instructions: Focus on auth." in task
|
||||
|
||||
|
||||
def test_build_root_task_diff_scope() -> None:
|
||||
config = {
|
||||
"targets": [],
|
||||
"diff_scope": {
|
||||
"active": True,
|
||||
"repos": [
|
||||
{
|
||||
"workspace_subdir": "repo",
|
||||
"analyzable_files_count": 3,
|
||||
"deleted_files_count": 2,
|
||||
},
|
||||
],
|
||||
},
|
||||
}
|
||||
task = build_root_task(config)
|
||||
|
||||
assert "Scope Constraints:" in task
|
||||
assert "3 changed file(s)" in task
|
||||
assert "2 deleted file(s)" in task
|
||||
@@ -0,0 +1,84 @@
|
||||
"""Tests for graceful handling of persistent RateLimitError in run_strix_scan."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import types
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
from openai import RateLimitError
|
||||
|
||||
import strix.tools.notes.tools as notes_tools
|
||||
import strix.tools.todo.tools as todo_tools
|
||||
from strix.core import runner
|
||||
from strix.core.agents import AgentCoordinator
|
||||
|
||||
|
||||
def _make_rate_limit_error() -> RateLimitError:
|
||||
request = httpx.Request("POST", "https://api.openai.com/v1/responses")
|
||||
response = httpx.Response(status_code=429, request=request)
|
||||
return RateLimitError("rate limited", response=response, body=None)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_persistent_rate_limit_stops_gracefully(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Any, caplog: pytest.LogCaptureFixture
|
||||
) -> None:
|
||||
"""A persistent RateLimitError stops the scan (root -> 'stopped') without raising."""
|
||||
monkeypatch.setattr(runner, "run_dir_for", lambda _scan_id: tmp_path)
|
||||
monkeypatch.setattr(runner, "runtime_state_dir", lambda _run_dir: tmp_path)
|
||||
monkeypatch.setattr(runner, "setup_scan_logging", lambda _run_dir: lambda: None)
|
||||
monkeypatch.setattr(runner, "set_scan_id", lambda _scan_id: None)
|
||||
|
||||
settings = types.SimpleNamespace(
|
||||
llm=types.SimpleNamespace(model="openai/gpt-4o", reasoning_effort="high")
|
||||
)
|
||||
monkeypatch.setattr(runner, "load_settings", lambda: settings)
|
||||
monkeypatch.setattr(runner, "configure_sdk_model_defaults", lambda _settings: None)
|
||||
monkeypatch.setattr(
|
||||
runner, "uses_chat_completions_tool_schema", lambda _model, _settings: False
|
||||
)
|
||||
|
||||
monkeypatch.setattr(todo_tools, "hydrate_todos_from_disk", lambda _state_dir: None)
|
||||
monkeypatch.setattr(notes_tools, "hydrate_notes_from_disk", lambda _state_dir: None)
|
||||
|
||||
async def _create_or_reuse(*_args: Any, **_kwargs: Any) -> dict[str, Any]:
|
||||
return {"client": object(), "session": object(), "caido_client": None}
|
||||
|
||||
async def _cleanup(*_args: Any, **_kwargs: Any) -> None:
|
||||
return None
|
||||
|
||||
monkeypatch.setattr(runner.session_manager, "create_or_reuse", _create_or_reuse)
|
||||
monkeypatch.setattr(runner.session_manager, "cleanup", _cleanup)
|
||||
|
||||
monkeypatch.setattr(runner, "build_root_task", lambda _scan_config: "task")
|
||||
monkeypatch.setattr(runner, "build_scope_context", lambda _scan_config: "")
|
||||
monkeypatch.setattr(runner, "make_model_settings", lambda *_args, **_kwargs: object())
|
||||
monkeypatch.setattr(runner, "build_strix_agent", lambda **_kwargs: object())
|
||||
monkeypatch.setattr(runner, "make_child_factory", lambda **_kwargs: lambda **_k: object())
|
||||
monkeypatch.setattr(runner, "open_agent_session", lambda _root_id, _db: object())
|
||||
|
||||
async def _raise_rate_limit(*_args: Any, **_kwargs: Any) -> None:
|
||||
raise _make_rate_limit_error()
|
||||
|
||||
monkeypatch.setattr(runner, "run_agent_loop", _raise_rate_limit)
|
||||
|
||||
coordinator = AgentCoordinator()
|
||||
|
||||
with caplog.at_level(logging.WARNING):
|
||||
result = await runner.run_strix_scan(
|
||||
scan_config={"targets": [], "scan_mode": "deep"},
|
||||
scan_id="scan-test",
|
||||
image="img",
|
||||
coordinator=coordinator,
|
||||
)
|
||||
|
||||
assert result is None
|
||||
root_ids = [aid for aid, parent in coordinator.parent_of.items() if parent is None]
|
||||
assert len(root_ids) == 1
|
||||
assert coordinator.statuses[root_ids[0]] == "stopped"
|
||||
# the resume hint must carry the real scan id, not a literal placeholder
|
||||
assert "strix --resume scan-test" in caplog.text
|
||||
assert "<run_name>" not in caplog.text
|
||||
Reference in New Issue
Block a user