[#45] Automated metadata file retrieval and parsing

Refactored base configuration model in order to provide a url and do all the fetching and parsing based on this. The urls are cached for a day and the xml file is updated via a command.
maykinmedia · Oct 13, 2023 · 3bfda54 · 3bfda54
1 parent e919dd6
commit 3bfda54
Show file tree

Hide file tree

Showing 12 changed files with 624 additions and 7 deletions.
diff --git a/digid_eherkenning/admin.py b/digid_eherkenning/admin.py
@@ -9,6 +9,7 @@
 
 @admin.register(DigidConfiguration)
 class DigidConfigurationAdmin(PrivateMediaMixin, SingletonModelAdmin):
+    readonly_fields = ("idp_metadata_file", "idp_service_entity_id")
     fieldsets = (
         (
             _("X.509 Certificate"),
@@ -23,8 +24,9 @@ class DigidConfigurationAdmin(PrivateMediaMixin, SingletonModelAdmin):
             _("Identity provider"),
             {
                 "fields": (
-                    "idp_metadata_file",
+                    "metadata_file_source",
                     "idp_service_entity_id",
+                    "idp_metadata_file",
                 ),
             },
         ),
@@ -72,6 +74,7 @@ class DigidConfigurationAdmin(PrivateMediaMixin, SingletonModelAdmin):
 
 @admin.register(EherkenningConfiguration)
 class EherkenningConfigurationAdmin(PrivateMediaMixin, SingletonModelAdmin):
+    readonly_fields = ("idp_metadata_file", "idp_service_entity_id")
     fieldsets = (
         (
             _("X.509 Certificate"),
@@ -86,8 +89,9 @@ class EherkenningConfigurationAdmin(PrivateMediaMixin, SingletonModelAdmin):
             _("Identity provider"),
             {
                 "fields": (
-                    "idp_metadata_file",
+                    "metadata_file_source",
                     "idp_service_entity_id",
+                    "idp_metadata_file",
                 ),
             },
         ),

diff --git a/digid_eherkenning/management/commands/update_stored_metadata.py b/digid_eherkenning/management/commands/update_stored_metadata.py
@@ -0,0 +1,42 @@
+from django.core.cache import cache
+from django.core.management import BaseCommand, CommandError
+
+from digid_eherkenning.models.digid import DigidConfiguration
+from digid_eherkenning.models.eherkenning import EherkenningConfiguration
+
+
+class Command(BaseCommand):
+    help = "Updates the stored metadata file and repopulates the db fields."
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "--digid",
+            action="store_true",
+            help="Update the DigiD configuration metadata.",
+        )
+        parser.add_argument(
+            "--eherkenning",
+            action="store_true",
+            help="Update the Eherkenning configuration metadata.",
+        )
+
+    def handle(self, **options):
+        if options["digid"]:
+            config = DigidConfiguration.get_solo()
+        elif options["eherkenning"]:
+            config = EherkenningConfiguration.get_solo()
+        else:
+            raise CommandError(
+                "A required argument is missing. Please provide either digid or eherkenning."
+            )
+
+        # delete the cache for the urls in order to trigger fetching and parsing xml again
+        if config.metadata_file_source and cache.get(config._meta.object_name):
+            cache.delete(config._meta.object_name)
+            config.save()
+
+            self.stdout.write(self.style.SUCCESS("Update was successful"))
+        else:
+            self.stdout.write(
+                self.style.WARNING("Update failed, no metadata file source found")
+            )
diff --git a/digid_eherkenning/migrations/0006_digidconfiguration_metadata_file_source_and_more.py b/digid_eherkenning/migrations/0006_digidconfiguration_metadata_file_source_and_more.py
@@ -0,0 +1,83 @@
+# Generated by Django 4.2.6 on 2023-10-12 14:36
+
+from django.db import migrations, models
+import privates.fields
+import privates.storages
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        (
+            "digid_eherkenning",
+            "0005_alter_eherkenningconfiguration_eh_service_instance_uuid_and_more",
+        ),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="digidconfiguration",
+            name="metadata_file_source",
+            field=models.URLField(
+                default="",
+                help_text="The URL-source where the XML metadata file can be retrieved from.",
+                max_length=255,
+                verbose_name="metadata file(XML) URL",
+            ),
+        ),
+        migrations.AddField(
+            model_name="eherkenningconfiguration",
+            name="metadata_file_source",
+            field=models.URLField(
+                default="",
+                help_text="The URL-source where the XML metadata file can be retrieved from.",
+                max_length=255,
+                verbose_name="metadata file(XML) URL",
+            ),
+        ),
+        migrations.AlterField(
+            model_name="digidconfiguration",
+            name="idp_metadata_file",
+            field=privates.fields.PrivateMediaFileField(
+                blank=True,
+                help_text="The metadata file of the identity provider. This is auto populated by the retrieved metadata XML file.",
+                null=True,
+                storage=privates.storages.PrivateMediaFileSystemStorage(),
+                upload_to="",
+                verbose_name="identity provider metadata",
+            ),
+        ),
+        migrations.AlterField(
+            model_name="digidconfiguration",
+            name="idp_service_entity_id",
+            field=models.CharField(
+                blank=True,
+                help_text="Example value: 'https://was-preprod1.digid.nl/saml/idp/metadata'. Note that this must match the 'entityID' attribute on the 'md:EntityDescriptor' node found in the Identity Provider's metadata. This is auto populated by the retrieved metadata XML file.",
+                max_length=255,
+                null=True,
+                verbose_name="identity provider service entity ID",
+            ),
+        ),
+        migrations.AlterField(
+            model_name="eherkenningconfiguration",
+            name="idp_metadata_file",
+            field=privates.fields.PrivateMediaFileField(
+                blank=True,
+                help_text="The metadata file of the identity provider. This is auto populated by the retrieved metadata XML file.",
+                null=True,
+                storage=privates.storages.PrivateMediaFileSystemStorage(),
+                upload_to="",
+                verbose_name="identity provider metadata",
+            ),
+        ),
+        migrations.AlterField(
+            model_name="eherkenningconfiguration",
+            name="idp_service_entity_id",
+            field=models.CharField(
+                blank=True,
+                help_text="Example value: 'https://was-preprod1.digid.nl/saml/idp/metadata'. Note that this must match the 'entityID' attribute on the 'md:EntityDescriptor' node found in the Identity Provider's metadata. This is auto populated by the retrieved metadata XML file.",
+                max_length=255,
+                null=True,
+                verbose_name="identity provider service entity ID",
+            ),
+        ),
+    ]
diff --git a/digid_eherkenning/models/base.py b/digid_eherkenning/models/base.py
@@ -1,13 +1,18 @@
+from django.core.cache import cache
 from django.core.exceptions import ValidationError
