Merge ~woutervb/snap-store-proxy-charm:SN-484_support_certificates into snap-store-proxy-charm:master

[Wouter van Bommel]

Add handling of tls certificates

When using certificates, we should:
 * Validate that the correct name is mentioned in the certificate
 * Validate that the key and certificate match
 * That we open the correct port
 * That we close the correct port
 * Validate that the correct encoding is used

We explicitly don't check the certificate chain or CA, as that is
something that is up to the endpoint (snapd) to validate and with custom
certificates there is a change that we don't have all the needed
information.

[Jonathan Hartley (tartley)]

This looks fabulous. I have a few nitwit ideas in the diffs, take any you like, but obviously ignore them if you don't.

Approved!

review: Approve

a854da0... by Wouter van Bommel on 2022-03-28

Reworked error handling in handle_certificate

As suggested in review feedback error handling could be simplified, have
done that now and code looks cleaner and is more consistend.

[Wouter van Bommel (woutervb)]

@Jonathan, thanks for the feedback, have taken over your suggestion around error handling as this simplifies complexity and improves consistency

diff --git a/Makefile b/Makefile
index cea8799..c852b89 100644
--- a/Makefile
+++ b/Makefile
@@ -55,11 +55,11 @@ black: $(ENV)
 .PHONY: lint
 lint: $(ENV)
 	@echo Running flake8
-	@$(ENV)/bin/flake8 src tests
+	@$(ENV)/bin/flake8 src tests conftest.py
 	@echo Running black
-	@$(ENV)/bin/black --check src tests
+	@$(ENV)/bin/black --check src tests conftest.py
 	@echo Running isort
-	@$(ENV)/bin/isort --profile black --check src tests
+	@$(ENV)/bin/isort --profile black --check src tests conftest.py
 	@echo Running documentation lint
 	@$(ENV)/bin/pymarkdown -d MD041,MD025 scan docs
 
diff --git a/config.yaml b/config.yaml
index 1bf660f..3ebebd2 100644
--- a/config.yaml
+++ b/config.yaml
@@ -11,4 +11,15 @@ options:
       Registration file generated via `store-admin`, added as ie \
       $(cat <generated file>)
     type: string
-
+  certificate:
+    default: null
+    description: |
+      The public part of the certificate that matches the hostname,\
+      PEM encoded. Add like $(cat <pem certificate>)
+    type: string
+  private_key:
+    default: null
+    description: |
+      The private part of the certificate that matches the hostname,\
+      PEM encoded, without an password. Add like $(cat <pem private>)
+    type: string
diff --git a/conftest.py b/conftest.py
index 31b44ee..1b06028 100644
--- a/conftest.py
+++ b/conftest.py
@@ -3,8 +3,24 @@
 #
 
 import pytest
+from ops.testing import Harness
 
 
 @pytest.fixture(autouse=True)
 def dummy_install_from_the_charmstore(mocker):
     mocker.patch("charm.install_from_the_charmstore")
+
+
[e-mail]()
+def harness():
+    # Helper so we have a 'new' instance every time, which is useful on
+    # occasion.
+    # We can't import charm at a module level, as the search paths aren't
+    # setup at the moment this file is read by pytest. We also can't import
+    # the SnapstoreProxyCharm object directly for similar reasons.
+    import charm
+
+    harness = Harness(charm.SnapstoreProxyCharm)
+    harness.begin()
+    yield harness
+    harness.cleanup()
diff --git a/requirements.txt b/requirements.txt
index c29abf0..ac74b07 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,2 +1,3 @@
 ops >= 1.2.0
 ops-lib-pgsql
+cryptography
\ No newline at end of file
diff --git a/src/charm.py b/src/charm.py
index 043114a..4a64d8f 100755
--- a/src/charm.py
+++ b/src/charm.py
@@ -11,6 +11,8 @@ import json
 import logging
 from urllib.parse import urlparse
 
