diff --git a/strix/interface/utils.py b/strix/interface/utils.py index 7ab55d8ee..9dbcee163 100644 --- a/strix/interface/utils.py +++ b/strix/interface/utils.py @@ -5,15 +5,18 @@ import os import re import secrets import shutil +import stat import subprocess import sys import tempfile +from collections.abc import Iterator from dataclasses import dataclass, field -from pathlib import Path -from typing import Any -from urllib.parse import parse_qs, urlparse +from pathlib import Path, PurePosixPath +from typing import Any, cast +from urllib.parse import parse_qs, unquote_to_bytes, urlparse, urlsplit import requests +import yaml from rich.console import Console from rich.panel import Panel from rich.text import Text @@ -1500,6 +1503,313 @@ 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" +_API_SPEC_EXTENSIONS = frozenset({".json", ".yaml", ".yml"}) +_API_SPEC_MAX_FILES = 100 +_API_SPEC_MAX_TOTAL_BYTES = 25 * 1024 * 1024 +_API_SPEC_MAX_REF_DEPTH = 25 +_API_SPEC_MAX_DOCUMENT_NODES = 100_000 + + +@dataclass(frozen=True) +class _ApiSpecBundle: + main_name: str + files: dict[Path, bytes] + + +def _safe_api_spec_label(path: Path) -> str: + label = re.sub(r"[^A-Za-z0-9._-]", "?", path.name) + return label[:80] or "referenced file" + + +def _parse_api_spec_document(content: bytes, label: str) -> Any: + try: + text = content.decode("utf-8") + except UnicodeDecodeError as exc: + raise ValueError(f"API spec file '{label}' must contain UTF-8 text.") from exc + try: + return json.loads(text) + except (json.JSONDecodeError, RecursionError): + try: + return yaml.safe_load(text) + except (yaml.YAMLError, RecursionError) as exc: + raise ValueError(f"API spec file '{label}' is not valid JSON or YAML.") from exc + + +def _iter_api_spec_refs(document: Any, label: str) -> Iterator[str]: + stack = [document] + seen_containers: set[int] = set() + nodes = 0 + while stack: + node = stack.pop() + nodes += 1 + if nodes > _API_SPEC_MAX_DOCUMENT_NODES: + raise ValueError(f"API spec file '{label}' is too complex to stage safely.") + if isinstance(node, dict): + mapping = cast("dict[str, Any]", node) + identity = id(mapping) + if identity in seen_containers: + continue + seen_containers.add(identity) + reference = mapping.get("$ref") + if isinstance(reference, str): + yield reference + stack.extend(mapping.values()) + elif isinstance(node, list): + sequence = cast("list[Any]", node) # type: ignore[redundant-cast] + identity = id(sequence) + if identity in seen_containers: + continue + seen_containers.add(identity) + stack.extend(sequence) + + +def _path_has_symlink(root: Path, path: Path) -> bool: + current = root + if current.is_symlink(): + return True + for part in path.relative_to(root).parts: + current /= part + if current.is_symlink(): + return True + return False + + +def _read_api_spec_file(path: Path, root: Path, remaining_bytes: int) -> bytes: + label = _safe_api_spec_label(path) + if _path_has_symlink(root, path): + raise ValueError(f"API spec file '{label}' is a symbolic link and cannot be staged.") + flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0) + try: + descriptor = os.open(path, flags) + except OSError as exc: + raise ValueError(f"Cannot safely read referenced API spec file '{label}'.") from exc + try: + file_stat = os.fstat(descriptor) + if not stat.S_ISREG(file_stat.st_mode): + raise ValueError(f"Referenced API spec file '{label}' is not a regular file.") + with os.fdopen(descriptor, "rb", closefd=False) as stream: + content = stream.read(remaining_bytes + 1) + finally: + os.close(descriptor) + if len(content) > remaining_bytes: + raise ValueError( + f"API spec dependencies exceed the {_API_SPEC_MAX_TOTAL_BYTES}-byte staging limit." + ) + return content + + +def _resolve_local_api_spec_ref(reference: str, referrer: Path, root: Path) -> Path | None: + try: + parsed = urlsplit(reference) + except ValueError as exc: + raise ValueError( + f"API spec file '{_safe_api_spec_label(referrer)}' has an invalid reference URI." + ) from exc + if parsed.scheme or parsed.netloc: + if parsed.scheme in {"http", "https"} and parsed.netloc: + return None + raise ValueError( + f"API spec file '{_safe_api_spec_label(referrer)}' uses an unsupported reference URI." + ) + if not parsed.path: + return None + if parsed.query: + raise ValueError( + f"API spec file '{_safe_api_spec_label(referrer)}' uses an unsupported local reference." + ) + try: + decoded = unquote_to_bytes(parsed.path).decode("utf-8") + except UnicodeDecodeError as exc: + raise ValueError( + f"API spec file '{_safe_api_spec_label(referrer)}' has an invalid local reference." + ) from exc + if not decoded or "\x00" in decoded or "\\" in decoded: + raise ValueError( + f"API spec file '{_safe_api_spec_label(referrer)}' has an invalid local reference." + ) + relative = PurePosixPath(decoded) + if relative.is_absolute(): + raise ValueError( + f"A local reference from '{_safe_api_spec_label(referrer)}' points outside the " + "API spec directory. Bundle the specification under one directory." + ) + if relative.suffix.lower() not in _API_SPEC_EXTENSIONS: + raise ValueError( + f"Local API spec dependency '{_safe_api_spec_label(Path(relative.name))}' must be " + "a JSON or YAML file." + ) + + candidate = referrer.parent.joinpath(*relative.parts).absolute() + if not candidate.is_relative_to(root): + raise ValueError( + f"A local reference from '{_safe_api_spec_label(referrer)}' points outside the " + "API spec directory. Bundle the specification under one directory." + ) + if _path_has_symlink(root, candidate): + raise ValueError( + f"API spec file '{_safe_api_spec_label(candidate)}' is a symbolic link and " + "cannot be staged." + ) + try: + resolved = candidate.resolve(strict=True) + except OSError as exc: + raise ValueError( + f"Referenced API spec file '{_safe_api_spec_label(candidate)}' does not exist " + "or cannot be read." + ) from exc + if not resolved.is_relative_to(root.resolve()): + raise ValueError( + f"A local reference from '{_safe_api_spec_label(referrer)}' points outside the " + "API spec directory. Bundle the specification under one directory." + ) + return resolved + + +def _collect_api_spec_bundle( + source: Path, + *, + follow_refs: bool, + available_files: int, + available_bytes: int, +) -> _ApiSpecBundle: + source = source.absolute() + if source.is_symlink(): + raise ValueError( + f"API spec file '{_safe_api_spec_label(source)}' is a symbolic link and " + "cannot be staged." + ) + try: + root = source.parent.resolve(strict=True) + source = source.resolve(strict=True) + except OSError as exc: + raise ValueError( + f"Cannot safely read API spec file '{_safe_api_spec_label(source)}'." + ) from exc + if source.suffix.lower() not in _API_SPEC_EXTENSIONS: + raise ValueError("API specifications and their dependencies must be JSON or YAML files.") + + queued: list[tuple[Path, int]] = [(source, 0)] + visited: set[Path] = set() + files: dict[Path, bytes] = {} + total_bytes = 0 + while queued: + path, depth = queued.pop() + if path in visited: + continue + if depth > _API_SPEC_MAX_REF_DEPTH: + raise ValueError( + f"API spec reference depth exceeds the {_API_SPEC_MAX_REF_DEPTH}-level limit." + ) + if len(visited) >= available_files: + raise ValueError( + f"API spec dependencies exceed the {_API_SPEC_MAX_FILES}-file staging limit." + ) + content = _read_api_spec_file(path, root, available_bytes - total_bytes) + total_bytes += len(content) + visited.add(path) + relative_path = path.relative_to(root) + files[relative_path] = content + if not follow_refs: + continue + label = _safe_api_spec_label(relative_path) + document = _parse_api_spec_document(content, label) + for reference in _iter_api_spec_refs(document, label): + dependency = _resolve_local_api_spec_ref(reference, path, root) + if dependency is not None and dependency not in visited: + queued.append((dependency, depth + 1)) + return _ApiSpecBundle(main_name=source.name, files=files) + + +def _unique_api_spec_name(name: str, used: set[str]) -> str: + stem, suffix = Path(name).stem, Path(name).suffix + candidate = name + count = 1 + while candidate.casefold() in used: + count += 1 + candidate = f"{stem}-{count}{suffix}" + used.add(candidate.casefold()) + return candidate + + +def _collect_api_spec_bundles(specs: list[dict[str, Any]]) -> list[_ApiSpecBundle]: + bundles: list[_ApiSpecBundle] = [] + remaining_files = _API_SPEC_MAX_FILES + remaining_bytes = _API_SPEC_MAX_TOTAL_BYTES + for target in specs: + bundle = _collect_api_spec_bundle( + Path(str(target["details"]["target_spec"])), + follow_refs=target["details"].get("spec_format") in {"openapi", "swagger"}, + available_files=remaining_files, + available_bytes=remaining_bytes, + ) + bundles.append(bundle) + remaining_files -= len(bundle.files) + remaining_bytes -= sum(len(content) for content in bundle.files.values()) + return bundles + + +def _layout_api_spec_bundles( + bundles: list[_ApiSpecBundle], +) -> list[tuple[Path, dict[Path, bytes]]]: + used_names: set[str] = set() + allocated_names = [_unique_api_spec_name(bundle.main_name, used_names) for bundle in bundles] + used_root_entries = { + name.casefold() + for name, bundle in zip(allocated_names, bundles, strict=True) + if len(bundle.files) == 1 + } + layouts: list[tuple[Path, dict[Path, bytes]]] = [] + for name, bundle in zip(allocated_names, bundles, strict=True): + if len(bundles) == 1 or len(bundle.files) == 1: + main_path = Path(name) + files = ( + {main_path: next(iter(bundle.files.values()))} + if len(bundle.files) == 1 + else dict(bundle.files) + ) + else: + namespace = _unique_api_spec_name(Path(name).stem, used_root_entries) + main_path = Path(namespace) / bundle.main_name + files = { + Path(namespace) / relative: content for relative, content in bundle.files.items() + } + layouts.append((main_path, files)) + return layouts + + +def _publish_api_spec_layouts(layouts: list[tuple[Path, dict[Path, bytes]]], run_name: str) -> Path: + temporary_root = Path(tempfile.gettempdir()) + try: + temporary_root.mkdir(parents=True, exist_ok=True) + temporary_root = temporary_root.resolve(strict=True) + staging_parent = temporary_root / "strix_api_specs" + if staging_parent.is_symlink(): + raise ValueError("API specification staging directory cannot be a symbolic link.") + staging_parent.mkdir(parents=True, exist_ok=True) + if staging_parent.resolve(strict=True) != staging_parent: + raise ValueError("API specification staging directory is not safe to use.") + except OSError as exc: + raise ValueError("Cannot create a safe API specification staging directory.") from exc + + staging = staging_parent / run_name + temporary = Path(tempfile.mkdtemp(prefix=f".{run_name}-", dir=staging_parent)) + try: + for _main_path, files in layouts: + for relative, content in files.items(): + destination = temporary / relative + destination.parent.mkdir(parents=True, exist_ok=True) + destination.write_bytes(content) + if staging.is_symlink() or staging.is_file(): + staging.unlink() + elif staging.exists(): + shutil.rmtree(staging) + temporary.rename(staging) + except Exception: + if temporary.exists(): + shutil.rmtree(temporary) + raise + return staging + def write_fetched_collection(collection: dict[str, Any], collection_uid: str) -> str: """Write a collection fetched from the Postman API to a local file. @@ -1515,33 +1825,29 @@ def write_fetched_collection(collection: dict[str, Any], collection_uid: str) -> def stage_api_specs(targets_info: list[dict[str, Any]], run_name: str) -> list[dict[str, Any]]: - """Copy every ``api_spec`` target into one directory for the sandbox. + """Stage API specs and their bounded local ``$ref`` closure for the sandbox. - A spec is a single file the agent reads, not a tree it works in, so it is - copied to a per-run staging directory that is exposed at - ``/workspace/api-specs`` instead of mounting its host location. Each target's - ``workspace_path`` records where the agent will find it. + Only explicit relative JSON/YAML references below each spec directory are + followed. Remote references remain unchanged, and nothing is published until + every local dependency has been validated and copied. """ specs = [t for t in targets_info if t.get("type") == "api_spec"] if not specs: return [] - - staging = Path(tempfile.gettempdir()) / "strix_api_specs" / run_name - staging.mkdir(parents=True, exist_ok=True) - - used: set[str] = set() for target in specs: - details = target["details"] - source = Path(str(details["target_spec"])) - name = source.name - stem, suffix = source.stem, source.suffix - count = 1 - while name in used: - count += 1 - name = f"{stem}-{count}{suffix}" - used.add(name) - shutil.copy2(source, staging / name) - details["workspace_path"] = f"/workspace/{API_SPEC_WORKSPACE_SUBDIR}/{name}" + target["details"].pop("workspace_path", None) + + safe_run_name = sanitize_name(run_name) + if safe_run_name != run_name or safe_run_name in {".", ".."}: + raise ValueError("Invalid run name for API specification staging.") + + layouts = _layout_api_spec_bundles(_collect_api_spec_bundles(specs)) + staging = _publish_api_spec_layouts(layouts, safe_run_name) + + for target, (main_path, _files) in zip(specs, layouts, strict=True): + target["details"]["workspace_path"] = ( + f"/workspace/{API_SPEC_WORKSPACE_SUBDIR}/{main_path.as_posix()}" + ) return [ { diff --git a/tests/test_api_spec_targets.py b/tests/test_api_spec_targets.py index bba0aba55..6cf4c3dd8 100644 --- a/tests/test_api_spec_targets.py +++ b/tests/test_api_spec_targets.py @@ -36,6 +36,23 @@ def _write_spec(directory: Path, name: str = "openapi.json") -> Path: return path +def _write_spec_with_ref(directory: Path, reference: str, name: str = "openapi.json") -> Path: + spec = { + **OPENAPI, + "components": {"schemas": {"User": {"$ref": reference}}}, + } + directory.mkdir(parents=True, exist_ok=True) + path = directory / name + path.write_text(json.dumps(spec), encoding="utf-8") + return path + + +@pytest.fixture(autouse=True) +def _isolate_staging(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + staging_root = tmp_path / "temporary" + monkeypatch.setattr("strix.interface.utils.tempfile.gettempdir", lambda: str(staging_root)) + + def _resolved_targets(*spec_paths: Path) -> list[dict[str, Any]]: """Run spec targets through the real setup path (detection + spec resolution).""" args = argparse.Namespace(target=[str(p) for p in spec_paths], target_list=None) @@ -105,12 +122,14 @@ def test_build_targets_info_rejects_unparseable_spec(tmp_path: Path) -> None: def test_stage_api_specs_copies_spec_into_workspace_dir(tmp_path: Path) -> None: - targets = _resolved_targets(_write_spec(tmp_path / "src")) + original = _write_spec(tmp_path / "src") + targets = _resolved_targets(original) (source,) = stage_api_specs(targets, "stage-run") assert source["workspace_subdir"] == "api-specs" staged = Path(source["source_path"]) / "openapi.json" assert json.loads(staged.read_text(encoding="utf-8"))["info"]["title"] == "Shop API" + assert staged.read_bytes() == original.read_bytes() assert targets[0]["details"]["workspace_path"] == "/workspace/api-specs/openapi.json" @@ -129,6 +148,256 @@ def test_stage_api_specs_disambiguates_same_filename(tmp_path: Path) -> None: assert (Path(source["source_path"]) / "openapi-2.json").is_file() +def test_stage_api_specs_copies_transitive_json_and_yaml_refs(tmp_path: Path) -> None: + source_dir = tmp_path / "src" + spec = _write_spec_with_ref(source_dir, "./schemas/User.json#/definitions/User") + schemas = source_dir / "schemas" + schemas.mkdir() + (schemas / "User.json").write_text( + json.dumps({"definitions": {"User": {"$ref": "../common.yaml#/User"}}}), + encoding="utf-8", + ) + (source_dir / "common.yaml").write_text("User:\n type: object\n", encoding="utf-8") + (source_dir / "unused.yaml").write_text("Secret: unreferenced\n", encoding="utf-8") + + targets = _resolved_targets(spec) + (staged_source,) = stage_api_specs(targets, "relative-refs") + staged = Path(staged_source["source_path"]) + + assert (staged / "openapi.json").is_file() + assert (staged / "schemas" / "User.json").is_file() + assert (staged / "common.yaml").is_file() + assert not (staged / "unused.yaml").exists() + assert (staged / "openapi.json").read_bytes() == spec.read_bytes() + assert staged_source["workspace_subdir"] == "api-specs" + assert targets[0]["details"]["workspace_path"] == "/workspace/api-specs/openapi.json" + + +def test_stage_api_specs_terminates_local_ref_cycles(tmp_path: Path) -> None: + source_dir = tmp_path / "src" + spec = _write_spec_with_ref(source_dir, "schemas/User.yaml#/User") + schemas = source_dir / "schemas" + schemas.mkdir() + (schemas / "User.yaml").write_text( + "User:\n allOf:\n - $ref: ../openapi.json#/components/schemas/User\n", + encoding="utf-8", + ) + + (staged_source,) = stage_api_specs(_resolved_targets(spec), "cyclic-refs") + staged = Path(staged_source["source_path"]) + + assert sorted(path.relative_to(staged) for path in staged.rglob("*") if path.is_file()) == [ + Path("openapi.json"), + Path("schemas/User.yaml"), + ] + + +def test_stage_api_specs_keeps_fragment_only_refs_internal(tmp_path: Path) -> None: + spec = _write_spec_with_ref(tmp_path / "src", "#/components/schemas/Local") + + (staged_source,) = stage_api_specs(_resolved_targets(spec), "fragment-ref") + staged = Path(staged_source["source_path"]) + + assert [path.relative_to(staged) for path in staged.rglob("*") if path.is_file()] == [ + Path("openapi.json") + ] + + +def test_stage_api_specs_does_not_fetch_remote_refs( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + spec = _write_spec_with_ref(tmp_path / "src", "https://schemas.example.test/User.json") + + def unexpected_fetch(*_args: object, **_kwargs: object) -> None: + raise AssertionError("remote reference was fetched") + + monkeypatch.setattr("strix.interface.utils.requests.get", unexpected_fetch) + (staged_source,) = stage_api_specs(_resolved_targets(spec), "remote-ref") + staged = Path(staged_source["source_path"]) + + assert [path.relative_to(staged) for path in staged.rglob("*") if path.is_file()] == [ + Path("openapi.json") + ] + + +def test_stage_api_specs_keeps_postman_collections_single_file(tmp_path: Path) -> None: + collection = { + "info": {"name": "Collection", "_postman_id": "fixture"}, + "item": [], + "metadata": {"$ref": "missing.yaml"}, + } + path = tmp_path / "collection.json" + path.write_text(json.dumps(collection), encoding="utf-8") + targets = _resolved_targets(path) + + (staged_source,) = stage_api_specs(targets, "postman-single-file") + staged = Path(staged_source["source_path"]) + + assert (staged / "collection.json").read_bytes() == path.read_bytes() + assert targets[0]["details"]["workspace_path"] == "/workspace/api-specs/collection.json" + + +def test_stage_api_specs_rejects_missing_local_ref_without_partial_stage(tmp_path: Path) -> None: + spec = _write_spec_with_ref(tmp_path / "src", "schemas/Missing.yaml#/User") + targets = _resolved_targets(spec) + + with pytest.raises(ValueError, match=r"Missing\.yaml"): + stage_api_specs(targets, "missing-ref") + + assert not (tmp_path / "temporary" / "strix_api_specs" / "missing-ref").exists() + assert "workspace_path" not in targets[0]["details"] + + +def test_stage_api_specs_rejects_ref_outside_spec_root(tmp_path: Path) -> None: + source_dir = tmp_path / "src" + spec = _write_spec_with_ref(source_dir, "../outside.yaml#/User") + (tmp_path / "outside.yaml").write_text("User: {type: object}\n", encoding="utf-8") + + with pytest.raises(ValueError, match="outside the API spec directory"): + stage_api_specs(_resolved_targets(spec), "escaping-ref") + + +def test_stage_api_specs_rejects_encoded_ref_escape(tmp_path: Path) -> None: + source_dir = tmp_path / "src" + spec = _write_spec_with_ref(source_dir, "..%2Foutside.yaml#/User") + (tmp_path / "outside.yaml").write_text("User: {type: object}\n", encoding="utf-8") + + with pytest.raises(ValueError, match="outside the API spec directory"): + stage_api_specs(_resolved_targets(spec), "encoded-escape") + + +def test_stage_api_specs_rejects_symlinked_ref(tmp_path: Path) -> None: + source_dir = tmp_path / "src" + spec = _write_spec_with_ref(source_dir, "schemas/User.yaml#/User") + schemas = source_dir / "schemas" + schemas.mkdir() + real_schema = schemas / "real.yaml" + real_schema.write_text("User: {type: object}\n", encoding="utf-8") + (schemas / "User.yaml").symlink_to(real_schema) + + with pytest.raises(ValueError, match="symbolic link"): + stage_api_specs(_resolved_targets(spec), "symlink-ref") + + +def test_stage_api_specs_isolates_colliding_multifile_specs(tmp_path: Path) -> None: + specs: list[Path] = [] + for directory, marker in (("a", "first"), ("b", "second")): + source_dir = tmp_path / directory + specs.append(_write_spec_with_ref(source_dir, "schemas/User.json")) + schemas = source_dir / "schemas" + schemas.mkdir() + (schemas / "User.json").write_text(json.dumps({"marker": marker}), encoding="utf-8") + + targets = _resolved_targets(*specs) + (staged_source,) = stage_api_specs(targets, "isolated-refs") + staged = Path(staged_source["source_path"]) + + assert [target["details"]["workspace_path"] for target in targets] == [ + "/workspace/api-specs/openapi/openapi.json", + "/workspace/api-specs/openapi-2/openapi.json", + ] + assert json.loads((staged / "openapi" / "schemas" / "User.json").read_text())["marker"] == ( + "first" + ) + assert ( + json.loads((staged / "openapi-2" / "schemas" / "User.json").read_text())["marker"] + == "second" + ) + + +def test_stage_api_specs_rejects_unsupported_local_ref_type(tmp_path: Path) -> None: + source_dir = tmp_path / "src" + spec = _write_spec_with_ref(source_dir, "schemas/User.txt") + schemas = source_dir / "schemas" + schemas.mkdir() + (schemas / "User.txt").write_text("not a JSON or YAML document", encoding="utf-8") + + with pytest.raises(ValueError, match="JSON or YAML"): + stage_api_specs(_resolved_targets(spec), "unsupported-ref") + + +def test_stage_api_specs_validates_all_targets_before_publishing(tmp_path: Path) -> None: + good = _write_spec(tmp_path / "good") + bad = _write_spec_with_ref(tmp_path / "bad", "missing.yaml") + targets = _resolved_targets(good, bad) + targets[0]["details"]["workspace_path"] = "/workspace/api-specs/stale.json" + + with pytest.raises(ValueError, match=r"missing\.yaml"): + stage_api_specs(targets, "atomic-stage") + + assert not (tmp_path / "temporary" / "strix_api_specs" / "atomic-stage").exists() + assert all("workspace_path" not in target["details"] for target in targets) + + +def test_stage_api_specs_enforces_file_limit( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + source_dir = tmp_path / "src" + spec = _write_spec_with_ref(source_dir, "one.yaml") + (source_dir / "one.yaml").write_text("$ref: two.yaml\n", encoding="utf-8") + (source_dir / "two.yaml").write_text("type: object\n", encoding="utf-8") + monkeypatch.setattr("strix.interface.utils._API_SPEC_MAX_FILES", 2) + + with pytest.raises(ValueError, match="2-file staging limit"): + stage_api_specs(_resolved_targets(spec), "file-limit") + + +def test_stage_api_specs_enforces_limits_across_targets( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + specs = [_write_spec(tmp_path / str(index)) for index in range(3)] + monkeypatch.setattr("strix.interface.utils._API_SPEC_MAX_FILES", 2) + + with pytest.raises(ValueError, match="2-file staging limit"): + stage_api_specs(_resolved_targets(*specs), "aggregate-file-limit") + + +def test_stage_api_specs_enforces_depth_limit( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + source_dir = tmp_path / "src" + spec = _write_spec_with_ref(source_dir, "one.yaml") + (source_dir / "one.yaml").write_text("$ref: two.yaml\n", encoding="utf-8") + (source_dir / "two.yaml").write_text("type: object\n", encoding="utf-8") + monkeypatch.setattr("strix.interface.utils._API_SPEC_MAX_REF_DEPTH", 1) + + with pytest.raises(ValueError, match="1-level limit"): + stage_api_specs(_resolved_targets(spec), "depth-limit") + + +def test_stage_api_specs_enforces_total_byte_limit( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + spec = _write_spec(tmp_path / "src") + monkeypatch.setattr("strix.interface.utils._API_SPEC_MAX_TOTAL_BYTES", 10) + + with pytest.raises(ValueError, match="10-byte staging limit"): + stage_api_specs(_resolved_targets(spec), "byte-limit") + + +def test_stage_api_specs_rejects_unsafe_run_name(tmp_path: Path) -> None: + targets = _resolved_targets(_write_spec(tmp_path / "src")) + + with pytest.raises(ValueError, match="Invalid run name"): + stage_api_specs(targets, "../outside") + + assert not (tmp_path / "outside").exists() + + +def test_stage_api_specs_rejects_symlinked_staging_directory(tmp_path: Path) -> None: + source = _write_spec(tmp_path / "src") + temporary_root = tmp_path / "temporary" + temporary_root.mkdir() + outside = tmp_path / "outside" + outside.mkdir() + (temporary_root / "strix_api_specs").symlink_to(outside, target_is_directory=True) + + with pytest.raises(ValueError, match="staging directory cannot be a symbolic link"): + stage_api_specs(_resolved_targets(source), "symlinked-stage") + + assert list(outside.iterdir()) == [] + + def test_stage_api_specs_without_specs_returns_nothing() -> None: assert stage_api_specs([{"type": "web_application", "details": {}}], "run") == []