mirror of
https://github.com/usestrix/strix.git
synced 2026-11-14 09:01:25 +00:00
fix(api-spec): preserve referenced Postman environment values
This commit is contained in:
parent
6b7ae2ebe4
commit
34e67e5578
5 changed files with 231 additions and 4 deletions
|
|
@ -21,7 +21,7 @@ strix (--target <target> | --target-list <path>) [options]
|
|||
</Note>
|
||||
|
||||
<Note>
|
||||
Fetching a Postman collection by id requires `POSTMAN_API_KEY`. Add `?env=<environment-uuid>` to also pull a Postman environment, which resolves `{{baseUrl}}` / token variables the collection references (e.g. `postman://<collection-uuid>?env=<environment-uid>`).
|
||||
Fetching a Postman collection by id requires `POSTMAN_API_KEY`. Add `?env=<environment-uuid>` to also pull a Postman environment (e.g. `postman://<collection-uuid>?env=<environment-uid>`). 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.
|
||||
</Note>
|
||||
</ParamField>
|
||||
|
||||
|
|
|
|||
|
|
@ -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"]))
|
||||
|
|
|
|||
|
|
@ -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__)
|
||||
|
|
@ -1509,8 +1510,11 @@ 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 = Path(tempfile.mkdtemp(prefix="collection-", dir=staging))
|
||||
path = snapshot / f"{sanitize_name(collection_uid)}.postman_collection.json"
|
||||
write_secret_text(path, json.dumps(collection, indent=2))
|
||||
return str(path)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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],
|
||||
*,
|
||||
|
|
|
|||
164
tests/test_postman_environment_handoff.py
Normal file
164
tests/test_postman_environment_handoff.py
Normal file
|
|
@ -0,0 +1,164 @@
|
|||
"""Postman environments must survive target resolution and sandbox staging."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import copy
|
||||
import json
|
||||
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"
|
||||
Loading…
Add table
Reference in a new issue