+from cryptography import x509
+from cryptography.hazmat.primitives import serialization
 from ops.charm import CharmBase
 from ops.framework import StoredState
 from ops.lib import use
@@ -168,7 +170,6 @@ class SnapstoreProxyCharm(CharmBase):
 
         for item in all_config_options.keys():
             logger.debug(f"Testing config item {item} for a new value")
-
             # try to get the value from the charm config, if set
             new_value = getattr(self._stored, item, default_values.get(item, None))
             if new_value is None:
@@ -195,8 +196,11 @@ class SnapstoreProxyCharm(CharmBase):
         actual snap entries"""
 
         has_registration_bundle = False
+        has_certificate_update = False
         for item in updates:
             has_registration_bundle |= item["name"] == "registration_bundle"
+            has_certificate_update |= item["name"] == "certificate"
+            has_certificate_update |= item["name"] == "private_key"
 
         if has_registration_bundle:
             for item in updates:
@@ -237,7 +241,7 @@ class SnapstoreProxyCharm(CharmBase):
             logger.debug(f"Status of tls is {self._stored.ssl}")
 
             # Now make 2 lists, option names and values where the indexes
-            # correspond with eachother
+            # correspond with each other
             options = []
             values = []
             for entry in data.keys():
@@ -263,6 +267,74 @@ class SnapstoreProxyCharm(CharmBase):
                     logger.debug("Stored the store id from the registration bundle")
             configure_proxy(options, values, force=True)
 
+        if has_certificate_update:
+            errors = self.handle_certificate()
+            if errors:
+                logger.error(errors)
+                self.errors.append(errors)
+
+    def handle_certificate(self):
+        # If we are called, then we know that the config has changed, and that
+        # self._stored["certificate|private_key"] contains the expected values
+
+        # See if the certificate matches the key, which is basically compare the
+        # modules of the 2 certificate parts, they should be equal
+        try:
+            raw_certificate = getattr(self._stored, "certificate", None)
+            if raw_certificate is None:
+                return "Missing TLS certificate"
+            cert = x509.load_pem_x509_certificate(raw_certificate.encode())
+        except (ValueError, TypeError) as exc:
+            logger.warning(f"Problem {str(exc)}, while loading certificate")
+            return "Problem loading the certificate, they should be PEM encoded strings"
+
+        try:
+            raw_key = getattr(self._stored, "private_key", None)
+            if raw_key is None:
+                self.errors.append("Missing TLS private key")
+                return
+            key = serialization.load_pem_private_key(raw_key.encode(), password=None)
+        except (ValueError, TypeError) as exc:
+            logger.warning(f"Problem {str(exc)}, while loading certificate")
+            return "Problem loading the private_key, they should be PEM encoded strings"
+
+        if cert.public_key().public_numbers() != key.private_numbers().public_numbers:
+            return "The certificate and the key don't match, skipping processing them"
+
+        # Build a list of names that are embedded in the certificate, start with the old
+        # (deprecated) CN
+        cert_names = set(
+            cn.value for cn in cert.subject.get_attributes_for_oid(x509.OID_COMMON_NAME)
+        )
+        # Now handle SANs
+        try:
+            san = cert.extensions.get_extension_for_oid(
+                x509.OID_SUBJECT_ALTERNATIVE_NAME
+            )
+        except x509.extensions.ExtensionNotFound:
+            pass
+        else:
+            cert_names.update(san.value.get_values_for_type(x509.DNSName))
+            cert_names.update(
+                str(ip) for ip in san.value.get_values_for_type(x509.IPAddress)
+            )
+
+        domain = getattr(self._stored, "domain")
+        if domain not in cert_names:
+            logger.warning(f"Could not find {domain} in list {cert_names}")
+            return "Certificate does not contain the proxy's registered name"
+
+        # Okay, the certificate and private key do make sense and match this host,
+        # so let's install them
+        configure_proxy(
+            ["proxy.tls.cert", "proxy.tls.key"],
+            [
+                getattr(self._stored, "certificate"),
+                getattr(self._stored, "private_key"),
+            ],
+        )
+        logger.info("Successfully installed the tls certificate on the proxy")
+
     def handle_pgsql_dsn_change(self):
         """If the database is configured, then we try to configure the snap."""
         installation_source = getattr(self._stored, "snap_install_source", None)
@@ -292,7 +364,12 @@ class SnapstoreProxyCharm(CharmBase):
             return
 
         # We are in a proper state, let's open the port
-        open_port(ssl=getattr(self._stored, "ssl", False))
+        # We only start listening if we have a certificate, so check for a
+        # certificate pair, and only then open the ssl port
+        is_ssl = (getattr(self._stored, "private_key", None) is not None) and (
+            getattr(self._stored, "certificate", None) is not None
+        )
+        open_port(ssl=is_ssl)
 
         self.unit.status = ActiveStatus(f"Running on: {self._stored.domain}")
 
diff --git a/src/helpers.py b/src/helpers.py
index f4000ac..9a82576 100644
--- a/src/helpers.py
+++ b/src/helpers.py
@@ -48,6 +48,10 @@ def config_options():
     for item in config["options"].keys():
         if item == "registration_bundle":
             type_dict.update({item: "base64+json"})
+        if item == "certificate":
+            type_dict.update({item: "string"})
+        if item == "private_key":
+            type_dict.update({item: "string"})
 
     return type_dict, default_value
 
@@ -56,12 +60,15 @@ def open_port(ssl=False):
     # Based on charm-helpers open_port function, as this is not provided by
     # the charm framework
     if ssl:
-        port = 443
+        open_port = 443
+        close_port = 80
     else:
-        port = 80
+        open_port = 80
+        close_port = 443
 
     # Don't catch exceptions, just lett them pass on
-    check_call(["open-port", f"{port}/tcp"])
+    check_call(["open-port", f"{open_port}/tcp"])
+    check_call(["close-port", f"{close_port}/tcp"])
 
 
 all_config_options, default_values = config_options()
diff --git a/tests/test_charm.py b/tests/test_charm.py
index b92efce..eff60e8 100644
--- a/tests/test_charm.py
+++ b/tests/test_charm.py
@@ -6,12 +6,20 @@
 
 import base64
 import hashlib
+import random
+import string
 import subprocess
+from datetime import datetime, timedelta
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
 import pytest
 import yaml
+from cryptography import x509
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives import hashes, serialization
+from cryptography.hazmat.primitives.asymmetric import rsa
+from cryptography.x509.oid import NameOID
 from ops.model import BlockedStatus
 from ops.testing import Harness
 
@@ -541,3 +549,213 @@ class TestInstallation(CharmTest):
         assert not hasattr(self.harness.charm._stored, "snap_md5")
         assert len(self.harness.charm.errors) == 1
         assert self.harness.charm.errors[0] == "Failed to install the snap-store-proxy"
+
+
+class TestTLSCertificate(CharmTest):
+    @staticmethod
+    def setup_harness(in_harness):
+        # Set some default values so we don't have to repeat them for every test
+        in_harness.charm._stored.db_uri = "postgresql://fake/database"
+        in_harness.charm._stored.snap_install_source = "store"
+        in_harness.charm._stored.database_created = True
+        in_harness.charm._stored.registered = True
+
+    @staticmethod
+    def create_x509_certificate(name, is_san):
+        """Helper to generate a certificate + private key
+
+        is is_san is true, then the name will be placed in the SAN section.
+        The certificate generated, will be self signed."""
+        key = rsa.generate_private_key(
+            public_exponent=65537, key_size=4048, backend=default_backend()
+        )
+
+        basic_contraints = x509.BasicConstraints(ca=True, path_length=0)
+        now = datetime.utcnow()
+        encoded_name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, name)])
+
+        if is_san:
+            alt_names = [x509.DNSName(name)]
+            san = x509.SubjectAlternativeName(alt_names)
+            # Just generate a string of 12 random characters
+            name = x509.Name(
+                [
+                    x509.NameAttribute(
+                        NameOID.COMMON_NAME,
+                        "".join(random.choice(string.ascii_letters) for x in range(12)),
+                    )
+                ]
+            )
+            cert = (
+                x509.CertificateBuilder()
+                .subject_name(name)
+                .issuer_name(encoded_name)
+                .public_key(key.public_key())
+                .serial_number(1000)
+                .not_valid_before(now)
+                .not_valid_after(now + timedelta(days=10 * 165))
+                .add_extension(basic_contraints, False)
+                .add_extension(san, False)
+                .sign(key, hashes.SHA256(), default_backend())
+            )
+        else:
+            cert = (
+                x509.CertificateBuilder()
+                .subject_name(encoded_name)
+                .issuer_name(encoded_name)
+                .public_key(key.public_key())
+                .serial_number(1000)
+                .not_valid_before(now)
+                .not_valid_after(now + timedelta(days=10 * 165))
+                .add_extension(basic_contraints, False)
+                .sign(key, hashes.SHA256(), default_backend())
+            )
+        cert_pem = cert.public_bytes(encoding=serialization.Encoding.PEM)
+        key_pem = key.private_bytes(
+            encoding=serialization.Encoding.PEM,
+            format=serialization.PrivateFormat.TraditionalOpenSSL,
+            encryption_algorithm=serialization.NoEncryption(),
+        )
+
+        return cert_pem.decode(), key_pem.decode()
+
+    @pytest.mark.parametrize("san", (True, False))
+    def test_all_okay(self, san, harness):
+
+        self.setup_harness(harness)
+
+        # Test for correct handling of missing private key
+        domain = "http://test.domain.just.invented"
+        harness.charm._stored.domain = domain
+        cert, key = self.create_x509_certificate(domain, san)
+
+        # Since valid values will be pushed on to the proxy, we need to stub that here
+        with patch("charm.configure_proxy") as m_configure, patch(
+            "charm.open_port"
+        ) as m_open_port:
+            # Ensure that the default values are set, as expected
+            harness.update_config({"certificate": cert, "private_key": key})
+
+        m_configure.assert_called_once_with(
+            ["proxy.tls.cert", "proxy.tls.key"], [cert, key]
+        )
+        m_open_port.assert_called_once_with(ssl=True)
+
+        assert len(harness.charm.errors) == 0
+
+    @pytest.mark.parametrize("san", (True, False))
+    def test_domain_mismatch(self, san, harness):
+
+        self.setup_harness(harness)
+
+        # Test for correct handling of missing private key
+        domain = "http://test.domain.just.invented"
+        harness.charm._stored.domain = domain + ".some.added.test"
+        cert, key = self.create_x509_certificate(domain, san)
+
+        # Since valid values will be pushed on to the proxy, we need to stub that here
+        with patch("charm.configure_proxy") as m_configure, patch(
+            "charm.open_port"
+        ) as m_open_port:
+            # Ensure that the default values are set, as expected
+            harness.update_config({"certificate": cert, "private_key": key})
+
+        m_configure.assert_not_called()
+        m_open_port.assert_not_called()
+
+        assert len(harness.charm.errors) == 1
+        assert harness.charm.errors == [
+            "Certificate does not contain the proxy's registered name"
+        ]
+
+    def test_only_certificate(self, harness):
+        self.setup_harness(harness)
+
+        # Test for correct handling of missing private key
+        cert, key = self.create_x509_certificate("https://test.domain", False)
+
+        # Since valid values will be pushed on to the proxy, we need to stub that here
+        with patch("charm.configure_proxy"):
+            # Ensure that the default values are set, as expected
+            harness.update_config({"certificate": cert})
+
+        assert harness.charm.errors == ["Missing TLS private key"]
+        assert len(harness.charm.errors) == 1
+
+    def test_only_private_key(self, harness):
+
+        self.setup_harness(harness)
+
+        # Test for correct handling of missing private key
+        cert, key = self.create_x509_certificate("https://test.domain", False)
+
+        # Since valid values will be pushed on to the proxy, we need to stub that here
+        with patch("charm.configure_proxy"):
+            # Ensure that the default values are set, as expected
+            harness.update_config({"private_key": key})
+
+        assert harness.charm.errors == ["Missing TLS certificate"]
+        assert len(harness.charm.errors) == 1
+
+    def test_key_mismatch(self, harness):
+
+        self.setup_harness(harness)
+
+        # Test for correct handling of missing private key
+        domain = "http://test.domain.just.invented"
+        harness.charm._stored.domain = domain
+        cert1, key1 = self.create_x509_certificate(domain, False)
+        cert2, key2 = self.create_x509_certificate(domain, False)
+
+        # Since valid values will be pushed on to the proxy, we need to stub that here
+        with patch("charm.configure_proxy") as m_configure, patch(
+            "charm.open_port"
+        ) as m_open_port:
+            # Ensure that the default values are set, as expected
+            harness.update_config({"certificate": cert1, "private_key": key2})
+
+        m_configure.assert_not_called()
+        m_open_port.assert_not_called()
+
+        assert len(harness.charm.errors) == 1
+        assert harness.charm.errors == [
+            "The certificate and the key don't match, skipping processing them"
+        ]
+
+    def test_private_key_bad_encoding(self, harness):
+
+        self.setup_harness(harness)
+
+        # Test for correct handling of missing private key
+        cert, key = self.create_x509_certificate("https://test.domain", False)
+
+        # Since valid values will be pushed on to the proxy, we need to stub that here
+        with patch("charm.configure_proxy"):
+            # Ensure that the default values are set, as expected
+            harness.update_config(
+                {"private_key": "My invalid key", "certificate": cert}
+            )
+
+        assert harness.charm.errors == [
+            "Problem loading the private_key, they should be PEM encoded strings"
+        ]
+        assert len(harness.charm.errors) == 1
+
+    def test_certificate_bad_encoding(self, harness):
+
+        self.setup_harness(harness)
+
+        # Test for correct handling of missing private key
+        cert, key = self.create_x509_certificate("https://test.domain", False)
+
+        # Since valid values will be pushed on to the proxy, we need to stub that here
+        with patch("charm.configure_proxy"):
+            # Ensure that the default values are set, as expected
+            harness.update_config(
+                {"private_key": key, "certificate": "My completely invalid certificate"}
+            )
+
+        assert harness.charm.errors == [
+            "Problem loading the certificate, they should be PEM encoded strings"
+        ]
+        assert len(harness.charm.errors) == 1
diff --git a/tests/test_helpers.py b/tests/test_helpers.py
index a62308c..b81b358 100644
--- a/tests/test_helpers.py
+++ b/tests/test_helpers.py
@@ -1,7 +1,5 @@
 from unittest import mock
 
-import pytest
-
 import helpers
 
 
@@ -58,9 +56,25 @@ def test_config_options():
         assert default_value_key in [key for key in all_options.keys()]
 
 
[e-mail]("ssl,port", [(True, 443), (False, 80)])
-def test_open_port(ssl, port):
+def test_open_ssl_port():
     with mock.patch("helpers.check_call") as mocked_call:
-        helpers.open_port(ssl)
+        helpers.open_port(True)
+
+    mocked_call.assert_has_calls(
+        [mock.call(["open-port", "443/tcp"]), mock.call(["close-port", "80/tcp"])],
+        any_order=False,
+    )
+
+    assert mocked_call.call_count == 2
+
+
+def test_open_http_port():
+    with mock.patch("helpers.check_call") as mocked_call:
+        helpers.open_port(False)
+
+    mocked_call.assert_has_calls(
+        [mock.call(["open-port", "80/tcp"]), mock.call(["close-port", "443/tcp"])],
+        any_order=False,
+    )
 
-    mocked_call.assert_called_once_with(["open-port", f"{port}/tcp"])
+    assert mocked_call.call_count == 2