Skip to content
Merged
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
1 change: 1 addition & 0 deletions CHANGES/7479.feature
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Added `--backend` option to the `add-signing-service` management command, enabling Sequoia (`sq`) as an alternative to GPG for key management. Use `--backend sq` to register signing services using Sequoia's key store.
123 changes: 100 additions & 23 deletions pulpcore/app/management/commands/add-signing-service.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import os
import subprocess
import warnings
from gettext import gettext as _
from pathlib import Path

Expand All @@ -9,6 +10,11 @@

from pulpcore.app.models.content import SigningService as BaseSigningService

ENV_DEFAULTS = {
"gpg": "GNUPGHOME",
"sq": "SEQUOIA_HOME",
}


class Command(BaseCommand):
"""
Expand All @@ -28,19 +34,35 @@ def add_arguments(self, parser):
)
parser.add_argument(
"key",
help=_("Key id of the public key."),
help=_("Key id or fingerprint of the public key."),
)
parser.add_argument(
"--class",
default="core:AsciiArmoredDetachedSigningService",
required=False,
help=_("Signing service class prefixed by the app label separated by a colon."),
)
parser.add_argument(
"--backend",
choices=["gpg", "sq"],
default="gpg",
required=False,
help=_("Key management backend to use for extracting key metadata. (default: gpg)"),
)
parser.add_argument(
"--home",
default=None,
required=False,
help=_(
"Home directory for the key management backend. "
"Defaults to $GNUPGHOME (gpg) or $SEQUOIA_HOME (sq)."
),
)
parser.add_argument(
"--gnupghome",
default=os.getenv("GNUPGHOME", ""),
default=None,
required=False,
help=_("A default GnuPG home directory to use during the initialization."),
help=_("Deprecated: use --home instead."),
)
parser.add_argument(
"--keyring",
Expand Down Expand Up @@ -68,11 +90,56 @@ def handle(self, *args, **options):
)
)

gpg_cmd = ["gpg"]
backend = options["backend"]

if options["home"] and options["gnupghome"]:
raise CommandError(_("--home and --gnupghome are mutually exclusive."))

if options["gnupghome"]:
gpg_cmd += ["--homedir", options["gnupghome"]]
if options["keyring"]:
gpg_cmd += ["--keyring", options["keyring"]]
warnings.warn(
"--gnupghome is deprecated; use --home instead.",
DeprecationWarning,
stacklevel=2,
)

home = options["home"] or options["gnupghome"] or os.getenv(ENV_DEFAULTS[backend], "")

if backend == "sq":
fingerprint, public_key = self._extract_key_from_sq(
key_id, home, options.get("keyring")
)
else:
fingerprint, public_key = self._extract_key_from_gpg(
key_id, home, options.get("keyring")
)

try:
script_path = Path(script).resolve(strict=True)
except FileNotFoundError as e:
raise CommandError(str(e))

try:
SigningService.objects.create(
name=name,
public_key=public_key,
pubkey_fingerprint=fingerprint,
script=script_path,
)
except IntegrityError as e:
raise CommandError(str(e))

print(
("Successfully added signing service {name} for key {fingerprint}.").format(
name=name, fingerprint=fingerprint
)
)

def _extract_key_from_gpg(self, key_id, home, keyring):
gpg_cmd = ["gpg"]
if home:
gpg_cmd += ["--homedir", home]
if keyring:
gpg_cmd += ["--keyring", keyring]

result = subprocess.run(
gpg_cmd + ["--with-colons", "--fingerprint", key_id],
Expand Down Expand Up @@ -103,23 +170,33 @@ def handle(self, *args, **options):
raise CommandError(result.stderr.strip())
public_key = result.stdout

try:
script_path = Path(script).resolve(strict=True)
except FileNotFoundError as e:
raise CommandError(str(e))
return fingerprint, public_key

def _extract_key_from_sq(self, key_id, home, keyring):
from pysequoia import Cert

sq_cmd = ["sq"]
if home:
sq_cmd += ["--home", home]
if keyring:
sq_cmd += ["--keyring", keyring]

result = subprocess.run(
sq_cmd + ["cert", "export", "--cert", key_id],
capture_output=True,
text=True,
)
if result.returncode != 0:
raise CommandError(result.stderr.strip())
public_key = result.stdout

try:
SigningService.objects.create(
name=name,
public_key=public_key,
pubkey_fingerprint=fingerprint,
script=script_path,
cert = Cert.from_bytes(public_key.encode("utf-8"))
except Exception as e:
raise CommandError(
_("Failed to parse exported certificate for '{}': {}").format(key_id, e)
)
except IntegrityError as e:
raise CommandError(str(e))

print(
("Successfully added signing service {name} for key {fingerprint}.").format(
name=name, fingerprint=fingerprint
)
)
fingerprint = cert.fingerprint.upper()

return fingerprint, public_key
152 changes: 127 additions & 25 deletions pulpcore/pytest_plugin.py
Original file line number Diff line number Diff line change
Expand Up @@ -1116,6 +1116,27 @@ def _dispatch_task_group(task_name, *args, **kwargs):
fi
"""

SQ_SIGNING_SCRIPT_STRING = """#!/usr/bin/env bash

FILE_PATH=$1
SIGNATURE_PATH="$1.asc"

SQ_HOME="{sq_home}"
SIGNER="{signer_fingerprint}"

# Create a detached signature using Sequoia (sq)
sq --home "${{SQ_HOME}}" sign --signer "${{SIGNER}}" \\
--signature-file="${{SIGNATURE_PATH}}" "${{FILE_PATH}}"

# Check the exit status
STATUS=$?
if [[ ${{STATUS}} -eq 0 ]]; then
echo '{{"file": "'${{FILE_PATH}}'", "signature": "'${{SIGNATURE_PATH}}'"}}'
else
exit ${{STATUS}}
fi
"""


