feat: support API specs and Postman collections as targets (#866)

This commit is contained in:
Anurag Mewar
2026-08-03 21:07:44 -07:00
committed by GitHub
parent ea6d53f4e9
commit 6719a70611
14 changed files with 1069 additions and 20 deletions
+22
View File
@@ -185,6 +185,28 @@ strix --target https://github.com/org/repo
strix --target https://your-app.com
```
### API Testing (OpenAPI / Swagger / Postman)
Point Strix at an API contract and it tests every declared endpoint instead of
having to discover them by crawling. Pair the spec with the live base URL so the
agent knows where to send traffic:
```bash
# OpenAPI / Swagger file (.json / .yaml)
strix --target ./openapi.yaml --target https://api.your-app.com
# Postman collection export
strix --target ./collection.postman_collection.json --target https://api.your-app.com
# Postman collection pulled live by id (no manual export)
export POSTMAN_API_KEY="PMAK-..."
strix --target postman://<collection-uuid>
# ...with a Postman environment to resolve {{baseUrl}} / token variables
strix --target "postman://<collection-uuid>?env=<environment-uuid>"
```
### Advanced Testing Scenarios
```bash
+4
View File
@@ -80,6 +80,10 @@ affecting the agents that do the actual testing.
API key for Perplexity AI. Enables real-time web search during scans for OSINT and vulnerability research.
</ParamField>
<ParamField path="POSTMAN_API_KEY" type="string">
Postman API key (`PMAK-…`). Enables fetching Postman collections by id as a target (`postman://<collection-uid>`), and Postman environments (`postman://<collection-uid>?env=<environment-uid>`) to resolve collection variables. Not needed when passing a local collection export file.
</ParamField>
<ParamField path="STRIX_TELEMETRY" default="1" type="string">
Telemetry toggle. Set to `0`, `false`, `no`, or `off` to disable telemetry (PostHog, Scarf, OTEL).
</ParamField>
+13 -1
View File
@@ -12,11 +12,17 @@ strix (--target <target> | --target-list <path>) [options]
## Options
<ParamField path="--target, -t" type="string">
Target to test. Accepts URLs, repositories, local directories, domains, or IP addresses. Can be specified multiple times. Fresh runs require at least one target source: `--target` or `--target-list`.
Target to test. Accepts URLs, repositories, local directories, domains, IP addresses, API spec files (OpenAPI/Swagger `.json`/`.yaml`, a Postman collection export), or a live Postman collection by id (`postman://<collection-uuid>`). Can be specified multiple times. Fresh 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.
<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>`).
</Note>
</ParamField>
<ParamField path="--target-list" type="string">
@@ -128,6 +134,12 @@ strix -n --target ./ --scan-mode quick --scope-mode diff --diff-base origin/main
# Multi-target white-box testing
strix -t https://github.com/org/app -t https://staging.example.com
# API spec + live target (OpenAPI/Swagger file or Postman collection)
strix -t ./openapi.yaml -t https://api.example.com
# Postman collection pulled live by id (+ optional environment)
strix -t "postman://<collection-uuid>?env=<environment-uuid>"
# Targets from a file
strix --target-list ./targets.txt
```
+2
View File
@@ -48,6 +48,7 @@ dependencies = [
# Cap <49: 49.x drops the universal2 macOS wheel (arm64-only), which breaks
# the Intel macOS (macos-x86_64) release build's `uv sync --frozen`.
"cryptography>=48.0.1,<49",
"pyyaml>=6.0",
]
[project.optional-dependencies]
@@ -133,6 +134,7 @@ module = [
"pydantic_settings.*",
"reportlab.*",
"pypdf.*",
"yaml.*",
"pygments.*",
]
ignore_missing_imports = true
+5
View File
@@ -122,6 +122,11 @@ class IntegrationSettings(BaseSettings):
alias="PERPLEXITY_API_KEY",
repr=False,
)
postman_api_key: str | None = Field(
default=None,
alias="POSTMAN_API_KEY",
repr=False,
)
class ViewerSettings(BaseSettings):
+57 -15
View File
@@ -33,6 +33,50 @@ def _accepts_required_tool_choice(model_name: str | None) -> bool:
return name.startswith("openai/") or is_known_openai_bare_model(name)
def _render_diff_scope(diff_scope: dict[str, Any]) -> list[str]:
"""Render pull-request diff-scope constraints as root-task lines."""
if not diff_scope.get("active"):
return []
parts: list[str] = [
"\n\nScope Constraints:",
"- Pull request diff-scope mode is active. Prioritize changed files "
"and use other files only for context.",
]
for repo_scope in diff_scope.get("repos", []) or []:
label = repo_scope.get("workspace_subdir") or repo_scope.get("source_path") or "repository"
changed = repo_scope.get("analyzable_files_count", 0)
deleted = repo_scope.get("deleted_files_count", 0)
parts.append(f"- {label}: {changed} changed file(s) in primary scope")
if deleted:
parts.append(f"- {label}: {deleted} deleted file(s) are context-only")
return parts
def _render_api_spec(details: dict[str, Any]) -> list[str]:
"""Render an API spec target as root-task lines.
The spec itself is in the workspace, so the task points at the file and lets
the agent read the contract rather than restating a parsed summary of it.
"""
title = details.get("spec_title") or details.get("target_spec", "API")
workspace_path = details.get("workspace_path", "")
lines = [
f"- {title} ({details.get('spec_format', 'api')} specification"
+ (f", available at: {workspace_path}" if workspace_path else "")
+ ")"
]
if base_urls := details.get("base_urls") or []:
lines.append(" - Base URL(s): " + ", ".join(base_urls))
lines.append(
" - Read the specification and test every operation it declares, using "
"its declared parameters, request bodies, and auth. Endpoints in the "
"specification are in scope even when nothing links to them. Load the "
"`api_spec_testing` skill for the methodology, or spawn a specialist "
"with it."
)
return lines
def build_root_task(scan_config: dict[str, Any]) -> str:
targets = scan_config.get("targets", []) or []
diff_scope = scan_config.get("diff_scope") or {}
@@ -43,6 +87,7 @@ def build_root_task(scan_config: dict[str, Any]) -> str:
"Local Codebases": [],
"URLs": [],
"IP Addresses": [],
"API Specifications": [],
}
for target in targets:
@@ -68,6 +113,8 @@ def build_root_task(scan_config: dict[str, Any]) -> str:
sections["URLs"].append(f"- {details.get('target_url', '')}")
elif ttype == "ip_address":
sections["IP Addresses"].append(f"- {details.get('target_ip', '')}")
elif ttype == "api_spec":
sections["API Specifications"].extend(_render_api_spec(details))
parts: list[str] = []
for label, items in sections.items():
@@ -92,21 +139,7 @@ def build_root_task(scan_config: dict[str, Any]) -> str:
"truth for what to do."
)
if diff_scope.get("active"):
parts.append("\n\nScope Constraints:")
parts.append(
"- Pull request diff-scope mode is active. Prioritize changed files "
"and use other files only for context.",
)
for repo_scope in diff_scope.get("repos", []) or []:
label = (
repo_scope.get("workspace_subdir") or repo_scope.get("source_path") or "repository"
)
changed = repo_scope.get("analyzable_files_count", 0)
deleted = repo_scope.get("deleted_files_count", 0)
parts.append(f"- {label}: {changed} changed file(s) in primary scope")
if deleted:
parts.append(f"- {label}: {deleted} deleted file(s) are context-only")
parts.extend(_render_diff_scope(diff_scope))
task = " ".join(parts)
if user_instructions:
@@ -121,6 +154,7 @@ def build_scope_context(scan_config: dict[str, Any]) -> dict[str, Any]:
"local_code": "target_path",
"web_application": "target_url",
"ip_address": "target_ip",
"api_spec": "target_spec",
}
for target in scan_config.get("targets", []) or []:
ttype = target.get("type", "unknown")
@@ -134,6 +168,14 @@ def build_scope_context(scan_config: dict[str, Any]) -> dict[str, Any]:
{"type": ttype, "value": value, "workspace_path": workspace_path},
)
# An API spec authorizes the hosts it declares as in-scope web targets
# so the agent can exercise every endpoint without expanding scope.
if ttype == "api_spec":
authorized.extend(
{"type": "web_application", "value": base_url, "workspace_path": ""}
for base_url in details.get("base_urls") or []
)
return {
"scope_source": "system_scan_config",
"authorization_source": "strix_platform_verified_targets",
+12 -2
View File
@@ -65,6 +65,14 @@ Examples:
# Local code analysis
strix --target ./my-project
# API spec test (OpenAPI/Swagger file or Postman collection export)
strix --target ./openapi.yaml --target https://api.example.com
strix --target ./collection.postman_collection.json
# Postman collection pulled live by id (needs POSTMAN_API_KEY); optional environment
strix --target postman://<collection-uuid> --target https://api.example.com
strix --target "postman://<collection-uuid>?env=<environment-uuid>"
# Domain penetration test
strix --target example.com
@@ -107,8 +115,10 @@ Examples:
"--target",
type=str,
action="append",
help="Target to test (URL, repository, local directory path, domain name, or IP address). "
"Local directories are mounted into the sandbox writable. "
help="Target to test: URL, repository, local directory path, domain name, IP address, "
"an API spec file (OpenAPI/Swagger .json/.yaml or a Postman collection export), or a "
"Postman collection by id (postman://<collection-uuid>[?env=<environment-uuid>], needs "
"POSTMAN_API_KEY). Local directories are mounted into the sandbox writable. "
"Can be specified multiple times for multi-target scans. "
"Fresh runs require --target or --target-list.",
)
+43 -1
View File
@@ -12,7 +12,7 @@ from __future__ import annotations
import asyncio
import logging
from datetime import UTC, datetime
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Any
from strix.config import Settings, codex, load_settings
from strix.core.paths import run_dir_for
@@ -28,8 +28,18 @@ from strix.interface.utils import (
read_target_list_file,
resolve_diff_scope_context,
rewrite_localhost_targets,
stage_api_specs,
write_fetched_collection,
)
from strix.telemetry import posthog, scarf
from strix.utils.api_spec import (
SpecParseError,
fetch_postman_collection,
fetch_postman_environment,
load_spec,
spec_base_urls,
spec_title,
)
if TYPE_CHECKING:
@@ -109,6 +119,9 @@ def build_targets_info(args: argparse.Namespace) -> None:
else:
display_target = target
if target_type == "api_spec":
_resolve_api_spec(target, target_dict)
args.targets_info.append(
{"type": target_type, "details": target_dict, "original": display_target}
)
@@ -119,6 +132,34 @@ def build_targets_info(args: argparse.Namespace) -> None:
rewrite_localhost_targets(args.targets_info, HOST_GATEWAY_HOSTNAME)
def _resolve_api_spec(target: str, details: dict[str, Any]) -> None:
"""Read the spec up front so bad input fails before the run starts.
Records the declared base URLs (the only thing scope authorization can take
from a spec) and, for a ``postman://`` target, downloads the collection to a
local file so the sandbox never needs the Postman API key.
"""
try:
if details.get("source") == "postman_api":
collection_uid = str(details["collection_uid"])
api_key = load_settings().integrations.postman_api_key or ""
raw = fetch_postman_collection(collection_uid, api_key)
environment_uid = str(details.get("environment_uid") or "")
extra_variables = (
fetch_postman_environment(environment_uid, api_key) if environment_uid else None
)
details["target_spec"] = write_fetched_collection(raw, collection_uid)
else:
raw = load_spec(str(details["target_spec"]))
extra_variables = None
base_urls = spec_base_urls(raw, extra_variables=extra_variables)
except SpecParseError as exc:
raise ValueError(f"Invalid API spec '{target}': {exc}") from None
details["spec_title"] = spec_title(raw)
details["base_urls"] = base_urls
def prepare_run(args: argparse.Namespace) -> None:
"""Resolve the run name, clone repos, compute diff-scope, and persist state.
@@ -139,6 +180,7 @@ def prepare_run(args: argparse.Namespace) -> None:
target_info["details"]["cloned_repo_path"] = cloned_path
args.local_sources = collect_local_sources(args.targets_info)
args.local_sources.extend(stage_api_specs(args.targets_info, args.run_name))
diff_scope = resolve_diff_scope_context(
local_sources=args.local_sources,
scope_mode=args.scope_mode,
+94 -1
View File
@@ -11,7 +11,7 @@ import tempfile
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
from urllib.parse import urlparse
from urllib.parse import parse_qs, urlparse
import docker
import requests
@@ -21,6 +21,7 @@ from rich.panel import Panel
from rich.text import Text
from strix.config import load_settings
from strix.utils.api_spec import detect_spec_format
logger = logging.getLogger(__name__)
@@ -484,6 +485,15 @@ def _derive_target_label_for_run_name(targets_info: list[dict[str, Any]] | None)
if target_type == "ip_address":
return str(details.get("target_ip", original) or original)
if target_type == "api_spec":
if details.get("source") == "postman_api":
return "postman-collection"
spec_path = details.get("target_spec", original)
try:
return str(Path(spec_path).stem or spec_path)
except Exception:
return str(spec_path)
return str(original or "pentest")
@@ -1113,6 +1123,24 @@ def infer_target_type(target: str) -> tuple[str, dict[str, str]]: # noqa: PLR09
return "repository", {"target_repo": target}
parsed = urlparse(target)
if parsed.scheme == "postman":
collection_uid = f"{parsed.netloc}{parsed.path}".strip("/")
if not collection_uid:
raise ValueError(
f"Missing Postman collection id in '{target}' (expected postman://<collection-uid>)"
)
details = {
"target_spec": target,
"spec_format": "postman",
"source": "postman_api",
"collection_uid": collection_uid,
}
query = parse_qs(parsed.query)
env_uid = (query.get("env") or query.get("environment") or [""])[0].strip()
if env_uid:
details["environment_uid"] = env_uid
return "api_spec", details
if parsed.scheme in ("http", "https"):
if parsed.username or parsed.password:
return "repository", {"target_repo": target}
@@ -1138,6 +1166,12 @@ def infer_target_type(target: str) -> tuple[str, dict[str, str]]: # noqa: PLR09
if path.is_dir():
check_mountable_dir(path)
return "local_code", {"target_path": str(path.resolve())}
spec_format = detect_spec_format(path)
if spec_format is not None:
return "api_spec", {
"target_spec": str(path.resolve()),
"spec_format": spec_format,
}
raise ValueError(f"Path exists but is not a directory: {target}")
except (OSError, RuntimeError) as e:
raise ValueError(f"Invalid path: {target} - {e!s}") from e
@@ -1164,6 +1198,9 @@ def infer_target_type(target: str) -> tuple[str, dict[str, str]]: # noqa: PLR09
"- A valid URL (http:// or https://)\n"
"- A Git repository URL (https://host/org/repo or git@host:org/repo.git)\n"
"- A local directory path\n"
"- An API spec file (OpenAPI/Swagger .json/.yaml or a Postman collection)\n"
"- A Postman collection by id (postman://<collection-uid>[?env=<environment-uid>], "
"needs POSTMAN_API_KEY)\n"
"- A domain name (e.g., example.com)\n"
"- An IP address (e.g., 192.168.1.10)"
)
@@ -1438,6 +1475,62 @@ def rewrite_localhost_targets(targets_info: list[dict[str, Any]], host_gateway:
details["target_ip"] = host_gateway
#: API spec targets are copied into one workspace directory rather than mounted
#: from wherever they happen to live on the host.
API_SPEC_WORKSPACE_SUBDIR = "api-specs"
def write_fetched_collection(collection: dict[str, Any], collection_uid: str) -> str:
"""Write a collection fetched from the Postman API to a local file.
Returns the file path, so a ``postman://`` target continues as an ordinary
spec file from here on and the API key never leaves the host.
"""
staging = Path(tempfile.gettempdir()) / "strix_api_specs" / "fetched"
staging.mkdir(parents=True, exist_ok=True)
path = staging / f"{sanitize_name(collection_uid)}.postman_collection.json"
path.write_text(json.dumps(collection, indent=2), encoding="utf-8")
return str(path)
def stage_api_specs(targets_info: list[dict[str, Any]], run_name: str) -> list[dict[str, Any]]:
"""Copy every ``api_spec`` target into one directory for the sandbox.
A spec is a single file the agent reads, not a tree it works in, so it is
copied to a per-run staging directory that is exposed at
``/workspace/api-specs`` instead of mounting its host location. Each target's
``workspace_path`` records where the agent will find it.
"""
specs = [t for t in targets_info if t.get("type") == "api_spec"]
if not specs:
return []
staging = Path(tempfile.gettempdir()) / "strix_api_specs" / run_name
staging.mkdir(parents=True, exist_ok=True)
used: set[str] = set()
for target in specs:
details = target["details"]
source = Path(str(details["target_spec"]))
name = source.name
stem, suffix = source.stem, source.suffix
count = 1
while name in used:
count += 1
name = f"{stem}-{count}{suffix}"
used.add(name)
shutil.copy2(source, staging / name)
details["workspace_path"] = f"/workspace/{API_SPEC_WORKSPACE_SUBDIR}/{name}"
return [
{
"source_path": str(staging),
"workspace_subdir": API_SPEC_WORKSPACE_SUBDIR,
"protect_metadata": False,
}
]
def clone_repository(repo_url: str, run_name: str, dest_name: str | None = None) -> str:
console = Console()
+61
View File
@@ -0,0 +1,61 @@
---
name: api_spec_testing
description: Spec-driven API pentesting — systematically exercise every endpoint from an ingested OpenAPI/Swagger/Postman inventory for authz, injection, and business-logic flaws
---
# API Spec Testing
When a target is an API specification (OpenAPI 3.x, Swagger 2.0, or a Postman
collection), the root task lists it under **API Specifications** with the path
to the spec file in the workspace and the authorized base URL(s). Read the spec
file first and build your own endpoint inventory from it — every operation with
its method, path, parameters, request-body schema (resolve `$ref`/`allOf`), and
auth scheme. Do not rediscover the surface by crawling. Walk the inventory
operation-by-operation and prove findings against the live base URL(s), which
are authorized in scope.
## Methodology
**1. Baseline the contract.** For each endpoint, send a well-formed request that
matches the declared schema and record the normal response (status, shape,
auth requirement). This baseline is what every abuse case is compared against.
**2. Enumerate coverage.** Track every `METHOD path` in the inventory and mark it
tested. Undocumented-but-implied siblings are worth probing too (e.g. if
`GET /users/{id}` exists, try `PUT`/`DELETE`/`PATCH` on the same path even when
the spec omits them — specs routinely under-document write operations).
**3. Prioritize by risk.** Object-scoped reads/writes, exports, admin/staff
operations, and anything touching billing, auth, or PII first.
## What to test per endpoint
Test the full range of API weaknesses against each operation, driven by what the
contract reveals — do not treat the following as an exhaustive checklist. The
highest-yield classes on APIs are **authorization** flaws, since the spec hands
you the object identifiers and privilege boundaries to abuse: examples include
BOLA/IDOR (swap `{id}`/`accountId`/`tenantId` across two accounts), BFLA
(privileged operations with a lower-privilege token), and missing/broken auth
(replay with the token stripped or expired against endpoints whose declared auth
says one is required). Beyond authorization, use the declared parameters and
body schema as a launch point for mass assignment and excessive data exposure,
injection and type-confusion on every parameter, and multi-step business-logic
and rate-limit abuse — and follow the contract wherever it suggests something
else worth probing.
## Validation
A finding is only real once reproduced against the live base URL with a
concrete request/response pair. Capture the exact HTTP request (method, path,
headers, body) and the response proving impact (another account's data, a
privileged action succeeding, an injected payload executing). Prefer two-account
diffs for authorization findings: same request, different token, unauthorized
success.
## Tips
- The base URL(s) from the spec are authorized targets — send real traffic.
- Path templates use `{param}`; substitute real values from your baseline.
- For Postman collections, saved example values and environment variables are
strong hints for valid inputs — use them to get past validation quickly.
- Keep a running coverage table so no operation in the inventory is skipped.
+312
View File
@@ -0,0 +1,312 @@
"""Recognize API specifications and extract the hosts they declare.
Supports OpenAPI 3.x, Swagger 2.0, and Postman Collection v2.1. Two things about
an API spec must be decided on the host, in code: whether a target file is a
spec at all (detection), and which base URLs it authorizes as in-scope hosts
(scope cannot be self-granted by the agent). Everything else about the contract
— operations, parameters, request bodies, auth — is left to the agent, which
reads the spec file directly in the sandbox, so ``$ref``, ``allOf``, and nested
schemas resolve properly instead of being re-parsed here. Collections held only
in Postman are fetched here too, so the API key stays on the host and never
enters the sandbox.
"""
from __future__ import annotations
import json
import logging
import re
from pathlib import Path
from typing import Any
from urllib.parse import urlsplit
import requests
import yaml
logger = logging.getLogger(__name__)
SPEC_EXTENSIONS = frozenset({".json", ".yaml", ".yml"})
#: Guard against pathological Postman folder nesting.
_MAX_POSTMAN_DEPTH = 25
class SpecParseError(ValueError):
"""Raised when a spec cannot be read, recognized, or fetched."""
def load_spec(path: str | Path) -> dict[str, Any]:
"""Load an API spec file as a mapping.
Raises :class:`SpecParseError` if the file cannot be read or is not a
JSON/YAML mapping.
"""
p = Path(path)
try:
text = p.read_text(encoding="utf-8")
except OSError as exc:
raise SpecParseError(f"Cannot read spec {p}: {exc}") from exc
# JSON is a subset of YAML, so safe_load parses both; try JSON first for a
# clearer error and to keep the fast path fast.
try:
data: Any = json.loads(text)
except json.JSONDecodeError:
try:
data = yaml.safe_load(text)
except yaml.YAMLError as exc:
raise SpecParseError(f"{p} is not valid JSON or YAML: {exc}") from exc
if not isinstance(data, dict):
raise SpecParseError(f"{p} does not contain a mapping at the top level")
return data
def classify_spec(raw: dict[str, Any]) -> str | None:
"""Return ``openapi`` / ``swagger`` / ``postman``, or ``None`` if unrecognized."""
if isinstance(raw.get("openapi"), str):
return "openapi"
if str(raw.get("swagger", "")).startswith("2"):
return "swagger"
info = raw.get("info")
if isinstance(info, dict) and ("_postman_id" in info or "item" in raw):
return "postman"
return None
def detect_spec_format(path: Path) -> str | None:
"""Return the spec format of *path*, or ``None`` if it is not a spec.
Only files whose extension is in :data:`SPEC_EXTENSIONS` are inspected; the
contents are then loaded to confirm, so an arbitrary ``.json`` config is not
mistaken for a spec.
"""
if path.suffix.lower() not in SPEC_EXTENSIONS:
return None
try:
raw = load_spec(path)
except SpecParseError:
return None
return classify_spec(raw)
def spec_title(raw: dict[str, Any]) -> str:
"""Return the spec's declared name, for display in the task and run record."""
info = raw.get("info")
if not isinstance(info, dict):
return "API"
name = info.get("title") or info.get("name") or "API"
return str(name).strip() or "API"
def _absolute_urls(candidates: list[str]) -> list[str]:
"""Keep absolute http(s) URLs, without trailing slashes, in declared order."""
urls: list[str] = []
for candidate in candidates:
split = urlsplit(candidate.strip())
if split.scheme in ("http", "https") and split.netloc:
urls.append(candidate.strip().rstrip("/"))
return list(dict.fromkeys(urls))
_SERVER_VAR_PATTERN = re.compile(r"\{([^{}/]+)\}")
def _resolve_server_url(url: str, variables: Any) -> str:
"""Substitute an OpenAPI server template's variables with their defaults."""
if "{" not in url or not isinstance(variables, dict):
return url
defaults: dict[str, str] = {}
for name, spec in variables.items():
if isinstance(spec, dict) and spec.get("default") is not None:
defaults[str(name)] = str(spec["default"])
return _SERVER_VAR_PATTERN.sub(lambda m: defaults.get(m.group(1), m.group(0)), url)
def _openapi_base_urls(raw: dict[str, Any]) -> list[str]:
servers = raw.get("servers")
if not isinstance(servers, list):
return []
return _absolute_urls(
[
_resolve_server_url(str(server["url"]), server.get("variables"))
for server in servers
if isinstance(server, dict) and server.get("url")
],
)
def _swagger_base_urls(raw: dict[str, Any]) -> list[str]:
host = str(raw.get("host", "")).strip()
if not host:
return []
base_path = str(raw.get("basePath", "")).strip()
schemes = [s for s in (raw.get("schemes") or ["https"]) if isinstance(s, str)]
return _absolute_urls([f"{scheme}://{host}{base_path}" for scheme in schemes])
_POSTMAN_VAR_PATTERN = re.compile(r"\{\{\s*([^}]+?)\s*\}\}")
def postman_variables(raw: dict[str, Any]) -> dict[str, str]:
"""Build a ``{name: value}`` map from a Postman ``variable`` block."""
variables: dict[str, str] = {}
entries = raw.get("variable")
if isinstance(entries, list):
for entry in entries:
if isinstance(entry, dict) and entry.get("key") is not None:
variables[str(entry["key"])] = str(entry.get("value", ""))
return variables
def _resolve_postman_vars(text: str, variables: dict[str, str]) -> str:
if not variables or "{{" not in text:
return text
return _POSTMAN_VAR_PATTERN.sub(lambda m: variables.get(m.group(1), m.group(0)), text)
def _postman_request_url(url: Any, variables: dict[str, str]) -> str:
if isinstance(url, str):
raw = url
elif isinstance(url, dict):
raw = str(url.get("raw", ""))
if not raw:
host = url.get("host")
raw = ".".join(str(h) for h in host) if isinstance(host, list) else str(host or "")
else:
return ""
return _resolve_postman_vars(raw, variables)
def _walk_postman_hosts(
items: Any,
variables: dict[str, str],
hosts: list[str],
depth: int = 0,
) -> None:
if depth > _MAX_POSTMAN_DEPTH or not isinstance(items, list):
return
for node in items:
if not isinstance(node, dict):
continue
if isinstance(node.get("item"), list):
_walk_postman_hosts(node["item"], variables, hosts, depth + 1)
continue
request = node.get("request")
if not isinstance(request, dict):
continue
url = _postman_request_url(request.get("url"), variables)
split = urlsplit(url)
if split.scheme and split.netloc:
hosts.append(f"{split.scheme}://{split.netloc}")
def _postman_base_urls(raw: dict[str, Any], extra_variables: dict[str, str] | None) -> list[str]:
variables = postman_variables(raw)
if extra_variables:
variables.update(extra_variables) # environment values override collection defaults
hosts: list[str] = []
_walk_postman_hosts(raw.get("item"), variables, hosts)
return _absolute_urls(sorted(set(hosts)))
def spec_base_urls(
raw: dict[str, Any],
*,
extra_variables: dict[str, str] | None = None,
) -> list[str]:
"""Return the absolute base URLs a spec declares, for scope authorization.
Relative and unresolved-template URLs are dropped: an unusable value would
otherwise be authorized as an in-scope host. Callers pair the spec with an
explicit ``--target`` host when the spec declares none.
"""
spec_format = classify_spec(raw)
if spec_format == "openapi":
return _openapi_base_urls(raw)
if spec_format == "swagger":
return _swagger_base_urls(raw)
if spec_format == "postman":
return _postman_base_urls(raw, extra_variables)
raise SpecParseError("File is not a recognized OpenAPI, Swagger, or Postman spec")
POSTMAN_API_BASE = "https://api.getpostman.com"
_POSTMAN_FETCH_TIMEOUT = 30
def _postman_api_json(url: str, api_key: str, label: str) -> dict[str, Any]:
"""GET a Postman API resource and return the parsed JSON payload.
Raises :class:`SpecParseError` with an actionable message on auth, network,
or shape errors.
"""
if not api_key:
raise SpecParseError(
"POSTMAN_API_KEY is not set. Export a Postman API key (PMAK-…) to "
"fetch from the Postman API, or pass a local collection file instead.",
)
try:
response = requests.get(
url,
headers={"X-Api-Key": api_key, "Accept": "application/json"},
timeout=_POSTMAN_FETCH_TIMEOUT,
)
except requests.RequestException as exc:
raise SpecParseError(f"Failed to reach the Postman API: {exc}") from exc
if response.status_code == 401:
raise SpecParseError("Postman API rejected the key (401). Check POSTMAN_API_KEY.")
if response.status_code == 404:
raise SpecParseError(
f"Postman {label} not found (404). Check the id and that the key can access it.",
)
if response.status_code != 200:
raise SpecParseError(f"Postman API returned HTTP {response.status_code} for {label}.")
try:
payload = response.json()
except ValueError as exc:
raise SpecParseError(f"Postman API returned non-JSON for {label}") from exc
if not isinstance(payload, dict):
raise SpecParseError(f"Unexpected Postman API response shape for {label}")
return payload
def fetch_postman_collection(collection_uid: str, api_key: str) -> dict[str, Any]:
"""Fetch a collection from the Postman API and return the raw collection dict.
Uses ``GET /collections/{uid}`` with the ``X-Api-Key`` header. The endpoint
wraps the collection under a ``collection`` key, unwrapped here so the result
matches an exported collection file.
"""
payload = _postman_api_json(
f"{POSTMAN_API_BASE}/collections/{collection_uid}",
api_key,
f"collection {collection_uid}",
)
collection = payload.get("collection", payload)
if not isinstance(collection, dict) or not collection:
raise SpecParseError(f"Postman collection {collection_uid} came back empty")
return collection
def fetch_postman_environment(environment_uid: str, api_key: str) -> dict[str, str]:
"""Fetch a Postman environment and return its enabled ``{key: value}`` pairs.
Disabled values are skipped, matching how Postman resolves an environment at
request time.
"""
payload = _postman_api_json(
f"{POSTMAN_API_BASE}/environments/{environment_uid}",
api_key,
f"environment {environment_uid}",
)
environment = payload.get("environment", payload)
values = environment.get("values") if isinstance(environment, dict) else None
if not isinstance(values, list):
return {}
return {
str(value["key"]): str(value.get("value", ""))
for value in values
if isinstance(value, dict) and value.get("key") and value.get("enabled", True)
}
+290
View File
@@ -0,0 +1,290 @@
"""Tests for spec recognition and base-URL extraction in strix.utils.api_spec."""
from __future__ import annotations
import json
from typing import TYPE_CHECKING, Any
import pytest
import requests
import yaml
from strix.utils.api_spec import (
SpecParseError,
classify_spec,
detect_spec_format,
fetch_postman_collection,
fetch_postman_environment,
load_spec,
spec_base_urls,
spec_title,
)
if TYPE_CHECKING:
from collections.abc import Callable
from pathlib import Path
OPENAPI_YAML = """
openapi: 3.0.1
info:
title: Shop API
version: 1.0.0
servers:
- url: https://{region}.api.shop.test/{ver}
variables:
region:
default: eu
ver:
default: v1
paths:
/users/{id}:
get:
summary: Get user
"""
SWAGGER_JSON = {
"swagger": "2.0",
"info": {"title": "Legacy"},
"host": "legacy.test",
"basePath": "/api",
"schemes": ["https"],
"paths": {"/orders": {"post": {"summary": "Create order"}}},
}
POSTMAN_JSON = {
"info": {"_postman_id": "abc-123", "name": "Pet Store"},
"item": [
{
"name": "Pets",
"item": [
{
"name": "List pets",
"request": {"method": "GET", "url": {"raw": "https://petstore.test/pets"}},
}
],
},
{
"name": "Add pet",
"request": {"method": "POST", "url": "https://petstore.test/pets"},
},
],
}
def _write(tmp_path: Path, name: str, content: str) -> Path:
path = tmp_path / name
path.write_text(content, encoding="utf-8")
return path
# --- detection -----------------------------------------------------------
def test_detect_openapi_yaml(tmp_path: Path) -> None:
assert detect_spec_format(_write(tmp_path, "openapi.yaml", OPENAPI_YAML)) == "openapi"
def test_detect_swagger_json(tmp_path: Path) -> None:
assert detect_spec_format(_write(tmp_path, "swagger.json", json.dumps(SWAGGER_JSON))) == (
"swagger"
)
def test_detect_postman_json(tmp_path: Path) -> None:
assert detect_spec_format(_write(tmp_path, "collection.json", json.dumps(POSTMAN_JSON))) == (
"postman"
)
def test_detect_ignores_non_spec_extension(tmp_path: Path) -> None:
assert detect_spec_format(_write(tmp_path, "notes.txt", OPENAPI_YAML)) is None
def test_detect_ignores_non_spec_json(tmp_path: Path) -> None:
assert detect_spec_format(_write(tmp_path, "config.json", json.dumps({"foo": "bar"}))) is None
def test_classify_unrecognized_is_none() -> None:
assert classify_spec({"foo": 1}) is None
# --- loading -------------------------------------------------------------
def test_load_spec_rejects_missing_file(tmp_path: Path) -> None:
with pytest.raises(SpecParseError, match="Cannot read"):
load_spec(tmp_path / "nope.yaml")
def test_load_spec_rejects_malformed_yaml(tmp_path: Path) -> None:
with pytest.raises(SpecParseError):
load_spec(_write(tmp_path, "broken.yaml", "openapi: 3.0.0\npaths: [unclosed"))
def test_load_spec_rejects_non_mapping(tmp_path: Path) -> None:
with pytest.raises(SpecParseError, match="mapping"):
load_spec(_write(tmp_path, "list.json", json.dumps([1, 2, 3])))
def test_spec_title_reads_openapi_and_postman() -> None:
assert spec_title(yaml.safe_load(OPENAPI_YAML)) == "Shop API"
assert spec_title(POSTMAN_JSON) == "Pet Store"
assert spec_title({"info": {}}) == "API"
# --- base URL extraction -------------------------------------------------
def test_openapi_base_urls_resolve_server_variables() -> None:
raw = yaml.safe_load(OPENAPI_YAML)
# {region}/{ver} substituted with their declared defaults
assert spec_base_urls(raw) == ["https://eu.api.shop.test/v1"]
def test_openapi_drops_unresolved_relative_server() -> None:
raw = {"openapi": "3.0.0", "info": {"title": "X"}, "servers": [{"url": "/v2"}]}
# relative URL is not an authorizable host
assert spec_base_urls(raw) == []
def test_swagger_base_urls_built_from_host() -> None:
assert spec_base_urls(SWAGGER_JSON) == ["https://legacy.test/api"]
def test_swagger_without_host_yields_no_base_urls() -> None:
assert spec_base_urls({"swagger": "2.0", "info": {}, "paths": {}}) == []
def test_postman_base_urls_from_request_hosts() -> None:
assert spec_base_urls(POSTMAN_JSON) == ["https://petstore.test"]
def test_spec_base_urls_rejects_unrecognized() -> None:
with pytest.raises(SpecParseError):
spec_base_urls({"foo": 1})
# --- Postman variable / environment resolution ---------------------------
POSTMAN_WITH_VARS = {
"info": {"_postman_id": "v-1", "name": "Var Collection"},
"variable": [{"key": "baseUrl", "value": "https://api.vars.test"}],
"item": [
{"name": "Get thing", "request": {"method": "GET", "url": {"raw": "{{baseUrl}}/things/1"}}}
],
}
POSTMAN_NEEDS_ENV = {
"info": {"_postman_id": "e-1", "name": "Env Collection"},
"item": [
{"name": "Get thing", "request": {"method": "GET", "url": {"raw": "{{baseUrl}}/things/1"}}}
],
}
def test_postman_resolves_collection_variables() -> None:
assert spec_base_urls(POSTMAN_WITH_VARS) == ["https://api.vars.test"]
def test_postman_without_env_leaves_variable_unresolved() -> None:
# {{baseUrl}} never resolves -> no absolute host recovered
assert spec_base_urls(POSTMAN_NEEDS_ENV) == []
def test_postman_environment_values_resolve_base_url() -> None:
resolved = spec_base_urls(
POSTMAN_NEEDS_ENV,
extra_variables={"baseUrl": "https://api.env.test"},
)
assert resolved == ["https://api.env.test"]
# --- Postman API fetch ---------------------------------------------------
class _FakeResponse:
def __init__(self, status_code: int, payload: Any) -> None:
self.status_code = status_code
self._payload = payload
def json(self) -> Any:
return self._payload
def test_fetch_postman_collection_unwraps(monkeypatch: pytest.MonkeyPatch) -> None:
captured: dict[str, Any] = {}
def fake_get(url: str, headers: dict[str, str], **_kwargs: Any) -> _FakeResponse:
captured["url"] = url
captured["headers"] = headers
return _FakeResponse(200, {"collection": POSTMAN_WITH_VARS})
monkeypatch.setattr(requests, "get", fake_get)
collection = fetch_postman_collection("abc-123", "PMAK-xyz")
assert collection["info"]["name"] == "Var Collection"
assert captured["url"].endswith("/collections/abc-123")
assert captured["headers"]["X-Api-Key"] == "PMAK-xyz"
def test_fetch_postman_missing_key_raises() -> None:
with pytest.raises(SpecParseError, match="POSTMAN_API_KEY"):
fetch_postman_collection("abc-123", "")
def test_fetch_postman_404_raises(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(requests, "get", lambda *_a, **_k: _FakeResponse(404, {}))
with pytest.raises(SpecParseError, match="not found"):
fetch_postman_collection("missing", "PMAK-xyz")
def test_fetch_postman_401_raises(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(requests, "get", lambda *_a, **_k: _FakeResponse(401, {}))
with pytest.raises(SpecParseError, match="rejected the key"):
fetch_postman_collection("abc-123", "bad-key")
def test_fetch_postman_empty_collection_raises(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(requests, "get", lambda *_a, **_k: _FakeResponse(200, {"collection": {}}))
with pytest.raises(SpecParseError, match="empty"):
fetch_postman_collection("abc-123", "PMAK-xyz")
def test_fetch_postman_environment_returns_enabled_values(
monkeypatch: pytest.MonkeyPatch,
) -> None:
payload = {
"environment": {
"name": "prod",
"values": [
{"key": "baseUrl", "value": "https://api.env.test", "enabled": True},
{"key": "secretToken", "value": "s3cr3t", "enabled": False},
],
}
}
monkeypatch.setattr(requests, "get", lambda *_a, **_k: _FakeResponse(200, payload))
values = fetch_postman_environment("env-1", "PMAK-xyz")
assert values == {"baseUrl": "https://api.env.test"} # disabled secret excluded
def _dispatch_get(
collection: dict[str, Any],
env: dict[str, Any],
) -> Callable[..., _FakeResponse]:
def fake_get(url: str, **_kwargs: Any) -> _FakeResponse:
if "/environments/" in url:
return _FakeResponse(200, env)
return _FakeResponse(200, {"collection": collection})
return fake_get
def test_fetch_then_resolve_from_environment(monkeypatch: pytest.MonkeyPatch) -> None:
env = {"environment": {"values": [{"key": "baseUrl", "value": "https://api.env.test"}]}}
monkeypatch.setattr(requests, "get", _dispatch_get(POSTMAN_NEEDS_ENV, env))
collection = fetch_postman_collection("coll-1", "PMAK-xyz")
variables = fetch_postman_environment("env-1", "PMAK-xyz")
assert spec_base_urls(collection, extra_variables=variables) == ["https://api.env.test"]
+152
View File
@@ -0,0 +1,152 @@
"""Integration of the ``api_spec`` target type into detection, staging, and inputs."""
from __future__ import annotations
import argparse
import json
from pathlib import Path
from typing import Any
import pytest
from strix.core.inputs import build_root_task, build_scope_context
from strix.interface.scan_setup import build_targets_info
from strix.interface.utils import infer_target_type, stage_api_specs
OPENAPI = {
"openapi": "3.0.0",
"info": {"title": "Shop API", "version": "1"},
"servers": [{"url": "https://api.shop.test/v1"}],
"paths": {
"/users/{id}": {
"get": {
"summary": "Get user",
"parameters": [{"name": "id", "in": "path", "schema": {"type": "string"}}],
}
}
},
}
def _write_spec(directory: Path, name: str = "openapi.json") -> Path:
directory.mkdir(parents=True, exist_ok=True)
path = directory / name
path.write_text(json.dumps(OPENAPI), encoding="utf-8")
return path
def _resolved_targets(*spec_paths: Path) -> list[dict[str, Any]]:
"""Run spec targets through the real setup path (detection + spec resolution)."""
args = argparse.Namespace(target=[str(p) for p in spec_paths], target_list=None)
build_targets_info(args)
targets: list[dict[str, Any]] = args.targets_info
return targets
def _staged_target(tmp_path: Path, run_name: str = "test-run") -> dict[str, Any]:
targets = _resolved_targets(_write_spec(tmp_path / "src"))
stage_api_specs(targets, run_name)
return targets[0]
def test_infer_target_type_detects_api_spec(tmp_path: Path) -> None:
path = _write_spec(tmp_path)
ttype, details = infer_target_type(str(path))
assert ttype == "api_spec"
assert details["spec_format"] == "openapi"
assert Path(details["target_spec"]).is_absolute()
def test_infer_target_type_still_rejects_non_spec_file(tmp_path: Path) -> None:
path = tmp_path / "data.json"
path.write_text(json.dumps({"foo": "bar"}), encoding="utf-8")
with pytest.raises(ValueError, match="not a directory"):
infer_target_type(str(path))
def test_infer_target_type_detects_postman_uri() -> None:
ttype, details = infer_target_type("postman://12345-abcdef-uid")
assert ttype == "api_spec"
assert details["source"] == "postman_api"
assert details["collection_uid"] == "12345-abcdef-uid"
assert details["spec_format"] == "postman"
def test_infer_target_type_rejects_empty_postman_uri() -> None:
with pytest.raises(ValueError, match="collection id"):
infer_target_type("postman://")
def test_infer_target_type_parses_postman_environment() -> None:
_ttype, details = infer_target_type("postman://coll-uid?env=env-uid")
assert details["collection_uid"] == "coll-uid"
assert details["environment_uid"] == "env-uid"
def test_infer_target_type_postman_without_env_omits_key() -> None:
_ttype, details = infer_target_type("postman://coll-uid")
assert "environment_uid" not in details
def test_build_targets_info_records_title_and_base_urls(tmp_path: Path) -> None:
(target,) = _resolved_targets(_write_spec(tmp_path))
assert target["details"]["spec_title"] == "Shop API"
assert target["details"]["base_urls"] == ["https://api.shop.test/v1"]
def test_build_targets_info_rejects_unparseable_spec(tmp_path: Path) -> None:
path = tmp_path / "openapi.json"
path.write_text('{"openapi": "3.0.0", "info": {"title": "X"}, "paths"', encoding="utf-8")
args = argparse.Namespace(target=[str(path)], target_list=None)
# a broken file is not recognized as a spec, so it fails as an unusable target
with pytest.raises(ValueError, match="Invalid target"):
build_targets_info(args)
def test_stage_api_specs_copies_spec_into_workspace_dir(tmp_path: Path) -> None:
targets = _resolved_targets(_write_spec(tmp_path / "src"))
(source,) = stage_api_specs(targets, "stage-run")
assert source["workspace_subdir"] == "api-specs"
staged = Path(source["source_path"]) / "openapi.json"
assert json.loads(staged.read_text(encoding="utf-8"))["info"]["title"] == "Shop API"
assert targets[0]["details"]["workspace_path"] == "/workspace/api-specs/openapi.json"
def test_stage_api_specs_disambiguates_same_filename(tmp_path: Path) -> None:
targets = _resolved_targets(
_write_spec(tmp_path / "a"),
_write_spec(tmp_path / "b"),
)
(source,) = stage_api_specs(targets, "dupe-run")
staged_paths = [t["details"]["workspace_path"] for t in targets]
assert staged_paths == [
"/workspace/api-specs/openapi.json",
"/workspace/api-specs/openapi-2.json",
]
assert (Path(source["source_path"]) / "openapi-2.json").is_file()
def test_stage_api_specs_without_specs_returns_nothing() -> None:
assert stage_api_specs([{"type": "web_application", "details": {}}], "run") == []
def test_build_root_task_points_at_the_spec_file(tmp_path: Path) -> None:
task = build_root_task({"targets": [_staged_target(tmp_path)]})
assert "API Specifications" in task
assert "Shop API (openapi specification" in task
assert "/workspace/api-specs/openapi.json" in task
assert "https://api.shop.test/v1" in task
assert "test every operation it declares" in task
def test_build_scope_context_authorizes_base_urls(tmp_path: Path) -> None:
context = build_scope_context({"targets": [_staged_target(tmp_path)]})
authorized = context["authorized_targets"]
types = {a["type"] for a in authorized}
assert "api_spec" in types
assert "web_application" in types
assert any(a["value"] == "https://api.shop.test/v1" for a in authorized)
Generated
+2
View File
@@ -2391,6 +2391,7 @@ dependencies = [
{ name = "pydantic" },
{ name = "pydantic-settings" },
{ name = "pypdf" },
{ name = "pyyaml" },
{ name = "reportlab" },
{ name = "requests" },
{ name = "rich" },
@@ -2431,6 +2432,7 @@ requires-dist = [
{ name = "pydantic", specifier = ">=2.11.3" },
{ name = "pydantic-settings", specifier = ">=2.13.0" },
{ name = "pypdf", specifier = ">=5.0" },
{ name = "pyyaml", specifier = ">=6.0" },
{ name = "reportlab", specifier = ">=4.0" },
{ name = "requests", specifier = ">=2.32.0" },
{ name = "rich" },