Switch X509CertGuard to db storage and urlencode

The X509CertGuard was requiring the user to perform newline stripping
from certificates, but this operation invalidates some certificates.
Therefore it is not possible to continue with this method.

This PR switches the expectation of certificate delivery to be
urlencoded and no longer with newlines striped.

This PR also stores the `X509CertGuard.ca_certificate` in the database
instead of on the filesystem.

This PR fully regenerates the migrations, which is a breaking change. It
comes with a .removal release note advising users as such.

https://pulp.plan.io/issues/6352
closes #6352

CHANGES/6352.feature

@@ -0,0 +1 @@
+``X509CertGuard.ca_certificate`` is now stored in the database and not on the filesystem.

CHANGES/6352.removal

@@ -0,0 +1,7 @@
+Migrations had to be regenerated from scratch due to a backwards incompatible change where
+``X509ContentGuard.ca_certificate`` is now stored in the database and not on the filesystem. Users
+who have already run migrations will need to drop the ``RHSMCertGuard`` and ``X509CertGuard`` tables
+manually from their databases, reapply migrations, and re-create their CertGuard objects.
+
+Also the submission of the client cert to the content app occurs via the `X-CLIENT-CERT` header, and
+is expected to be urlencoded.

pulp_certguard/app/migrations/0001_initial.py

@@ -1,24 +1,34 @@
-# Generated by Django 2.2.3 on 2019-09-04 19:05
+# Generated by Django 2.2.11 on 2020-04-01 20:52
 
 from django.db import migrations, models
 import django.db.models.deletion
