IDP dynamic configuration

conftest.py

@@ -16,6 +16,7 @@ def saml_request_minimal() -> str:
 
 @pytest.fixture()
 @lru_cache()
-def sp_metadata_xml() -> str:
-    with (XML_ROOT / "metadata/sp_metadata.xml").open("r") as f:
+def sp_metadata_xml(request) -> str:
+    file_name = getattr(request, "param", "sp_metadata")
+    with (XML_ROOT / f"metadata/{file_name}.xml").open("r") as f:
         return f.read()

djangosaml2idp/conf.py

@@ -0,0 +1,37 @@
+import copy
+from typing import Callable, Optional, Union
+
+from django.conf import settings
+from django.core.exceptions import ImproperlyConfigured
+from django.http import HttpRequest
+from django.utils.module_loading import import_string
+
+
+def get_callable(path: Union[Callable, str]) -> Callable:
+    """ Import the function at a given path and return it
+    """
+    if callable(path):
+        return path
+
+    try:
+        config_loader = import_string(path)
+    except ImportError as e:
+        raise ImproperlyConfigured(f'Error importing SAML config loader {path}: "{e}"')
+
+    if not callable(config_loader):
+        raise ImproperlyConfigured("SAML config loader must be a callable object.")
+
+    return config_loader
+
+
+def get_config(config_loader_path: Optional[Union[Callable, str]] = None, request: Optional[HttpRequest] = None) -> dict:
+    """ Load a config_loader function if necessary, and call that function with the request as argument.
+        If the config_loader_path is a callable instead of a string, no importing is necessary and it will be used directly.
+        Return the resulting SPConfig.
+    """
+    static_config = copy.deepcopy(settings.SAML_IDP_CONFIG)
+
+    if config_loader_path is None:
+        return static_config or {}
+    else:
+        return get_callable(config_loader_path)(static_config, request)

djangosaml2idp/idp.py

@@ -1,53 +1,68 @@
-import copy
-
 from django.conf import settings
 from django.core.exceptions import ImproperlyConfigured
+from django.http import HttpRequest
 from django.utils.translation import gettext as _
 from saml2.config import IdPConfig
 from saml2.metadata import entity_descriptor
 from saml2.server import Server
+from typing import Callable, Dict, Optional, Union
+
+from .conf import get_callable, get_config
 
 
 class IDP:
     """ Access point for the IDP Server instance
     """
-    _server_instance: Server = None
+    _server_instances: Dict[str, Server] = {}
 
     @classmethod
-    def construct_metadata(cls, with_local_sp: bool = True) -> dict:
+    def construct_metadata(cls, idp_conf: dict, request: Optional[HttpRequest] = None, with_local_sp: bool = True) -> IdPConfig:
         """ Get the config including the metadata for all the configured service providers. """
+        conf = IdPConfig()
+
         from .models import ServiceProvider
-        idp_config = copy.deepcopy(settings.SAML_IDP_CONFIG)
-        if idp_config:
-            idp_config['metadata'] = {  # type: ignore
-                'local': (
-                    [sp.metadata_path() for sp in ServiceProvider.objects.filter(active=True)]
-                    if with_local_sp else []),
-            }
-        return idp_config
+        sp_queryset = ServiceProvider.objects.none()
+        if with_local_sp:
+            sp_queryset = ServiceProvider.objects.filter(active=True)
+            if getattr(settings, "SAML_IDP_FILTER_SP_QUERYSET", None) is not None:
+                sp_queryset = get_callable(settings.SAML_IDP_FILTER_SP_QUERYSET)(sp_queryset, request)
+
+        idp_conf['metadata'] = {  # type: ignore
+            'local': (
+                [sp.metadata_path() for sp in sp_queryset]
+                if with_local_sp else []
+            ),
+        }
+        try:
+            conf.load(idp_conf)
+        except Exception as e:
+            raise ImproperlyConfigured(_('Could not instantiate an IDP based on the SAML_IDP_CONFIG settings and configured ServiceProviders: {}').format(str(e)))
+        return conf
+
+    @classmethod
+    def load(cls, request: Optional[HttpRequest] = None, config_loader_path: Optional[Union[Callable, str]] = None) -> Server:
+        idp_conf = get_config(config_loader_path, request)
+        if "entityid" not in idp_conf:
+            raise ImproperlyConfigured('The configuration must contain an entityid')
+        entity_id = idp_conf["entityid"]
+
+        if entity_id not in cls._server_instances:
+            # actually initialize the IdP server and cache it
+            conf = cls.construct_metadata(idp_conf, request)
+            cls._server_instances[entity_id] = Server(config=conf)
+
+        return cls._server_instances[entity_id]
 
     @classmethod
