Compare commits

..
46 changed files with 274 additions and 1538 deletions
+4 -4
View File
@@ -8,7 +8,7 @@ Configure Strix using environment variables or a config file.
## LLM Configuration
<ParamField path="STRIX_LLM" type="string" required>
Model name in LiteLLM format (e.g., `openai/gpt-5.4`, `anthropic/claude-sonnet-4-6`).
Model name in LiteLLM format (for example, `openai/gpt-5.4`, `anthropic/claude-sonnet-4-6`).
</ParamField>
<ParamField path="LLM_API_KEY" type="string">
@@ -20,7 +20,7 @@ Configure Strix using environment variables or a config file.
</ParamField>
<ParamField path="LLM_EXTRA_HEADERS" type="string">
Extra HTTP headers sent on every LLM request, as a JSON object (e.g.
Extra HTTP headers sent on every LLM request, as a JSON object (for example
`{"X-Feature-Key":"value","X-Tenant":"acme"}`). Useful for OpenAI-compatible
gateways that require attribution or routing headers in addition to the bearer
token. The bearer token itself still comes from `LLM_API_KEY`. Applies to both
@@ -65,8 +65,8 @@ affecting the agents that do the actual testing.
<ParamField path="DEDUPE_LLM_EXTRA_HEADERS" type="string">
Optional JSON object of extra HTTP headers sent on every deduplication-model
request, e.g. `{"X-Feature-Key":"value"}`. A dedicated dedupe model never
inherits `LLM_EXTRA_HEADERS`; set this when its endpoint needs custom headers.
request, for example `{"X-Feature-Key":"value"}`. A dedicated dedupe model never
inherits `LLM_EXTRA_HEADERS`. Set this when its endpoint needs custom headers.
</ParamField>
<ParamField path="STRIX_DEDUPE_REASONING_EFFORT" type="string">
+8 -8
View File
@@ -7,7 +7,7 @@ Skills are structured knowledge packages that give Strix agents deep expertise i
## The Idea
LLMs have broad but shallow security knowledge. They know _about_ SQL injection, but lack the nuanced techniques that experienced pentesters useparser quirks, bypass methods, validation tricks, and chain attacks.
LLMs have broad but shallow security knowledge. They know _about_ SQL injection, but lack the nuanced techniques that experienced pentesters use: parser quirks, bypass methods, validation tricks, and chain attacks.
Skills inject this deep, specialized knowledge directly into the agent's context, transforming it from a generalist into a specialist for the task at hand.
@@ -25,9 +25,9 @@ create_agent(
The skills are injected into the agent's system prompt, giving it access to:
- **Advanced techniques** Non-obvious methods beyond standard testing
- **Working payloads** Practical examples with variations
- **Validation methods** How to confirm findings and avoid false positives
- **Advanced techniques**: Non-obvious methods beyond standard testing
- **Working payloads**: Practical examples with variations
- **Validation methods**: How to confirm findings and avoid false positives
## Skill Categories
@@ -138,7 +138,7 @@ How to confirm findings and avoid false positives.
Community contributions are welcome. Create a `.md` file in the appropriate category with YAML frontmatter (`name` and `description` fields). Good skills include:
1. **Real-world techniques** Methods that work in practice
2. **Practical payloads** Working examples with variations
3. **Validation steps** How to confirm without false positives
4. **Context awareness** Version/environment-specific behavior
1. **Real-world techniques**: Methods that work in practice
2. **Practical payloads**: Working examples with variations
3. **Validation steps**: How to confirm without false positives
4. **Context awareness**: Version/environment-specific behavior
+4 -4
View File
@@ -24,10 +24,10 @@ Skip the setup. Run Strix in the cloud at [app.strix.ai](https://app.strix.ai).
## What You Get
- **Penetration test reports** Validated findings with PoCs
- **Shareable dashboards** Collaborate with your team
- **CI/CD integration** Block risky changes automatically
- **Continuous monitoring** Catch new vulnerabilities quickly
- **Penetration test reports**: Validated findings with PoCs
- **Shareable dashboards**: Collaborate with your team
- **CI/CD integration**: Block risky changes automatically
- **Continuous monitoring**: Catch new vulnerabilities quickly
## Getting Started
+9 -9
View File
@@ -52,20 +52,20 @@ Skills are specialized knowledge packages that enhance agent capabilities. They
1. Choose the right category
2. Create a `.md` file with YAML frontmatter (`name` and `description` fields)
3. Include practical examplesworking payloads, commands, test cases
3. Include practical examples: working payloads, commands, test cases
4. Provide validation methods to confirm findings
5. Submit via PR
5. Submit through PR
## Contributing Code
### Pull Request Process
1. **Create an issue first** Describe the problem or feature
2. **Fork and branch** Work from `main`
3. **Make changes** Follow existing code style
4. **Write tests** Ensure coverage for new features
5. **Run checks** `make check-all` should pass
6. **Submit PR** Link to issue and provide context
1. **Create an issue first**: Describe the problem or feature
2. **Fork and branch**: Work from `main`
3. **Make changes**: Follow existing code style
4. **Write tests**: Ensure coverage for new features
5. **Run checks**: `make check-all` should pass
6. **Submit PR**: Link to issue and provide context
### Code Style
@@ -77,7 +77,7 @@ Skills are specialized knowledge packages that enhance agent capabilities. They
## Package Builds
Editable installs do not require Go; they run the TUI from source (`go run`).
Editable installs do not require Go. They run the TUI from source (`go run`).
Wheels are intentionally strict: they always bundle the matching Go sidecar and
are platform-specific.
+12 -12
View File
@@ -3,7 +3,7 @@ title: "Introduction"
description: "Open-source AI hackers to secure your apps"
---
Strix are autonomous AI agents that act like real hackersthey run your code dynamically, find vulnerabilities, and validate them with 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 agents that act like real hackers: they run your code dynamically, find vulnerabilities, and validate them with 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.
<Frame>
<img src="/images/screenshot.png" alt="Strix Demo" />
@@ -26,17 +26,17 @@ Strix are autonomous AI agents that act like real hackers—they run your code d
## Use Cases
- **Application Security Testing** Detect and validate critical vulnerabilities in your applications
- **Rapid Penetration Testing** Get penetration tests done in hours, not weeks
- **Bug Bounty Automation** Automate research and generate PoCs for faster reporting
- **CI/CD Integration** Block vulnerabilities before they reach production
- **Application Security Testing**: Detect and validate critical vulnerabilities in your applications
- **Rapid Penetration Testing**: Get penetration tests done in hours, not weeks
- **Bug Bounty Automation**: Automate research and generate PoCs for faster reporting
- **CI/CD Integration**: Block vulnerabilities before they reach production
## Key Capabilities
- **Full hacker toolkit** Browser automation, HTTP proxy, terminal, Python runtime
- **Real validation** PoCs, not false positives
- **Multi-agent orchestration** Specialized agents collaborate on complex targets
- **Developer-first CLI** Interactive TUI or headless mode for automation
- **Full hacker toolkit**: Browser automation, HTTP proxy, terminal, Python runtime
- **Real validation**: PoCs, not false positives
- **Multi-agent orchestration**: Specialized agents collaborate on complex targets
- **Developer-first CLI**: Interactive TUI or headless mode for automation
## Security Tools
@@ -67,9 +67,9 @@ Strix agents come equipped with a comprehensive toolkit:
Strix uses a graph of specialized agents for comprehensive security 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 Workflows**: Specialized agents for different attacks and assets
- **Scalable Testing**: Parallel execution for fast comprehensive coverage
- **Dynamic Coordination**: Agents collaborate and share discoveries
## Quick Example
+12 -12
View File
@@ -7,7 +7,7 @@ Strix is built to be driven by AI coding agents. Install the official agent skil
## Install the Skills
Works with any agent that supports the open [SKILL.md standard](https://agentskills.io) Claude Code, Cursor, Codex, Gemini CLI, OpenCode, and dozens more:
Works with any agent that supports the open [SKILL.md standard](https://agentskills.io): Claude Code, Cursor, Codex, Gemini CLI, OpenCode, and dozens more:
```bash
npx skills add usestrix/strix
@@ -15,8 +15,8 @@ npx skills add usestrix/strix
| Skill | What your agent learns |
|-------|------------------------|
| `penetration-testing-with-strix` | Run headless scans against code, URLs, domains, or IPs self-hosted CLI or managed cloud with budget caps, and read the results |
| `managed-pentesting-with-strix` | Drive the managed [app.strix.ai](https://app.strix.ai) platform over REST no local Docker or LLM key needed |
| `penetration-testing-with-strix` | Run headless scans against code, URLs, domains, or IPs (self-hosted CLI or managed cloud) with budget caps, and read the results |
| `managed-pentesting-with-strix` | Drive the managed [app.strix.ai](https://app.strix.ai) platform over REST: no local Docker or LLM key needed |
| `fix-security-vulnerabilities-with-strix` | Triage findings, fix root causes, and re-run Strix to verify each fix |
| `ci-security-scanning-with-strix` | Add PR security scanning to GitHub Actions or any CI (self-hosted CLI or managed app) |
@@ -26,23 +26,23 @@ Install a single skill with `npx skills add usestrix/strix --skill penetration-t
npx skills use usestrix/strix@penetration-testing-with-strix | claude
```
## Two ways to run self-hosted or managed
## Two ways to run: self-hosted or managed
Both use the same engine and produce the same validated findings and SARIF, so agents can pick per situation or combine them:
- **Open-source CLI (self-hosted)** runs locally in a Docker sandbox with your own LLM key. Free, fully local, air-gap capable. Best for local dev loops and full control.
- **Managed cloud** runs on Strix's infrastructure via the [app.strix.ai REST API](https://docs.app.strix.ai). No Docker, no LLM key, no local install; adds team dashboards, scheduling, PR reviews, and downloadable PDF/DOCX reports (Enterprise plan). Best in sandboxed/CI environments and for teams. Create an API token under **Settings → API Access**; the `managed-pentesting-with-strix` skill has the full flow.
- **Open-source CLI (self-hosted)**: runs locally in a Docker sandbox with your own LLM key. Free, fully local, air-gap capable. Best for local dev loops and full control.
- **Managed cloud**: runs on Strix's infrastructure through the [app.strix.ai REST API](https://docs.app.strix.ai). No Docker, no LLM key, no local install, adds team dashboards, scheduling, PR reviews, and downloadable PDF/DOCX reports (Enterprise plan). Best in sandboxed/CI environments and for teams. Create an API token under **Settings → API Access**. The `managed-pentesting-with-strix` skill has the full flow.
## Agent-Friendly Interfaces
Everything an agent needs is machine-readable:
- **Headless CLI** `strix -n` runs without the TUI and exits with `0` (clean), `1` (error), or `2` (vulnerabilities found).
- **REST API** the managed platform exposes a documented [OpenAPI](https://docs.app.strix.ai/openapi.json) at `https://app.strix.ai/api/v1` (scans, vulnerabilities, assets, PR reviews, schedules, webhooks) with bearer tokens and scopes.
- **Structured results** every run writes `vulnerabilities.json`, `vulnerabilities.csv`, `findings.sarif` (SARIF 2.1.0), and per-finding Markdown under `strix_runs/<run-name>/`; the cloud exposes the same as JSON plus SARIF export.
- **Budget controls** `--max-budget` and `--max-turns` give agents hard cost/time caps.
- **`AGENTS.md`** the [repository's agent guide](https://github.com/usestrix/strix/blob/main/AGENTS.md) with a quick reference.
- **`llms.txt`** this documentation is indexed at [docs.strix.ai/llms.txt](https://docs.strix.ai/llms.txt) and fully exported at [docs.strix.ai/llms-full.txt](https://docs.strix.ai/llms-full.txt); every page is also available as Markdown by appending `.md` to its URL.
- **Headless CLI**: `strix -n` runs without the TUI and exits with `0` (clean), `1` (error), or `2` (vulnerabilities found).
- **REST API**: the managed platform exposes a documented [OpenAPI](https://docs.app.strix.ai/openapi.json) at `https://app.strix.ai/api/v1` (scans, vulnerabilities, assets, PR reviews, schedules, webhooks) with bearer tokens and scopes.
- **Structured results**: every run writes `vulnerabilities.json`, `vulnerabilities.csv`, `findings.sarif` (SARIF 2.1.0), and per-finding Markdown under `strix_runs/<run-name>/`. The cloud exposes the same as JSON plus SARIF export.
- **Budget controls**: `--max-budget` and `--max-turns` give agents hard cost/time caps.
- **`AGENTS.md`**: the [repository's agent guide](https://github.com/usestrix/strix/blob/main/AGENTS.md) with a quick reference.
- **`llms.txt`**: this documentation is indexed at [docs.strix.ai/llms.txt](https://docs.strix.ai/llms.txt) and fully exported at [docs.strix.ai/llms-full.txt](https://docs.strix.ai/llms-full.txt). Every page is also available as Markdown by appending `.md` to its URL.
## Example Prompts
+3 -3
View File
@@ -37,7 +37,7 @@ Add these secrets to your repository:
| Secret | Description |
|--------|-------------|
| `STRIX_LLM` | Model name (e.g., `openai/gpt-5.4`) |
| `STRIX_LLM` | Model name (for example, `openai/gpt-5.4`) |
| `LLM_API_KEY` | API key for your LLM provider |
## Exit Codes
@@ -46,8 +46,8 @@ The workflow fails when vulnerabilities are found:
| Code | Result |
|------|--------|
| 0 | Pass No vulnerabilities |
| 2 | Fail Vulnerabilities found |
| 0 | Pass: No vulnerabilities |
| 2 | Fail: Vulnerabilities found |
## Scan Modes for CI
+3 -3
View File
@@ -1,6 +1,6 @@
---
title: "Azure OpenAI"
description: "Configure Strix with OpenAI models via Azure"
description: "Configure Strix with OpenAI models through Azure"
---
## Setup
@@ -19,7 +19,7 @@ export AZURE_API_VERSION="2025-11-01-preview"
| `STRIX_LLM` | `azure/<your-deployment-name>` |
| `AZURE_API_KEY` | Your Azure OpenAI API key |
| `AZURE_API_BASE` | Your Azure OpenAI endpoint URL |
| `AZURE_API_VERSION` | API version (e.g., `2025-11-01-preview`) |
| `AZURE_API_VERSION` | API version (for example, `2025-11-01-preview`) |
## Example
@@ -33,5 +33,5 @@ export AZURE_API_VERSION="2025-11-01-preview"
## Prerequisites
1. Create an Azure OpenAI resource
2. Deploy a model (e.g., GPT-5.4)
2. Deploy a model (for example, GPT-5.4)
3. Get the endpoint URL and API key from the Azure portal
+2 -2
View File
@@ -1,6 +1,6 @@
---
title: "AWS Bedrock"
description: "Configure Strix with models via AWS Bedrock"
description: "Configure Strix with models through AWS Bedrock"
---
## Installation
@@ -17,7 +17,7 @@ pipx install "strix-agent[bedrock]"
export STRIX_LLM="bedrock/anthropic.claude-4-5-sonnet-20251022-v1:0"
```
No API key requireduses AWS credentials from environment.
No API key required: uses AWS credentials from environment.
## Authentication
+9 -9
View File
@@ -59,7 +59,7 @@ export LLM_API_BASE="http://localhost:1234/v1" # Adjust port as needed
Some OpenAI-compatible gateways require extra HTTP headers (for attribution or
tenant routing) alongside the bearer token. Set them with `LLM_EXTRA_HEADERS` as
a JSON object — they are sent on every request:
a JSON object. They are sent on every request:
```bash
export STRIX_LLM="openai/your-model"
@@ -69,12 +69,12 @@ export LLM_EXTRA_HEADERS='{"X-Feature-Key":"value","X-Tenant":"acme"}'
```
For endpoints behind a private CA, point Strix at your certificate bundle with
the standard `SSL_CERT_FILE=/path/to/ca-bundle.pem` never disable TLS
the standard `SSL_CERT_FILE=/path/to/ca-bundle.pem`: never disable TLS
verification against a real endpoint.
## Tool calling must return structured `tool_calls`
Strix is entirely tool-driven: every working turn must be a **native** function/tool call. If your inference server returns the tool call as plain assistant text instead of a structured `tool_calls` field, Strix never sees a call it can execute, so the agent makes no real progress — it re-prompts the model for a tool call and gives up once its recovery attempts are exhausted.
Strix is entirely tool-driven: every working turn must be a **native** function/tool call. If your inference server returns the tool call as plain assistant text instead of a structured `tool_calls` field, Strix never sees a call it can execute, so the agent makes no real progress. It re-prompts the model for a tool call and gives up once its recovery attempts are exhausted.
This is almost always an **inference-server configuration** problem, not a model or Strix problem. Common symptoms are the model printing a call as text such as:
@@ -84,19 +84,19 @@ exec_command(cmd="nmap ...", timeout=180)
{"action": "exec_command", "params": {"cmd": "nmap ..."}}
```
The fix belongs on the inference server: it must be configured to parse the model's tool tokens into structured `tool_calls`. A correctly configured endpoint either returns a structured call or rejects the request outright — it never leaks the call as text.
The fix belongs on the inference server: it must be configured to parse the model's tool tokens into structured `tool_calls`. A correctly configured endpoint either returns a structured call or rejects the request outright. It never leaks the call as text.
### Fixes by server
**llama.cpp (`llama-server`)**
- Run with `--jinja` and a correct tool-use chat template (`--chat-template` / `--chat-template-file` matching the model). Recent builds enable `--jinja` by default **upgrade** if yours doesn't.
- For thinking models, align or disable reasoning (`--reasoning-format`, `-rea off`) so it doesn't break tool-call parsing.
- A low temperature (e.g. `--temp 0.2`) improves tool-call reliability.
- Run with `--jinja` and a correct tool-use chat template (`--chat-template` / `--chat-template-file` matching the model). Recent builds enable `--jinja` by default. **Upgrade** if yours does not.
- For thinking models, align or disable reasoning (`--reasoning-format`, `-rea off`) so it does not break tool-call parsing.
- A low temperature (for example `--temp 0.2`) improves tool-call reliability.
**Ollama**
- Use a recent Ollama and a model whose template wires tools. Modern Ollama refuses tools (`tools param requires --jinja flag`) if the template lacks tool support.
- For reasoning models (e.g. qwen3), disable the model's **thinking** mode thinking left on frequently pushes the tool call into the text `content` instead of the structured `tool_calls` field. Turn it off on the Ollama side (a non-thinking model variant, or `think: false` in the model's parameters / `Modelfile`).
- Raise **`num_ctx`** to at least 16k32k. Strix sends a large system prompt plus many tool schemas; at Ollama's small default context the tool definitions are truncated out of the prompt and the model stops emitting valid calls. A short test prompt can look fine while a real scan fails, so set this explicitly rather than inferring it from a quick check.
- For reasoning models (for example qwen3), disable the model's **thinking** mode: thinking left on frequently pushes the tool call into the text `content` instead of the structured `tool_calls` field. Turn it off on the Ollama side (a non-thinking model variant, or `think: false` in the model's parameters / `Modelfile`).
- Raise **`num_ctx`** to at least 16k32k. Strix sends a large system prompt plus many tool schemas. At Ollama's small default context the tool definitions are truncated out of the prompt and the model stops emitting valid calls. A short test prompt can look fine while a real scan fails, so set this explicitly rather than inferring it from a quick check.
**vLLM**
- Start with `--enable-auto-tool-choice`, a matching `--tool-call-parser` (`hermes`, `qwen3_xml`, or `llama3_json`), and a matching `--reasoning-parser` for reasoning models.
+5 -5
View File
@@ -3,7 +3,7 @@ title: "Novita AI"
description: "Configure Strix with Novita AI models"
---
[Novita AI](https://novita.ai) provides fast, cost-efficient inference for open-source models via an OpenAI-compatible API.
[Novita AI](https://novita.ai) provides fast, cost-efficient inference for open-source models through an OpenAI-compatible API.
## Setup
@@ -29,7 +29,7 @@ export LLM_API_BASE="https://api.novita.ai/openai"
## Benefits
- **Cost-efficient** Competitive pricing with per-token billing
- **OpenAI-compatible** Drop-in replacement using `LLM_API_BASE`
- **Large context** Models support up to 262k token context windows
- **Function calling** All listed models support tool/function calling
- **Cost-efficient**: Competitive pricing with per-token billing
- **OpenAI-compatible**: Drop-in replacement using `LLM_API_BASE`
- **Large context**: Models support up to 262k token context windows
- **Function calling**: All listed models support tool/function calling
+5 -5
View File
@@ -1,6 +1,6 @@
---
title: "OpenRouter"
description: "Configure Strix with models via OpenRouter"
description: "Configure Strix with models through OpenRouter"
---
[OpenRouter](https://openrouter.ai) provides access to 100+ models from multiple providers through a single API.
@@ -31,7 +31,7 @@ Access any model on OpenRouter using the format `openrouter/<provider>/<model>`:
## Benefits
- **Single API** Access models from OpenAI, Anthropic, Google, Meta, and more
- **Fallback routing** Automatic failover between providers
- **Cost tracking** Monitor usage across all models
- **Higher rate limits** OpenRouter handles provider limits for you
- **Single API**: Access models from OpenAI, Anthropic, Google, Meta, and more
- **Fallback routing**: Automatic failover between providers
- **Cost tracking**: Monitor usage across all models
- **Higher rate limits**: OpenRouter handles provider limits for you
+3 -3
View File
@@ -44,13 +44,13 @@ See the [Local Models guide](/llm-providers/local) for setup instructions and re
Access 100+ models through a single API.
</Card>
<Card title="Google Vertex AI" href="/llm-providers/vertex">
Gemini 3 models via Google Cloud.
Gemini 3 models through Google Cloud.
</Card>
<Card title="AWS Bedrock" href="/llm-providers/bedrock">
Claude and Titan models via AWS.
Claude and Titan models through AWS.
</Card>
<Card title="Azure OpenAI" href="/llm-providers/azure">
GPT-5.4 via Azure.
GPT-5.4 through Azure.
</Card>
<Card title="Local Models" href="/llm-providers/local">
Llama 4, Mistral, and self-hosted models.
+2 -2
View File
@@ -1,6 +1,6 @@
---
title: "Google Vertex AI"
description: "Configure Strix with Gemini models via Google Cloud"
description: "Configure Strix with Gemini models through Google Cloud"
---
## Installation
@@ -17,7 +17,7 @@ pipx install "strix-agent[vertex]"
export STRIX_LLM="vertex_ai/gemini-3-pro-preview"
```
No API key requireduses Google Cloud Application Default Credentials.
No API key required: uses Google Cloud Application Default Credentials.
## Authentication
+1 -1
View File
@@ -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.)
- An LLM API key from any [supported provider](/llm-providers/overview) (OpenAI, Anthropic, Google)
## Installation
+1 -1
View File
@@ -3,7 +3,7 @@ title: "Browser"
description: "Playwright-powered Chrome for web application testing"
---
Strix uses a headless Chrome browser via Playwright to interact with web applications exactly like a real user would.
Strix uses a headless Chrome browser through Playwright to interact with web applications exactly like a real user would.
## How It Works
+1 -1
View File
@@ -28,6 +28,6 @@ Strix agents use specialized tools to test your applications like a real penetra
| -------------- | ---------------------------------------- |
| Python Runtime | Write and execute custom exploit scripts |
| File Editor | Read and modify source code |
| Web Search | Real-time OSINT via Perplexity |
| Web Search | Real-time OSINT through Perplexity |
| Notes | Document findings during the scan |
| Reporting | Generate vulnerability reports with PoCs |
+10 -10
View File
@@ -70,9 +70,9 @@ asyncio.run(main())
| `view_sitemap_entry()` | Inspect one sitemap entry + its related requests |
| `scope_rules()` | Manage proxy scope (allowlist/denylist) |
For one-off arbitrary requests, use shell tooling like `curl` the
For one-off arbitrary requests, use shell tooling like `curl`: the
sandbox's `HTTP_PROXY` env routes the traffic through Caido
automatically, so it lands in `list_requests` and can be replayed via
automatically, so it lands in `list_requests` and can be replayed through
`repeat_request`.
### Example: Automated IDOR Testing
@@ -106,24 +106,24 @@ asyncio.run(main())
## Human-in-the-Loop
Strix exposes the Caido proxy to your host machine, so you can interact with it alongside the automated scan. When the sandbox starts, the Caido URL is displayed in the TUI sidebar click it to copy, then open it in Caido Desktop.
Strix exposes the Caido proxy to your host machine, so you can interact with it alongside the automated scan. When the sandbox starts, the Caido URL is displayed in the TUI sidebar: click it to copy, then open it in Caido Desktop.
### Accessing Caido
1. Start a scan as usual
2. Look for the **Caido** URL in the sidebar stats panel (e.g. `localhost:52341`)
2. Look for the **Caido** URL in the sidebar stats panel (for example `localhost:52341`)
3. Open the URL in Caido Desktop
4. Click **Continue as guest** to access the instance
### What You Can Do
- **Inspect traffic** Browse all HTTP/HTTPS requests the agent is making in real time
- **Replay requests** Take any captured request and resend it with your own modifications
- **Intercept and modify** Pause requests mid-flight, edit them, then forward
- **Explore the sitemap** See the full attack surface the agent has discovered
- **Manual testing** Use Caido's tools to test findings the agent reports, or explore areas it hasn't reached
- **Inspect traffic**: Browse all HTTP/HTTPS requests the agent is making in real time
- **Replay requests**: Take any captured request and resend it with your own modifications
- **Intercept and modify**: Pause requests mid-flight, edit them, then forward
- **Explore the sitemap**: See the full attack surface the agent has discovered
- **Manual testing**: Use Caido's tools to test findings the agent reports, or explore areas it has not reached
This turns Strix from a fully automated scanner into a collaborative tool the agent handles the heavy lifting while you focus on the interesting parts.
This turns Strix from a fully automated scanner into a collaborative tool: the agent handles the heavy lifting while you focus on the interesting parts.
## Scope
+7 -7
View File
@@ -14,14 +14,14 @@ strix (--target <target> | --target-list <path>) [options]
<ParamField path="--target, -t" type="string">
Target to test. Accepts URLs, repositories, local directories, domains, IP addresses, API spec files (OpenAPI/Swagger `.json`/`.yaml`, a Postman collection export), or a live Postman collection by id (`postman://<collection-uuid>`). Can be specified multiple times. Fresh runs require at least one target source: `--target` or `--target-list`.
When the target is an API spec, Strix copies it into the agent's workspace and authorizes the base URLs it declares (including those resolved from a Postman environment) as in-scope hosts - so the agent reads the contract and tests the full declared surface instead of discovering endpoints by crawling. Pair the spec with the deployed base URL (e.g. `--target ./openapi.yaml --target https://api.example.com`) so the agent has a reachable host to attack.
When the target is an API spec, Strix copies it into the agent's workspace and authorizes the base URLs it declares (including those resolved from a Postman environment) as in-scope hosts - so the agent reads the contract and tests the full declared surface instead of discovering endpoints by crawling. Pair the spec with the deployed base URL (for example `--target ./openapi.yaml --target https://api.example.com`) so the agent has a reachable host to attack.
<Note>
A local directory is mounted into the sandbox live and **writable**, so the agent edits your real files (`.git` excepted). Commit or stash first.
</Note>
<Note>
Fetching a Postman collection by id requires `POSTMAN_API_KEY`. Add `?env=<environment-uuid>` to also pull a Postman environment, which resolves `{{baseUrl}}` / token variables the collection references (e.g. `postman://<collection-uuid>?env=<environment-uid>`).
Fetching a Postman collection by id requires `POSTMAN_API_KEY`. Add `?env=<environment-uuid>` to also pull a Postman environment, which resolves `{{baseUrl}}` / token variables the collection references (for example `postman://<collection-uuid>?env=<environment-uid>`).
</Note>
</ParamField>
@@ -46,7 +46,7 @@ strix (--target <target> | --target-list <path>) [options]
</ParamField>
<ParamField path="--diff-base" type="string">
Target branch or commit to compare against (e.g., `origin/main`). Defaults to the repository's default branch.
Target branch or commit to compare against (for example, `origin/main`). Defaults to the repository's default branch.
</ParamField>
<ParamField path="--non-interactive, -n" type="boolean">
@@ -88,8 +88,8 @@ strix (--target <target> | --target-list <path>) [options]
slightly overshoot the limit by any calls already in flight when the
threshold is crossed (most relevant with several child agents running
concurrently).
- Cost is a best-effort estimate derived from token usage and model pricing;
providers that do not expose priced usage may under-count.
- Cost is a best-effort estimate derived from token usage and model pricing.
Providers that do not expose priced usage may under-count.
- For LiteLLM-routed models, Strix enables streaming success callbacks to
capture provider-reported cost. Message content remains excluded, but
third-party LiteLLM callbacks configured in the same process can receive
@@ -148,6 +148,6 @@ strix --target-list ./targets.txt
| Code | Meaning |
|------|---------|
| 0 | Scan completed successfully (interactive mode always exits `0`; in headless mode, `0` means no vulnerabilities were found) |
| 1 | A fatal error occurred before or during the scan (e.g. missing environment variables, Docker unavailable, invalid config file, diff-scope resolution failure, or an unhandled error) |
| 0 | Scan completed successfully (interactive mode always exits `0`, in headless mode, `0` means no vulnerabilities were found) |
| 1 | A fatal error occurred before or during the scan (for example missing environment variables, Docker unavailable, invalid config file, diff-scope resolution failure, or an unhandled error) |
| 2 | Vulnerabilities found (headless mode only) |
-7
View File
@@ -230,7 +230,6 @@ ignore = [
# args they intentionally ignore.
"tests/test_viewer_auth.py" = ["S105", "S106", "ARG001"]
"tests/test_codex_auth.py" = ["S105", "S106", "SLF001"]
"tests/test_grok_auth.py" = ["S105", "S106", "SLF001"]
# Hatchling loads the build hook by path, not as an importable package.
"scripts/tui_sidecar_hook.py" = ["INP001"]
# Stdlib HTTP handler overrides (do_GET/do_POST).
@@ -245,9 +244,6 @@ ignore = [
# Stdlib HTTP handler overrides (do_GET/do_POST) and lazy imports that avoid a
# circular dependency with strix.telemetry / strix.interface.viewer.report_pdf.
"strix/interface/viewer/server.py" = ["N802", "PLC0415"]
# Lazy import of the TUI live-view projection so importing the viewer does not
# eagerly pull in the Textual TUI.
"strix/interface/viewer/transcript.py" = ["PLC0415"]
# Lazy telemetry import to avoid importing PostHog before the viewer starts.
"strix/interface/viewer/cli.py" = ["PLC0415"]
# Lazy imports inside functions to avoid circular dependency with
@@ -292,9 +288,6 @@ ignore = [
# Heavy inference deps (httpx, openai) imported lazily so auth-status checks
# don't pull them in.
"strix/config/codex.py" = ["PLC0415"]
"strix/config/grok.py" = ["PLC0415"]
# Lazy ``import fcntl`` so the module imports on non-POSIX platforms.
"strix/config/subscription_store.py" = ["PLC0415"]
# Interface utility branches per scope-mode / target-type combination;
# splitting would obscure the decision tree without simplifying it.
"strix/interface/utils.py" = ["PLR0912", "BLE001", "PLC0415"]
+47 -18
View File
@@ -16,6 +16,7 @@ import hashlib
import json
import logging
import secrets
import threading
import time
import urllib.parse
from pathlib import Path
@@ -23,7 +24,7 @@ from typing import TYPE_CHECKING, Any
import requests
from strix.config import subscription_store
from strix.utils.secret_files import write_secret_text
if TYPE_CHECKING:
@@ -53,12 +54,26 @@ _ACCOUNT_CLAIM = "https://api.openai.com/auth"
_TOKEN_TIMEOUT = 30
_EXPIRY_SKEW_S = 300
_refresh_lock = threading.Lock()
# Kept separate from cli-config.json so OAuth tokens never land in the env-var config.
AUTH_PATH = Path.home() / ".strix" / "subscription-auth.json"
def _read_store() -> dict[str, Any]:
try:
data = json.loads(AUTH_PATH.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
return {}
return data if isinstance(data, dict) else {}
def _write_store(data: dict[str, Any]) -> None:
write_secret_text(AUTH_PATH, json.dumps(data, indent=2))
def read_record() -> dict[str, Any] | None:
record = subscription_store.read(AUTH_PATH).get(PROVIDER)
record = _read_store().get(PROVIDER)
if not isinstance(record, dict) or record.get("type") != "oauth":
return None
if not (record.get("access") and record.get("refresh") and record.get("account_id")):
@@ -71,31 +86,45 @@ def is_authenticated() -> bool:
def save_record(record: dict[str, Any]) -> None:
with subscription_store.guard(AUTH_PATH):
data = subscription_store.read(AUTH_PATH)
data[PROVIDER] = record
subscription_store.write(AUTH_PATH, data)
data = _read_store()
data[PROVIDER] = record
_write_store(data)
def logout() -> None:
with subscription_store.guard(AUTH_PATH):
data = subscription_store.read(AUTH_PATH)
if PROVIDER not in data:
return
del data[PROVIDER]
if data:
subscription_store.write(AUTH_PATH, data)
return
with contextlib.suppress(OSError):
AUTH_PATH.unlink()
data = _read_store()
if PROVIDER not in data:
return
del data[PROVIDER]
if data:
_write_store(data)
return
with contextlib.suppress(OSError):
AUTH_PATH.unlink()
@contextlib.contextmanager
def _refresh_guard() -> Iterator[None]:
"""Serialize token refresh within (lock) and across (flock) Strix processes,
so concurrent runs can't both spend the single-use refresh token."""
with subscription_store.guard(AUTH_PATH):
yield
with _refresh_lock:
try:
import fcntl
lock_path = AUTH_PATH.with_suffix(".lock")
lock_path.parent.mkdir(parents=True, exist_ok=True)
handle = lock_path.open("w")
except (ImportError, OSError):
yield
return
try:
with contextlib.suppress(OSError):
fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
yield
finally:
with contextlib.suppress(OSError):
fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
handle.close()
class CodexAuthError(Exception):
-310
View File
@@ -1,310 +0,0 @@
"""Grok (xAI) subscription auth: OAuth login, token refresh, and the OpenAI
client that routes inference through xAI's API.
Mirrors xAI's Grok CLI: OAuth 2.0 + PKCE against ``auth.x.ai``, with the access
token sent as a ``Bearer`` token to ``api.x.ai/v1`` (OpenAI-compatible, so the
subscription and a metered API key share one endpoint — only the bearer differs).
Using a Grok/SuperGrok subscription outside xAI's own products is not officially
supported by xAI; the user chooses this path knowingly. The OAuth constants are
xAI's own Grok CLI values (the backend only accepts that client).
"""
from __future__ import annotations
import base64
import contextlib
import hashlib
import json
import logging
import secrets
import time
import urllib.parse
from pathlib import Path
from typing import TYPE_CHECKING, Any
import requests
from strix.config import subscription_store
if TYPE_CHECKING:
from collections.abc import Iterator
from openai import AsyncOpenAI
logger = logging.getLogger(__name__)
PROVIDER = "grok"
CLIENT_ID = "b1a00492-073a-47ea-816f-4c329264a828"
AUTHORIZE_URL = "https://auth.x.ai/oauth2/authorize"
TOKEN_URL = "https://auth.x.ai/oauth2/token" # noqa: S105 # nosec B105 - URL, not a secret
CALLBACK_HOST = "127.0.0.1"
CALLBACK_PORT = 56121
CALLBACK_PATH = "/callback"
REDIRECT_URI = f"http://{CALLBACK_HOST}:{CALLBACK_PORT}{CALLBACK_PATH}"
SCOPE = "openid profile email offline_access grok-cli:access api:access"
XAI_BASE_URL = "https://api.x.ai/v1"
_TOKEN_TIMEOUT = 30
_EXPIRY_SKEW_S = 300
# Shared with the other subscription providers; kept separate from cli-config.json
# so OAuth tokens never land in the env-var config.
AUTH_PATH = Path.home() / ".strix" / "subscription-auth.json"
def read_record() -> dict[str, Any] | None:
record = subscription_store.read(AUTH_PATH).get(PROVIDER)
if not isinstance(record, dict) or record.get("type") != "oauth":
return None
if not (record.get("access") and record.get("refresh")):
return None
return record
def is_authenticated() -> bool:
return read_record() is not None
def save_record(record: dict[str, Any]) -> None:
with subscription_store.guard(AUTH_PATH):
data = subscription_store.read(AUTH_PATH)
data[PROVIDER] = record
subscription_store.write(AUTH_PATH, data)
def logout() -> None:
with subscription_store.guard(AUTH_PATH):
data = subscription_store.read(AUTH_PATH)
if PROVIDER not in data:
return
del data[PROVIDER]
if data:
subscription_store.write(AUTH_PATH, data)
return
with contextlib.suppress(OSError):
AUTH_PATH.unlink()
@contextlib.contextmanager
def _refresh_guard() -> Iterator[None]:
"""Serialize token refresh within (lock) and across (flock) Strix processes,
so concurrent runs can't both spend the single-use refresh token."""
with subscription_store.guard(AUTH_PATH):
yield
class GrokAuthError(Exception):
def __init__(self, code: str, message: str | None = None) -> None:
self.code = code
super().__init__(message or code)
def _b64url(raw: bytes) -> str:
return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii")
def generate_pkce() -> tuple[str, str]:
verifier = _b64url(secrets.token_bytes(64))
challenge = _b64url(hashlib.sha256(verifier.encode("ascii")).digest())
return verifier, challenge
def create_state() -> str:
return secrets.token_hex(16)
def build_authorize_url(challenge: str, state: str) -> str:
params = {
"response_type": "code",
"client_id": CLIENT_ID,
"redirect_uri": REDIRECT_URI,
"scope": SCOPE,
"code_challenge": challenge,
"code_challenge_method": "S256",
"state": state,
}
return f"{AUTHORIZE_URL}?{urllib.parse.urlencode(params)}"
def parse_redirect_input(value: str) -> tuple[str | None, str | None]:
"""Extract ``(code, state)`` from a pasted redirect URL, ``code#state``,
query string, or bare code."""
value = (value or "").strip()
if not value:
return None, None
with contextlib.suppress(ValueError):
parsed = urllib.parse.urlparse(value)
if parsed.scheme and parsed.query:
query = urllib.parse.parse_qs(parsed.query)
return _first(query, "code"), _first(query, "state")
if "#" in value:
code, _, state = value.partition("#")
return code or None, state or None
if "code=" in value:
query = urllib.parse.parse_qs(value)
return _first(query, "code"), _first(query, "state")
return value, None
def _first(query: dict[str, list[str]], key: str) -> str | None:
values = query.get(key)
return values[0] if values else None
def _post_form(payload: dict[str, str]) -> dict[str, Any]:
try:
response = requests.post(
TOKEN_URL,
data=payload,
headers={"Accept": "application/json"},
timeout=_TOKEN_TIMEOUT,
)
except requests.RequestException as exc:
raise GrokAuthError("unavailable", str(exc)) from exc
if response.status_code >= 400:
detail = response.text[:300]
raise GrokAuthError("token_http_error", f"HTTP {response.status_code}: {detail}")
data = json.loads(response.content or b"{}")
if not isinstance(data, dict):
raise GrokAuthError("bad_response", "token endpoint returned non-object")
return data
def _record_from_token_response(
data: dict[str, Any], refresh_fallback: str | None = None
) -> dict[str, Any]:
access = data.get("access_token")
# A refresh response may omit refresh_token when it isn't rotated; keep the old one.
refresh = data.get("refresh_token") or refresh_fallback
expires_in = data.get("expires_in")
if not isinstance(access, str) or not access:
raise GrokAuthError("bad_response", "token response missing access_token")
if not isinstance(refresh, str) or not refresh:
raise GrokAuthError("bad_response", "token response missing refresh_token")
ttl = expires_in if isinstance(expires_in, int | float) else 3600
return {
"type": "oauth",
"provider": PROVIDER,
"access": access,
"refresh": refresh,
"expires_at": time.time() + ttl,
}
def exchange_code(code: str, verifier: str) -> dict[str, Any]:
data = _post_form(
{
"grant_type": "authorization_code",
"client_id": CLIENT_ID,
"code": code,
"code_verifier": verifier,
"redirect_uri": REDIRECT_URI,
}
)
return _record_from_token_response(data)
def refresh_tokens(refresh_token: str) -> dict[str, Any]:
data = _post_form(
{
"grant_type": "refresh_token",
"client_id": CLIENT_ID,
"refresh_token": refresh_token,
}
)
return _record_from_token_response(data, refresh_fallback=refresh_token)
def _access_token(record: dict[str, Any]) -> str:
access = record["access"]
if not isinstance(access, str) or not access:
raise GrokAuthError("bad_response", "stored access token is missing or malformed")
return access
def _near_expiry(record: dict[str, Any]) -> bool:
expires_at = record.get("expires_at")
if not isinstance(expires_at, int | float):
return True
return expires_at - _EXPIRY_SKEW_S <= time.time()
def get_valid_token() -> str:
"""Return a valid access token, refreshing under the cross-process guard if
near expiry."""
record = read_record()
if record is None:
raise GrokAuthError("not_authenticated", "not signed in; run: strix auth login grok")
if not _near_expiry(record):
return _access_token(record)
with _refresh_guard():
record = read_record()
if record is None:
raise GrokAuthError("not_authenticated", "not signed in; run: strix auth login grok")
if not _near_expiry(record):
return _access_token(record)
try:
refreshed = refresh_tokens(record["refresh"])
except GrokAuthError:
# A peer process may have already spent this single-use refresh token.
latest = read_record()
if latest and latest["refresh"] != record["refresh"] and not _near_expiry(latest):
return _access_token(latest)
raise
save_record(refreshed)
return _access_token(refreshed)
def build_openai_client() -> AsyncOpenAI:
"""An ``AsyncOpenAI`` for xAI's API. A per-request hook re-stamps a fresh
bearer token so long scans survive token expiry."""
import asyncio
import httpx
from openai import AsyncOpenAI
get_valid_token() # fail fast at configure time if the sign-in is dead
async def _auth_hook(request: httpx.Request) -> None:
access = await asyncio.to_thread(get_valid_token)
request.headers["Authorization"] = f"Bearer {access}"
http_client = httpx.AsyncClient(
timeout=httpx.Timeout(600.0, connect=30.0),
event_hooks={"request": [_auth_hook]},
)
return AsyncOpenAI(
api_key="strix-grok-oauth", # placeholder; the hook overwrites Authorization
base_url=XAI_BASE_URL,
http_client=http_client,
)
_subscription_client: AsyncOpenAI | None = None
def get_subscription_client() -> AsyncOpenAI:
global _subscription_client # noqa: PLW0603
if _subscription_client is None:
_subscription_client = build_openai_client()
return _subscription_client
SUBSCRIPTION_PREFIX = "grok/"
def subscription_model(model_name: str | None) -> str | None:
"""The model slug behind a ``grok/<model>`` STRIX_LLM, or None."""
name = (model_name or "").strip()
if not name.lower().startswith(SUBSCRIPTION_PREFIX):
return None
return name[len(SUBSCRIPTION_PREFIX) :] or None
def auth_mode(model_name: str | None) -> str:
return "subscription" if subscription_model(model_name) else "api_key"
+2 -7
View File
@@ -20,7 +20,6 @@ from agents.model_settings import ModelSettings
from agents.models.fake_id import FAKE_RESPONSES_ID
from agents.models.interface import Model
from agents.models.multi_provider import MultiProvider
from agents.models.openai_chatcompletions import OpenAIChatCompletionsModel
from agents.models.openai_responses import OpenAIResponsesModel
from agents.retry import (
ModelRetryBackoffSettings,
@@ -37,7 +36,7 @@ from openai.types.responses import (
from openai.types.responses.response_usage import ResponseUsage
from openai.types.shared import Reasoning
from strix.config import codex, grok
from strix.config import codex
from strix.config.loader import load_settings
from strix.config.tool_call_ids import TurnCallIdRewriter, dedupe_input
from strix.config.tool_call_limits import TurnToolCallLimiter
@@ -482,10 +481,6 @@ class StrixProvider(MultiProvider):
codex.get_subscription_client(),
reasoning_effort=llm.reasoning_effort,
)
elif grok_slug := grok.subscription_model(model_name):
# xAI's API is OpenAI chat-completions compatible; the subscription
# bearer is stamped per-request by the client's auth hook.
model = OpenAIChatCompletionsModel(grok_slug, grok.get_subscription_client())
else:
model = super().get_model(model_name)
if llm.disable_streaming:
@@ -561,7 +556,7 @@ def configure_sdk_model_defaults(settings: Settings) -> None:
"""Apply Strix config to SDK-native defaults."""
llm = settings.llm
set_tracing_disabled(True)
if codex.subscription_model(llm.model) or grok.subscription_model(llm.model):
if codex.subscription_model(llm.model):
return
_configure_litellm_compatibility()
_configure_openrouter_attribution(llm.model)
-45
View File
@@ -1,45 +0,0 @@
"""Shared helpers across model-subscription providers (ChatGPT/Codex and Grok).
Each provider module (:mod:`strix.config.codex`, :mod:`strix.config.grok`)
exposes the same small surface — ``subscription_model``, ``auth_mode``,
``is_authenticated`` — so callers that only care "is this run on a subscription,
and which provider?" can stay provider-agnostic.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from strix.config import codex, grok
if TYPE_CHECKING:
from types import ModuleType
_PROVIDERS: tuple[ModuleType, ...] = (codex, grok)
# Human-facing provider names keyed by each module's ``PROVIDER`` constant.
_DISPLAY_NAMES: dict[str, str] = {codex.PROVIDER: "ChatGPT", grok.PROVIDER: "Grok"}
def provider_for_model(model_name: str | None) -> ModuleType | None:
"""Return the subscription provider module that owns ``model_name``'s prefix,
or None when the model isn't a subscription model."""
for provider in _PROVIDERS:
if provider.subscription_model(model_name):
return provider
return None
def auth_mode(model_name: str | None) -> str:
return "subscription" if provider_for_model(model_name) is not None else "api_key"
def provider_label(model_name: str | None) -> str | None:
"""Human-facing name of the subscription provider for ``model_name`` (e.g.
"ChatGPT" or "Grok"), or None when the model isn't a subscription model."""
provider = provider_for_model(model_name)
if provider is None:
return None
return _DISPLAY_NAMES.get(provider.PROVIDER)
-152
View File
@@ -1,152 +0,0 @@
"""Shared on-disk store for subscription OAuth credentials.
Every subscription provider (ChatGPT/Codex, Grok) keeps its record under its own
key in a single ``~/.strix/subscription-auth.json`` file. Reads and writes go
through here so that:
* tokens are written owner-only (mode 0600) from the moment the file is created,
never briefly exposed with umask-derived permissions, and
* concurrent read-modify-write mutations — even across different providers or
processes — are serialized, so one provider's update can't clobber another's.
The lock is reentrant, so a provider may nest a ``save`` inside a longer
``guard`` (e.g. refreshing a token then persisting it) without deadlocking.
"""
from __future__ import annotations
import contextlib
import json
import os
import tempfile
import threading
from pathlib import Path
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from collections.abc import Iterator
from io import TextIOWrapper
class StoreLockError(RuntimeError):
"""The cross-process store lock could not be acquired.
Raised instead of silently proceeding, so a read-modify-write never runs
unlocked (which would let concurrent provider logins/refreshes/logouts race).
"""
def read(path: Path) -> dict[str, Any]:
"""The store's contents, or an empty dict when absent/unreadable."""
try:
data = json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
return {}
return data if isinstance(data, dict) else {}
def write(path: Path, data: dict[str, Any]) -> None:
"""Atomically replace the store, owner-only from creation.
The temp file is created with a random name via ``mkstemp`` (mode 0600, no
symlink following), so a local attacker can't pre-plant a symlink at a
predictable path to divert the token write.
"""
path.parent.mkdir(parents=True, exist_ok=True)
fd, tmp_name = tempfile.mkstemp(dir=path.parent, prefix=path.name, suffix=".tmp")
tmp = Path(tmp_name)
try:
with os.fdopen(fd, "w", encoding="utf-8") as handle:
json.dump(data, handle, indent=2)
tmp.replace(path)
except BaseException:
with contextlib.suppress(OSError):
tmp.unlink()
raise
with contextlib.suppress(OSError):
path.chmod(0o600)
class _StoreLock:
"""A reentrant lock serializing store mutations within (thread lock) and
across (flock) Strix processes. Nesting reuses the single held file lock, so
a provider can persist a record inside a longer critical section."""
def __init__(self) -> None:
self._thread_lock = threading.RLock()
self._flock_handle: TextIOWrapper | None = None
self._depth = 0
@contextlib.contextmanager
def hold(self, path: Path) -> Iterator[None]:
with self._thread_lock:
if self._depth == 0:
self._flock_handle = _acquire_flock(path)
self._depth += 1
try:
yield
finally:
self._depth -= 1
if self._depth == 0:
self._release_flock()
def _release_flock(self) -> None:
handle = self._flock_handle
self._flock_handle = None
if handle is None:
return
try:
import fcntl
with contextlib.suppress(OSError):
fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
except ImportError:
pass
finally:
handle.close()
_store_lock = _StoreLock()
def guard(path: Path) -> contextlib.AbstractContextManager[None]:
"""Serialize store mutation across threads and processes (reentrant)."""
return _store_lock.hold(path)
def _acquire_flock(path: Path) -> TextIOWrapper:
"""Hold an exclusive cross-process lock on the store, or raise.
Never returns without the lock held: a missing ``fcntl`` or a failed
``flock`` raises :class:`StoreLockError` so the caller aborts rather than
mutating the store unlocked.
"""
try:
import fcntl
except ImportError as exc: # pragma: no cover - non-POSIX
msg = "cross-process credential locking requires fcntl (a POSIX platform)"
raise StoreLockError(msg) from exc
lock_path = path.with_suffix(".lock")
lock_path.parent.mkdir(parents=True, exist_ok=True)
# O_NOFOLLOW rejects a pre-positioned symlink at the predictable lock path
# (so an attacker can't redirect the open), and no O_TRUNC since the lock
# file is only an flock anchor whose contents we never use.
try:
fd = os.open(str(lock_path), os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600)
except OSError as exc:
msg = f"could not open lock file {lock_path}: {exc}"
raise StoreLockError(msg) from exc
handle = os.fdopen(fd, "r+")
try:
while True:
try:
fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
break
except InterruptedError: # EINTR — retry the blocking acquire
continue
except OSError as exc:
handle.close()
msg = f"could not lock {lock_path}: {exc}"
raise StoreLockError(msg) from exc
return handle
+69 -168
View File
@@ -1,8 +1,8 @@
"""`strix auth` — model-subscription sign-in (login / status / logout).
"""`strix auth` — ChatGPT subscription sign-in (login / status / logout).
Signing in only stores OAuth tokens (``~/.strix/subscription-auth.json``); model
selection stays with ``STRIX_LLM``. A ``chatgpt/<model>`` STRIX_LLM runs on a
ChatGPT subscription and a ``grok/<model>`` one on a Grok/SuperGrok subscription.
selection stays with ``STRIX_LLM``. A ``chatgpt/<model>`` STRIX_LLM runs on the
subscription.
"""
from __future__ import annotations
@@ -12,7 +12,6 @@ import base64
import logging
import threading
import webbrowser
from dataclasses import dataclass
from http.server import BaseHTTPRequestHandler, HTTPServer
from pathlib import Path
from typing import TYPE_CHECKING, Any
@@ -22,76 +21,24 @@ from rich.console import Console
from rich.panel import Panel
from rich.text import Text
from strix.config import codex, grok, load_settings, subscription_store
from strix.config import codex, load_settings
if TYPE_CHECKING:
from collections.abc import Callable
from types import ModuleType
logger = logging.getLogger(__name__)
_CALLBACK_TIMEOUT_S = 300
# CLI-facing name for the login provider. Internally this is the Codex OAuth
# flow (``codex.PROVIDER``), but users know it as ChatGPT, so that's what the
# command and messaging say. ``codex`` is accepted as an alias.
LOGIN_PROVIDER = "chatgpt"
_ACCEPTED_PROVIDERS = frozenset({LOGIN_PROVIDER, codex.PROVIDER})
@dataclass(frozen=True)
class _Provider:
"""A model-subscription provider the ``strix auth`` command can sign into.
``module`` is the provider's OAuth module (:mod:`strix.config.codex` or
:mod:`strix.config.grok`); both expose the same login surface. ``error`` is
that module's auth-error class, caught to report a clean failure.
"""
name: str
module: ModuleType
error: type[Exception]
display: str
example_model: str
blurb: str
_PROVIDERS: dict[str, _Provider] = {
"chatgpt": _Provider(
name="chatgpt",
module=codex,
error=codex.CodexAuthError,
display="ChatGPT",
example_model="chatgpt/gpt-5.4",
blurb="This uses your ChatGPT Plus/Pro plan for inference instead of a metered API key.",
),
"grok": _Provider(
name="grok",
module=grok,
error=grok.GrokAuthError,
display="Grok",
example_model="grok/grok-4",
blurb="This uses your Grok/SuperGrok plan for inference instead of a metered API key.",
),
}
# Internal OAuth provider ids and common vendor names accepted as aliases.
_PROVIDER_ALIASES: dict[str, str] = {
codex.PROVIDER: "chatgpt",
grok.PROVIDER: "grok",
"xai": "grok",
"supergrok": "grok",
}
_DEFAULT_PROVIDER = "chatgpt"
_USAGE = (
"Usage:\n"
" strix auth login [chatgpt|grok] [--manual]\n"
" strix auth status\n"
" strix auth logout [chatgpt|grok]"
)
def _resolve_provider(name: str) -> _Provider | None:
key = _PROVIDER_ALIASES.get(name.lower(), name.lower())
return _PROVIDERS.get(key)
_USAGE = "Usage:\n strix auth login chatgpt [--manual]\n strix auth status\n strix auth logout"
def run_auth(argv: list[str]) -> int:
@@ -102,20 +49,20 @@ def run_auth(argv: list[str]) -> int:
rest = argv[1:]
if subcommand in ("-h", "--help", "help"):
console.print(_USAGE, markup=False)
console.print(_USAGE)
return 0
handlers: dict[str, Callable[[], int]] = {
"login": lambda: _login(console, rest),
"status": lambda: _status(console),
"logout": lambda: _logout(console, rest),
"logout": lambda: _logout(console),
}
handler = handlers.get(subcommand)
if handler is not None:
return handler()
console.print(f"[red]Unknown auth command:[/] {subcommand}\n")
console.print(_USAGE, markup=False)
console.print(_USAGE)
return 2
@@ -124,8 +71,8 @@ def _login(console: Console, argv: list[str]) -> int:
parser.add_argument(
"provider",
nargs="?",
default=_DEFAULT_PROVIDER,
help="Model provider to sign in with (chatgpt or grok; default: chatgpt).",
default=LOGIN_PROVIDER,
help="Model provider to sign in with (default: chatgpt).",
)
parser.add_argument(
"--manual",
@@ -137,42 +84,39 @@ def _login(console: Console, argv: list[str]) -> int:
except SystemExit as exc: # argparse already printed the message
return int(exc.code or 2)
provider = _resolve_provider(args.provider)
if provider is None:
supported = ", ".join(f"'{name}'" for name in _PROVIDERS)
console.print(f"[red]Unsupported provider:[/] {args.provider}. Supported: {supported}.")
if args.provider.lower() not in _ACCEPTED_PROVIDERS:
console.print(
f"[red]Unsupported provider:[/] {args.provider}. "
f"Only '{LOGIN_PROVIDER}' (ChatGPT subscription) is supported."
)
return 2
module = provider.module
verifier, challenge = module.generate_pkce()
state = module.create_state()
authorize_url = module.build_authorize_url(challenge, state)
verifier, challenge = codex.generate_pkce()
state = codex.create_state()
authorize_url = codex.build_authorize_url(challenge, state)
console.print()
console.print("[bold]Signing in with ChatGPT[/] [dim](provider: chatgpt)[/]")
console.print(
f"[bold]Signing in with {provider.display}[/] [dim](provider: {provider.name})[/]"
"[dim]This uses your ChatGPT Plus/Pro plan for inference instead of a metered API key.[/]"
)
console.print(f"[dim]{provider.blurb}[/]")
console.print()
try:
record = _run_oauth_flow(
console, provider, authorize_url, verifier, state, manual=args.manual
)
except provider.error as exc:
record = _run_oauth_flow(console, authorize_url, verifier, state, manual=args.manual)
except codex.CodexAuthError as exc:
return _fail(console, exc)
except KeyboardInterrupt:
console.print("\n[yellow]Sign-in cancelled.[/]")
return 130
module.save_record(record)
_print_success(console, provider)
codex.save_record(record)
_print_success(console)
return 0
def _run_oauth_flow(
console: Console,
provider: _Provider,
authorize_url: str,
verifier: str,
state: str,
@@ -180,10 +124,7 @@ def _run_oauth_flow(
manual: bool,
) -> dict[str, Any]:
"""Drive the browser (or manual) OAuth flow and return a token record."""
module = provider.module
server = (
None if manual else _try_start_callback_server(module.CALLBACK_PORT, module.CALLBACK_PATH)
)
server = None if manual else _try_start_callback_server()
console.print("Open this URL in your browser to authorize:")
console.print(f"[cyan]{authorize_url}[/]")
@@ -201,8 +142,8 @@ def _run_oauth_flow(
if result is not None:
code, returned_state, error = result
if error:
raise provider.error("oauth_error", error)
return _finish(provider, code, returned_state, verifier, state, require_state=True)
raise codex.CodexAuthError("oauth_error", error)
return _finish(code, returned_state, verifier, state, require_state=True)
console.print("[yellow]Timed out waiting for the browser. Falling back to manual paste.[/]")
# Manual fallback: the user completes sign-in and pastes the redirect URL
@@ -212,13 +153,12 @@ def _run_oauth_flow(
try:
pasted = console.input("Paste the full redirect URL (or code#state): ").strip()
except EOFError as exc:
raise provider.error("no_input", "no redirect URL provided") from exc
code, returned_state = module.parse_redirect_input(pasted)
return _finish(provider, code, returned_state, verifier, state, require_state=False)
raise codex.CodexAuthError("no_input", "no redirect URL provided") from exc
code, returned_state = codex.parse_redirect_input(pasted)
return _finish(code, returned_state, verifier, state, require_state=False)
def _finish(
provider: _Provider,
code: str | None,
returned_state: str | None,
verifier: str,
@@ -227,17 +167,16 @@ def _finish(
require_state: bool,
) -> dict[str, Any]:
if not code:
raise provider.error("no_code", "no authorization code found in the redirect")
# The loopback callback from the provider always carries state, so a missing
# or mismatched value there is forged (CSRF) and must be rejected. Manual
# paste is user-initiated (the user copies their own redirect), so state is
# only validated when the pasted value includes it.
raise codex.CodexAuthError("no_code", "no authorization code found in the redirect")
# The loopback callback from OpenAI always carries state, so a missing or
# mismatched value there is forged (CSRF) and must be rejected. Manual paste
# is user-initiated (the user copies their own redirect), so state is only
# validated when the pasted value includes it.
if require_state and returned_state is None:
raise provider.error("state_mismatch", "missing state in callback; possible CSRF")
raise codex.CodexAuthError("state_mismatch", "missing state in callback; possible CSRF")
if returned_state is not None and returned_state != expected_state:
raise provider.error("state_mismatch", "state did not match; possible CSRF")
record: dict[str, Any] = provider.module.exchange_code(code, verifier)
return record
raise codex.CodexAuthError("state_mismatch", "state did not match; possible CSRF")
return codex.exchange_code(code, verifier)
class _CallbackServer:
@@ -264,7 +203,7 @@ class _CallbackServer:
self._httpd.server_close()
def _try_start_callback_server(port: int, path: str) -> _CallbackServer | None:
def _try_start_callback_server() -> _CallbackServer | None:
event = threading.Event()
holder: dict[str, Any] = {}
@@ -274,7 +213,7 @@ def _try_start_callback_server(port: int, path: str) -> _CallbackServer | None:
def do_GET(self) -> None:
parsed = urlparse(self.path)
if parsed.path != path:
if parsed.path != codex.CALLBACK_PATH:
self.send_response(404)
self.end_headers()
return
@@ -291,9 +230,9 @@ def _try_start_callback_server(port: int, path: str) -> _CallbackServer | None:
event.set()
try:
httpd = HTTPServer(("127.0.0.1", port), Handler)
httpd = HTTPServer(("127.0.0.1", codex.CALLBACK_PORT), Handler)
except OSError:
logger.debug("could not bind callback port %d", port, exc_info=True)
logger.debug("could not bind callback port %d", codex.CALLBACK_PORT, exc_info=True)
return None
return _CallbackServer(httpd, event, holder)
@@ -304,67 +243,30 @@ def _first(query: dict[str, list[str]], key: str) -> str | None:
def _status(console: Console) -> int:
settings = load_settings()
active_model = settings.llm.model
signed_in_any = False
for provider in _PROVIDERS.values():
record = provider.module.read_record()
if record is None:
continue
signed_in_any = True
console.print(f"[green]Signed in[/] with a {provider.display} subscription.")
account_id = record.get("account_id")
if account_id:
console.print(f" Account: [bold]{account_id}[/]")
if provider.module.subscription_model(active_model):
console.print(f" Runs use the subscription (STRIX_LLM=[bold]{active_model}[/]).")
else:
console.print(
f" [yellow]Note:[/] set [cyan]STRIX_LLM[/] to e.g. "
f"[cyan]{provider.example_model}[/] to run on this subscription."
)
if not signed_in_any:
console.print(
"[yellow]Not signed in.[/] Run [cyan]strix auth login chatgpt[/] "
"or [cyan]strix auth login grok[/] to sign in."
)
record = codex.read_record()
if record is None:
console.print("[yellow]Not signed in.[/] Run [cyan]strix auth login chatgpt[/] to sign in.")
return 1
settings = load_settings()
console.print("[green]Signed in[/] with a ChatGPT subscription.")
console.print(f" Account: [bold]{record.get('account_id')}[/]")
if codex.subscription_model(settings.llm.model):
console.print(f" Runs use the subscription (STRIX_LLM=[bold]{settings.llm.model}[/]).")
else:
console.print(
" [yellow]Note:[/] set [cyan]STRIX_LLM[/] to e.g. [cyan]chatgpt/gpt-5.4[/] "
"to run on the subscription."
)
return 0
def _logout(console: Console, argv: list[str]) -> int:
parser = argparse.ArgumentParser(prog="strix auth logout", add_help=True)
parser.add_argument(
"provider",
nargs="?",
default=None,
help="Provider to sign out of (chatgpt or grok; default: all).",
)
try:
args = parser.parse_args(argv)
except SystemExit as exc:
return int(exc.code or 2)
if args.provider is None:
# Hold the store lock across every provider so a concurrent save/refresh
# can't slip a credential back in between removals (logout-all is atomic).
with subscription_store.guard(codex.AUTH_PATH):
for provider in _PROVIDERS.values():
provider.module.logout()
console.print("[green]Signed out.[/] Stored subscription credentials removed.")
return 0
target = _resolve_provider(args.provider)
if target is None:
supported = ", ".join(f"'{name}'" for name in _PROVIDERS)
console.print(f"[red]Unsupported provider:[/] {args.provider}. Supported: {supported}.")
return 2
target.module.logout()
console.print(f"[green]Signed out of {target.display}.[/] Stored credentials removed.")
def _logout(console: Console) -> int:
codex.logout()
console.print("[green]Signed out.[/] Stored subscription credentials removed.")
return 0
def _fail(console: Console, exc: Exception) -> int:
def _fail(console: Console, exc: codex.CodexAuthError) -> int:
error_text = Text()
error_text.append("SIGN-IN FAILED", style="bold red")
error_text.append("\n\n", style="white")
@@ -382,18 +284,17 @@ def _fail(console: Console, exc: Exception) -> int:
return 1
def _print_success(console: Console, provider: _Provider) -> None:
prefix = provider.module.SUBSCRIPTION_PREFIX
def _print_success(console: Console) -> None:
text = Text()
text.append(f"Signed in with your {provider.display} subscription", style="bold #22c55e")
text.append("Signed in with your ChatGPT subscription", style="bold #22c55e")
text.append("\n\n", style="white")
text.append("Set ", style="white")
text.append("STRIX_LLM", style="bold white")
text.append(" to a ", style="white")
text.append(prefix, style="bold cyan")
text.append("chatgpt/", style="bold cyan")
text.append(" model (e.g. ", style="white")
text.append(provider.example_model, style="bold cyan")
text.append(f") — runs are billed to your {provider.display} plan.", style="white")
text.append("chatgpt/gpt-5.4", style="bold cyan")
text.append(") — runs are billed to your ChatGPT plan.", style="white")
text.append("\n\n", style="white")
text.append("Run a scan as usual, e.g. ", style="white")
text.append("strix --target https://example.com", style="bold cyan")
+1 -11
View File
@@ -8,7 +8,7 @@ from rich.console import Console
from rich.panel import Panel
from rich.text import Text
from strix.config import codex, grok, load_settings
from strix.config import codex, load_settings
from strix.interface.utils import (
check_docker_connection,
image_exists,
@@ -37,16 +37,6 @@ def validate_environment() -> None:
logger.info("Environment OK (ChatGPT subscription)")
return
if grok.subscription_model(settings.llm.model):
if not grok.is_authenticated():
console.print(
f"[red]STRIX_LLM={settings.llm.model} uses your Grok subscription, "
"but you're not signed in.[/] Run [cyan]strix auth login grok[/] first."
)
sys.exit(1)
logger.info("Environment OK (Grok subscription)")
return
if not settings.llm.model:
missing_required_vars.append("STRIX_LLM")
+1 -10
View File
@@ -436,16 +436,7 @@ def main() -> None:
start_background_check()
if not args.non_interactive and prompt_update_if_available(Console()):
if is_binary_install() and sys.platform != "win32":
# The PyInstaller onefile bootloader passes its state to the child
# process via environment variables; if they leak into the re-exec,
# the new binary reuses the old extracted application instead of
# unpacking itself, so the pre-update version runs again.
env = {
key: value
for key, value in os.environ.items()
if not key.startswith("_PYI_") and key != "_MEIPASS2"
}
os.execve(sys.executable, sys.argv, env) # noqa: S606 # nosec B606
os.execv(sys.executable, sys.argv) # noqa: S606 # nosec B606
sys.exit(0)
check_docker_installed()
+3 -5
View File
@@ -14,7 +14,7 @@ import logging
from datetime import UTC, datetime
from typing import TYPE_CHECKING, Any
from strix.config import Settings, load_settings, subscription
from strix.config import Settings, codex, load_settings
from strix.core.paths import run_dir_for
from strix.interface.utils import (
assign_workspace_subdirs,
@@ -226,7 +226,7 @@ def telemetry_start(args: argparse.Namespace) -> None:
model = load_settings().llm.model
kwargs = {
"model": model,
"auth_mode": subscription.auth_mode(model),
"auth_mode": codex.auth_mode(model),
"scan_mode": args.scan_mode,
"is_whitebox": is_whitebox_scan(args.targets_info),
"interactive": not args.non_interactive,
@@ -241,15 +241,13 @@ def _persist_run_record(args: argparse.Namespace) -> None:
run_dir = run_dir_for(args.run_name)
run_dir.mkdir(parents=True, exist_ok=True)
model = load_settings().llm.model
run_record = {
"run_id": args.run_name,
"run_name": args.run_name,
"status": "running",
"start_time": datetime.now(UTC).isoformat(),
"end_time": None,
"auth_mode": subscription.auth_mode(model),
"subscription_provider": subscription.provider_label(model),
"auth_mode": codex.auth_mode(load_settings().llm.model),
"targets_info": args.targets_info,
"scan_mode": args.scan_mode,
"instruction": args.instruction,
+1 -5
View File
@@ -24,7 +24,7 @@ from strix.interface.tui.backend.projection import (
sanitize_terminal_text,
terminal_projection,
)
from strix.interface.utils import is_subscription_run, subscription_label
from strix.interface.utils import is_subscription_run
if TYPE_CHECKING:
@@ -162,11 +162,8 @@ class TuiController:
if self.report_state is not None:
usage = dict(self.report_state.get_total_llm_usage())
subscription = False
subscription_name = ""
with contextlib.suppress(Exception):
subscription = is_subscription_run(self.report_state)
if subscription:
subscription_name = subscription_label(self.report_state)
model_warning = ""
if model and not is_recommended_or_frontier_model(model):
model_warning = (
@@ -203,7 +200,6 @@ class TuiController:
],
"usage": terminal_projection(usage, max_string=256, max_items=20),
"subscription": subscription,
"subscription_label": terminal_projection(subscription_name, max_string=64),
"viewer_status": self.viewer_status,
"viewer_url": terminal_projection(self.viewer_url, max_string=1024),
"error": terminal_projection(self.error, max_string=2 * 1024),
@@ -175,7 +175,6 @@ def bounded_state_projection(state: dict[str, Any]) -> dict[str, Any]:
"messages": [],
"usage": {},
"subscription": state["subscription"],
"subscription_label": state["subscription_label"],
"viewer_status": state["viewer_status"],
"viewer_url": None,
"error": terminal_projection(state["error"], max_string=256),
+2 -13
View File
@@ -1067,12 +1067,11 @@ func TestBudgetPauseShowsOneWarningToastUntilResumed(t *testing.T) {
func TestStatsViewShowsSubscription(t *testing.T) {
model := New(nil)
model.snapshot.Model = "grok/grok-4"
model.snapshot.Model = "gpt-5"
model.snapshot.Subscription = true
model.snapshot.SubscriptionLabel = "Grok subscription"
model.snapshot.Usage = map[string]any{"total_tokens": float64(1200), "cost": 3.5}
stats := ansi.Strip(model.statsView())
if !strings.Contains(stats, "Grok subscription") {
if !strings.Contains(stats, "ChatGPT subscription") {
t.Fatalf("stats missing subscription line: %q", stats)
}
if strings.Contains(stats, "$") {
@@ -1080,16 +1079,6 @@ func TestStatsViewShowsSubscription(t *testing.T) {
}
}
func TestStatsViewSubscriptionFallsBackWithoutLabel(t *testing.T) {
model := New(nil)
model.snapshot.Model = "gpt-5"
model.snapshot.Subscription = true
stats := ansi.Strip(model.statsView())
if !strings.Contains(stats, "Subscription") {
t.Fatalf("stats missing generic subscription line: %q", stats)
}
}
func TestVulnerabilityMarkdownReport(t *testing.T) {
report := vulnerabilityMarkdownReport(map[string]any{
"title": "SQLi in login",
+1 -5
View File
@@ -596,11 +596,7 @@ func (m Model) statsView() string {
if b.Len() > 0 {
b.WriteString("\n")
}
label := m.snapshot.SubscriptionLabel
if label == "" {
label = "Subscription"
}
b.WriteString(lipgloss.NewStyle().Foreground(green).Render(label))
b.WriteString(lipgloss.NewStyle().Foreground(green).Render("ChatGPT subscription"))
}
total := numberValue(m.snapshot.Usage["total_tokens"])
if total > 0 {
@@ -68,7 +68,6 @@ type Snapshot struct {
Vulnerabilities []map[string]any `json:"-"`
Usage map[string]any `json:"usage"`
Subscription bool `json:"subscription"`
SubscriptionLabel string `json:"subscription_label"`
ViewerStatus string `json:"viewer_status"`
ViewerURL *string `json:"viewer_url"`
Error *string `json:"error"`
+4 -22
View File
@@ -262,27 +262,9 @@ def is_subscription_run(report_state: Any) -> bool:
record = getattr(report_state, "run_record", None)
if isinstance(record, dict) and record.get("auth_mode"):
return record.get("auth_mode") == "subscription"
from strix.config import subscription
from strix.config import codex
return subscription.auth_mode(load_settings().llm.model) == "subscription"
def subscription_label(report_state: Any) -> str:
"""Human label for the active model subscription (e.g. "Grok subscription").
Prefers the persisted run record so a resumed run keeps its original provider
even if STRIX_LLM later points at a different one; falls back to current
settings.
"""
record = getattr(report_state, "run_record", None)
if isinstance(record, dict):
provider = record.get("subscription_provider")
if isinstance(provider, str) and provider:
return f"{provider} subscription"
from strix.config import subscription
label = subscription.provider_label(load_settings().llm.model)
return f"{label} subscription" if label else "Subscription"
return codex.auth_mode(load_settings().llm.model) == "subscription"
def _int_stat(usage: dict[str, Any], key: str) -> int:
@@ -386,7 +368,7 @@ def build_live_stats_text(report_state: Any) -> Text:
stats_text.append(str(model), style="white")
if is_subscription_run(report_state):
stats_text.append(" · ", style="dim white")
stats_text.append(subscription_label(report_state), style="#22c55e")
stats_text.append("ChatGPT subscription", style="#22c55e")
stats_text.append("\n")
vuln_count = len(report_state.vulnerability_reports)
@@ -432,7 +414,7 @@ def build_tui_stats_text(report_state: Any) -> Text:
subscription = is_subscription_run(report_state)
if subscription:
stats_text.append("\n")
stats_text.append(subscription_label(report_state), style="#22c55e")
stats_text.append("ChatGPT subscription", style="#22c55e")
usage = _llm_usage(report_state)
if usage and _int_stat(usage, "total_tokens") > 0:
@@ -101,7 +101,6 @@ export function RunDetails({
const totalTokens = num(usage.total_tokens);
const cost = num(usage.cost);
const subscription = str(raw.auth_mode) === "subscription";
const subscriptionProvider = str(raw.subscription_provider);
const sub = (n: number, word: string) => (
<span className="text-[#666]"> ({formatNumber(n)} {word})</span>
@@ -181,7 +180,7 @@ export function RunDetails({
<Field label="Provider">
<span className="inline-flex items-center gap-1.5">
<span className="rounded-full border border-[#22c55e]/40 bg-[#22c55e]/10 px-2 py-0.5 text-[11px] text-[#22c55e]">
{subscriptionProvider ? `${subscriptionProvider} subscription` : "Subscription"}
ChatGPT subscription
</span>
</span>
</Field>
File diff suppressed because one or more lines are too long
+1 -1
View File
@@ -6,7 +6,7 @@
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<meta name="color-scheme" content="dark" />
<title>Strix Results</title>
<script type="module" crossorigin src="./assets/index-XDX3roAH.js"></script>
<script type="module" crossorigin src="./assets/index-DBJ-RJqo.js"></script>
<link rel="stylesheet" crossorigin href="./assets/index-DKbLYAbP.css">
</head>
<body>
+1 -34
View File
@@ -6,7 +6,6 @@ import json
import logging
from typing import TYPE_CHECKING, Any
from strix.config import subscription
from strix.core.paths import run_record_path
from strix.interface.tui.live_view import TuiLiveView
@@ -58,39 +57,7 @@ def read_run_summary(run_dir: Path) -> dict[str, Any]:
record = {}
status = record.get("status")
finished = status in _TERMINAL_STATUSES and bool(record.get("end_time"))
summary = {**record, "finished": finished}
_backfill_subscription_provider(summary)
return summary
def _first_recorded_model(record: dict[str, Any]) -> str | None:
"""The first non-empty per-agent model slug in a run record, or None."""
usage = record.get("llm_usage")
if not isinstance(usage, dict):
return None
agents = usage.get("agents")
if not isinstance(agents, list):
return None
for agent in agents:
if isinstance(agent, dict):
model = agent.get("model")
if isinstance(model, str) and model:
return model
return None
def _backfill_subscription_provider(record: dict[str, Any]) -> None:
"""Name the subscription provider for runs recorded before that field
existed, deriving it from the recorded ``provider/model`` slug so the viewer
labels them correctly without a rescan. Newer runs already carry the field.
"""
if record.get("subscription_provider"):
return
if record.get("auth_mode") != "subscription":
return
label = subscription.provider_label(_first_recorded_model(record))
if label:
record["subscription_provider"] = label
return {**record, "finished": finished}
def primary_target(record: dict[str, Any]) -> str | None:
+2 -4
View File
@@ -11,7 +11,7 @@ from uuid import uuid4
from agents.usage import Usage
from strix.config import subscription
from strix.config import codex
from strix.config.loader import load_settings
from strix.core.paths import run_dir_for
from strix.report.pricing import resolve_litellm_model
@@ -123,8 +123,7 @@ class ReportState:
self.scan_results: dict[str, Any] | None = None
self.scan_config: dict[str, Any] | None = None
self._llm_usage = LLMUsageLedger()
model = load_settings().llm.model
auth_mode = subscription.auth_mode(model)
auth_mode = codex.auth_mode(load_settings().llm.model)
self._llm_usage.zero_cost = auth_mode == "subscription"
self.run_record: dict[str, Any] = {
"run_id": self.run_id,
@@ -133,7 +132,6 @@ class ReportState:
"end_time": None,
"status": "running",
"auth_mode": auth_mode,
"subscription_provider": subscription.provider_label(model),
"targets_info": [],
"llm_usage": self._build_llm_usage_record(),
}
+16 -52
View File
@@ -6,34 +6,23 @@ from typing import TYPE_CHECKING, Any
import pytest
from strix.config import codex, grok
from strix.config import codex
from strix.interface import auth_cli
if TYPE_CHECKING:
from pathlib import Path
_CHATGPT = auth_cli._PROVIDERS["chatgpt"]
@pytest.fixture(autouse=True)
def _tmp_store(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
store = tmp_path / "home" / ".strix" / "subscription-auth.json"
monkeypatch.setattr(codex, "AUTH_PATH", store)
monkeypatch.setattr(grok, "AUTH_PATH", store)
monkeypatch.setattr(codex, "AUTH_PATH", tmp_path / "home" / ".strix" / "subscription-auth.json")
def test_default_provider_is_chatgpt() -> None:
assert auth_cli._DEFAULT_PROVIDER == "chatgpt"
assert set(auth_cli._PROVIDERS) == {"chatgpt", "grok"}
def test_provider_aliases_resolve() -> None:
assert auth_cli._resolve_provider(codex.PROVIDER) is _CHATGPT
assert auth_cli._resolve_provider("ChatGPT") is _CHATGPT
assert auth_cli._resolve_provider("grok") is auth_cli._PROVIDERS["grok"]
assert auth_cli._resolve_provider("xai") is auth_cli._PROVIDERS["grok"]
assert auth_cli._resolve_provider("gemini") is None
def test_login_provider_is_chatgpt() -> None:
assert auth_cli.LOGIN_PROVIDER == "chatgpt"
assert codex.PROVIDER in auth_cli._ACCEPTED_PROVIDERS
assert "chatgpt" in auth_cli._ACCEPTED_PROVIDERS
def test_unknown_subcommand_returns_usage_error() -> None:
@@ -62,32 +51,32 @@ def test_finish_requires_state_on_loopback(monkeypatch: pytest.MonkeyPatch) -> N
# Loopback (require_state=True): missing or mismatched state is rejected.
with pytest.raises(codex.CodexAuthError) as missing:
auth_cli._finish(_CHATGPT, "code", None, "verifier", "expected", require_state=True)
auth_cli._finish("code", None, "verifier", "expected", require_state=True)
assert missing.value.code == "state_mismatch"
with pytest.raises(codex.CodexAuthError) as mismatch:
auth_cli._finish(_CHATGPT, "code", "wrong", "verifier", "expected", require_state=True)
auth_cli._finish("code", "wrong", "verifier", "expected", require_state=True)
assert mismatch.value.code == "state_mismatch"
# Matching state proceeds to the exchange.
assert auth_cli._finish(
_CHATGPT, "code", "expected", "verifier", "expected", require_state=True
) == {"ok": True}
assert auth_cli._finish("code", "expected", "verifier", "expected", require_state=True) == {
"ok": True
}
def test_finish_manual_paste_allows_absent_state(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(codex, "exchange_code", lambda *_: {"ok": True})
# Manual paste (require_state=False): a bare code with no state is accepted,
# but a present-and-wrong state is still rejected.
assert auth_cli._finish(
_CHATGPT, "code", None, "verifier", "expected", require_state=False
) == {"ok": True}
assert auth_cli._finish("code", None, "verifier", "expected", require_state=False) == {
"ok": True
}
with pytest.raises(codex.CodexAuthError):
auth_cli._finish(_CHATGPT, "code", "wrong", "verifier", "expected", require_state=False)
auth_cli._finish("code", "wrong", "verifier", "expected", require_state=False)
def test_finish_rejects_missing_code() -> None:
with pytest.raises(codex.CodexAuthError) as exc:
auth_cli._finish(_CHATGPT, None, "expected", "verifier", "expected", require_state=True)
auth_cli._finish(None, "expected", "verifier", "expected", require_state=True)
assert exc.value.code == "no_code"
@@ -95,31 +84,6 @@ def test_model_subcommand_removed() -> None:
assert auth_cli.run_auth(["model", "gpt-5.5"]) == 2
def _sign_in_both() -> None:
codex.save_record({"type": "oauth", "access": "c", "refresh": "r", "account_id": "a"})
grok.save_record({"type": "oauth", "access": "g", "refresh": "r"})
def test_logout_all_removes_every_provider() -> None:
_sign_in_both()
assert codex.is_authenticated()
assert grok.is_authenticated()
assert auth_cli.run_auth(["logout"]) == 0
assert not codex.is_authenticated()
assert not grok.is_authenticated()
def test_logout_single_provider_leaves_the_other() -> None:
_sign_in_both()
assert auth_cli.run_auth(["logout", "grok"]) == 0
assert codex.is_authenticated()
assert not grok.is_authenticated()
@pytest.mark.parametrize("provider", ["chatgpt", "codex", "ChatGPT"])
def test_login_accepts_provider_aliases(provider: str, monkeypatch: pytest.MonkeyPatch) -> None:
reached = {"flow": False}
-265
View File
@@ -1,265 +0,0 @@
"""Tests for Grok (xAI) subscription auth: PKCE, token handling, store."""
from __future__ import annotations
import base64
import hashlib
import json
import time
from typing import TYPE_CHECKING, Any
from unittest import mock
import pytest
import requests
from strix.config import grok
if TYPE_CHECKING:
from pathlib import Path
@pytest.fixture(autouse=True)
def _tmp_store(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
path = tmp_path / "home" / ".strix" / "subscription-auth.json"
monkeypatch.setattr(grok, "AUTH_PATH", path)
return path
def test_pkce_challenge_matches_verifier_and_is_unpadded() -> None:
verifier, challenge = grok.generate_pkce()
expected = (
base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).rstrip(b"=").decode()
)
assert challenge == expected
assert "=" not in verifier
assert "=" not in challenge
def test_authorize_url_carries_pkce_client_and_grok_scope() -> None:
url = grok.build_authorize_url("chal", "st8")
assert grok.AUTHORIZE_URL in url
assert "code_challenge=chal" in url
assert "code_challenge_method=S256" in url
assert f"client_id={grok.CLIENT_ID}" in url
assert "state=st8" in url
# The Grok-CLI scope is what unlocks subscription inference.
assert "grok-cli%3Aaccess" in url
assert "offline_access" in url
def test_redirect_uri_is_loopback() -> None:
assert grok.REDIRECT_URI == "http://127.0.0.1:56121/callback"
def test_post_form_returns_parsed_body() -> None:
resp = mock.MagicMock()
resp.status_code = 200
resp.content = b'{"access_token": "tok"}'
with mock.patch.object(requests, "post", return_value=resp) as post:
data = grok._post_form({"grant_type": "refresh_token"})
assert data == {"access_token": "tok"}
assert post.call_args.kwargs["timeout"] == grok._TOKEN_TIMEOUT
def test_post_form_raises_on_http_error() -> None:
resp = mock.MagicMock()
resp.status_code = 400
resp.text = "invalid_grant"
with (
mock.patch.object(requests, "post", return_value=resp),
pytest.raises(grok.GrokAuthError) as exc,
):
grok._post_form({"grant_type": "refresh_token"})
assert exc.value.code == "token_http_error"
@pytest.mark.parametrize(
("value", "expected"),
[
("http://127.0.0.1:56121/callback?code=AAA&state=BBB", ("AAA", "BBB")),
("AAA#BBB", ("AAA", "BBB")),
("code=AAA&state=BBB", ("AAA", "BBB")),
("AAA", ("AAA", None)),
("", (None, None)),
],
)
def test_parse_redirect_input(value: str, expected: tuple[str | None, str | None]) -> None:
assert grok.parse_redirect_input(value) == expected
@pytest.mark.parametrize(
("model", "expected"),
[
("grok/grok-4", "grok-4"),
("Grok/Grok-4", "Grok-4"),
(" grok/grok-4 ", "grok-4"),
("xai/grok-4", None), # metered API path
("chatgpt/gpt-5.4", None),
("grok-4", None),
("grok/", None),
("", None),
(None, None),
],
)
def test_subscription_model(model: str | None, expected: str | None) -> None:
assert grok.subscription_model(model) == expected
def test_auth_mode() -> None:
assert grok.auth_mode("grok/grok-4") == "subscription"
assert grok.auth_mode("xai/grok-4") == "api_key"
assert grok.auth_mode("chatgpt/gpt-5.4") == "api_key"
assert grok.auth_mode(None) == "api_key"
def _record(access: str, refresh: str, expires_at: float) -> dict[str, Any]:
return {
"type": "oauth",
"provider": "grok",
"access": access,
"refresh": refresh,
"expires_at": expires_at,
}
def test_store_roundtrip_and_logout() -> None:
assert grok.read_record() is None
assert grok.is_authenticated() is False
grok.save_record(_record("a1", "r1", time.time() + 3600))
record = grok.read_record()
assert record is not None
assert record["access"] == "a1"
assert grok.is_authenticated() is True
grok.logout()
assert grok.read_record() is None
grok.logout() # no-op when already gone
def test_store_file_permissions_are_owner_only(_tmp_store: Path) -> None:
grok.save_record(_record("a1", "r1", time.time() + 3600))
assert (_tmp_store.stat().st_mode & 0o777) == 0o600
def test_store_shares_file_with_other_providers(_tmp_store: Path) -> None:
# Grok must not clobber a co-resident ChatGPT record in the shared store.
_tmp_store.parent.mkdir(parents=True, exist_ok=True)
_tmp_store.write_text(json.dumps({"codex": {"type": "oauth", "access": "x"}}))
grok.save_record(_record("a1", "r1", time.time() + 3600))
on_disk = json.loads(_tmp_store.read_text())
assert on_disk["codex"] == {"type": "oauth", "access": "x"}
assert on_disk["grok"]["access"] == "a1"
grok.logout()
# Removing grok leaves the other provider's record and the file intact.
assert json.loads(_tmp_store.read_text()) == {"codex": {"type": "oauth", "access": "x"}}
def test_read_record_rejects_incomplete_records() -> None:
grok.save_record({"type": "oauth", "access": "a"}) # missing refresh
assert grok.read_record() is None
assert grok.is_authenticated() is False
def test_get_valid_token_returns_stored_when_fresh(monkeypatch: pytest.MonkeyPatch) -> None:
def _boom(_payload: dict[str, str]) -> dict[str, Any]:
msg = "should not refresh a fresh token"
raise AssertionError(msg)
monkeypatch.setattr(grok, "_post_form", _boom)
grok.save_record(_record("access-fresh", "r1", time.time() + 3600))
assert grok.get_valid_token() == "access-fresh"
def test_get_valid_token_refreshes_and_persists_rotation(monkeypatch: pytest.MonkeyPatch) -> None:
calls = {"n": 0}
def _fake_post(payload: dict[str, str]) -> dict[str, Any]:
calls["n"] += 1
assert payload["grant_type"] == "refresh_token"
assert payload["refresh_token"] == "r1"
return {"access_token": "access-new", "refresh_token": "r2", "expires_in": 3600}
monkeypatch.setattr(grok, "_post_form", _fake_post)
grok.save_record(_record("stale", "r1", time.time() - 10)) # already expired
assert grok.get_valid_token() == "access-new"
assert calls["n"] == 1
record = grok.read_record()
assert record is not None
assert record["refresh"] == "r2" # rotated refresh written back
def test_refresh_keeps_old_refresh_when_response_omits_it(monkeypatch: pytest.MonkeyPatch) -> None:
def _fake_post(_payload: dict[str, str]) -> dict[str, Any]:
return {"access_token": "access-new", "expires_in": 3600} # no refresh_token
monkeypatch.setattr(grok, "_post_form", _fake_post)
grok.save_record(_record("stale", "r1", time.time() - 10))
assert grok.get_valid_token() == "access-new"
record = grok.read_record()
assert record is not None
assert record["refresh"] == "r1" # fell back to the prior refresh token
def test_get_valid_token_uses_token_rotated_by_another_process(
monkeypatch: pytest.MonkeyPatch,
) -> None:
records = [
_record("stale", "r1", time.time() - 10),
_record("fresh-from-other-process", "r2", time.time() + 3600),
]
calls = {"n": 0}
def _fake_read() -> dict[str, Any]:
record = records[min(calls["n"], len(records) - 1)]
calls["n"] += 1
return record
def _boom(_payload: dict[str, str]) -> dict[str, Any]:
msg = "must not refresh a token another process already rotated"
raise AssertionError(msg)
monkeypatch.setattr(grok, "read_record", _fake_read)
monkeypatch.setattr(grok, "_post_form", _boom)
assert grok.get_valid_token() == "fresh-from-other-process"
def test_get_valid_token_recovers_when_refresh_loses_race(
monkeypatch: pytest.MonkeyPatch,
) -> None:
grok.save_record(_record("stale", "r1", time.time() - 10))
def _fake_post(_payload: dict[str, str]) -> dict[str, Any]:
grok.save_record(_record("fresh-from-peer", "r2", time.time() + 3600))
raise grok.GrokAuthError("token_http_error", "HTTP 400: invalid_grant")
monkeypatch.setattr(grok, "_post_form", _fake_post)
assert grok.get_valid_token() == "fresh-from-peer"
def test_get_valid_token_reraises_refresh_error_without_rotation(
monkeypatch: pytest.MonkeyPatch,
) -> None:
grok.save_record(_record("stale", "r1", time.time() - 10))
def _fake_post(_payload: dict[str, str]) -> dict[str, Any]:
raise grok.GrokAuthError("token_http_error", "HTTP 400: invalid_grant")
monkeypatch.setattr(grok, "_post_form", _fake_post)
with pytest.raises(grok.GrokAuthError):
grok.get_valid_token()
def test_get_valid_token_raises_when_not_signed_in() -> None:
with pytest.raises(grok.GrokAuthError) as exc:
grok.get_valid_token()
assert exc.value.code == "not_authenticated"
-102
View File
@@ -1,102 +0,0 @@
"""Grok subscription routing through StrixProvider.get_model."""
from __future__ import annotations
import argparse
from unittest import mock
from agents.models.openai_chatcompletions import OpenAIChatCompletionsModel
from strix.config import grok, subscription
from strix.config.models import StrixProvider, _TurnGuardModel
from strix.interface import scan_setup, utils
from strix.report import state as state_mod
def test_grok_prefix_routes_to_chat_completions(monkeypatch) -> None: # type: ignore[no-untyped-def]
client = mock.MagicMock()
monkeypatch.setattr(grok, "get_subscription_client", lambda: client)
model = StrixProvider().get_model("grok/grok-4")
assert isinstance(model, _TurnGuardModel)
assert isinstance(model._inner, OpenAIChatCompletionsModel)
# The provider strips the grok/ prefix and passes xAI's bare model slug.
assert model._inner.model == "grok-4"
def test_non_subscription_model_is_not_hijacked_by_grok(monkeypatch) -> None: # type: ignore[no-untyped-def]
def _boom() -> object:
msg = "grok client must not be built for a non-grok model"
raise AssertionError(msg)
monkeypatch.setattr(grok, "get_subscription_client", _boom)
# A metered xai/* key model must fall through to the normal provider path,
# not the subscription route.
model = StrixProvider().get_model("xai/grok-4")
assert isinstance(model, _TurnGuardModel)
assert not isinstance(model._inner, OpenAIChatCompletionsModel)
def test_provider_label_names_the_subscription() -> None:
assert subscription.provider_label("grok/grok-4") == "Grok"
assert subscription.provider_label("chatgpt/gpt-5.4") == "ChatGPT"
# Metered API-key models are not subscriptions.
assert subscription.provider_label("xai/grok-4") is None
assert subscription.provider_label("openai/gpt-5.4") is None
def test_run_record_reports_grok_provider(monkeypatch) -> None: # type: ignore[no-untyped-def]
settings = mock.MagicMock()
settings.llm.model = "grok/grok-4"
monkeypatch.setattr(state_mod, "load_settings", lambda: settings)
record = state_mod.ReportState(run_name="run-test").run_record
assert record["auth_mode"] == "subscription"
assert record["subscription_provider"] == "Grok"
def test_subscription_label_prefers_persisted_provider(monkeypatch) -> None: # type: ignore[no-untyped-def]
settings = mock.MagicMock()
settings.llm.model = "chatgpt/gpt-5.4" # current settings point at ChatGPT
monkeypatch.setattr(utils, "load_settings", lambda: settings)
# A resumed Grok run keeps its persisted provider even though settings changed.
resumed = mock.MagicMock(
run_record={"auth_mode": "subscription", "subscription_provider": "Grok"}
)
assert utils.subscription_label(resumed) == "Grok subscription"
# With no persisted provider, it derives the label from settings (not a
# hardcoded default).
fresh = mock.MagicMock(run_record={})
assert utils.subscription_label(fresh) == "ChatGPT subscription"
def test_persisted_run_record_carries_provider(tmp_path, monkeypatch) -> None: # type: ignore[no-untyped-def]
settings = mock.MagicMock()
settings.llm.model = "grok/grok-4"
monkeypatch.setattr(scan_setup, "load_settings", lambda: settings)
monkeypatch.setattr(scan_setup, "run_dir_for", lambda _name: tmp_path)
captured: dict[str, object] = {}
monkeypatch.setattr(
"strix.report.writer.write_run_record", lambda _dir, rec: captured.update(rec)
)
args = argparse.Namespace(
run_name="run-test",
targets_info=[],
scan_mode="scan",
instruction=None,
non_interactive=True,
local_sources=[],
diff_scope={"active": False},
scope_mode="mode",
diff_base=None,
)
scan_setup._persist_run_record(args)
# The resume/viewer record must carry the provider so resumed runs stay labeled.
assert captured["auth_mode"] == "subscription"
assert captured["subscription_provider"] == "Grok"
-107
View File
@@ -1,107 +0,0 @@
"""Shared subscription credential store: secure writes and cross-provider locking."""
from __future__ import annotations
import fcntl
import stat
from typing import TYPE_CHECKING
import pytest
from strix.config import codex, grok, subscription_store
if TYPE_CHECKING:
from pathlib import Path
def test_write_creates_owner_only_file(tmp_path: Path) -> None:
path = tmp_path / ".strix" / "subscription-auth.json"
subscription_store.write(path, {"grok": {"type": "oauth", "access": "a", "refresh": "r"}})
assert stat.S_IMODE(path.stat().st_mode) == 0o600
# No stray temp file is left behind.
assert not path.with_suffix(".json.tmp").exists()
def test_write_does_not_follow_a_symlink_at_target(tmp_path: Path) -> None:
store_dir = tmp_path / ".strix"
store_dir.mkdir()
outside = tmp_path / "attacker-target.json"
path = store_dir / "subscription-auth.json"
path.symlink_to(outside) # attacker pre-plants a symlink at the store path
subscription_store.write(path, {"grok": {"type": "oauth", "access": "a", "refresh": "r"}})
# The atomic rename replaced the symlink with a real file; nothing was
# written through it to the attacker-chosen location.
assert not path.is_symlink()
assert not outside.exists()
assert subscription_store.read(path)["grok"]["access"] == "a"
def test_providers_share_store_without_clobbering(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
store = tmp_path / ".strix" / "subscription-auth.json"
monkeypatch.setattr(codex, "AUTH_PATH", store)
monkeypatch.setattr(grok, "AUTH_PATH", store)
codex.save_record({"type": "oauth", "access": "c", "refresh": "r", "account_id": "acct"})
grok.save_record({"type": "oauth", "access": "g", "refresh": "r"})
data = subscription_store.read(store)
assert data["codex"]["access"] == "c"
assert data["grok"]["access"] == "g"
# Logging one provider out leaves the other's credential intact.
grok.logout()
remaining = subscription_store.read(store)
assert "grok" not in remaining
assert remaining["codex"]["access"] == "c"
def test_guard_is_reentrant(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
store = tmp_path / ".strix" / "subscription-auth.json"
monkeypatch.setattr(grok, "AUTH_PATH", store)
# Persisting while already holding the guard must not deadlock — this mirrors
# a token refresh saving its new record inside the refresh critical section.
with subscription_store.guard(store):
grok.save_record({"type": "oauth", "access": "g", "refresh": "r"})
record = grok.read_record()
assert record is not None
assert record["access"] == "g"
def test_mutation_aborts_when_lock_cannot_be_acquired(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
store = tmp_path / ".strix" / "subscription-auth.json"
monkeypatch.setattr(grok, "AUTH_PATH", store)
def _no_lock(*_args: object, **_kwargs: object) -> None:
raise OSError("no locks available")
monkeypatch.setattr(fcntl, "flock", _no_lock)
# Rather than silently doing an unlocked read-modify-write, the store raises
# and writes nothing.
with pytest.raises(subscription_store.StoreLockError):
grok.save_record({"type": "oauth", "access": "g", "refresh": "r"})
assert not store.exists()
def test_lock_file_rejects_a_pre_positioned_symlink(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
store_dir = tmp_path / ".strix"
store_dir.mkdir()
store = store_dir / "subscription-auth.json"
monkeypatch.setattr(grok, "AUTH_PATH", store)
# Attacker pre-plants a symlink where the lock file would be created.
outside = tmp_path / "attacker-target"
store.with_suffix(".lock").symlink_to(outside)
with pytest.raises(subscription_store.StoreLockError):
grok.save_record({"type": "oauth", "access": "g", "refresh": "r"})
# The symlink target was never created/truncated through the lock open.
assert not outside.exists()
-22
View File
@@ -110,28 +110,6 @@ def test_state_populates_model_warning_for_non_frontier_model() -> None:
assert "not a recommended frontier model" in warning
def test_snapshot_carries_the_subscription_provider_label() -> None:
os.environ["STRIX_LLM"] = "grok/grok-4"
loader._cached = None
snapshot = TuiController(args()).snapshot()
# The TUI renders this label, so it must name the actual provider rather
# than assuming ChatGPT.
assert snapshot["subscription"] is True
assert snapshot["subscription_label"] == "Grok subscription"
def test_snapshot_has_no_subscription_label_for_api_key_runs() -> None:
os.environ["STRIX_LLM"] = "openai/gpt-5.4"
loader._cached = None
snapshot = TuiController(args()).snapshot()
assert snapshot["subscription"] is False
assert snapshot["subscription_label"] == ""
def test_setup_restores_prepared_cli_targets() -> None:
setup_args = args()
setup_args.targets_info = [
-47
View File
@@ -70,53 +70,6 @@ def test_read_run_summary_finished_flag(tmp_path: Path) -> None:
assert read_run_summary(partial)["finished"] is False
def _write_record(base: Path, name: str, record: dict[str, object]) -> Path:
run_dir = base / "strix_runs" / name
run_dir.mkdir(parents=True)
(run_dir / "run.json").write_text(json.dumps(record), encoding="utf-8")
return run_dir
def test_read_run_summary_backfills_subscription_provider(tmp_path: Path) -> None:
# An older subscription run recorded no provider name; it is derived from
# the recorded provider/model slug so the viewer can label it.
run_dir = _write_record(
tmp_path,
"grok-run",
{
"auth_mode": "subscription",
"llm_usage": {"agents": [{"agent_id": "root", "model": "grok/grok-4"}]},
},
)
assert read_run_summary(run_dir)["subscription_provider"] == "Grok"
def test_read_run_summary_keeps_explicit_provider(tmp_path: Path) -> None:
run_dir = _write_record(
tmp_path,
"chatgpt-run",
{
"auth_mode": "subscription",
"subscription_provider": "ChatGPT",
"llm_usage": {"agents": [{"agent_id": "root", "model": "grok/grok-4"}]},
},
)
# An explicit field is authoritative and never overwritten by the slug.
assert read_run_summary(run_dir)["subscription_provider"] == "ChatGPT"
def test_read_run_summary_ignores_api_key_runs(tmp_path: Path) -> None:
run_dir = _write_record(
tmp_path,
"api-key-run",
{
"auth_mode": "api_key",
"llm_usage": {"agents": [{"agent_id": "root", "model": "openai/gpt-5.4"}]},
},
)
assert "subscription_provider" not in read_run_summary(run_dir)
def test_read_missing_artifacts_return_defaults(tmp_path: Path) -> None:
run_dir = _make_run(tmp_path, "empty", status="running", end_time=None)
assert read_vulnerabilities(run_dir) == []