Improve post-quantum signature test coverage

coverage.md

@@ -38,7 +38,7 @@ This file contains list of features and their test coverage.
 | **Publish** |  |  |
 | As a user, I can publish repodata with specific checksum type | PART | "on_demand is not covered, https://pulp.plan.io/issues/6503" |
 | As a user, I have the published root directory containing the ‘Package’ directory and packages in alphabetical order inside it. | YES | testing with modularity and kickstarter repositories, contains test if no extra files are present |
-| As a user, I can sign repository metadata using a signing service and publish such repo | PART |  |
+| As a user, I can sign repository metadata using a signing service and publish such repo | YES |  |
 | As a user, I can have a config.repo file generated for any distribution at runtime | YES |  |
 | As a user I can set/update repo_gpgcheck and gpg_check options | YES | |
 | **Upload** |  |  |

pulp_rpm/tests/functional/api/test_character_encoding.py

@@ -3,7 +3,6 @@
 import uuid
 
 import pytest
-import requests
 
 from pulpcore.tests.functional.utils import PulpTaskError
 
@@ -13,6 +12,7 @@
     RPM_WITH_NON_UTF_8_NAME,
     RPM_WITH_NON_UTF_8_URL,
 )
+from pulp_rpm.tests.functional.utils import fetch_url
 
 """Test upload of RPMs with different character encoding.
 
@@ -28,7 +28,7 @@ def test_upload_non_ascii(
 ):
     """Test whether one can upload an RPM with non-ascii metadata."""
     temp_file = tmp_path / str(uuid.uuid4())
-    temp_file.write_bytes(requests.get(RPM_WITH_NON_ASCII_URL).content)
+    temp_file.write_bytes(fetch_url(RPM_WITH_NON_ASCII_URL))
     artifact = pulpcore_bindings.ArtifactsApi.create(str(temp_file))
     response = rpm_package_api.create(
         artifact=artifact.pulp_href,
@@ -43,7 +43,7 @@ def test_upload_non_utf8(
 ):
     """Test whether an exception is raised when non-utf-8 is uploaded."""
     temp_file = tmp_path / str(uuid.uuid4())
-    temp_file.write_bytes(requests.get(RPM_WITH_NON_UTF_8_URL).content)
+    temp_file.write_bytes(fetch_url(RPM_WITH_NON_UTF_8_URL))
     artifact = pulpcore_bindings.ArtifactsApi.create(str(temp_file))
     with pytest.raises(PulpTaskError) as ctx:
         response = rpm_package_api.create(

pulp_rpm/tests/functional/api/test_checkpoint.py

@@ -1,39 +1,43 @@
 """Tests for checkpoint distribution and publications."""
 
 import re
+import uuid
 from datetime import datetime, timedelta
-from tempfile import NamedTemporaryFile
 from time import sleep
 from urllib.parse import urlparse
 
 import pytest
-import requests
 from aiohttp import ClientResponseError
 
-from pulp_rpm.tests.functional.constants import RPM_SIGNED_URL
+from pulp_rpm.tests.functional.utils import Nevra, build_rpm, fetch_url
 
 
 @pytest.fixture(scope="class")
 def rpm_package_factory_class(
     gen_object_with_cleanup,
     pulp_domain_enabled,
     rpm_package_api,
+    tmp_path_factory,
 ):
     """Return a Package created from uploading an RPM file."""
 
-    def _rpm_package_factory_class(url=RPM_SIGNED_URL, pulp_domain=None):
-        with NamedTemporaryFile() as file_to_upload:
-            file_to_upload.write(requests.get(url).content)
-            file_to_upload.flush()
-            upload_attrs = {"file": file_to_upload.name}
-
-            kwargs = {}
-            if pulp_domain:
-                if not pulp_domain_enabled:
-                    raise RuntimeError("Server does not have domains enabled.")
-                kwargs["pulp_domain"] = pulp_domain
-
-            return gen_object_with_cleanup(rpm_package_api, **upload_attrs, **kwargs)
+    def _rpm_package_factory_class(url=None, pulp_domain=None):
+        tmp_path = tmp_path_factory.mktemp("rpm_pkg")
+        uid = uuid.uuid4().hex[:8]
+        rpm_file = tmp_path / f"test-pkg-{uid}-1.0-1.noarch.rpm"
+        if url is not None:
+            rpm_file.write_bytes(fetch_url(url))
+        else:
+            build_rpm(Nevra(f"test-pkg-{uid}", 0, "1.0", "1", "noarch"), rpm_file)
+        upload_attrs = {"file": str(rpm_file)}
+
+        kwargs = {}
+        if pulp_domain:
+            if not pulp_domain_enabled:
+                raise RuntimeError("Server does not have domains enabled.")
+            kwargs["pulp_domain"] = pulp_domain
+
+        return gen_object_with_cleanup(rpm_package_api, **upload_attrs, **kwargs)
 
     return _rpm_package_factory_class
 

pulp_rpm/tests/functional/api/test_crud_content_unit.py

@@ -16,16 +16,14 @@
     RPM_MODULAR_MODULES_CONTENT_NAME,
     RPM_MODULES_OBSOLETE_CONTENT_NAME,
     RPM_PACKAGE_CONTENT_NAME,
-    RPM_PACKAGE_FILENAME,
-    RPM_PACKAGE_FILENAME2,
     RPM_REPO_METADATA_FIXTURE_URL,
 )
 from pulp_rpm.tests.functional.utils import gen_rpm_content_attrs
 
 
+@pytest.mark.parallel
 def test_crud_content_unit(
-    delete_orphans_pre,
-    signed_artifact,
+    rpm_artifact_factory,
     gen_object_with_cleanup,
     rpm_package_api,
     rpm_repository_api,
@@ -35,8 +33,10 @@ def test_crud_content_unit(
 ):
     """Test creating, reading, updating, and deleting a content unit of package type."""
     # Create content unit
+    artifact = rpm_artifact_factory()
+    relative_path = "test-crud-content-unit-1.0-1.noarch.rpm"
 
-    attrs = gen_rpm_content_attrs(signed_artifact, RPM_PACKAGE_FILENAME)
+    attrs = gen_rpm_content_attrs(artifact, relative_path)
     response = rpm_package_api.create(**attrs)
     content_unit = rpm_package_api.read(monitor_task(response.task).created_resources[0])
     # rpm package doesn't keep relative_path but the location href
@@ -55,14 +55,14 @@ def test_crud_content_unit(
     assert page.results[0] == content_unit
 
     # Attempt to update a content unit using HTTP PATCH
-    attrs = gen_rpm_content_attrs(signed_artifact, RPM_PACKAGE_FILENAME2)
+    attrs = gen_rpm_content_attrs(artifact, relative_path)
     with pytest.raises(AttributeError) as exc:
         rpm_package_api.partial_update(content_unit.pulp_href, attrs)
     msg = "object has no attribute 'partial_update'"
     assert msg in str(exc)
 
     # Attempt to update a content unit using HTTP PUT
-    attrs = gen_rpm_content_attrs(signed_artifact, RPM_PACKAGE_FILENAME2)
+    attrs = gen_rpm_content_attrs(artifact, relative_path)
     with pytest.raises(AttributeError) as exc:
         rpm_package_api.update(content_unit.pulp_href, attrs)
     msg = "object has no attribute 'update'"
@@ -75,14 +75,14 @@ def test_crud_content_unit(
     assert msg in str(exc)
 
     # Attempt to create duplicate package without specifying a repository
-    attrs = gen_rpm_content_attrs(signed_artifact, RPM_PACKAGE_FILENAME)
+    attrs = gen_rpm_content_attrs(artifact, relative_path)
     response = rpm_package_api.create(**attrs)
     duplicate = rpm_package_api.read(monitor_task(response.task).created_resources[0])
     assert duplicate.pulp_href == content_unit.pulp_href
 
     # Attempt to create duplicate package while specifying a repository
     repo = rpm_repository_factory()
-    attrs = gen_rpm_content_attrs(signed_artifact, RPM_PACKAGE_FILENAME)
+    attrs = gen_rpm_content_attrs(artifact, relative_path)
     attrs["repository"] = repo.pulp_href
     response = rpm_package_api.create(**attrs)
     monitored_response = monitor_task(response.task)

pulp_rpm/tests/functional/api/test_domains.py

@@ -11,6 +11,7 @@
     RPM_SIGNED_FIXTURE_URL,
 )
 from pulp_rpm.tests.functional.utils import (
+    Nevra,
     get_package_repo_path,
 )
 
@@ -181,8 +182,12 @@ def test_artifact_from_file(
     pulpcore_bindings,
     gen_object_with_cleanup,
     rpm_artifact_factory,
+    rpm_create_package,
 ):
     """Test uploading artifacts in separate domains."""
+    # Build one RPM and upload the same file into each domain.
+    rpm_path = rpm_create_package(Nevra(f"package-{uuid.uuid4().hex[:8]}", 0, "1.0", "1", "noarch"))
+
     body = {
         "name": str(uuid.uuid4()),
         "storage_class": "pulpcore.app.models.storage.FileSystem",
@@ -198,13 +203,13 @@ def test_artifact_from_file(
     domain2 = gen_object_with_cleanup(pulpcore_bindings.DomainsApi, body)
 
     # Create as-artifact in domain1
-    domain1_artifact = rpm_artifact_factory(pulp_domain=domain1.name)
+    domain1_artifact = rpm_artifact_factory(path=rpm_path, pulp_domain=domain1.name)
     artifacts = pulpcore_bindings.ArtifactsApi.list(pulp_domain=domain1.name)
     assert artifacts.count == 1
     assert domain1_artifact.pulp_href == artifacts.results[0].pulp_href
 
     # Create as-artifact in domain2
-    domain2_artifact = rpm_artifact_factory(pulp_domain=domain2.name)
+    domain2_artifact = rpm_artifact_factory(path=rpm_path, pulp_domain=domain2.name)
     artifacts = pulpcore_bindings.ArtifactsApi.list(pulp_domain=domain2.name)
     assert artifacts.count == 1
     assert domain2_artifact.pulp_href == artifacts.results[0].pulp_href
@@ -215,7 +220,7 @@ def test_artifact_from_file(
 
     # Show that duplicate artifact can not be uploaded in same domain
     with pytest.raises(CoreApiException) as e:
-        rpm_artifact_factory(pulp_domain=domain1.name)
+        rpm_artifact_factory(path=rpm_path, pulp_domain=domain1.name)
     assert e.value.status == 400
     assert json.loads(e.value.body) == {
         "non_field_errors": [
@@ -229,6 +234,7 @@ def test_rpm_from_file(
     cleanup_domains,
     pulpcore_bindings,
     rpm_package_factory,
+    rpm_create_package,
     gen_object_with_cleanup,
     rpm_package_api,
 ):
@@ -240,9 +246,12 @@ def test_rpm_from_file(
     }
     domain = gen_object_with_cleanup(pulpcore_bindings.DomainsApi, body)
 
+    # Build one RPM and upload the same file into the default and custom domains.
+    rpm_path = rpm_create_package(Nevra(f"content-{uuid.uuid4().hex[:8]}", 0, "1.0", "1", "noarch"))
+
     try:
-        default_content = rpm_package_factory()
-        domain_content = rpm_package_factory(pulp_domain=domain.name)
+        default_content = rpm_package_factory(path=rpm_path)
+        domain_content = rpm_package_factory(path=rpm_path, pulp_domain=domain.name)
         assert default_content.pulp_href != domain_content.pulp_href
         assert default_content.sha256 == domain_content.sha256
 

pulp_rpm/tests/functional/api/test_download_content.py

@@ -5,12 +5,12 @@
 from urllib.parse import urljoin
 
 import pytest
-import requests
 
 from pulpcore.client.pulp_rpm import RpmRpmPublication
 
 from pulp_rpm.tests.functional.constants import RPM_UNSIGNED_FIXTURE_URL
 from pulp_rpm.tests.functional.utils import (
+    fetch_url,
     get_package_repo_path,
 )
 
@@ -59,7 +59,7 @@ def test_all(
     package_paths = [p.location_href for p in packages.results]
     unit_path = choice(package_paths)
     fixture_hash = hashlib.sha256(
-        requests.get(urljoin(RPM_UNSIGNED_FIXTURE_URL, unit_path)).content
+        fetch_url(urljoin(RPM_UNSIGNED_FIXTURE_URL, unit_path))
     ).hexdigest()
 
     # …and Pulp.

pulp_rpm/tests/functional/api/test_metadata_signing.py

@@ -0,0 +1,89 @@
+import pysequoia
+import pytest
+import requests
+
+from pulp_rpm.tests.functional.conftest import (
+    create_signing_service,
+    import_signing_key,
+    make_signing_script,
+    remove_signing_service,
+)
+from pulp_rpm.tests.functional.constants import (
+    KEY_V4_RSA4K,
+    KEY_V6_MLDSA65_ED25519,
+    RPM_UNSIGNED_FIXTURE_URL,
+)
+
+
+@pytest.fixture
+def metadata_signing_service(request, tmp_path, pulpcore_bindings):
+    """Create a metadata signing service for the given (key, backend) pair."""
+    key, backend = request.param
+    home = tmp_path / "signing"
+    home.mkdir(mode=0o700)
+
+    _, fingerprint, _ = import_signing_key(key.private_url, home, backend=backend)
+    script_path = make_signing_script(home, fingerprint, tmp_path, backend=backend)
+    service_name = create_signing_service(home, fingerprint, script_path, backend=backend)
+
+    service = pulpcore_bindings.SigningServicesApi.list(name=service_name).results[0]
+    yield service, key
+    remove_signing_service(service_name)
+
+
+@pytest.mark.parallel
+@pytest.mark.parametrize(
+    "metadata_signing_service",
+    [
+        (KEY_V4_RSA4K, "gpg"),
+        (KEY_V6_MLDSA65_ED25519, "sq"),
+    ],
+    indirect=True,
+)
+def test_publish_signed_repo_metadata(
+    metadata_signing_service,
+    rpm_repository_factory,
+    init_and_sync,
+    rpm_publication_factory,
+    rpm_distribution_factory,
+    distribution_base_url,
+):
+    """Verify that publishing with a metadata signing service produces a signed repomd.xml.
+
+    After syncing and publishing with a metadata signing service attached, the
+    distribution should serve `repomd.xml.asc` (detached signature) and
+    `repomd.xml.key` (public key) alongside `repomd.xml`, and the
+    detached signature should be verifiable with the published public key.
+    """
+    service, _key = metadata_signing_service
+
+    repo = rpm_repository_factory(metadata_signing_service=service.pulp_href)
+    repo, _ = init_and_sync(
+        repository=repo,
+        url=RPM_UNSIGNED_FIXTURE_URL,
+        policy="on_demand",
+    )
+
+    publication = rpm_publication_factory(repository=repo.pulp_href)
+    distribution = rpm_distribution_factory(publication=publication.pulp_href)
+    base_url = distribution_base_url(distribution.base_url)
+
+    repomd_resp = requests.get(f"{base_url}/repodata/repomd.xml")
+    assert repomd_resp.status_code == 200
+
+    asc_resp = requests.get(f"{base_url}/repodata/repomd.xml.asc")
+    assert asc_resp.status_code == 200
+    assert len(asc_resp.content) > 0, "repomd.xml.asc is empty"
+
+    key_resp = requests.get(f"{base_url}/repodata/repomd.xml.key")
+    assert key_resp.status_code == 200
+    assert len(key_resp.content) > 0, "repomd.xml.key is empty"
+
+    # Verify the detached signature using pysequoia
+    cert = pysequoia.Cert.from_bytes(key_resp.content)
+    sig = pysequoia.Sig.from_bytes(asc_resp.content)
+    pysequoia.verify(
+        bytes=repomd_resp.content,
+        signature=sig,
+        store=lambda _key_ids: [cert],
+    )