From a7c5067fa67c7592be4b42378172305533b04e1b Mon Sep 17 00:00:00 2001 From: Ryan Northey Date: Mon, 19 Jul 2021 10:00:02 +0100 Subject: [PATCH] dist: Add GPG identity util Signed-off-by: Ryan Northey --- .github/dependabot.yml | 5 + bazel/repositories_extra.bzl | 5 + tools/gpg/BUILD | 12 + tools/gpg/identity.py | 143 +++++++++++ tools/gpg/requirements.txt | 10 + tools/gpg/tests/test_identity.py | 405 +++++++++++++++++++++++++++++++ tools/testing/BUILD | 4 +- 7 files changed, 582 insertions(+), 2 deletions(-) create mode 100644 tools/gpg/BUILD create mode 100644 tools/gpg/identity.py create mode 100644 tools/gpg/requirements.txt create mode 100644 tools/gpg/tests/test_identity.py diff --git a/.github/dependabot.yml b/.github/dependabot.yml index 758bb344b9774..2a85dba110736 100644 --- a/.github/dependabot.yml +++ b/.github/dependabot.yml @@ -61,6 +61,11 @@ updates: schedule: interval: "daily" +- package-ecosystem: "pip" + directory: "/tools/gpg" + schedule: + interval: "daily" + - package-ecosystem: "pip" directory: "/tools/protodoc" schedule: diff --git a/bazel/repositories_extra.bzl b/bazel/repositories_extra.bzl index 93d442f7f5859..76e44809c964c 100644 --- a/bazel/repositories_extra.bzl +++ b/bazel/repositories_extra.bzl @@ -62,6 +62,11 @@ def _python_deps(): requirements = "@envoy//tools/git:requirements.txt", extra_pip_args = ["--require-hashes"], ) + pip_install( + name = "gpg_pip3", + requirements = "@envoy//tools/gpg:requirements.txt", + extra_pip_args = ["--require-hashes"], + ) pip_install( name = "kafka_pip3", requirements = "@envoy//source/extensions/filters/network/kafka:requirements.txt", diff --git a/tools/gpg/BUILD b/tools/gpg/BUILD new file mode 100644 index 0000000000000..50a3dd91ff14e --- /dev/null +++ b/tools/gpg/BUILD @@ -0,0 +1,12 @@ +load("//bazel:envoy_build_system.bzl", "envoy_package") +load("@gpg_pip3//:requirements.bzl", "requirement") +load("//tools/base:envoy_python.bzl", "envoy_py_library") + +licenses(["notice"]) # Apache 2 + +envoy_package() + +envoy_py_library( + name = "tools.gpg.identity", + deps = [requirement("python-gnupg")], +) diff --git a/tools/gpg/identity.py b/tools/gpg/identity.py new file mode 100644 index 0000000000000..179adece3874e --- /dev/null +++ b/tools/gpg/identity.py @@ -0,0 +1,143 @@ +import logging +import os +import pwd +from functools import cached_property +from email.utils import formataddr, parseaddr +from typing import Optional + +import gnupg + + +class GPGError(Exception): + pass + + +class GPGIdentity(object): + """A GPG identity with a signing key + + The signing key is found either by matching provided name/email, + or by retrieving the first private key. + """ + + def __init__( + self, + name: Optional[str] = None, + email: Optional[str] = None, + log: Optional[logging.Logger] = None): + self._provided_name = name + self._provided_email = email + self._log = log + + def __str__(self) -> str: + return self.uid + + @cached_property + def email(self) -> str: + """Email parsed from the signing key""" + return parseaddr(self.uid)[1] + + @property + def fingerprint(self) -> str: + """GPG key fingerprint""" + return self.signing_key["fingerprint"] + + @cached_property + def gpg(self) -> gnupg.GPG: + return gnupg.GPG() + + @property + def gnupg_home(self) -> str: + return os.path.join(self.home, ".gnupg") + + @cached_property + def home(self) -> str: + """Gets *and sets if required* the `HOME` env var""" + os.environ["HOME"] = os.environ.get("HOME", pwd.getpwuid(os.getuid()).pw_dir) + return os.environ["HOME"] + + @cached_property + def log(self) -> logging.Logger: + return self._log or logging.getLogger(self.__class__.__name__) + + @property + def provided_email(self) -> Optional[str]: + """Provided email for the identity""" + return self._provided_email + + @cached_property + def provided_id(self) -> Optional[str]: + """Provided name and/or email for the identity""" + if not (self.provided_name or self.provided_email): + return + return ( + formataddr(self.provided_name, self.provided_email) if + (self.provided_name and self.provided_email) else + (self.provided_name or self.provided_email)) + + @property + def provided_name(self) -> Optional[str]: + """Provided name for the identity""" + return self._provided_name + + @cached_property + def name(self) -> str: + """Name parsed from the signing key""" + return parseaddr(self.uid)[0] + + @cached_property + def signing_key(self) -> dict: + """A `dict` representing the GPG key to sign with""" + # if name and/or email are provided the list of keys is pre-filtered + # but we still need to figure out which uid matched for the found key + for key in self.gpg.list_keys(True, keys=self.provided_id): + key = self.match(key) + if key: + return key + raise GPGError( + f"No key found for '{self.provided_id}'" if self.provided_id else "No available key") + + @property + def uid(self) -> str: + """UID of the identity's signing key""" + return self.signing_key["uid"] + + def match(self, key: dict) -> Optional[dict]: + """Match a signing key + + The key is found either by matching provided name/email + or the first available private key + + the matching `uid` (or first) is added as `uid` to the dict + """ + if self.provided_id: + key["uid"] = self._match_key(key["uids"]) + return key if key["uid"] else None + if self.log: + self.log.warning("No GPG name/email supplied, signing with first available key") + key["uid"] = key["uids"][0] + return key + + def _match_email(self, uids: list) -> Optional[str]: + """Match only the email""" + for uid in uids: + if parseaddr(uid)[1] == self.provided_email: + return uid + + def _match_key(self, uids: dict) -> Optional[str]: + """If either/both name or email are supplied it tries to match either/both""" + if self.provided_name and self.provided_email: + return self._match_uid(uids) + elif self.provided_name: + return self._match_name(uids) + elif self.provided_email: + return self._match_email(uids) + + def _match_name(self, uids: list) -> Optional[str]: + """Match only the name""" + for uid in uids: + if parseaddr(uid)[0] == self.provided_name: + return uid + + def _match_uid(self, uids: list) -> Optional[str]: + """Match the whole uid - ie `Name `""" + return self.provided_id if self.provided_id in uids else None diff --git a/tools/gpg/requirements.txt b/tools/gpg/requirements.txt new file mode 100644 index 0000000000000..f405a325f70be --- /dev/null +++ b/tools/gpg/requirements.txt @@ -0,0 +1,10 @@ +# +# This file is autogenerated by pip-compile +# To update, run: +# +# pip-compile --generate-hashes tools/gpg/requirements.txt +# +python-gnupg==0.4.7 \ + --hash=sha256:2061f56b1942c29b92727bf9aecbd3cea3893acc9cccbdc7eb4604285efe4ac7 \ + --hash=sha256:3ff5b1bf5e397de6e1fe41a7c0f403dad4e242ac92b345f440eaecfb72a7ebae + # via -r tools/gpg/requirements.txt diff --git a/tools/gpg/tests/test_identity.py b/tools/gpg/tests/test_identity.py new file mode 100644 index 0000000000000..d8ccc8280207a --- /dev/null +++ b/tools/gpg/tests/test_identity.py @@ -0,0 +1,405 @@ +from unittest.mock import MagicMock, PropertyMock + +import pytest + +from tools.gpg import identity + + +@pytest.mark.parametrize("name", ["NAME", None]) +@pytest.mark.parametrize("email", ["EMAIL", None]) +@pytest.mark.parametrize("log", ["LOG", None]) +def test_identity_constructor(name, email, log): + gpg = identity.GPGIdentity(name, email, log) + assert gpg.provided_name == name + assert gpg.provided_email == email + assert gpg._log == log + + +def test_identity_dunder_str(patches): + gpg = identity.GPGIdentity() + patched = patches( + ("GPGIdentity.uid", dict(new_callable=PropertyMock)), + prefix="tools.gpg.identity") + + with patched as (m_uid, ): + m_uid.return_value = "SOME BODY" + assert str(gpg) == "SOME BODY" + + +def test_identity_email(patches): + gpg = identity.GPGIdentity() + patched = patches( + "parseaddr", + ("GPGIdentity.uid", dict(new_callable=PropertyMock)), + prefix="tools.gpg.identity") + + with patched as (m_parse, m_uid): + assert gpg.email == m_parse.return_value.__getitem__.return_value + + assert ( + list(m_parse.return_value.__getitem__.call_args) + == [(1,), {}]) + assert ( + list(m_parse.call_args) + == [(m_uid.return_value,), {}]) + assert "email" in gpg.__dict__ + + +def test_identity_fingerprint(patches): + gpg = identity.GPGIdentity() + patched = patches( + ("GPGIdentity.signing_key", dict(new_callable=PropertyMock)), + prefix="tools.gpg.identity") + + with patched as (m_key, ): + assert gpg.fingerprint == m_key.return_value.__getitem__.return_value + + assert ( + list(m_key.return_value.__getitem__.call_args) + == [('fingerprint',), {}]) + + assert "fingerprint" not in gpg.__dict__ + + +def test_identity_gpg(patches): + gpg = identity.GPGIdentity() + patched = patches( + "gnupg.GPG", + prefix="tools.gpg.identity") + + with patched as (m_gpg, ): + assert gpg.gpg == m_gpg.return_value + + assert ( + list(m_gpg.call_args) + == [(), {}]) + + assert "gpg" in gpg.__dict__ + + +def test_identity_gnupg_home(patches): + gpg = identity.GPGIdentity() + patched = patches( + "os", + ("GPGIdentity.home", dict(new_callable=PropertyMock)), + prefix="tools.gpg.identity") + + with patched as (m_os, m_home): + assert gpg.gnupg_home == m_os.path.join.return_value + + assert ( + list(m_os.path.join.call_args) + == [(m_home.return_value, '.gnupg'), {}]) + + assert "gnupg_home" not in gpg.__dict__ + + +def test_identity_home(patches): + gpg = identity.GPGIdentity() + patched = patches( + "os", + "pwd", + prefix="tools.gpg.identity") + + with patched as (m_os, m_pwd): + assert gpg.home == m_os.environ.__getitem__.return_value + + assert ( + list(m_os.environ.__getitem__.call_args) + == [('HOME', ), {}]) + assert ( + list(m_os.environ.__setitem__.call_args) + == [('HOME', m_os.environ.get.return_value), {}]) + assert ( + list(m_os.environ.get.call_args) + == [('HOME', m_pwd.getpwuid.return_value.pw_dir), {}]) + assert ( + list(m_pwd.getpwuid.call_args) + == [(m_os.getuid.return_value,), {}]) + assert ( + list(m_os.getuid.call_args) + == [(), {}]) + + assert "home" in gpg.__dict__ + + +@pytest.mark.parametrize("log", ["LOGGER", None]) +def test_identity_log(patches, log): + gpg = identity.GPGIdentity() + patched = patches( + "logging", + prefix="tools.gpg.identity") + + gpg._log = log + + with patched as (m_log, ): + if log: + assert gpg.log == log + assert not m_log.getLogger.called + else: + assert gpg.log == m_log.getLogger.return_value + assert ( + list(m_log.getLogger.call_args) + == [(gpg.__class__.__name__, ), {}]) + + +@pytest.mark.parametrize("name", ["NAME", None]) +@pytest.mark.parametrize("email", ["EMAIL", None]) +def test_identity_identity_id(patches, name, email): + gpg = identity.GPGIdentity() + patched = patches( + "formataddr", + ("GPGIdentity.provided_name", dict(new_callable=PropertyMock)), + ("GPGIdentity.provided_email", dict(new_callable=PropertyMock)), + prefix="tools.gpg.identity") + + with patched as (m_format, m_name, m_email): + m_name.return_value = name + m_email.return_value = email + result = gpg.provided_id + + assert "provided_id" in gpg.__dict__ + + if name and email: + assert ( + list(m_format.call_args) + == [('NAME', 'EMAIL'), {}]) + assert result == m_format.return_value + return + + assert not m_format.called + assert result == name or email + + +def test_identity_name(patches): + gpg = identity.GPGIdentity() + patched = patches( + "parseaddr", + ("GPGIdentity.uid", dict(new_callable=PropertyMock)), + prefix="tools.gpg.identity") + + with patched as (m_parse, m_uid): + assert gpg.name == m_parse.return_value.__getitem__.return_value + + assert ( + list(m_parse.return_value.__getitem__.call_args) + == [(0,), {}]) + assert ( + list(m_parse.call_args) + == [(m_uid.return_value,), {}]) + assert "name" in gpg.__dict__ + + +@pytest.mark.parametrize("key", ["KEY1", "KEY2", "KEY5"]) +@pytest.mark.parametrize("name", ["NAME", None]) +@pytest.mark.parametrize("email", ["EMAIL", None]) +def test_identity_signing_key(patches, key, name, email): + packager = MagicMock() + gpg = identity.GPGIdentity() + _keys = ["KEY1", "KEY2", "KEY3"] + patched = patches( + "GPGIdentity.match", + ("GPGIdentity.gpg", dict(new_callable=PropertyMock)), + ("GPGIdentity.provided_id", dict(new_callable=PropertyMock)), + prefix="tools.gpg.identity") + + with patched as (m_match, m_gpg, m_id): + if not name and not email: + m_id.return_value = None + m_match.side_effect = lambda k: (k == key and f"MATCH {k}") + m_gpg.return_value.list_keys.return_value = _keys + if key in _keys: + assert gpg.signing_key == f"MATCH {key}" + _match_attempts = _keys[:_keys.index(key) + 1] + else: + with pytest.raises(identity.GPGError) as e: + gpg.signing_key + if name or email: + assert ( + e.value.args[0] + == f"No key found for '{m_id.return_value}'") + else: + assert ( + e.value.args[0] + == 'No available key') + _match_attempts = _keys + + assert ( + list(m_gpg.return_value.list_keys.call_args) + == [(True, ), dict(keys=m_id.return_value)]) + assert ( + list(list(c) for c in m_match.call_args_list) + == [[(k,), {}] for k in _match_attempts]) + + +def test_identity_uid(patches): + gpg = identity.GPGIdentity() + patched = patches( + ("GPGIdentity.signing_key", dict(new_callable=PropertyMock)), + prefix="tools.gpg.identity") + + with patched as (m_key, ): + assert gpg.uid == m_key.return_value.__getitem__.return_value + + assert ( + list(m_key.return_value.__getitem__.call_args) + == [('uid',), {}]) + + assert "uid" not in gpg.__dict__ + + +@pytest.mark.parametrize("name", ["NAME", None]) +@pytest.mark.parametrize("email", ["EMAIL", None]) +@pytest.mark.parametrize("match", ["MATCH", None]) +@pytest.mark.parametrize("log", [True, False]) +def test_identity_match(patches, name, email, match, log): + gpg = identity.GPGIdentity() + _keys = ["KEY1", "KEY2", "KEY3"] + patched = patches( + "GPGIdentity._match_key", + ("GPGIdentity.provided_id", dict(new_callable=PropertyMock)), + ("GPGIdentity.log", dict(new_callable=PropertyMock)), + prefix="tools.gpg.identity") + key = dict(uids=["UID1", "UID2"]) + + with patched as (m_match, m_id, m_log): + if not log: + m_log.return_value = None + m_match.return_value = match + m_id.return_value = name or email + result = gpg.match(key) + + if not name and not email: + assert not m_match.called + if log: + assert ( + list(m_log.return_value.warning.call_args) + == [('No GPG name/email supplied, signing with first available key',), {}]) + assert ( + result + == {'uids': ['UID1', 'UID2'], 'uid': 'UID1'}) + return + assert ( + list(m_match.call_args) + == [(key["uids"],), {}]) + if log: + assert not m_log.return_value.warning.called + if match: + assert ( + result + == {'uids': ['UID1', 'UID2'], 'uid': 'MATCH'}) + else: + assert not result + + +@pytest.mark.parametrize("uids", [[], ["UID1"], ["UID1", "UID2"]]) +@pytest.mark.parametrize("email", [None, "UID1", "UID1", "UID2", "UID3"]) +def test_identity__match_email(patches, uids, email): + gpg = identity.GPGIdentity() + patched = patches( + "parseaddr", + ("GPGIdentity.provided_email", dict(new_callable=PropertyMock)), + prefix="tools.gpg.identity") + + with patched as (m_parse, m_email): + m_parse.side_effect = lambda _email: ("NAME", _email) + m_email.return_value = email + result = gpg._match_email(uids) + + if email in uids: + assert result == email + assert ( + list(list(c) for c in m_parse.call_args_list) + == [[(uid,), {}] for uid in uids[:uids.index(email) + 1]]) + return + + assert not result + assert ( + list(list(c) for c in m_parse.call_args_list) + == [[(uid,), {}] for uid in uids]) + + +@pytest.mark.parametrize("name", ["NAME", None]) +@pytest.mark.parametrize("email", ["EMAIL", None]) +def test_identity__match_key(patches, name, email): + gpg = identity.GPGIdentity() + _keys = ["KEY1", "KEY2", "KEY3"] + patched = patches( + "GPGIdentity._match_email", + "GPGIdentity._match_name", + "GPGIdentity._match_uid", + ("GPGIdentity.provided_email", dict(new_callable=PropertyMock)), + ("GPGIdentity.provided_name", dict(new_callable=PropertyMock)), + prefix="tools.gpg.identity") + key = dict(uids=["UID1", "UID2"]) + + with patched as (m_email, m_name, m_uid, m_pemail, m_pname): + m_pemail.return_value = email + m_pname.return_value = name + result = gpg._match_key(key) + + if name and email: + assert ( + list(m_uid.call_args) + == [(dict(uids=key["uids"]),), {}]) + assert not m_email.called + assert not m_name.called + assert result == m_uid.return_value + elif name: + assert ( + list(m_name.call_args) + == [(dict(uids=key["uids"]),), {}]) + assert not m_email.called + assert not m_uid.called + assert result == m_name.return_value + elif email: + assert ( + list(m_email.call_args) + == [(dict(uids=key["uids"]),), {}]) + assert not m_name.called + assert not m_uid.called + assert result == m_email.return_value + + +@pytest.mark.parametrize("uids", [[], ["UID1"], ["UID1", "UID2"]]) +@pytest.mark.parametrize("name", [None, "UID1", "UID1", "UID2", "UID3"]) +def test_identity__match_name(patches, uids, name): + gpg = identity.GPGIdentity() + patched = patches( + "parseaddr", + ("GPGIdentity.provided_name", dict(new_callable=PropertyMock)), + prefix="tools.gpg.identity") + + with patched as (m_parse, m_name): + m_parse.side_effect = lambda _name: (_name, "EMAIL") + m_name.return_value = name + result = gpg._match_name(uids) + + if name in uids: + assert result == name + assert ( + list(list(c) for c in m_parse.call_args_list) + == [[(uid,), {}] for uid in uids[:uids.index(name) + 1]]) + return + + assert not result + assert ( + list(list(c) for c in m_parse.call_args_list) + == [[(uid,), {}] for uid in uids]) + + +@pytest.mark.parametrize("uid", ["UID1", "UID7"]) +def test_identity__match_uid(patches, uid): + gpg = identity.GPGIdentity() + uids = [f"UID{i}" for i in range(5)] + matches = uid in uids + patched = patches( + ("GPGIdentity.provided_id", dict(new_callable=PropertyMock)), + prefix="tools.gpg.identity") + + with patched as (m_id, ): + m_id.return_value = uid + if matches: + assert gpg._match_uid(uids) == uid + else: + assert not gpg._match_uid(uids) diff --git a/tools/testing/BUILD b/tools/testing/BUILD index 2f8bd93d6105f..ab36b91e2212c 100644 --- a/tools/testing/BUILD +++ b/tools/testing/BUILD @@ -19,13 +19,13 @@ envoy_py_binary( ":plugin", "//:.coveragerc", "//:pytest.ini", - "//tools/base:runner", - "//tools/base:utils", ], deps = [ requirement("pytest"), requirement("pytest-asyncio"), requirement("pytest-cov"), + "//tools/base:runner", + "//tools/base:utils", ], )