-    def load(cls, force_refresh: bool = False) -> Server:
-        """ Instantiate a IDP Server instance based on the config defined in the SAML_IDP_CONFIG settings.
-            Throws an ImproperlyConfigured exception if it could not do so for any reason.
-        """
-        if cls._server_instance is None or force_refresh:
-            conf = IdPConfig()
-            md = cls.construct_metadata()
-            try:
-                conf.load(md)
-                cls._server_instance = Server(config=conf)
-            except Exception as e:
-                raise ImproperlyConfigured(_('Could not instantiate an IDP based on the SAML_IDP_CONFIG settings and configured ServiceProviders: {}').format(str(e)))
-        return cls._server_instance
+    def flush(cls):
+        cls._server_instances = {}
 
     @classmethod
-    def metadata(cls) -> str:
+    def metadata(cls, request: Optional[HttpRequest] = None, config_loader_path: Optional[Union[Callable, str]] = None) -> str:
         """ Get the IDP metadata as a string. """
-        conf = IdPConfig()
         try:
-            conf.load(cls.construct_metadata(with_local_sp=False))
+            conf = cls.construct_metadata(get_config(config_loader_path, request), request, with_local_sp=False)
             metadata = entity_descriptor(conf)
         except Exception as e:
-            raise ImproperlyConfigured(_('Could not instantiate IDP metadata based on the SAML_IDP_CONFIG settings and configured ServiceProviders: {}').format(str(e)))
+            raise ImproperlyConfigured(_('Could not instantiate IDP metadata: {}').format(str(e)))
         return str(metadata)

djangosaml2idp/models.py

@@ -173,7 +173,7 @@ def save(self, *args, **kwargs):
         if not self.metadata_expiration_dt:
             self.metadata_expiration_dt = extract_validuntil_from_metadata(self.local_metadata)
         super().save(*args, **kwargs)
-        IDP.load(force_refresh=True)
+        IDP.flush()
 
     @property
     def attribute_mapping(self) -> Dict[str, str]:
@@ -228,14 +228,10 @@ def metadata_path(self) -> str:
 
     @property
     def sign_response(self) -> bool:
-        if self._sign_response is None:
-            return getattr(IDP.load().config, "sign_response", False)
         return self._sign_response
 
     @property
     def sign_assertion(self) -> bool:
-        if self._sign_assertion is None:
-            return getattr(IDP.load().config, "sign_assertion", False)
         return self._sign_assertion
 
     @property

djangosaml2idp/urls.py

@@ -10,5 +10,5 @@
     path('login/process/', views.LoginProcessView.as_view(), name='saml_login_process'),
     path('login/process_multi_factor/', views.get_multifactor, name='saml_multi_factor'),
     path('slo/<str:binding>/', views.LogoutProcessView.as_view(), name="saml_logout_binding"),
-    path('metadata/', views.metadata, name='saml2_idp_metadata'),
+    path('metadata/', views.MetadataView.as_view(), name='saml2_idp_metadata'),
 ]

djangosaml2idp/views.py

@@ -25,6 +25,7 @@
 from saml2.authn_context import PASSWORD, AuthnBroker, authn_context_class_ref
 from saml2.ident import NameID
 from saml2.saml import NAMEID_FORMAT_UNSPECIFIED
+from saml2.server import Server
 
 from .error_views import error_cbv
 from .idp import IDP
@@ -83,15 +84,22 @@ def check_access(processor: BaseProcessor, request: HttpRequest) -> None:
         raise PermissionDenied(_("You do not have access to this resource"))
 
 