@pytest.fixture(scope="session")
def signing_script_path(signing_script_temp_dir, signing_gpg_homedir_path, signing_gpg_metadata):
Expand Down Expand Up @@ -1207,63 +1228,144 @@ def ascii_armored_detached_signing_service(
).results[0]


def import_signing_key(key_url, gpg_home):
"""Import a PGP key into a GPG home directory and trust it.
@pytest.fixture(scope="session")
def sq_signing_home_path(tmp_path_factory):
return tmp_path_factory.mktemp("sq_home")

Returns ``(gpg, fingerprint, keyid)``.
"""
try:
import gnupg
except ImportError:
pytest.skip("python-gnupg not installed")

gpg = gnupg.GPG(gnupghome=gpg_home)
@pytest.fixture(scope="session")
def sq_signing_script_temp_dir(tmp_path_factory):
return tmp_path_factory.mktemp("sq_signing_script_dir")


@pytest.fixture(scope="session")
def sq_signing_metadata(sq_signing_home_path):
"""A fixture that returns Sequoia signing metadata (i.e., fingerprint, keyid)."""
return import_signing_key(KEY_V6_ED25519_PRIVATE, sq_signing_home_path, backend="sq")

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is a non-PQC, OpenPGPv6 key, so it should work with older versions of sequoia but still test something which GPG cannot do.



@pytest.fixture(scope="session")
def sq_signing_script_path(sq_signing_script_temp_dir, sq_signing_home_path, sq_signing_metadata):
_sq, fingerprint, _keyid = sq_signing_metadata
return make_signing_script(
sq_signing_home_path, fingerprint, sq_signing_script_temp_dir, backend="sq"
)


@pytest.fixture(scope="session")
def _sq_ascii_armored_detached_signing_service_name(
sq_signing_script_path,
sq_signing_metadata,
sq_signing_home_path,
):
_sq, fingerprint, _keyid = sq_signing_metadata
service_name = create_signing_service(
sq_signing_home_path, fingerprint, sq_signing_script_path, backend="sq"
)

yield service_name

remove_signing_service(service_name)


@pytest.fixture(scope="session")
def sq_ascii_armored_detached_signing_service(
_sq_ascii_armored_detached_signing_service_name, pulpcore_bindings
):
return pulpcore_bindings.SigningServicesApi.list(
name=_sq_ascii_armored_detached_signing_service_name
).results[0]


def import_signing_key(key_url, home, *, backend="gpg"):
"""Import a PGP key into a keyring and return metadata.

Returns `(gpg_instance_or_none, fingerprint, keyid)`. The first element
is a `gnupg.GPG` instance when `backend` is `"gpg"`, or `None` when
`backend` is `"sq"`.
"""
response = requests.get(key_url)
response.raise_for_status()
result = gpg.import_keys(response.content)
assert result.count >= 1, f"Failed to import key from {key_url}"

key_info = gpg.list_keys()[0]
fingerprint = key_info["fingerprint"]
keyid = key_info["keyid"]
gpg.trust_keys(fingerprint, "TRUST_ULTIMATE")
if backend == "sq":
from pysequoia import Cert

completed = subprocess.run(
("sq", "--home", str(home), "key", "import"),
input=response.content,
capture_output=True,
)
assert completed.returncode == 0, completed.stderr.decode()

cert = Cert.from_bytes(response.content)
fingerprint = cert.fingerprint.upper()
keyid = fingerprint[-16:]

return None, fingerprint, keyid
else:
try:
import gnupg
except ImportError:
pytest.skip("python-gnupg not installed")

gpg = gnupg.GPG(gnupghome=home)

result = gpg.import_keys(response.content)
assert result.count >= 1, f"Failed to import key from {key_url}"

return gpg, fingerprint, keyid
key_info = gpg.list_keys()[0]
fingerprint = key_info["fingerprint"]
keyid = key_info["keyid"]
gpg.trust_keys(fingerprint, "TRUST_ULTIMATE")

return gpg, fingerprint, keyid

def make_signing_script(gpg_home, fingerprint, script_dir=None):

def make_signing_script(home, fingerprint, script_dir=None, *, backend="gpg"):
"""Create a detached-signature signing script.

Returns the script path.
"""
if script_dir is None:
script_dir = gpg_home
script_path = script_dir / "sign.sh"
script_path.write_text(SIGNING_SCRIPT_STRING.format(gpg_home=gpg_home, gpg_key_id=fingerprint))
script_dir = home
if backend == "sq":
script_path = script_dir / "sq_sign.sh"
script_path.write_text(
SQ_SIGNING_SCRIPT_STRING.format(sq_home=home, signer_fingerprint=fingerprint)
)
else:
script_path = script_dir / "sign.sh"
script_path.write_text(SIGNING_SCRIPT_STRING.format(gpg_home=home, gpg_key_id=fingerprint))
script_path.chmod(0o755)
return script_path


def create_signing_service(
gpg_home, fingerprint, script_path, *, service_class="core:AsciiArmoredDetachedSigningService"
home,
fingerprint,
script_path,
*,
backend="gpg",
service_class="core:AsciiArmoredDetachedSigningService",
):
"""Register a signing service via pulpcore-manager.

Returns the service name.
"""
service_name = str(uuid.uuid4())
cmd = (
cmd = [
"pulpcore-manager",
"add-signing-service",
service_name,
str(script_path),
fingerprint,
"--class",
service_class,
"--gnupghome",
str(gpg_home),
)
"--backend",
backend,
"--home",
str(home),
]
completed = subprocess.run(cmd, capture_output=True, text=True)
assert completed.returncode == 0, completed.stderr

Expand Down
Loading
Loading