libs: Add envoy.gpg.sign (#33)

Signed-off-by: Ryan Northey <ryan@synca.io>

.github/workflows/ci.yml

@@ -21,6 +21,7 @@ jobs:
         - envoy.github.abstract
         - envoy.github.release
         - envoy.gpg.identity
+        - envoy.gpg.sign
         - mypy-abstracts
         - pytest-patches
         python-version:
@@ -100,6 +101,7 @@ jobs:
         - envoy.github.abstract
         - envoy.github.release
         - envoy.gpg.identity
+        - envoy.gpg.sign
         - mypy-abstracts
         - pytest-patches
     steps:

envoy.gpg.sign/README.rst

@@ -0,0 +1,5 @@
+
+envoy.gpg.sign
+==============
+
+GPG signing util used in Envoy proxy's CI

envoy.gpg.sign/VERSION

@@ -0,0 +1 @@
+0.0.1

envoy.gpg.sign/envoy/gpg/sign/__init__.py

@@ -0,0 +1,18 @@
+
+from .exceptions import SigningError
+from .util import DirectorySigningUtil
+from .deb import DebChangesFiles, DebSigningUtil
+from .rpm import RPMMacro, RPMSigningUtil
+from .runner import PackageSigningRunner
+from .cmd import cmd
+
+
+__all__ = (
+    "cmd",
+    "DebChangesFiles",
+    "DebSigningUtil",
+    "DirectorySigningUtil",
+    "PackageSigningRunner",
+    "RPMMacro",
+    "RPMSigningUtil",
+    "SigningError")

envoy.gpg.sign/envoy/gpg/sign/cmd.py

@@ -0,0 +1,24 @@
+
+import sys
+
+from .runner import PackageSigningRunner
+from .deb import DebSigningUtil
+from .rpm import RPMSigningUtil
+
+
+def _register_utils() -> None:
+    PackageSigningRunner.register_util("deb", DebSigningUtil)
+    PackageSigningRunner.register_util("rpm", RPMSigningUtil)
+
+
+def main(*args) -> int:
+    _register_utils()
+    return PackageSigningRunner(*args).run()
+
+
+def cmd():
+    sys.exit(main(*sys.argv[1:]))
+
+
+if __name__ == "__main__":
+    cmd()

envoy.gpg.sign/envoy/gpg/sign/deb.py

@@ -0,0 +1,106 @@
+
+import pathlib
+from functools import cached_property
+from itertools import chain
+from typing import Iterator, Type
+
+from .exceptions import SigningError
+from .util import DirectorySigningUtil
+
+
+class DebChangesFiles(object):
+    """Creates a set of `changes` files for specific distros from a src
+    `changes` file.
+
+    eg, if src changes file is `envoy_1.100.changes` and `Distribution:`
+    field is `buster bullseye`, it creates:
+
+        `envoy_1.100.changes` -> `envoy_1.100.buster.changes`
+        `envoy_1.100.changes` -> `envoy_1.100.bullseye.changes`
+
+    while replacing any instances of the original distribution name in
+    the respective changes files, eg:
+
+        `buster bullseye` -> `buster`
+        `buster bullseye` -> `bullseye`
+
+    finally, it removes the src changes file.
+    """
+
+    def __init__(self, src):
+        self.src = src
+
+    def __iter__(self) -> Iterator[pathlib.Path]:
+        """Iterate the required changes files, creating them, yielding the paths
+        of the newly created files, and deleting the original
+        """
+        for path in self.files:
+            yield path
+        self.src.unlink()
+
+    @cached_property
+    def distributions(self) -> str:
+        """Find and parse the `Distributions` header in the `changes` file"""
+        with open(self.src) as f:
+            line = f.readline()
+            while line:
+                if not line.startswith("Distribution:"):
+                    line = f.readline()
+                    continue
+                return line.split(":")[1].strip()
+        raise SigningError(
+            f"Did not find Distribution field in changes file {self.src}")
+
+    @property
+    def files(self) -> Iterator[pathlib.Path]:
+        """Create changes files for each distro, yielding the paths"""
+        for distro in self.distributions.split():
+            yield self.changes_file(distro)
+
+    def changes_file(self, distro: str) -> pathlib.Path:
+        """Create a `changes` file for a specific distro"""
+        target = self.changes_file_path(distro)
+        target.write_text(
+            self.src.read_text().replace(
+                self.distributions,
+                distro))
+        return target
+
+    def changes_file_path(self, distro: str) -> pathlib.Path:
+        """Path to write the new changes file to"""
+        return self.src.with_suffix(f".{distro}.changes")
+
+
+class DebSigningUtil(DirectorySigningUtil):
+    """Sign all `changes` packages in a given directory
+
+    the `.changes` spec allows a single `.changes` file to have multiple
+    `Distributions` listed.
+
+    but, most package repos require a single signed `.change` file per
+    distribution, with only one distribution listed.
+
+    this extracts the `.changes` files to -> per-distro
+    `filename.distro.changes`, and removes the original, before signing the
+    files.
+    """
+
+    command_name = "debsign"
+    ext = "changes"
+    _package_type = "deb"
+
+    @cached_property
+    def command_args(self) -> tuple:
+        return ("-k", self.maintainer.fingerprint)
+
+    @property
+    def changes_files(self) -> Type[DebChangesFiles]:
+        return DebChangesFiles
+
+    @cached_property
+    def pkg_files(self) -> tuple:
+        """Mangled .changes paths"""
+        return tuple(
+            chain.from_iterable(
+                self.changes_files(src)
+                for src in super().pkg_files))

envoy.gpg.sign/envoy/gpg/sign/exceptions.py

@@ -0,0 +1,4 @@
+
+
+class SigningError(Exception):
+    pass

envoy.gpg.sign/envoy/gpg/sign/rpm.py

@@ -0,0 +1,87 @@
+
+import pathlib
+from functools import cached_property
+from typing import Type, Union
+
+from .exceptions import SigningError
+from .util import DirectorySigningUtil
+
+
+class RPMMacro:
+    """`.rpmmacros` configuration for rpmsign"""
+
+    _macro_filename = ".rpmmacros"
+
+    def __init__(
+            self,
+            home: Union[pathlib.Path, str],
+            overwrite: bool = False, **kwargs):
+        self._home = home
+        self.overwrite = bool(overwrite)
+        self.kwargs = kwargs
+
+    @property
+    def home(self) -> pathlib.Path:
+        return pathlib.Path(self._home)
+
+    @property
+    def path(self) -> pathlib.Path:
+        return self.home.joinpath(self._macro_filename)
+
+    @property
+    def macro(self) -> str:
+        macro = self.template
+        for k, v in self.kwargs.items():
+            macro = macro.replace(f"__{k.upper()}__", str(v))
+        return macro
+
+    @property
+    def template(self) -> str:
+        return pathlib.Path(
+            __file__).parent.joinpath(
+                "rpm_macro.tmpl").read_text()
+
+    def write(self) -> None:
+        if not self.overwrite and self.path.exists():
+            return
+        self.path.write_text(self.macro)
+
+
+class RPMSigningUtil(DirectorySigningUtil):
+    """Sign all RPM packages in a given directory"""
+
+    command_name = "rpmsign"
+    ext = "rpm"
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.setup()
+
+    @cached_property
+    def command(self) -> str:
+        gpg2_available = (
+            self.maintainer.gpg_bin
+            and self.maintainer.gpg_bin.name == "gpg2")
+        if not gpg2_available:
+            raise SigningError("GPG2 is required to sign RPM packages")
+        return super().command
+
+    @cached_property
+    def command_args(self) -> tuple:
+        return ("--key-id", self.maintainer.fingerprint, "--addsign")
+
+    @property
+    def rpmmacro(self) -> Type[RPMMacro]:
+        return RPMMacro
+
+    def setup(self) -> None:
+        """Create the .rpmmacros file if it doesn't exist"""
+        self.rpmmacro(
+            self.maintainer.home,
+            maintainer=self.maintainer.name,
+            gpg_bin=self.maintainer.gpg_bin,
+            gpg_config=self.maintainer.gnupg_home).write()
+
+    def sign_pkg(self, pkg_file: pathlib.Path) -> None:
+        pkg_file.chmod(0o755)
+        super().sign_pkg(pkg_file)

envoy.gpg.sign/envoy/gpg/sign/rpm_macro.tmpl

@@ -0,0 +1,5 @@
+%_signature gpg
+%_gpg_path __GPG_CONFIG__
+%_gpg_name __MAINTAINER__
+%_gpgbin __GPG_BIN__
+%__gpg_sign_cmd %{__gpg} gpg --force-v3-sigs --batch --verbose --no-armor --no-secmem-warning -u "%{_gpg_name}" -sbo %{__signature_filename} --digest-algo sha256 %{__plaintext_filename}'