From ba76057720d769d2602da8ecbd1d6b8a1da22d91 Mon Sep 17 00:00:00 2001 From: Arseniy Dugin Date: Mon, 13 Jul 2026 13:43:28 +0000 Subject: [PATCH] feat(artifacts): list and download pipeline-run artifacts Extend tangle sdk artifacts get with list, out-dir, only and include-children modes. Inline values are JSON-encoded; signed-URL content streams to owner-only (POSIX 0o600) files with path, collision, symlink, redirect and internal-address checks. Signed URLs are redacted from errors. Query top-level and nested shapes are validated as clean errors, null per-entry output filters select all outputs (same as an empty list), and an explicit empty --query object stays distinct from an absent one. A public TangleApiClient.request_raw seam keeps artifact fetching independent of private client methods. --- .../tangle-cli/src/tangle_cli/artifacts.py | 780 ++++++++- .../src/tangle_cli/artifacts_cli.py | 293 +++- packages/tangle-cli/src/tangle_cli/client.py | 18 + tests/test_artifacts.py | 1450 ++++++++++++++++- tests/test_artifacts_cli.py | 466 ++++++ tests/test_client.py | 15 + 6 files changed, 2980 insertions(+), 42 deletions(-) diff --git a/packages/tangle-cli/src/tangle_cli/artifacts.py b/packages/tangle-cli/src/tangle_cli/artifacts.py index 438bfab..2949c32 100644 --- a/packages/tangle-cli/src/tangle_cli/artifacts.py +++ b/packages/tangle-cli/src/tangle_cli/artifacts.py @@ -1,19 +1,98 @@ -"""Read-only artifact lookup helpers for Tangle pipeline runs. +"""Artifact lookup and download helpers for Tangle pipeline runs. -This module intentionally resolves artifact metadata only. It does not fetch -signed URLs, download remote objects, write local files, or mutate artifacts. +This module resolves artifact metadata and can fetch artifact contents (direct +data route, inline metadata value, or signed URL) to a caller-provided +directory. It never mutates or deletes artifacts. """ from __future__ import annotations +import ipaddress +import json +import os +import re +from collections.abc import Iterable from dataclasses import asdict, dataclass, field, is_dataclass -from typing import Any, Protocol +from pathlib import Path +from typing import Any, BinaryIO, Protocol +from urllib.parse import quote, urljoin, urlparse + +import requests from .handler import TangleCliHandler +_SAFE_NAME_RE = re.compile(r"[^A-Za-z0-9._-]+") + +# Artifact bytes are streamed to disk in fixed-size chunks so that large +# artifacts are never fully buffered in memory. +_DOWNLOAD_CHUNK_SIZE = 1024 * 1024 + +# ``O_NOFOLLOW`` is absent on some platforms (e.g. Windows); fall back to 0 +# there. ``O_EXCL`` alone still refuses pre-existing symlinks on POSIX. +_O_NOFOLLOW = getattr(os, "O_NOFOLLOW", 0) + +# Status codes from the direct ``/data`` route that mean "the bytes are not +# available here, try another path" rather than a hard failure: 404 (no direct +# data route for this artifact), 403 (direct download forbidden — bytes live +# behind a signed URL), and 410 (direct route gone). For all of these the +# inline metadata ``value`` or a signed URL is the correct fallback. +_DATA_FALLBACK_STATUS_CODES = frozenset({403, 404, 410}) + +# Redirect status codes. The direct ``/data`` route can 3xx-redirect to a +# cross-origin storage URL (e.g. GCS/S3). The authenticated client refuses to +# forward Tangle credentials off-origin and raises an ``HTTPError`` carrying the +# redirect response, so a redirect status observed on that pre-response error +# means "the bytes live off-origin, fall back to the signed URL" rather than a +# terminal failure. +_REDIRECT_STATUS_CODES = frozenset({301, 302, 303, 307, 308}) + +# Redirect-follow bound for absolute signed-URL downloads. Redirects there are +# followed manually (never automatically by ``requests``) so every hop can be +# rechecked against the http(s) allowlist before it is fetched. +_MAX_SIGNED_URL_REDIRECTS = 5 + +# Connect/read timeout for each unauthenticated signed-URL fetch (per hop). +_SIGNED_URL_TIMEOUT_SECONDS = 60.0 + +# Absolute signed URLs (and their redirect targets) pointing at internal hosts +# — loopback, link-local (incl. the cloud metadata IP), RFC1918/unique-local +# private ranges, and the unspecified address — are rejected by default so a +# hostile signed URL cannot aim the CLI at services reachable only from the +# local network (SSRF). Self-hosted/local deployments legitimately serve +# loopback signed URLs; this env var opts back in. +_ALLOW_INTERNAL_HOSTS_ENV = "TANGLE_ALLOW_INTERNAL_ARTIFACT_HOSTS" + +# The socket layer (``inet_aton`` semantics via ``getaddrinfo``) accepts IPv4 +# spellings that ``ipaddress.ip_address`` rejects: pure decimal (2130706433), +# octal (0177.0.0.1), hex (0x7f000001, 0x7f.0.0.1), and dotted short forms +# (127.1) — each of which can smuggle a loopback/metadata address past a +# canonical-literal check. Any host made only of such numeric labels that is +# *not* a canonical IP literal is rejected outright rather than re-implementing +# ``inet_aton`` value semantics: no legitimate signed URL spells its host that +# way, and an all-numeric label is not a valid DNS name. +_INET_ATON_NUMERIC_HOST_RE = re.compile(r"(?i)^(0x[0-9a-f]*|\d+)(\.(0x[0-9a-f]*|\d+)){0,3}$") + +# NAT64 translation prefixes (RFC 6052 well-known 64:ff9b::/96 and RFC 8215 +# local-use 64:ff9b:1::/48) embed an IPv4 address in the low 32 bits; a NAT64 +# gateway would connect to that embedded IPv4, so it is what gets classified. +_NAT64_PREFIXES = ( + ipaddress.ip_network("64:ff9b::/96"), + ipaddress.ip_network("64:ff9b:1::/48"), +) + +# Carrier-grade NAT shared address space (RFC 6598). ``is_private`` is False +# for it (it is neither private nor globally routable), so it needs its own +# membership check. +_CGNAT_NETWORK = ipaddress.ip_network("100.64.0.0/10") + +# Sentinel distinguishing an absent inline ``value`` field (no bytes here — fall +# back to the signed URL) from an explicit inline ``null`` (a real value written +# as JSON ``null``). ``None`` alone cannot tell these apart. +_MISSING = object() + class ArtifactClient(Protocol): - """Subset of the static API client used for read-only artifact lookup.""" + """Subset of the static API client used for artifact lookup and download.""" def get_run_details(self, run_id: str) -> Any: ... @@ -21,6 +100,10 @@ def get_execution_details(self, execution_id: str) -> Any: ... def artifacts_get(self, artifact_id: str) -> Any: ... + def request_raw(self, method: str, path: str, **kwargs: Any) -> Any: ... + + def artifacts_signed_artifact_url(self, artifact_id: str) -> Any: ... + @dataclass class ArtifactComponentQuery: @@ -82,12 +165,13 @@ def from_response(cls, response: Any, *, key: str = "") -> ArtifactInfo: class ArtifactManager(TangleCliHandler): - """Read-only artifact metadata manager. + """Artifact metadata, listing, and download manager. Downstream packages can inject an already-authenticated client or a lazy - ``client_factory`` (for example, one that applies provider auth). The manager - keeps the same read-only constraints as the module-level helpers: it never - downloads artifact contents, signs URLs, writes files, or mutates artifacts. + ``client_factory`` (for example, one that applies provider auth). In addition + to resolving metadata, the manager can download artifact contents (direct + data route, inline metadata value, or signed URL) and write them to a + caller-provided directory. It never mutates or deletes artifacts. """ def __init__( @@ -103,7 +187,7 @@ def __init__( def collect_artifacts( self, execution: Any, - tasks_query: dict[str, list[str]], + tasks_query: dict[str, list[str] | None], components_query: list[ArtifactComponentQuery], prefix: str = "", ) -> dict[str, str]: @@ -122,7 +206,9 @@ def collect_artifacts( for query_name in (task_name, key_prefix): if query_name in tasks_query: - output_filters.append(tasks_query[query_name]) + # A ``null`` filter means all outputs, same as ``[]`` — and + # the same as ``executions``/``components`` null filters. + output_filters.append(tasks_query[query_name] or []) break child_digest = _mapping_or_attr(child_task, "digest") @@ -162,7 +248,7 @@ def collect_artifacts( def collect_execution_artifacts( self, - execution_ids: dict[str, list[str]], + execution_ids: dict[str, list[str] | None], ) -> dict[str, str]: """Collect artifact IDs directly from execution IDs.""" @@ -189,11 +275,19 @@ def get_artifacts( - ``executions``: ``{: []}`` - ``artifact_ids``: ``[, ...]`` - Empty output lists mean all outputs. Per-artifact lookup failures are - returned as ``ArtifactInfo(error=...)`` entries instead of failing the - whole command. + Empty (or ``null``) output lists mean all outputs. Per-artifact + lookup failures are returned as ``ArtifactInfo(error=...)`` entries + instead of failing the whole command. """ + # ``--query`` must be a JSON object; valid JSON that is not an object + # (a list, string, or number) is rejected as a concise CLI error before + # any client/network work, and nested key shapes are checked the same + # way so a malformed inner value cannot raise from mid-walk. + if not isinstance(query, dict): + raise RuntimeError("--query must be a JSON object") + _validate_query_shape(query) + artifact_ids: dict[str, str] = {} for artifact_id in query.get("artifact_ids", []) or []: @@ -228,6 +322,348 @@ def get_artifacts( return artifacts + def _resolve_root_execution(self, run_id: str) -> Any: + """Resolve a pipeline run id to its root execution tree. + + Artifact listing/download operate on the run's *root execution*, but the + CLI (like the ``--query`` path) accepts a *run* id; run ids and execution + ids are distinct namespaces. ``get_run_details`` maps the run id to its + root execution (falling back to treating the id as an execution id when + no run record exists), so the artifact endpoints always receive the + execution data even when the run and root-execution ids differ — instead + of passing the run id straight to ``get_execution_details``. + """ + + details = self._require_client().get_run_details(run_id) + execution = _mapping_or_attr(details, "execution") + if not execution: + raise RuntimeError(f"No execution details found for run {run_id}") + return execution + + def list_result_artifacts( + self, + run_id: str, + *, + include_children: bool = False, + ) -> list[dict[str, str]]: + """List root output artifacts and optionally direct child outputs.""" + + try: + root = self._resolve_root_execution(run_id) + root_artifacts = _artifact_id_map(_mapping_or_attr(root, "output_artifacts", {})) + rows = [ + {"owner": "root", "output": output_name, "artifact_id": artifact_id} + for output_name, artifact_id in root_artifacts.items() + ] + if not include_children: + return rows + + child_executions = _direct_child_executions(root) + if not child_executions: + client = self._require_client() + child_executions = { + task_name: client.get_execution_details(execution_id) + for task_name, execution_id in _child_execution_ids(root).items() + } + except (RuntimeError, requests.HTTPError): + # Already clean / formatted by the CLI's HTTP-error surfacing. + raise + except Exception as exc: + # Any other client/transport failure would escape the CLI's + # RuntimeError handler as a raw traceback. Surface it concisely. + raise RuntimeError( + f"Failed to list artifacts for run {run_id}: " + f"{_http_error_detail(_exc_status(exc), exc)}" + ) from exc + + rows.extend( + {"owner": task_name, "output": output_name, "artifact_id": artifact_id} + for task_name, child in child_executions.items() + for output_name, artifact_id in _artifact_id_map( + _mapping_or_attr(child, "output_artifacts", {}) + ).items() + ) + return rows + + def download_result_artifacts( + self, + run_id: str, + *, + out_dir: str | Path, + only: Iterable[str] | None = None, + include_children: bool = False, + ) -> dict[str, Path]: + """Download root output artifacts and optionally direct child outputs.""" + + out_dir_path = Path(out_dir) + try: + out_dir_path.mkdir(parents=True, exist_ok=True) + except OSError as exc: + # --out-dir pointing at an existing file (FileExistsError / + # NotADirectoryError) or an otherwise uncreatable path would + # otherwise raise a raw traceback past the CLI's RuntimeError handler. + raise RuntimeError( + f"Cannot create output directory {out_dir_path}: " + f"{exc.strerror or exc}" + ) from exc + only_set = set(only) if only else None + results: dict[str, Path] = {} + seen: dict[str, Path] = {} + # Distinct artifact ids can sanitize/truncate to the same filename + # (``artifact_id[:12]``). Track each claimed filename so a collision + # between *different* artifacts fails cleanly instead of silently + # overwriting an already-downloaded file. + claimed: dict[str, str] = {} + + for row in self.list_result_artifacts( + run_id, + include_children=include_children, + ): + output_name = row["output"] + if only_set is not None and output_name not in only_set: + continue + artifact_id = row["artifact_id"] + path = seen.get(artifact_id) + if path is None: + filename = _artifact_filename(row["owner"], output_name, artifact_id) + prior = claimed.get(filename) + if prior is not None and prior != artifact_id: + raise RuntimeError( + f"Refusing to download artifact {artifact_id}: target " + f"filename {filename!r} already used by artifact {prior}" + ) + path = self._download_artifact_to(artifact_id, out_dir_path / filename) + claimed[filename] = artifact_id + seen[artifact_id] = path + results[f"{row['owner']}::{output_name}"] = path + return results + + def _download_artifact_to(self, artifact_id: str, dest: Path) -> Path: + """Stream artifact bytes to ``dest``, returning the path written. + + Tries the direct ``/data`` route first; on a fallback status the bytes + come from an inline metadata value (JSON-encoded, written with a + ``.json`` suffix appended to ``dest``) or a signed URL (raw bytes, + written to ``dest`` as-is). The destination is opened exclusively and + without following symlinks, so an existing file/dir/symlink at the + target fails cleanly rather than being overwritten or followed. + """ + + client = self._require_client() + request = getattr(client, "request_raw", None) + if not callable(request): + raise RuntimeError("Artifact downloads require a client with raw request support") + endpoint = _data_endpoint(artifact_id) + try: + response = request("GET", endpoint, stream=True) + except requests.RequestException as exc: + status = _exc_status(exc) + if status in _REDIRECT_STATUS_CODES: + # The direct data route redirected to cross-origin storage; the + # client refused to forward Tangle credentials off-origin and + # raised before returning a body. Fall back to the signed-URL + # path (fetched unauthenticated) instead of stopping at the 302. + self._download_signed_artifact_to(artifact_id, dest, exc, status) + return dest + # Transport failure before a response object exists (DNS/TLS/ + # timeout/connection) on the direct data route. Without this the + # raw requests exception would escape the CLI's RuntimeError handler. + raise RuntimeError( + f"Artifact download failed: GET {endpoint} transport failed: " + f"{_http_error_detail(status, exc)}" + ) from exc + try: + try: + response.raise_for_status() + except Exception as exc: + status = _response_status_code(response) + if status not in _DATA_FALLBACK_STATUS_CODES: + # Surface as a concise CLI error instead of a raw requests + # traceback (e.g. a 500 on the direct data route). + raise RuntimeError( + f"Artifact download failed: GET {endpoint} returned " + f"{_http_error_detail(status, exc)}" + ) from exc + metadata_value = self._artifact_value_bytes(artifact_id) + if metadata_value is not None: + json_dest = dest.with_name(dest.name + ".json") + _write_new_file_bytes(json_dest, metadata_value) + return json_dest + self._download_signed_artifact_to(artifact_id, dest, exc, status) + return dest + try: + _stream_response_to_file(response, dest) + except requests.RequestException as exc: + # A transport failure mid-stream (e.g. a dropped connection) + # would otherwise escape as a raw requests traceback. + raise RuntimeError( + f"Artifact download failed: GET {endpoint} streaming failed: " + f"{_http_error_detail(_exc_status(exc), exc)}" + ) from exc + finally: + _close_quietly(response) + return dest + + def _artifact_value_bytes(self, artifact_id: str) -> bytes | None: + try: + artifact = self._require_client().artifacts_get(artifact_id) + except Exception: + # Fetching the inline value is a best-effort step in the fallback + # chain (direct /data -> inline value -> signed URL). Any failure + # here means "no inline value from this step"; the caller then tries + # the signed URL, which is the authoritative final attempt and + # surfaces its own clean error if it also fails. Narrowing this catch + # would turn a recoverable inline-fetch failure into a hard error and + # skip the signed-URL fallback. + return None + artifact_data = _mapping_or_attr(artifact, "artifact_data") + value = _mapping_or_attr(artifact_data, "value", _MISSING) + if value is None and not isinstance(artifact_data, dict): + # The shipped generated client deserializes ``artifact_data`` as a + # plain dict (the response field is typed ``Any``), so ``dict.get`` + # above already tells absent from explicit null. A typed model + # shaped like the generated ``ArtifactData`` defaults ``value`` to + # ``None``, which ``getattr`` cannot distinguish from a returned + # inline null — pydantic's set-field tracking can: a never-set + # ``value`` falls through to the signed URL, while an explicitly + # returned null stays a real inline value. + fields_set = getattr(artifact_data, "model_fields_set", None) + if fields_set is None: + fields_set = getattr(artifact_data, "__fields_set__", None) + if fields_set is not None and "value" not in fields_set: + value = _MISSING + # A genuinely absent ``value`` field means there are no inline bytes + # here, so fall through to the signed-URL path. An explicit inline + # ``null`` is a real value and is written as JSON ``null``. + if value is _MISSING: + return None + # Every inline value (strings included) is JSON-encoded so the written + # ``.json`` file is always valid JSON; ``str()`` would emit a Python + # repr, and a raw string would not round-trip as JSON. + return json.dumps(value).encode("utf-8") + + def _download_signed_artifact_to( + self, artifact_id: str, dest: Path, data_error: Exception, data_status: int | None + ) -> None: + endpoint = _data_endpoint(artifact_id) + direct = f"GET {endpoint} returned {_http_error_detail(data_status, data_error)}" + try: + signed_response = self._require_client().artifacts_signed_artifact_url(artifact_id) + except Exception as exc: + raise RuntimeError( + f"Artifact {artifact_id} download failed: {direct}; " + f"signed-URL request failed: {_http_error_detail(_exc_status(exc), exc)}" + ) from exc + signed_url = _mapping_or_attr(signed_response, "signed_url") + if not signed_url: + raise RuntimeError( + f"Artifact {artifact_id} download failed: {direct}; no signed URL available" + ) from data_error + signed_url = str(signed_url) + if not (_is_absolute_http_url(signed_url) or _is_relative_path_url(signed_url)): + # A non-http(s) absolute URL (file://, gs://, s3://) or a + # protocol-relative //host/... URL is neither fetchable + # unauthenticated as http(s) nor a same-origin path; handing it to + # the client's raw request raises a ValueError that would escape the + # CLI as a traceback, and fetching it could forward credentials to + # an unexpected host. Reject cleanly (without echoing the URL, which + # can carry signed credentials). + raise RuntimeError( + f"Artifact {artifact_id} download failed: {direct}; unsupported " + f"signed URL (expected an http(s) URL or a same-origin path)" + ) from data_error + try: + response = self._download_signed_url(signed_url) + except RuntimeError as exc: + # Signed-URL policy failures raised below (internal host, + # disallowed redirect target, redirect without Location, too many + # redirects, missing raw-request support) get the same artifact/ + # direct context as the transport failures they sit alongside. + raise RuntimeError( + f"Artifact {artifact_id} download failed: {direct}; {exc}" + ) from exc + except requests.RequestException as exc: + # DNS/TLS/timeout/connection failure before any response object + # exists (e.g. absolute signed URL ``requests.get``). Without this + # the raw requests exception would escape the CLI's RuntimeError + # handler as a traceback. + raise RuntimeError( + f"Artifact {artifact_id} download failed: {direct}; " + f"signed-URL transport failed: {_signed_url_error_detail(_exc_status(exc), exc)}" + ) from exc + try: + try: + response.raise_for_status() + except Exception as exc: + raise RuntimeError( + f"Artifact {artifact_id} download failed: {direct}; signed-URL download " + f"returned {_signed_url_error_detail(_response_status_code(response), exc)}" + ) from exc + try: + _stream_response_to_file(response, dest) + except requests.RequestException as exc: + raise RuntimeError( + f"Artifact {artifact_id} download failed: {direct}; signed-URL " + f"download streaming failed: {_signed_url_error_detail(_exc_status(exc), exc)}" + ) from exc + finally: + _close_quietly(response) + + def _download_signed_url(self, signed_url: str) -> requests.Response: + # Absolute signed URLs point at the storage origin (e.g. GCS), so they + # are fetched with an unauthenticated, streamed ``requests.get`` — Tangle + # API auth headers/tokens are never sent off-origin. Relative signed + # paths stay same-origin via the authenticated client (which itself + # refuses cross-origin redirects). + if _is_absolute_http_url(signed_url): + return self._download_absolute_signed_url(signed_url) + client = self._require_client() + request = getattr(client, "request_raw", None) + if not callable(request): + raise RuntimeError("Relative signed artifact URLs require raw request support") + return request("GET", signed_url, stream=True) + + @staticmethod + def _download_absolute_signed_url(signed_url: str) -> requests.Response: + """Fetch an absolute signed URL, rechecking each redirect hop. + + Redirects are followed manually (``allow_redirects=False``) so every + hop is re-validated against the http(s) allowlist that admitted the + initial signed URL. An automatic follow would hand a non-http(s) + redirect target straight to ``requests``, whose ``InvalidSchema`` error + echoes the target URL — and redirect targets derived from signed URLs + can carry signed credentials. Rejection messages never echo the URL. + + The initial URL and every redirect target are also checked against the + internal-host blocklist before being fetched (see + ``_reject_internal_host``). + """ + + _reject_internal_host(signed_url, context="signed URL host") + current_url = signed_url + for _ in range(_MAX_SIGNED_URL_REDIRECTS + 1): + response = requests.get( + current_url, timeout=_SIGNED_URL_TIMEOUT_SECONDS, stream=True, allow_redirects=False + ) + if response.status_code not in _REDIRECT_STATUS_CODES: + return response + location = response.headers.get("Location") + _close_quietly(response) + if not location: + raise RuntimeError( + f"signed-URL download returned HTTP {response.status_code} " + "with no redirect Location" + ) + current_url = urljoin(current_url, location) + if not _is_absolute_http_url(current_url): + raise RuntimeError( + "signed URL redirected to an unsupported target (expected an http(s) URL)" + ) + _reject_internal_host(current_url, context="signed URL redirected to host") + raise RuntimeError( + f"signed URL exceeded {_MAX_SIGNED_URL_REDIRECTS} redirects" + ) + @staticmethod def serialize_artifacts(artifacts: dict[str, ArtifactInfo]) -> list[dict[str, Any]]: """Serialize artifact dict to a JSON-friendly list, dropping ``None`` fields.""" @@ -268,6 +704,66 @@ def _artifact_id_map(raw_artifacts: Any) -> dict[str, str]: return artifact_ids +def _query_shape_error(location: str, expected: str, value: Any) -> RuntimeError: + return RuntimeError(f"--query {location} must be {expected}, got {type(value).__name__}") + + +def _validate_output_names(location: str, outputs: Any) -> None: + if outputs is None: + return + if not isinstance(outputs, list) or not all(isinstance(name, str) for name in outputs): + raise _query_shape_error(location, "a list of output-name strings", outputs) + + +def _validate_query_shape(query: dict[str, Any]) -> None: + """Reject malformed nested ``--query`` values before any API call. + + Each supported key has a fixed shape (documented on ``get_artifacts``). + Anything else would otherwise surface as an ``AttributeError`` or + ``TypeError`` traceback from deep inside the artifact walk (for example + ``.items()`` on a list) instead of a clean error. ``null`` top-level + sections are treated as absent, and ``null`` per-entry output lists mean + all outputs (same as ``[]``), matching the ``or``-defaults applied + downstream. + """ + + for key in ("tasks", "executions"): + section = query.get(key) + if section is None: + continue + if not isinstance(section, dict): + raise _query_shape_error( + f"'{key}'", "an object mapping names to output-name lists", section + ) + for name, outputs in section.items(): + _validate_output_names(f"'{key}' entry {name!r}", outputs) + + components = query.get("components") + if components is not None: + if not isinstance(components, list): + raise _query_shape_error("'components'", "a list of objects", components) + for index, component in enumerate(components): + if not isinstance(component, dict): + raise _query_shape_error(f"'components' entry {index}", "an object", component) + for field_name in ("name", "digest"): + field_value = component.get(field_name) + if field_value is not None and not isinstance(field_value, str): + raise _query_shape_error( + f"'components' entry {index} key '{field_name}'", "a string", field_value + ) + _validate_output_names( + f"'components' entry {index} key 'outputs'", component.get("outputs") + ) + + artifact_ids = query.get("artifact_ids") + if artifact_ids is not None: + if not isinstance(artifact_ids, list): + raise _query_shape_error("'artifact_ids'", "a list of artifact-id strings", artifact_ids) + for index, artifact_id in enumerate(artifact_ids): + if not isinstance(artifact_id, str): + raise _query_shape_error(f"'artifact_ids' entry {index}", "a string", artifact_id) + + def _component_queries(raw_components: list[dict[str, Any]]) -> list[ArtifactComponentQuery]: return [ ArtifactComponentQuery( @@ -285,6 +781,260 @@ def _artifact_info_from_response(response: Any, *, artifact_id: str, key: str) - return ArtifactInfo.from_response(response, key=key) +def _child_execution_ids(execution: Any) -> dict[str, str]: + """Map direct child task names to their execution ids.""" + + raw = _mapping_or_attr(execution, "child_task_execution_ids", {}) + if not isinstance(raw, dict): + return {} + return {str(name): str(execution_id) for name, execution_id in raw.items() if execution_id} + + +def _direct_child_executions(execution: Any) -> dict[str, Any]: + """Return already-inlined direct child executions keyed by task name. + + Some responses embed the child executions directly (``child_executions``), + which avoids a follow-up fetch per child. When absent, the caller resolves + ids via ``_child_execution_ids`` instead. + """ + + raw = _mapping_or_attr(execution, "child_executions", {}) + if not isinstance(raw, dict): + return {} + return {str(name): child for name, child in raw.items() if child} + + +def _data_endpoint(artifact_id: str) -> str: + # Percent-encode the id (``safe=''`` encodes ``/`` too) so an id containing + # a slash or other path-significant characters cannot escape the intended + # ``/api/artifacts/{id}/data`` path. + return f"/api/artifacts/{quote(artifact_id, safe='')}/data" + + +def _safe_name(name: str) -> str: + """Reduce an arbitrary label to filesystem-safe characters.""" + + cleaned = _SAFE_NAME_RE.sub("_", str(name)).strip("._") + return cleaned or "artifact" + + +def _artifact_filename(owner: str, output_name: str, artifact_id: str) -> str: + # A short id suffix keeps names readable while disambiguating same-named + # outputs across owners; the collision guard in download handles the rare + # case where two distinct ids share a 12-char prefix. No extension is + # forced: inline JSON values gain a ``.json`` suffix at write time. + return f"{_safe_name(owner)}__{_safe_name(output_name)}__{_safe_name(artifact_id[:12])}" + + +def _open_new_artifact_file(dest: Path) -> BinaryIO: + """Open ``dest`` for writing, refusing overwrite and never following symlinks. + + ``O_CREAT | O_EXCL`` fails atomically if the path already exists (a file, + directory, or symlink), so downloads never clobber existing data or follow a + symlink to write outside the output directory, with no check-then-write + TOCTOU window. + """ + + try: + fd = os.open(dest, os.O_WRONLY | os.O_CREAT | os.O_EXCL | _O_NOFOLLOW, 0o600) + except FileExistsError as exc: + raise RuntimeError(f"Refusing to overwrite existing path: {dest.name}") from exc + except OSError as exc: + raise RuntimeError( + f"Cannot write artifact to {dest.name}: {exc.strerror or exc}" + ) from exc + return os.fdopen(fd, "wb") + + +def _write_new_file_bytes(dest: Path, data: bytes) -> None: + handle = _open_new_artifact_file(dest) + try: + handle.write(data) + handle.close() + except OSError as exc: + # Same clean-error contract as the streaming path: a local write failure + # is a concise RuntimeError, and the partial file is removed. + _discard_partial_download(handle, dest) + raise RuntimeError( + f"Cannot write artifact to {dest.name}: {exc.strerror or exc}" + ) from exc + except BaseException: + _discard_partial_download(handle, dest) + raise + + +def _stream_response_to_file(response: requests.Response, dest: Path) -> None: + """Stream a response body to ``dest`` in chunks, cleaning up on failure.""" + + handle = _open_new_artifact_file(dest) + try: + for chunk in response.iter_content(chunk_size=_DOWNLOAD_CHUNK_SIZE): + if chunk: + handle.write(chunk) + # Close on the success path so a buffered-flush error surfaces as a + # write failure handled below, not from a ``finally`` block. + handle.close() + except requests.RequestException: + # Mid-stream transport failures get route context from the caller. + # ``RequestException`` subclasses ``OSError``, so it must be re-raised + # before the OSError handler below can claim it. + _discard_partial_download(handle, dest) + raise + except OSError as exc: + # A mid-stream local write failure (e.g. disk full) would otherwise + # escape the CLI's RuntimeError handler as a raw traceback. + _discard_partial_download(handle, dest) + raise RuntimeError( + f"Cannot write artifact to {dest.name}: {exc.strerror or exc}" + ) from exc + except BaseException: + _discard_partial_download(handle, dest) + raise + + +def _discard_partial_download(handle: BinaryIO, dest: Path) -> None: + _close_quietly(handle) + dest.unlink(missing_ok=True) + + +def _close_quietly(response: Any) -> None: + close = getattr(response, "close", None) + if callable(close): + try: + close() + except Exception: + # Closing a response is best-effort resource cleanup in ``finally`` + # blocks; a close failure must never replace the real error. + pass + + +def _response_status_code(response: Any) -> int | None: + status = getattr(response, "status_code", None) + return status if isinstance(status, int) else None + + +def _exc_status(exc: Exception) -> int | None: + return _response_status_code(getattr(exc, "response", None)) + + +def _http_error_detail(status: int | None, exc: Exception) -> str: + """Concise, traceback-free description of a failed HTTP request.""" + + if status is not None: + return f"HTTP {status}" + message = str(exc).strip() + return message or exc.__class__.__name__ + + +def _signed_url_error_detail(status: int | None, exc: Exception) -> str: + """Describe a signed-URL failure without echoing the (credential-bearing) URL. + + ``requests`` embeds the full request URL in many exception messages; a + signed URL carries credentials in its query string, so only the status and + exception type are reported — never ``str(exc)``. + """ + + if status is not None: + return f"HTTP {status}" + return exc.__class__.__name__ + + +def _is_absolute_http_url(url: str) -> bool: + parsed = urlparse(url) + return parsed.scheme in ("http", "https") and bool(parsed.netloc) + + +def _is_relative_path_url(url: str) -> bool: + """True for a same-origin path (no scheme, no host) fetched via the client. + + A protocol-relative ``//host/...`` URL has a netloc and a non-http(s) + absolute URL (``file://``, ``gs://``, ``s3://``) has a scheme, so both are + excluded here — they are neither fetchable unauthenticated as http(s) nor a + safe same-origin path for the authenticated client. + """ + + parsed = urlparse(url) + return not parsed.scheme and not parsed.netloc + + +def _internal_artifact_hosts_allowed() -> bool: + value = os.environ.get(_ALLOW_INTERNAL_HOSTS_ENV, "") + return value.strip().lower() in ("1", "true", "yes", "on") + + +def _embedded_ipv4( + ip: ipaddress.IPv4Address | ipaddress.IPv6Address, +) -> ipaddress.IPv4Address | None: + """Extract the IPv4 address embedded in an IPv4-mapped/NAT64 IPv6 address.""" + + if not isinstance(ip, ipaddress.IPv6Address): + return None + if ip.ipv4_mapped is not None: + return ip.ipv4_mapped + if any(ip in prefix for prefix in _NAT64_PREFIXES): + return ipaddress.IPv4Address(int(ip) & 0xFFFFFFFF) + return None + + +def _host_rejection_reason(host: str | None) -> str | None: + """Why ``host`` must not be fetched, or ``None`` for an external host. + + Checks the host literal only — no DNS resolution — so this is not a + defense against DNS rebinding (a public hostname resolving to an internal + address is not caught). Rejects ``localhost`` names, loopback (127.0.0.0/8, + ::1), link-local (169.254.0.0/16 including the cloud metadata IP, + fe80::/10), RFC1918 and IPv6 unique-local (fc00::/7) private ranges, + carrier-grade NAT (100.64.0.0/10), the unspecified address (0.0.0.0, ::), + and non-canonical ``inet_aton`` numeric spellings (see + ``_INET_ATON_NUMERIC_HOST_RE``). + Trailing dots are stripped before matching, and IPv4 addresses embedded in + IPv4-mapped (``::ffff:127.0.0.1``) or NAT64 (``64:ff9b::a9fe:a9fe``) IPv6 + literals are unwrapped and classified by their IPv4 rules. + """ + + if not host: + return None + host = host.rstrip(".") + if host == "localhost" or host.endswith(".localhost"): + return "is an internal address" + try: + ip = ipaddress.ip_address(host) + except ValueError: + if _INET_ATON_NUMERIC_HOST_RE.match(host): + return "is a non-canonical numeric address" + return None + embedded = _embedded_ipv4(ip) + if embedded is not None: + ip = embedded + if ip.is_loopback or ip.is_link_local or ip.is_private or ip.is_unspecified: + return "is an internal address" + # CGNAT space (RFC 6598) is not covered by is_private but is only routable + # inside a provider's network, so treat it as internal too. + if ip.version == 4 and ip in _CGNAT_NETWORK: + return "is an internal address" + return None + + +def _reject_internal_host(url: str, *, context: str) -> None: + """Raise unless ``url``'s host is external or the env override is set. + + ``urlparse().hostname`` lowercases the host and strips IPv6 brackets, so + the checks in ``_host_rejection_reason`` see a normalized literal. The + message names the host (not the full URL, which can carry signed + credentials). + """ + + host = urlparse(url).hostname + reason = _host_rejection_reason(host) + if reason is None or _internal_artifact_hosts_allowed(): + return + raise RuntimeError( + f"{context} {host!r} {reason}; refusing to fetch " + f"(set {_ALLOW_INTERNAL_HOSTS_ENV}=1 to allow internal hosts, " + "e.g. for local/dev environments)" + ) + + __all__ = [ "ArtifactClient", "ArtifactComponentQuery", diff --git a/packages/tangle-cli/src/tangle_cli/artifacts_cli.py b/packages/tangle-cli/src/tangle_cli/artifacts_cli.py index a6d59e1..34fd92e 100644 --- a/packages/tangle-cli/src/tangle_cli/artifacts_cli.py +++ b/packages/tangle-cli/src/tangle_cli/artifacts_cli.py @@ -1,11 +1,17 @@ -"""`tangle sdk artifacts` read-only artifact commands.""" +"""`tangle sdk artifacts` artifact metadata, listing, and download commands.""" from __future__ import annotations -from typing import Annotated, Any +import json +from collections.abc import Iterator +from contextlib import contextmanager +from pathlib import Path +from typing import Annotated, Any, Callable +import requests from cyclopts import App, Parameter +from .args_container import ConfigFileError from .cli_helpers import ( LazyTangleApiClient, api_arg_specs, @@ -31,22 +37,194 @@ help=( "JSON query with optional keys: " "'tasks', 'components', 'executions', and 'artifact_ids'. " - "Empty output lists mean all outputs." + "Empty output lists mean all outputs. " + "Mutually exclusive with --list and --out-dir." ), ), ] +ListOption = Annotated[ + bool, + Parameter( + name="--list", + help="List result artifact metadata for the run instead of querying by task/component.", + ), +] + +OutDirOption = Annotated[ + str | None, + Parameter( + name="--out-dir", + help="Fetch result artifact bytes into this directory (created if missing).", + ), +] + +OnlyOption = Annotated[ + list[str] | None, + Parameter( + name="--only", + help="Restrict --out-dir output to these output names (repeatable). Requires --out-dir.", + ), +] + +IncludeChildrenOption = Annotated[ + bool, + Parameter( + name="--include-children", + help="Include direct child task outputs. Valid only with --list or --out-dir.", + ), +] + app = App( name="artifacts", - help="Read artifact metadata for Tangle pipeline runs.", + help="Resolve, list, and fetch artifacts for Tangle pipeline runs.", ) +_HTTP_ERROR_BODY_LIMIT = 2000 + + +def _format_http_failure(exc: requests.HTTPError) -> str: + """Render an API HTTP failure as a concise one-line CLI message. + + The status, reason, attempted method/URL, and a trimmed response body are + what a caller needs to act on. Only authenticated metadata endpoints raise + ``HTTPError`` through this layer; signed-URL failures are redacted and + re-raised as ``RuntimeError`` inside ``ArtifactManager`` and never reach + this formatter, so echoing the URL here cannot leak signed credentials. + """ + + response = exc.response + if response is None: + return f"Tangle API request failed: {exc}" + request = getattr(response, "request", None) + target = ( + f"{request.method} {request.url}" + if request is not None and getattr(request, "url", None) + else (response.url or "Tangle API") + ) + reason = f" {response.reason}" if response.reason else "" + summary = f"Tangle API request failed ({response.status_code}{reason}) for {target}" + body = (response.text or "").strip() + if not body: + return summary + if len(body) > _HTTP_ERROR_BODY_LIMIT: + body = f"{body[:_HTTP_ERROR_BODY_LIMIT]}... (truncated)" + return f"{summary}: {body}" + + +@contextmanager +def _clean_artifact_errors() -> Iterator[None]: + """Re-raise API HTTP/transport failures as ``RuntimeError`` for CLI output. + + ``ArtifactManager`` lets ``requests`` exceptions from metadata endpoints + propagate. This boundary converts them into the ``RuntimeError`` the + command handler reports as a JSON error with a nonzero exit, instead of + letting a raw traceback reach the interpreter. + """ + + try: + yield + except requests.HTTPError as exc: + raise RuntimeError(_format_http_failure(exc)) from exc + except requests.RequestException as exc: + raise RuntimeError(f"Tangle API request failed: {exc}") from exc + + +def _bool_config_converter(field_name: str) -> Callable[[Any], bool]: + def convert(value: Any) -> bool: + if not isinstance(value, bool): + raise ConfigFileError( + f"{field_name} must be a boolean (true/false), got {type(value).__name__}" + ) + return value + + return convert + + +def _str_list_config_converter(field_name: str) -> Callable[[Any], list[str]]: + def convert(value: Any) -> list[str]: + if not isinstance(value, list) or not all(isinstance(item, str) for item in value): + raise ConfigFileError(f"{field_name} must be a list of strings") + return value + + return convert + + +def _query_json_converter(field_name: str) -> Callable[[Any], Any]: + """JSON converter that keeps an explicit empty object distinct from absent. + + The shared JSON converter maps ``"{}"``/``"[]"`` to ``None``, which would + turn an explicit ``--query '{}'`` into "no query passed" and surface the + misleading "--query is required" error. Here every JSON string is parsed + as-is, so ``'{}'`` stays an (empty, valid) object query and non-object + values reach the object-shape check with their real type. + """ + + def convert(value: Any) -> Any: + if isinstance(value, (dict, list)): + return value + if isinstance(value, str): + try: + return json.loads(value) + except json.JSONDecodeError as exc: + raise ConfigFileError(f"Invalid JSON for {field_name}: {exc}") from exc + raise ConfigFileError( + f"{field_name} must be a dict, list, or JSON string, got {type(value).__name__}" + ) + + return convert + + +def _path_config_converter(field_name: str) -> Callable[[Any], str]: + def convert(value: Any) -> str: + if not isinstance(value, (str, Path)): + raise ConfigFileError(f"{field_name} must be a filesystem path string") + return str(value) + + return convert + + +def _require_single_mode(*, query: Any, list_outputs: bool, out_dir: Any) -> None: + """Exactly one of --query / --list / --out-dir must be selected.""" + + selected = [] + if query is not None: + selected.append("--query") + if list_outputs: + selected.append("--list") + if out_dir is not None: + selected.append("--out-dir") + if len(selected) > 1: + raise RuntimeError(f"{' / '.join(selected)} are mutually exclusive; pass only one") + if not selected: + raise RuntimeError("--query is required unless --list or --out-dir is set") + + +def _require_flag_modes( + *, list_outputs: bool, out_dir: Any, only: Any, include_children: bool +) -> None: + """--only / --include-children are valid only alongside download/list modes.""" + + if only and out_dir is None: + raise RuntimeError("--only is only valid with --out-dir") + if include_children and not (list_outputs or out_dir is not None): + raise RuntimeError("--include-children is only valid with --list or --out-dir") + + +def _require_object_query(query: Any) -> None: + if query is not None and not isinstance(query, dict): + raise RuntimeError("--query must be a JSON object") + @app.command(name="get") def artifacts_get( run_id: str | None = None, *, query: QueryOption = None, + list_outputs: ListOption = False, + out_dir: OutDirOption = None, + only: OnlyOption = None, + include_children: IncludeChildrenOption = False, base_url: BaseUrlOption = None, token: TokenOption = None, auth_header: AuthHeaderOption = None, @@ -54,12 +232,35 @@ def artifacts_get( config: ConfigOption = None, log_type: LogTypeOption = "console", ) -> None: - """Get artifact metadata for tasks/components in a pipeline run.""" + """Resolve, list, or fetch artifacts for a pipeline run. + + Selects exactly one mode: ``--query`` (resolve artifact metadata for + tasks/components/executions/artifact ids), ``--list`` (list result artifact + metadata), or ``--out-dir`` (fetch output artifact bytes to a directory). + + Downloaded files are named ``____`` and + written with owner-only permissions on POSIX (mode ``0o600``; Windows + ACLs are not adjusted); inline artifact values are + JSON-encoded and gain a ``.json`` suffix, while streamed and signed-URL + downloads keep the name as-is. + """ all_args = load_args_or_exit( config, run_id=("run_id", run_id, None, False, True), - query=("query", query, None, True, True), + # query is optional here: --list / --out-dir modes do not use it. + query=("query", query, None, False, False, _query_json_converter("query")), + list_outputs=("list", list_outputs, False, False, False, _bool_config_converter("list")), + out_dir=("out_dir", out_dir, None, False, False, _path_config_converter("out_dir")), + only=("only", only, None, False, False, _str_list_config_converter("only")), + include_children=( + "include_children", + include_children, + False, + False, + False, + _bool_config_converter("include_children"), + ), log_type=(log_type, "console"), **api_arg_specs( base_url=base_url, @@ -73,6 +274,25 @@ def artifacts_get( for args in all_args: logger, finalize_logs = logger_for_log_type(args.log_type) try: + # Mode/shape validation runs before any client construction so an + # invalid flag combination fails without touching the network. + try: + _require_single_mode( + query=args.query, + list_outputs=args.list_outputs, + out_dir=args.out_dir, + ) + _require_flag_modes( + list_outputs=args.list_outputs, + out_dir=args.out_dir, + only=args.only, + include_children=args.include_children, + ) + _require_object_query(args.query) + except RuntimeError as exc: + print_json({"status": "error", "error": str(exc)}) + raise SystemExit(1) from exc + client = LazyTangleApiClient( base_url=args.base_url, token=args.token, @@ -80,29 +300,68 @@ def artifacts_get( header=args.header, include_env_credentials=include_env_credentials_for_args(args, base_url), command_name="artifact commands", + logger=logger, ) if require_available := getattr(client, "require_available", None): require_available() from .artifacts import ArtifactManager - manager = ArtifactManager(client=client) + manager = ArtifactManager(client=client, logger=logger) try: - artifacts = manager.get_artifacts(args.run_id, args.query) + with _clean_artifact_errors(): + if args.out_dir is not None: + results.append(_download(manager, args)) + elif args.list_outputs: + results.append(_list(manager, args)) + else: + results.append(_query(manager, args)) except RuntimeError as exc: print_json({"status": "error", "error": str(exc)}) raise SystemExit(1) from exc - - results.append( - { - "status": "success", - "run_id": args.run_id, - "count": len(artifacts), - "artifacts": ArtifactManager.serialize_artifacts(artifacts), - } - ) finally: finalize_logs() print_json( results[0] if len(results) == 1 else {"status": "success", "results": results} ) + + +def _query(manager: Any, args: Any) -> dict[str, Any]: + from .artifacts import ArtifactManager + + artifacts = manager.get_artifacts(args.run_id, args.query) + return { + "status": "success", + "run_id": args.run_id, + "count": len(artifacts), + "artifacts": ArtifactManager.serialize_artifacts(artifacts), + } + + +def _list(manager: Any, args: Any) -> dict[str, Any]: + rows = manager.list_result_artifacts( + args.run_id, + include_children=args.include_children, + ) + return { + "status": "success", + "run_id": args.run_id, + "count": len(rows), + "artifacts": rows, + } + + +def _download(manager: Any, args: Any) -> dict[str, Any]: + artifacts = manager.download_result_artifacts( + args.run_id, + out_dir=args.out_dir, + only=args.only, + include_children=args.include_children, + ) + return { + "status": "success", + "run_id": args.run_id, + "count": len(artifacts), + "out_dir": str(Path(args.out_dir).resolve()), + "artifacts": {key: str(path.resolve()) for key, path in artifacts.items()}, + } diff --git a/packages/tangle-cli/src/tangle_cli/client.py b/packages/tangle-cli/src/tangle_cli/client.py index 060c960..5f3ae15 100644 --- a/packages/tangle-cli/src/tangle_cli/client.py +++ b/packages/tangle-cli/src/tangle_cli/client.py @@ -144,6 +144,24 @@ def _make_request( ) return response + def request_raw( + self, + method: str, + path: str, + params: Mapping[str, Any] | None = None, + json_data: Any = None, + **kwargs: Any, + ) -> requests.Response: + """Public alias for :meth:`_make_request`. + + Issues an HTTP request against an API path and returns the raw + ``requests.Response`` (retries, auth refresh, and redirect handling + included). Standard ``requests`` keyword arguments such as + ``stream=True`` are forwarded. + """ + + return self._make_request(method, path, params=params, json_data=json_data, **kwargs) + def _request_with_rate_limit_retries( self, method: str, diff --git a/tests/test_artifacts.py b/tests/test_artifacts.py index 6347feb..00c8e70 100644 --- a/tests/test_artifacts.py +++ b/tests/test_artifacts.py @@ -1,13 +1,75 @@ from __future__ import annotations +import json from types import SimpleNamespace from typing import Any import pytest +import requests +from pydantic import BaseModel +from tangle_cli import artifacts as tangle_artifacts from tangle_cli.artifacts import ArtifactManager +def _manager(client: Any) -> ArtifactManager: + return ArtifactManager(client=client) + + +class FakeDataResponse: + def __init__( + self, + content: bytes, + status_code: int = 200, + headers: dict[str, str] | None = None, + ) -> None: + self.content = content + self.status_code = status_code + self.headers = headers or {} + self.closed = False + + def raise_for_status(self) -> None: + if self.status_code >= 400: + # Mirror ``requests``: raise an ``HTTPError`` carrying the response so + # the download path can surface a concise status without a traceback. + raise requests.HTTPError(f"{self.status_code} Server Error", response=self) + return None + + def iter_content(self, chunk_size: int = 1) -> Any: + for start in range(0, len(self.content), max(chunk_size, 1)): + yield self.content[start : start + chunk_size] + + def close(self) -> None: + self.closed = True + + +class StreamingOnlyResponse: + """Response that yields chunks but raises if the body is buffered via ``.content``. + + Proves the download path streams to disk and never materializes the full + artifact in memory. + """ + + def __init__(self, chunks: list[bytes], status_code: int = 200) -> None: + self._chunks = chunks + self.status_code = status_code + self.iter_calls: list[int] = [] + + @property + def content(self) -> bytes: + raise AssertionError("streamed download must not access response.content") + + def raise_for_status(self) -> None: + return None + + def iter_content(self, chunk_size: int = 1) -> Any: + self.iter_calls.append(chunk_size) + yield from self._chunks + + def close(self) -> None: + return None + + def _artifact_response(artifact_id: str, uri: str) -> SimpleNamespace: return SimpleNamespace( id=artifact_id, @@ -26,10 +88,19 @@ def __init__(self) -> None: self.artifact_responses: dict[str, Any] = {} self.artifact_errors: dict[str, Exception] = {} self.execution_details: dict[str, Any] = {} + self.execution_errors: dict[str, Exception] = {} self.run_details: dict[str, Any] = {} self.artifacts_get_calls: list[str] = [] self.get_execution_details_calls: list[str] = [] self.get_run_details_calls: list[str] = [] + self.data_responses: dict[str, bytes] = {} + self.data_status_codes: dict[str, int] = {} + self.signed_data_status_codes: dict[str, int] = {} + self.signed_urls: dict[str, str] = {} + self.signed_url_errors: dict[str, Exception] = {} + self.make_request_calls: list[tuple[str, str]] = [] + self.stream_responses: dict[str, StreamingOnlyResponse] = {} + self.artifacts_signed_url_calls: list[str] = [] def artifacts_get(self, artifact_id: str) -> Any: self.artifacts_get_calls.append(artifact_id) @@ -39,11 +110,46 @@ def artifacts_get(self, artifact_id: str) -> Any: def get_execution_details(self, execution_id: str) -> Any: self.get_execution_details_calls.append(execution_id) + if execution_id in self.execution_errors: + raise self.execution_errors[execution_id] return self.execution_details[execution_id] def get_run_details(self, run_id: str) -> Any: self.get_run_details_calls.append(run_id) - return self.run_details[run_id] + if run_id in self.run_details: + return self.run_details[run_id] + if run_id in self.execution_errors: + raise self.execution_errors[run_id] + # Model the client's run->root-execution resolution: with no explicit + # run record, fall back to treating the id as the root execution id. + return SimpleNamespace( + run=SimpleNamespace(root_execution_id=run_id), + execution=self.execution_details.get(run_id), + ) + + def request_raw(self, method: str, path: str, **kwargs: Any) -> Any: + self.make_request_calls.append((method, path)) + if path.startswith("http"): + return FakeDataResponse(self.data_responses[path]) + if path.endswith("/signed-data"): + artifact_id = path.removeprefix("/api/artifacts/").removesuffix("/signed-data") + return FakeDataResponse( + self.data_responses.get(artifact_id, b""), + self.signed_data_status_codes.get(artifact_id, 200), + ) + artifact_id = path.removeprefix("/api/artifacts/").removesuffix("/data") + if artifact_id in self.stream_responses: + return self.stream_responses[artifact_id] + return FakeDataResponse( + self.data_responses.get(artifact_id, b""), + self.data_status_codes.get(artifact_id, 200), + ) + + def artifacts_signed_artifact_url(self, artifact_id: str) -> Any: + self.artifacts_signed_url_calls.append(artifact_id) + if artifact_id in self.signed_url_errors: + raise self.signed_url_errors[artifact_id] + return SimpleNamespace(signed_url=self.signed_urls[artifact_id]) def _task( @@ -61,11 +167,30 @@ def _task( ) -def _execution(graph_tasks: dict[str, Any], child_executions: dict[str, Any] | None = None) -> SimpleNamespace: +def _execution( + graph_tasks: dict[str, Any], + child_executions: dict[str, Any] | None = None, + output_artifacts: dict[str, Any] | None = None, +) -> SimpleNamespace: return SimpleNamespace( task_spec=SimpleNamespace(graph_tasks=graph_tasks), child_executions=child_executions or {}, + output_artifacts=output_artifacts or {}, + ) + + +def _single_root_artifact_client(artifact_id: str = "artifact-model") -> FakeArtifactClient: + client = FakeArtifactClient() + client.execution_details["root-exec"] = _execution( + {}, + output_artifacts={"model": {"id": artifact_id}}, ) + return client + + +# --------------------------------------------------------------------------- +# Metadata query resolution +# --------------------------------------------------------------------------- def test_get_artifacts_resolves_direct_artifact_ids_without_run_tree() -> None: @@ -75,7 +200,7 @@ def test_get_artifacts_resolves_direct_artifact_ids_without_run_tree() -> None: "artifact-2": _artifact_response("artifact-2", "gs://bucket/artifact-2"), } - artifacts = ArtifactManager(client=client).get_artifacts( + artifacts = _manager(client).get_artifacts( "run-1", {"artifact_ids": ["artifact-1", "artifact-2"]}, ) @@ -94,7 +219,7 @@ def test_get_artifacts_resolves_execution_output_lookup() -> None: ) client.artifact_responses["artifact-model"] = _artifact_response("artifact-model", "gs://bucket/model") - artifacts = ArtifactManager(client=client).get_artifacts("run-1", {"executions": {"exec-1": ["model"]}}) + artifacts = _manager(client).get_artifacts("run-1", {"executions": {"exec-1": ["model"]}}) assert list(artifacts) == ["exec-1/model"] assert artifacts["exec-1/model"].id == "artifact-model" @@ -116,7 +241,7 @@ def test_get_artifacts_resolves_task_query_from_run_details_tree() -> None: ) client.artifact_responses["artifact-model"] = _artifact_response("artifact-model", "gs://bucket/model") - artifacts = ArtifactManager(client=client).get_artifacts("run-1", {"tasks": {"Train": ["model"]}}) + artifacts = _manager(client).get_artifacts("run-1", {"tasks": {"Train": ["model"]}}) assert list(artifacts) == ["Train/model"] assert artifacts["Train/model"].uri == "gs://bucket/model" @@ -136,7 +261,7 @@ def test_get_artifacts_resolves_component_name_and_digest_queries() -> None: client.artifact_responses["artifact-vectors"] = _artifact_response("artifact-vectors", "gs://bucket/vectors") client.artifact_responses["artifact-scores"] = _artifact_response("artifact-scores", "gs://bucket/scores") - artifacts = ArtifactManager(client=client).get_artifacts( + artifacts = _manager(client).get_artifacts( "run-1", { "components": [ @@ -167,7 +292,7 @@ def test_get_artifacts_unions_outputs_from_multiple_matching_selectors() -> None client.artifact_responses["artifact-model"] = _artifact_response("artifact-model", "gs://bucket/model") client.artifact_responses["artifact-metrics"] = _artifact_response("artifact-metrics", "gs://bucket/metrics") - artifacts = ArtifactManager(client=client).get_artifacts( + artifacts = _manager(client).get_artifacts( "run-1", { "tasks": {"Train": ["model"]}, @@ -193,7 +318,7 @@ def test_get_artifacts_resolves_nested_subgraph_task_paths() -> None: ) client.artifact_responses["artifact-inner"] = _artifact_response("artifact-inner", "gs://bucket/inner") - artifacts = ArtifactManager(client=client).get_artifacts("run-1", {"tasks": {"Subgraph/Inner": ["out"]}}) + artifacts = _manager(client).get_artifacts("run-1", {"tasks": {"Subgraph/Inner": ["out"]}}) assert list(artifacts) == ["Subgraph/Inner/out"] assert artifacts["Subgraph/Inner/out"].uri == "gs://bucket/inner" @@ -204,7 +329,7 @@ def test_get_artifacts_serializes_per_artifact_lookup_errors() -> None: client.artifact_responses["good"] = _artifact_response("good", "gs://bucket/good") client.artifact_errors["bad"] = RuntimeError("not found") - artifacts = ArtifactManager(client=client).get_artifacts("run-1", {"artifact_ids": ["good", "bad"]}) + artifacts = _manager(client).get_artifacts("run-1", {"artifact_ids": ["good", "bad"]}) assert artifacts["good"].uri == "gs://bucket/good" assert artifacts["bad"].id == "bad" @@ -227,4 +352,1309 @@ def test_get_artifacts_requires_execution_details_for_task_queries() -> None: client.run_details["run-1"] = SimpleNamespace(execution=None) with pytest.raises(RuntimeError, match="No execution details"): - ArtifactManager(client=client).get_artifacts("run-1", {"tasks": {"Train": []}}) + _manager(client).get_artifacts("run-1", {"tasks": {"Train": []}}) + + +@pytest.mark.parametrize("query", [[1, 2, 3], "a string", 42]) +def test_get_artifacts_rejects_non_object_query_without_network(query: Any) -> None: + client = FakeArtifactClient() + + with pytest.raises(RuntimeError, match="--query must be a JSON object"): + _manager(client).get_artifacts("run-1", query) + + assert client.get_run_details_calls == [] + assert client.get_execution_details_calls == [] + assert client.artifacts_get_calls == [] + + +@pytest.mark.parametrize( + "query, match", + [ + ({"executions": [1]}, "'executions' must be an object"), + ({"executions": {"exec-1": "model"}}, "'executions' entry 'exec-1' must be a list"), + ({"executions": {"exec-1": [1]}}, "'executions' entry 'exec-1' must be a list"), + ({"tasks": ["Train"]}, "'tasks' must be an object"), + ({"tasks": {"Train": "model"}}, "'tasks' entry 'Train' must be a list"), + ({"components": {"name": "Train"}}, "'components' must be a list"), + ({"components": ["Train"]}, "'components' entry 0 must be an object"), + ({"components": [{"name": 1}]}, "'components' entry 0 key 'name' must be a string"), + ({"components": [{"digest": 1}]}, "'components' entry 0 key 'digest' must be a string"), + ( + {"components": [{"name": "Train", "outputs": "model"}]}, + "'components' entry 0 key 'outputs' must be a list", + ), + ({"artifact_ids": "artifact-1"}, "'artifact_ids' must be a list"), + ({"artifact_ids": [1]}, "'artifact_ids' entry 0 must be a string"), + ({"artifact_ids": [None]}, "'artifact_ids' entry 0 must be a string"), + ], +) +def test_get_artifacts_rejects_malformed_nested_query_without_network( + query: dict[str, Any], match: str +) -> None: + # Nested shapes are checked up front: a malformed inner value must raise a + # clean RuntimeError, not an AttributeError/TypeError traceback from deep + # inside the artifact walk (e.g. ``.items()`` on a list). + client = FakeArtifactClient() + + with pytest.raises(RuntimeError, match=match): + _manager(client).get_artifacts("run-1", query) + + assert client.get_run_details_calls == [] + assert client.get_execution_details_calls == [] + assert client.artifacts_get_calls == [] + + +@pytest.mark.parametrize( + "query", + [ + {}, + {"tasks": None, "executions": None, "components": None, "artifact_ids": None}, + ], +) +def test_get_artifacts_accepts_empty_query_object_without_network( + query: dict[str, Any], +) -> None: + # An explicit empty object (and explicit nulls, which mean "absent") is a + # valid query that selects nothing and touches no endpoint. + client = FakeArtifactClient() + + assert _manager(client).get_artifacts("run-1", query) == {} + + assert client.get_run_details_calls == [] + assert client.get_execution_details_calls == [] + assert client.artifacts_get_calls == [] + + +@pytest.mark.parametrize( + "query, expected_keys", + [ + # Regression: a null tasks filter passed validation but raised a raw + # ``TypeError: 'NoneType' object is not iterable`` from the artifact + # walk once the named task matched and had output artifacts. + ({"tasks": {"Train": None}}, ["Train/model", "Train/metrics"]), + ({"executions": {"exec-1": None}}, ["exec-1/model", "exec-1/metrics"]), + ( + {"components": [{"name": "Trainer", "outputs": None}]}, + ["Train/model", "Train/metrics"], + ), + ], +) +def test_get_artifacts_treats_null_entry_output_filter_as_all_outputs( + query: dict[str, Any], expected_keys: list[str] +) -> None: + # A ``null`` per-entry output filter means all outputs, exactly like + # ``[]``, for every nested query section. + client = FakeArtifactClient() + client.run_details["run-1"] = SimpleNamespace( + execution=_execution( + { + "Train": _task( + name="Trainer", + output_artifacts={"model": "artifact-model", "metrics": "artifact-metrics"}, + ) + } + ) + ) + client.execution_details["exec-1"] = SimpleNamespace( + output_artifacts={"model": {"id": "artifact-model"}, "metrics": {"id": "artifact-metrics"}} + ) + client.artifact_responses["artifact-model"] = _artifact_response("artifact-model", "gs://bucket/model") + client.artifact_responses["artifact-metrics"] = _artifact_response( + "artifact-metrics", "gs://bucket/metrics" + ) + + artifacts = _manager(client).get_artifacts("run-1", query) + + assert list(artifacts) == expected_keys + assert all(artifact.error is None for artifact in artifacts.values()) + + +# --------------------------------------------------------------------------- +# Listing +# --------------------------------------------------------------------------- + + +def test_list_result_artifacts_includes_root_and_direct_children() -> None: + client = FakeArtifactClient() + child_execution = _execution( + {}, + output_artifacts={"child output": {"id": "artifact-child"}}, + ) + client.execution_details["root-exec"] = _execution( + {}, + child_executions={"Train": child_execution}, + output_artifacts={"final": {"id": "artifact-root"}}, + ) + + artifacts = _manager(client).list_result_artifacts("root-exec", include_children=True) + + assert artifacts == [ + {"owner": "root", "output": "final", "artifact_id": "artifact-root"}, + {"owner": "Train", "output": "child output", "artifact_id": "artifact-child"}, + ] + # Root execution is resolved through the run lookup, not a raw execution + # lookup; direct children are inline so no per-child fetch is needed. + assert client.get_run_details_calls == ["root-exec"] + assert client.get_execution_details_calls == [] + + +def test_list_result_artifacts_fetches_direct_children_from_raw_child_ids() -> None: + client = FakeArtifactClient() + client.execution_details["root-exec"] = SimpleNamespace( + output_artifacts={"final": {"id": "artifact-root"}}, + child_task_execution_ids={"Train": "child-exec"}, + ) + client.execution_details["child-exec"] = SimpleNamespace( + output_artifacts={"child": {"id": "artifact-child"}} + ) + + artifacts = _manager(client).list_result_artifacts("root-exec", include_children=True) + + assert artifacts == [ + {"owner": "root", "output": "final", "artifact_id": "artifact-root"}, + {"owner": "Train", "output": "child", "artifact_id": "artifact-child"}, + ] + # Root execution comes from the run lookup; only the raw child id is fetched + # via get_execution_details (children are already execution ids). + assert client.get_run_details_calls == ["root-exec"] + assert client.get_execution_details_calls == ["child-exec"] + + +def test_list_result_artifacts_resolves_run_id_to_root_execution() -> None: + # The CLI accepts a run id, but the artifact tree lives under the run's root + # execution (a distinct id). The run id must be resolved via get_run_details + # and must not be passed straight to get_execution_details. + client = FakeArtifactClient() + client.run_details["run-1"] = SimpleNamespace( + run=SimpleNamespace(root_execution_id="root-exec-9"), + execution=_execution({}, output_artifacts={"final": {"id": "artifact-root"}}), + ) + + rows = _manager(client).list_result_artifacts("run-1") + + assert rows == [{"owner": "root", "output": "final", "artifact_id": "artifact-root"}] + assert client.get_run_details_calls == ["run-1"] + # The run id is never conflated with an execution id. + assert client.get_execution_details_calls == [] + + +def test_list_result_artifacts_missing_execution_for_run_is_clean_error() -> None: + client = FakeArtifactClient() + client.run_details["run-1"] = SimpleNamespace( + run=SimpleNamespace(root_execution_id="root-exec-9"), + execution=None, + ) + + with pytest.raises(RuntimeError) as exc_info: + _manager(client).list_result_artifacts("run-1") + assert "No execution details found for run run-1" in str(exc_info.value) + + +def test_list_result_artifacts_surfaces_clean_error_on_non_request_child_error() -> None: + # A non-RequestException from get_execution_details (e.g. a client-specific + # error) must be wrapped as a concise RuntimeError instead of escaping the + # CLI's RuntimeError handler as a raw traceback. + client = FakeArtifactClient() + client.execution_details["root-exec"] = SimpleNamespace( + output_artifacts={}, + child_task_execution_ids={"child": "child-exec"}, + ) + client.execution_errors["child-exec"] = ValueError("unexpected client failure") + + with pytest.raises(RuntimeError) as exc_info: + _manager(client).list_result_artifacts("root-exec", include_children=True) + message = str(exc_info.value) + assert "Failed to list artifacts for run root-exec" in message + assert "Traceback" not in message + + +def test_list_result_artifacts_surfaces_clean_error_on_http_error() -> None: + client = FakeArtifactClient() + client.execution_errors["root-exec"] = requests.HTTPError( + "403 Forbidden", response=FakeDataResponse(b"", status_code=403) + ) + + # An HTTPError propagates so the CLI's clean-error boundary formats it. + with pytest.raises(requests.HTTPError): + _manager(client).list_result_artifacts("root-exec") + + +# --------------------------------------------------------------------------- +# Download: resolution, fallbacks, dedup, filtering +# --------------------------------------------------------------------------- + + +def test_download_result_artifacts_resolves_run_id_to_root_execution(tmp_path) -> None: + client = FakeArtifactClient() + client.run_details["run-1"] = SimpleNamespace( + run=SimpleNamespace(root_execution_id="root-exec-9"), + execution=_execution({}, output_artifacts={"model": {"id": "artifact-model"}}), + ) + client.data_responses = {"artifact-model": b'{"model": true}'} + + artifacts = _manager(client).download_result_artifacts("run-1", out_dir=tmp_path) + + assert set(artifacts) == {"root::model"} + assert artifacts["root::model"].read_bytes() == b'{"model": true}' + assert client.get_run_details_calls == ["run-1"] + assert client.get_execution_details_calls == [] + assert client.make_request_calls == [("GET", "/api/artifacts/artifact-model/data")] + + +def test_download_result_artifacts_filters_and_deduplicates_by_artifact_id(tmp_path) -> None: + client = FakeArtifactClient() + child_execution = _execution( + {}, + output_artifacts={"model": {"id": "artifact-shared"}}, + ) + client.execution_details["root-exec"] = _execution( + {}, + child_executions={"Train Task": child_execution}, + output_artifacts={ + "model": {"id": "artifact-shared"}, + "metrics": {"id": "artifact-metrics"}, + }, + ) + client.data_responses = { + "artifact-shared": b'{"model": true}', + "artifact-metrics": b'{"metrics": true}', + } + + artifacts = _manager(client).download_result_artifacts( + "root-exec", + out_dir=tmp_path, + only=["model"], + include_children=True, + ) + + assert set(artifacts) == {"root::model", "Train Task::model"} + assert artifacts["root::model"] == artifacts["Train Task::model"] + assert artifacts["root::model"].name == "root__model__artifact-sha" + assert artifacts["root::model"].read_bytes() == b'{"model": true}' + assert client.make_request_calls == [("GET", "/api/artifacts/artifact-shared/data")] + + +def test_download_result_artifacts_falls_back_to_metadata_value_after_404(tmp_path) -> None: + client = FakeArtifactClient() + client.execution_details["root-exec"] = _execution( + {}, + output_artifacts={"model": {"id": "artifact-model"}}, + ) + client.data_status_codes["artifact-model"] = 404 + client.artifact_responses["artifact-model"] = SimpleNamespace( + id="artifact-model", + artifact_data=SimpleNamespace(value="inline bytes"), + ) + + artifacts = _manager(client).download_result_artifacts("root-exec", out_dir=tmp_path) + + # Inline values are JSON-encoded and gain a .json suffix at write time. + assert artifacts["root::model"].name == "root__model__artifact-mod.json" + assert artifacts["root::model"].read_bytes() == b'"inline bytes"' + assert client.artifacts_get_calls == ["artifact-model"] + assert client.make_request_calls == [("GET", "/api/artifacts/artifact-model/data")] + + +@pytest.mark.parametrize( + "value, expected", + [ + ({"nested": {"k": "v"}, "n": 1, "ok": True}, b'{"nested": {"k": "v"}, "n": 1, "ok": true}'), + ([1, "two", {"three": 3}], b'[1, "two", {"three": 3}]'), + (42, b"42"), + (True, b"true"), + # An explicit inline null is a real value and must be written as JSON + # ``null`` — not treated as "absent" and routed to the signed URL. The + # client below has no signed URL, so a fall-through would error out. + (None, b"null"), + ], +) +def test_download_json_encodes_structured_inline_value(value: Any, expected: bytes, tmp_path) -> None: + # A structured inline metadata value (dict/list/number/bool/null) must be + # written as valid JSON, not the Python repr that ``str()`` would produce for + # a dict or list (single quotes, ``True``/``None`` tokens) into a ``.json`` file. + client = _single_root_artifact_client() + client.data_status_codes["artifact-model"] = 404 + client.artifact_responses["artifact-model"] = SimpleNamespace( + id="artifact-model", + artifact_data=SimpleNamespace(value=value), + ) + + artifacts = _manager(client).download_result_artifacts("root-exec", out_dir=tmp_path) + + written = artifacts["root::model"].read_bytes() + assert written == expected + assert artifacts["root::model"].name.endswith(".json") + # The written file round-trips as JSON; a str()-based repr would not. + assert json.loads(written) == value + + +def test_download_absent_inline_value_falls_back_to_signed_url(tmp_path) -> None: + # A genuinely absent ``value`` field (no such attribute) is not an inline + # value: the download must fall through to the signed URL rather than + # writing anything. Counterpart to the explicit inline-null case above. + client = _single_root_artifact_client() + client.data_status_codes["artifact-model"] = 404 + client.artifact_responses["artifact-model"] = SimpleNamespace( + id="artifact-model", + artifact_data=SimpleNamespace(), # no ``value`` attribute + ) + client.signed_urls["artifact-model"] = "/api/artifacts/artifact-model/signed-data" + client.data_responses["artifact-model"] = b"relative signed bytes" + + artifacts = _manager(client).download_result_artifacts("root-exec", out_dir=tmp_path) + + assert artifacts["root::model"].read_bytes() == b"relative signed bytes" + assert client.artifacts_signed_url_calls == ["artifact-model"] + assert client.make_request_calls == [ + ("GET", "/api/artifacts/artifact-model/data"), + ("GET", "/api/artifacts/artifact-model/signed-data"), + ] + + +class TypedArtifactData(BaseModel): + """Pydantic stand-in mirroring the generated ``ArtifactData`` model. + + The shipped generated client deserializes ``artifact_data`` as a plain + dict (the response field is typed ``Any``), where ``dict.get`` already + distinguishes absent from explicit null. A custom client may return a + typed model whose ``value`` defaults to ``None`` at class level; these + tests pin that pydantic's set-field tracking keeps the distinction. + """ + + uri: Any = None + value: Any = None + + +def test_download_typed_artifact_data_unset_value_falls_back_to_signed_url(tmp_path) -> None: + # A typed model with ``value`` left at its class default has no inline + # value: the ``None`` default must not be mistaken for an explicit inline + # null and written as a ``null`` .json file. + client = _single_root_artifact_client() + client.data_status_codes["artifact-model"] = 404 + client.artifact_responses["artifact-model"] = SimpleNamespace( + id="artifact-model", + artifact_data=TypedArtifactData(uri="gs://bucket/model"), + ) + client.signed_urls["artifact-model"] = "/api/artifacts/artifact-model/signed-data" + client.data_responses["artifact-model"] = b"signed bytes" + + artifacts = _manager(client).download_result_artifacts("root-exec", out_dir=tmp_path) + + assert artifacts["root::model"].read_bytes() == b"signed bytes" + assert not artifacts["root::model"].name.endswith(".json") + assert client.artifacts_signed_url_calls == ["artifact-model"] + + +def test_download_typed_artifact_data_explicit_null_written_as_json_null(tmp_path) -> None: + # An explicitly-set ``value=None`` on a typed model is a real inline null + # and must be written as JSON ``null``, not routed to the signed URL. The + # client below has no signed URL, so a fall-through would error out. + client = _single_root_artifact_client() + client.data_status_codes["artifact-model"] = 404 + client.artifact_responses["artifact-model"] = SimpleNamespace( + id="artifact-model", + artifact_data=TypedArtifactData(value=None), + ) + + artifacts = _manager(client).download_result_artifacts("root-exec", out_dir=tmp_path) + + assert artifacts["root::model"].name.endswith(".json") + assert artifacts["root::model"].read_bytes() == b"null" + assert client.artifacts_signed_url_calls == [] + + +def test_download_transient_metadata_error_falls_back_to_signed_url(tmp_path) -> None: + # A failure fetching the inline metadata value after a fallback status is a + # best-effort step: the download must continue to the signed URL (the + # authoritative final attempt) rather than stopping on the inline-fetch error. + client = _single_root_artifact_client() + client.data_status_codes["artifact-model"] = 404 + client.artifact_errors["artifact-model"] = requests.ConnectionError("boom") + client.signed_urls["artifact-model"] = "/api/artifacts/artifact-model/signed-data" + client.data_responses["artifact-model"] = b"signed bytes" + + artifacts = _manager(client).download_result_artifacts("root-exec", out_dir=tmp_path) + + assert artifacts["root::model"].read_bytes() == b"signed bytes" + assert client.artifacts_signed_url_calls == ["artifact-model"] + + +@pytest.mark.parametrize("status_code", [403, 410]) +def test_download_result_artifacts_falls_back_to_metadata_value_for_fallback_status( + status_code: int, + tmp_path, +) -> None: + client = FakeArtifactClient() + client.execution_details["root-exec"] = _execution( + {}, + output_artifacts={"model": {"id": "artifact-model"}}, + ) + client.data_status_codes["artifact-model"] = status_code + client.artifact_responses["artifact-model"] = SimpleNamespace( + id="artifact-model", + artifact_data=SimpleNamespace(value="inline bytes"), + ) + + artifacts = _manager(client).download_result_artifacts("root-exec", out_dir=tmp_path) + + assert artifacts["root::model"].read_bytes() == b'"inline bytes"' + assert client.make_request_calls == [("GET", "/api/artifacts/artifact-model/data")] + + +@pytest.mark.parametrize("status_code", [403, 410]) +def test_download_result_artifacts_falls_back_to_signed_url_for_fallback_status( + status_code: int, + tmp_path, +) -> None: + client = FakeArtifactClient() + client.execution_details["root-exec"] = _execution( + {}, + output_artifacts={"model": {"id": "artifact-model"}}, + ) + client.data_status_codes["artifact-model"] = status_code + client.signed_urls["artifact-model"] = "/api/artifacts/artifact-model/signed-data" + client.data_responses["artifact-model"] = b"relative signed bytes" + + artifacts = _manager(client).download_result_artifacts("root-exec", out_dir=tmp_path) + + assert artifacts["root::model"].read_bytes() == b"relative signed bytes" + assert client.make_request_calls == [ + ("GET", "/api/artifacts/artifact-model/data"), + ("GET", "/api/artifacts/artifact-model/signed-data"), + ] + + +def test_download_result_artifacts_reraises_non_fallback_status(tmp_path) -> None: + client = FakeArtifactClient() + client.execution_details["root-exec"] = _execution( + {}, + output_artifacts={"model": {"id": "artifact-model"}}, + ) + client.data_status_codes["artifact-model"] = 500 + + with pytest.raises(RuntimeError) as exc_info: + _manager(client).download_result_artifacts("root-exec", out_dir=tmp_path) + message = str(exc_info.value) + assert "Artifact download failed" in message + assert "/api/artifacts/artifact-model/data" in message + assert "HTTP 500" in message + assert "Traceback" not in message + assert client.artifacts_get_calls == [] + + +def test_download_result_artifacts_uses_client_for_relative_signed_url(tmp_path) -> None: + client = FakeArtifactClient() + client.execution_details["root-exec"] = _execution( + {}, + output_artifacts={"model": {"id": "artifact-model"}}, + ) + client.data_status_codes["artifact-model"] = 404 + client.signed_urls["artifact-model"] = "/api/artifacts/artifact-model/signed-data" + client.data_responses["artifact-model"] = b"relative signed bytes" + + artifacts = _manager(client).download_result_artifacts("root-exec", out_dir=tmp_path) + + assert artifacts["root::model"].read_bytes() == b"relative signed bytes" + assert client.make_request_calls == [ + ("GET", "/api/artifacts/artifact-model/data"), + ("GET", "/api/artifacts/artifact-model/signed-data"), + ] + + +# --------------------------------------------------------------------------- +# Signed-URL failure handling (clean errors, no URL echo) +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("status_code", [403, 404]) +def test_download_signed_url_fallback_failure_surfaces_clean_error( + status_code: int, + tmp_path, +) -> None: + # Direct /data is forbidden/missing and the signed-URL download itself fails. + # The user must see a concise message naming both the direct status and the + # signed-URL status, not a raw requests traceback. + client = FakeArtifactClient() + client.execution_details["root-exec"] = _execution( + {}, + output_artifacts={"model": {"id": "artifact-model"}}, + ) + client.data_status_codes["artifact-model"] = status_code + client.signed_urls["artifact-model"] = "/api/artifacts/artifact-model/signed-data" + client.signed_data_status_codes["artifact-model"] = 500 + + with pytest.raises(RuntimeError) as exc_info: + _manager(client).download_result_artifacts("root-exec", out_dir=tmp_path) + message = str(exc_info.value) + assert "artifact-model download failed" in message + assert f"HTTP {status_code}" in message + assert "signed-URL download returned HTTP 500" in message + assert "Traceback" not in message + + +def test_download_signed_url_request_failure_surfaces_clean_error(tmp_path) -> None: + # /data returns 403 and requesting the signed URL itself errors; the message + # must cover both, with no traceback. + client = FakeArtifactClient() + client.execution_details["root-exec"] = _execution( + {}, + output_artifacts={"model": {"id": "artifact-model"}}, + ) + client.data_status_codes["artifact-model"] = 403 + client.signed_url_errors["artifact-model"] = requests.HTTPError( + "403 Forbidden", response=FakeDataResponse(b"", status_code=403) + ) + + with pytest.raises(RuntimeError) as exc_info: + _manager(client).download_result_artifacts("root-exec", out_dir=tmp_path) + message = str(exc_info.value) + assert "artifact-model download failed" in message + assert "signed-URL request failed: HTTP 403" in message + assert "Traceback" not in message + + +@pytest.mark.parametrize( + "signed_url", + [ + "gs://bucket/artifact-model", + "s3://bucket/artifact-model", + "file:///etc/passwd", + "//storage.example/artifact-model", # protocol-relative + ], +) +def test_download_rejects_non_http_signed_url_cleanly(signed_url: str, tmp_path) -> None: + # A non-http(s) or protocol-relative signed URL must be rejected as a clean + # RuntimeError before it reaches the client's raw request (whose relative-URL + # join raises a ValueError that would escape the CLI as a traceback). The + # error must not echo the signed URL, which can carry credentials. + client = _single_root_artifact_client() + client.data_status_codes["artifact-model"] = 404 + client.signed_urls["artifact-model"] = signed_url + + with pytest.raises(RuntimeError) as exc_info: + _manager(client).download_result_artifacts("root-exec", out_dir=tmp_path) + message = str(exc_info.value) + assert "unsupported signed URL" in message + assert "Traceback" not in message + assert signed_url not in message + # Only the direct /data probe was issued; the bad URL was never fetched. + assert client.make_request_calls == [("GET", "/api/artifacts/artifact-model/data")] + + +# --------------------------------------------------------------------------- +# Signed-URL fetch: unauthenticated absolute GET, redirect handling +# --------------------------------------------------------------------------- + + +def test_download_result_artifacts_uses_unauthenticated_get_for_absolute_signed_url( + monkeypatch, + tmp_path, +) -> None: + client = FakeArtifactClient() + client.execution_details["root-exec"] = _execution( + {}, + output_artifacts={"model": {"id": "artifact-model"}}, + ) + client.data_status_codes["artifact-model"] = 404 + client.signed_urls["artifact-model"] = "https://storage.example/artifact-model" + external_get_calls: list[tuple[str, dict[str, Any]]] = [] + + def fake_get(url: str, **kwargs: Any) -> FakeDataResponse: + external_get_calls.append((url, kwargs)) + return FakeDataResponse(b"signed bytes") + + monkeypatch.setattr("tangle_cli.artifacts.requests.get", fake_get) + + artifacts = _manager(client).download_result_artifacts("root-exec", out_dir=tmp_path) + + assert artifacts["root::model"].read_bytes() == b"signed bytes" + assert external_get_calls == [ + ( + "https://storage.example/artifact-model", + {"timeout": 60, "stream": True, "allow_redirects": False}, + ) + ] + assert client.make_request_calls == [("GET", "/api/artifacts/artifact-model/data")] + + +def test_download_falls_back_to_signed_url_on_cross_origin_redirect(tmp_path, monkeypatch) -> None: + # The direct /data route 302-redirects to cross-origin storage. The client + # refuses to forward Tangle credentials off-origin and raises an HTTPError + # carrying the redirect response *before* a body is returned. The download + # must not stop at the 302: it falls back to the signed URL, which for an + # absolute storage URL is fetched with no auth headers (no credential leak). + client = _single_root_artifact_client() + client.signed_urls["artifact-model"] = "https://storage.example/artifact-model" + + redirect_response = FakeDataResponse(b"", status_code=302) + original_make_request = client.request_raw + + def make_request(method: str, path: str, **kwargs: Any) -> Any: + if path.endswith("/data"): + raise requests.HTTPError( + "Refusing to follow cross-origin redirect from ... to ...", + response=redirect_response, + ) + return original_make_request(method, path, **kwargs) + + client.request_raw = make_request # type: ignore[method-assign] + + external_get_calls: list[tuple[str, dict[str, Any]]] = [] + + def fake_get(url: str, **kwargs: Any) -> FakeDataResponse: + external_get_calls.append((url, kwargs)) + return FakeDataResponse(b"storage bytes") + + monkeypatch.setattr("tangle_cli.artifacts.requests.get", fake_get) + + artifacts = _manager(client).download_result_artifacts("root-exec", out_dir=tmp_path) + + assert artifacts["root::model"].read_bytes() == b"storage bytes" + # The off-origin storage URL is fetched with only timeout/stream — no auth + # headers, cookies, or tokens are forwarded across origins. + assert external_get_calls == [ + ( + "https://storage.example/artifact-model", + {"timeout": 60, "stream": True, "allow_redirects": False}, + ) + ] + assert client.artifacts_signed_url_calls == ["artifact-model"] + + +def test_download_signed_url_redirect_to_non_http_target_rejected_cleanly( + monkeypatch, + tmp_path, +) -> None: + # An absolute signed URL that redirects to a non-http(s) target must be + # rejected against the same allowlist that admitted the initial URL — an + # automatic follow would raise requests.InvalidSchema, whose message echoes + # the redirect URL (which can carry signed credentials). + client = _single_root_artifact_client() + client.data_status_codes["artifact-model"] = 404 + client.signed_urls["artifact-model"] = "https://storage.example/artifact-model" + redirect_target = "file:///etc/passwd" + external_get_calls: list[tuple[str, dict[str, Any]]] = [] + + def fake_get(url: str, **kwargs: Any) -> FakeDataResponse: + external_get_calls.append((url, kwargs)) + return FakeDataResponse(b"", status_code=302, headers={"Location": redirect_target}) + + monkeypatch.setattr("tangle_cli.artifacts.requests.get", fake_get) + + with pytest.raises(RuntimeError) as exc_info: + _manager(client).download_result_artifacts("root-exec", out_dir=tmp_path) + message = str(exc_info.value) + assert "Artifact artifact-model download failed" in message + assert "unsupported target" in message + assert redirect_target not in message + assert "Traceback" not in message + # Only the initial signed URL was fetched; the disallowed target never was. + assert [url for url, _ in external_get_calls] == ["https://storage.example/artifact-model"] + + +def test_download_signed_url_follows_http_redirect_with_recheck( + monkeypatch, + tmp_path, +) -> None: + # An http(s) → http(s) signed-URL redirect is followed, but manually: every + # hop is fetched with allow_redirects=False, unauthenticated, and rechecked + # against the http(s) allowlist. + client = _single_root_artifact_client() + client.data_status_codes["artifact-model"] = 404 + client.signed_urls["artifact-model"] = "https://storage.example/artifact-model" + responses = [ + FakeDataResponse( + b"", status_code=302, headers={"Location": "https://cdn.example/blob"} + ), + FakeDataResponse(b"redirected bytes"), + ] + external_get_calls: list[tuple[str, dict[str, Any]]] = [] + + def fake_get(url: str, **kwargs: Any) -> FakeDataResponse: + external_get_calls.append((url, kwargs)) + return responses[len(external_get_calls) - 1] + + monkeypatch.setattr("tangle_cli.artifacts.requests.get", fake_get) + + artifacts = _manager(client).download_result_artifacts("root-exec", out_dir=tmp_path) + + assert artifacts["root::model"].read_bytes() == b"redirected bytes" + expected_kwargs = {"timeout": 60, "stream": True, "allow_redirects": False} + assert external_get_calls == [ + ("https://storage.example/artifact-model", expected_kwargs), + ("https://cdn.example/blob", expected_kwargs), + ] + # The redirect response was closed once its Location was consumed. + assert responses[0].closed + + +def test_download_signed_url_redirect_loop_surfaces_clean_error( + monkeypatch, + tmp_path, +) -> None: + client = _single_root_artifact_client() + client.data_status_codes["artifact-model"] = 404 + client.signed_urls["artifact-model"] = "https://storage.example/artifact-model" + + def fake_get(url: str, **kwargs: Any) -> FakeDataResponse: + return FakeDataResponse(b"", status_code=302, headers={"Location": url}) + + monkeypatch.setattr("tangle_cli.artifacts.requests.get", fake_get) + + with pytest.raises(RuntimeError) as exc_info: + _manager(client).download_result_artifacts("root-exec", out_dir=tmp_path) + message = str(exc_info.value) + assert "exceeded" in message + assert "redirects" in message + assert "Traceback" not in message + + +# --------------------------------------------------------------------------- +# Internal-host guard (SSRF) +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "signed_url", + [ + "http://127.0.0.1/artifact-model", + "http://127.9.8.7:9000/artifact-model", + "HTTP://LOCALHOST/artifact-model", + "http://internal.localhost/artifact-model", + "http://[::1]/artifact-model", + "http://[::ffff:127.0.0.1]/artifact-model", + "http://169.254.169.254/latest/meta-data", + "http://[fe80::1]/artifact-model", + "http://10.0.0.5/artifact-model", + "http://172.16.0.5:8443/artifact-model", + "http://192.168.1.5/artifact-model", + "http://100.64.0.1/artifact-model", + "http://100.127.255.254/artifact-model", + "http://[fd00::5]/artifact-model", + "http://0.0.0.0/artifact-model", + "http://127.0.0.1./artifact-model", + "http://localhost./artifact-model", + "http://[64:ff9b::a9fe:a9fe]/artifact-model", + "http://[64:ff9b::7f00:1]/artifact-model", + "http://[64:ff9b:1::7f00:1]/artifact-model", + ], +) +def test_download_rejects_internal_signed_url_host_by_default( + signed_url: str, + monkeypatch, + tmp_path, +) -> None: + # A signed URL aimed at an internal host (loopback, link-local/metadata, + # RFC1918, CGNAT, unique-local, unspecified) must be rejected before any fetch — + # a hostile signed URL must not turn the CLI into a local-network SSRF + # client. The message names the host and the override, never the full URL. + monkeypatch.delenv("TANGLE_ALLOW_INTERNAL_ARTIFACT_HOSTS", raising=False) + client = _single_root_artifact_client() + client.data_status_codes["artifact-model"] = 404 + client.signed_urls["artifact-model"] = signed_url + external_get_calls: list[str] = [] + + def fake_get(url: str, **kwargs: Any) -> FakeDataResponse: + external_get_calls.append(url) + return FakeDataResponse(b"must not be fetched") + + monkeypatch.setattr("tangle_cli.artifacts.requests.get", fake_get) + + with pytest.raises(RuntimeError) as exc_info: + _manager(client).download_result_artifacts("root-exec", out_dir=tmp_path) + message = str(exc_info.value) + assert "Artifact artifact-model download failed" in message + assert "is an internal address" in message + assert "TANGLE_ALLOW_INTERNAL_ARTIFACT_HOSTS" in message + assert "Traceback" not in message + assert signed_url not in message + assert external_get_calls == [] + + +def test_download_allows_internal_signed_url_host_with_override( + monkeypatch, + tmp_path, +) -> None: + # Local/dev/kind stands legitimately serve loopback signed URLs; the env + # override restores that behavior, including for redirect hops. + monkeypatch.setenv("TANGLE_ALLOW_INTERNAL_ARTIFACT_HOSTS", "1") + client = _single_root_artifact_client() + client.data_status_codes["artifact-model"] = 404 + client.signed_urls["artifact-model"] = "http://127.0.0.1:9000/artifact-model" + responses = [ + FakeDataResponse( + b"", status_code=302, headers={"Location": "http://127.0.0.1:9001/blob"} + ), + FakeDataResponse(b"loopback bytes"), + ] + external_get_calls: list[str] = [] + + def fake_get(url: str, **kwargs: Any) -> FakeDataResponse: + external_get_calls.append(url) + return responses[len(external_get_calls) - 1] + + monkeypatch.setattr("tangle_cli.artifacts.requests.get", fake_get) + + artifacts = _manager(client).download_result_artifacts("root-exec", out_dir=tmp_path) + + assert artifacts["root::model"].read_bytes() == b"loopback bytes" + assert external_get_calls == [ + "http://127.0.0.1:9000/artifact-model", + "http://127.0.0.1:9001/blob", + ] + + +def test_download_rejects_signed_url_redirect_to_internal_host( + monkeypatch, + tmp_path, +) -> None: + # An external signed URL must not be able to hop to an internal host via a + # redirect: each hop is rechecked, and the internal target is never fetched. + monkeypatch.delenv("TANGLE_ALLOW_INTERNAL_ARTIFACT_HOSTS", raising=False) + client = _single_root_artifact_client() + client.data_status_codes["artifact-model"] = 404 + client.signed_urls["artifact-model"] = "https://storage.example/artifact-model" + external_get_calls: list[str] = [] + + def fake_get(url: str, **kwargs: Any) -> FakeDataResponse: + external_get_calls.append(url) + return FakeDataResponse( + b"", status_code=302, headers={"Location": "http://169.254.169.254/latest"} + ) + + monkeypatch.setattr("tangle_cli.artifacts.requests.get", fake_get) + + with pytest.raises(RuntimeError) as exc_info: + _manager(client).download_result_artifacts("root-exec", out_dir=tmp_path) + message = str(exc_info.value) + assert "signed URL redirected to host" in message + assert "is an internal address" in message + assert "Traceback" not in message + assert external_get_calls == ["https://storage.example/artifact-model"] + + +@pytest.mark.parametrize( + "signed_url", + [ + "http://2130706433/artifact-model", + "http://0177.0.0.1/artifact-model", + "http://0x7f000001/artifact-model", + "HTTP://0X7F000001/artifact-model", + "http://0x7f.0.0.1/artifact-model", + "http://017700000001/artifact-model", + "http://127.1/artifact-model", + "http://0/artifact-model", + "http://0x0/artifact-model", + "http://134744072/artifact-model", + ], +) +def test_download_rejects_non_canonical_numeric_signed_url_host( + signed_url: str, + monkeypatch, + tmp_path, +) -> None: + # The socket layer's inet_aton semantics accept decimal/octal/hex and + # dotted short-form IPv4 spellings that ipaddress rejects, so these could + # smuggle a loopback/metadata address past a canonical-literal check. Any + # non-canonical numeric host is rejected outright — including spellings of + # external addresses (134744072 is 8.8.8.8): no legitimate signed URL + # spells its host that way. + monkeypatch.delenv("TANGLE_ALLOW_INTERNAL_ARTIFACT_HOSTS", raising=False) + client = _single_root_artifact_client() + client.data_status_codes["artifact-model"] = 404 + client.signed_urls["artifact-model"] = signed_url + external_get_calls: list[str] = [] + + def fake_get(url: str, **kwargs: Any) -> FakeDataResponse: + external_get_calls.append(url) + return FakeDataResponse(b"must not be fetched") + + monkeypatch.setattr("tangle_cli.artifacts.requests.get", fake_get) + + with pytest.raises(RuntimeError) as exc_info: + _manager(client).download_result_artifacts("root-exec", out_dir=tmp_path) + message = str(exc_info.value) + assert "is a non-canonical numeric address" in message + assert "TANGLE_ALLOW_INTERNAL_ARTIFACT_HOSTS" in message + assert "Traceback" not in message + assert external_get_calls == [] + + +@pytest.mark.parametrize( + "host", + [ + "8.8.8.8", + "storage.example", + "storage.example.", + "100.63.255.255", # just below the CGNAT range + "100.128.0.0", # just above the CGNAT range + "2600::1", + "64:ff9b::808:808", # NAT64-embedded 8.8.8.8 is external + "::ffff:0x7f000001", # not parseable by the socket layer either + "", + None, + ], +) +def test_host_rejection_reason_allows_external_hosts(host: str | None) -> None: + assert tangle_artifacts._host_rejection_reason(host) is None + + +# --------------------------------------------------------------------------- +# Download hardening: streaming, safe writes, collision/overwrite/symlink +# --------------------------------------------------------------------------- + + +class _DiskFullHandle: + """Real artifact file handle whose writes fail like a full disk.""" + + def __init__(self, handle: Any) -> None: + self._handle = handle + + def write(self, data: bytes) -> int: + raise OSError(28, "No space left on device") + + def close(self) -> None: + self._handle.close() + + +def test_download_mid_stream_write_failure_surfaces_clean_error( + monkeypatch, + tmp_path, +) -> None: + # A local write failure mid-download (e.g. disk full) must surface as a + # concise RuntimeError — the CLI only translates RuntimeError into a clean + # error, so a raw OSError would escape as a traceback — and the partial + # file must be cleaned up. + client = _single_root_artifact_client() + client.data_responses["artifact-model"] = b"artifact bytes" + real_open = tangle_artifacts._open_new_artifact_file + monkeypatch.setattr( + tangle_artifacts, + "_open_new_artifact_file", + lambda dest: _DiskFullHandle(real_open(dest)), + ) + + with pytest.raises(RuntimeError) as exc_info: + _manager(client).download_result_artifacts("root-exec", out_dir=tmp_path) + message = str(exc_info.value) + assert "Cannot write artifact to" in message + assert "No space left on device" in message + assert "Traceback" not in message + # The partially-created file was removed. + assert list(tmp_path.iterdir()) == [] + + +def test_download_streams_to_disk_without_buffering_full_body(tmp_path) -> None: + client = _single_root_artifact_client() + client.stream_responses["artifact-model"] = StreamingOnlyResponse( + [b"chunk-a", b"chunk-b", b"chunk-c"] + ) + + artifacts = _manager(client).download_result_artifacts("root-exec", out_dir=tmp_path) + + # The body is reassembled from streamed chunks; the production path never + # touches ``response.content`` (which would raise here). + assert artifacts["root::model"].read_bytes() == b"chunk-achunk-bchunk-c" + assert client.stream_responses["artifact-model"].iter_calls # streaming was used + + +class _FailingStreamResponse: + """200 response whose body read fails mid-stream like a dropped connection.""" + + def __init__(self, message: str = "connection reset") -> None: + self.status_code = 200 + self.closed = False + self._message = message + + def raise_for_status(self) -> None: + return None + + def iter_content(self, chunk_size: int = 1) -> Any: + yield b"partial" + raise requests.ConnectionError(self._message) + + def close(self) -> None: + self.closed = True + + +# Fake signed-URL credential embedded in transport-error messages below. Real +# ``requests`` transport errors echo the request URL (``Max retries exceeded +# with url: ...``), so signed-URL failure paths must not include ``str(exc)``. +_SIGNED_URL_SECRET = "X-Signature=super-secret-signature" + + +def test_download_direct_stream_transport_error_is_clean(tmp_path) -> None: + client = _single_root_artifact_client() + client.stream_responses["artifact-model"] = _FailingStreamResponse() + + with pytest.raises(RuntimeError) as exc_info: + _manager(client).download_result_artifacts("root-exec", out_dir=tmp_path) + message = str(exc_info.value) + assert "streaming failed" in message + assert "Traceback" not in message + # The partial file is cleaned up. + assert list(tmp_path.iterdir()) == [] + + +def test_download_signed_url_stream_transport_error_is_clean(tmp_path, monkeypatch) -> None: + client = _single_root_artifact_client() + client.data_status_codes["artifact-model"] = 404 + client.signed_urls["artifact-model"] = "https://storage.example/artifact-model" + monkeypatch.setattr( + "tangle_cli.artifacts.requests.get", + lambda *a, **k: _FailingStreamResponse( + f"Connection broken for url: https://storage.example/artifact-model?{_SIGNED_URL_SECRET}" + ), + ) + + with pytest.raises(RuntimeError) as exc_info: + _manager(client).download_result_artifacts("root-exec", out_dir=tmp_path) + message = str(exc_info.value) + assert "signed-URL download streaming failed: ConnectionError" in message + assert _SIGNED_URL_SECRET not in message + assert "Traceback" not in message + assert list(tmp_path.iterdir()) == [] + + +@pytest.mark.parametrize( + "exc", [requests.ConnectionError("dns failure"), requests.Timeout("timed out")] +) +def test_download_direct_data_pre_response_transport_error_is_clean( + exc: Exception, tmp_path +) -> None: + # The initial direct /data request fails before returning a response + # (DNS/TLS/timeout/connection). The raw requests exception must not escape. + client = _single_root_artifact_client() + + def make_request(method: str, path: str, **kwargs: Any) -> Any: + raise exc + + client.request_raw = make_request # type: ignore[method-assign] + + with pytest.raises(RuntimeError) as exc_info: + _manager(client).download_result_artifacts("root-exec", out_dir=tmp_path) + message = str(exc_info.value) + assert "GET /api/artifacts/artifact-model/data transport failed" in message + assert "Traceback" not in message + assert list(tmp_path.iterdir()) == [] + + +@pytest.mark.parametrize( + "exc, detail", + [ + ( + requests.ConnectionError( + f"Max retries exceeded with url: /artifact-model?{_SIGNED_URL_SECRET}" + ), + "ConnectionError", + ), + (requests.Timeout(f"timed out for url: /artifact-model?{_SIGNED_URL_SECRET}"), "Timeout"), + ], +) +def test_download_absolute_signed_url_pre_response_transport_error_is_clean( + exc: Exception, detail: str, tmp_path, monkeypatch +) -> None: + # Absolute signed URL fetch fails before any response object exists (DNS, + # TLS, timeout). The raw requests exception must not escape the CLI's + # RuntimeError handler, and its message (which echoes the request URL, + # including signed credentials) must not be included. + client = _single_root_artifact_client() + client.data_status_codes["artifact-model"] = 404 + client.signed_urls["artifact-model"] = "https://storage.example/artifact-model" + + def fail_get(*args: Any, **kwargs: Any) -> Any: + raise exc + + monkeypatch.setattr("tangle_cli.artifacts.requests.get", fail_get) + + with pytest.raises(RuntimeError) as exc_info: + _manager(client).download_result_artifacts("root-exec", out_dir=tmp_path) + message = str(exc_info.value) + assert f"signed-URL transport failed: {detail}" in message + assert "GET /api/artifacts/artifact-model/data returned HTTP 404" in message + assert _SIGNED_URL_SECRET not in message + assert "Traceback" not in message + assert list(tmp_path.iterdir()) == [] + + +def test_download_relative_signed_url_pre_response_transport_error_is_clean(tmp_path) -> None: + # Relative signed path fetch via the client raises before a response. + client = _single_root_artifact_client() + client.data_status_codes["artifact-model"] = 404 + client.signed_urls["artifact-model"] = "/api/artifacts/artifact-model/signed-data" + + original_make_request = client.request_raw + + def make_request(method: str, path: str, **kwargs: Any) -> Any: + if path.endswith("/signed-data"): + raise requests.ConnectionError( + f"connection refused for url: {path}?{_SIGNED_URL_SECRET}" + ) + return original_make_request(method, path, **kwargs) + + client.request_raw = make_request # type: ignore[method-assign] + + with pytest.raises(RuntimeError) as exc_info: + _manager(client).download_result_artifacts("root-exec", out_dir=tmp_path) + message = str(exc_info.value) + assert "signed-URL transport failed: ConnectionError" in message + assert _SIGNED_URL_SECRET not in message + assert "Traceback" not in message + assert list(tmp_path.iterdir()) == [] + + +@pytest.mark.parametrize( + "artifact_id, encoded", + [ + ("a/b", "a%2Fb"), + ("../secret", "..%2Fsecret"), + ("a?b#c", "a%3Fb%23c"), + ("with space", "with%20space"), + ], +) +def test_download_percent_encodes_artifact_id_path_segment( + artifact_id: str, encoded: str, tmp_path +) -> None: + # A hostile/unusual artifact id must be percent-encoded as a single path + # segment (matching the generated client) so it cannot alter the request + # path structure or inject query/fragment components. + client = FakeArtifactClient() + client.execution_details["root-exec"] = _execution( + {}, output_artifacts={"out": {"id": artifact_id}} + ) + + _manager(client).download_result_artifacts("root-exec", out_dir=tmp_path) + + expected = f"/api/artifacts/{encoded}/data" + assert ("GET", expected) in client.make_request_calls + # The raw id is never interpolated unescaped, and there is exactly one + # artifacts path prefix (no extra segments / query / fragment injected). + requested = [path for _, path in client.make_request_calls] + assert f"/api/artifacts/{artifact_id}/data" not in requested + assert all(path.count("/api/artifacts/") == 1 for path in requested) + assert all("?" not in path and "#" not in path for path in requested) + + +def test_download_relative_signed_url_is_not_double_encoded(tmp_path) -> None: + # A backend-supplied relative signed URL is passed to the client verbatim; + # any percent-escapes it already contains must not be re-encoded. + client = FakeArtifactClient() + client.execution_details["root-exec"] = _execution( + {}, output_artifacts={"model": {"id": "artifact-model"}} + ) + client.data_status_codes["artifact-model"] = 404 + # An already-escaped relative signed path (note the literal %2F). + signed_path = "/api/artifacts/artifact%2Fmodel/signed-data" + client.signed_urls["artifact-model"] = signed_path + client.data_responses["artifact%2Fmodel"] = b"signed bytes" + + artifacts = _manager(client).download_result_artifacts("root-exec", out_dir=tmp_path) + + assert artifacts["root::model"].read_bytes() == b"signed bytes" + assert ("GET", signed_path) in client.make_request_calls + # The pre-existing %2F must not become %252F. + assert "%252F" not in "".join(p for _, p in client.make_request_calls) + + +def test_download_out_dir_is_existing_file_is_clean_error(tmp_path) -> None: + # --out-dir pointing at an existing file must surface a concise error, not a + # raw FileExistsError/NotADirectoryError traceback. + client = _single_root_artifact_client() + client.data_responses = {"artifact-model": b"bytes"} + out_file = tmp_path / "not-a-dir" + out_file.write_text("i am a file", encoding="utf-8") + + with pytest.raises(RuntimeError) as exc_info: + _manager(client).download_result_artifacts("root-exec", out_dir=out_file) + message = str(exc_info.value) + assert "Cannot create output directory" in message + assert "Traceback" not in message + # No download was attempted. + assert client.make_request_calls == [] + + +def test_download_refuses_to_overwrite_existing_file(tmp_path) -> None: + client = _single_root_artifact_client() + client.data_responses = {"artifact-model": b"new bytes"} + dest = tmp_path / "root__model__artifact-mod" + dest.write_bytes(b"original") + + with pytest.raises(RuntimeError, match="Refusing to overwrite existing path"): + _manager(client).download_result_artifacts("root-exec", out_dir=tmp_path) + # The pre-existing file is left untouched. + assert dest.read_bytes() == b"original" + + +def test_download_refuses_directory_target(tmp_path) -> None: + client = _single_root_artifact_client() + client.data_responses = {"artifact-model": b"new bytes"} + (tmp_path / "root__model__artifact-mod").mkdir() + + with pytest.raises(RuntimeError, match="Refusing to overwrite existing path"): + _manager(client).download_result_artifacts("root-exec", out_dir=tmp_path) + + +def test_download_refuses_symlink_target_without_following(tmp_path) -> None: + client = _single_root_artifact_client() + client.data_responses = {"artifact-model": b"new bytes"} + outside = tmp_path / "outside.txt" + outside.write_bytes(b"secret") + link = tmp_path / "root__model__artifact-mod" + link.symlink_to(outside) + + with pytest.raises(RuntimeError, match="Refusing to overwrite existing path"): + _manager(client).download_result_artifacts("root-exec", out_dir=tmp_path) + # The symlink target must not be written through. + assert outside.read_bytes() == b"secret" + + +def test_download_collision_between_distinct_artifacts_is_refused(tmp_path) -> None: + # Two distinct artifact ids that sanitize/truncate to the same filename + # (same owner+output+12-char prefix) must not silently overwrite each other. + client = FakeArtifactClient() + child = _execution( + {}, + output_artifacts={"out": {"id": "abcdefghijkl-second"}}, + ) + client.execution_details["root-exec"] = _execution( + {}, + child_executions={"root": child}, + output_artifacts={"out": {"id": "abcdefghijkl-first"}}, + ) + client.data_responses = { + "abcdefghijkl-first": b"first", + "abcdefghijkl-second": b"second", + } + + with pytest.raises(RuntimeError, match="already used by artifact"): + _manager(client).download_result_artifacts( + "root-exec", out_dir=tmp_path, include_children=True + ) + + +def test_download_sanitizes_artifact_id_and_stays_in_out_dir(tmp_path) -> None: + # A hostile artifact id / owner containing path separators must not escape + # the output directory; every filename segment is sanitized. + client = FakeArtifactClient() + client.execution_details["root-exec"] = _execution( + {}, + child_executions={ + "../escape": _execution({}, output_artifacts={"o": {"id": "../../etc/pwn"}}) + }, + output_artifacts={}, + ) + # The direct /data request percent-encodes the id, so the fake client sees + # the encoded segment as the lookup key. + client.data_responses = {"..%2F..%2Fetc%2Fpwn": b"payload"} + + artifacts = _manager(client).download_result_artifacts( + "root-exec", out_dir=tmp_path, include_children=True + ) + + (written,) = artifacts.values() + assert written.parent == tmp_path + assert written.read_bytes() == b"payload" + # The request path encodes the traversal id as one segment. + assert ("GET", "/api/artifacts/..%2F..%2Fetc%2Fpwn/data") in client.make_request_calls + # No traversal outside the output directory. + assert not (tmp_path.parent / "etc").exists() diff --git a/tests/test_artifacts_cli.py b/tests/test_artifacts_cli.py index b7b3ec1..6b9d3cb 100644 --- a/tests/test_artifacts_cli.py +++ b/tests/test_artifacts_cli.py @@ -4,10 +4,16 @@ import importlib import json import sys +from pathlib import Path +from types import SimpleNamespace from typing import Any +import pytest +import requests + from tangle_cli import artifacts as artifacts_module from tangle_cli import artifacts_cli, cli +from tangle_cli.logger import get_default_logger def run_app(app: Any, args: list[str]) -> None: @@ -18,6 +24,28 @@ def run_app(app: Any, args: list[str]) -> None: raise +def _stub_client(monkeypatch: Any) -> list[dict[str, Any]]: + """Stub client construction so mode tests never touch the network.""" + + client_calls: list[dict[str, Any]] = [] + + def fake_client_from_options(**kwargs: Any) -> object: + client_calls.append(kwargs) + return object() + + monkeypatch.setattr(artifacts_cli, "LazyTangleApiClient", fake_client_from_options) + return client_calls + + +def _run_expecting_error(app: Any, args: list[str]) -> int: + try: + app(args) + except SystemExit as exc: + assert exc.code not in (0, None) + return exc.code # type: ignore[return-value] + raise AssertionError("expected the command to fail") + + def test_sdk_artifacts_help_lists_get_only(capsys) -> None: app = cli.build_app() @@ -80,6 +108,7 @@ def fake_get_artifacts(self, run_id: str, query: dict[str, Any]) -> dict[str, ob "header": ["X-Config: yes"], "include_env_credentials": False, "command_name": "artifact commands", + "logger": get_default_logger(), } ] assert get_calls == [ @@ -195,3 +224,440 @@ def guarded_import(name: str, *args: Any, **kwargs: Any) -> Any: module = importlib.import_module("tangle_cli.artifacts_cli") assert module.app.name == ("artifacts",) + + +# --------------------------------------------------------------------------- +# --list mode +# --------------------------------------------------------------------------- + + +def test_sdk_artifacts_list_mode_calls_manager_and_reports_rows(monkeypatch, capsys) -> None: + _stub_client(monkeypatch) + list_calls: list[dict[str, Any]] = [] + + def fake_list(self, run_id: str, *, include_children: bool) -> list[dict[str, str]]: + list_calls.append({"run_id": run_id, "include_children": include_children}) + return [{"owner": "root", "output": "model", "artifact_id": "artifact-model"}] + + monkeypatch.setattr(artifacts_module.ArtifactManager, "list_result_artifacts", fake_list) + + app = cli.build_app() + run_app(app, ["sdk", "artifacts", "get", "run-1", "--list", "--include-children"]) + + result = json.loads(capsys.readouterr().out) + assert result == { + "status": "success", + "run_id": "run-1", + "count": 1, + "artifacts": [{"owner": "root", "output": "model", "artifact_id": "artifact-model"}], + } + assert list_calls == [{"run_id": "run-1", "include_children": True}] + + +# --------------------------------------------------------------------------- +# --out-dir mode +# --------------------------------------------------------------------------- + + +def test_sdk_artifacts_download_mode_reports_resolved_paths(monkeypatch, tmp_path, capsys) -> None: + _stub_client(monkeypatch) + download_calls: list[dict[str, Any]] = [] + written = tmp_path / "root__model__artifact-mod.json" + written.write_bytes(b"{}") + + def fake_download( + self, run_id: str, *, out_dir: Any, only: Any, include_children: bool + ) -> dict[str, Path]: + download_calls.append( + { + "run_id": run_id, + "out_dir": out_dir, + "only": only, + "include_children": include_children, + } + ) + return {"root::model": written} + + monkeypatch.setattr(artifacts_module.ArtifactManager, "download_result_artifacts", fake_download) + + app = cli.build_app() + run_app( + app, + [ + "sdk", + "artifacts", + "get", + "run-1", + "--out-dir", + str(tmp_path), + "--only", + "model", + "--include-children", + ], + ) + + result = json.loads(capsys.readouterr().out) + assert result == { + "status": "success", + "run_id": "run-1", + "count": 1, + "out_dir": str(tmp_path.resolve()), + "artifacts": {"root::model": str(written.resolve())}, + } + assert download_calls == [ + { + "run_id": "run-1", + "out_dir": str(tmp_path), + "only": ["model"], + "include_children": True, + } + ] + + +def test_sdk_artifacts_download_end_to_end_writes_streamed_bytes(monkeypatch, tmp_path, capsys) -> None: + # Full command path with only the client boundary faked: the real + # ArtifactManager resolves the run, streams the direct /data response, and + # the CLI reports the written path. + class _StreamResponse: + status_code = 200 + + def raise_for_status(self) -> None: + pass + + def iter_content(self, chunk_size: int): + yield b'{"model":' + yield b" true}" + + def close(self) -> None: + pass + + class _FakeClient: + def __init__(self, **kwargs: Any) -> None: + pass + + def get_run_details(self, run_id: str) -> Any: + return { + "execution": {"output_artifacts": {"model": {"id": "artifact-model"}}} + } + + def request_raw(self, method: str, path: str, **kwargs: Any) -> Any: + assert (method, path) == ("GET", "/api/artifacts/artifact-model/data") + return _StreamResponse() + + monkeypatch.setattr(artifacts_cli, "LazyTangleApiClient", _FakeClient) + + app = cli.build_app() + run_app(app, ["sdk", "artifacts", "get", "run-1", "--out-dir", str(tmp_path)]) + + result = json.loads(capsys.readouterr().out) + # Streamed downloads keep the bare filename; only inline JSON values gain .json. + written = tmp_path / "root__model__artifact-mod" + assert result == { + "status": "success", + "run_id": "run-1", + "count": 1, + "out_dir": str(tmp_path.resolve()), + "artifacts": {"root::model": str(written.resolve())}, + } + assert written.read_bytes() == b'{"model": true}' + + +# --------------------------------------------------------------------------- +# API error surfacing +# --------------------------------------------------------------------------- + + +def test_sdk_artifacts_http_error_reports_clean_json_error(monkeypatch, capsys) -> None: + # An HTTPError from a metadata endpoint must surface as a one-line JSON + # error with a nonzero exit, not a raw requests traceback. + _stub_client(monkeypatch) + response = SimpleNamespace( + status_code=500, + reason="Internal Server Error", + url="https://api.test/api/executions/root-exec/details", + text="backend exploded", + request=SimpleNamespace( + method="GET", url="https://api.test/api/executions/root-exec/details" + ), + ) + + def fake_list(self, run_id: str, *, include_children: bool) -> list[dict[str, str]]: + raise requests.HTTPError("500 Server Error", response=response) # type: ignore[arg-type] + + monkeypatch.setattr(artifacts_module.ArtifactManager, "list_result_artifacts", fake_list) + + app = cli.build_app() + payload = _error_output(app, ["sdk", "artifacts", "get", "run-1", "--list"], capsys) + assert payload["status"] == "error" + assert ( + "Tangle API request failed (500 Internal Server Error) for " + "GET https://api.test/api/executions/root-exec/details: backend exploded" + ) == payload["error"] + + +def test_sdk_artifacts_transport_error_reports_clean_json_error(monkeypatch, capsys) -> None: + _stub_client(monkeypatch) + + def fake_list(self, run_id: str, *, include_children: bool) -> list[dict[str, str]]: + raise requests.ConnectionError("connection refused") + + monkeypatch.setattr(artifacts_module.ArtifactManager, "list_result_artifacts", fake_list) + + app = cli.build_app() + payload = _error_output(app, ["sdk", "artifacts", "get", "run-1", "--list"], capsys) + assert payload["status"] == "error" + assert payload["error"] == "Tangle API request failed: connection refused" + + +# --------------------------------------------------------------------------- +# Mode validation (fails before client construction / any network call) +# --------------------------------------------------------------------------- + + +def _error_output(app: Any, args: list[str], capsys: Any) -> dict[str, Any]: + _run_expecting_error(app, args) + return json.loads(capsys.readouterr().out) + + +def test_mode_validation_rejects_query_and_list_together(monkeypatch, capsys) -> None: + client_calls = _stub_client(monkeypatch) + app = cli.build_app() + + payload = _error_output( + app, + ["sdk", "artifacts", "get", "run-1", "--query", '{"artifact_ids": ["a"]}', "--list"], + capsys, + ) + assert payload["status"] == "error" + assert "mutually exclusive" in payload["error"] + # Validation short-circuits before any client is built. + assert client_calls == [] + + +def test_mode_validation_rejects_no_mode(monkeypatch, capsys) -> None: + client_calls = _stub_client(monkeypatch) + app = cli.build_app() + + payload = _error_output(app, ["sdk", "artifacts", "get", "run-1"], capsys) + assert payload["status"] == "error" + assert "--query is required unless --list or --out-dir is set" in payload["error"] + assert client_calls == [] + + +def test_mode_validation_rejects_only_without_out_dir(monkeypatch, capsys) -> None: + client_calls = _stub_client(monkeypatch) + app = cli.build_app() + + payload = _error_output( + app, + ["sdk", "artifacts", "get", "run-1", "--list", "--only", "model"], + capsys, + ) + assert payload["status"] == "error" + assert "--only is only valid with --out-dir" in payload["error"] + assert client_calls == [] + + +def test_mode_validation_rejects_include_children_with_query(monkeypatch, capsys) -> None: + client_calls = _stub_client(monkeypatch) + app = cli.build_app() + + payload = _error_output( + app, + [ + "sdk", + "artifacts", + "get", + "run-1", + "--query", + '{"artifact_ids": ["a"]}', + "--include-children", + ], + capsys, + ) + assert payload["status"] == "error" + assert "--include-children is only valid with --list or --out-dir" in payload["error"] + assert client_calls == [] + + +def test_mode_validation_rejects_non_object_query(monkeypatch, capsys) -> None: + client_calls = _stub_client(monkeypatch) + app = cli.build_app() + + payload = _error_output( + app, + ["sdk", "artifacts", "get", "run-1", "--query", "[1, 2, 3]"], + capsys, + ) + assert payload["status"] == "error" + assert "--query must be a JSON object" in payload["error"] + assert client_calls == [] + + +def test_empty_object_query_is_accepted_as_query_mode(monkeypatch, capsys) -> None: + # ``--query '{}'`` is an explicit (empty) object query, not a missing + # ``--query``: it selects query mode and succeeds with zero artifacts + # instead of surfacing the misleading "--query is required" error. + _stub_client(monkeypatch) + app = cli.build_app() + + run_app(app, ["sdk", "artifacts", "get", "run-1", "--query", "{}"]) + + payload = json.loads(capsys.readouterr().out) + assert payload == {"status": "success", "run_id": "run-1", "count": 0, "artifacts": []} + + +def test_empty_object_query_conflicts_with_list_mode(monkeypatch, capsys) -> None: + # An explicit empty object counts as a selected --query for the + # mutual-exclusion check, same as any other object value. + client_calls = _stub_client(monkeypatch) + app = cli.build_app() + + payload = _error_output( + app, + ["sdk", "artifacts", "get", "run-1", "--query", "{}", "--list"], + capsys, + ) + assert payload["status"] == "error" + assert "mutually exclusive" in payload["error"] + assert client_calls == [] + + +def test_empty_array_query_is_rejected_as_non_object(monkeypatch, capsys) -> None: + client_calls = _stub_client(monkeypatch) + app = cli.build_app() + + payload = _error_output( + app, + ["sdk", "artifacts", "get", "run-1", "--query", "[]"], + capsys, + ) + assert payload["status"] == "error" + assert "--query must be a JSON object" in payload["error"] + assert client_calls == [] + + +@pytest.mark.parametrize( + "query, fragment", + [ + ('{"executions": [1]}', "--query 'executions' must be an object"), + ('{"tasks": {"Train": "model"}}', "--query 'tasks' entry 'Train' must be a list"), + ('{"artifact_ids": "artifact-1"}', "--query 'artifact_ids' must be a list"), + ('{"artifact_ids": [null]}', "--query 'artifact_ids' entry 0 must be a string"), + ], +) +def test_malformed_nested_query_reports_clean_error( + monkeypatch, capsys, query: str, fragment: str +) -> None: + # Malformed nested query values must exit with the JSON error contract, + # not an AttributeError/TypeError traceback from inside the artifact walk. + _stub_client(monkeypatch) + app = cli.build_app() + + payload = _error_output( + app, + ["sdk", "artifacts", "get", "run-1", "--query", query], + capsys, + ) + assert payload["status"] == "error" + assert fragment in payload["error"] + + +def test_null_task_entry_query_resolves_all_outputs_end_to_end(monkeypatch, capsys) -> None: + # Regression: ``--query '{"tasks": {"train": null}}'`` against a run tree + # where the task exists and has outputs used to escape validation and + # crash with a raw ``TypeError`` from the artifact walk. A ``null`` filter + # must behave like ``[]`` (all outputs) through the full command path. + class _FakeClient: + def __init__(self, **kwargs: Any) -> None: + pass + + def get_run_details(self, run_id: str) -> Any: + return { + "execution": { + "task_spec": { + "graph_tasks": { + "train": { + "execution_output_artifacts": {"model": "artifact-model"} + } + } + } + } + } + + def artifacts_get(self, artifact_id: str) -> Any: + assert artifact_id == "artifact-model" + return { + "id": artifact_id, + "artifact_data": {"uri": "gs://bucket/model", "total_size": 3, "is_dir": False}, + } + + monkeypatch.setattr(artifacts_cli, "LazyTangleApiClient", _FakeClient) + + app = cli.build_app() + run_app(app, ["sdk", "artifacts", "get", "run-1", "--query", '{"tasks": {"train": null}}']) + + result = json.loads(capsys.readouterr().out) + assert result == { + "status": "success", + "run_id": "run-1", + "count": 1, + "artifacts": [ + { + "id": "artifact-model", + "uri": "gs://bucket/model", + "key": "train/model", + "total_size": 3, + "is_dir": False, + } + ], + } + + +# --------------------------------------------------------------------------- +# Config-file type converters +# --------------------------------------------------------------------------- + + +def test_config_list_flag_must_be_boolean(monkeypatch, tmp_path) -> None: + _stub_client(monkeypatch) + config = tmp_path / "artifacts.yaml" + config.write_text("run_id: run-1\nlist: not-a-bool\n", encoding="utf-8") + app = cli.build_app() + + code = _run_expecting_error(app, ["sdk", "artifacts", "get", "--config", str(config)]) + assert code not in (0, None) + + +def test_config_only_must_be_string_list(monkeypatch, tmp_path) -> None: + _stub_client(monkeypatch) + config = tmp_path / "artifacts.yaml" + config.write_text( + "run_id: run-1\nout_dir: /tmp/out\nonly: model\n", encoding="utf-8" + ) + app = cli.build_app() + + code = _run_expecting_error(app, ["sdk", "artifacts", "get", "--config", str(config)]) + assert code not in (0, None) + + +def test_config_list_and_out_dir_from_file(monkeypatch, tmp_path, capsys) -> None: + _stub_client(monkeypatch) + config = tmp_path / "artifacts.yaml" + config.write_text( + "run_id: run-config\nlist: true\ninclude_children: true\n", encoding="utf-8" + ) + list_calls: list[dict[str, Any]] = [] + + def fake_list(self, run_id: str, *, include_children: bool) -> list[dict[str, str]]: + list_calls.append({"run_id": run_id, "include_children": include_children}) + return [] + + monkeypatch.setattr(artifacts_module.ArtifactManager, "list_result_artifacts", fake_list) + + app = cli.build_app() + run_app(app, ["sdk", "artifacts", "get", "--config", str(config)]) + + result = json.loads(capsys.readouterr().out) + assert result["status"] == "success" + assert list_calls == [{"run_id": "run-config", "include_children": True}] diff --git a/tests/test_client.py b/tests/test_client.py index 5456ac0..05f4a56 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -53,3 +53,18 @@ def fail_from_dict(*args, **kwargs): assert client.get_run_pipeline_spec("run-1") is task_spec client.executions_details.assert_called_once_with("root-exec-1") + + +def test_request_raw_is_public_alias_for_make_request() -> None: + session = MagicMock() + response = MagicMock(status_code=200) + session.request.return_value = response + client = TangleApiClient("https://api.test", session=session) + + result = client.request_raw("GET", "/api/test", stream=True) + + assert result is response + call = session.request.call_args + assert call.args[0] == "GET" + assert call.args[1] == "https://api.test/api/test" + assert call.kwargs["stream"] is True