Add tests for signing manifests with a service

Required PR: pulp/pulpcore#2915

closes pulp#687

CHANGES/687.misc

@@ -0,0 +1 @@
+Added a test case for signing manifests by leveraging a signing service.

pulp_container/app/models.py

@@ -1,6 +1,5 @@
 from gettext import gettext as _
 
-import gnupg
 import json
 import os
 import re
@@ -26,6 +25,7 @@
     Upload as CoreUpload,
 )
 from pulpcore.plugin.repo_version_utils import remove_duplicates, validate_repo_version
+from pulpcore.plugin.util import verify_signature
 
 
 from . import downloaders
@@ -452,19 +452,7 @@ def validate(self):
                 manifest_file.name, env_vars={"REFERENCE": "test", "SIG_PATH": sig_path}
             )
 
-            with open(signed["signature_path"], "rb") as fp:
-                gpg = gnupg.GPG()
-                verified = gpg.verify_file(fp)
-                if verified.trust_level is None or verified.trust_level < verified.TRUST_FULLY:
-                    raise RuntimeError(
-                        "The signature could not be verified or the trust level is too "
-                        "low. The signing script may generate invalid signatures."
-                    )
-                elif verified.pubkey_fingerprint != self.pubkey_fingerprint:
-                    raise RuntimeError(
-                        "Fingerprints of the provided public key and the verified public "
-                        "key are not equal. The signing script is probably not valid."
-                    )
+            verify_signature(signed["signature_path"], self.public_key)
 
 
 class ContainerRepository(

pulp_container/tests/functional/api/test_push_signatures.py

@@ -1,57 +1,28 @@
 """Tests that verify that an image signature can be pushed to Pulp."""
-import gnupg
 import base64
 import json
-import requests
 import pytest
-import tempfile
 
 from pulp_container.tests.functional.constants import REGISTRY_V2_REPO_PULP
 from pulp_container.constants import SIGNATURE_TYPE
 
-KEY = "https://raw.githubusercontent.com/pulp/pulp-fixtures/master/common/GPG-PRIVATE-KEY-pulp-qe"
-
-
-@pytest.fixture(scope="session")
-def tempdir():
-    """A temporary directory."""
-    with tempfile.TemporaryDirectory() as directory:
-        yield directory
-
-
-@pytest.fixture(scope="session")
-def gpg_metadata(tempdir):
-    """A fixture for a GPG instance and related metadata."""
-    private_key = requests.get(KEY).text
-
-    gpg = gnupg.GPG(gnupghome=tempdir)
-
-    gpg.import_keys(private_key)
-
-    fingerprint = gpg.list_keys()[0]["fingerprint"]
-    keyid = gpg.list_keys()[0]["keyid"]
-
-    gpg.trust_keys(fingerprint, "TRUST_ULTIMATE")
-
-    return gpg, fingerprint, keyid
-
 
 @pytest.fixture
 def distribution(
     registry_client,
     local_registry,
     container_distribution_api,
-    gpg_metadata,
+    signing_gpg_metadata,
     add_to_cleanup,
 ):
     """Return a distribution created after pushing a signed content to the Pulp Registry."""
     if registry_client.name != "podman":
         pytest.skip("This test requires podman to sign pulled content", allow_module_level=True)
 
-    registry_client.pull(f"{REGISTRY_V2_REPO_PULP}:manifest_a")
-
     image_path = f"{REGISTRY_V2_REPO_PULP}:manifest_a"
-    gpg, fingerprint, keyid = gpg_metadata
+    registry_client.pull(image_path)
+
+    gpg, fingerprint, keyid = signing_gpg_metadata
 
     with registry_client._client.machine.env(GNUPGHOME=gpg.gnupghome):
         local_registry.tag_and_push(image_path, "test-1:manifest_a", "--sign-by", keyid)
@@ -70,11 +41,11 @@ def test_assert_signed_image(
     container_push_repository_api,
     container_manifest_api,
     container_signature_api,
-    gpg_metadata,
+    signing_gpg_metadata,
     distribution,
 ):
     """Test whether an admin user can fetch a signature from the Pulp Registry."""
-    gpg, fingerprint, keyid = gpg_metadata
+    gpg, fingerprint, keyid = signing_gpg_metadata
 
     repository = container_push_repository_api.read(distribution.repository)
     manifest = container_manifest_api.list(

pulp_container/tests/functional/api/test_sign_manifests.py

@@ -0,0 +1,57 @@
+import pytest
+
+from pulp_smash.pulp3.bindings import monitor_task
+
+from pulpcore.client.pulp_container import RepositorySign
+
+from pulp_container.constants import SIGNATURE_TYPE
+from pulp_container.tests.functional.constants import REGISTRY_V2_REPO_PULP
+
+MANIFEST_TAG = "manifest_a"
+
+
+@pytest.fixture
+def distribution(registry_client, local_registry, container_distribution_api, add_to_cleanup):
+    """The fixture for a distribution that references a repository of the push type."""
+    image_path = f"{REGISTRY_V2_REPO_PULP}:{MANIFEST_TAG}"
+    registry_client.pull(image_path)
+    local_registry.tag_and_push(image_path, f"test-1:{MANIFEST_TAG}")
+
+    distribution = container_distribution_api.list(name="test-1").results[0]
+    add_to_cleanup(container_distribution_api, distribution.pulp_href)
+
+    return distribution
+
+
+def test_sign_manifest(
+    signing_gpg_metadata,
+    distribution,
+    signing_service,
+    container_push_repository_api,
+    container_signature_api,
+    container_tag_api,
+    container_manifest_api,
+):
+    """Test whether a user can sign a manifest by leveraging a signing service."""
+    _, _, keyid = signing_gpg_metadata
+    sign_data = RepositorySign(signing_service.pulp_href)
+
+    response = container_push_repository_api.sign(distribution.repository, sign_data)
+    created_resources = monitor_task(response.task).created_resources
+
+    tags = container_tag_api.list(repository_version=created_resources[0])
+    assert tags.count == 1
+
+    tag = tags.results[0]
+    assert tag.name == MANIFEST_TAG
+
+    signatures = container_signature_api.list()
+    assert signatures.count == 1
+
+    signature = signatures.results[0]
+    assert signature.key_id == keyid
+    assert signature.type == SIGNATURE_TYPE.ATOMIC_SHORT
+
+    manifest = container_manifest_api.read(tag.tagged_manifest)
+    assert signature.signed_manifest == manifest.pulp_href
+    assert signature.name.startswith(manifest.digest)

pulp_container/tests/functional/conftest.py

@@ -1,9 +1,14 @@
+import os
+import stat
 import pytest
+import tempfile
+import requests
 
 from urllib.parse import urljoin, urlparse
 
-import requests
+from pulp_smash.utils import execute_pulpcore_python, uuid4
 from pulp_smash.cli import RegistryClient
+
 from pulpcore.client.pulp_container import (
     ApiClient,
     PulpContainerNamespacesApi,
@@ -145,6 +150,78 @@ def inspect(local_url):
     return _LocalRegistry()
 
 
+@pytest.fixture(scope="session")
+def signing_script_filename(signing_gpg_homedir_path):
+    """A fixture for a script that is suited for signing manifests."""
+    raw_script = (
+        "#!/usr/bin/env bash",
+        "",
+        "# use the side channel to set the GNUPGHOME variable",
+        f'export GNUPGHOME="{signing_gpg_homedir_path}"',
+        "",
+        "MANIFEST_PATH=$1",
+        'FINGEPRINT="$PULP_SIGNING_KEY_FINGERPRINT"',
+        "",
+        "skopeo standalone-sign $MANIFEST_PATH $REFERENCE $FINGEPRINT -o $SIG_PATH",
+        "",
+        "STATUS=$?",
+        "if [ $STATUS -eq 0 ]; then",
+        '   echo {\\"signature_path\\": \\"$SIG_PATH\\"}',
+        "else",
+        "   exit $STATUS",
+        "fi",
+        "",
+    )
+
+    with open(os.path.join(signing_gpg_homedir_path, "bash-script.sh"), "w") as f:
+        f.write("\n".join(raw_script))
+
+    return f.name
+
+
+@pytest.fixture
+def signing_service(
+    cli_client,
+    signing_gpg_metadata,
+    signing_script_filename,
+    signing_service_api_client,
+):
+    """A fixture for a signing service."""
+    st = os.stat(signing_script_filename)
+    os.chmod(signing_script_filename, st.st_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH)
+
+    gpg, fingerprint, keyid = signing_gpg_metadata
+
+    service_name = uuid4()
+    cmd = (
+        "pulpcore-manager",
+        "add-signing-service",
+        service_name,
+        signing_script_filename,
+        keyid,
+        "--class",
+        "container:ManifestSigningService",
+        "--gnupghome",
+        gpg.gnupghome,
+    )
+
+    response = cli_client.run(cmd)
+
+    assert response.returncode == 0
+
+    signing_service = signing_service_api_client.list(name=service_name).results[0]
+    assert signing_service.pubkey_fingerprint == fingerprint
+    assert signing_service.public_key == gpg.export_keys(keyid)
+
+    yield signing_service
+
+    cmd = (
+        "from pulpcore.app.models import SigningService;"
+        f"SigningService.objects.filter(name='{service_name}').delete()"
+    )
+    execute_pulpcore_python(cli_client, cmd)
+
+
 @pytest.fixture(scope="session")
 def container_client(bindings_cfg):
     """Fixture for container_client."""