+from django.core.files.base import ContentFile
 from django.db import models
 from django.utils.encoding import force_str
 from django.utils.translation import gettext_lazy as _
 
+from onelogin.saml2.constants import OneLogin_Saml2_Constants
+from onelogin.saml2.idp_metadata_parser import OneLogin_Saml2_IdPMetadataParser
 from privates.fields import PrivateMediaFileField
 from simple_certmanager.models import Certificate
 from solo.models import SingletonModel
 
 from ..choices import DigestAlgorithms, SignatureAlgorithms, XMLContentTypes
+from ..settings import get_setting
 
 
 class ConfigurationManager(models.Manager):
@@ -29,17 +34,31 @@ class BaseConfiguration(SingletonModel):
     )
     idp_metadata_file = PrivateMediaFileField(
         _("identity provider metadata"),
-        blank=False,
-        help_text=_("The metadata file of the identity provider."),
+        blank=True,
+        null=True,
+        help_text=_(
+            "The metadata file of the identity provider. This is auto populated "
+            "by the retrieved metadata XML file."
+        ),
     )
     idp_service_entity_id = models.CharField(
         _("identity provider service entity ID"),
         max_length=255,
-        blank=False,
+        blank=True,
+        null=True,
         help_text=_(
             "Example value: 'https://was-preprod1.digid.nl/saml/idp/metadata'. Note "
             "that this must match the 'entityID' attribute on the "
-            "'md:EntityDescriptor' node found in the Identity Provider's metadata."
+            "'md:EntityDescriptor' node found in the Identity Provider's metadata. "
+            "This is auto populated by the retrieved metadata XML file."
+        ),
+    )
+    metadata_file_source = models.URLField(
+        _("metadata file(XML) URL"),
+        max_length=255,
+        default="",
+        help_text=_(
+            "The URL-source where the XML metadata file can be retrieved from."
         ),
     )
     want_assertions_signed = models.BooleanField(
@@ -166,7 +185,78 @@ class Meta:
     def __str__(self):
         return force_str(self._meta.verbose_name)
 
+    def populate_xml_fields(self, urls, xml):
+        """
+        Populates the idp_metadata_file and idp_service_entity_id fields based on the
+        fetched xml metadata
+        """
+        self.idp_service_entity_id = urls["entityId"]
+        content = ContentFile(xml.encode("utf-8"))
+        self.idp_metadata_file.save("metadata.xml", content, save=False)
+
+    def fetch_xml_metadata(self):
+        """
+        Retrieves the xml metadata from the provided url (metadata_file_source)
+
+        :return a string of the xml metadata.
+        """
+        return OneLogin_Saml2_IdPMetadataParser.get_metadata(self.metadata_file_source)
+
+    def parse_data_from_xml_source(self):
+        """
+        Parses the xml metadata
+
+        :return a tuple of a dictionary with the useful urls and the xml string itself.
+        """
+        urls = {}
+        xml = ""
+
+        try:
+            xml = self.fetch_xml_metadata()
+            parsed_idp_metadata = OneLogin_Saml2_IdPMetadataParser.parse(
+                xml,
+                required_sso_binding=OneLogin_Saml2_Constants.BINDING_HTTP_POST,
+                required_slo_binding=OneLogin_Saml2_Constants.BINDING_HTTP_POST,
+            )
+        except Exception:
+            raise ValidationError(
+                _(
+                    "An error has occured while processing the xml file. Make sure the file is valid"
+                )
+            )
+
+        if idp := parsed_idp_metadata.get("idp"):
+            urls = {
+                "entityId": (
+                    idp.get("entityId")
+                    if not idp.get("entityId").startswith("urn")
+                    else self.metadata_file_source
+                ),
+                "sso_url": idp.get("singleSignOnService", {}).get("url"),
+                "slo_url": idp.get("singleLogoutService", {}).get("url"),
+            }
+
+            if cache.get(self._meta.object_name):
+                cache.delete(self._meta.object_name)
+
+        return (urls, xml)
+
     def save(self, *args, **kwargs):
+        if self.pk and self.metadata_file_source:
+            if (
+                cache.get(self._meta.object_name, {}).get("entityId")
+                != self.metadata_file_source
+            ):
+                urls, xml = self.parse_data_from_xml_source()
+
+                if urls and xml:
+                    self.populate_xml_fields(urls, xml)
+                    cache.set(
+                        self._meta.object_name,
+                        urls,
+                        get_setting("METADATA_URLS_CACHE_TIMEOUT"),
+                    )
+
         if self.base_url.endswith("/"):
             self.base_url = self.base_url[:-1]
         super().save(*args, **kwargs)

diff --git a/digid_eherkenning/settings.py b/digid_eherkenning/settings.py
@@ -12,6 +12,7 @@
 # Public settings
 class Defaults:
     DIGID_SESSION_AGE: int = 60 * 15  # 15 minutes, in seconds
+    METADATA_URLS_CACHE_TIMEOUT = 86400
 
 
 def get_setting(name: str):

diff --git a/docs/cli.rst b/docs/cli.rst
@@ -9,6 +9,7 @@ django-digid-eherkenning ships with a couple of Django management commands:
 * ``generate_digid_metadata``
 * ``generate_eherkenning_metadata``
 * ``generate_eherkenning_dienstcatalogus``
+* ``update_stored_metadata``
 
 For details, call:
 

diff --git a/docs/configuration.rst b/docs/configuration.rst
@@ -24,7 +24,7 @@ If this header is not set or empty, we instead get the value from ``REMOTE_ADDR`
 
 **Protecting metadata endpoints**
 
-The metdata URLs are open by design to facilitate sharing these URLs with identity
+The metadata URLs are open by design to facilitate sharing these URLs with identity
 providers or other interested parties. Because the metadata is generated on the fly,
 there is a Denial-of-Service risk. We recommend to protect these URLs at the web-server
 level by:
@@ -57,3 +57,7 @@ Django settings
   .. note:: This setting is a last resort and it will expire after 15 minutes even if
      there is user activity. Typically you want to define a middleware in your project
      to extend the session duration while there is still activity.
+
+``METADATA_URLS_CACHE_TIMEOUT``
+  The library uses django cache in order to store some useful urls. This prevents reading an XML file
+  if this has not been updated. Defaults to 86400 (1 day).
diff --git a/tests/conftest.py b/tests/conftest.py
@@ -12,6 +12,13 @@
 
 
 DIGID_TEST_METADATA_FILE = BASE_DIR / "files" / "digid" / "metadata"
+DIGID_TEST_METADATA_FILE_SLO_POST = (
+    BASE_DIR / "files" / "digid" / "metadata_with_slo_POST"
+)
+DIGID_TEST_METADATA_FILE_SLO_POST_2 = (
+    BASE_DIR / "files" / "digid" / "metadata_with_slo_POST_2"
+)
+
 DIGID_TEST_KEY_FILE = BASE_DIR / "files" / "snakeoil-cert" / "ssl-cert-snakeoil.key"
 DIGID_TEST_CERTIFICATE_FILE = (
     BASE_DIR / "files" / "snakeoil-cert" / "ssl-cert-snakeoil.pem"
@@ -20,6 +27,7 @@
     "base_url": "https://sp.example.nl",
     "entity_id": "sp.example.nl/digid",
     "idp_service_entity_id": "https://was-preprod1.digid.nl/saml/idp/metadata",
+    "metadata_file_source": "https://was-preprod1.digid.nl/saml/idp/metadata",
     "attribute_consuming_service_index": "1",
     "service_name": "Example",
     "requested_attributes": [],