feat(ISV-7116): create script for direct container signing (#819)

A new script rh_direct_sign_image.py is created to support direct
container signing method. The script will be used in the new Tekton task
that replaces current signing method.

The script resolves all digests for components, verifies if images are
already signed and filter them out. Whatever remains is signed using
internal request and new direct signing pipeline.

- PyxisSignature/SigningItem dataclasses for structured data
- Multi-arch digest collection via skopeo inspect --raw
- Source container resolution via select-oci-auth + oras resolve
- ConfigMap reading for SIG_KEY_NAME(S)
- Greedy base64 batching with --batch-max-size and --output args
- Full unit test suite (34 tests)

Assisted-by: Claude Sonnet 4.6

Signed-off-by: Ales Raszka <araszka@redhat.com>
Co-authored-by: Jan Koscielniak <jakoscie@redhat.com>

Author: Allda

Dockerfile

@@ -172,6 +172,7 @@ ENV PATH="$PATH:/home/pubtools-pulp-wrapper"
 ENV PATH="$PATH:/home/pubtools-marketplacesvm-wrapper"
 ENV PATH="$PATH:/home/developer-portal-wrapper"
 ENV PATH="$PATH:/home/publish-to-cgw-wrapper"
+ENV PATH="$PATH:/home/scripts/python/tasks/managed"
 # Flat imports: helpers and task scripts must be importable.
 # Tests use the same layout via pyproject [tool.pytest.ini_options] pythonpath.
 # Keep /home for other modules (e.g. pyxis, sbom) that expect it.

pyproject.toml

@@ -34,6 +34,7 @@ line-length = 95
 pythonpath = [
   ".",
   "integration-tests/lib",
+  "pyxis",
   "scripts/python/helpers",
   "scripts/python/tasks/internal",
   "scripts/python/tasks/managed",
@@ -45,6 +46,7 @@ pythonpath = [
 [dependency-groups]
 dev = [
     "black>=25.11.0",
+    "flake8>=7.3.0",
     "pytest>=8.4.2",
     "pytest-cov>=6.0",
 ]

scripts/python/helpers/kubectl.py

@@ -0,0 +1,27 @@
+"""Helpers for interacting with Kubernetes via kubectl."""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from subprocess_cmd import run_cmd
+
+
+def get_configmap(name: str) -> dict[str, Any]:
+    """Fetch a Kubernetes ConfigMap by name and return its parsed JSON.
+
+    Args:
+        name: The ConfigMap resource name to retrieve.
+
+    Returns:
+        The full ConfigMap object as a parsed dictionary.
+
+    Raises:
+        RuntimeError: If kubectl exits with a non-zero return code.
+
+    """
+    result = run_cmd(["kubectl", "get", f"cm/{name}", "-ojson"], check=False)
+    if result.returncode != 0:
+        raise RuntimeError(f"Failed to retrieve ConfigMap '{name}': {result.stderr.strip()}")
+    return json.loads(result.stdout)

scripts/python/helpers/oras_utils.py

@@ -4,8 +4,37 @@
 
 import re
 import subprocess
+import tempfile
 from pathlib import Path
 
+from subprocess_cmd import run_cmd
+
+
+def oras_resolve(reference: str) -> str:
+    """Resolve the digest of an OCI image reference using oras.
+
+    Obtains registry credentials via ``select-oci-auth``, writes them to a
+    temporary auth file, then runs ``oras resolve`` and returns the digest.
+
+    Raises ``RuntimeError`` if oras resolve exits non-zero.
+    """
+    with tempfile.NamedTemporaryFile(mode="w", suffix=".json") as auth_file:
+        select_auth = run_cmd(["select-oci-auth", reference])
+        auth_file.write(select_auth.stdout)
+        auth_file.flush()
+
+        result = run_cmd(
+            ["oras", "resolve", "--registry-config", auth_file.name, reference],
+            check=False,
+        )
+
+    if result.returncode != 0:
+        raise RuntimeError(
+            f"oras resolve failed for {reference!r} (exit {result.returncode}):"
+            f" {result.stderr.strip()}"
+        )
+    return result.stdout.strip()
+
 
 def oras_login(registry: str, username: str, password: str) -> None:
     """Log in to an OCI registry via oras using username/password credentials.

scripts/python/helpers/subprocess_cmd.py

@@ -26,7 +26,7 @@ def run_cmd(
     """Run *cmd*; capture stdout as text; optionally append stderr to *stderr_path*."""
     # Child must inherit pod env (PATH, KUBECONFIG, etc.); only overlay *env*.
     merged: dict[str, str] = {**os.environ, **dict(env or {})}
-    err_f: Any = subprocess.DEVNULL
+    err_f: Any = subprocess.PIPE
     fh: Any = None
     try:
         if stderr_path is not None:

scripts/python/helpers/test_kubectl.py

@@ -0,0 +1,33 @@
+"""Tests for `kubect`."""
+
+from __future__ import annotations
+
+import json
+from unittest.mock import MagicMock, patch
+
+from kubectl import get_configmap
+
+
+def test_get_configmap_runs_kubectl_and_returns_parsed_json() -> None:
+    """Kubectl is called with the correct arguments and its output is parsed as JSON."""
+    cm_json = json.dumps({"data": {"SIG_KEY_NAME": "some-key"}})
+    with patch("kubectl.run_cmd") as mock_run:
+        mock_run.return_value = MagicMock(stdout=cm_json, returncode=0)
+        result = get_configmap("signing-config-map")
+
+    mock_run.assert_called_once_with(
+        ["kubectl", "get", "cm/signing-config-map", "-ojson"], check=False
+    )
+    assert result == {"data": {"SIG_KEY_NAME": "some-key"}}
+
+
+def test_get_configmap_raises_on_kubectl_failure() -> None:
+    """RuntimeError is raised with the configmap name and stderr when kubectl fails."""
+    import pytest
+
+    with patch("kubectl.run_cmd") as mock_run:
+        mock_run.return_value = MagicMock(
+            returncode=1, stderr="Error from server (NotFound): configmaps not found"
+        )
+        with pytest.raises(RuntimeError, match="signing-config-map"):
+            get_configmap("signing-config-map")

scripts/python/helpers/test_oras_utils.py

@@ -0,0 +1,62 @@
+"""Unit tests for oras_utils."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, call, patch
+
+import pytest
+
+from oras_utils import oras_resolve
+
+
+def test_oras_resolve_calls_select_oci_auth_with_reference() -> None:
+    """select-oci-auth is called with the image reference."""
+    with patch("oras_utils.run_cmd") as mock_run:
+        mock_run.side_effect = [
+            MagicMock(stdout='{"auths": {}}'),
+            MagicMock(returncode=0, stdout="sha256:abc\n"),
+        ]
+        oras_resolve("registry.io/repo:tag")
+
+    first_call = mock_run.call_args_list[0]
+    assert first_call == call(["select-oci-auth", "registry.io/repo:tag"])
+
+
+def test_oras_resolve_passes_auth_file_to_oras() -> None:
+    """Oras resolve is called with --registry-config pointing to the auth temp file."""
+    with patch("oras_utils.run_cmd") as mock_run:
+        mock_run.side_effect = [
+            MagicMock(stdout='{"auths": {}}'),
+            MagicMock(returncode=0, stdout="sha256:abc\n"),
+        ]
+        oras_resolve("registry.io/repo:tag")
+
+    second_call = mock_run.call_args_list[1]
+    cmd = second_call.args[0]
+    assert cmd[0] == "oras"
+    assert cmd[1] == "resolve"
+    assert "--registry-config" in cmd
+    assert "registry.io/repo:tag" in cmd
+
+
+def test_oras_resolve_returns_stripped_digest() -> None:
+    """Returns the digest from oras resolve output, stripped of whitespace."""
+    with patch("oras_utils.run_cmd") as mock_run:
+        mock_run.side_effect = [
+            MagicMock(stdout="{}"),
+            MagicMock(returncode=0, stdout="sha256:deadbeef\n"),
+        ]
+        result = oras_resolve("registry.io/repo:tag")
+
+    assert result == "sha256:deadbeef"
+
+
+def test_oras_resolve_raises_on_nonzero_returncode() -> None:
+    """Raises RuntimeError when oras resolve exits non-zero."""
+    with patch("oras_utils.run_cmd") as mock_run:
+        mock_run.side_effect = [
+            MagicMock(stdout="{}"),
+            MagicMock(returncode=1, stdout="", stderr="unauthorized"),
+        ]
+        with pytest.raises(RuntimeError):
+            oras_resolve("registry.io/repo:tag")

scripts/python/helpers/test_subprocess_cmd.py

@@ -41,6 +41,12 @@ def test_run_cmd_stderr_path_on_success(tmp_path) -> None:
     assert log.read_text(encoding="utf-8") == ""
 
 
+def test_run_cmd_captures_stderr_when_no_path_given() -> None:
+    """Stderr is captured in the return value when no stderr_path is provided."""
+    r = subprocess_cmd.run_cmd(["sh", "-c", "echo error-msg >&2"], check=True)
+    assert "error-msg" in r.stderr
+
+
 def test_run_cmd_text_success() -> None:
     """``run_cmd_text`` returns stdout from a successful subprocess."""
     with mock.patch("subprocess_cmd.subprocess.run") as run: