Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 1 addition & 2 deletions functest_requirements.txt
Original file line number Diff line number Diff line change
@@ -1,6 +1,5 @@
pytest<10
python-gnupg
pytest-xdist
pytest-timeout
pytest-custom_exit_code
trustme~=1.2.1
trustme~=1.2.1
66 changes: 44 additions & 22 deletions pulp_container/tests/functional/api/test_push_signatures.py
Original file line number Diff line number Diff line change
@@ -1,55 +1,78 @@
"""Tests that verify that an image signature can be pushed to Pulp."""

import base64
import json
import subprocess

import pytest

from pulp_container.constants import SIGNATURE_TYPE
from pulp_container.tests.functional.conftest import verify_inline_signature
from pulp_container.tests.functional.constants import REGISTRY_V2_REPO_PULP


def _podman_supports_sq_signing():
"""Return True if the local podman build supports --sign-by-sq-fingerprint."""
result = subprocess.run(
("podman", "push", "--help"),
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
)
return "--sign-by-sq-fingerprint" in result.stdout.decode()


@pytest.fixture
def distribution(
def signed_distribution(
signing_key_home,
registry_client,
local_registry,
container_distribution_api,
signing_gpg_metadata,
container_namespace_api,
add_to_cleanup,
full_path,
):
"""Return a distribution created after pushing a signed content to the Pulp Registry."""
"""Push an image signed with a Sequoia key (twice, for two distinct signatures).

Parameterized (via `signing_key_home`) over a key that signs with its primary key and
one that signs with a dedicated subkey.
"""
if registry_client.name != "podman":
pytest.skip("This test requires podman to sign pulled content", allow_module_level=True)
if not _podman_supports_sq_signing():
pytest.skip("This podman build does not support --sign-by-sq-fingerprint")

image_path = f"{REGISTRY_V2_REPO_PULP}:manifest_a"
registry_client.pull(image_path)

gpg, fingerprint, keyid = signing_gpg_metadata

with registry_client.set_env(GNUPGHOME=str(gpg.gnupghome)):
local_registry.tag_and_push(image_path, full_path("test-1:manifest_a"), "--sign-by", keyid)

# push the same image for the second time with a different signature (timestamp)
local_registry.tag_and_push(image_path, full_path("test-1:manifest_a"), "--sign-by", keyid)
# Point the Sequoia integration at the home holding our imported key.
sign_args = ("--sign-by-sq-fingerprint", signing_key_home.fingerprint)
with registry_client.set_env(SEQUOIA_HOME=str(signing_key_home.home)):
local_registry.tag_and_push(image_path, full_path("test-1:manifest_a"), *sign_args)
# push a second time to produce a distinct signature (timestamp)
local_registry.tag_and_push(image_path, full_path("test-1:manifest_a"), *sign_args)

distribution = container_distribution_api.list(name="test-1").results[0]
add_to_cleanup(container_distribution_api, distribution.pulp_href)
# Clean up the namespace, which cascades to the distribution and the push repository.
add_to_cleanup(container_namespace_api, distribution.namespace)

return distribution


def test_assert_signed_image(
signing_key_home,
local_registry,
container_repository_api,
container_manifest_api,
container_signature_api,
signing_gpg_metadata,
distribution,
signed_distribution,
full_path,
):
"""Test whether an admin user can fetch a signature from the Pulp Registry."""
gpg, fingerprint, keyid = signing_gpg_metadata
"""Test whether an admin user can fetch a signature from the Pulp Registry.

Runs against both primary-key and subkey signing.
"""
distribution = signed_distribution
fingerprint = signing_key_home.signing_fingerprint
keyid = signing_key_home.signing_keyid

repository = container_repository_api.read(distribution.repository)
manifest = container_manifest_api.list(
Expand All @@ -75,13 +98,12 @@ def test_assert_signed_image(
timestamps = []
for s in signatures:
raw_s = base64.b64decode(s["content"])
decrypted = gpg.decrypt(raw_s)

assert decrypted.key_id == keyid
assert decrypted.fingerprint == fingerprint
assert decrypted.status == "signature valid"
sig_fingerprint, sig_key_id, json_s = verify_inline_signature(
signing_key_home.public_key, raw_s
)

json_s = json.loads(decrypted.data)
assert sig_key_id == keyid
assert sig_fingerprint == fingerprint

image_path = json_s["critical"]["identity"]["docker-reference"]
assert image_path == f"{local_registry.name}/{full_path(distribution)}:manifest_a"
Expand Down
86 changes: 77 additions & 9 deletions pulp_container/tests/functional/api/test_sign_manifests.py
Original file line number Diff line number Diff line change
@@ -1,50 +1,118 @@
import pytest

from pulpcore.pytest_plugin import create_signing_service, remove_signing_service

from pulp_container.constants import SIGNATURE_TYPE
from pulp_container.tests.functional.constants import REGISTRY_V2_REPO_PULP

MANIFEST_TAG = "manifest_a"

# Builds an atomic container signature payload for the passed manifest and signs
# it with a Sequoia (sq) key, emitting an inline-signed binary OpenPGP message.
# See https://github.com/pulp/pulp_container/issues/2280.
SIGNING_SCRIPT_STRING = """#!/usr/bin/env bash

set -e

MANIFEST_PATH=$1
FINGERPRINT="$PULP_SIGNING_KEY_FINGERPRINT"
SQ_HOME="{sq_home}"

DIGEST="sha256:$(sha256sum "$MANIFEST_PATH" | awk '{{print $1}}')"

PAYLOAD_FILE="$(mktemp)"
cat > "$PAYLOAD_FILE" <<EOF
{{"critical": {{"type": "atomic container signature", \
"image": {{"docker-manifest-digest": "$DIGEST"}}, \
"identity": {{"docker-reference": "$REFERENCE"}}}}, \
"optional": {{"creator": "pulp sq test"}}}}
EOF

sq --home "$SQ_HOME" sign --signer "$FINGERPRINT" --message --binary \
--output "$SIG_PATH" "$PAYLOAD_FILE"

echo "{{\\"signature_path\\": \\"$SIG_PATH\\"}}"
"""


@pytest.fixture
def distribution(
registry_client, local_registry, container_distribution_api, full_path, add_to_cleanup
registry_client,
local_registry,
container_distribution_api,
container_namespace_api,
full_path,
add_to_cleanup,
):
"""The fixture for a distribution created by pushing an image to the registry."""
image_path = f"{REGISTRY_V2_REPO_PULP}:{MANIFEST_TAG}"
registry_client.pull(image_path)
local_registry.tag_and_push(image_path, full_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)
# Clean up the namespace, which cascades to the distribution and the push repository.
add_to_cleanup(container_namespace_api, distribution.namespace)

return distribution


@pytest.fixture
def manifest_signing_service(signing_key_home, tmp_path, pulpcore_bindings):
"""Register a ManifestSigningService backed by a Sequoia key.

Parameterized (via `signing_key_home`) over a key that signs with its primary key and
one that signs with a dedicated subkey.
"""
script_path = tmp_path / "sign-manifest.sh"
script_path.write_text(SIGNING_SCRIPT_STRING.format(sq_home=signing_key_home.home))
script_path.chmod(0o755)

try:
service_name = create_signing_service(
signing_key_home.home,
signing_key_home.fingerprint,
script_path,
backend="sq",
service_class="container:ManifestSigningService",
)
except TypeError:
pytest.skip("This pulpcore release does not support the Sequoia (sq) signing backend")

service = pulpcore_bindings.SigningServicesApi.list(name=service_name).results[0]
assert service.pubkey_fingerprint == signing_key_home.fingerprint

yield service, signing_key_home.signing_fingerprint, signing_key_home.signing_keyid

remove_signing_service(service_name, service_class="container:ManifestSigningService")


def test_sign_manifest(
signing_gpg_metadata,
manifest_signing_service,
distribution,
container_signing_service,
container_repository_api,
container_signature_api,
container_tag_api,
container_manifest_api,
monitor_task,
):
"""Test whether a user can sign a manifest by leveraging a signing service."""
_, fingerprint, keyid = signing_gpg_metadata
sign_data = {"manifest_signing_service": container_signing_service.pulp_href}
"""Test whether a user can sign a manifest by leveraging a signing service.

Runs against both primary-key and subkey signing.
"""
service, fingerprint, keyid = manifest_signing_service
sign_data = {"manifest_signing_service": service.pulp_href}

response = container_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])
repository_version = created_resources[0]
tags = container_tag_api.list(repository_version=repository_version)
assert tags.count == 1

tag = tags.results[0]
assert tag.name == MANIFEST_TAG

signatures = container_signature_api.list()
signatures = container_signature_api.list(repository_version=repository_version)
assert signatures.count == 1

signature = signatures.results[0]
Expand Down
103 changes: 103 additions & 0 deletions pulp_container/tests/functional/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,12 +3,18 @@
import stat
import subprocess
from contextlib import contextmanager, suppress
from types import SimpleNamespace
from urllib.parse import urljoin, urlparse
from uuid import uuid4

import pytest
import requests

from pulpcore.pytest_plugin import (
KEY_V6_MLDSA65_ED25519_PRIVATE,
KEY_V6_MLDSA65_ED25519_PUBLIC,
import_signing_key,
)
from pulpcore.tests.functional.utils import BindingsNamespace

from pulp_container.tests.functional.constants import PULP_HELLO_WORLD_REPO, REGISTRY_V2_FEED_URL
Expand All @@ -17,6 +23,103 @@
BearerTokenAuth,
)

# A GPG-generated fixture key. Unlike the Sequoia-generated ML-DSA key (which signs with a
# dedicated subkey), this key signs with its primary key. Whether a certificate signs with its
# primary key or a subkey is a matter of key configuration, not tooling -- pulp_container must
# record the actual signing (sub)key in both cases, which the parameterization below verifies.
_GPG_FIXTURE_KEY_PRIVATE = (
"https://raw.githubusercontent.com/pulp/pulp-fixtures/master/common/"
"GPG-PRIVATE-KEY-fixture-signing"
)
_GPG_FIXTURE_KEY_PUBLIC = (
"https://raw.githubusercontent.com/pulp/pulp-fixtures/master/common/GPG-KEY-fixture-signing"
)


@pytest.fixture(scope="session", params=["primary", "subkey"])
def signing_key_home(request, tmp_path_factory):
"""Import a signing key into a Sequoia home, exercising primary- and subkey-signing.

Yields a namespace with the Sequoia `home`, the certificate `fingerprint` (primary), the
actual `signing_fingerprint`/`signing_keyid` used when signing, and the ascii-armored
`public_key`.
"""
keys = {
# GPG-generated key -> signs with its primary key.
"primary": (_GPG_FIXTURE_KEY_PRIVATE, _GPG_FIXTURE_KEY_PUBLIC),
# Sequoia ML-DSA (post-quantum) key -> signs with a dedicated subkey.
"subkey": (KEY_V6_MLDSA65_ED25519_PRIVATE, KEY_V6_MLDSA65_ED25519_PUBLIC),
}
private_url, public_url = keys[request.param]

home = tmp_path_factory.mktemp(f"sq_home_{request.param}")
try:
_sq, fingerprint, _keyid = import_signing_key(private_url, home, backend="sq")
except TypeError:
pytest.skip("This pulpcore release does not support the Sequoia (sq) signing backend")

public_key = requests.get(public_url).content.decode("utf-8")
signing_fingerprint, signing_keyid = sq_signing_identity(home, fingerprint, public_key)

return SimpleNamespace(
home=home,
fingerprint=fingerprint,
signing_fingerprint=signing_fingerprint,
signing_keyid=signing_keyid,
public_key=public_key,
)


def _keyid_from_fingerprint(fingerprint):
"""Derive the key ID from an OpenPGP fingerprint, matching pulp_container's logic.

For v4 fingerprints (40 hex chars) the key ID is the last 16 chars; for v6 (64 hex
chars) it is the first 16 chars.
"""
if len(fingerprint) == 40:
return fingerprint[-16:]
elif len(fingerprint) == 64:
return fingerprint[:16]
raise ValueError(f"Unexpected fingerprint length: {len(fingerprint)}")


def _verified_signing_key(public_key, raw_signature):
"""Verify an inline-signed OpenPGP message and return (signing_key_fpr, payload_bytes).

Works for both classic (RSA/ed25519) and post-quantum (ML-DSA) keys. Uses pysequoia
directly rather than pulpcore's gpg_verify because the latter pulls in Django models,
which aren't configured in the functional-test client process.
"""
from pysequoia import Cert, verify

certs = Cert.split_bytes(public_key.encode("utf-8"))
result = verify(bytes=raw_signature, store=lambda key_ids: certs)
valid_sig = result.valid_sigs[0]
return valid_sig.signing_key.upper(), bytes(result.bytes)


def verify_inline_signature(public_key, raw_signature):
"""Verify a signature blob and return (fingerprint, key_id, payload_dict)."""
fingerprint, payload_bytes = _verified_signing_key(public_key, raw_signature)
return fingerprint, _keyid_from_fingerprint(fingerprint), json.loads(payload_bytes)


def sq_signing_identity(sq_home, signer, public_key):
"""Return the (fingerprint, key_id) actually used when signing with `signer`.

A certificate may sign with a dedicated subkey rather than its primary key, so the
fingerprint recorded on produced signatures can differ from the certificate's primary
fingerprint. This signs throwaway data to discover the real signing (sub)key.
"""
completed = subprocess.run(
("sq", "--home", str(sq_home), "sign", "--signer", signer, "--message", "--binary"),
input=b"probe",
capture_output=True,
)
completed.check_returncode()
fingerprint, _payload = _verified_signing_key(public_key, completed.stdout)
return fingerprint, _keyid_from_fingerprint(fingerprint)


def gen_container_remote(url=REGISTRY_V2_FEED_URL, **kwargs):
"""Return a semi-random dict for use in creating a container Remote.
Expand Down
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ dependencies = [
"jsonschema>=4.4,<4.27",
"pulpcore>=3.111.0,<3.130",
"pyjwt[crypto]>=2.4,<2.14",
"pysequoia>=0.1.33,<0.2.0"
"pysequoia>=0.1.35,<0.2.0"
]

[project.urls]
Expand Down
Loading