-import pulp_certguard.app.models
 
 
 class Migration(migrations.Migration):
 
     initial = True
 
     dependencies = [
-        ('core', '0004_add_duplicated_reserved_resources'),
+        ('core', '0026_task_group'),
     ]
 
     operations = [
+        migrations.CreateModel(
+            name='RHSMCertGuard',
+            fields=[
+                ('contentguard_ptr', models.OneToOneField(auto_created=True, on_delete=django.db.models.deletion.CASCADE, parent_link=True, primary_key=True, related_name='certguard_rhsmcertguard', serialize=False, to='core.ContentGuard')),
+                ('ca_certificate', models.TextField()),
+            ],
+            options={
+                'default_related_name': '%(app_label)s_%(model_name)s',
+            },
+            bases=('core.contentguard',),
+        ),
         migrations.CreateModel(
             name='X509CertGuard',
             fields=[
                 ('contentguard_ptr', models.OneToOneField(auto_created=True, on_delete=django.db.models.deletion.CASCADE, parent_link=True, primary_key=True, related_name='certguard_x509certguard', serialize=False, to='core.ContentGuard')),
-                ('ca_certificate', models.FileField(max_length=255, upload_to=pulp_certguard.app.models.X509CertGuard._certpath)),
+                ('ca_certificate', models.TextField()),
             ],
             options={
                 'default_related_name': '%(app_label)s_%(model_name)s',

pulp_certguard/app/models.py

@@ -6,7 +6,6 @@
 
 from OpenSSL import crypto as openssl
 
-from pulpcore.plugin import storage
 from pulpcore.plugin.models import ContentGuard
 
 from pulp_certguard.app.utils import get_rhsm
@@ -20,7 +19,55 @@
 log = getLogger(__name__)
 
 
-class RHSMCertGuard(ContentGuard):
+class BaseCertGuard(ContentGuard):
+    """A Base class all CertGuard implementations should derive from."""
+
+    ca_certificate = models.TextField()
+
+    @staticmethod
+    def _get_client_cert_header(request):
+        try:
+            client_cert_data = request.headers["X-CLIENT-CERT"]
+        except KeyError:
+            msg = _("A client certificate was not received via the `X-CLIENT-CERT` header.")
+            raise PermissionError(msg)
+        return unquote(client_cert_data)
+
+    def _ensure_client_cert_is_trusted(self, unquoted_certificate):
+        try:
+            openssl_ca_cert = openssl.load_certificate(
+                openssl.FILETYPE_PEM, buffer=self.ca_certificate
+            )
+        except openssl.Error as exc:
+            raise PermissionError(str(exc))
+
+        try:
+            openssl_client_cert = openssl.load_certificate(
+                openssl.FILETYPE_PEM, buffer=unquoted_certificate
+            )
+        except openssl.Error as exc:
+            raise PermissionError(str(exc))
+
+        trust_store = openssl.X509Store()
+        trust_store.add_cert(openssl_ca_cert)
+
+        try:
+            context = openssl.X509StoreContext(
+                certificate=openssl_client_cert,
+                store=trust_store,
+            )
+            context.verify_certificate()
+        except openssl.X509StoreContextError:
+            msg = _("Client certificate is not signed by the stored 'ca_certificate'.")
+            raise PermissionError(msg)
+        except openssl.Error as exc:
+            raise PermissionError(str(exc))
+
+    class Meta:
+        abstract = True
+
+
+class RHSMCertGuard(BaseCertGuard):
     """
     A content-guard validating on a RHSM Certificate validated by `python-rhsm`.
 
@@ -39,8 +86,6 @@ class RHSMCertGuard(ContentGuard):
 
     TYPE = 'rhsm'
 
-    ca_certificate = models.TextField()
-
     def __init__(self, *args, **kwargs):
         """Initialize a RHSMCertGuard and ensure this system has python-rhsm on it."""
         get_rhsm()  # Validate that rhsm is installed
@@ -67,45 +112,6 @@ def permit(self, request):
         rhsm_cert = self._create_rhsm_cert_from_pem(unquoted_certificate)
         self._check_paths(rhsm_cert, request.path)
 
-    @staticmethod
-    def _get_client_cert_header(request):
-        try:
-            client_cert_data = request.headers["X-CLIENT-CERT"]
-        except KeyError:
-            msg = _("A client certificate was not received via the `X-CLIENT-CERT` header.")
-            raise PermissionError(msg)
-        return unquote(client_cert_data)
-
-    def _ensure_client_cert_is_trusted(self, unquoted_certificate):
-        try:
-            openssl_ca_cert = openssl.load_certificate(
-                openssl.FILETYPE_PEM, buffer=self.ca_certificate
-            )
-        except openssl.Error as exc:
-            raise PermissionError(str(exc))
-
-        try:
-            openssl_client_cert = openssl.load_certificate(
-                openssl.FILETYPE_PEM, buffer=unquoted_certificate
-            )
-        except openssl.Error as exc:
-            raise PermissionError(str(exc))
-
-        trust_store = openssl.X509Store()
-        trust_store.add_cert(openssl_ca_cert)
-
-        try:
-            context = openssl.X509StoreContext(
-                certificate=openssl_client_cert,
-                store=trust_store,
-            )
-            context.verify_certificate()
-        except openssl.X509StoreContextError:
-            msg = _("Client certificate is not signed by the stored 'ca_certificate'.")
-            raise PermissionError(msg)
-        except openssl.Error as exc:
-            raise PermissionError(str(exc))
-
     @staticmethod
     def _create_rhsm_cert_from_pem(unquoted_certificate):
         try:
@@ -122,7 +128,7 @@ def _check_paths(rhsm_cert, path):
             raise PermissionError(msg)
 
 
-class X509CertGuard(ContentGuard):
+class X509CertGuard(BaseCertGuard):
     """
     A content-guard that authenticates the request based on a client provided X.509 Certificate.
 
@@ -133,137 +139,19 @@ class X509CertGuard(ContentGuard):
 
     TYPE = 'x509'
 
-    def _certpath(self, name):
-        return storage.get_tls_path(self, name)
-
-    ca_certificate = models.FileField(max_length=255, upload_to=_certpath)
-
     def permit(self, request):
-        """Authorize the specified web request.
+        """
+        Validate the client cert is trusted.
 
         Args:
-            request (aiohttp.web.Request): A request for a published file.
+            request: The request from the user.
 
         Raises:
-            PermissionError: When the request cannot be authorized.
-
+            PermissionError: If the client certificate is not trusted from the CA certificated
+                stored as `ca_certificate`.
         """
-        ca = self.ca_certificate.read()
-        validator = X509Validator(ca.decode('utf8'))
-        validator(request)
+        unquoted_certificate = unquote(self._get_client_cert_header(request))
+        self._ensure_client_cert_is_trusted(unquoted_certificate)
 
     class Meta:
         default_related_name = "%(app_label)s_%(model_name)s"
-
-
-class X509Validator:
-    """An X.509 certificate validator."""
-
-    CERT_HEADER_NAME = 'X-CLIENT-CERT'
-
-    @staticmethod
-    def format(pem):
-        """Ensure the PEM encoded certificate is properly formatted.
-
-        The certificate is passed as an HTTP header which does not permit newlines.
-
-        Args:
-            pem (str): A PEM encoded certificate.
-
-        Returns:
-            str: A properly PEM formatted certificate.
-
-        """
-        header = '-----BEGIN CERTIFICATE-----'
-        footer = '-----END CERTIFICATE-----'
-        body = pem.replace(header, '')
-        body = body.replace(footer, '')
-        body = body.strip(' \n\r')
-        return '\n'.join((header, body, footer))
-
-    @staticmethod
-    def load(pem):
-        """Load the PEM encoded certificate.
-
-        Encapsulates complexity of OpenSSL.
-
-        Args:
-            pem (str): A PEM encoded certificate.
-
-        Returns:
-            openssl.X509: The loaded certificate.
-
-        Raises:
-            ValueError: On load failed.
-
-        """
-        try:
-            return openssl.load_certificate(openssl.FILETYPE_PEM, buffer=X509Validator.format(pem))
-        except Exception as le:
-            raise ValueError(str(le))
-
-    def client_certificate(self, request):
-        """Extract and load the client certificate passed in the X-CLIENT-CERT header.
-
-        Args:
-            request (aiohttp.web.Request): A request for a published file.
-
-        Returns:
-            openssl.X509: The loaded certificate.
-
-        Raises:
-            KeyError: When the client certificate header has not
-                been passed in the request.
-
-        """
-        name = self.CERT_HEADER_NAME
-        try:
-            certificate = request.headers[name]
-        except KeyError:
-            reason = _('HTTP header "{h}" not found.').format(h=name)
-            raise KeyError(reason)
-        else:
-            return X509Validator.load(certificate)
-
-    def __init__(self, ca_certificate):
-        """Inits a new validator.
-
-        Args:
-            ca_certificate (str): A PEM encoded CA certificate.
-
-        """
-        self.ca_certificate = self.load(ca_certificate)
-
-    @property
-    def store(self):
-        """A X.509 certificate (trust) store.
-
-        Returns:
-            openssl.X509Store: A store containing the CA certificate.
-
-        """
-        store = openssl.X509Store()
-        store.add_cert(self.ca_certificate)
-        return store
-
-    def __call__(self, request):
-        """Validate the client X.509 certificate passed in the request.
-
-        Args:
-            request (aiohttp.web.Request): A request for a published file.
-
-        Raises:
-            PermissionError: When validation the client certificate
-                cannot be validated.
-
-        """
-        try:
-            context = openssl.X509StoreContext(
-                certificate=self.client_certificate(request),
-                store=self.store,
-            )
-            context.verify_certificate()
-        except openssl.X509StoreContextError:
-            raise PermissionError(_('Client certificate cannot be validated.'))
-        except Exception as pe:
-            raise PermissionError(str(pe))