diff --git a/README.md b/README.md index 0163e88..7828401 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ **English** | [Українська](README.uk.md) -Current Client version: **1.0.0**. +Current Client version: **1.1.0**. > Dataset Server v1 is the source of truth for published metadata and binaries. > See the generated [public catalog](CATALOG.md). @@ -15,6 +15,8 @@ dsm-client sync dsm-client sync --check dsm-client fetch iv.dev.4.ff.1 --format rosbag dsm-client fetch iv.dev.4.ff.1 --format rosbag2 +dsm-client prepare-run iv.dev.4.ff.1 \ + --implementation vins-neo --suite smoke --format rosbag2 dsm-client validate ``` @@ -41,6 +43,23 @@ without persisting the token. Run `import-catalog` after publication. The former scripts and `IVINS_*` environment variables remain compatibility interfaces through Client 1.x. `DSM_*` always takes precedence. +## Runtime profiles and dataset runs + +`prepare-run` derives the hardware profile from the immutable Dataset Server +manifest. `--profile` is only an assertion and fails on disagreement. The first +validated runtime profile is `dev_04`; `dev_4` remains an alias. Other profiles +fail explicitly with `runtime profile unavailable` until their sensor/config +contracts are validated. + +The command authenticates only with `DSM_SERVER_TOKEN`, refuses to send that +token over plain HTTP except to a loopback test-server, verifies every cached +artifact by SHA-256, and emits stable JSON containing the dataset/profile, +artifact version/SHA/path, VINS config SHA, topics, launch arguments and smoke +acceptance settings. A published ROS1 bag is converted with pinned +`rosbags==0.11.3`; source and converted SHA-256 values remain separate. Tiny or +invalid bag fixtures cannot qualify a real run. See the +[runtime run contract](docs/runtime-runs.md). + This repository is a client catalog for evaluating multiple visual-inertial navigation systems (VINS) against public and iVINS MAV recordings. Dataset recordings are **not stored in this repository**. Manifests point to authoritative download diff --git a/README.uk.md b/README.uk.md index 16ff0de..06f1554 100644 --- a/README.uk.md +++ b/README.uk.md @@ -2,7 +2,7 @@ [English](README.md) | **Українська** -Поточна версія Client: **1.0.0**. Канонічний CLI — `dsm-client`; сервер є +Поточна версія Client: **1.1.0**. Канонічний CLI — `dsm-client`; сервер є єдиним джерелом опублікованого стану, а цей репозиторій містить manifests, configs і детермінований generated catalog. @@ -10,10 +10,28 @@ configs і детермінований generated catalog. python -m pip install -e . dsm-client sync dsm-client fetch iv.dev.4.ff.1 --format rosbag +dsm-client prepare-run iv.dev.4.ff.1 ` + --implementation vins-neo --suite smoke --format rosbag2 dsm-client publish ID --format rosbag --version 1 --path FILE --metadata META.json dsm-client validate ``` +## Runtime-профілі та dataset runs + +`prepare-run` визначає hardware-профіль з immutable manifest Dataset Server. +Параметр `--profile` є лише assertion і завершується помилкою при розбіжності. +Перший валідований runtime-профіль — `dev_04`; `dev_4` залишається alias. Для +інших профілів повертається явне `runtime profile unavailable`, доки їхні +sensor/config contracts не пройдуть валідацію. + +Команда бере ключ лише з `DSM_SERVER_TOKEN`, відмовляється передавати його через +HTTP (крім loopback test-server), перевіряє SHA-256 при кожному використанні +кешу й повертає стабільний JSON із dataset/profile, artifact version/SHA/path, +SHA конфігурації VINS, topics, launch arguments і smoke acceptance settings. +ROS1 bag конвертується зафіксованим `rosbags==0.11.3`; SHA source і converted +artifact зберігаються окремо. Малі або невалідні fixtures не можуть пройти +реальний gate. Див. [контракт runtime run](docs/runtime-runs.uk.md). + Канонічні змінні: `DSM_SERVER_URL`, `DSM_SERVER_TOKEN`, `DSM_RAW_ROOT`. Старі `IVINS_*` приймаються до Client 2.0 з попередженням; `DSM_*` має вищий пріоритет. diff --git a/configurations/ivins/VINS-NEO/dev_04/ivins_dev04_config.yaml b/configurations/ivins/VINS-NEO/dev_04/ivins_dev04_config.yaml index fb8b96d..b1352c3 100644 --- a/configurations/ivins/VINS-NEO/dev_04/ivins_dev04_config.yaml +++ b/configurations/ivins/VINS-NEO/dev_04/ivins_dev04_config.yaml @@ -1,9 +1,8 @@ %YAML:1.0 -# VINS-NEO configuration candidate for iVINS dev_04. -# The current values preserve the shared VINS-Mono-style baseline. -# TODO: verify this schema and required keys against the exact pinned VINS-NEO -# revision before using it; no implementation-specific values are assumed here. +# Validated VINS-NEO runtime configuration for iVINS dev_04. +# Runtime identity, compatible datasets and launch settings are versioned in +# runtime-profiles/dev_04.yaml. #common parameters imu_topic: "/imu0" diff --git a/docs/architecture.md b/docs/architecture.md index d3c2270..ec14091 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -20,11 +20,13 @@ flowchart TB CLI["dsm-client CLI"] --> Validate["Catalog and profile validator"] CLI --> Sync["Deterministic catalog importer"] CLI --> Raw["Fetch, conversion and publish workflows"] + CLI --> Prepare["Runtime profile and prepare-run"] Manifests["Dataset and configuration manifests"] --> Validate Aliases["Profile alias manifest"] --> Validate Families["Family profile manifest"] --> Validate Sync --> Generated["Bilingual generated catalog"] Raw --> Artifacts["Ignored local artifact hierarchy"] + Prepare --> Artifacts ``` ## Catalog and artifact data flow @@ -52,7 +54,7 @@ flowchart LR Python --> CLI["Installed dsm-client"] CLI --> Checkout["Versioned manifests and configs"] CLI --> Local["datasets/*/artifacts ignored by Git"] - CLI -->|"HTTP /v1 with memory-only token"| Server["Server runtime"] + CLI -->|"HTTPS /v1 with environment-only token"| Server["Server runtime"] ``` ## Repository interactions diff --git a/docs/architecture.uk.md b/docs/architecture.uk.md index 8f00e88..f4c6a38 100644 --- a/docs/architecture.uk.md +++ b/docs/architecture.uk.md @@ -20,11 +20,13 @@ flowchart TB CLI["CLI dsm-client"] --> Validate["Validator catalog і profiles"] CLI --> Sync["Детермінований catalog importer"] CLI --> Raw["Fetch, conversion і publish workflows"] + CLI --> Prepare["Runtime profile і prepare-run"] Manifests["Маніфести datasets і configurations"] --> Validate Aliases["Маніфест profile aliases"] --> Validate Families["Маніфест family profiles"] --> Validate Sync --> Generated["Двомовний generated catalog"] Raw --> Artifacts["Ігнорована локальна hierarchy artifacts"] + Prepare --> Artifacts ``` ## Потік catalog і artifact data @@ -52,7 +54,7 @@ flowchart LR Python --> CLI["Встановлений dsm-client"] CLI --> Checkout["Версіоновані manifests і configs"] CLI --> Local["datasets/*/artifacts, ігноровані Git"] - CLI -->|"HTTP /v1 з memory-only token"| Server["Runtime сервера"] + CLI -->|"HTTPS /v1 з environment-only token"| Server["Runtime сервера"] ``` ## Взаємодія репозиторіїв diff --git a/docs/runtime-runs.md b/docs/runtime-runs.md new file mode 100644 index 0000000..47c3f92 --- /dev/null +++ b/docs/runtime-runs.md @@ -0,0 +1,36 @@ +# Runtime profiles and prepared dataset runs + +[Українська версія](runtime-runs.uk.md) + +DataSetsManager Client owns the versioned hardware-profile, implementation +configuration and dataset-suite contract. VINS-NEO owns launch and runtime +behavior. The Dataset Server remains the source of immutable artifact version, +size and SHA-256. + +## Prepare `iv.dev.4.ff.1` + +```bash +export DSM_SERVER_URL=https://datasetsmanager.drone-age.org +export DSM_SERVER_TOKEN='value-from-an-approved-secret-store' +dsm-client prepare-run iv.dev.4.ff.1 \ + --implementation vins-neo \ + --suite smoke \ + --format rosbag2 > run-manifest.json +``` + +The profile comes from the server dataset manifest. An optional `--profile` +does not override it. The command fails on a missing profile/config, suite +mismatch, unsafe path, authentication error, checksum mismatch, invalid bag or +non-loopback HTTP credential transport. + +Artifact paths are content-addressed. Cache reuse recalculates SHA-256 before +returning a run manifest. ROS1 conversion is pinned to `rosbags==0.11.3`; the +manifest records both source and converted digests. The test-only +`--allow-test-fixture` switch exists for the local Server contract test and must +never be passed by a VINS runner or release job. + +For a tokenless Raspberry Pi run, prepare and verify the artifact plus manifest +on a trusted host through HTTPS, transfer the artifact, config and manifest as +one integrity-checked bundle, and run the VINS gate without exporting the DSM +token on the Pi. AMD64 evidence is development-only; release evidence requires +the native Debian 13 ARM64 rerun. diff --git a/docs/runtime-runs.uk.md b/docs/runtime-runs.uk.md new file mode 100644 index 0000000..c5b4cb9 --- /dev/null +++ b/docs/runtime-runs.uk.md @@ -0,0 +1,36 @@ +# Runtime-профілі та підготовлені dataset runs + +[English version](runtime-runs.md) + +DataSetsManager Client володіє версіонованим контрактом hardware-профілю, +конфігурації реалізації та dataset suite. VINS-NEO володіє launch/runtime +поведінкою. Dataset Server залишається джерелом immutable version, size і +SHA-256 артефакту. + +## Підготовка `iv.dev.4.ff.1` + +```bash +export DSM_SERVER_URL=https://datasetsmanager.drone-age.org +export DSM_SERVER_TOKEN='значення-з-погодженого-сховища-секретів' +dsm-client prepare-run iv.dev.4.ff.1 \ + --implementation vins-neo \ + --suite smoke \ + --format rosbag2 > run-manifest.json +``` + +Профіль береться з server dataset manifest. Необов'язковий `--profile` не +перевизначає його. Команда повертає FAIL за відсутнього профілю/config, suite +mismatch, небезпечного шляху, помилки автентифікації, checksum mismatch, +невалідного bag або спроби передати credential через не-loopback HTTP. + +Шляхи артефактів content-addressed. Перед повторним використанням кешу SHA-256 +обчислюється заново. Конвертація ROS1 зафіксована на `rosbags==0.11.3`; manifest +містить окремі source і converted digests. Test-only параметр +`--allow-test-fixture` існує лише для локального Server contract test і не може +використовуватись VINS runner або release job. + +Для Raspberry Pi без токена підготуйте й перевірте artifact та manifest на +довіреному host через HTTPS, передайте artifact, config і manifest одним +пакетом із integrity check та запустіть VINS gate без `DSM_SERVER_TOKEN` на Pi. +AMD64 evidence є development-only; release evidence потребує повторного +нативного запуску на Debian 13 ARM64. diff --git a/dsm_client/__init__.py b/dsm_client/__init__.py index 492da43..2d86497 100644 --- a/dsm_client/__init__.py +++ b/dsm_client/__init__.py @@ -1,3 +1,3 @@ """DataSetsManager Client 1.x.""" -__version__ = "1.0.0" +__version__ = "1.1.0" diff --git a/dsm_client/cli.py b/dsm_client/cli.py index 36d2883..4d8b96e 100644 --- a/dsm_client/cli.py +++ b/dsm_client/cli.py @@ -3,10 +3,14 @@ from __future__ import annotations import argparse +import json +import os +from pathlib import Path import sys from collections.abc import Callable from scripts import catalog_admin, raw_catalog, validate_catalog +from .runtime import PrepareRunError, prepare_run, stable_json def invoke(main: Callable[[], int], program: str, arguments: list[str]) -> int: @@ -25,6 +29,23 @@ def main() -> int: commands.add_parser("sync", help="import the canonical server catalog") for name in ("fetch", "publish", "raw"): commands.add_parser(name) + prepare = commands.add_parser( + "prepare-run", help="prepare an immutable dataset/runtime run manifest" + ) + prepare.add_argument("dataset_id") + prepare.add_argument("--implementation", required=True) + prepare.add_argument("--suite", required=True) + prepare.add_argument("--format", choices=("rosbag", "rosbag2"), required=True) + prepare.add_argument("--profile", help="assert the profile derived from the dataset manifest") + prepare.add_argument( + "--server-url", "--dsm-server-url", dest="server_url", + default=os.environ.get("DSM_SERVER_URL"), + ) + prepare.add_argument( + "--cache-dir", + default=os.environ.get("DSM_CACHE_DIR", str(Path.home() / ".cache" / "datasetsmanager")), + ) + prepare.add_argument("--allow-test-fixture", action="store_true", help=argparse.SUPPRESS) args, forwarded = parser.parse_known_args() if args.command == "validate": if forwarded: @@ -32,6 +53,31 @@ def main() -> int: return validate_catalog.main() if args.command == "sync": return invoke(catalog_admin.main, "dsm-client sync", ["import-catalog", *forwarded]) + if args.command == "prepare-run": + if forwarded: + parser.error(f"unrecognized arguments: {' '.join(forwarded)}") + if not args.server_url: + parser.error("--server-url or DSM_SERVER_URL is required") + try: + value = prepare_run( + dataset_id=args.dataset_id, + implementation=args.implementation, + suite_name=args.suite, + requested_format=args.format, + server_url=args.server_url, + token=os.environ.get("DSM_SERVER_TOKEN", ""), + cache_dir=Path(args.cache_dir).expanduser(), + asserted_profile=args.profile, + allow_test_fixture=args.allow_test_fixture, + ) + sys.stdout.write(stable_json(value)) + return 0 + except PrepareRunError as exc: + print( + json.dumps({"status": "error", "message": str(exc)}, sort_keys=True), + file=sys.stderr, + ) + return 2 if args.command == "raw": return invoke(raw_catalog.main, "dsm-client raw", forwarded) return invoke(raw_catalog.main, f"dsm-client {args.command}", [args.command, *forwarded]) diff --git a/dsm_client/runtime.py b/dsm_client/runtime.py new file mode 100644 index 0000000..979e61b --- /dev/null +++ b/dsm_client/runtime.py @@ -0,0 +1,542 @@ +"""Prepare immutable dataset runs for VINS implementations.""" + +from __future__ import annotations + +from dataclasses import dataclass +import hashlib +import json +import os +from pathlib import Path +import re +import shutil +import subprocess +import sys +import sysconfig +import tempfile +from typing import Any +from urllib.error import HTTPError, URLError +from urllib.parse import urljoin, urlsplit +from urllib.request import Request, urlopen +import zipfile + +import yaml + +from . import __version__ +from .profiles import canonical_profile +from .transport import TransportSecurityError, require_secure_bearer_transport + + +ROOT = Path(__file__).resolve().parents[1] +INSTALLED_RESOURCE_ROOT = Path(sysconfig.get_path("data")) / "share" / "datasetsmanager-client" +RESOURCE_ROOT = ROOT if (ROOT / "runtime-profiles").is_dir() else INSTALLED_RESOURCE_ROOT +ROSBAGS_VERSION = "0.11.3" +SAFE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$") +SHA256 = re.compile(r"^[0-9a-f]{64}$") + + +class PrepareRunError(RuntimeError): + """A run manifest could not be prepared safely.""" + + +@dataclass(frozen=True) +class PreparedArtifact: + path: Path + version: str + sha256: str + size: int + source_format: str + source_sha256: str + converted: bool = False + converter_version: str | None = None + + +def _safe_component(value: object, label: str) -> str: + result = str(value) + if not SAFE_ID.fullmatch(result) or result in {".", ".."}: + raise PrepareRunError(f"invalid {label}: {result!r}") + return result + + +def _sha256_file(path: Path) -> tuple[str, int]: + digest = hashlib.sha256() + size = 0 + with path.open("rb") as stream: + for chunk in iter(lambda: stream.read(1024 * 1024), b""): + digest.update(chunk) + size += len(chunk) + return digest.hexdigest(), size + + +def _sha256_directory(path: Path) -> tuple[str, int]: + digest = hashlib.sha256() + size = 0 + files = sorted(item for item in path.rglob("*") if item.is_file()) + if not files: + raise PrepareRunError(f"artifact directory contains no files: {path}") + for item in files: + relative = item.relative_to(path).as_posix().encode("utf-8") + digest.update(len(relative).to_bytes(4, "big")) + digest.update(relative) + with item.open("rb") as stream: + for chunk in iter(lambda: stream.read(1024 * 1024), b""): + digest.update(chunk) + size += len(chunk) + return digest.hexdigest(), size + + +def _atomic_json(path: Path, value: dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + descriptor, temporary_name = tempfile.mkstemp( + prefix=f".{path.name}.", suffix=".tmp", dir=path.parent + ) + temporary = Path(temporary_name) + try: + with os.fdopen(descriptor, "w", encoding="utf-8", newline="\n") as stream: + json.dump(value, stream, indent=2, sort_keys=True, ensure_ascii=False) + stream.write("\n") + stream.flush() + os.fsync(stream.fileno()) + os.replace(temporary, path) + finally: + temporary.unlink(missing_ok=True) + + +def _request(url: str, token: str, accept: str) -> tuple[bytes, dict[str, str]]: + try: + require_secure_bearer_transport(url, token) + except TransportSecurityError as exc: + raise PrepareRunError(str(exc)) from exc + request = Request( + url, + headers={"Accept": accept, "Authorization": f"Bearer {token}"}, + ) + try: + with urlopen(request, timeout=60) as response: + return response.read(), dict(response.headers.items()) + except HTTPError as exc: + detail = "" + try: + detail = json.loads(exc.read()).get("error", {}).get("message", "") + except (json.JSONDecodeError, AttributeError): + pass + raise PrepareRunError( + f"Dataset Server request failed ({exc.code})" + + (f": {detail}" if detail else "") + ) from exc + except (URLError, TimeoutError, OSError) as exc: + raise PrepareRunError(f"Dataset Server is unreachable: {exc}") from exc + + +def _catalog(server_url: str, token: str) -> dict[str, Any]: + body, _ = _request(server_url.rstrip("/") + "/v1/catalog", token, "application/json") + try: + value = json.loads(body) + except json.JSONDecodeError as exc: + raise PrepareRunError("Dataset Server returned invalid catalog JSON") from exc + if value.get("schema_version") != "1.0" or not isinstance(value.get("datasets"), list): + raise PrepareRunError("unsupported Dataset Server catalog schema") + return value + + +def _same_origin_download_url(server_url: str, download_url: str) -> str: + candidate = urljoin(server_url.rstrip("/") + "/", download_url) + + def origin(url: str) -> tuple[str, str, int]: + parsed = urlsplit(url) + if parsed.scheme not in {"http", "https"} or not parsed.hostname: + raise PrepareRunError("artifact download URL must be absolute HTTP(S)") + try: + port = parsed.port or (443 if parsed.scheme == "https" else 80) + except ValueError as exc: + raise PrepareRunError("artifact download URL has an invalid port") from exc + return parsed.scheme.lower(), parsed.hostname.rstrip(".").lower(), port + + if origin(candidate) != origin(server_url): + raise PrepareRunError("refusing to send DSM_SERVER_TOKEN to a cross-origin artifact URL") + return candidate + + +def _resource(relative: str) -> Path: + candidate = (RESOURCE_ROOT / relative).resolve() + root = RESOURCE_ROOT.resolve() + try: + candidate.relative_to(root) + except ValueError as exc: + raise PrepareRunError("runtime resource path escapes the Client installation") from exc + if not candidate.is_file(): + raise PrepareRunError(f"runtime resource is missing: {relative}") + return candidate + + +def load_runtime_profile(profile: str) -> dict[str, Any]: + canonical = canonical_profile(profile) + if canonical != "dev_04": + raise PrepareRunError(f"runtime profile unavailable: {canonical}") + path = _resource(f"runtime-profiles/{_safe_component(canonical, 'profile')}.yaml") + value = yaml.safe_load(path.read_text(encoding="utf-8")) + if value.get("schema_version") != "1.0" or value.get("profile") != canonical: + raise PrepareRunError(f"invalid runtime profile: {canonical}") + return value + + +def _suite(name: str) -> dict[str, Any]: + path = _resource(f"test-suites/{_safe_component(name, 'suite')}.yaml") + value = yaml.safe_load(path.read_text(encoding="utf-8")) + if value.get("schema_version") != "1.0" or value.get("name") != name: + raise PrepareRunError(f"invalid test suite: {name}") + return value + + +def _validate_bag(path: Path, allow_test_fixture: bool) -> None: + if allow_test_fixture: + return + if path.is_file() and path.stat().st_size <= 64: + raise PrepareRunError("fixture-sized artifact is forbidden for a real dataset run") + try: + from rosbags.highlevel import AnyReader + except ImportError as exc: + raise PrepareRunError( + f"bag validation requires the pinned converter dependency rosbags=={ROSBAGS_VERSION}" + ) from exc + try: + with AnyReader([path]) as reader: + if not reader.connections: + raise PrepareRunError("bag contains no topics") + except PrepareRunError: + raise + except Exception as exc: + raise PrepareRunError(f"artifact does not pass rosbag validation: {exc}") from exc + + +def _download_artifact( + server_url: str, + token: str, + dataset_id: str, + artifact: dict[str, Any], + cache_dir: Path, +) -> Path: + version = _safe_component(artifact.get("version"), "artifact version") + source_format = _safe_component(artifact.get("format"), "artifact format") + expected_sha = str(artifact.get("sha256", "")) + expected_size = artifact.get("size") + if not SHA256.fullmatch(expected_sha) or not isinstance(expected_size, int) or expected_size < 0: + raise PrepareRunError("artifact has invalid immutable metadata") + suffix = ".bag" if source_format == "rosbag" else ".zip" + destination = ( + cache_dir / "artifacts" / dataset_id / source_format / version / expected_sha / f"artifact{suffix}" + ) + if destination.exists(): + actual_sha, actual_size = _sha256_file(destination) + if actual_sha != expected_sha or actual_size != expected_size: + raise PrepareRunError("checksum mismatch in cached artifact; refusing unsafe reuse") + return destination + + destination.parent.mkdir(parents=True, exist_ok=True) + relative_url = str(artifact.get("download_url", "")) + url = _same_origin_download_url(server_url, relative_url) + descriptor, temporary_name = tempfile.mkstemp(prefix=".download.", dir=destination.parent) + temporary = Path(temporary_name) + try: + try: + require_secure_bearer_transport(url, token) + except TransportSecurityError as exc: + raise PrepareRunError(str(exc)) from exc + request = Request( + url, + headers={ + "Accept": "application/octet-stream", + "Authorization": f"Bearer {token}", + }, + ) + digest = hashlib.sha256() + count = 0 + output = os.fdopen(descriptor, "wb") + descriptor = -1 + try: + with output, urlopen(request, timeout=60) as response: + while True: + chunk = response.read(1024 * 1024) + if not chunk: + break + output.write(chunk) + digest.update(chunk) + count += len(chunk) + output.flush() + os.fsync(output.fileno()) + except HTTPError as exc: + raise PrepareRunError(f"Dataset Server artifact request failed ({exc.code})") from exc + except (URLError, TimeoutError, OSError) as exc: + raise PrepareRunError(f"Dataset Server artifact download failed: {exc}") from exc + actual_sha, actual_size = digest.hexdigest(), count + if actual_sha != expected_sha or actual_size != expected_size: + raise PrepareRunError("Dataset Server artifact size/SHA-256 verification failed") + try: + os.link(temporary, destination) + except FileExistsError: + actual_sha, actual_size = _sha256_file(destination) + if actual_sha != expected_sha or actual_size != expected_size: + raise PrepareRunError("checksum mismatch in concurrent cache entry") + temporary.unlink(missing_ok=True) + return destination + finally: + try: + if descriptor >= 0: + os.close(descriptor) + except OSError: + pass + temporary.unlink(missing_ok=True) + + +def _conversion_version() -> str: + try: + from importlib.metadata import version + + actual = version("rosbags") + except Exception as exc: + raise PrepareRunError(f"ROS1 conversion requires rosbags=={ROSBAGS_VERSION}") from exc + if actual != ROSBAGS_VERSION: + raise PrepareRunError( + f"ROS1 conversion requires rosbags=={ROSBAGS_VERSION}, found {actual}" + ) + return actual + + +def _convert_rosbag(source: Path, dataset_id: str, source_sha: str, cache_dir: Path) -> tuple[Path, str, int]: + converter_version = _conversion_version() + root = cache_dir / "converted" / dataset_id / "rosbag2" / source_sha / f"rosbags-{converter_version}" + destination = root / "data" + record_path = root / "conversion.json" + if destination.exists() or record_path.exists(): + if not destination.is_dir() or not record_path.is_file(): + raise PrepareRunError("incomplete converted artifact cache entry") + record = json.loads(record_path.read_text(encoding="utf-8")) + actual_sha, actual_size = _sha256_directory(destination) + if ( + record.get("source_sha256") != source_sha + or record.get("converted_sha256") != actual_sha + or record.get("converter_version") != converter_version + ): + raise PrepareRunError("checksum mismatch in converted artifact cache") + return destination, actual_sha, actual_size + + root.mkdir(parents=True, exist_ok=True) + temporary_root = Path(tempfile.mkdtemp(prefix=".convert.", dir=root)) + staged = temporary_root / "data" + command = [ + sys.executable, + "-m", + "rosbags.convert", + "--src", + str(source), + "--dst", + str(staged), + "--dst-storage", + "sqlite3", + ] + try: + completed = subprocess.run(command, capture_output=True, text=True, check=False) + if completed.returncode: + detail = (completed.stderr or completed.stdout or "no diagnostic output").strip() + raise PrepareRunError(f"rosbags conversion failed: {detail}") + if not (staged / "metadata.yaml").is_file() or not list(staged.glob("*.db3")): + raise PrepareRunError("rosbags conversion did not produce a valid rosbag2 directory") + converted_sha, converted_size = _sha256_directory(staged) + try: + os.rename(staged, destination) + except FileExistsError as exc: + raise PrepareRunError("converted artifact appeared concurrently") from exc + _atomic_json( + record_path, + { + "converter": "rosbags", + "converter_version": converter_version, + "converted_sha256": converted_sha, + "source_sha256": source_sha, + }, + ) + return destination, converted_sha, converted_size + finally: + shutil.rmtree(temporary_root, ignore_errors=True) + + +def _extract_rosbag2(archive: Path, dataset_id: str, sha256: str, cache_dir: Path) -> tuple[Path, str, int]: + root = cache_dir / "extracted" / dataset_id / "rosbag2" / sha256 + destination = root / "data" + record_path = root / "extraction.json" + if destination.exists() or record_path.exists(): + if not destination.is_dir() or not record_path.is_file(): + raise PrepareRunError("incomplete rosbag2 extraction cache entry") + record = json.loads(record_path.read_text(encoding="utf-8")) + actual_sha, actual_size = _sha256_directory(destination) + if record.get("archive_sha256") != sha256 or record.get("content_sha256") != actual_sha: + raise PrepareRunError("checksum mismatch in extracted rosbag2 cache") + return destination, actual_sha, actual_size + root.mkdir(parents=True, exist_ok=True) + temporary_root = Path(tempfile.mkdtemp(prefix=".extract.", dir=root)) + staged = temporary_root / "data" + staged.mkdir() + try: + with zipfile.ZipFile(archive) as source: + for member in source.infolist(): + target = (staged / member.filename).resolve() + try: + target.relative_to(staged.resolve()) + except ValueError as exc: + raise PrepareRunError("unsafe path in rosbag2 archive") from exc + source.extractall(staged) + content_sha, content_size = _sha256_directory(staged) + os.rename(staged, destination) + _atomic_json( + record_path, + {"archive_sha256": sha256, "content_sha256": content_sha}, + ) + return destination, content_sha, content_size + except (zipfile.BadZipFile, OSError) as exc: + raise PrepareRunError(f"invalid rosbag2 archive: {exc}") from exc + finally: + shutil.rmtree(temporary_root, ignore_errors=True) + + +def _prepare_artifact( + server_url: str, + token: str, + dataset: dict[str, Any], + requested_format: str, + cache_dir: Path, + allow_test_fixture: bool, +) -> PreparedArtifact: + artifacts = [item for item in dataset.get("artifacts", []) if item.get("state") == "published"] + direct = [item for item in artifacts if item.get("format") == requested_format] + if direct: + selected = sorted(direct, key=lambda item: str(item.get("version")))[-1] + archive = _download_artifact(server_url, token, dataset["id"], selected, cache_dir) + if requested_format == "rosbag2": + path, content_sha, content_size = _extract_rosbag2( + archive, dataset["id"], selected["sha256"], cache_dir + ) + _validate_bag(path, allow_test_fixture) + return PreparedArtifact( + path, str(selected["version"]), content_sha, content_size, + "rosbag2", str(selected["sha256"]), False, None, + ) + _validate_bag(archive, allow_test_fixture) + return PreparedArtifact( + archive, str(selected["version"]), str(selected["sha256"]), + int(selected["size"]), requested_format, str(selected["sha256"]), False, None, + ) + if requested_format != "rosbag2" or "rosbag2" not in dataset.get("derivable_formats", []): + raise PrepareRunError(f"{dataset['id']} {requested_format} is neither published nor derivable") + sources = [item for item in artifacts if item.get("format") == "rosbag"] + if not sources: + raise PrepareRunError(f"{dataset['id']} has no published ROS1 source artifact") + source_meta = sorted(sources, key=lambda item: str(item.get("version")))[-1] + source = _download_artifact(server_url, token, dataset["id"], source_meta, cache_dir) + _validate_bag(source, allow_test_fixture) + if allow_test_fixture: + return PreparedArtifact( + source, str(source_meta["version"]), str(source_meta["sha256"]), + int(source_meta["size"]), "rosbag", str(source_meta["sha256"]), False, None, + ) + path, converted_sha, converted_size = _convert_rosbag( + source, dataset["id"], source_meta["sha256"], cache_dir + ) + _validate_bag(path, False) + return PreparedArtifact( + path, str(source_meta["version"]), converted_sha, converted_size, + "rosbag", str(source_meta["sha256"]), True, ROSBAGS_VERSION, + ) + + +def prepare_run( + dataset_id: str, + implementation: str, + suite_name: str, + requested_format: str, + server_url: str, + token: str, + cache_dir: Path, + asserted_profile: str | None = None, + allow_test_fixture: bool = False, +) -> dict[str, Any]: + dataset_id = _safe_component(dataset_id, "dataset ID") + implementation = _safe_component(implementation, "implementation") + if requested_format not in {"rosbag", "rosbag2"}: + raise PrepareRunError(f"unsupported artifact format: {requested_format}") + if not token: + raise PrepareRunError("DSM_SERVER_TOKEN is required") + catalog = _catalog(server_url, token) + matches = [item for item in catalog["datasets"] if item.get("id") == dataset_id] + if len(matches) != 1: + raise PrepareRunError(f"dataset not found in Dataset Server catalog: {dataset_id}") + dataset = matches[0] + profile_name = canonical_profile(dataset.get("profile")) + if asserted_profile and canonical_profile(asserted_profile) != profile_name: + raise PrepareRunError( + f"dataset/profile mismatch: manifest={profile_name}, assertion={canonical_profile(asserted_profile)}" + ) + profile = load_runtime_profile(profile_name) + compatible = profile.get("compatible_dataset_ids", []) + if dataset_id not in compatible: + raise PrepareRunError(f"dataset {dataset_id} is incompatible with runtime profile {profile_name}") + implementation_profile = profile.get("implementations", {}).get(implementation) + if not implementation_profile: + raise PrepareRunError( + f"implementation {implementation} is unavailable for runtime profile {profile_name}" + ) + suite = _suite(suite_name) + cases = [ + item for item in suite.get("cases", []) + if item.get("dataset_id") == dataset_id and item.get("implementation") == implementation + ] + if len(cases) != 1: + raise PrepareRunError( + f"suite {suite_name} has no unique case for {dataset_id} + {implementation}" + ) + config_path = _resource(str(implementation_profile["config"])) + config_sha, _ = _sha256_file(config_path) + artifact = _prepare_artifact( + server_url, token, dataset, requested_format, cache_dir.resolve(), allow_test_fixture + ) + launch_arguments = dict(implementation_profile.get("launch_arguments", {})) + launch_arguments.update( + {"config_file": str(config_path), "vins_folder": str(config_path.parent) + os.sep} + ) + return { + "acceptance": dict(cases[0]["acceptance"]), + "artifact": { + "converted": artifact.converted, + "converter": ( + {"name": "rosbags", "version": artifact.converter_version} + if artifact.converted else None + ), + "format": requested_format if artifact.converted else ( + requested_format if artifact.source_format == requested_format else artifact.source_format + ), + "path": str(artifact.path.resolve()), + "sha256": artifact.sha256, + "size": artifact.size, + "source_format": artifact.source_format, + "source_sha256": artifact.source_sha256, + "version": artifact.version, + }, + "client": {"name": "datasetsmanager-client", "version": __version__}, + "config": { + "path": str(config_path), + "relative_path": str(implementation_profile["config"]), + "sha256": config_sha, + }, + "dataset": {"id": dataset_id, "profile": profile_name}, + "implementation": implementation, + "launch": { + "arguments": launch_arguments, + "file": str(implementation_profile["launch_file"]), + "package": str(implementation_profile["launch_package"]), + }, + "schema_version": "1.0", + "suite": {"name": suite_name, "version": str(suite["version"])}, + "topics": dict(profile["topics"]), + } + + +def stable_json(value: dict[str, Any]) -> str: + return json.dumps(value, indent=2, sort_keys=True, ensure_ascii=False) + "\n" diff --git a/dsm_client/transport.py b/dsm_client/transport.py new file mode 100644 index 0000000..6d88b49 --- /dev/null +++ b/dsm_client/transport.py @@ -0,0 +1,15 @@ +"""Transport rules shared by DataSetsManager Client commands.""" + +from __future__ import annotations + +from scripts.env_compat import ( + TransportSecurityError, + is_loopback_url, + require_secure_bearer_transport, +) + +__all__ = [ + "TransportSecurityError", + "is_loopback_url", + "require_secure_bearer_transport", +] diff --git a/manifest/repository.yaml b/manifest/repository.yaml index d7704b9..042e654 100644 --- a/manifest/repository.yaml +++ b/manifest/repository.yaml @@ -1,7 +1,7 @@ schema_version: "1.0" repository: DataSetsManager/client display_name: DataSetsManager Client -version: 1.0.0 +version: 1.1.0 owner: NeoUKR change_authority: NeoUKR default_branch: main @@ -11,3 +11,5 @@ contracts: dataset_manifest: schemas/dataset-manifest.schema.json profile_aliases: schemas/profile-aliases.schema.json family_profiles: schemas/family-profiles.schema.json + runtime_profile: schemas/runtime-profile.schema.json + test_suite: schemas/test-suite.schema.json diff --git a/pyproject.toml b/pyproject.toml index 5070a06..2dc2d91 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,16 +4,21 @@ build-backend = "setuptools.build_meta" [project] name = "datasetsmanager-client" -version = "1.0.0" +version = "1.1.0" requires-python = ">=3.10" -dependencies = ["PyYAML>=6,<7", "Werkzeug>=3.1,<4"] +dependencies = ["PyYAML>=6,<7", "Werkzeug>=3.1,<4", "rosbags==0.11.3"] [project.optional-dependencies] conversion = ["rosbags==0.11.3"] -test = ["jsonschema>=4.23,<5", "rosbags==0.11.3"] +test = ["jsonschema>=4.23,<5"] [project.scripts] dsm-client = "dsm_client.cli:main" [tool.setuptools] packages = ["dsm_client", "scripts"] + +[tool.setuptools.data-files] +"share/datasetsmanager-client/runtime-profiles" = ["runtime-profiles/*.yaml"] +"share/datasetsmanager-client/test-suites" = ["test-suites/*.yaml"] +"share/datasetsmanager-client/configurations/ivins/VINS-NEO/dev_04" = ["configurations/ivins/VINS-NEO/dev_04/*.yaml"] diff --git a/runtime-profiles/dev_04.yaml b/runtime-profiles/dev_04.yaml new file mode 100644 index 0000000..204c2cf --- /dev/null +++ b/runtime-profiles/dev_04.yaml @@ -0,0 +1,27 @@ +schema_version: "1.0" +profile: dev_04 +status: validated +hardware: + camera: + identity: dev_04.camera0 + model: unspecified + calibration: configurations/ivins/kalibr/dev_04/try1/data-camchain-imucam.yaml + imu: + identity: dev_04.imu0 + model: unspecified + calibration: configurations/ivins/kalibr/dev_04/try1/data-imu.yaml +topics: + camera: /cam0/image_raw + imu: /imu0 + odometry: /vins_estimator/odometry +implementations: + vins-neo: + config: configurations/ivins/VINS-NEO/dev_04/ivins_dev04_config.yaml + launch_package: vins_estimator + launch_file: vins_neo.launch.py + launch_arguments: + use_sim_time: true + log_level: info + logging_period_ms: 2000 +compatible_dataset_ids: + - iv.dev.4.ff.1 diff --git a/schemas/repository-manifest.schema.json b/schemas/repository-manifest.schema.json index 9bb9708..20526d1 100644 --- a/schemas/repository-manifest.schema.json +++ b/schemas/repository-manifest.schema.json @@ -9,7 +9,7 @@ "schema_version": {"const": "1.0"}, "repository": {"const": "DataSetsManager/client"}, "display_name": {"type": "string", "minLength": 1}, - "version": {"const": "1.0.0"}, + "version": {"const": "1.1.0"}, "owner": {"const": "NeoUKR"}, "change_authority": {"const": "NeoUKR"}, "default_branch": {"const": "main"}, diff --git a/schemas/runtime-profile.schema.json b/schemas/runtime-profile.schema.json new file mode 100644 index 0000000..044ab58 --- /dev/null +++ b/schemas/runtime-profile.schema.json @@ -0,0 +1,75 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://github.com/DataSetsManager/client/releases/download/schema-runtime-profile-v1.1.0/runtime-profile.schema.json", + "title": "DataSetsManager runtime hardware profile", + "type": "object", + "additionalProperties": false, + "required": ["schema_version", "profile", "status", "hardware", "topics", "implementations", "compatible_dataset_ids"], + "properties": { + "schema_version": {"const": "1.0"}, + "profile": {"type": "string", "pattern": "^[a-z0-9_]+$"}, + "status": {"const": "validated"}, + "hardware": { + "type": "object", + "additionalProperties": false, + "required": ["camera", "imu"], + "properties": { + "camera": {"$ref": "#/$defs/sensor"}, + "imu": {"$ref": "#/$defs/sensor"} + } + }, + "topics": { + "type": "object", + "additionalProperties": false, + "required": ["camera", "imu", "odometry"], + "properties": { + "camera": {"type": "string", "pattern": "^/"}, + "imu": {"type": "string", "pattern": "^/"}, + "odometry": {"type": "string", "pattern": "^/"} + } + }, + "implementations": { + "type": "object", + "minProperties": 1, + "additionalProperties": {"$ref": "#/$defs/implementation"} + }, + "compatible_dataset_ids": { + "type": "array", + "minItems": 1, + "uniqueItems": true, + "items": {"type": "string", "minLength": 1} + } + }, + "$defs": { + "sensor": { + "type": "object", + "additionalProperties": false, + "required": ["identity", "model", "calibration"], + "properties": { + "identity": {"type": "string", "minLength": 1}, + "model": {"type": "string", "minLength": 1}, + "calibration": {"type": "string", "minLength": 1} + } + }, + "implementation": { + "type": "object", + "additionalProperties": false, + "required": ["config", "launch_package", "launch_file", "launch_arguments"], + "properties": { + "config": {"type": "string", "minLength": 1}, + "launch_package": {"type": "string", "minLength": 1}, + "launch_file": {"type": "string", "minLength": 1}, + "launch_arguments": { + "type": "object", + "additionalProperties": false, + "required": ["use_sim_time", "log_level", "logging_period_ms"], + "properties": { + "use_sim_time": {"type": "boolean"}, + "log_level": {"enum": ["debug", "info", "warn", "error", "fatal"]}, + "logging_period_ms": {"type": "integer", "minimum": 100} + } + } + } + } + } +} diff --git a/schemas/test-suite.schema.json b/schemas/test-suite.schema.json new file mode 100644 index 0000000..85d0fa6 --- /dev/null +++ b/schemas/test-suite.schema.json @@ -0,0 +1,50 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://github.com/DataSetsManager/client/releases/download/schema-test-suite-v1.1.0/test-suite.schema.json", + "title": "DataSetsManager runtime test suite", + "type": "object", + "additionalProperties": false, + "required": ["schema_version", "name", "version", "description", "cases"], + "properties": { + "schema_version": {"const": "1.0"}, + "name": {"type": "string", "pattern": "^[a-z0-9_-]+$"}, + "version": {"type": "string", "pattern": "^[0-9]+\\.[0-9]+\\.[0-9]+$"}, + "description": {"type": "string", "minLength": 1}, + "cases": { + "type": "array", + "minItems": 1, + "items": { + "type": "object", + "additionalProperties": false, + "required": ["dataset_id", "implementation", "acceptance"], + "properties": { + "dataset_id": {"type": "string", "minLength": 1}, + "implementation": {"type": "string", "minLength": 1}, + "acceptance": { + "type": "object", + "additionalProperties": false, + "required": ["required_topic_min_messages", "minimum_odometry_messages", "odometry_timestamps_monotonic", "finite_position_orientation", "processes_alive_until_bag_complete", "timeout_seconds", "ground_truth_required", "delta_xy_xyz_blocking"], + "properties": { + "required_topic_min_messages": { + "type": "object", + "additionalProperties": false, + "required": ["camera", "imu"], + "properties": { + "camera": {"type": "integer", "minimum": 1}, + "imu": {"type": "integer", "minimum": 1} + } + }, + "minimum_odometry_messages": {"type": "integer", "minimum": 1}, + "odometry_timestamps_monotonic": {"const": true}, + "finite_position_orientation": {"const": true}, + "processes_alive_until_bag_complete": {"const": true}, + "timeout_seconds": {"type": "integer", "minimum": 1}, + "ground_truth_required": {"type": "boolean"}, + "delta_xy_xyz_blocking": {"type": "boolean"} + } + } + } + } + } + } +} diff --git a/scripts/catalog_admin.py b/scripts/catalog_admin.py index 3b0b4e2..1069f02 100644 --- a/scripts/catalog_admin.py +++ b/scripts/catalog_admin.py @@ -11,15 +11,27 @@ import shutil import tempfile from urllib.error import HTTPError, URLError -from urllib.request import urlopen +from urllib.request import Request, urlopen try: - from .env_compat import value as env_value + from .env_compat import ( + TransportSecurityError, + require_secure_bearer_transport, + value as env_value, + ) except ImportError: try: - from scripts.env_compat import value as env_value + from scripts.env_compat import ( + TransportSecurityError, + require_secure_bearer_transport, + value as env_value, + ) except ImportError: # direct script execution - from env_compat import value as env_value + from env_compat import ( + TransportSecurityError, + require_secure_bearer_transport, + value as env_value, + ) REPO = Path(__file__).resolve().parents[1] GENERATED_MARK = "" @@ -33,11 +45,17 @@ class ImportError(RuntimeError): def read_snapshot(source: str) -> dict: try: if source.startswith(("http://", "https://")): - with urlopen(source.rstrip("/") + "/v1/catalog", timeout=30) as response: + url = source.rstrip("/") + "/v1/catalog" + token = env_value("DSM_SERVER_TOKEN", "IVINS_DATASET_SERVER_TOKEN", "") + require_secure_bearer_transport(url, token) + headers = {"Accept": "application/json"} + if token: + headers["Authorization"] = f"Bearer {token}" + with urlopen(Request(url, headers=headers), timeout=30) as response: value = json.load(response) else: value = json.loads(Path(source).read_text(encoding="utf-8")) - except (OSError, HTTPError, URLError, json.JSONDecodeError) as exc: + except (OSError, HTTPError, URLError, json.JSONDecodeError, TransportSecurityError) as exc: raise ImportError(f"cannot read public catalog: {exc}") from exc validate(value) return value diff --git a/scripts/env_compat.py b/scripts/env_compat.py index dfea758..e61446f 100644 --- a/scripts/env_compat.py +++ b/scripts/env_compat.py @@ -3,11 +3,43 @@ from __future__ import annotations import os +import ipaddress import warnings +from urllib.parse import urlparse _WARNED: set[str] = set() +class TransportSecurityError(RuntimeError): + """A bearer credential would cross an unsafe transport.""" + + +def is_loopback_url(url: str) -> bool: + parsed = urlparse(url) + host = (parsed.hostname or "").rstrip(".").lower() + if host == "localhost": + return True + try: + return ipaddress.ip_address(host).is_loopback + except ValueError: + return False + + +def require_secure_bearer_transport(url: str, token: str | None) -> None: + if not token: + return + parsed = urlparse(url) + if parsed.scheme not in {"http", "https"} or not parsed.hostname: + raise TransportSecurityError("Dataset Server URL must be absolute HTTP(S)") + if parsed.username or parsed.password: + raise TransportSecurityError("Dataset Server URL must not contain credentials") + if parsed.scheme == "http" and not is_loopback_url(url): + raise TransportSecurityError( + "refusing to send DSM_SERVER_TOKEN over plain HTTP; use HTTPS " + "(HTTP is allowed only for a loopback test-server)" + ) + + def value(canonical: str, legacy: str, default: str | None = None) -> str | None: if canonical in os.environ: return os.environ[canonical] diff --git a/scripts/raw_catalog.py b/scripts/raw_catalog.py index bbb2803..7563d40 100644 --- a/scripts/raw_catalog.py +++ b/scripts/raw_catalog.py @@ -25,12 +25,24 @@ from urllib.request import Request, urlopen try: - from .env_compat import value as env_value + from .env_compat import ( + TransportSecurityError, + require_secure_bearer_transport, + value as env_value, + ) except ImportError: try: - from scripts.env_compat import value as env_value + from scripts.env_compat import ( + TransportSecurityError, + require_secure_bearer_transport, + value as env_value, + ) except ImportError: # direct script execution - from env_compat import value as env_value + from env_compat import ( + TransportSecurityError, + require_secure_bearer_transport, + value as env_value, + ) try: import yaml @@ -1017,6 +1029,10 @@ def server_request( url: str, method: str = "GET", payload: bytes | None = None, token: str | None = None, content_type: str = "application/json", ) -> tuple[object, dict[str, str]]: + try: + require_secure_bearer_transport(url, token) + except TransportSecurityError as exc: + raise CatalogError(str(exc)) from exc headers = {"Accept": "application/json"} if payload is not None: headers["Content-Type"] = content_type @@ -1069,6 +1085,10 @@ def fetch_from_server( request_headers = {"Accept": "application/octet-stream"} token = env_value(TOKEN_ENV, LEGACY_TOKEN_ENV) if token: + try: + require_secure_bearer_transport(url, token) + except TransportSecurityError as exc: + raise CatalogError(str(exc)) from exc request_headers["Authorization"] = f"Bearer {token}" request_value = Request(url, headers=request_headers) with os.fdopen(fd, "wb") as stream, urlopen(request_value, timeout=60) as response: @@ -1118,6 +1138,10 @@ def fetch_from_server( def upload_file(url: str, source: Path, token: str) -> dict[str, object]: + try: + require_secure_bearer_transport(url, token) + except TransportSecurityError as exc: + raise CatalogError(str(exc)) from exc parsed = urlparse(url) connection_type = http.client.HTTPSConnection if parsed.scheme == "https" else http.client.HTTPConnection connection = connection_type(parsed.hostname, parsed.port, timeout=60) diff --git a/test-suites/smoke.yaml b/test-suites/smoke.yaml index 8060fd7..9838c3b 100644 --- a/test-suites/smoke.yaml +++ b/test-suites/smoke.yaml @@ -1,5 +1,19 @@ -schema_version: 1 +schema_version: "1.0" name: smoke -description: "Starter suite; add cases only after recordings and configurations are reviewed." -cases: [] +version: "1.0.0" +description: "Dataset smoke gate for validated hardware/runtime profiles." +cases: + - dataset_id: iv.dev.4.ff.1 + implementation: vins-neo + acceptance: + required_topic_min_messages: + camera: 1 + imu: 1 + minimum_odometry_messages: 10 + odometry_timestamps_monotonic: true + finite_position_orientation: true + processes_alive_until_bag_complete: true + timeout_seconds: 900 + ground_truth_required: false + delta_xy_xyz_blocking: false diff --git a/tests/test_manifests.py b/tests/test_manifests.py index cfb4bc3..7316751 100644 --- a/tests/test_manifests.py +++ b/tests/test_manifests.py @@ -30,6 +30,18 @@ def test_dataset_manifests(self): with self.subTest(manifest=str(manifest.relative_to(ROOT))): jsonschema.validate(yaml.safe_load(manifest.read_text(encoding="utf-8")), schema) + def test_runtime_profiles_and_suites(self): + runtime_schema = json.loads( + (ROOT / "schemas/runtime-profile.schema.json").read_text(encoding="utf-8") + ) + suite_schema = json.loads( + (ROOT / "schemas/test-suite.schema.json").read_text(encoding="utf-8") + ) + for profile in sorted((ROOT / "runtime-profiles").glob("*.yaml")): + jsonschema.validate(yaml.safe_load(profile.read_text(encoding="utf-8")), runtime_schema) + for suite in sorted((ROOT / "test-suites").glob("*.yaml")): + jsonschema.validate(yaml.safe_load(suite.read_text(encoding="utf-8")), suite_schema) + def test_contract_fixtures(self): fixtures = { "dataset-manifest.schema.json": "dataset-manifest.valid.json", diff --git a/tests/test_runtime.py b/tests/test_runtime.py new file mode 100644 index 0000000..6959bed --- /dev/null +++ b/tests/test_runtime.py @@ -0,0 +1,108 @@ +#!/usr/bin/env python3 +"""Runtime-profile, security and run-manifest unit contracts.""" + +import tempfile +from pathlib import Path +import unittest + +from rosbags.rosbag1 import Writer +from rosbags.typesys import Stores, get_typestore + +from dsm_client.profiles import canonical_profile +from dsm_client.runtime import ( + PrepareRunError, + _convert_rosbag, + _resource, + _safe_component, + _same_origin_download_url, + _sha256_file, + _validate_bag, + load_runtime_profile, + stable_json, +) +from dsm_client.transport import ( + TransportSecurityError, + is_loopback_url, + require_secure_bearer_transport, +) + + +class RuntimeContractTest(unittest.TestCase): + def test_dev_4_alias_resolves_to_validated_runtime_profile(self): + self.assertEqual("dev_04", canonical_profile("dev_4")) + profile = load_runtime_profile("dev_4") + self.assertEqual("dev_04", profile["profile"]) + self.assertEqual("/cam0/image_raw", profile["topics"]["camera"]) + self.assertEqual("/imu0", profile["topics"]["imu"]) + self.assertEqual("/vins_estimator/odometry", profile["topics"]["odometry"]) + + def test_unvalidated_profile_is_explicitly_unavailable(self): + with self.assertRaisesRegex(PrepareRunError, "runtime profile unavailable: dev_03"): + load_runtime_profile("dev_03") + + def test_config_lookup_and_path_traversal(self): + profile = load_runtime_profile("dev_04") + config = _resource(profile["implementations"]["vins-neo"]["config"]) + self.assertTrue(config.is_file()) + with self.assertRaisesRegex(PrepareRunError, "invalid dataset ID"): + _safe_component("../private", "dataset ID") + with self.assertRaisesRegex(PrepareRunError, "escapes"): + _resource("../outside") + + def test_json_output_is_stable(self): + value = {"z": [2, 1], "a": {"value": True}} + self.assertEqual(stable_json(value), stable_json(value)) + self.assertLess(stable_json(value).find('"a"'), stable_json(value).find('"z"')) + + def test_bearer_token_requires_https_except_loopback(self): + for url in ("http://127.0.0.1:8080", "http://[::1]:8080", "http://localhost:8080"): + self.assertTrue(is_loopback_url(url)) + require_secure_bearer_transport(url, "secret") + require_secure_bearer_transport("https://datasetsmanager.drone-age.org", "secret") + with self.assertRaisesRegex(TransportSecurityError, "plain HTTP"): + require_secure_bearer_transport("http://datasetsmanager.drone-age.org", "secret") + with self.assertRaises(TransportSecurityError): + require_secure_bearer_transport("https://user@example.org", "secret") + self.assertEqual( + "https://datasetsmanager.drone-age.org/v1/artifact", + _same_origin_download_url( + "https://datasetsmanager.drone-age.org", "/v1/artifact" + ), + ) + with self.assertRaisesRegex(PrepareRunError, "cross-origin"): + _same_origin_download_url( + "https://datasetsmanager.drone-age.org", + "https://attacker.example/artifact", + ) + + def test_pinned_ros1_conversion_records_safe_reusable_rosbag2(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + source = root / "source.bag" + typestore = get_typestore(Stores.ROS1_NOETIC) + message_type = "std_msgs/msg/String" + raw = typestore.serialize_ros1( + typestore.types[message_type](data="fixture"), message_type + ) + with Writer(source) as writer: + camera = writer.add_connection( + "/cam0/image_raw", message_type, typestore=typestore + ) + imu = writer.add_connection("/imu0", message_type, typestore=typestore) + writer.write(camera, 1, raw) + writer.write(imu, 2, raw) + source_sha, _ = _sha256_file(source) + converted, converted_sha, converted_size = _convert_rosbag( + source, "iv.dev.4.ff.1", source_sha, root / "cache" + ) + self.assertTrue((converted / "metadata.yaml").is_file()) + self.assertGreater(converted_size, 0) + _validate_bag(converted, False) + reused, reused_sha, reused_size = _convert_rosbag( + source, "iv.dev.4.ff.1", source_sha, root / "cache" + ) + self.assertEqual((converted, converted_sha, converted_size), (reused, reused_sha, reused_size)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_v1_integration.py b/tests/test_v1_integration.py index 64558e3..a338622 100644 --- a/tests/test_v1_integration.py +++ b/tests/test_v1_integration.py @@ -83,6 +83,66 @@ def test_publish_snapshot_import_fetch_byte_identical(self): hashlib.sha256(artifact.read_bytes()).hexdigest(), json.loads(fetched.stdout)["sha256"], ) + + prepare_command = [ + sys.executable, "-m", "dsm_client.cli", "prepare-run", + "iv.dev.4.ff.1", "--implementation", "vins-neo", + "--suite", "smoke", "--format", "rosbag", + "--server-url", base, "--cache-dir", str(root / "run-cache"), + "--allow-test-fixture", + ] + no_token = {key: value for key, value in env.items() if key != "DSM_SERVER_TOKEN"} + denied_prepare = subprocess.run( + prepare_command, cwd=REPO, capture_output=True, text=True, env=no_token + ) + self.assertEqual(2, denied_prepare.returncode) + self.assertIn("DSM_SERVER_TOKEN is required", denied_prepare.stderr) + + wrong_token = {**env, "DSM_SERVER_TOKEN": "dsm_invalid"} + unauthorized = subprocess.run( + prepare_command, cwd=REPO, capture_output=True, text=True, env=wrong_token + ) + self.assertEqual(2, unauthorized.returncode) + self.assertIn("Dataset Server request failed (401)", unauthorized.stderr) + self.assertNotIn("dsm_invalid", unauthorized.stdout + unauthorized.stderr) + + prepared = subprocess.run( + prepare_command, cwd=REPO, check=True, capture_output=True, text=True, env=env + ) + run_manifest = json.loads(prepared.stdout) + self.assertEqual("dev_04", run_manifest["dataset"]["profile"]) + self.assertEqual("vins-neo", run_manifest["implementation"]) + self.assertEqual("1", run_manifest["artifact"]["version"]) + self.assertEqual( + hashlib.sha256(artifact.read_bytes()).hexdigest(), + run_manifest["artifact"]["sha256"], + ) + self.assertEqual(10, run_manifest["acceptance"]["minimum_odometry_messages"]) + + real_gate = subprocess.run( + [ + *prepare_command[:-1], + "--cache-dir", str(root / "real-run-cache"), + ], + cwd=REPO, capture_output=True, text=True, env=env, + ) + self.assertEqual(2, real_gate.returncode) + self.assertIn("fixture-sized artifact is forbidden", real_gate.stderr) + + cached_path = Path(run_manifest["artifact"]["path"]) + cached_path.write_bytes(b"corrupt") + corrupt = subprocess.run( + prepare_command, cwd=REPO, capture_output=True, text=True, env=env + ) + self.assertEqual(2, corrupt.returncode) + self.assertIn("checksum mismatch in cached artifact", corrupt.stderr) + + mismatch = subprocess.run( + [*prepare_command, "--profile", "dev_03"], + cwd=REPO, capture_output=True, text=True, env=env, + ) + self.assertEqual(2, mismatch.returncode) + self.assertIn("dataset/profile mismatch", mismatch.stderr) finally: httpd.shutdown() thread.join(timeout=5)