diff --git a/README.md b/README.md index 982a635de..58281588f 100644 --- a/README.md +++ b/README.md @@ -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. 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) +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 +283,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 fe911907c..a90b2fce3 100644 --- a/docs/advanced/configuration.mdx +++ b/docs/advanced/configuration.mdx @@ -66,6 +66,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. + + 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 3b10f9dea..37ec9f3e6 100644 --- a/docs/usage/cli.mdx +++ b/docs/usage/cli.mdx @@ -12,7 +12,14 @@ 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. + +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=`). + @@ -105,6 +112,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 2aae0541c..4b90db486 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.*", "pygments.*", ] ignore_missing_imports = true diff --git a/strix/config/settings.py b/strix/config/settings.py index 98d68934d..2e60e7d8c 100644 --- a/strix/config/settings.py +++ b/strix/config/settings.py @@ -82,6 +82,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..0782ef2ec --- /dev/null +++ b/strix/core/api_spec.py @@ -0,0 +1,657 @@ +"""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 8ca10009c..df80a8ead 100644 --- a/strix/interface/main.py +++ b/strix/interface/main.py @@ -31,6 +31,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 @@ -497,6 +498,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 + + # Postman collection pulled live by id (needs POSTMAN_API_KEY); optional environment + strix --target postman:// --target https://api.example.com + strix --target "postman://?env=" + # Large local repository (bind-mounted read-only instead of copied) strix --mount ./huge-monorepo @@ -542,8 +551,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( @@ -714,6 +725,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_uid"], + postman_key, + target_dict.get("environment_uid", ""), + ) + 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 44a570d1a..e66b033a1 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__) @@ -480,6 +481,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") @@ -1110,6 +1120,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} @@ -1134,6 +1163,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 @@ -1160,6 +1195,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)" ) 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 775dc08b2..2dc82cc1d 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" },