-def get_sp_config(sp_entity_id: str) -> ServiceProvider:
-    """ Get a dict with the configuration for a SP according to the SAML_IDP_SPCONFIG settings.
+def get_sp_config(sp_entity_id: str, idp_server: Server) -> ServiceProvider:
+    """ Get a dict with the configuration for a SP according to the SAML_IDP_SPCONFIG settings and the SP model.
         Raises an exception if no SP matching the given entity id can be found.
     """
     try:
+        if sp_entity_id not in idp_server.metadata.keys():
+            raise ObjectDoesNotExist()
         sp = ServiceProvider.objects.get(entity_id=sp_entity_id, active=True)
     except ObjectDoesNotExist:
-        raise ImproperlyConfigured(_("No active Service Provider object matching the entity_id '{}' found").format(sp_entity_id))
-    return sp
+        raise ObjectDoesNotExist(
+            _("No active Service Provider object matching the entity_id '{}' found for the Identity Provider '{}").format(
+                sp_entity_id, idp_server.ident.name_qualifier
+            )
+        )
+    else:
+        return sp
 
 
 def get_authn(req_info=None):
@@ -101,7 +109,7 @@ def get_authn(req_info=None):
     return broker.get_authn_by_accr(req_authn_context)
 
 
-def build_authn_response(user: User, authn, resp_args, service_provider: ServiceProvider) -> list:  # type: ignore
+def build_authn_response(user: User, authn, resp_args, service_provider: ServiceProvider, idp_server: Server) -> list:  # type: ignore
     """ pysaml2 server.Server.create_authn_response wrapper
     """
     policy = resp_args.get('name_id_policy', None)
@@ -110,7 +118,6 @@ def build_authn_response(user: User, authn, resp_args, service_provider: Service
     else:
         name_id_format = policy.format
 
-    idp_server = IDP.load()
     idp_name_id_format_list = idp_server.config.getattr("name_id_format", "idp") or [NAMEID_FORMAT_UNSPECIFIED]
 
     if name_id_format not in idp_name_id_format_list:
@@ -127,8 +134,8 @@ def build_authn_response(user: User, authn, resp_args, service_provider: Service
         userid=user_id,
         sp_entity_id=service_provider.entity_id,
         # Signing
-        sign_response=service_provider.sign_response,
-        sign_assertion=service_provider.sign_assertion,
+        sign_response=service_provider.sign_response if service_provider.sign_response is not None else getattr(idp_server, 'sign_response', False),
+        sign_assertion=service_provider.sign_assertion if service_provider.sign_assertion is not None else getattr(idp_server, 'sign_assertion', False),
         sign_alg=service_provider.signing_algorithm,
         digest_alg=service_provider.digest_algorithm,
         # Encryption
@@ -139,8 +146,18 @@ def build_authn_response(user: User, authn, resp_args, service_provider: Service
 
 
 class IdPHandlerViewMixin:
-    """ Contains some methods used by multiple views """
+    config_loader_path = getattr(settings, 'SAML_IDP_CONFIG_LOADER', None)
+
+    def get_config_loader_path(self, request: HttpRequest):
+        return self.config_loader_path
+
+    def get_idp_server(self, request: HttpRequest) -> Server:
+        return IDP.load(request, self.get_config_loader_path(request))
 
+    def get_idp_metadata(self, request: HttpRequest) -> str:
+        return IDP.metadata(request, self.get_config_loader_path(request))
+
+    """ Contains some methods used by multiple views """
     def render_login_html_to_string(self, context=None, request=None, using=None):
         """ Render the html response for the login action. Can be using a custom html template if set on the view. """
         default_login_template_name = 'djangosaml2idp/login.html'
@@ -179,7 +196,7 @@ def create_html_response(self, request: HttpRequest, binding, authn_resp, destin
                 "type": "POST",
             }
         else:
-            idp_server = IDP.load()
+            idp_server = self.get_idp_server(request)
             http_args = idp_server.apply_binding(
                 binding=binding,
                 msg_str=authn_resp,
@@ -230,7 +247,7 @@ def get(self, request, *args, **kwargs):
         # TODO: would it be better to store SAML info in request objects?
         # AuthBackend takes request obj as argument...
         try:
-            idp_server = IDP.load()
+            idp_server = self.get_idp_server(request)
 
             # Parse incoming request
             req_info = idp_server.parse_authn_request(request.session['SAMLRequest'], binding)
@@ -245,15 +262,17 @@ def get(self, request, *args, **kwargs):
             resp_args = idp_server.response_args(req_info.message)
             # Set SP and Processor
             sp_entity_id = resp_args.pop('sp_entity_id')
-            service_provider = get_sp_config(sp_entity_id)
+            service_provider = get_sp_config(sp_entity_id, idp_server)
             # Check if user has access
             try:
                 # Check if user has access to SP
                 check_access(service_provider.processor, request)
+            except (ObjectDoesNotExist) as excp:
+                return error_cbv.handle_error(request, exception=excp, status_code=404)
             except PermissionDenied as excp:
                 return error_cbv.handle_error(request, exception=excp, status_code=403)
             # Construct SamlResponse message
-            authn_resp = build_authn_response(request.user, get_authn(), resp_args, service_provider)
+            authn_resp = build_authn_response(request.user, get_authn(), resp_args, service_provider, idp_server)
         except Exception as e:
             return error_cbv.handle_error(request, exception=e, status_code=500)
 
@@ -280,11 +299,15 @@ def get(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:
         request_data = request.POST or request.GET
         passed_data: Dict[str, Union[str, List[str]]] = request_data.copy().dict()
 
+        idp_server = self.get_idp_server(request)
+
         try:
             # get sp information from the parameters
             sp_entity_id = str(passed_data['sp'])
-            service_provider = get_sp_config(sp_entity_id)
+            service_provider = get_sp_config(sp_entity_id, idp_server)
             processor: BaseProcessor = service_provider.processor  # type: ignore
+        except (ObjectDoesNotExist) as excp:
+            return error_cbv.handle_error(request, exception=excp, status_code=404)
         except (KeyError, ImproperlyConfigured) as excp:
             return error_cbv.handle_error(request, exception=excp, status_code=400)
 
@@ -294,8 +317,6 @@ def get(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:
         except PermissionDenied as excp:
             return error_cbv.handle_error(request, exception=excp, status_code=403)
 
-        idp_server = IDP.load()
-
         binding_out, destination = idp_server.pick_binding(
             service="assertion_consumer_service",
             entity_id=sp_entity_id)
@@ -305,7 +326,7 @@ def get(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:
         passed_data['in_response_to'] = "IdP_Initiated_Login"
 
         # Construct SamlResponse messages
-        authn_resp = build_authn_response(request.user, get_authn(), passed_data, service_provider)
+        authn_resp = build_authn_response(request.user, get_authn(), passed_data, service_provider, idp_server)
 
         html_response = self.create_html_response(request, binding_out, authn_resp, destination, passed_data.get('RelayState', ""))
         return self.render_response(request, html_response, processor)
@@ -354,7 +375,7 @@ def get(self, request: HttpRequest, *args, **kwargs):
         relay_state = request.session['RelayState']
         logger.debug("--- {} requested [\n{}] to IDP ---".format(self.__service_name, binding))
 
-        idp_server = IDP.load()
+        idp_server = self.get_idp_server(request)
 
         # adapted from pysaml2 examples/idp2/idp_uwsgi.py
         try:
@@ -414,18 +435,20 @@ def get(self, request: HttpRequest, *args, **kwargs):
         return self.render_response(request, html_response, None)
 
 
+@method_decorator(never_cache, name="dispatch")
+class MetadataView(IdPHandlerViewMixin, View):
+    def get(self, request: HttpRequest, *args, **kwargs) -> HttpResponse:
+        """ Returns an XML with the SAML 2.0 metadata for this Idp.
+            The metadata is constructed on-the-fly based on the config dict in the django settings.
+        """
+        metadata = self.get_idp_metadata(request)
+        return HttpResponse(content=metadata.encode("utf-8"), content_type="text/xml; charset=utf8",)
+
+
 @never_cache
 def get_multifactor(request: HttpRequest) -> HttpResponse:
     if hasattr(settings, "SAML_IDP_MULTIFACTOR_VIEW"):
         multifactor_class = import_string(getattr(settings, "SAML_IDP_MULTIFACTOR_VIEW"))
     else:
         multifactor_class = ProcessMultiFactorView
     return multifactor_class.as_view()(request)
-
-
-@never_cache
-def metadata(request: HttpRequest) -> HttpResponse:
-    """ Returns an XML with the SAML 2.0 metadata for this Idp.
-        The metadata is constructed on-the-fly based on the config dict in the django settings.
-    """
-    return HttpResponse(content=IDP.metadata().encode('utf-8'), content_type="text/xml; charset=utf8")