From 8a17173a78d1c9766027ef85fa95be36bbb43d77 Mon Sep 17 00:00:00 2001 From: Anurag Mewar Date: Thu, 23 Jul 2026 19:09:41 +0530 Subject: [PATCH 01/11] feat: support API specs and Postman collections as targets --- README.md | 22 + docs/advanced/configuration.mdx | 4 + docs/usage/cli.mdx | 14 +- pyproject.toml | 2 + strix/config/settings.py | 1 + strix/core/api_spec.py | 656 ++++++++++++++++++++++++ strix/core/inputs.py | 56 +- strix/interface/main.py | 30 +- strix/interface/utils.py | 44 +- strix/skills/custom/api_spec_testing.md | 75 +++ tests/test_api_spec.py | 450 ++++++++++++++++ tests/test_api_spec_targets.py | 89 ++++ uv.lock | 2 + 13 files changed, 1423 insertions(+), 22 deletions(-) create mode 100644 strix/core/api_spec.py create mode 100644 strix/skills/custom/api_spec_testing.md create mode 100644 tests/test_api_spec.py create mode 100644 tests/test_api_spec_targets.py diff --git a/README.md b/README.md index 2122d04b4..026dda5a0 100644 --- a/README.md +++ b/README.md @@ -185,6 +185,27 @@ 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:// + +# ...with a Postman environment to resolve {{baseUrl}} / token variables +strix --target "postman://?env=" +``` + ### Advanced Testing Scenarios ```bash @@ -261,6 +282,7 @@ export LLM_API_KEY="your-api-key" # Optional export LLM_API_BASE="your-api-base-url" # if using a local model, e.g. Ollama, LMStudio export PERPLEXITY_API_KEY="your-api-key" # for search capabilities +export POSTMAN_API_KEY="PMAK-..." # to fetch Postman collections by id (postman://) export STRIX_REASONING_EFFORT="high" # control thinking effort (default: high, quick scan: medium) ``` diff --git a/docs/advanced/configuration.mdx b/docs/advanced/configuration.mdx index 5d56c9a16..6c32b04ae 100644 --- a/docs/advanced/configuration.mdx +++ b/docs/advanced/configuration.mdx @@ -41,6 +41,10 @@ Configure Strix using environment variables or a config file. API key for Perplexity AI. Enables real-time web search during scans for OSINT and vulnerability research. + + Postman API key (`PMAK-…`). Enables fetching Postman collections by id as a target (`postman://`), and Postman environments (`postman://?env=`) to resolve collection variables. Not needed when passing a local collection export file. + + Telemetry toggle. Set to `0`, `false`, `no`, or `off` to disable telemetry (PostHog, Scarf, OTEL). diff --git a/docs/usage/cli.mdx b/docs/usage/cli.mdx index 68f7d5edd..2936d52b7 100644 --- a/docs/usage/cli.mdx +++ b/docs/usage/cli.mdx @@ -12,7 +12,13 @@ strix (--target | --target-list | --mount ) [options] ## Options - 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`, `--target-list`, or `--mount`. + 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://`). Can be specified multiple times. Fresh runs require at least one target source: `--target`, `--target-list`, or `--mount`. + + When the target is an API spec, Strix parses the declared endpoints (method, path, parameters, body fields, auth) into the agent's task and authorizes the spec's base URLs as in-scope hosts - so it 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. + + + Fetching a Postman collection by id requires `POSTMAN_API_KEY`. Add `?env=` to also pull a Postman environment, which resolves `{{baseUrl}}` / token variables the collection references (e.g. `postman://?env=`). + @@ -105,6 +111,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://?env=" + # Targets from a file strix --target-list ./targets.txt diff --git a/pyproject.toml b/pyproject.toml index d389bd67e..36da134c0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -49,6 +49,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] @@ -122,6 +123,7 @@ module = [ "pydantic_settings.*", "reportlab.*", "pypdf.*", + "yaml.*", ] ignore_missing_imports = true disable_error_code = ["import-untyped"] diff --git a/strix/config/settings.py b/strix/config/settings.py index 387e13f3d..2e741a7e2 100644 --- a/strix/config/settings.py +++ b/strix/config/settings.py @@ -70,6 +70,7 @@ class IntegrationSettings(BaseSettings): model_config = _BASE_CONFIG perplexity_api_key: str | None = Field(default=None, alias="PERPLEXITY_API_KEY") + postman_api_key: str | None = Field(default=None, alias="POSTMAN_API_KEY") class ViewerSettings(BaseSettings): diff --git a/strix/core/api_spec.py b/strix/core/api_spec.py new file mode 100644 index 000000000..8b4a41935 --- /dev/null +++ b/strix/core/api_spec.py @@ -0,0 +1,656 @@ +"""Parse API specifications into a normalized endpoint inventory. + +Supports OpenAPI 3.x, Swagger 2.0, and Postman Collection v2.1. The goal is +not full schema validation but extracting the *attack surface* an agent needs: +every operation (method + path), its parameters and body fields, the declared +auth, and the base server URLs. + +The inventory feeds two places: + +* :func:`strix.core.inputs.build_root_task` renders it as an endpoint list so + the agent tests every operation instead of discovering them by crawling. +* :func:`strix.core.inputs.build_scope_context` uses the base URLs to mark the + API hosts as authorized, in-scope targets. + +Parsing is intentionally defensive: a malformed or partial spec yields whatever +could be recovered rather than raising, so a single bad operation never sinks a +whole run. +""" + +from __future__ import annotations + +import json +import logging +import re +from dataclasses import dataclass, field +from functools import lru_cache +from pathlib import Path +from typing import Any +from urllib.parse import urlsplit + +import requests +import yaml + +from strix.config import load_settings + + +logger = logging.getLogger(__name__) + + +SPEC_EXTENSIONS = frozenset({".json", ".yaml", ".yml"}) +_HTTP_METHODS = frozenset( + {"get", "put", "post", "delete", "options", "head", "patch", "trace"}, +) + +#: Cap the number of endpoints rendered into a prompt so a large spec cannot +#: blow up the context window. The full inventory is still persisted. +MAX_ENDPOINTS_RENDERED = 200 + +#: Recursion / size guards for pathological Postman trees and huge specs. +_MAX_POSTMAN_DEPTH = 25 +_MAX_ENDPOINTS_PARSED = 5000 + + +@dataclass +class Endpoint: + """A single API operation.""" + + method: str + path: str + summary: str = "" + parameters: list[str] = field(default_factory=list) + body_fields: list[str] = field(default_factory=list) + security: list[str] = field(default_factory=list) + + def render(self) -> str: + line = f"{self.method} {self.path}" + if self.summary: + line += f" — {self.summary}" + extras: list[str] = [] + if self.parameters: + extras.append("params: " + ", ".join(self.parameters)) + if self.body_fields: + extras.append("body: " + ", ".join(self.body_fields)) + if self.security: + extras.append("auth: " + ", ".join(self.security)) + if extras: + line += " (" + "; ".join(extras) + ")" + return line + + +@dataclass +class ApiSpecInventory: + """Normalized view of an API specification.""" + + spec_format: str + title: str + base_urls: list[str] + endpoints: list[Endpoint] + + def to_details(self) -> dict[str, Any]: + """Return a JSON-serializable form for the run record / viewer.""" + return { + "spec_format": self.spec_format, + "title": self.title, + "base_urls": list(self.base_urls), + "endpoint_count": len(self.endpoints), + "endpoints": [ + { + "method": e.method, + "path": e.path, + "summary": e.summary, + "parameters": e.parameters, + "body_fields": e.body_fields, + "security": e.security, + } + for e in self.endpoints + ], + } + + +class SpecParseError(ValueError): + """Raised when a file cannot be recognized or parsed as an API spec.""" + + +def _load_raw(path: Path) -> dict[str, Any]: + text = path.read_text(encoding="utf-8") + # 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"{path} is not valid JSON or YAML: {exc}") from exc + if not isinstance(data, dict): + raise SpecParseError(f"{path} does not contain a mapping at the top level") + return data + + +def _classify(raw: dict[str, Any]) -> str | None: + 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 ``openapi`` / ``swagger`` / ``postman`` for *path*, else ``None``. + + 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_raw(path) + except (OSError, SpecParseError): + return None + return _classify(raw) + + +def _schema_type(schema: Any) -> str: + if not isinstance(schema, dict): + return "" + typ = schema.get("type") + if isinstance(typ, str): + return typ + if "$ref" in schema: + return str(schema["$ref"]).rsplit("/", 1)[-1] + return "" + + +def _format_param(param: dict[str, Any]) -> str: + name = str(param.get("name", "")).strip() + if not name: + return "" + location = str(param.get("in", "")).strip() + # OpenAPI 3 nests the type under "schema"; Swagger 2 puts it inline. + typ = _schema_type(param.get("schema")) or _schema_type(param) + required = "*" if param.get("required") else "" + label = name + required + meta = ":".join(p for p in (location, typ) if p) + return f"{label} ({meta})" if meta else label + + +def _security_names(entries: Any) -> list[str]: + names: list[str] = [] + if isinstance(entries, list): + for entry in entries: + if isinstance(entry, dict): + names.extend(str(k) for k in entry) + # De-duplicate while preserving order. + return list(dict.fromkeys(names)) + + +def _body_fields(request_body: Any) -> list[str]: + if not isinstance(request_body, dict): + return [] + content = request_body.get("content") + if not isinstance(content, dict): + return [] + for media in content.values(): + schema = media.get("schema") if isinstance(media, dict) else None + props = schema.get("properties") if isinstance(schema, dict) else None + if isinstance(props, dict): + return list(props.keys()) + return [] + + +def _iter_operations(item: dict[str, Any]) -> list[tuple[str, dict[str, Any]]]: + return [ + (method, op) + for method, op in item.items() + if method.lower() in _HTTP_METHODS and isinstance(op, dict) + ] + + +def _parse_openapi(raw: dict[str, Any]) -> ApiSpecInventory: + info = raw.get("info") if isinstance(raw.get("info"), dict) else {} + title = str(info.get("title", "API")) if isinstance(info, dict) else "API" + base_urls: list[str] = [ + str(server["url"]) + for server in raw.get("servers", []) or [] + if isinstance(server, dict) and server.get("url") + ] + + global_security = _security_names(raw.get("security")) + endpoints: list[Endpoint] = [] + paths = raw.get("paths") + if isinstance(paths, dict): + for path, item in paths.items(): + if not isinstance(item, dict): + continue + shared_params = item.get("parameters", []) or [] + for method, op in _iter_operations(item): + params = [ + _format_param(p) + for p in (*shared_params, *(op.get("parameters", []) or [])) + if isinstance(p, dict) + ] + op_security = _security_names(op.get("security")) or global_security + endpoints.append( + Endpoint( + method=method.upper(), + path=str(path), + summary=str(op.get("summary", "")).strip(), + parameters=[p for p in params if p], + body_fields=_body_fields(op.get("requestBody")), + security=op_security, + ), + ) + if len(endpoints) >= _MAX_ENDPOINTS_PARSED: + return ApiSpecInventory("openapi", title, base_urls, endpoints) + return ApiSpecInventory("openapi", title, base_urls, endpoints) + + +def _parse_swagger(raw: dict[str, Any]) -> ApiSpecInventory: + info = raw.get("info") if isinstance(raw.get("info"), dict) else {} + title = str(info.get("title", "API")) if isinstance(info, dict) else "API" + host = str(raw.get("host", "")).strip() + base_path = str(raw.get("basePath", "")).strip() + schemes = [s for s in (raw.get("schemes") or ["https"]) if isinstance(s, str)] + base_urls: list[str] = [] + if host: + base_urls = [f"{scheme}://{host}{base_path}" for scheme in schemes] + elif base_path: + base_urls = [base_path] + + global_security = _security_names(raw.get("security")) + endpoints: list[Endpoint] = [] + paths = raw.get("paths") + if isinstance(paths, dict): + for path, item in paths.items(): + if not isinstance(item, dict): + continue + shared_params = item.get("parameters", []) or [] + for method, op in _iter_operations(item): + merged = [ + p + for p in (*shared_params, *(op.get("parameters", []) or [])) + if isinstance(p, dict) + ] + params = [_format_param(p) for p in merged if p.get("in") != "body"] + body_fields: list[str] = [] + for p in merged: + if p.get("in") == "body": + schema = p.get("schema") + props = schema.get("properties") if isinstance(schema, dict) else None + if isinstance(props, dict): + body_fields = list(props.keys()) + op_security = _security_names(op.get("security")) or global_security + endpoints.append( + Endpoint( + method=method.upper(), + path=str(path), + summary=str(op.get("summary", "")).strip(), + parameters=[p for p in params if p], + body_fields=body_fields, + security=op_security, + ), + ) + if len(endpoints) >= _MAX_ENDPOINTS_PARSED: + return ApiSpecInventory("swagger", title, base_urls, endpoints) + return ApiSpecInventory("swagger", title, base_urls, endpoints) + + +_VAR_PATTERN = re.compile(r"\{\{\s*([^}]+?)\s*\}\}") + + +def _collect_variables(entries: Any) -> dict[str, str]: + """Build a ``{name: value}`` map from a Postman ``variable`` block.""" + variables: dict[str, str] = {} + 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_vars(text: str, variables: dict[str, str]) -> str: + """Substitute ``{{var}}`` placeholders using the collection variables.""" + if not variables or "{{" not in text: + return text + return _VAR_PATTERN.sub(lambda m: variables.get(m.group(1), m.group(0)), text) + + +def _postman_url(url: Any, variables: dict[str, str]) -> tuple[str, str]: + """Return ``(base_url, path)`` from a Postman request ``url`` node.""" + if isinstance(url, str): + raw = url + elif isinstance(url, dict): + raw = str(url.get("raw", "")) + if not raw: + host = url.get("host") + host_str = ".".join(host) if isinstance(host, list) else str(host or "") + path = url.get("path") + path_str = "/".join(str(p) for p in path) if isinstance(path, list) else str(path or "") + raw = f"{host_str}/{path_str}".strip("/") + else: + return "", "" + raw = _resolve_vars(raw, variables) + split = urlsplit(raw) + if split.scheme and split.netloc: + return f"{split.scheme}://{split.netloc}", split.path or "/" + return "", raw + + +_PATH_VAR_PATTERN = re.compile(r":([A-Za-z0-9_]+)") + + +def _normalize_postman_path(path: str) -> tuple[str, list[str]]: + """Turn Postman ``:var`` path segments into ``{var}`` and list them as params. + + Mirrors the OpenAPI path-template style so path variables read consistently + and surface as object references worth testing (IDOR/BOLA candidates). + """ + names = _PATH_VAR_PATTERN.findall(path) + normalized = _PATH_VAR_PATTERN.sub(lambda m: "{" + m.group(1) + "}", path) + return normalized, [f"{name} (path)" for name in names] + + +def _postman_query_params(url: Any, variables: dict[str, str]) -> list[str]: + if not isinstance(url, dict) or not isinstance(url.get("query"), list): + return [] + return [ + f"{_resolve_vars(str(q['key']), variables)} (query)" + for q in url["query"] + if isinstance(q, dict) and q.get("key") and not q.get("disabled") + ] + + +def _postman_header_params(request: dict[str, Any], variables: dict[str, str]) -> list[str]: + headers = request.get("header") + if not isinstance(headers, list): + return [] + return [ + f"{_resolve_vars(str(entry['key']), variables)} (header)" + for entry in headers + if isinstance(entry, dict) and entry.get("key") and not entry.get("disabled") + ] + + +def _postman_body_fields(request: dict[str, Any], variables: dict[str, str]) -> list[str]: + body = request.get("body") + if not isinstance(body, dict): + return [] + mode = body.get("mode") + if mode == "raw": + raw = _resolve_vars(str(body.get("raw", "")), variables) + try: + parsed = json.loads(raw) + except ValueError: # JSONDecodeError is a ValueError subclass + return [] + return list(parsed.keys()) if isinstance(parsed, dict) else [] + if mode in ("urlencoded", "formdata"): + entries = body.get(mode) + if isinstance(entries, list): + return [ + str(e["key"]) + for e in entries + if isinstance(e, dict) and e.get("key") and not e.get("disabled") + ] + return [] + + +def _postman_auth(auth: Any, inherited: list[str]) -> list[str]: + """Return the auth scheme name for a request/folder, falling back to inherited.""" + if isinstance(auth, dict): + auth_type = str(auth.get("type", "")).strip() + if auth_type and auth_type != "noauth": + return [auth_type] + if auth_type == "noauth": + return [] + return inherited + + +def _walk_postman( + items: Any, + endpoints: list[Endpoint], + base_urls: set[str], + variables: dict[str, str], + inherited_auth: 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): + folder_auth = _postman_auth(node.get("auth"), inherited_auth) + _walk_postman(node["item"], endpoints, base_urls, variables, folder_auth, depth + 1) + continue + request = node.get("request") + if not isinstance(request, dict): + continue + method = str(request.get("method", "GET")).upper() + base, raw_path = _postman_url(request.get("url"), variables) + if base: + base_urls.add(base) + path, path_params = _normalize_postman_path(raw_path or "/") + params = ( + path_params + + _postman_query_params(request.get("url"), variables) + + _postman_header_params(request, variables) + ) + endpoints.append( + Endpoint( + method=method, + path=path, + summary=str(node.get("name", "")).strip(), + parameters=params, + body_fields=_postman_body_fields(request, variables), + security=_postman_auth(request.get("auth"), inherited_auth), + ), + ) + if len(endpoints) >= _MAX_ENDPOINTS_PARSED: + return + + +def _parse_postman( + raw: dict[str, Any], + extra_variables: dict[str, str] | None = None, +) -> ApiSpecInventory: + info = raw.get("info") if isinstance(raw.get("info"), dict) else {} + title = str(info.get("name", "Postman Collection")) if isinstance(info, dict) else "Postman" + variables = _collect_variables(raw.get("variable")) + if extra_variables: + variables.update(extra_variables) # environment values override collection defaults + endpoints: list[Endpoint] = [] + base_urls: set[str] = set() + collection_auth = _postman_auth(raw.get("auth"), []) + _walk_postman(raw.get("item"), endpoints, base_urls, variables, collection_auth) + return ApiSpecInventory("postman", title, sorted(base_urls), endpoints) + + +def _parse_raw(raw: dict[str, Any]) -> ApiSpecInventory: + spec_format = _classify(raw) + if spec_format == "openapi": + return _parse_openapi(raw) + if spec_format == "swagger": + return _parse_swagger(raw) + if spec_format == "postman": + return _parse_postman(raw) + raise SpecParseError("File is not a recognized OpenAPI, Swagger, or Postman spec") + + +def parse_api_spec(path: str | Path) -> ApiSpecInventory: + """Parse the spec at *path* into an :class:`ApiSpecInventory`. + + Raises :class:`SpecParseError` if the file cannot be read or recognized. + """ + p = Path(path) + try: + raw = _load_raw(p) + except OSError as exc: + raise SpecParseError(f"Cannot read spec {p}: {exc}") from exc + return _parse_raw(raw) + + +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") + if not isinstance(collection, dict): + raise SpecParseError(f"Unexpected Postman API response shape for '{collection_uid}'") + 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}`` map. + + Environments hold the base URLs, hosts, and tokens that collections + reference as ``{{variables}}``; pulling them in lets those placeholders + resolve to real values for the scan. Optional — only used when an + environment id is supplied. + """ + payload = _postman_api_json( + f"{POSTMAN_API_BASE}/environments/{environment_uid}", + api_key, + f"environment '{environment_uid}'", + ) + environment = payload.get("environment") + values = environment.get("values") if isinstance(environment, dict) else None + resolved: dict[str, str] = {} + if isinstance(values, list): + for entry in values: + if ( + isinstance(entry, dict) + and entry.get("key") is not None + and entry.get("enabled", True) + ): + resolved[str(entry["key"])] = str(entry.get("value", "")) + return resolved + + +def parse_postman_api( + collection_uid: str, + api_key: str, + environment_uid: str = "", +) -> ApiSpecInventory: + """Fetch a Postman collection (and optional environment) and parse it.""" + collection = fetch_postman_collection(collection_uid, api_key) + extra_variables = fetch_postman_environment(environment_uid, api_key) if environment_uid else {} + return _parse_postman(collection, extra_variables=extra_variables) + + +@lru_cache(maxsize=64) +def _parse_cached(resolved: str, mtime: float) -> ApiSpecInventory: + del mtime # part of the cache key only; invalidates when the file changes + return parse_api_spec(resolved) + + +@lru_cache(maxsize=16) +def _fetch_cached(collection_uid: str, api_key: str, environment_uid: str) -> ApiSpecInventory: + return parse_postman_api(collection_uid, api_key, environment_uid) + + +def load_inventory(details: dict[str, Any]) -> ApiSpecInventory | None: + """Return the inventory for an ``api_spec`` target's ``details`` block. + + Handles both sources: a local spec file (``source`` absent or ``"file"``) + and a collection pulled from the Postman API (``source == "postman_api"``). + Results are memoized so the root-task and scope-context builders do not + re-read the file or re-hit the API. Returns ``None`` (and logs) on any + failure so callers degrade gracefully. + """ + if details.get("source") == "postman_api": + uid = details.get("collection_uid") + if not uid: + return None + env_uid = str(details.get("environment_uid") or "") + try: + api_key = load_settings().integrations.postman_api_key or "" + return _fetch_cached(str(uid), api_key, env_uid) + except SpecParseError as exc: + logger.warning("api_spec: failed to fetch Postman collection %s: %s", uid, exc) + return None + + spec = details.get("target_spec") + if not spec: + return None + try: + path = Path(str(spec)).resolve() + mtime = path.stat().st_mtime + except OSError as exc: + logger.warning("api_spec: cannot stat %s: %s", spec, exc) + return None + try: + return _parse_cached(str(path), mtime) + except SpecParseError as exc: + logger.warning("api_spec: failed to parse %s: %s", spec, exc) + return None + + +def render_inventory( + inventory: ApiSpecInventory, + *, + max_endpoints: int = MAX_ENDPOINTS_RENDERED, +) -> str: + """Render an inventory as a markdown block for the root task prompt.""" + lines: list[str] = [f"- {inventory.title} ({inventory.spec_format})"] + if inventory.base_urls: + lines.append(" - Base URL(s): " + ", ".join(inventory.base_urls)) + lines.append(f" - Endpoints ({len(inventory.endpoints)}):") + lines.extend(f" - {endpoint.render()}" for endpoint in inventory.endpoints[:max_endpoints]) + remaining = len(inventory.endpoints) - max_endpoints + if remaining > 0: + lines.append(f" - … and {remaining} more endpoint(s) (see run record)") + return "\n".join(lines) diff --git a/strix/core/inputs.py b/strix/core/inputs.py index de922d07d..f1b523c5c 100644 --- a/strix/core/inputs.py +++ b/strix/core/inputs.py @@ -14,6 +14,7 @@ model_supports_reasoning, request_timeout_extra_args, ) +from strix.core.api_spec import load_inventory, render_inventory from strix.core.sessions import scrub_images_from_items @@ -33,6 +34,25 @@ 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 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 +63,7 @@ def build_root_task(scan_config: dict[str, Any]) -> str: "Local Codebases": [], "URLs": [], "IP Addresses": [], + "API Specifications": [], } for target in targets: @@ -65,6 +86,14 @@ 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": + inventory = load_inventory(details) + if inventory is not None: + sections["API Specifications"].append(render_inventory(inventory)) + else: + sections["API Specifications"].append( + f"- {details.get('target_spec', 'unknown spec')} (could not be parsed)", + ) parts: list[str] = [] for label, items in sections.items(): @@ -72,21 +101,7 @@ def build_root_task(scan_config: dict[str, Any]) -> str: parts.append(f"\n\n{label}:") parts.extend(items) - 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: @@ -101,6 +116,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") @@ -114,6 +130,16 @@ 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": + inventory = load_inventory(details) + if inventory is not None: + authorized.extend( + {"type": "web_application", "value": base_url, "workspace_path": ""} + for base_url in inventory.base_urls + ) + return { "scope_source": "system_scan_config", "authorization_source": "strix_platform_verified_targets", diff --git a/strix/interface/main.py b/strix/interface/main.py index b52be0fcd..3b9059a3f 100644 --- a/strix/interface/main.py +++ b/strix/interface/main.py @@ -30,6 +30,7 @@ is_known_openai_bare_model, is_recommended_or_frontier_model, ) +from strix.core.api_spec import SpecParseError, parse_api_spec, parse_postman_api from strix.core.paths import run_dir_for, runtime_state_dir from strix.interface.cli import run_cli from strix.interface.tui import run_tui @@ -423,6 +424,14 @@ def parse_arguments() -> argparse.Namespace: # 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 --target https://api.example.com + + # Postman collection pulled live by id (needs POSTMAN_API_KEY); optional environment + strix --target postman:// --target https://api.example.com + strix --target "postman://?env=" --target https://api.example.com + # Large local repository (bind-mounted read-only instead of copied) strix --mount ./huge-monorepo @@ -468,8 +477,10 @@ def parse_arguments() -> argparse.Namespace: "--target", type=str, action="append", - help="Target to test (URL, repository, local directory path, domain name, or IP address). " - "Can be specified multiple times for multi-target scans. " + 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://[?env=], needs " + "POSTMAN_API_KEY). Can be specified multiple times for multi-target scans. " "Fresh runs require at least one of --target, --target-list, or --mount.", ) parser.add_argument( @@ -640,6 +651,21 @@ def parse_arguments() -> argparse.Namespace: else: display_target = target + if target_type == "api_spec": + try: + if target_dict.get("source") == "postman_api": + postman_key = load_settings().integrations.postman_api_key or "" + inventory = parse_postman_api( + target_dict["collection_uuid"], + postman_key, + target_dict.get("environment_uuid", ""), + ) + else: + inventory = parse_api_spec(target_dict["target_spec"]) + except SpecParseError as exc: + parser.error(f"Invalid API spec '{target}': {exc}") + if not inventory.endpoints: + parser.error(f"API spec '{target}' contains no endpoints to test") args.targets_info.append( {"type": target_type, "details": target_dict, "original": display_target} ) diff --git a/strix/interface/utils.py b/strix/interface/utils.py index 27c10a468..b7980325e 100644 --- a/strix/interface/utils.py +++ b/strix/interface/utils.py @@ -12,7 +12,7 @@ from pathlib import Path from typing import Any from urllib.error import HTTPError, URLError -from urllib.parse import urlparse +from urllib.parse import parse_qs, urlparse from urllib.request import Request, urlopen import docker @@ -22,6 +22,7 @@ from rich.text import Text from strix.config import load_settings +from strix.core.api_spec import detect_spec_format logger = logging.getLogger(__name__) @@ -446,6 +447,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") @@ -1076,6 +1086,25 @@ 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://)" + ) + 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} @@ -1100,6 +1129,12 @@ def infer_target_type(target: str) -> tuple[str, dict[str, str]]: # noqa: PLR09 if path.exists(): if path.is_dir(): 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 @@ -1126,6 +1161,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://[?env=], " + "needs POSTMAN_API_KEY)\n" "- A domain name (e.g., example.com)\n" "- An IP address (e.g., 192.168.1.10)" ) @@ -1147,9 +1185,7 @@ def read_target_list_file(path_str: str) -> list[str]: if (target := line.strip()) and not target.startswith("#") ] except UnicodeDecodeError as e: - raise ValueError( - f"Target list file '{path_str}' must be valid UTF-8 text: {e!s}" - ) from e + raise ValueError(f"Target list file '{path_str}' must be valid UTF-8 text: {e!s}") from e except OSError as e: raise ValueError(f"Failed to read target list file '{path_str}': {e!s}") from e diff --git a/strix/skills/custom/api_spec_testing.md b/strix/skills/custom/api_spec_testing.md new file mode 100644 index 000000000..6a82bf6f5 --- /dev/null +++ b/strix/skills/custom/api_spec_testing.md @@ -0,0 +1,75 @@ +--- +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 ingested API specification (OpenAPI 3.x, Swagger 2.0, or a +Postman collection), the root task already contains a normalized endpoint +inventory under **API Specifications** — every operation with its method, path, +declared parameters, request-body fields, 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 + +**Authorization (highest yield on APIs)** +- BOLA/IDOR: swap object identifiers in path/query/body across two accounts; + confirm cross-account read or state change. Every `{id}`, `parentId`, + `accountId`, `tenantId` in the inventory is a candidate. +- BFLA: call privileged operations (declared `auth` scopes, admin paths) with a + lower-privilege token; confirm the action succeeds. +- Missing auth: replay each endpoint with the token stripped and with an expired + token; the declared `auth: …` in the inventory tells you what should be + required — flag any endpoint that returns data without it. + +**Mass assignment / excessive data exposure** +- Use the declared `body:` fields as a starting point, then add sensitive fields + the schema omits (`role`, `isAdmin`, `verified`, `balance`, `ownerId`) and + confirm they are honored. +- Check responses for fields beyond what the caller should see. + +**Injection & parameter abuse** +- For every parameter, test against its declared type: send strings where + integers/UUIDs are expected, oversized values, and injection payloads + (SQLi/NoSQLi/command/SSTI depending on backend). Type confusion often bypasses + validation. +- Test `fields`/`include`/`expand`/`filter` style knobs for authorization + bypass in resolvers/serializers. + +**Business logic & rate limits** +- Chain operations across endpoints (create → approve → withdraw) looking for + workflow/state bypass and race conditions on money/quota mutations. +- Confirm rate limiting on auth and expensive endpoints. + +## 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. diff --git a/tests/test_api_spec.py b/tests/test_api_spec.py new file mode 100644 index 000000000..122154b35 --- /dev/null +++ b/tests/test_api_spec.py @@ -0,0 +1,450 @@ +"""Tests for the API-spec parser in strix.core.api_spec.""" + +from __future__ import annotations + +import json +from collections.abc import Callable +from typing import TYPE_CHECKING, Any + +import pytest + +import strix.core.api_spec as api_spec +from strix.core.api_spec import ( + MAX_ENDPOINTS_RENDERED, + SpecParseError, + detect_spec_format, + fetch_postman_collection, + fetch_postman_environment, + load_inventory, + parse_api_spec, + parse_postman_api, + render_inventory, +) + + +if TYPE_CHECKING: + from pathlib import Path + + +OPENAPI_YAML = """ +openapi: 3.0.1 +info: + title: Shop API + version: 1.0.0 +servers: + - url: https://api.shop.test/v1 +security: + - bearerAuth: [] +paths: + /users/{id}: + parameters: + - name: id + in: path + required: true + schema: + type: string + get: + summary: Get user + parameters: + - name: expand + in: query + schema: + type: string + put: + summary: Update user + requestBody: + content: + application/json: + schema: + type: object + properties: + name: {} + role: {} +""" + +SWAGGER_JSON = { + "swagger": "2.0", + "info": {"title": "Legacy"}, + "host": "legacy.test", + "basePath": "/api", + "schemes": ["https"], + "paths": { + "/orders": { + "post": { + "summary": "Create order", + "parameters": [ + {"name": "X-Trace", "in": "header", "type": "string"}, + { + "name": "body", + "in": "body", + "schema": {"properties": {"amount": {}, "currency": {}}}, + }, + ], + } + } + }, +} + +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?limit=10", + "query": [{"key": "limit"}], + }, + }, + } + ], + }, + { + "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 + + +def test_detect_openapi_yaml(tmp_path: Path) -> None: + path = _write(tmp_path, "openapi.yaml", OPENAPI_YAML) + assert detect_spec_format(path) == "openapi" + + +def test_detect_swagger_json(tmp_path: Path) -> None: + path = _write(tmp_path, "swagger.json", json.dumps(SWAGGER_JSON)) + assert detect_spec_format(path) == "swagger" + + +def test_detect_postman_json(tmp_path: Path) -> None: + path = _write(tmp_path, "collection.json", json.dumps(POSTMAN_JSON)) + assert detect_spec_format(path) == "postman" + + +def test_detect_ignores_non_spec_extension(tmp_path: Path) -> None: + path = _write(tmp_path, "notes.txt", OPENAPI_YAML) + assert detect_spec_format(path) is None + + +def test_detect_ignores_non_spec_json(tmp_path: Path) -> None: + path = _write(tmp_path, "config.json", json.dumps({"foo": "bar"})) + assert detect_spec_format(path) is None + + +def test_parse_openapi_endpoints_and_params(tmp_path: Path) -> None: + inventory = parse_api_spec(_write(tmp_path, "openapi.yaml", OPENAPI_YAML)) + + assert inventory.spec_format == "openapi" + assert inventory.title == "Shop API" + assert inventory.base_urls == ["https://api.shop.test/v1"] + assert len(inventory.endpoints) == 2 + + get = next(e for e in inventory.endpoints if e.method == "GET") + # path-level and operation-level params are merged + assert any(p.startswith("id*") and "path" in p for p in get.parameters) + assert any("expand" in p for p in get.parameters) + assert get.security == ["bearerAuth"] + + put = next(e for e in inventory.endpoints if e.method == "PUT") + assert put.body_fields == ["name", "role"] + + +def test_parse_swagger_builds_base_url_and_body(tmp_path: Path) -> None: + inventory = parse_api_spec(_write(tmp_path, "swagger.json", json.dumps(SWAGGER_JSON))) + + assert inventory.spec_format == "swagger" + assert inventory.base_urls == ["https://legacy.test/api"] + (endpoint,) = inventory.endpoints + assert endpoint.method == "POST" + assert endpoint.path == "/orders" + assert endpoint.body_fields == ["amount", "currency"] + # body param is not rendered as a query/path param + assert all("body" not in p for p in endpoint.parameters) + assert any("X-Trace" in p for p in endpoint.parameters) + + +def test_parse_postman_walks_folders(tmp_path: Path) -> None: + inventory = parse_api_spec(_write(tmp_path, "collection.json", json.dumps(POSTMAN_JSON))) + + assert inventory.spec_format == "postman" + assert inventory.base_urls == ["https://petstore.test"] + methods = {(e.method, e.path) for e in inventory.endpoints} + assert ("GET", "/pets") in methods + assert ("POST", "/pets") in methods + + +def test_parse_rejects_unrecognized_file(tmp_path: Path) -> None: + path = _write(tmp_path, "thing.json", json.dumps({"foo": 1})) + with pytest.raises(SpecParseError): + parse_api_spec(path) + + +def test_parse_rejects_malformed_yaml(tmp_path: Path) -> None: + path = _write(tmp_path, "broken.yaml", "openapi: 3.0.0\npaths: [unclosed") + with pytest.raises(SpecParseError): + parse_api_spec(path) + + +def test_render_inventory_caps_endpoints(tmp_path: Path) -> None: + paths = {f"/r{i}": {"get": {"summary": f"op {i}"}} for i in range(MAX_ENDPOINTS_RENDERED + 25)} + spec = {"openapi": "3.0.0", "info": {"title": "Big"}, "paths": paths} + inventory = parse_api_spec(_write(tmp_path, "big.json", json.dumps(spec))) + + rendered = render_inventory(inventory) + assert "and 25 more endpoint(s)" in rendered + assert rendered.count("\n - GET") == MAX_ENDPOINTS_RENDERED + + +def test_to_details_is_json_serializable(tmp_path: Path) -> None: + inventory = parse_api_spec(_write(tmp_path, "openapi.yaml", OPENAPI_YAML)) + details = inventory.to_details() + # round-trips through json without error + assert json.loads(json.dumps(details))["endpoint_count"] == 2 + + +def test_load_inventory_returns_none_for_missing_spec() -> None: + assert load_inventory({"target_spec": "/nonexistent/spec.yaml"}) is None + + +def test_load_inventory_returns_none_without_spec_key() -> None: + assert load_inventory({}) is None + + +def test_load_inventory_parses_and_caches(tmp_path: Path) -> None: + path = _write(tmp_path, "openapi.yaml", OPENAPI_YAML) + details = {"target_spec": str(path)} + first = load_inventory(details) + second = load_inventory(details) + assert first is not None + assert second is first # served from the mtime-keyed cache + + +# --- Postman variable resolution + API fetch ----------------------------- + +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"}, + }, + } + ], +} + + +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_postman_resolves_collection_variables(tmp_path: Path) -> None: + inventory = parse_api_spec(_write(tmp_path, "vars.json", json.dumps(POSTMAN_WITH_VARS))) + assert inventory.base_urls == ["https://api.vars.test"] + assert inventory.endpoints[0].path == "/things/1" + + +def test_fetch_postman_collection_unwraps(monkeypatch: pytest.MonkeyPatch) -> None: + captured: dict[str, Any] = {} + + def fake_get(url: str, headers: dict[str, str], timeout: int) -> _FakeResponse: + captured["url"] = url + captured["headers"] = headers + return _FakeResponse(200, {"collection": POSTMAN_WITH_VARS}) + + monkeypatch.setattr(api_spec.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_parse_postman_api_end_to_end(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr( + api_spec.requests, + "get", + lambda *a, **k: _FakeResponse(200, {"collection": POSTMAN_WITH_VARS}), + ) + inventory = parse_postman_api("abc-123", "PMAK-xyz") + assert inventory.spec_format == "postman" + assert inventory.base_urls == ["https://api.vars.test"] + + +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( + api_spec.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( + api_spec.requests, + "get", + lambda *a, **k: _FakeResponse(401, {}), + ) + with pytest.raises(SpecParseError, match="rejected the key"): + fetch_postman_collection("abc-123", "bad-key") + + +# --- Postman environment resolution -------------------------------------- + +POSTMAN_NEEDS_ENV = { + "info": {"_postman_id": "e-1", "name": "Env Collection"}, + "item": [ + {"name": "Get thing", "request": {"method": "GET", "url": {"raw": "{{baseUrl}}/things/1"}}} + ], +} + +ENV_PAYLOAD = { + "environment": { + "name": "prod", + "values": [ + {"key": "baseUrl", "value": "https://api.env.test", "enabled": True}, + {"key": "secretToken", "value": "s3cr3t", "enabled": False}, + ], + } +} + + +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_postman_environment_returns_enabled_values( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(api_spec.requests, "get", lambda *a, **k: _FakeResponse(200, ENV_PAYLOAD)) + values = fetch_postman_environment("env-1", "PMAK-xyz") + assert values == {"baseUrl": "https://api.env.test"} # disabled secret excluded + + +def test_parse_postman_api_resolves_from_environment(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(api_spec.requests, "get", _dispatch_get(POSTMAN_NEEDS_ENV, ENV_PAYLOAD)) + inventory = parse_postman_api("coll-1", "PMAK-xyz", "env-1") + assert inventory.base_urls == ["https://api.env.test"] + assert inventory.endpoints[0].path == "/things/1" + + +def test_parse_postman_api_without_env_leaves_variable_unresolved( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr( + api_spec.requests, + "get", + lambda *a, **k: _FakeResponse(200, {"collection": POSTMAN_NEEDS_ENV}), + ) + inventory = parse_postman_api("coll-1", "PMAK-xyz") + # no environment supplied -> no concrete base URL recovered + assert inventory.base_urls == [] + + +# --- Postman header / body / auth extraction ----------------------------- + +POSTMAN_RICH = { + "info": {"_postman_id": "r-1", "name": "Rich"}, + "variable": [ + {"key": "baseUrl", "value": "https://api.rich.test"}, + {"key": "petName", "value": "Rex"}, + ], + "auth": {"type": "bearer", "bearer": [{"key": "token", "value": "{{token}}"}]}, + "item": [ + { + "name": "Create pet", + "request": { + "method": "POST", + "header": [ + {"key": "Content-Type", "value": "application/json"}, + {"key": "X-Trace", "value": "1", "disabled": True}, + ], + "body": {"mode": "raw", "raw": '{"name": "{{petName}}", "species": "dog"}'}, + "url": {"raw": "{{baseUrl}}/pets"}, + }, + }, + { + "name": "Admin", + "request": { + "method": "GET", + "auth": {"type": "apikey", "apikey": [{"key": "key", "value": "X-Admin"}]}, + "url": {"raw": "{{baseUrl}}/admin"}, + }, + }, + ], +} + + +def test_postman_extracts_headers_body_and_auth(tmp_path: Path) -> None: + inventory = parse_api_spec(_write(tmp_path, "rich.json", json.dumps(POSTMAN_RICH))) + assert inventory.base_urls == ["https://api.rich.test"] + + create = next(e for e in inventory.endpoints if e.path == "/pets") + assert create.body_fields == ["name", "species"] # body raw parsed, {{petName}} resolved + assert "Content-Type (header)" in create.parameters + assert all("X-Trace" not in p for p in create.parameters) # disabled header skipped + assert create.security == ["bearer"] # inherited from collection-level auth + + admin = next(e for e in inventory.endpoints if e.path == "/admin") + assert admin.security == ["apikey"] # request-level auth overrides inherited + + +POSTMAN_PATH_VARS = { + "info": {"_postman_id": "p-1", "name": "Path Vars"}, + "variable": [{"key": "baseUrl", "value": "https://api.pv.test"}], + "item": [ + { + "name": "Get order", + "request": { + "method": "GET", + "url": { + "raw": "{{baseUrl}}/users/:userId/orders/:orderId?expand=items", + "query": [{"key": "expand", "value": "items"}], + }, + }, + } + ], +} + + +def test_postman_normalizes_colon_path_variables(tmp_path: Path) -> None: + inventory = parse_api_spec(_write(tmp_path, "pv.json", json.dumps(POSTMAN_PATH_VARS))) + (endpoint,) = inventory.endpoints + # :var segments become {var}, matching the OpenAPI template style + assert endpoint.path == "/users/{userId}/orders/{orderId}" + assert "userId (path)" in endpoint.parameters + assert "orderId (path)" in endpoint.parameters + assert "expand (query)" in endpoint.parameters diff --git a/tests/test_api_spec_targets.py b/tests/test_api_spec_targets.py new file mode 100644 index 000000000..78fdf743f --- /dev/null +++ b/tests/test_api_spec_targets.py @@ -0,0 +1,89 @@ +"""Integration of the ``api_spec`` target type into detection and input builders.""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from strix.core.inputs import build_root_task, build_scope_context +from strix.interface.utils import infer_target_type + + +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 _spec_target(tmp_path: Path) -> dict[str, object]: + path = tmp_path / "openapi.json" + path.write_text(json.dumps(OPENAPI), encoding="utf-8") + ttype, details = infer_target_type(str(path)) + return {"type": ttype, "details": details, "original": str(path)} + + +def test_infer_target_type_detects_api_spec(tmp_path: Path) -> None: + target = _spec_target(tmp_path) + assert target["type"] == "api_spec" + assert target["details"]["spec_format"] == "openapi" + assert Path(target["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_root_task_renders_endpoint_inventory(tmp_path: Path) -> None: + task = build_root_task({"targets": [_spec_target(tmp_path)]}) + assert "API Specifications" in task + assert "Shop API (openapi)" in task + assert "https://api.shop.test/v1" in task + assert "GET /users/{id}" in task + + +def test_build_scope_context_authorizes_base_urls(tmp_path: Path) -> None: + context = build_scope_context({"targets": [_spec_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) diff --git a/uv.lock b/uv.lock index 46ed7d6a1..2fa40cc60 100644 --- a/uv.lock +++ b/uv.lock @@ -2424,6 +2424,7 @@ dependencies = [ { name = "pydantic" }, { name = "pydantic-settings" }, { name = "pypdf" }, + { name = "pyyaml" }, { name = "reportlab" }, { name = "requests" }, { name = "rich" }, @@ -2464,6 +2465,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" }, From 37ee873148877298e5a44d93d5cf7f818d2714a6 Mon Sep 17 00:00:00 2001 From: Anurag Mewar <51066119+5h4d0wr007@users.noreply.github.com> Date: Thu, 23 Jul 2026 21:17:09 +0530 Subject: [PATCH 02/11] Update main.py --- strix/interface/main.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/strix/interface/main.py b/strix/interface/main.py index 3b9059a3f..ba09d3e81 100644 --- a/strix/interface/main.py +++ b/strix/interface/main.py @@ -656,9 +656,9 @@ def parse_arguments() -> argparse.Namespace: if target_dict.get("source") == "postman_api": postman_key = load_settings().integrations.postman_api_key or "" inventory = parse_postman_api( - target_dict["collection_uuid"], + target_dict["collection_uid"], postman_key, - target_dict.get("environment_uuid", ""), + target_dict.get("environment_uid", ""), ) else: inventory = parse_api_spec(target_dict["target_spec"]) From 6ec302d7ef826c3acbe4c8fc516b9d34f8376727 Mon Sep 17 00:00:00 2001 From: Anurag Mewar <51066119+5h4d0wr007@users.noreply.github.com> Date: Sun, 26 Jul 2026 09:38:58 +0530 Subject: [PATCH 03/11] Update api_spec.py --- strix/core/api_spec.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/strix/core/api_spec.py b/strix/core/api_spec.py index 8b4a41935..7b6971f6d 100644 --- a/strix/core/api_spec.py +++ b/strix/core/api_spec.py @@ -234,7 +234,7 @@ def _parse_openapi(raw: dict[str, Any]) -> ApiSpecInventory: for p in (*shared_params, *(op.get("parameters", []) or [])) if isinstance(p, dict) ] - op_security = _security_names(op.get("security")) or global_security + op_security = _security_names(op.get("security")) if "security" in op else global_security endpoints.append( Endpoint( method=method.upper(), @@ -284,7 +284,7 @@ def _parse_swagger(raw: dict[str, Any]) -> ApiSpecInventory: props = schema.get("properties") if isinstance(schema, dict) else None if isinstance(props, dict): body_fields = list(props.keys()) - op_security = _security_names(op.get("security")) or global_security + op_security = _security_names(op.get("security")) if "security" in op else global_security endpoints.append( Endpoint( method=method.upper(), From bd644a992a3d942645d6a50adfc5d12c9bd0ccce Mon Sep 17 00:00:00 2001 From: Anurag Mewar <51066119+5h4d0wr007@users.noreply.github.com> Date: Sun, 26 Jul 2026 09:46:57 +0530 Subject: [PATCH 04/11] Update api_spec.py --- strix/core/api_spec.py | 81 ++++++++++++++++++++++++++++++++++++------ 1 file changed, 70 insertions(+), 11 deletions(-) diff --git a/strix/core/api_spec.py b/strix/core/api_spec.py index 7b6971f6d..58ee1ded8 100644 --- a/strix/core/api_spec.py +++ b/strix/core/api_spec.py @@ -189,17 +189,67 @@ def _security_names(entries: Any) -> list[str]: return list(dict.fromkeys(names)) -def _body_fields(request_body: Any) -> list[str]: +def _resolve_ref(ref: Any, root: dict[str, Any]) -> Any: + """Resolve a local JSON-Pointer ``$ref`` (``#/a/b``) against *root*. + + Handles both OpenAPI (``#/components/schemas/…``) and Swagger + (``#/definitions/…``) roots. Remote refs (``other.json#/…``) are not + resolved and yield ``None``. + """ + if not isinstance(ref, str) or not ref.startswith("#/"): + return None + node: Any = root + for part in ref[2:].split("/"): + part = part.replace("~1", "/").replace("~0", "~") # JSON Pointer unescape + if not isinstance(node, dict) or part not in node: + return None + node = node[part] + return node + + +def _schema_property_names( + schema: Any, + root: dict[str, Any], + seen: set[str] | None = None, +) -> list[str]: + """Collect property names from *schema*, following ``$ref`` and ``allOf`` etc. + + Recurses through ``$ref`` and the ``allOf``/``oneOf``/``anyOf`` composition + keywords so bodies declared by reference or composition still surface their + fields. ``seen`` guards against circular ``$ref`` chains. + """ + if not isinstance(schema, dict): + return [] + seen = seen if seen is not None else set() + ref = schema.get("$ref") + if isinstance(ref, str): + if ref in seen: + return [] + seen.add(ref) + return _schema_property_names(_resolve_ref(ref, root), root, seen) + names: list[str] = list(schema.get("properties", {}) or {}) + for combiner in ("allOf", "oneOf", "anyOf"): + for sub in schema.get(combiner, []) or []: + names.extend(_schema_property_names(sub, root, seen)) + return list(dict.fromkeys(names)) # de-duplicate, preserve order + + +def _body_fields(request_body: Any, root: dict[str, Any]) -> list[str]: if not isinstance(request_body, dict): return [] + # An OpenAPI requestBody may itself be a $ref into components.requestBodies. + ref = request_body.get("$ref") + if isinstance(ref, str): + resolved = _resolve_ref(ref, root) + return _body_fields(resolved, root) if isinstance(resolved, dict) else [] content = request_body.get("content") if not isinstance(content, dict): return [] for media in content.values(): schema = media.get("schema") if isinstance(media, dict) else None - props = schema.get("properties") if isinstance(schema, dict) else None - if isinstance(props, dict): - return list(props.keys()) + names = _schema_property_names(schema, root) + if names: + return names return [] @@ -234,14 +284,20 @@ def _parse_openapi(raw: dict[str, Any]) -> ApiSpecInventory: for p in (*shared_params, *(op.get("parameters", []) or [])) if isinstance(p, dict) ] - op_security = _security_names(op.get("security")) if "security" in op else global_security + # An explicit ``security: []`` opts an operation out of auth and + # must override the global policy; only a *missing* key inherits it. + op_security = ( + _security_names(op.get("security")) + if "security" in op + else global_security + ) endpoints.append( Endpoint( method=method.upper(), path=str(path), summary=str(op.get("summary", "")).strip(), parameters=[p for p in params if p], - body_fields=_body_fields(op.get("requestBody")), + body_fields=_body_fields(op.get("requestBody"), raw), security=op_security, ), ) @@ -280,11 +336,14 @@ def _parse_swagger(raw: dict[str, Any]) -> ApiSpecInventory: body_fields: list[str] = [] for p in merged: if p.get("in") == "body": - schema = p.get("schema") - props = schema.get("properties") if isinstance(schema, dict) else None - if isinstance(props, dict): - body_fields = list(props.keys()) - op_security = _security_names(op.get("security")) if "security" in op else global_security + body_fields = _schema_property_names(p.get("schema"), raw) + # An explicit ``security: []`` opts an operation out of auth and + # must override the global policy; only a *missing* key inherits it. + op_security = ( + _security_names(op.get("security")) + if "security" in op + else global_security + ) endpoints.append( Endpoint( method=method.upper(), From c641f7639b8cb7a9db242210538c5276d323eaed Mon Sep 17 00:00:00 2001 From: Anurag Mewar <51066119+5h4d0wr007@users.noreply.github.com> Date: Sun, 26 Jul 2026 09:47:51 +0530 Subject: [PATCH 05/11] Update inputs.py --- strix/core/inputs.py | 30 ++++++++++++++++++++++++++++-- 1 file changed, 28 insertions(+), 2 deletions(-) diff --git a/strix/core/inputs.py b/strix/core/inputs.py index f1b523c5c..844de5a71 100644 --- a/strix/core/inputs.py +++ b/strix/core/inputs.py @@ -4,6 +4,7 @@ import json from typing import TYPE_CHECKING, Any +from urllib.parse import urlsplit from agents.model_settings import ModelSettings from openai.types.shared import Reasoning @@ -109,8 +110,30 @@ def build_root_task(scan_config: dict[str, Any]) -> str: return task +def _explicit_target_hosts(scan_config: dict[str, Any]) -> set[str]: + """Hosts the user explicitly supplied as web/IP targets, lower-cased. + + A spec's declared base URLs are only trusted for authorization when their + host matches one of these — a spec must not silently widen scope to a host + the user never selected. + """ + hosts: set[str] = set() + for target in scan_config.get("targets", []) or []: + details = target.get("details") or {} + ttype = target.get("type") + host: str | None = None + if ttype == "web_application": + host = urlsplit(str(details.get("target_url", ""))).hostname + elif ttype == "ip_address": + host = str(details.get("target_ip", "")).strip() or None + if host: + hosts.add(host.lower()) + return hosts + + def build_scope_context(scan_config: dict[str, Any]) -> dict[str, Any]: authorized: list[dict[str, str]] = [] + explicit_hosts = _explicit_target_hosts(scan_config) value_keys = { "repository": "target_repo", "local_code": "target_path", @@ -130,14 +153,17 @@ 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. + # A spec's declared base URLs are authorized only when their host was + # also supplied explicitly as a target, so a spec cannot expand scope to + # an unrelated host the user never selected. if ttype == "api_spec": inventory = load_inventory(details) if inventory is not None: authorized.extend( {"type": "web_application", "value": base_url, "workspace_path": ""} for base_url in inventory.base_urls + if (host := urlsplit(base_url).hostname) + and host.lower() in explicit_hosts ) return { From b1d704a06cba733a8cc004a4ae4a7723b7e6f1ba Mon Sep 17 00:00:00 2001 From: Anurag Mewar <51066119+5h4d0wr007@users.noreply.github.com> Date: Sun, 26 Jul 2026 09:56:49 +0530 Subject: [PATCH 06/11] Update api_spec.py --- strix/core/api_spec.py | 82 +++++++----------------------------------- 1 file changed, 12 insertions(+), 70 deletions(-) diff --git a/strix/core/api_spec.py b/strix/core/api_spec.py index 58ee1ded8..0782ef2ec 100644 --- a/strix/core/api_spec.py +++ b/strix/core/api_spec.py @@ -9,6 +9,7 @@ * :func:`strix.core.inputs.build_root_task` renders it as an endpoint list so the agent tests every operation instead of discovering them by crawling. + * :func:`strix.core.inputs.build_scope_context` uses the base URLs to mark the API hosts as authorized, in-scope targets. @@ -189,67 +190,17 @@ def _security_names(entries: Any) -> list[str]: return list(dict.fromkeys(names)) -def _resolve_ref(ref: Any, root: dict[str, Any]) -> Any: - """Resolve a local JSON-Pointer ``$ref`` (``#/a/b``) against *root*. - - Handles both OpenAPI (``#/components/schemas/…``) and Swagger - (``#/definitions/…``) roots. Remote refs (``other.json#/…``) are not - resolved and yield ``None``. - """ - if not isinstance(ref, str) or not ref.startswith("#/"): - return None - node: Any = root - for part in ref[2:].split("/"): - part = part.replace("~1", "/").replace("~0", "~") # JSON Pointer unescape - if not isinstance(node, dict) or part not in node: - return None - node = node[part] - return node - - -def _schema_property_names( - schema: Any, - root: dict[str, Any], - seen: set[str] | None = None, -) -> list[str]: - """Collect property names from *schema*, following ``$ref`` and ``allOf`` etc. - - Recurses through ``$ref`` and the ``allOf``/``oneOf``/``anyOf`` composition - keywords so bodies declared by reference or composition still surface their - fields. ``seen`` guards against circular ``$ref`` chains. - """ - if not isinstance(schema, dict): - return [] - seen = seen if seen is not None else set() - ref = schema.get("$ref") - if isinstance(ref, str): - if ref in seen: - return [] - seen.add(ref) - return _schema_property_names(_resolve_ref(ref, root), root, seen) - names: list[str] = list(schema.get("properties", {}) or {}) - for combiner in ("allOf", "oneOf", "anyOf"): - for sub in schema.get(combiner, []) or []: - names.extend(_schema_property_names(sub, root, seen)) - return list(dict.fromkeys(names)) # de-duplicate, preserve order - - -def _body_fields(request_body: Any, root: dict[str, Any]) -> list[str]: +def _body_fields(request_body: Any) -> list[str]: if not isinstance(request_body, dict): return [] - # An OpenAPI requestBody may itself be a $ref into components.requestBodies. - ref = request_body.get("$ref") - if isinstance(ref, str): - resolved = _resolve_ref(ref, root) - return _body_fields(resolved, root) if isinstance(resolved, dict) else [] content = request_body.get("content") if not isinstance(content, dict): return [] for media in content.values(): schema = media.get("schema") if isinstance(media, dict) else None - names = _schema_property_names(schema, root) - if names: - return names + props = schema.get("properties") if isinstance(schema, dict) else None + if isinstance(props, dict): + return list(props.keys()) return [] @@ -284,20 +235,14 @@ def _parse_openapi(raw: dict[str, Any]) -> ApiSpecInventory: for p in (*shared_params, *(op.get("parameters", []) or [])) if isinstance(p, dict) ] - # An explicit ``security: []`` opts an operation out of auth and - # must override the global policy; only a *missing* key inherits it. - op_security = ( - _security_names(op.get("security")) - if "security" in op - else global_security - ) + op_security = _security_names(op.get("security")) or global_security endpoints.append( Endpoint( method=method.upper(), path=str(path), summary=str(op.get("summary", "")).strip(), parameters=[p for p in params if p], - body_fields=_body_fields(op.get("requestBody"), raw), + body_fields=_body_fields(op.get("requestBody")), security=op_security, ), ) @@ -336,14 +281,11 @@ def _parse_swagger(raw: dict[str, Any]) -> ApiSpecInventory: body_fields: list[str] = [] for p in merged: if p.get("in") == "body": - body_fields = _schema_property_names(p.get("schema"), raw) - # An explicit ``security: []`` opts an operation out of auth and - # must override the global policy; only a *missing* key inherits it. - op_security = ( - _security_names(op.get("security")) - if "security" in op - else global_security - ) + schema = p.get("schema") + props = schema.get("properties") if isinstance(schema, dict) else None + if isinstance(props, dict): + body_fields = list(props.keys()) + op_security = _security_names(op.get("security")) or global_security endpoints.append( Endpoint( method=method.upper(), From 2027f43feb83dd52eb0bf242813c2a536f7f5db3 Mon Sep 17 00:00:00 2001 From: Anurag Mewar <51066119+5h4d0wr007@users.noreply.github.com> Date: Sun, 26 Jul 2026 09:57:42 +0530 Subject: [PATCH 07/11] Update inputs.py --- strix/core/inputs.py | 30 ++---------------------------- 1 file changed, 2 insertions(+), 28 deletions(-) diff --git a/strix/core/inputs.py b/strix/core/inputs.py index 844de5a71..f1b523c5c 100644 --- a/strix/core/inputs.py +++ b/strix/core/inputs.py @@ -4,7 +4,6 @@ import json from typing import TYPE_CHECKING, Any -from urllib.parse import urlsplit from agents.model_settings import ModelSettings from openai.types.shared import Reasoning @@ -110,30 +109,8 @@ def build_root_task(scan_config: dict[str, Any]) -> str: return task -def _explicit_target_hosts(scan_config: dict[str, Any]) -> set[str]: - """Hosts the user explicitly supplied as web/IP targets, lower-cased. - - A spec's declared base URLs are only trusted for authorization when their - host matches one of these — a spec must not silently widen scope to a host - the user never selected. - """ - hosts: set[str] = set() - for target in scan_config.get("targets", []) or []: - details = target.get("details") or {} - ttype = target.get("type") - host: str | None = None - if ttype == "web_application": - host = urlsplit(str(details.get("target_url", ""))).hostname - elif ttype == "ip_address": - host = str(details.get("target_ip", "")).strip() or None - if host: - hosts.add(host.lower()) - return hosts - - def build_scope_context(scan_config: dict[str, Any]) -> dict[str, Any]: authorized: list[dict[str, str]] = [] - explicit_hosts = _explicit_target_hosts(scan_config) value_keys = { "repository": "target_repo", "local_code": "target_path", @@ -153,17 +130,14 @@ def build_scope_context(scan_config: dict[str, Any]) -> dict[str, Any]: {"type": ttype, "value": value, "workspace_path": workspace_path}, ) - # A spec's declared base URLs are authorized only when their host was - # also supplied explicitly as a target, so a spec cannot expand scope to - # an unrelated host the user never selected. + # 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": inventory = load_inventory(details) if inventory is not None: authorized.extend( {"type": "web_application", "value": base_url, "workspace_path": ""} for base_url in inventory.base_urls - if (host := urlsplit(base_url).hostname) - and host.lower() in explicit_hosts ) return { From 72aab3d4ffc7ffb3b6ed98c748d7309eb5943f89 Mon Sep 17 00:00:00 2001 From: Anurag Mewar <51066119+5h4d0wr007@users.noreply.github.com> Date: Sun, 26 Jul 2026 10:02:03 +0530 Subject: [PATCH 08/11] Update test_api_spec_targets.py From 0fee4054207432b091b47e264dae3ce10321d0fb Mon Sep 17 00:00:00 2001 From: Anurag Mewar <51066119+5h4d0wr007@users.noreply.github.com> Date: Sun, 26 Jul 2026 11:15:11 +0530 Subject: [PATCH 09/11] Update cli.mdx --- docs/usage/cli.mdx | 1 + 1 file changed, 1 insertion(+) diff --git a/docs/usage/cli.mdx b/docs/usage/cli.mdx index 2936d52b7..2dbc1bcc7 100644 --- a/docs/usage/cli.mdx +++ b/docs/usage/cli.mdx @@ -16,6 +16,7 @@ strix (--target | --target-list | --mount ) [options] When the target is an API spec, Strix parses the declared endpoints (method, path, parameters, body fields, auth) into the agent's task and authorizes the spec's base URLs as in-scope hosts - so it 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. +Strix authorizes the base URLs the spec/collection declares (including those resolved from a Postman environment). If you also pass explicit `--target` hosts, they act as an allow-list that narrows scope to just those hosts, this is useful to keep the agent off example or production URLs a spec happens to declare. Fetching a Postman collection by id requires `POSTMAN_API_KEY`. Add `?env=` to also pull a Postman environment, which resolves `{{baseUrl}}` / token variables the collection references (e.g. `postman://?env=`). From 2527b710fa8385571cc202053a42eff9fbe5f312 Mon Sep 17 00:00:00 2001 From: Anurag Mewar <51066119+5h4d0wr007@users.noreply.github.com> Date: Sun, 26 Jul 2026 11:17:21 +0530 Subject: [PATCH 10/11] Update README.md --- README.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index 026dda5a0..7f1395915 100644 --- a/README.md +++ b/README.md @@ -189,7 +189,7 @@ strix --target https://your-app.com 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: +agent knows where to send traffic. Strix authorizes all base URLs the spec/collection declares (including those resolved from a Postman environment), but pass an explicit --target alongside the spec to narrow testing to just that host: ```bash # OpenAPI / Swagger file (.json / .yaml) @@ -206,6 +206,7 @@ strix --target postman:// strix --target "postman://?env=" ``` + ### Advanced Testing Scenarios ```bash From f210830ae2c9ab23114c1085d8f121ec1cfd4bca Mon Sep 17 00:00:00 2001 From: Anurag Mewar <51066119+5h4d0wr007@users.noreply.github.com> Date: Sun, 26 Jul 2026 11:25:44 +0530 Subject: [PATCH 11/11] Update main.py --- strix/interface/main.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/strix/interface/main.py b/strix/interface/main.py index ba09d3e81..2e614f1b0 100644 --- a/strix/interface/main.py +++ b/strix/interface/main.py @@ -426,11 +426,11 @@ def parse_arguments() -> argparse.Namespace: # 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 --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:// --target https://api.example.com - strix --target "postman://?env=" --target https://api.example.com + strix --target "postman://?env=" # Large local repository (bind-mounted read-only instead of copied) strix --mount ./huge-monorepo