diff --git a/docs/usage/cli.mdx b/docs/usage/cli.mdx index 2fc492e9b..965f6f91b 100644 --- a/docs/usage/cli.mdx +++ b/docs/usage/cli.mdx @@ -21,7 +21,7 @@ strix (--target | --target-list ) [options] - 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=`). + Fetching a Postman collection by id requires `POSTMAN_API_KEY`. Add `?env=` to also pull a Postman environment (e.g. `postman://?env=`). Enabled values referenced by `{{baseUrl}}`, `{{token}}`, or other templates are included in the staged collection's variable block, overriding collection defaults. Dependencies of referenced variables are included too; unrelated environment values and the host's Postman API key are not copied. Environment variables accessed only through script APIs such as `pm.environment.get(...)` are not inferred; provide the values needed for those requests explicitly. diff --git a/strix/interface/scan_setup.py b/strix/interface/scan_setup.py index 62176f761..a1847bed3 100644 --- a/strix/interface/scan_setup.py +++ b/strix/interface/scan_setup.py @@ -40,6 +40,7 @@ from strix.utils.api_spec import ( load_spec, spec_base_urls, spec_title, + with_postman_environment, ) @@ -224,6 +225,8 @@ def _resolve_api_spec(target: str, details: dict[str, Any]) -> None: extra_variables = ( fetch_postman_environment(environment_uid, api_key) if environment_uid else None ) + if extra_variables: + raw = with_postman_environment(raw, extra_variables) details["target_spec"] = write_fetched_collection(raw, collection_uid) else: raw = load_spec(str(details["target_spec"])) diff --git a/strix/interface/utils.py b/strix/interface/utils.py index 7ab55d8ee..8ae51a870 100644 --- a/strix/interface/utils.py +++ b/strix/interface/utils.py @@ -21,6 +21,7 @@ from rich.text import Text from strix.config import load_settings from strix.telemetry import report_error from strix.utils.api_spec import detect_spec_format +from strix.utils.secret_files import write_secret_text logger = logging.getLogger(__name__) @@ -1500,6 +1501,10 @@ def rewrite_localhost_targets(targets_info: list[dict[str, Any]], host_gateway: #: from wherever they happen to live on the host. API_SPEC_WORKSPACE_SUBDIR = "api-specs" +# Keep ownership of downloads until staging succeeds. TemporaryDirectory also +# removes pending downloads at normal process exit if setup is abandoned. +_fetched_collections: dict[str, tempfile.TemporaryDirectory[str]] = {} + def write_fetched_collection(collection: dict[str, Any], collection_uid: str) -> str: """Write a collection fetched from the Postman API to a local file. @@ -1509,8 +1514,16 @@ def write_fetched_collection(collection: dict[str, Any], collection_uid: str) -> """ staging = Path(tempfile.gettempdir()) / "strix_api_specs" / "fetched" staging.mkdir(parents=True, exist_ok=True) - path = staging / f"{sanitize_name(collection_uid)}.postman_collection.json" - path.write_text(json.dumps(collection, indent=2), encoding="utf-8") + # One collection can be targeted with multiple environments in the same run. + # Keep each resolved snapshot independent, including across concurrent runs. + snapshot = tempfile.TemporaryDirectory(prefix="collection-", dir=staging) + path = Path(snapshot.name) / f"{sanitize_name(collection_uid)}.postman_collection.json" + try: + write_secret_text(path, json.dumps(collection, indent=2)) + except BaseException: + snapshot.cleanup() + raise + _fetched_collections[str(path)] = snapshot return str(path) @@ -1540,9 +1553,22 @@ def stage_api_specs(targets_info: list[dict[str, Any]], run_name: str) -> list[d count += 1 name = f"{stem}-{count}{suffix}" used.add(name) - shutil.copy2(source, staging / name) + destination = staging / name + if source.resolve() != destination.resolve(): + shutil.copy2(source, destination) details["workspace_path"] = f"/workspace/{API_SPEC_WORKSPACE_SUBDIR}/{name}" + # Release only downloads this process created, after every copy succeeds. + # Retain the per-run staged files: resume still needs those snapshots. + for target in specs: + details = target["details"] + downloaded = str(details["target_spec"]) + snapshot = _fetched_collections.get(downloaded) + if snapshot is not None: + details["target_spec"] = str(staging / Path(details["workspace_path"]).name) + snapshot.cleanup() + del _fetched_collections[downloaded] + return [ { "source_path": str(staging), diff --git a/strix/utils/api_spec.py b/strix/utils/api_spec.py index db3a77388..730609033 100644 --- a/strix/utils/api_spec.py +++ b/strix/utils/api_spec.py @@ -17,13 +17,17 @@ import json import logging import re from pathlib import Path -from typing import Any +from typing import TYPE_CHECKING, Any, cast from urllib.parse import urlsplit import requests import yaml +if TYPE_CHECKING: + from collections.abc import Sequence + + logger = logging.getLogger(__name__) @@ -210,6 +214,58 @@ def _postman_base_urls(raw: dict[str, Any], extra_variables: dict[str, str] | No return _absolute_urls(sorted(set(hosts))) +def _postman_variable_references(value: Any) -> set[str]: + references: set[str] = set() + pending = [value] + while pending: + node = pending.pop() + if isinstance(node, str): + references.update(match.group(1) for match in _POSTMAN_VAR_PATTERN.finditer(node)) + elif isinstance(node, dict): + pending.extend(cast("dict[str, Any]", node).values()) + elif isinstance(node, list): + pending.extend(cast("Sequence[Any]", node)) + return references + + +def with_postman_environment(raw: dict[str, Any], environment: dict[str, str]) -> dict[str, Any]: + """Carry referenced environment values in the collection's variable block. + + Preserve request templates and collection metadata. Only include environment + values referenced by ``{{name}}`` templates, including dependencies of used + variables; unused environment credentials must not enter the sandbox. + """ + variables = postman_variables(raw) | environment + pending = list(_postman_variable_references({k: v for k, v in raw.items() if k != "variable"})) + referenced: set[str] = set() + while pending: + name = pending.pop() + if name in referenced: + continue + referenced.add(name) + pending.extend(_postman_variable_references(variables.get(name, "")) - referenced) + + selected = {name: environment[name] for name in sorted(referenced & environment.keys())} + if not selected: + return raw + raw_entries = raw.get("variable") + entries = cast("Sequence[Any]", raw_entries) if isinstance(raw_entries, list) else () + merged: list[Any] = [] + for entry in entries: + if isinstance(entry, dict): + variable = cast("dict[str, Any]", entry) + key = variable.get("key") + if key is not None and str(key) in selected: + merged.append({**variable, "value": selected[str(key)]}) + continue + merged.append(entry) + existing = postman_variables(raw) + merged.extend( + {"key": name, "value": value} for name, value in selected.items() if name not in existing + ) + return {**raw, "variable": merged} + + def spec_base_urls( raw: dict[str, Any], *, diff --git a/tests/test_postman_environment_handoff.py b/tests/test_postman_environment_handoff.py new file mode 100644 index 000000000..52bba6108 --- /dev/null +++ b/tests/test_postman_environment_handoff.py @@ -0,0 +1,199 @@ +"""Postman environments must survive target resolution and sandbox staging.""" + +from __future__ import annotations + +import argparse +import copy +import json +import shutil +from pathlib import Path +from types import SimpleNamespace +from typing import Any + +import pytest +import requests + +from strix.interface.scan_setup import build_targets_info +from strix.interface.utils import stage_api_specs +from strix.utils.api_spec import postman_variables + + +COLLECTION = { + "info": {"_postman_id": "collection-1", "name": "Environment handoff"}, + "variable": [ + {"key": "baseUrl", "value": "https://default.test", "description": "API host"}, + {"key": "fallback", "value": "collection fallback"}, + {"key": "unusedDefault", "value": "{{unusedSecret}}"}, + ], + "auth": {"type": "bearer", "bearer": [{"key": "token", "value": "{{ token }}"}]}, + "item": [ + { + "name": "Nested folder", + "item": [ + { + "request": { + "method": "POST", + "url": "{{baseUrl}}/tenants/{{tenant}}", + "header": [{"key": "X-Fallback", "value": "{{fallback}}"}], + "body": {"mode": "raw", "raw": '{"account": "{{account}}"}'}, + } + } + ], + } + ], +} + + +@pytest.fixture +def postman_api(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> dict[str, Any]: + collection = copy.deepcopy(COLLECTION) + payloads: dict[str, Any] = { + "collection": collection, + "environments": { + "env-1": [ + {"key": "baseUrl", "value": "https://one.test"}, + {"key": "token", "value": "{{accessToken}}"}, + {"key": "accessToken", "value": "scan-only-token"}, + {"key": "tenant", "value": "tenant-one"}, + {"key": "account", "value": ""}, + {"key": "fallback", "value": "disabled value", "enabled": False}, + {"key": "unusedSecret", "value": "unrelated-secret"}, + ], + "env-2": [ + {"key": "baseUrl", "value": "https://two.test"}, + {"key": "tenant", "value": "tenant-two"}, + ], + }, + "calls": [], + } + + def get(url: str, **kwargs: Any) -> SimpleNamespace: + assert kwargs["headers"]["X-Api-Key"] == "host-only-api-key" + payloads["calls"].append(url) + if "/environments/" in url: + body = {"environment": {"values": payloads["environments"][url.rsplit("/", 1)[1]]}} + else: + body = {"collection": collection} + return SimpleNamespace(status_code=200, json=lambda: body) + + monkeypatch.setattr(requests, "get", get) + monkeypatch.setattr("strix.interface.utils.tempfile.gettempdir", lambda: str(tmp_path)) + monkeypatch.setattr( + "strix.interface.scan_setup.load_settings", + lambda: SimpleNamespace(integrations=SimpleNamespace(postman_api_key="host-only-api-key")), + ) + return payloads + + +def _targets(*uris: str) -> list[dict[str, Any]]: + args = argparse.Namespace(target=list(uris), target_list=None) + build_targets_info(args) + targets: list[dict[str, Any]] = args.targets_info + return targets + + +def test_environment_values_reach_the_staged_collection(postman_api: dict[str, Any]) -> None: + targets = _targets("postman://collection-1?env=env-1") + (source,) = stage_api_specs(targets, "environment-handoff") + details = targets[0]["details"] + staged = Path(source["source_path"]) / Path(details["workspace_path"]).name + raw = json.loads(staged.read_text()) + variables = postman_variables(raw) + + assert details["base_urls"] == ["https://one.test"] + assert variables["baseUrl"] == "https://one.test" + assert variables["token"] == "{{accessToken}}" # noqa: S105 - unresolved template + assert variables["accessToken"] == "scan-only-token" + assert variables["tenant"] == "tenant-one" + assert variables["account"] == "" + assert variables["fallback"] == "collection fallback" + assert "unusedSecret" not in variables + assert "host-only-api-key" not in staged.read_text() + assert "unrelated-secret" not in staged.read_text() + assert "scan-only-token" not in json.dumps(targets) + assert raw["auth"] == COLLECTION["auth"] + assert raw["item"] == COLLECTION["item"] + assert raw["variable"][0]["description"] == "API host" + assert postman_api["collection"] == COLLECTION + + +@pytest.mark.usefixtures("postman_api") +def test_two_environments_of_one_collection_keep_distinct_snapshots() -> None: + targets = _targets("postman://collection-1?env=env-1", "postman://collection-1?env=env-2") + assert targets[0]["details"]["target_spec"] != targets[1]["details"]["target_spec"] + (source,) = stage_api_specs(targets, "two-environments") + assert targets[0]["details"]["workspace_path"] != targets[1]["details"]["workspace_path"] + for target, expected in zip(targets, ("tenant-one", "tenant-two"), strict=True): + staged = Path(source["source_path"]) / Path(target["details"]["workspace_path"]).name + assert postman_variables(json.loads(staged.read_text()))["tenant"] == expected + + +@pytest.mark.usefixtures("postman_api") +def test_environment_snapshot_is_private_before_and_after_staging() -> None: + targets = _targets("postman://collection-1?env=env-1") + assert Path(targets[0]["details"]["target_spec"]).stat().st_mode & 0o777 == 0o600 + (source,) = stage_api_specs(targets, "private-environment") + staged = Path(source["source_path"]) / Path(targets[0]["details"]["workspace_path"]).name + assert staged.stat().st_mode & 0o777 == 0o600 + + +def test_collection_without_environment_keeps_its_contents(postman_api: dict[str, Any]) -> None: + (target,) = _targets("postman://collection-1") + assert json.loads(Path(target["details"]["target_spec"]).read_text()) == COLLECTION + assert len(postman_api["calls"]) == 1 + + +def test_referenced_variable_cycles_do_not_expand_forever(postman_api: dict[str, Any]) -> None: + postman_api["environments"]["env-1"] = [ + {"key": "token", "value": "{{cycle}}"}, + {"key": "cycle", "value": "{{token}}"}, + ] + (target,) = _targets("postman://collection-1?env=env-1") + variables = postman_variables(json.loads(Path(target["details"]["target_spec"]).read_text())) + assert variables["token"] == "{{cycle}}" # noqa: S105 - unresolved template + assert variables["cycle"] == "{{token}}" + + +def test_references_in_collection_values_find_environment_dependencies( + postman_api: dict[str, Any], +) -> None: + postman_api["collection"]["variable"].append({"key": "token", "value": "{{accessToken}}"}) + postman_api["environments"]["env-1"] = [{"key": "accessToken", "value": "scan-only-token"}] + (target,) = _targets("postman://collection-1?env=env-1") + variables = postman_variables(json.loads(Path(target["details"]["target_spec"]).read_text())) + assert variables["token"] == "{{accessToken}}" # noqa: S105 - unresolved template + assert variables["accessToken"] == "scan-only-token" + + +@pytest.mark.usefixtures("postman_api") +def test_staging_releases_the_download_snapshot_and_keeps_the_run_copy() -> None: + targets = _targets("postman://collection-1?env=env-1") + downloaded = Path(targets[0]["details"]["target_spec"]) + (source,) = stage_api_specs(targets, "snapshot-cleanup") + + assert not downloaded.parent.exists() + staged = Path(targets[0]["details"]["target_spec"]) + assert staged.parent == Path(source["source_path"]) + assert postman_variables(json.loads(staged.read_text()))["tenant"] == "tenant-one" + # Repeating setup uses the retained run snapshot rather than a removed input. + assert stage_api_specs(targets, "snapshot-cleanup") == [source] + + +@pytest.mark.usefixtures("postman_api") +def test_failed_staging_keeps_downloads_available_for_retry( + monkeypatch: pytest.MonkeyPatch, +) -> None: + targets = _targets("postman://collection-1?env=env-1") + downloaded = Path(targets[0]["details"]["target_spec"]) + copy_file = shutil.copy2 + + def fail_copy(*_args: Any, **_kwargs: Any) -> None: + raise OSError("staging unavailable") + + monkeypatch.setattr(shutil, "copy2", fail_copy) + with pytest.raises(OSError, match="staging unavailable"): + stage_api_specs(targets, "failed-staging") + assert downloaded.is_file() + monkeypatch.setattr(shutil, "copy2", copy_file) + stage_api_specs(targets, "failed-staging") + assert not downloaded.parent.exists()