mirror of
https://github.com/usestrix/strix.git
synced 2026-10-29 06:32:26 +00:00
POC : Added native MCP support to Strix,
enabling seamless integration with coding agents for enhanced functionality and automation. This commit includes new modules for agent launching, MCP installation, server management, and service handling, along with updated documentation and test cases to ensure robust performance.
This commit is contained in:
kaddour-youcef
parent
302efedca6
commit
1cfeb960bd
21 changed files with 1550 additions and 19 deletions
14
README.md
14
README.md
|
|
@ -72,7 +72,7 @@ Strix are autonomous AI penetration testing agents that act just like real hacke
|
|||
|
||||
**Prerequisites:**
|
||||
- Docker (running)
|
||||
- An LLM API key from any [supported provider](https://docs.strix.ai/llm-providers/overview) (OpenAI, Anthropic, Google, etc.)
|
||||
- Codex or Claude Code authenticated with your normal account, or an LLM API key for legacy provider mode
|
||||
|
||||
### Installation & First Scan
|
||||
|
||||
|
|
@ -80,14 +80,12 @@ Strix are autonomous AI penetration testing agents that act just like real hacke
|
|||
# Install Strix
|
||||
curl -sSL https://strix.ai/install | bash
|
||||
|
||||
# Configure your AI provider
|
||||
export STRIX_LLM="openai/gpt-5.4"
|
||||
export LLM_API_KEY="your-api-key"
|
||||
|
||||
# Run your first security assessment
|
||||
# Run through Codex or Claude Code; no model API key is needed
|
||||
strix --target ./app-directory
|
||||
```
|
||||
|
||||
When `STRIX_LLM` is unset, Strix launches an installed Codex or Claude Code CLI and exposes its security runtime through the local `strix-mcp` server. Every model call stays inside that coding agent. Use `--agent codex` or `--agent claude` to select one explicitly. See [coding-agent integration](docs/integrations/coding-agents.mdx) for standalone MCP and skill installation.
|
||||
|
||||
> [!NOTE]
|
||||
> First run automatically pulls the sandbox Docker image. Results are saved to `strix_runs/<run-name>`
|
||||
|
||||
|
|
@ -223,7 +221,9 @@ jobs:
|
|||
> If diff-scope cannot resolve, ensure checkout uses full history (`fetch-depth: 0`) or pass
|
||||
> `--diff-base` explicitly.
|
||||
|
||||
### Configuration
|
||||
### Legacy Provider Configuration (Optional)
|
||||
|
||||
Codex/Claude agent mode requires no configuration here. For direct provider mode, set:
|
||||
|
||||
```bash
|
||||
export STRIX_LLM="openai/gpt-5.4"
|
||||
|
|
|
|||
|
|
@ -45,6 +45,7 @@
|
|||
{
|
||||
"group": "Integrations",
|
||||
"pages": [
|
||||
"integrations/coding-agents",
|
||||
"integrations/github-actions",
|
||||
"integrations/ci-cd"
|
||||
]
|
||||
|
|
|
|||
|
|
@ -77,11 +77,7 @@ Strix uses a graph of specialized agents for comprehensive security testing:
|
|||
# Install
|
||||
curl -sSL https://strix.ai/install | bash
|
||||
|
||||
# Configure
|
||||
export STRIX_LLM="openai/gpt-5.4"
|
||||
export LLM_API_KEY="your-api-key"
|
||||
|
||||
# Scan
|
||||
# Scan through an authenticated Codex or Claude Code CLI
|
||||
strix --target ./your-app
|
||||
```
|
||||
|
||||
|
|
|
|||
393
docs/integrations/coding-agents.mdx
Normal file
393
docs/integrations/coding-agents.mdx
Normal file
|
|
@ -0,0 +1,393 @@
|
|||
---
|
||||
title: "Coding-agent runtime"
|
||||
description: "Architecture, installation, operation, and limitations of the Strix skill and MCP proof of concept"
|
||||
---
|
||||
|
||||
<Warning>
|
||||
This integration is a skeleton and proof of concept. It demonstrates that Strix can move model reasoning into Codex, Claude Code, or another MCP-capable coding agent while retaining the Strix sandbox, proxy, security knowledge, findings, and report artifacts. It is usable for experimentation, but it does not yet provide complete behavioral parity with the legacy Strix multi-agent runtime. Review the limitations and roadmap before relying on it for production assessments.
|
||||
</Warning>
|
||||
|
||||
## Goal
|
||||
|
||||
The coding-agent runtime removes the requirement for Strix to hold a separate model-provider API key. Instead of constructing and running an OpenAI Agents SDK model inside Strix, it uses an already authenticated coding agent as the reasoning and orchestration layer.
|
||||
|
||||
The implementation separates the application into two parts:
|
||||
|
||||
- **Skill:** teaches the host coding agent how to conduct a Strix assessment, enforce scope, select security knowledge, validate findings, and finish the report.
|
||||
- **MCP server:** exposes model-free Strix operations such as scan initialization, Docker sandbox execution, Caido traffic inspection, security knowledge retrieval, finding persistence, and report generation.
|
||||
|
||||
All LLM inference in this mode happens inside the selected coding agent. The `strix-mcp` process does not instantiate a model, call a model endpoint, or require `LLM_API_KEY`, `OPENAI_API_KEY`, or `ANTHROPIC_API_KEY`.
|
||||
|
||||
The coding agent still needs its normal authentication, subscription, or provider configuration. “No model API key” means that Strix does not need an additional provider credential; it does not mean that Codex or Claude Code can operate without being authenticated.
|
||||
|
||||
## Current status
|
||||
|
||||
This proof of concept establishes the following architecture and compatibility points:
|
||||
|
||||
- The existing `strix --target ...` entry point can launch Codex or Claude Code.
|
||||
- The same target types, scan modes, instruction input, diff scoping, Docker sandbox, Caido proxy, vulnerability artifacts, and final report format are reused.
|
||||
- The skill and MCP server can be installed separately and invoked directly from an agent.
|
||||
- The MCP server has a real stdio protocol boundary and can be used by clients other than the built-in launcher.
|
||||
- MCP-mode vulnerability deduplication cannot silently fall back to a Strix-owned model call.
|
||||
|
||||
It does **not** yet mean that the legacy agent graph, interactive TUI, conversation persistence, cost accounting, or autonomous subagent behavior have been reproduced exactly. Those gaps are intentional and documented below.
|
||||
|
||||
## Architecture
|
||||
|
||||
```mermaid
|
||||
flowchart LR
|
||||
U[User] --> CLI[Strix CLI]
|
||||
CLI --> L[Agent launcher]
|
||||
L --> A[Codex / Claude Code]
|
||||
S[Strix security skill] --> A
|
||||
A <-->|MCP over stdio| M[strix-mcp]
|
||||
M --> K[Security knowledge]
|
||||
M --> D[Docker sandbox]
|
||||
D --> C[Caido proxy]
|
||||
M --> R[Findings and reports]
|
||||
A --> O[Agent-native final response]
|
||||
```
|
||||
|
||||
### Responsibility split
|
||||
|
||||
| Component | Responsibility | Does it call an LLM? |
|
||||
| --- | --- | --- |
|
||||
| `strix` CLI | Parse existing arguments, select a runtime, ensure Docker is available, and launch the coding agent | No |
|
||||
| Agent launcher | Build an ephemeral MCP configuration and a scan prompt for Codex or Claude Code | No |
|
||||
| Codex / Claude Code | Plan the assessment, reason over evidence, choose tools, optionally coordinate native subagents, and write report narratives | Yes, through the agent's own authenticated runtime |
|
||||
| `strix-security` skill | Provide the scan workflow, safety constraints, reporting quality bar, and MCP usage instructions | No |
|
||||
| `strix-mcp` | Expose Strix capabilities through MCP and own one active scan lifecycle | No |
|
||||
| Docker sandbox | Execute security tooling in isolation and route HTTP traffic through Caido | No |
|
||||
| Report state | Persist targets, findings, SARIF, CSV/JSON/Markdown, and final report sections | No |
|
||||
|
||||
## Runtime selection
|
||||
|
||||
The CLI supports four values for `--agent`:
|
||||
|
||||
| Value | Behavior |
|
||||
| --- | --- |
|
||||
| `auto` | Use legacy mode when `STRIX_LLM` is configured; otherwise select a coding-agent CLI |
|
||||
| `codex` | Explicitly launch Codex and inject an ephemeral Strix MCP server |
|
||||
| `claude` | Explicitly launch Claude Code and inject an ephemeral Strix MCP server |
|
||||
| `legacy` | Use the existing in-process provider and OpenAI Agents SDK runtime |
|
||||
|
||||
When coding-agent selection is automatic, `STRIX_AGENT` is checked first. If it is not set, Strix selects `codex` when available, followed by `claude`.
|
||||
|
||||
```bash
|
||||
# Automatically use an installed coding agent when STRIX_LLM is unset
|
||||
strix --target ./your-app
|
||||
|
||||
# Select a host explicitly
|
||||
strix --agent codex --target ./your-app
|
||||
strix --agent claude --target https://staging.example.com
|
||||
|
||||
# Force the previous direct-provider runtime
|
||||
strix --agent legacy --target ./your-app
|
||||
```
|
||||
|
||||
## End-to-end execution flow
|
||||
|
||||
### 1. CLI parsing and compatibility
|
||||
|
||||
The existing argument parser still processes targets, instructions, scan mode, scope mode, diff base, mounts, budget, and resume options. In coding-agent mode, the launcher converts those values into a structured `start_scan` call embedded in the initial agent prompt.
|
||||
|
||||
For Codex, the launcher passes an ephemeral `mcp_servers.strix` configuration through Codex configuration overrides. For Claude Code, it passes an inline JSON MCP configuration with `--mcp-config`. The prompt is separated from Claude's variadic config arguments with `--`, preventing Claude from interpreting the prompt as another configuration filename.
|
||||
|
||||
### 2. Skill loading
|
||||
|
||||
The launcher tells the coding agent to read the packaged `strix-security/SKILL.md` before acting. The same skill can also be installed into the agent's normal skill directory for direct invocation.
|
||||
|
||||
The skill requires the agent to:
|
||||
|
||||
1. Respect the verified target scope.
|
||||
2. Start the MCP scan lifecycle once.
|
||||
3. Load only relevant Strix security knowledge.
|
||||
4. Build and execute a test plan.
|
||||
5. Validate suspected issues with safe, reproducible proofs of concept.
|
||||
6. Review existing findings before filing another.
|
||||
7. Produce a customer-facing final report.
|
||||
8. Stop and persist the run if the assessment cannot be completed.
|
||||
|
||||
The skill is intentionally concise. Detailed tool selection and CVSS field guidance live in `references/mcp-tools.md` and are loaded only when needed.
|
||||
|
||||
### 3. Scan initialization
|
||||
|
||||
The agent calls `start_scan`. The MCP service then:
|
||||
|
||||
1. Validates scan mode and target input.
|
||||
2. Infers each target type using existing Strix logic.
|
||||
3. Resolves and deduplicates local paths and mounts.
|
||||
4. Clones remote repository targets when necessary.
|
||||
5. Rewrites localhost targets to `host.docker.internal` for access from Docker.
|
||||
6. Applies the existing local-copy size limit.
|
||||
7. Resolves full, automatic, or explicit diff scope.
|
||||
8. Creates or hydrates `ReportState` under `strix_runs/<run-name>`.
|
||||
9. Starts the existing Strix Docker sandbox and Caido sidecar.
|
||||
10. Returns the authorized scope, task, output directory, and `/workspace/<name>` mappings to the coding agent.
|
||||
|
||||
One MCP server process owns one active scan at a time. A second `start_scan` call is rejected until the current scan is finished or stopped.
|
||||
|
||||
### 4. Security execution
|
||||
|
||||
The coding agent calls `sandbox_exec` with an argv array. Commands execute inside the existing Strix security container rather than directly in the host-agent workspace.
|
||||
|
||||
The sandbox retains the existing tooling model:
|
||||
|
||||
- source targets are copied or mounted under `/workspace`;
|
||||
- common reconnaissance and security tools are available;
|
||||
- browser automation runs through the `agent-browser` CLI;
|
||||
- Python and curl can be used for custom validation;
|
||||
- HTTP and HTTPS traffic are routed through the in-container Caido proxy.
|
||||
|
||||
The proof of concept currently supports bounded, synchronous command execution. It does not expose the legacy `write_stdin`/interactive-process lifecycle through MCP yet.
|
||||
|
||||
### 5. Proxy inspection and replay
|
||||
|
||||
Traffic produced in the sandbox is available through MCP wrappers around the existing Caido API:
|
||||
|
||||
- list captured requests with HTTPQL filters;
|
||||
- inspect request or response bodies with pagination or regex search;
|
||||
- replay a request with URL, parameter, header, cookie, or body changes;
|
||||
- browse the generated sitemap;
|
||||
- create and manage Caido scope rules.
|
||||
|
||||
This preserves the existing browser → capture → inspect → modify → validate workflow while allowing the host coding agent to drive it.
|
||||
|
||||
### 6. Knowledge loading
|
||||
|
||||
Strix's existing security Markdown library remains the knowledge source. `list_knowledge` exposes available framework, technology, protocol, tooling, cloud, and vulnerability modules. `load_knowledge` returns one exact module on demand.
|
||||
|
||||
Scan-mode guidance is also loaded through the same mechanism, for example:
|
||||
|
||||
```text
|
||||
scan_modes/quick
|
||||
scan_modes/standard
|
||||
scan_modes/deep
|
||||
frameworks/nextjs
|
||||
protocols/graphql
|
||||
vulnerabilities/idor
|
||||
tooling/agent_browser
|
||||
```
|
||||
|
||||
The host agent's native web-search capability should be used for current external research. The Perplexity-backed legacy `web_search` tool is not exposed by this MCP skeleton.
|
||||
|
||||
### 7. Finding creation
|
||||
|
||||
`create_vulnerability_report` reuses the existing Strix validation, CVSS calculation, artifact generation, and `ReportState` persistence code.
|
||||
|
||||
MCP mode explicitly sets `allow_model_dedupe=false`. This is important: the existing semantic dedupe implementation can call the configured Strix model, which would violate the agent-native model boundary. The proof of concept therefore performs exact normalized duplicate detection on title, target, endpoint, and method. The coding agent is responsible for semantic review through `list_findings` before filing.
|
||||
|
||||
Reports still require:
|
||||
|
||||
- a verified title and affected target;
|
||||
- technical analysis and impact;
|
||||
- a working PoC description and payload/code;
|
||||
- remediation guidance;
|
||||
- all eight CVSS 3.1 base metrics;
|
||||
- repository-relative code locations for white-box findings.
|
||||
|
||||
### 8. Finalization and artifacts
|
||||
|
||||
The agent calls `finish_scan` with four report sections:
|
||||
|
||||
- executive summary;
|
||||
- methodology;
|
||||
- consolidated technical analysis;
|
||||
- prioritized recommendations.
|
||||
|
||||
Strix marks the run complete, writes its normal artifacts, and tears down the sandbox. `stop_scan` provides the incomplete-run path and preserves state before cleanup.
|
||||
|
||||
Artifacts remain under `strix_runs/<run-name>`:
|
||||
|
||||
```text
|
||||
strix_runs/<run-name>/
|
||||
├── run.json
|
||||
├── penetration_test_report.md
|
||||
├── findings.sarif
|
||||
├── vulnerabilities.csv
|
||||
├── vulnerabilities.json
|
||||
└── vulnerabilities/
|
||||
└── vuln-0001.md
|
||||
```
|
||||
|
||||
## MCP tool surface
|
||||
|
||||
The proof-of-concept server exposes 15 tools:
|
||||
|
||||
| Tool | Purpose |
|
||||
| --- | --- |
|
||||
| `start_scan` | Create or resume report state and start the sandbox |
|
||||
| `sandbox_exec` | Execute argv-form commands inside the sandbox |
|
||||
| `list_knowledge` | List built-in Strix security modules |
|
||||
| `load_knowledge` | Load one knowledge module |
|
||||
| `list_proxy_requests` | Query captured traffic using HTTPQL |
|
||||
| `view_proxy_request` | Read a captured request or response |
|
||||
| `repeat_proxy_request` | Replay captured traffic with modifications |
|
||||
| `list_sitemap` | Browse the Caido sitemap hierarchy |
|
||||
| `view_sitemap_entry` | Inspect one sitemap entry and recent traffic |
|
||||
| `manage_scope` | Manage Caido allow/deny scope rules |
|
||||
| `create_vulnerability_report` | Validate and persist one verified finding |
|
||||
| `list_findings` | Review already persisted findings |
|
||||
| `scan_status` | Inspect lifecycle state and output location |
|
||||
| `finish_scan` | Complete reports and clean up the sandbox |
|
||||
| `stop_scan` | Persist an incomplete run and clean up the sandbox |
|
||||
|
||||
The server uses newline-delimited JSON-RPC over stdio. Its local transport reads and writes pipe file descriptors through the event loop because the worker-thread-backed stdio wrapper can hang in some managed shells. The protocol and FastMCP server core remain standard MCP.
|
||||
|
||||
## Installation
|
||||
|
||||
### Prerequisites
|
||||
|
||||
- Docker installed and running.
|
||||
- Strix installed in the current Python environment.
|
||||
- Codex or Claude Code installed and authenticated.
|
||||
- The coding agent allowed to start the local stdio MCP process.
|
||||
|
||||
The packaged commands are:
|
||||
|
||||
```text
|
||||
strix Existing CLI plus coding-agent runtime selection
|
||||
strix-mcp Local MCP stdio server
|
||||
strix-skill-path Prints the packaged portable skill directory
|
||||
```
|
||||
|
||||
### Use the normal CLI
|
||||
|
||||
No separate skill or MCP registration is required when using `strix --agent ...`; the launcher injects both the prompt and an ephemeral MCP configuration.
|
||||
|
||||
```bash
|
||||
strix --agent codex -t ./app -t https://staging.example.com
|
||||
strix --agent claude -t ./app -t http://localhost:8080
|
||||
strix --agent codex --resume previous-run-name
|
||||
```
|
||||
|
||||
### Install into Codex
|
||||
|
||||
```bash
|
||||
codex mcp add strix -- strix-mcp
|
||||
mkdir -p ~/.codex/skills
|
||||
cp -R "$(strix-skill-path)" ~/.codex/skills/strix-security
|
||||
```
|
||||
|
||||
Then ask:
|
||||
|
||||
```text
|
||||
Use $strix-security to run an authorized assessment of ./your-app.
|
||||
```
|
||||
|
||||
### Install into Claude Code
|
||||
|
||||
```bash
|
||||
claude mcp add --scope user strix -- strix-mcp
|
||||
mkdir -p ~/.claude/skills
|
||||
cp -R "$(strix-skill-path)" ~/.claude/skills/strix-security
|
||||
```
|
||||
|
||||
Then ask:
|
||||
|
||||
```text
|
||||
Use /strix-security to run an authorized assessment of ./your-app.
|
||||
```
|
||||
|
||||
### Install into another coding agent
|
||||
|
||||
Configure a local stdio MCP server with `strix-mcp` as its command. Attach `strix-security/SKILL.md` through that agent's reusable instruction or skill mechanism. The minimum client requirements are:
|
||||
|
||||
- MCP tool discovery and invocation over stdio;
|
||||
- local subprocess support;
|
||||
- permission to start Docker-backed operations;
|
||||
- sufficient tool-call and context limits for a security assessment.
|
||||
|
||||
The built-in launcher currently has adapters only for Codex and Claude Code. Other agents use manual MCP/skill configuration in this skeleton.
|
||||
|
||||
## What remains compatible
|
||||
|
||||
- Local code, Git repository, URL, domain, IP, and combined target input.
|
||||
- `quick`, `standard`, and `deep` scan modes.
|
||||
- Inline and file-based instructions.
|
||||
- Full, automatic, and explicit diff scope.
|
||||
- Read-only mounts for large repositories.
|
||||
- Localhost target access from the sandbox.
|
||||
- Docker isolation, browser tooling, and Caido capture/replay.
|
||||
- PoC, CVSS, code-location, remediation, SARIF, CSV, JSON, and Markdown outputs.
|
||||
- Report-state hydration through `--resume` or `start_scan(resume=true)`.
|
||||
|
||||
## Proof-of-concept limitations
|
||||
|
||||
This integration should be treated as a foundation to perfect, not as finished runtime parity.
|
||||
|
||||
### Orchestration
|
||||
|
||||
- The host coding agent replaces the Strix `AgentCoordinator` and SDK runner.
|
||||
- Strix does not currently persist or restore the host agent's conversation.
|
||||
- `--resume` restores Strix targets, findings, configuration, and reports, but starts a new coding-agent reasoning session unless the user separately resumes that agent's session.
|
||||
- Native Codex/Claude subagents may be used, but their topology and status are not recorded in Strix's legacy `agents.json` graph.
|
||||
- There is no MCP equivalent of the legacy agent messaging, waiting, interruption, or child lifecycle tools.
|
||||
|
||||
### User experience
|
||||
|
||||
- The coding agent owns the interactive UI; the Strix TUI and live agent graph are not active in this mode.
|
||||
- Output, approval prompts, token visibility, and cancellation behavior differ by host agent.
|
||||
- `--max-budget-usd` is forwarded only where the host exposes an equivalent option. Host subscription usage is not tracked in Strix's LLM ledger.
|
||||
|
||||
### Tool parity
|
||||
|
||||
- `sandbox_exec` is synchronous and bounded; background process handles and `write_stdin` are not exposed.
|
||||
- Strix notes and todos are not exposed because the host agent normally provides planning primitives.
|
||||
- Legacy Perplexity web search is not exposed; the host's native web capability is expected instead.
|
||||
- Semantic model-based finding deduplication is replaced by deterministic exact matching plus agent review.
|
||||
|
||||
### Portability and hardening
|
||||
|
||||
- Only Codex and Claude Code have built-in launcher adapters.
|
||||
- Client-specific permission policies and MCP timeouts require more compatibility testing.
|
||||
- Windows and macOS installation, Docker networking, and path behavior need dedicated end-to-end coverage.
|
||||
- The server currently keeps scan state in process-global objects and supports one active scan per process.
|
||||
- Abrupt host-agent termination can leave a container running until cleanup is invoked or Docker reaps it.
|
||||
- Automated end-to-end tests should launch a real sandbox, generate proxy traffic, file a finding, finish, resume, and verify cleanup.
|
||||
|
||||
## Recommended next steps
|
||||
|
||||
1. Add a runtime abstraction so legacy and MCP orchestration share an explicit interface instead of branching at the CLI boundary.
|
||||
2. Add background command sessions and stdin streaming to the MCP execution API.
|
||||
3. Persist agent-runtime metadata and provide a host-session resume contract.
|
||||
4. Add deterministic and/or host-assisted semantic deduplication without a hidden Strix model call.
|
||||
5. Restore observability with host-reported token/usage metadata where clients expose it.
|
||||
6. Add structured progress events for TUI or agent UI integration.
|
||||
7. Add adapters and conformance tests for additional coding agents.
|
||||
8. Add Docker-backed end-to-end CI for source, web, proxy replay, reporting, resume, and cleanup.
|
||||
9. Threat-model MCP arguments, report content, host permissions, and sandbox escape boundaries.
|
||||
10. Decide whether legacy provider mode remains supported long-term or becomes a separate optional package/runtime.
|
||||
|
||||
## Troubleshooting
|
||||
|
||||
### Inspect the packaged skill
|
||||
|
||||
```bash
|
||||
strix-skill-path
|
||||
```
|
||||
|
||||
### Verify agent MCP registration
|
||||
|
||||
```bash
|
||||
codex mcp get strix
|
||||
claude mcp get strix
|
||||
```
|
||||
|
||||
### Run the MCP server directly
|
||||
|
||||
`strix-mcp` is a stdio process and normally appears to wait silently. That is expected: it is waiting for an MCP client to send an initialize request. Do not use ordinary terminal output as a health check.
|
||||
|
||||
### Force a runtime
|
||||
|
||||
If a saved `STRIX_LLM` configuration causes `auto` to choose legacy mode, explicitly pass:
|
||||
|
||||
```bash
|
||||
strix --agent codex --target ./your-app
|
||||
# or
|
||||
strix --agent claude --target ./your-app
|
||||
```
|
||||
|
||||
### Preserve incomplete work
|
||||
|
||||
If the agent is still responsive, instruct it to call `stop_scan`. This persists the run record before tearing down the container. If the process was killed, inspect Docker for a stranded sandbox and use the normal Strix/Docker cleanup procedure.
|
||||
|
|
@ -6,7 +6,7 @@ description: "Install Strix and run your first security scan"
|
|||
## Prerequisites
|
||||
|
||||
- Docker (running)
|
||||
- An LLM API key from any [supported provider](/llm-providers/overview) (OpenAI, Anthropic, Google, etc.)
|
||||
- Codex or Claude Code authenticated with your normal account, or an LLM API key for [legacy provider mode](/llm-providers/overview)
|
||||
|
||||
## Installation
|
||||
|
||||
|
|
@ -25,7 +25,13 @@ description: "Install Strix and run your first security scan"
|
|||
|
||||
## Configuration
|
||||
|
||||
Set your LLM provider:
|
||||
No model API key is required when Codex or Claude Code is installed and authenticated. Strix automatically selects a coding agent when `STRIX_LLM` is unset:
|
||||
|
||||
```bash
|
||||
strix --target ./your-app
|
||||
```
|
||||
|
||||
To use legacy direct-provider mode instead, set:
|
||||
|
||||
```bash
|
||||
export STRIX_LLM="openai/gpt-5.4"
|
||||
|
|
@ -42,6 +48,8 @@ For best results, use `openai/gpt-5.4`, `anthropic/claude-opus-4-6`, or `openai/
|
|||
strix --target ./your-app
|
||||
```
|
||||
|
||||
See [Codex and Claude Code](/integrations/coding-agents) for explicit agent selection and standalone skill/MCP installation.
|
||||
|
||||
<Note>
|
||||
First run pulls the Docker sandbox image automatically. Results are saved to `strix_runs/<run-name>`.
|
||||
</Note>
|
||||
|
|
|
|||
|
|
@ -34,6 +34,7 @@ classifiers = [
|
|||
]
|
||||
dependencies = [
|
||||
"openai-agents[litellm]==0.14.6",
|
||||
"mcp>=1.28.1",
|
||||
"pydantic>=2.11.3",
|
||||
"pydantic-settings>=2.13.0",
|
||||
"rich",
|
||||
|
|
@ -46,6 +47,8 @@ dependencies = [
|
|||
|
||||
[project.scripts]
|
||||
strix = "strix.interface.main:main"
|
||||
strix-mcp = "strix.mcp.server:main"
|
||||
strix-skill-path = "strix.mcp.install:print_skill_path"
|
||||
|
||||
[dependency-groups]
|
||||
dev = [
|
||||
|
|
@ -69,6 +72,9 @@ build-backend = "hatchling.build"
|
|||
[tool.hatch.build.targets.wheel]
|
||||
packages = ["strix"]
|
||||
|
||||
[tool.hatch.build.targets.wheel.force-include]
|
||||
"skills/strix-security" = "strix/agent_skills/strix-security"
|
||||
|
||||
# ============================================================================
|
||||
# Type Checking Configuration
|
||||
# ============================================================================
|
||||
|
|
|
|||
29
skills/strix-security/SKILL.md
Normal file
29
skills/strix-security/SKILL.md
Normal file
|
|
@ -0,0 +1,29 @@
|
|||
---
|
||||
name: strix-security
|
||||
description: Run authorized Strix application-security assessments through a coding agent and the Strix MCP server, including source review, reconnaissance, browser/API testing, exploit validation, vulnerability reporting, and resumable final reports. Use when asked to pentest, security-test, scan, assess, or find and validate vulnerabilities in a local codebase, repository, web application, domain, or IP address with Strix.
|
||||
---
|
||||
|
||||
# Strix Security
|
||||
|
||||
Use the connected coding agent for all reasoning. Use only Strix MCP tools for the scan lifecycle, isolated offensive commands, proxy operations, findings, and report persistence. Never request a separate model API key or invoke a model API from Strix.
|
||||
|
||||
## Run the assessment
|
||||
|
||||
1. Confirm the targets are explicitly authorized. Treat the targets returned by `start_scan` as the hard scope; user instructions may narrow but never expand it.
|
||||
2. Call `start_scan` once. For a prior run, pass its name with `resume=true`. Record the returned sandbox paths and output directory.
|
||||
3. Call `load_knowledge` for `scan_modes/<mode>`. Load technology, framework, protocol, tooling, and vulnerability modules only when evidence makes them relevant.
|
||||
4. Build a short test plan covering attack-surface discovery, authentication/authorization, input handling, data exposure, business logic, dependencies, and infrastructure as applicable.
|
||||
5. Execute security commands with `sandbox_exec`. Source targets are under the returned `/workspace/<name>` paths. Drive the browser with the `agent-browser` CLI through `sandbox_exec`; inspect and replay its captured traffic with the proxy tools.
|
||||
6. Validate every suspected issue with a concrete, reproducible proof of concept. Prefer safe demonstrations and avoid destructive actions, persistence, denial of service, or access beyond the authorized scope.
|
||||
7. Call `list_findings` before filing. Call `create_vulnerability_report` once per distinct verified root cause. For white-box findings, include repository-relative `code_locations`, exact line ranges, and actionable fixes.
|
||||
8. Continue until the selected scan mode is satisfied. Call `finish_scan` with customer-facing executive summary, methodology, consolidated technical analysis, and prioritized recommendations. Report the returned output directory.
|
||||
|
||||
## Quality bar
|
||||
|
||||
- Do not report scanner output, weak signals, version banners, missing headers, or theoretical code paths without exploitability and impact evidence.
|
||||
- Distinguish separate endpoints, parameters, privileges, and root causes. Do not duplicate a finding with different wording.
|
||||
- Keep customer-facing reports free of internal paths, agent/tool names, prompts, sandbox details, and model commentary.
|
||||
- Preserve evidence in commands and reports, but never expose credentials or unrelated sensitive data.
|
||||
- If blocked or interrupted, call `stop_scan` so artifacts and resume state remain usable.
|
||||
|
||||
Read [references/mcp-tools.md](references/mcp-tools.md) when tool selection, arguments, or the browser/proxy workflow is unclear.
|
||||
10
skills/strix-security/agents/openai.yaml
Normal file
10
skills/strix-security/agents/openai.yaml
Normal file
|
|
@ -0,0 +1,10 @@
|
|||
interface:
|
||||
display_name: "Strix Security"
|
||||
short_description: "Agent-native application security testing"
|
||||
default_prompt: "Use $strix-security to run an authorized security assessment of this project."
|
||||
dependencies:
|
||||
tools:
|
||||
- type: "mcp"
|
||||
value: "strix"
|
||||
description: "Local Strix security sandbox, proxy, knowledge, and reporting server"
|
||||
transport: "stdio"
|
||||
39
skills/strix-security/references/mcp-tools.md
Normal file
39
skills/strix-security/references/mcp-tools.md
Normal file
|
|
@ -0,0 +1,39 @@
|
|||
# Strix MCP tools
|
||||
|
||||
## Lifecycle
|
||||
|
||||
- `start_scan`: initialize targets, report state, source mappings, and the Docker sandbox. Pass `targets`, `mounts`, `scan_mode`, `instruction`, and optional `run_name`. To resume, pass `run_name` and `resume=true`.
|
||||
- `scan_status`: inspect active state, output path, configuration, and finding count.
|
||||
- `stop_scan`: persist an incomplete run and tear down its sandbox.
|
||||
- `finish_scan`: write `penetration_test_report.md`, `run.json`, vulnerability Markdown/CSV/JSON, and SARIF, then tear down the sandbox.
|
||||
|
||||
## Execution and knowledge
|
||||
|
||||
- `sandbox_exec`: pass an argv array, never a shell command string. Examples: `["bash", "-lc", "cd /workspace/app && semgrep scan --config auto"]`, `["agent-browser", "open", "https://target.example"]`.
|
||||
- `list_knowledge`: list built-in Strix modules.
|
||||
- `load_knowledge`: load a module by exact name, such as `frameworks/nextjs`, `protocols/graphql`, `vulnerabilities/idor`, or `tooling/agent_browser`.
|
||||
|
||||
## HTTP proxy
|
||||
|
||||
All HTTP(S) traffic from sandbox commands is routed through Caido.
|
||||
|
||||
1. Generate traffic with curl, Python, or `agent-browser` through `sandbox_exec`.
|
||||
2. Use `list_proxy_requests` with an optional Caido HTTPQL filter.
|
||||
3. Use `view_proxy_request` to inspect request or response content.
|
||||
4. Use `repeat_proxy_request` with field-level modifications to test authorization, input handling, cookies, headers, or bodies.
|
||||
5. Use `list_sitemap` and `view_sitemap_entry` to enumerate captured application structure.
|
||||
6. Use `manage_scope` to create or update proxy allowlists and denylists.
|
||||
|
||||
## Findings
|
||||
|
||||
- `list_findings`: review existing reports before filing another.
|
||||
- `create_vulnerability_report`: requires title, description, impact, target, technical analysis, PoC description, actual PoC code/payload, remediation, and all eight CVSS 3.1 base metrics. Optional fields include endpoint, method, CVE, CWE, and code locations.
|
||||
|
||||
CVSS keys and values:
|
||||
|
||||
- `attack_vector`: `N`, `A`, `L`, or `P`
|
||||
- `attack_complexity`: `L` or `H`
|
||||
- `privileges_required`: `N`, `L`, or `H`
|
||||
- `user_interaction`: `N` or `R`
|
||||
- `scope`: `U` or `C`
|
||||
- `confidentiality`, `integrity`, `availability`: `N`, `L`, or `H`
|
||||
|
|
@ -1,4 +1,15 @@
|
|||
from .main import main
|
||||
"""User interfaces for Strix.
|
||||
|
||||
Keep package import lightweight: the MCP service reuses target utilities and
|
||||
must not initialize the legacy model stack during stdio startup.
|
||||
"""
|
||||
|
||||
|
||||
def main() -> None:
|
||||
"""Load and run the CLI entry point lazily."""
|
||||
from .main import main as run
|
||||
|
||||
run()
|
||||
|
||||
|
||||
__all__ = ["main"]
|
||||
|
|
|
|||
120
strix/interface/agent_launcher.py
Normal file
120
strix/interface/agent_launcher.py
Normal file
|
|
@ -0,0 +1,120 @@
|
|||
"""Launch Strix through an authenticated coding-agent CLI."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from strix.config import load_settings
|
||||
from strix.mcp.install import skill_path
|
||||
|
||||
|
||||
def uses_coding_agent(args: Any) -> bool:
|
||||
"""Return whether this invocation should delegate reasoning to a coding agent."""
|
||||
requested = str(getattr(args, "agent", "auto") or "auto")
|
||||
if requested == "legacy":
|
||||
return False
|
||||
if requested in {"codex", "claude"}:
|
||||
return True
|
||||
return not bool(load_settings().llm.model)
|
||||
|
||||
|
||||
def resolve_agent(requested: str) -> str:
|
||||
"""Resolve ``auto`` to an installed, authenticated-agent-capable CLI."""
|
||||
if requested != "auto":
|
||||
if shutil.which(requested) is None:
|
||||
raise RuntimeError(f"Requested coding agent CLI is not installed: {requested}")
|
||||
return requested
|
||||
for candidate in (os.environ.get("STRIX_AGENT"), "codex", "claude"):
|
||||
if candidate and candidate in {"codex", "claude"} and shutil.which(candidate):
|
||||
return candidate
|
||||
raise RuntimeError(
|
||||
"No coding agent CLI found. Install Codex or Claude Code, or configure legacy "
|
||||
"provider mode with STRIX_LLM."
|
||||
)
|
||||
|
||||
|
||||
def _scan_prompt(args: Any) -> str:
|
||||
instructions_path = skill_path() / "SKILL.md"
|
||||
targets = [] if args.resume else list(args.target or [])
|
||||
mounts = [] if args.resume else list(args.mount or [])
|
||||
call = {
|
||||
"targets": targets,
|
||||
"mounts": mounts,
|
||||
"scan_mode": args.scan_mode,
|
||||
"instruction": args.instruction or "",
|
||||
"run_name": args.resume,
|
||||
"resume": bool(args.resume),
|
||||
"scope_mode": args.scope_mode,
|
||||
"diff_base": args.diff_base,
|
||||
"non_interactive": args.non_interactive,
|
||||
}
|
||||
return (
|
||||
"Run a Strix security assessment now. Read the complete skill file at "
|
||||
f"{instructions_path} and follow it exactly. Use the connected Strix MCP tools for the "
|
||||
"scan "
|
||||
"lifecycle, isolated security commands, proxy traffic, findings, and final report. "
|
||||
"Do not call any model API or ask for a model API key. Begin by calling start_scan with "
|
||||
f"these arguments: {json.dumps(call, ensure_ascii=False)}. Continue until finish_scan "
|
||||
"succeeds, unless authorization or a material blocker requires user input."
|
||||
)
|
||||
|
||||
|
||||
def _mcp_command() -> tuple[str, list[str]]:
|
||||
return sys.executable, ["-m", "strix.mcp.server"]
|
||||
|
||||
|
||||
def build_agent_command(args: Any, agent: str) -> list[str]:
|
||||
"""Build a coding-agent command with an ephemeral Strix MCP server."""
|
||||
python, server_args = _mcp_command()
|
||||
prompt = _scan_prompt(args)
|
||||
cwd = str(Path.cwd())
|
||||
if agent == "codex":
|
||||
command = [
|
||||
"codex",
|
||||
"-C",
|
||||
cwd,
|
||||
"-c",
|
||||
f"mcp_servers.strix.command={json.dumps(python)}",
|
||||
"-c",
|
||||
f"mcp_servers.strix.args={json.dumps(server_args)}",
|
||||
"-c",
|
||||
"mcp_servers.strix.startup_timeout_sec=120",
|
||||
"-c",
|
||||
"mcp_servers.strix.tool_timeout_sec=900",
|
||||
]
|
||||
if args.non_interactive:
|
||||
command.insert(1, "exec")
|
||||
command.append(prompt)
|
||||
return command
|
||||
|
||||
config = json.dumps(
|
||||
{
|
||||
"mcpServers": {
|
||||
"strix": {"type": "stdio", "command": python, "args": server_args, "env": {}}
|
||||
}
|
||||
}
|
||||
)
|
||||
command = ["claude", "--mcp-config", config]
|
||||
if args.non_interactive:
|
||||
command.append("--print")
|
||||
if args.max_budget_usd is not None:
|
||||
command.extend(["--max-budget-usd", str(args.max_budget_usd)])
|
||||
# ``--mcp-config`` accepts multiple values. Without an option terminator,
|
||||
# Claude consumes the positional prompt as another config path.
|
||||
command.extend(["--", prompt])
|
||||
return command
|
||||
|
||||
|
||||
def launch_agent_scan(args: Any) -> int:
|
||||
"""Run Strix in the selected coding agent and return its exit status."""
|
||||
agent = resolve_agent(str(args.agent or "auto"))
|
||||
command = build_agent_command(args, agent)
|
||||
command[0] = shutil.which(agent) or command[0]
|
||||
result = subprocess.run(command, check=False) # noqa: S603
|
||||
return result.returncode
|
||||
|
|
@ -457,6 +457,16 @@ Examples:
|
|||
help="Maximum LLM cost in USD (> 0). The scan stops cleanly when this limit is reached.",
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
"--agent",
|
||||
choices=["auto", "codex", "claude", "legacy"],
|
||||
default="auto",
|
||||
help=(
|
||||
"Reasoning runtime. 'auto' uses STRIX_LLM when configured, otherwise an installed "
|
||||
"Codex or Claude Code CLI. Agent mode needs no separate model API key."
|
||||
),
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
"--resume",
|
||||
type=str,
|
||||
|
|
@ -495,7 +505,7 @@ Examples:
|
|||
)
|
||||
_load_resume_state(args, parser)
|
||||
agents_path = runtime_state_dir(run_dir_for(args.resume)) / "agents.json"
|
||||
if not agents_path.exists():
|
||||
if args.agent == "legacy" and not agents_path.exists():
|
||||
parser.error(
|
||||
f"--resume {args.resume}: missing {agents_path}. The run was "
|
||||
f"persisted but never reached its first agent snapshot — "
|
||||
|
|
@ -748,6 +758,21 @@ def main() -> None:
|
|||
if args.config:
|
||||
apply_config_override(validate_config_file(args.config))
|
||||
|
||||
from strix.interface.agent_launcher import launch_agent_scan, uses_coding_agent
|
||||
|
||||
if uses_coding_agent(args):
|
||||
check_docker_installed()
|
||||
pull_docker_image()
|
||||
raise SystemExit(launch_agent_scan(args))
|
||||
|
||||
if args.resume:
|
||||
agents_path = runtime_state_dir(run_dir_for(args.resume)) / "agents.json"
|
||||
if not agents_path.exists():
|
||||
raise SystemExit(
|
||||
f"--resume {args.resume}: missing {agents_path}; use --agent codex/claude "
|
||||
"for an MCP-mode run or start a fresh legacy run."
|
||||
)
|
||||
|
||||
check_docker_installed()
|
||||
pull_docker_image()
|
||||
|
||||
|
|
|
|||
1
strix/mcp/__init__.py
Normal file
1
strix/mcp/__init__.py
Normal file
|
|
@ -0,0 +1 @@
|
|||
"""Agent-native Model Context Protocol integration for Strix."""
|
||||
27
strix/mcp/install.py
Normal file
27
strix/mcp/install.py
Normal file
|
|
@ -0,0 +1,27 @@
|
|||
"""Installation helpers for the portable Strix agent skill."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def skill_path() -> Path:
|
||||
"""Return the skill directory in a checkout or installed wheel."""
|
||||
module_path = Path(__file__).resolve()
|
||||
candidates = (
|
||||
module_path.parents[2] / "skills" / "strix-security",
|
||||
module_path.parents[1] / "agent_skills" / "strix-security",
|
||||
)
|
||||
for candidate in candidates:
|
||||
if (candidate / "SKILL.md").is_file():
|
||||
return candidate
|
||||
raise FileNotFoundError("The packaged strix-security skill could not be located")
|
||||
|
||||
|
||||
def print_skill_path() -> None:
|
||||
"""Print a shell-safe-to-quote path for agent skill installation."""
|
||||
print(skill_path()) # noqa: T201 - this command intentionally prints its result.
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
print_skill_path()
|
||||
316
strix/mcp/server.py
Normal file
316
strix/mcp/server.py
Normal file
|
|
@ -0,0 +1,316 @@
|
|||
"""stdio MCP server for agent-native Strix scans."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import logging
|
||||
from contextlib import asynccontextmanager
|
||||
from typing import Any
|
||||
|
||||
import anyio
|
||||
import mcp.types as mcp_types
|
||||
from mcp.server.fastmcp import FastMCP
|
||||
from mcp.shared.message import SessionMessage
|
||||
|
||||
from strix.mcp.service import StrixMCPService
|
||||
from strix.tools.proxy.tools import (
|
||||
list_requests as _list_requests,
|
||||
list_sitemap as _list_sitemap,
|
||||
repeat_request as _repeat_request,
|
||||
scope_rules as _scope_rules,
|
||||
view_request as _view_request,
|
||||
view_sitemap_entry as _view_sitemap_entry,
|
||||
)
|
||||
|
||||
|
||||
logging.basicConfig(level=logging.WARNING)
|
||||
service = StrixMCPService()
|
||||
mcp = FastMCP(
|
||||
"Strix Security",
|
||||
instructions=(
|
||||
"Authorized application-security testing tools. The connected coding agent performs all "
|
||||
"reasoning; this server never invokes a language model. Call start_scan first and "
|
||||
"finish_scan last."
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
async def start_scan(
|
||||
targets: list[str],
|
||||
scan_mode: str = "deep",
|
||||
instruction: str = "",
|
||||
run_name: str | None = None,
|
||||
mounts: list[str] | None = None,
|
||||
resume: bool = False,
|
||||
scope_mode: str = "auto",
|
||||
diff_base: str | None = None,
|
||||
non_interactive: bool = False,
|
||||
) -> dict[str, Any]:
|
||||
"""Start or resume an authorized Strix scan and isolated sandbox."""
|
||||
return await service.start_scan(
|
||||
targets,
|
||||
scan_mode=scan_mode,
|
||||
instruction=instruction,
|
||||
run_name=run_name,
|
||||
mounts=mounts,
|
||||
resume=resume,
|
||||
scope_mode=scope_mode,
|
||||
diff_base=diff_base,
|
||||
non_interactive=non_interactive,
|
||||
)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
async def sandbox_exec(argv: list[str], timeout: int = 120) -> dict[str, Any]:
|
||||
"""Run an argv-form command in the isolated pentesting sandbox.
|
||||
|
||||
The sandbox includes common reconnaissance tools, Python, curl, and the
|
||||
agent-browser CLI. Traffic is captured by the integrated Caido proxy.
|
||||
"""
|
||||
return await service.sandbox_exec(argv, timeout)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def list_knowledge() -> dict[str, Any]:
|
||||
"""List Strix security knowledge modules available to load on demand."""
|
||||
return service.list_knowledge()
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def load_knowledge(name: str) -> dict[str, Any]:
|
||||
"""Load one Strix knowledge module, such as vulnerabilities/xss."""
|
||||
return service.load_knowledge(name)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
async def list_proxy_requests(
|
||||
httpql_filter: str | None = None,
|
||||
first: int = 50,
|
||||
after: str | None = None,
|
||||
sort_by: str = "timestamp",
|
||||
sort_order: str = "desc",
|
||||
scope_id: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""List HTTP requests captured from sandbox traffic by Caido."""
|
||||
return await service.proxy_call(
|
||||
_list_requests,
|
||||
{
|
||||
"httpql_filter": httpql_filter,
|
||||
"first": first,
|
||||
"after": after,
|
||||
"sort_by": sort_by,
|
||||
"sort_order": sort_order,
|
||||
"scope_id": scope_id,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
async def view_proxy_request(
|
||||
request_id: str,
|
||||
part: str = "request",
|
||||
search_pattern: str | None = None,
|
||||
page: int = 1,
|
||||
page_size: int = 50,
|
||||
) -> dict[str, Any]:
|
||||
"""Read a captured HTTP request or response with optional regex search."""
|
||||
return await service.proxy_call(
|
||||
_view_request,
|
||||
{
|
||||
"request_id": request_id,
|
||||
"part": part,
|
||||
"search_pattern": search_pattern,
|
||||
"page": page,
|
||||
"page_size": page_size,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
async def repeat_proxy_request(
|
||||
request_id: str,
|
||||
modifications: dict[str, Any] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Replay a captured request with URL, parameter, header, cookie, or body changes."""
|
||||
return await service.proxy_call(
|
||||
_repeat_request,
|
||||
{"request_id": request_id, "modifications": modifications},
|
||||
)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
async def list_sitemap(
|
||||
scope_id: str | None = None,
|
||||
parent_id: str | None = None,
|
||||
depth: str = "DIRECT",
|
||||
page: int = 1,
|
||||
) -> dict[str, Any]:
|
||||
"""Browse the hierarchical sitemap built from captured proxy traffic."""
|
||||
return await service.proxy_call(
|
||||
_list_sitemap,
|
||||
{"scope_id": scope_id, "parent_id": parent_id, "depth": depth, "page": page},
|
||||
)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
async def view_sitemap_entry(entry_id: str) -> dict[str, Any]:
|
||||
"""View one proxy sitemap entry and its recent requests."""
|
||||
return await service.proxy_call(_view_sitemap_entry, {"entry_id": entry_id})
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
async def manage_scope(
|
||||
action: str,
|
||||
allowlist: list[str] | None = None,
|
||||
denylist: list[str] | None = None,
|
||||
scope_id: str | None = None,
|
||||
scope_name: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Create, list, update, get, or delete Caido proxy scope rules."""
|
||||
return await service.proxy_call(
|
||||
_scope_rules,
|
||||
{
|
||||
"action": action,
|
||||
"allowlist": allowlist,
|
||||
"denylist": denylist,
|
||||
"scope_id": scope_id,
|
||||
"scope_name": scope_name,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
async def create_vulnerability_report(
|
||||
title: str,
|
||||
description: str,
|
||||
impact: str,
|
||||
target: str,
|
||||
technical_analysis: str,
|
||||
poc_description: str,
|
||||
poc_script_code: str,
|
||||
remediation_steps: str,
|
||||
cvss_breakdown: dict[str, str],
|
||||
endpoint: str | None = None,
|
||||
method: str | None = None,
|
||||
cve: str | None = None,
|
||||
cwe: str | None = None,
|
||||
code_locations: list[dict[str, Any]] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Persist one distinct, verified vulnerability and its working proof of concept."""
|
||||
return await service.create_finding(
|
||||
title=title,
|
||||
description=description,
|
||||
impact=impact,
|
||||
target=target,
|
||||
technical_analysis=technical_analysis,
|
||||
poc_description=poc_description,
|
||||
poc_script_code=poc_script_code,
|
||||
remediation_steps=remediation_steps,
|
||||
cvss_breakdown=cvss_breakdown,
|
||||
endpoint=endpoint,
|
||||
method=method,
|
||||
cve=cve,
|
||||
cwe=cwe,
|
||||
code_locations=code_locations,
|
||||
)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def list_findings() -> dict[str, Any]:
|
||||
"""List findings already persisted for the active scan."""
|
||||
return service.list_findings()
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def scan_status() -> dict[str, Any]:
|
||||
"""Return active scan metadata, output path, and finding count."""
|
||||
return service.status()
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
async def finish_scan(
|
||||
executive_summary: str,
|
||||
methodology: str,
|
||||
technical_analysis: str,
|
||||
recommendations: str,
|
||||
) -> dict[str, Any]:
|
||||
"""Finalize the customer-facing report and tear down the sandbox."""
|
||||
return await service.finish_scan(
|
||||
executive_summary=executive_summary,
|
||||
methodology=methodology,
|
||||
technical_analysis=technical_analysis,
|
||||
recommendations=recommendations,
|
||||
)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
async def stop_scan(status: str = "stopped") -> dict[str, Any]:
|
||||
"""Persist an incomplete scan and tear down its sandbox."""
|
||||
return await service.stop_scan(status)
|
||||
|
||||
|
||||
def main() -> None:
|
||||
"""Run the local stdio MCP server."""
|
||||
anyio.run(_run_stdio)
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def _event_loop_stdio() -> Any:
|
||||
"""Provide stdio streams without AnyIO's worker-thread file wrapper.
|
||||
|
||||
Some managed shells block AnyIO worker-thread jobs indefinitely. Reading
|
||||
and writing the pipe file descriptors through the event loop keeps the MCP
|
||||
transport portable while preserving newline-delimited JSON-RPC semantics.
|
||||
"""
|
||||
read_writer, read_stream = anyio.create_memory_object_stream(0)
|
||||
write_stream, write_reader = anyio.create_memory_object_stream(0)
|
||||
|
||||
async def read_stdin() -> None:
|
||||
buffered = b""
|
||||
async with read_writer:
|
||||
while True:
|
||||
await anyio.wait_readable(0)
|
||||
chunk = os.read(0, 65536)
|
||||
if not chunk:
|
||||
break
|
||||
buffered += chunk
|
||||
while b"\n" in buffered:
|
||||
line, buffered = buffered.split(b"\n", 1)
|
||||
if not line:
|
||||
continue
|
||||
try:
|
||||
message = mcp_types.JSONRPCMessage.model_validate_json(line)
|
||||
except Exception as exc: # noqa: BLE001 - protocol parse errors are messages.
|
||||
await read_writer.send(exc)
|
||||
else:
|
||||
await read_writer.send(SessionMessage(message))
|
||||
|
||||
async def write_stdout() -> None:
|
||||
async with write_reader:
|
||||
async for session_message in write_reader:
|
||||
payload = (
|
||||
session_message.message.model_dump_json(by_alias=True, exclude_none=True)
|
||||
+ "\n"
|
||||
).encode()
|
||||
while payload:
|
||||
await anyio.wait_writable(1)
|
||||
payload = payload[os.write(1, payload) :]
|
||||
|
||||
async with anyio.create_task_group() as tasks:
|
||||
tasks.start_soon(read_stdin)
|
||||
tasks.start_soon(write_stdout)
|
||||
yield read_stream, write_stream
|
||||
|
||||
|
||||
async def _run_stdio() -> None:
|
||||
async with _event_loop_stdio() as (read_stream, write_stream):
|
||||
await mcp._mcp_server.run( # noqa: SLF001 - mirrors FastMCP.run_stdio_async.
|
||||
read_stream,
|
||||
write_stream,
|
||||
mcp._mcp_server.create_initialization_options(), # noqa: SLF001
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
364
strix/mcp/service.py
Normal file
364
strix/mcp/service.py
Normal file
|
|
@ -0,0 +1,364 @@
|
|||
"""Stateful, model-free Strix service exposed by the MCP server."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
import shutil
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from typing import Any
|
||||
|
||||
from strix.config import load_settings
|
||||
from strix.core.inputs import build_root_task, build_scope_context
|
||||
from strix.core.paths import run_dir_for
|
||||
from strix.interface.utils import (
|
||||
assign_workspace_subdirs,
|
||||
build_mount_targets_info,
|
||||
collect_local_sources,
|
||||
dedupe_local_targets,
|
||||
find_oversized_local_targets,
|
||||
generate_run_name,
|
||||
infer_target_type,
|
||||
resolve_diff_scope_context,
|
||||
rewrite_localhost_targets,
|
||||
)
|
||||
from strix.report.state import ReportState, set_global_report_state
|
||||
from strix.runtime import session_manager
|
||||
from strix.skills import get_available_skills, load_skills
|
||||
from strix.tools.finish.tool import _do_finish
|
||||
from strix.tools.reporting.tool import _do_create
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class StrixMCPService:
|
||||
"""Own one agent-driven scan for the lifetime of an MCP process.
|
||||
|
||||
This class deliberately contains no model provider or model call. The MCP
|
||||
client (Codex, Claude Code, or another coding agent) supplies all reasoning.
|
||||
"""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.scan_id: str | None = None
|
||||
self.scan_config: dict[str, Any] | None = None
|
||||
self.report_state: ReportState | None = None
|
||||
self.bundle: dict[str, Any] | None = None
|
||||
|
||||
async def start_scan(
|
||||
self,
|
||||
targets: list[str],
|
||||
*,
|
||||
scan_mode: str = "deep",
|
||||
instruction: str = "",
|
||||
run_name: str | None = None,
|
||||
mounts: list[str] | None = None,
|
||||
resume: bool = False,
|
||||
scope_mode: str = "auto",
|
||||
diff_base: str | None = None,
|
||||
non_interactive: bool = False,
|
||||
) -> dict[str, Any]:
|
||||
"""Create or resume a scan and start its isolated tool sandbox."""
|
||||
if self.bundle is not None:
|
||||
return {
|
||||
"success": False,
|
||||
"error": "A scan is already active in this MCP session",
|
||||
"scan_id": self.scan_id,
|
||||
}
|
||||
if scan_mode not in {"quick", "standard", "deep"}:
|
||||
return {"success": False, "error": f"Unsupported scan mode: {scan_mode}"}
|
||||
|
||||
if resume:
|
||||
if not run_name:
|
||||
return {"success": False, "error": "run_name is required when resume=true"}
|
||||
state = ReportState(run_name)
|
||||
state.hydrate_from_run_dir()
|
||||
persisted = state.run_record
|
||||
targets_info = persisted.get("targets_info") or []
|
||||
if not targets_info:
|
||||
return {"success": False, "error": f"Run {run_name!r} has no persisted targets"}
|
||||
local_sources = persisted.get("local_sources") or collect_local_sources(targets_info)
|
||||
scan_mode = str(persisted.get("scan_mode") or scan_mode)
|
||||
scope_mode = str(persisted.get("scope_mode") or scope_mode)
|
||||
diff_base = persisted.get("diff_base") or diff_base
|
||||
diff_scope = persisted.get("diff_scope") or {"active": False}
|
||||
original_instruction = str(persisted.get("instruction") or "")
|
||||
instruction = "\n\n".join(x for x in (original_instruction, instruction) if x)
|
||||
else:
|
||||
if not targets and not mounts:
|
||||
return {"success": False, "error": "At least one target or mount is required"}
|
||||
try:
|
||||
targets_info = await self._prepare_targets(targets, mounts or [], run_name)
|
||||
except (OSError, RuntimeError, ValueError) as exc:
|
||||
return {"success": False, "error": str(exc)}
|
||||
run_name = run_name or generate_run_name(targets_info)
|
||||
local_sources = collect_local_sources(targets_info)
|
||||
try:
|
||||
resolved_scope = resolve_diff_scope_context(
|
||||
local_sources=local_sources,
|
||||
scope_mode=scope_mode,
|
||||
diff_base=diff_base,
|
||||
non_interactive=non_interactive,
|
||||
)
|
||||
except ValueError as exc:
|
||||
return {"success": False, "error": str(exc)}
|
||||
diff_scope = resolved_scope.metadata
|
||||
if resolved_scope.instruction_block:
|
||||
instruction = "\n\n".join(
|
||||
part for part in (resolved_scope.instruction_block, instruction) if part
|
||||
)
|
||||
state = ReportState(run_name)
|
||||
|
||||
assert run_name is not None
|
||||
scan_config = {
|
||||
"targets": targets_info,
|
||||
"scan_mode": scan_mode,
|
||||
"user_instructions": instruction,
|
||||
"local_sources": local_sources,
|
||||
"scope_mode": scope_mode,
|
||||
"diff_base": diff_base,
|
||||
"diff_scope": diff_scope,
|
||||
"non_interactive": non_interactive,
|
||||
"agent_runtime": "mcp",
|
||||
}
|
||||
state.set_scan_config(scan_config)
|
||||
state.save_run_data()
|
||||
set_global_report_state(state)
|
||||
|
||||
try:
|
||||
bundle = await session_manager.create_or_reuse(
|
||||
run_name,
|
||||
image=load_settings().runtime.image,
|
||||
local_sources=local_sources,
|
||||
)
|
||||
except Exception as exc: # noqa: BLE001 - return a clean MCP error.
|
||||
state.save_run_data(status="failed")
|
||||
logger.exception("Could not create Strix sandbox")
|
||||
return {"success": False, "error": f"Could not create Strix sandbox: {exc}"}
|
||||
|
||||
self.scan_id = run_name
|
||||
self.scan_config = scan_config
|
||||
self.report_state = state
|
||||
self.bundle = bundle
|
||||
return {
|
||||
"success": True,
|
||||
"scan_id": run_name,
|
||||
"run_dir": str(run_dir_for(run_name).resolve()),
|
||||
"task": build_root_task(scan_config),
|
||||
"scope": build_scope_context(scan_config),
|
||||
"scan_mode": scan_mode,
|
||||
"workspace_targets": self._workspace_targets(targets_info),
|
||||
"next": (
|
||||
"Follow the strix-security skill workflow. Use sandbox_exec for testing, "
|
||||
"load_knowledge when a technology or vulnerability class is relevant, file "
|
||||
"only verified findings, then call finish_scan."
|
||||
),
|
||||
}
|
||||
|
||||
async def _prepare_targets(
|
||||
self,
|
||||
targets: list[str],
|
||||
mounts: list[str],
|
||||
run_name: str | None,
|
||||
) -> list[dict[str, Any]]:
|
||||
targets_info: list[dict[str, Any]] = []
|
||||
for raw_target in targets:
|
||||
target_type, details = await asyncio.to_thread(infer_target_type, raw_target)
|
||||
original = (
|
||||
details.get("target_path", raw_target)
|
||||
if target_type == "local_code"
|
||||
else raw_target
|
||||
)
|
||||
targets_info.append({"type": target_type, "details": details, "original": original})
|
||||
targets_info.extend(build_mount_targets_info(mounts))
|
||||
targets_info = dedupe_local_targets(targets_info)
|
||||
assign_workspace_subdirs(targets_info)
|
||||
rewrite_localhost_targets(targets_info, "host.docker.internal")
|
||||
|
||||
max_mb = load_settings().runtime.max_local_copy_mb
|
||||
oversized = find_oversized_local_targets(targets_info, max_mb * 1024 * 1024)
|
||||
if oversized:
|
||||
details = "; ".join(
|
||||
f"{path} ({size / (1024 * 1024):.0f} MB)" for path, size in oversized
|
||||
)
|
||||
raise ValueError(
|
||||
f"Local target too large to copy: {details}. Use the mounts argument or raise "
|
||||
"STRIX_MAX_LOCAL_COPY_MB."
|
||||
)
|
||||
|
||||
scan_name = run_name or generate_run_name(targets_info)
|
||||
for target_info in targets_info:
|
||||
if target_info["type"] != "repository":
|
||||
continue
|
||||
details = target_info["details"]
|
||||
details["cloned_repo_path"] = await self._clone_repository(
|
||||
details["target_repo"],
|
||||
scan_name,
|
||||
details.get("workspace_subdir"),
|
||||
)
|
||||
return targets_info
|
||||
|
||||
async def _clone_repository(self, url: str, run_name: str, subdir: str | None) -> str:
|
||||
name = subdir or re.sub(r"[^A-Za-z0-9._-]", "-", Path(url).stem) or "repository"
|
||||
destination = Path(tempfile.gettempdir()) / "strix_repos" / run_name / name
|
||||
destination.parent.mkdir(parents=True, exist_ok=True)
|
||||
if destination.exists():
|
||||
raise ValueError(
|
||||
f"Clone destination already exists: {destination}. "
|
||||
"Use resume=true or another run_name."
|
||||
)
|
||||
git = shutil.which("git")
|
||||
if git is None:
|
||||
raise RuntimeError("Git is required to clone repository targets")
|
||||
process = await asyncio.create_subprocess_exec(
|
||||
git,
|
||||
"clone",
|
||||
url,
|
||||
str(destination),
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.PIPE,
|
||||
) # noqa: S603 - argv is fixed apart from the authorized URL/path.
|
||||
_stdout, stderr = await process.communicate()
|
||||
if process.returncode != 0:
|
||||
detail = stderr.decode("utf-8", errors="replace")[-2000:]
|
||||
raise RuntimeError(f"Could not clone {url}: {detail}")
|
||||
return str(destination.resolve())
|
||||
|
||||
@staticmethod
|
||||
def _workspace_targets(targets_info: list[dict[str, Any]]) -> list[dict[str, str]]:
|
||||
mapped: list[dict[str, str]] = []
|
||||
for target in targets_info:
|
||||
details = target.get("details") or {}
|
||||
subdir = details.get("workspace_subdir")
|
||||
mapped.append(
|
||||
{
|
||||
"type": str(target.get("type", "unknown")),
|
||||
"target": str(target.get("original", "")),
|
||||
"sandbox_path": f"/workspace/{subdir}" if subdir else "",
|
||||
}
|
||||
)
|
||||
return mapped
|
||||
|
||||
async def sandbox_exec(self, argv: list[str], timeout: int = 120) -> dict[str, Any]:
|
||||
"""Execute one argv-form command inside the Strix security sandbox."""
|
||||
if self.bundle is None:
|
||||
return {"success": False, "error": "Call start_scan first"}
|
||||
if not argv or any(not isinstance(part, str) or not part for part in argv):
|
||||
return {"success": False, "error": "argv must contain non-empty strings"}
|
||||
timeout = min(max(timeout, 1), 900)
|
||||
try:
|
||||
result = await self.bundle["session"].exec(*argv, timeout=timeout)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
return {"success": False, "error": str(exc)}
|
||||
return {
|
||||
"success": bool(result.ok()),
|
||||
"exit_code": result.exit_code,
|
||||
"stdout": self._decode(result.stdout),
|
||||
"stderr": self._decode(result.stderr),
|
||||
}
|
||||
|
||||
@staticmethod
|
||||
def _decode(value: Any) -> str:
|
||||
if isinstance(value, bytes):
|
||||
return value.decode("utf-8", errors="replace")
|
||||
return str(value or "")
|
||||
|
||||
def list_knowledge(self) -> dict[str, Any]:
|
||||
return {"success": True, "skills": get_available_skills()}
|
||||
|
||||
def load_knowledge(self, name: str) -> dict[str, Any]:
|
||||
content = load_skills([name])
|
||||
if not content:
|
||||
return {"success": False, "error": f"Knowledge module not found: {name}"}
|
||||
return {"success": True, "name": name, "content": next(iter(content.values()))}
|
||||
|
||||
async def proxy_call(self, tool: Any, arguments: dict[str, Any]) -> dict[str, Any]:
|
||||
if self.bundle is None:
|
||||
return {"success": False, "error": "Call start_scan first"}
|
||||
context = SimpleNamespace(context={"caido_client": self.bundle["caido_client"]})
|
||||
try:
|
||||
raw = await tool.on_invoke_tool(context, json.dumps(arguments))
|
||||
return json.loads(raw) if isinstance(raw, str) else {"success": True, "result": raw}
|
||||
except Exception as exc: # noqa: BLE001
|
||||
logger.exception("Proxy tool call failed")
|
||||
return {"success": False, "error": str(exc)}
|
||||
|
||||
async def create_finding(self, **finding: Any) -> dict[str, Any]:
|
||||
if self.report_state is None:
|
||||
return {"success": False, "error": "Call start_scan first"}
|
||||
finding["allow_model_dedupe"] = False
|
||||
finding.setdefault("agent_id", "coding-agent")
|
||||
finding.setdefault("agent_name", "coding-agent")
|
||||
result: Any = await _do_create(**finding)
|
||||
if not isinstance(result, dict):
|
||||
return {"success": False, "error": "Unexpected report result"}
|
||||
return result
|
||||
|
||||
def list_findings(self) -> dict[str, Any]:
|
||||
if self.report_state is None:
|
||||
return {"success": False, "error": "Call start_scan first"}
|
||||
return {
|
||||
"success": True,
|
||||
"count": len(self.report_state.vulnerability_reports),
|
||||
"findings": self.report_state.get_existing_vulnerabilities(),
|
||||
}
|
||||
|
||||
def status(self) -> dict[str, Any]:
|
||||
if self.report_state is None:
|
||||
return {"success": True, "active": False}
|
||||
return {
|
||||
"success": True,
|
||||
"active": self.bundle is not None,
|
||||
"scan_id": self.scan_id,
|
||||
"run_dir": str(self.report_state.get_run_dir().resolve()),
|
||||
"status": self.report_state.run_record.get("status"),
|
||||
"findings": len(self.report_state.vulnerability_reports),
|
||||
"scan_config": self.scan_config,
|
||||
}
|
||||
|
||||
async def finish_scan(
|
||||
self,
|
||||
*,
|
||||
executive_summary: str,
|
||||
methodology: str,
|
||||
technical_analysis: str,
|
||||
recommendations: str,
|
||||
) -> dict[str, Any]:
|
||||
if self.report_state is None:
|
||||
return {"success": False, "error": "Call start_scan first"}
|
||||
raw_result: Any = await asyncio.to_thread(
|
||||
_do_finish,
|
||||
parent_id=None,
|
||||
executive_summary=executive_summary,
|
||||
methodology=methodology,
|
||||
technical_analysis=technical_analysis,
|
||||
recommendations=recommendations,
|
||||
)
|
||||
if not isinstance(raw_result, dict):
|
||||
return {"success": False, "error": "Unexpected finish result"}
|
||||
result = raw_result
|
||||
if result.get("success"):
|
||||
await self._cleanup()
|
||||
result["run_dir"] = str(self.report_state.get_run_dir().resolve())
|
||||
return result
|
||||
|
||||
async def stop_scan(self, status: str = "stopped") -> dict[str, Any]:
|
||||
if self.report_state is None:
|
||||
return {"success": True, "message": "No active scan"}
|
||||
self.report_state.save_run_data(status=status)
|
||||
await self._cleanup()
|
||||
return {
|
||||
"success": True,
|
||||
"scan_id": self.scan_id,
|
||||
"status": self.report_state.run_record.get("status"),
|
||||
"run_dir": str(self.report_state.get_run_dir().resolve()),
|
||||
}
|
||||
|
||||
async def _cleanup(self) -> None:
|
||||
if self.bundle is not None and self.scan_id is not None:
|
||||
await session_manager.cleanup(self.scan_id)
|
||||
self.bundle = None
|
||||
|
|
@ -169,6 +169,7 @@ async def _do_create( # noqa: PLR0912
|
|||
code_locations: list[dict[str, Any]] | None,
|
||||
agent_id: str | None = None,
|
||||
agent_name: str | None = None,
|
||||
allow_model_dedupe: bool = True,
|
||||
) -> dict[str, Any]:
|
||||
errors: list[str] = []
|
||||
fields = {
|
||||
|
|
@ -225,8 +226,6 @@ async def _do_create( # noqa: PLR0912
|
|||
"warning": "Report could not be persisted - report state unavailable",
|
||||
}
|
||||
|
||||
from strix.report.dedupe import check_duplicate
|
||||
|
||||
existing = report_state.get_existing_vulnerabilities()
|
||||
candidate = {
|
||||
"title": title,
|
||||
|
|
@ -239,7 +238,36 @@ async def _do_create( # noqa: PLR0912
|
|||
"endpoint": endpoint,
|
||||
"method": method,
|
||||
}
|
||||
dedupe = await check_duplicate(candidate, existing)
|
||||
if allow_model_dedupe:
|
||||
from strix.report.dedupe import check_duplicate
|
||||
|
||||
dedupe = await check_duplicate(candidate, existing)
|
||||
else:
|
||||
# MCP mode must never call a model. Exact normalized identity catches
|
||||
# accidental retries while the host coding agent handles semantic
|
||||
# deduplication with ``list_findings``.
|
||||
identity = tuple(
|
||||
str(candidate.get(field) or "").strip().casefold()
|
||||
for field in ("title", "target", "endpoint", "method")
|
||||
)
|
||||
duplicate = next(
|
||||
(
|
||||
report
|
||||
for report in existing
|
||||
if tuple(
|
||||
str(report.get(field) or "").strip().casefold()
|
||||
for field in ("title", "target", "endpoint", "method")
|
||||
)
|
||||
== identity
|
||||
),
|
||||
None,
|
||||
)
|
||||
dedupe = {
|
||||
"is_duplicate": duplicate is not None,
|
||||
"duplicate_id": duplicate.get("id", "") if duplicate else "",
|
||||
"confidence": 1.0,
|
||||
"reason": "Exact normalized finding identity already exists",
|
||||
}
|
||||
if dedupe.get("is_duplicate"):
|
||||
duplicate_id = dedupe.get("duplicate_id", "")
|
||||
duplicate_title = next(
|
||||
|
|
|
|||
76
tests/test_agent_launcher.py
Normal file
76
tests/test_agent_launcher.py
Normal file
|
|
@ -0,0 +1,76 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from argparse import Namespace
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from strix.interface import agent_launcher
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
import pytest
|
||||
|
||||
|
||||
def _args(**overrides: object) -> Namespace:
|
||||
values: dict[str, object] = {
|
||||
"agent": "codex",
|
||||
"target": ["./app"],
|
||||
"mount": None,
|
||||
"scan_mode": "standard",
|
||||
"instruction": "Focus on authorization",
|
||||
"resume": None,
|
||||
"non_interactive": False,
|
||||
"max_budget_usd": None,
|
||||
"scope_mode": "auto",
|
||||
"diff_base": None,
|
||||
}
|
||||
values.update(overrides)
|
||||
return Namespace(**values)
|
||||
|
||||
|
||||
def test_auto_uses_coding_agent_without_strix_model(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(
|
||||
agent_launcher,
|
||||
"load_settings",
|
||||
lambda: SimpleNamespace(llm=SimpleNamespace(model=None)),
|
||||
)
|
||||
|
||||
assert agent_launcher.uses_coding_agent(_args(agent="auto")) is True
|
||||
|
||||
|
||||
def test_explicit_legacy_mode_does_not_launch_agent(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(
|
||||
agent_launcher,
|
||||
"load_settings",
|
||||
lambda: SimpleNamespace(llm=SimpleNamespace(model=None)),
|
||||
)
|
||||
|
||||
assert agent_launcher.uses_coding_agent(_args(agent="legacy")) is False
|
||||
|
||||
|
||||
def test_codex_command_embeds_ephemeral_mcp_config(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(Path, "cwd", classmethod(lambda cls: Path("/tmp/project")))
|
||||
|
||||
command = agent_launcher.build_agent_command(_args(non_interactive=True), "codex")
|
||||
|
||||
assert command[:2] == ["codex", "exec"]
|
||||
assert any("mcp_servers.strix.command" in item for item in command)
|
||||
assert any("strix.mcp.server" in item for item in command)
|
||||
assert "start_scan" in command[-1]
|
||||
assert "Focus on authorization" in command[-1]
|
||||
|
||||
|
||||
def test_claude_command_uses_stdio_mcp_config() -> None:
|
||||
command = agent_launcher.build_agent_command(
|
||||
_args(agent="claude", non_interactive=True, max_budget_usd=5.0),
|
||||
"claude",
|
||||
)
|
||||
|
||||
assert command[0] == "claude"
|
||||
assert "--mcp-config" in command
|
||||
assert "strix.mcp.server" in command[command.index("--mcp-config") + 1]
|
||||
assert "--print" in command
|
||||
assert command[command.index("--max-budget-usd") + 1] == "5.0"
|
||||
assert command[-2] == "--"
|
||||
assert "start_scan" in command[-1]
|
||||
9
tests/test_mcp_install.py
Normal file
9
tests/test_mcp_install.py
Normal file
|
|
@ -0,0 +1,9 @@
|
|||
from strix.mcp.install import skill_path
|
||||
|
||||
|
||||
def test_packaged_skill_path_contains_valid_skill() -> None:
|
||||
path = skill_path()
|
||||
|
||||
assert path.name == "strix-security"
|
||||
assert (path / "SKILL.md").is_file()
|
||||
assert (path / "references" / "mcp-tools.md").is_file()
|
||||
70
tests/test_mcp_service.py
Normal file
70
tests/test_mcp_service.py
Normal file
|
|
@ -0,0 +1,70 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from types import SimpleNamespace
|
||||
from typing import TYPE_CHECKING, Any, cast
|
||||
|
||||
from strix.mcp import service as service_module
|
||||
from strix.mcp.service import StrixMCPService
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
import pytest
|
||||
|
||||
from strix.report.state import ReportState
|
||||
|
||||
|
||||
async def test_sandbox_exec_requires_active_scan() -> None:
|
||||
service = StrixMCPService()
|
||||
|
||||
result = await service.sandbox_exec(["id"])
|
||||
|
||||
assert result == {"success": False, "error": "Call start_scan first"}
|
||||
|
||||
|
||||
async def test_sandbox_exec_returns_decoded_result() -> None:
|
||||
class Session:
|
||||
async def exec(self, *args: str, timeout: int) -> Any:
|
||||
assert args == ("printf", "ok")
|
||||
assert timeout == 10
|
||||
return SimpleNamespace(
|
||||
ok=lambda: True,
|
||||
exit_code=0,
|
||||
stdout=b"ok",
|
||||
stderr=b"",
|
||||
)
|
||||
|
||||
service = StrixMCPService()
|
||||
service.bundle = {"session": Session()}
|
||||
|
||||
result = await service.sandbox_exec(["printf", "ok"], timeout=10)
|
||||
|
||||
assert result == {"success": True, "exit_code": 0, "stdout": "ok", "stderr": ""}
|
||||
|
||||
|
||||
async def test_create_finding_forces_model_free_dedupe(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
captured: dict[str, Any] = {}
|
||||
|
||||
async def fake_create(**kwargs: Any) -> dict[str, bool]:
|
||||
captured.update(kwargs)
|
||||
return {"success": True}
|
||||
|
||||
monkeypatch.setattr(service_module, "_do_create", fake_create)
|
||||
service = StrixMCPService()
|
||||
service.report_state = cast("ReportState", SimpleNamespace())
|
||||
|
||||
result = await service.create_finding(title="Verified issue")
|
||||
|
||||
assert result == {"success": True}
|
||||
assert captured["allow_model_dedupe"] is False
|
||||
assert captured["agent_id"] == "coding-agent"
|
||||
|
||||
|
||||
def test_load_knowledge_returns_scan_mode() -> None:
|
||||
service = StrixMCPService()
|
||||
|
||||
result = service.load_knowledge("scan_modes/quick")
|
||||
|
||||
assert result["success"] is True
|
||||
assert "quick" in result["content"].lower()
|
||||
2
uv.lock
generated
2
uv.lock
generated
|
|
@ -2214,6 +2214,7 @@ dependencies = [
|
|||
{ name = "caido-sdk-client" },
|
||||
{ name = "cvss" },
|
||||
{ name = "docker" },
|
||||
{ name = "mcp" },
|
||||
{ name = "openai-agents", extra = ["litellm"] },
|
||||
{ name = "pydantic" },
|
||||
{ name = "pydantic-settings" },
|
||||
|
|
@ -2239,6 +2240,7 @@ requires-dist = [
|
|||
{ name = "caido-sdk-client", specifier = ">=0.2.0" },
|
||||
{ name = "cvss", specifier = ">=3.2" },
|
||||
{ name = "docker", specifier = ">=7.1.0" },
|
||||
{ name = "mcp", specifier = ">=1.28.1" },
|
||||
{ name = "openai-agents", extras = ["litellm"], specifier = "==0.14.6" },
|
||||
{ name = "pydantic", specifier = ">=2.11.3" },
|
||||
{ name = "pydantic-settings", specifier = ">=2.13.0" },
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue