feat(RELEASE-2509): add python scripts for publish-to-nrrc task

add python scripts for publish-to-nrrc task

Signed-off-by: Elena German <elgerman@redhat.com>
Assisted-by: Claude

Author: elenagerman

pyproject.toml

@@ -25,6 +25,7 @@ dependencies = [
     "diffused-lib==0.3.0",
     "confluent-kafka",
     "python-gitlab>=4.0",
+    "python-dotenv>=1.0.0",
 ]
 
 [tool.black]

scripts/python/helpers/file.py

@@ -7,7 +7,10 @@
 import gzip
 import io
 import os
+import re
+import subprocess
 import tempfile
+from collections.abc import Callable, Sequence
 from pathlib import Path
 from typing import Any
 
@@ -23,6 +26,7 @@ def load_json_dict(path: Path) -> dict[str, Any]:
 
 
 _GZIP_READ_CHUNK_SIZE = 64 * 1024
+_ARCHIVE_TYPE = re.compile(r"(gzip compressed data|POSIX tar archive)")
 
 
 def sha256(path: Path) -> str:
@@ -57,6 +61,33 @@ def path_from_env_variable(
     return default if isinstance(default, Path) else Path(default)
 
 
+def resolve_path_under_base(base: Path, relative: str | Path) -> Path:
+    """Resolve *relative* under *base* and ensure the result stays inside *base*.
+
+    Rejects absolute paths and ``..`` traversal after resolution. Typical use:
+    Tekton passes a path relative to a data directory (e.g. charon env/config files).
+    """
+    text = str(relative).strip()
+    if not text:
+        raise ValueError(f"path must be relative to {base}: {relative!r}")
+    rel = Path(text)
+    if rel.is_absolute():
+        raise ValueError(f"path must be relative to {base}: {relative!r}")
+    root = base.resolve()
+    candidate = (root / rel).resolve()
+    if not candidate.is_relative_to(root):
+        raise ValueError(f"path must stay under {base}: {relative!r}")
+    return candidate
+
+
+NRRC_WORK_DIR_DEFAULT = Path("/var/workdir/nrrc")
+
+
+def nrrc_work_dir() -> Path:
+    """Return the NRRC staging directory from ``WORK_DIR`` or ``/var/workdir/nrrc``."""
+    return path_from_env_variable("WORK_DIR", NRRC_WORK_DIR_DEFAULT)
+
+
 def make_tempfile_path(
     prefix: str,
     data: bytes | None = None,
@@ -97,3 +128,13 @@ def decompress_gzip_bounded(data: bytes, *, max_bytes: int) -> bytes:
                 msg = f"decompressed data exceeds {max_bytes} bytes (possible gzip bomb)"
                 raise ValueError(msg)
     return bytes(output)
+
+
+def is_gzip_or_tar_archive(
+    path: Path,
+    *,
+    file_cmd: Callable[[Sequence[str | Path]], subprocess.CompletedProcess[str]],
+) -> bool:
+    """Return True when ``file -b`` reports gzip or tar content for *path*."""
+    result = file_cmd(["file", "-b", str(path)])
+    return _ARCHIVE_TYPE.search(result.stdout) is not None

scripts/python/helpers/test_file.py

@@ -4,6 +4,8 @@
 
 import json
 import gzip
+import subprocess
+from collections.abc import Sequence
 from pathlib import Path
 
 import file
@@ -63,6 +65,43 @@ def test_load_json_dict_rejects_non_object(tmp_path: Path) -> None:
         file.load_json_dict(path)
 
 
+def test_resolve_path_under_base_relative_file(tmp_path: Path) -> None:
+    """A normal relative path resolves under *base*."""
+    target = tmp_path / "uid" / "charon.env"
+    target.parent.mkdir(parents=True)
+    assert file.resolve_path_under_base(tmp_path, "uid/charon.env") == target.resolve()
+
+
+def test_resolve_path_under_base_rejects_absolute(tmp_path: Path) -> None:
+    """Absolute paths are rejected even if they exist."""
+    with pytest.raises(ValueError, match="must be relative"):
+        file.resolve_path_under_base(tmp_path, "/etc/passwd")
+
+
+def test_resolve_path_under_base_rejects_traversal(tmp_path: Path) -> None:
+    """``..`` segments that escape *base* are rejected."""
+    with pytest.raises(ValueError, match="must stay under"):
+        file.resolve_path_under_base(tmp_path, "../outside")
+
+
+def test_resolve_path_under_base_rejects_blank(tmp_path: Path) -> None:
+    """Blank relative paths are rejected."""
+    with pytest.raises(ValueError, match="must be relative"):
+        file.resolve_path_under_base(tmp_path, "   ")
+
+
+def test_nrrc_work_dir_default() -> None:
+    """Default NRRC work directory uses the image writable root."""
+    assert file.nrrc_work_dir() == Path("/var/workdir/nrrc")
+
+
+def test_nrrc_work_dir_from_env(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    """``WORK_DIR`` overrides the default NRRC staging directory."""
+    custom = tmp_path / "staging"
+    monkeypatch.setenv("WORK_DIR", str(custom))
+    assert file.nrrc_work_dir() == custom
+
+
 def test_make_tempfile_path_empty_file() -> None:
     """A `None` payload leaves the created file with zero length."""
     p = file.make_tempfile_path("t-", None)
@@ -94,3 +133,35 @@ def test_decompress_gzip_bounded_rejects_oversized_output() -> None:
     compressed = gzip.compress(raw)
     with pytest.raises(ValueError, match="gzip bomb"):
         file.decompress_gzip_bounded(compressed, max_bytes=1000)
+
+
+def test_is_gzip_or_tar_archive_posix_tar() -> None:
+    """POSIX tar archives are recognized."""
+
+    def fake_file_cmd(
+        cmd: Sequence[str | Path],
+    ) -> subprocess.CompletedProcess[str]:
+        return subprocess.CompletedProcess(
+            [str(x) for x in cmd],
+            0,
+            stdout="POSIX tar archive\n",
+            stderr="",
+        )
+
+    assert file.is_gzip_or_tar_archive(Path("/tmp/archive.tar"), file_cmd=fake_file_cmd)
+
+
+def test_is_gzip_or_tar_archive_rejects_other_types() -> None:
+    """Non-archive ``file -b`` output returns False."""
+
+    def fake_file_cmd(
+        cmd: Sequence[str | Path],
+    ) -> subprocess.CompletedProcess[str]:
+        return subprocess.CompletedProcess(
+            [str(x) for x in cmd],
+            0,
+            stdout="ASCII text\n",
+            stderr="",
+        )
+
+    assert not file.is_gzip_or_tar_archive(Path("/tmp/readme.txt"), file_cmd=fake_file_cmd)

scripts/python/tasks/managed/charon_env.py

@@ -0,0 +1,72 @@
+"""Parse charon parameter files used by NRRC/MRRC publish tasks."""
+
+from __future__ import annotations
+
+import re
+import shutil
+from pathlib import Path
+
+from dotenv import dotenv_values
+
+_REGISTRY_SPLIT = re.compile(r"%")
+
+
+def load_charon_env(path: Path) -> dict[str, str]:
+    """Load charon parameters from a dotenv file (``KEY=value`` lines)."""
+    if not path.is_file():
+        raise FileNotFoundError(f"charon env file not found: {path}")
+    values = dotenv_values(path, encoding="utf-8")
+    return {key: value for key, value in values.items() if value is not None}
+
+
+def split_oci_registries(value: str) -> list[str]:
+    """Split ``CHARON_OCI_REGISTRY`` on ``%`` into non-empty registry references."""
+    return [part.strip() for part in _REGISTRY_SPLIT.split(value) if part.strip()]
+
+
+def short_sha256_prefix(registry: str) -> str:
+    """Return the first six characters of the digest in *registry*."""
+    marker = "@sha256:"
+    if marker not in registry:
+        raise ValueError(f"registry reference missing @sha256: digest: {registry!r}")
+    return registry.split(marker, 1)[1][:6]
+
+
+def source_repo(registry: str) -> str:
+    """Return the repository part of an OCI reference (before ``@sha256:``)."""
+    return registry.split("@sha256:", 1)[0]
+
+
+def require_env_keys(env: dict[str, str], *keys: str) -> None:
+    """Raise ValueError when any *keys* are missing from *env*."""
+    for key in keys:
+        if key not in env:
+            raise ValueError(f"missing required charon env variable: {key}")
+
+
+def require_oci_registries(env: dict[str, str]) -> list[str]:
+    """Return non-empty ``CHARON_OCI_REGISTRY`` entries from *env*."""
+    try:
+        value = env["CHARON_OCI_REGISTRY"]
+    except KeyError as e:
+        raise ValueError("CHARON_OCI_REGISTRY is required in charon env file") from e
+    registries = split_oci_registries(value)
+    if not registries:
+        raise ValueError("CHARON_OCI_REGISTRY must list at least one registry reference")
+    return registries
+
+
+def charon_config_path(*, home: Path | None = None) -> Path:
+    """Return the default charon configuration file path under *home* or ``Path.home()``."""
+    root = home if home is not None else Path.home()
+    return root / ".charon" / "charon.yaml"
+
+
+def install_charon_config(config_source: Path, *, home: Path | None = None) -> Path:
+    """Copy the charon config into ``$HOME/.charon/charon.yaml``."""
+    if not config_source.is_file():
+        raise FileNotFoundError(f"charon config file not found: {config_source}")
+    dest = charon_config_path(home=home)
+    dest.parent.mkdir(parents=True, exist_ok=True)
+    shutil.copy2(config_source, dest)
+    return dest

scripts/python/tasks/managed/publish_to_nrrc_prepare_repo.py

@@ -0,0 +1,92 @@
+#!/usr/bin/env python3
+"""Download npm archives from OCI registries for publish-to-nrrc.
+
+Tekton injects ``DATA_DIR``, ``CHARON_PARAM_FILE_PATH``, and optionally
+``WORK_DIR`` (default ``/var/workdir/nrrc``; catalog sets ``/workdir/nrrc``) via env.
+"""
+
+from __future__ import annotations
+
+import subprocess
+from collections.abc import Callable
+from pathlib import Path
+
+import charon_env
+import file
+import subprocess_cmd
+import tekton
+from logger import logger
+
+
+def prepare_repo(
+    *,
+    charon_param_file: Path,
+    work_dir: Path,
+    run_cmd: Callable[..., subprocess.CompletedProcess[str]] = subprocess_cmd.run_cmd,
+) -> None:
+    """Download OCI archives and collect gzip/tar files under ``shared/``."""
+    env = charon_env.load_charon_env(charon_param_file)
+    registries = charon_env.require_oci_registries(env)
+
+    repo_dir = work_dir
+    shared_repo = repo_dir / "shared"
+    shared_repo.mkdir(parents=True, exist_ok=True)
+
+    for registry in registries:
+        logger.info("Downloading the npm archive from %s", registry)
+        short_hash = charon_env.short_sha256_prefix(registry)
+        subdir = repo_dir / short_hash
+        subdir.mkdir(parents=True, exist_ok=True)
+
+        source_repo = charon_env.source_repo(registry)
+        auth_file = file.make_tempfile_path("oci-auth-", None)
+        try:
+            auth = run_cmd(
+                ["select-oci-auth", source_repo],
+                check=True,
+            ).stdout
+            auth_file.write_text(auth, encoding="utf-8")
+            run_cmd(
+                [
+                    "oras",
+                    "pull",
+                    "--registry-config",
+                    str(auth_file),
+                    registry,
+                    "-o",
+                    str(subdir),
+                ],
+                check=True,
+            )
+        finally:
+            auth_file.unlink(missing_ok=True)
+
+        for found in subdir.rglob("*"):
+            if not found.is_file():
+                continue
+            if not file.is_gzip_or_tar_archive(found, file_cmd=run_cmd):
+                continue
+            move_to = shared_repo / f"{short_hash}_{found.name}"
+            if move_to.exists():
+                logger.warning("%s already exists, skipped", move_to)
+                continue
+            found.rename(move_to)
+
+
+def main() -> int:
+    """Read Tekton env vars and prepare npm archives for upload."""
+    data_dir = Path(tekton.require_env("DATA_DIR"))
+    work_dir = file.nrrc_work_dir()
+    charon_param_file = file.resolve_path_under_base(
+        data_dir,
+        tekton.require_env("CHARON_PARAM_FILE_PATH"),
+    )
+    prepare_repo(
+        charon_param_file=charon_param_file,
+        work_dir=work_dir,
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())