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/charon_env.py

@@ -0,0 +1,82 @@
+"""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
+
+import file
+
+_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
+
+
+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 file.path_from_env_variable("WORK_DIR", NRRC_WORK_DIR_DEFAULT)

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,25 @@ 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
+
+
 def make_tempfile_path(
     prefix: str,
     data: bytes | None = None,
@@ -97,3 +120,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_charon_env.py

@@ -0,0 +1,163 @@
+"""Tests for ``charon_env``."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import charon_env
+import pytest
+
+
+def test_load_charon_env_parses_dotenv_lines(tmp_path: Path) -> None:
+    """Dotenv lines are parsed into a key/value mapping."""
+    env_file = tmp_path / "charon.env"
+    env_file.write_text(
+        "\nCHARON_OCI_REGISTRY=repo@sha256:abcdef0123456789\n"
+        "CHARON_TARGET=dev-npm-ga\n"
+        'CHARON_PRODUCT_NAME="Test Product"\n',
+        encoding="utf-8",
+    )
+    env = charon_env.load_charon_env(env_file)
+    assert env["CHARON_OCI_REGISTRY"] == "repo@sha256:abcdef0123456789"
+    assert env["CHARON_TARGET"] == "dev-npm-ga"
+    assert env["CHARON_PRODUCT_NAME"] == "Test Product"
+
+
+def test_load_charon_env_missing_file(tmp_path: Path) -> None:
+    """Missing env files raise FileNotFoundError."""
+    with pytest.raises(FileNotFoundError):
+        charon_env.load_charon_env(tmp_path / "missing.env")
+
+
+def test_load_charon_env_rejects_invalid_utf8(tmp_path: Path) -> None:
+    """Invalid UTF-8 bytes fail instead of being silently replaced."""
+    env_file = tmp_path / "charon.env"
+    env_file.write_bytes(b"CHARON_TARGET=dev\xff\n")
+    with pytest.raises(UnicodeDecodeError):
+        charon_env.load_charon_env(env_file)
+
+
+def test_split_oci_registries() -> None:
+    """Registry references are split on percent signs."""
+    value = "quay.io/a@sha256:111111%quay.io/b@sha256:222222"
+    assert charon_env.split_oci_registries(value) == [
+        "quay.io/a@sha256:111111",
+        "quay.io/b@sha256:222222",
+    ]
+
+
+def test_short_sha256_prefix() -> None:
+    """Short hash uses the first six digest characters."""
+    assert charon_env.short_sha256_prefix("repo@sha256:0b15aad24f1b847") == "0b15aa"
+
+
+def test_source_repo() -> None:
+    """Source repo strips the digest suffix."""
+    assert charon_env.source_repo("quay.io/org/app@sha256:abc") == "quay.io/org/app"
+
+
+def test_load_charon_env_skips_malformed_lines(tmp_path: Path) -> None:
+    """Lines without ``=`` are ignored."""
+    env_file = tmp_path / "charon.env"
+    env_file.write_text("CHARON_TARGET=dev\nnot-a-variable\n", encoding="utf-8")
+    env = charon_env.load_charon_env(env_file)
+    assert env == {"CHARON_TARGET": "dev"}
+
+
+def test_short_sha256_prefix_requires_digest() -> None:
+    """Registry references without a digest raise ValueError."""
+    with pytest.raises(ValueError, match="@sha256:"):
+        charon_env.short_sha256_prefix("quay.io/org/app:latest")
+
+
+def test_split_oci_registries_ignores_empty_segments() -> None:
+    """Trailing or duplicate ``%`` separators yield no empty entries."""
+    assert charon_env.split_oci_registries("quay.io/a@sha256:111111%") == [
+        "quay.io/a@sha256:111111",
+    ]
+    assert charon_env.split_oci_registries("%quay.io/a@sha256:111111%") == [
+        "quay.io/a@sha256:111111",
+    ]
+
+
+def test_load_charon_env_skips_comments_and_blank_lines(tmp_path: Path) -> None:
+    """Comments and blank lines are ignored."""
+    env_file = tmp_path / "charon.env"
+    env_file.write_text(
+        "# comment\n\nCHARON_TARGET=dev\n# CHARON_FOO=bar\n",
+        encoding="utf-8",
+    )
+    assert charon_env.load_charon_env(env_file) == {"CHARON_TARGET": "dev"}
+
+
+def test_load_charon_env_strips_single_quotes(tmp_path: Path) -> None:
+    """Single-quoted values are unquoted."""
+    env_file = tmp_path / "charon.env"
+    env_file.write_text("CHARON_PRODUCT_NAME='Test Product'\n", encoding="utf-8")
+    assert charon_env.load_charon_env(env_file)["CHARON_PRODUCT_NAME"] == "Test Product"
+
+
+def test_require_env_keys_raises_for_missing_key() -> None:
+    """Missing required keys raise ValueError."""
+    with pytest.raises(ValueError, match="missing required charon env variable"):
+        charon_env.require_env_keys({"CHARON_TARGET": "dev"}, "CHARON_PRODUCT_NAME")
+
+
+def test_require_oci_registries(tmp_path: Path) -> None:
+    """Non-empty registry lists are returned from parsed env data."""
+    env_file = tmp_path / "charon.env"
+    env_file.write_text(
+        "CHARON_OCI_REGISTRY=quay.io/a@sha256:111111%quay.io/b@sha256:222222\n",
+        encoding="utf-8",
+    )
+    env = charon_env.load_charon_env(env_file)
+    assert charon_env.require_oci_registries(env) == [
+        "quay.io/a@sha256:111111",
+        "quay.io/b@sha256:222222",
+    ]
+
+
+def test_require_oci_registries_missing_key() -> None:
+    """Missing ``CHARON_OCI_REGISTRY`` raises ValueError."""
+    with pytest.raises(ValueError, match="CHARON_OCI_REGISTRY is required"):
+        charon_env.require_oci_registries({})
+
+
+def test_require_oci_registries_empty_value() -> None:
+    """Blank ``CHARON_OCI_REGISTRY`` values raise ValueError."""
+    with pytest.raises(ValueError, match="at least one registry"):
+        charon_env.require_oci_registries({"CHARON_OCI_REGISTRY": ""})
+
+
+def test_nrrc_work_dir_default() -> None:
+    """Default NRRC work directory uses the image writable root."""
+    assert charon_env.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 charon_env.nrrc_work_dir() == custom
+
+
+def test_charon_config_path_uses_explicit_home(tmp_path: Path) -> None:
+    """An explicit *home* overrides ``Path.home()``."""
+    assert charon_env.charon_config_path(home=tmp_path) == tmp_path / ".charon" / "charon.yaml"
+
+
+def test_install_charon_config(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """Charon config is copied into ``$HOME/.charon/charon.yaml``."""
+    monkeypatch.setenv("HOME", str(tmp_path))
+    config_path = charon_env.charon_config_path()
+    source = tmp_path / "input.yaml"
+    source.write_text("charon-config\n", encoding="utf-8")
+    dest = charon_env.install_charon_config(source)
+    assert dest == config_path
+    assert config_path.read_text(encoding="utf-8") == "charon-config\n"
+
+
+def test_install_charon_config_missing_source(tmp_path: Path) -> None:
+    """Missing config sources raise FileNotFoundError."""
+    with pytest.raises(FileNotFoundError, match="charon config file not found"):
+        charon_env.install_charon_config(tmp_path / "missing.yaml")

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,31 @@ 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_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 +121,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)