[IMP][14.0] auth_saml: download the provider metadata

auth_saml/models/auth_saml_provider.py

@@ -8,14 +8,16 @@
 import os
 import tempfile
 import urllib.parse
+import requests
 
 # dependency name is pysaml2 # pylint: disable=W7936
 import saml2
 import saml2.xmldsig as ds
 from saml2.client import Saml2Client
 from saml2.config import Config as Saml2Config
-
+from saml2.sigver import SignatureError
 from odoo import api, fields, models
+from odoo.exceptions import UserError
 
 _logger = logging.getLogger(__name__)
 
@@ -42,6 +44,13 @@ class AuthSamlProvider(models.Model):
         ),
         required=True,
     )
+    idp_metadata_url = fields.Char(
+        string="Identity Provider Metadata URL",
+        help="Some SAML providers, notably Office365 can have a metadata "
+        "document which changes over time, and they provide a URL to the "
+        "document instead. When this field is set, the metadata can be "
+        "fetched from the provided URL.",
+    )
     sp_baseurl = fields.Text(
         string="Override Base URL",
         help="""Base URL sent to Odoo with this, rather than automatically
@@ -229,10 +238,19 @@ def _get_config_for_provider(self, base_url: str = None):
             "cert_file": self._get_cert_key_path("sp_pem_public"),
             "key_file": self._get_cert_key_path("sp_pem_private"),
         }
-        sp_config = Saml2Config()
-        sp_config.load(settings)
-        sp_config.allow_unknown_attributes = True
-        return sp_config
+        try:
+            sp_config = Saml2Config()
+            sp_config.load(settings)
+            sp_config.allow_unknown_attributes = True
+            return sp_config
+        except saml2.SAMLError:
+            if self.env.context.get("saml2_retry_after_refresh_metadata", False):
+                raise
+            # Retry after refresh metadata
+            self.action_refresh_metadata_from_url()
+            return self.with_context(
+                saml2_retry_after_refresh_metatata=1
+            )._get_config_for_provider(base_url)
 
     def _get_client_for_provider(self, base_url: str = None):
         sp_config = self._get_config_for_provider(base_url)
@@ -278,12 +296,26 @@ def _validate_auth_response(self, token: str, base_url: str = None):
         """return the validation data corresponding to the access token"""
         self.ensure_one()
 
-        client = self._get_client_for_provider(base_url)
-        response = client.parse_authn_request_response(
-            token,
-            saml2.entity.BINDING_HTTP_POST,
-            self._get_outstanding_requests_dict(),
-        )
+        try:
+            client = self._get_client_for_provider(base_url)
+            response = client.parse_authn_request_response(
+                token,
+                saml2.entity.BINDING_HTTP_POST,
+                self._get_outstanding_requests_dict(),
+            )
+        except SignatureError:
+            # we have a metadata url: try to refresh the metadata document
+            if self.idp_metadata_url:
+                self.action_refresh_metadata_from_url()
+                # retry: if it fails again, we let the exception flow
+                client = self._get_client_for_provider(base_url)
+                response = client.parse_authn_request_response(
+                    token,
+                    saml2.entity.BINDING_HTTP_POST,
+                    self._get_outstanding_requests_dict(),
+                )
+            else:
+                raise
         matching_value = None
 
         if self.matching_attribute == "subject.nameId":
@@ -367,3 +399,28 @@ def _hook_validate_auth_response(self, response, matching_value):
             vals[attribute.field_name] = attribute_value
 
         return {"mapped_attrs": vals}
+
+    def action_refresh_metadata_from_url(self):
+        providers = self.search(
+            [("idp_metadata_url", "ilike", "http%"), ("id", "in", self.ids)]
+        )
+        if not providers:
+            return False
+        # lock the records we might update, so that multiple simultaneous login
+        # attempts will not cause concurrent updates
+        self.env.cr.execute(
+            "SELECT id FROM auth_saml_provider WHERE id in %s FOR UPDATE",
+            (tuple(providers.ids),),
+        )
+        updated = False
+        for provider in providers:
+            document = requests.get(provider.idp_metadata_url)
+            if document.status_code != 200:
+                raise UserError(
+                    f"Unable to download the metadata for {provider.name}: {document.reason}"
+                )
+            if document.text != provider.idp_metadata:
+                provider.idp_metadata = document.text
+                _logger.info("Updated provider metadata for %s", provider.name)
+                updated = True
+        return updated

auth_saml/readme/CONFIGURE.rst

@@ -15,3 +15,7 @@ with the highest priority. It is still possible to access the login without redi
 by using the query parameter ``disable_autoredirect``, as in
 ``https://example.com/web/login?disable_autoredirect=`` The login is also displayed if
 there is an error with SAML login, in order to display any error message.
+
+If you are using Office365 as identity provider, set up the federation metadata document
+rather than the document itself. This will allow the module to refresh the document when
+needed. 

auth_saml/tests/fake_idp.py

@@ -105,11 +105,11 @@ def _unpack(self, ver="SAMLResponse"):
 
 
 class FakeIDP(Server):
-    def __init__(self, metadatas=None):
-        settings = CONFIG
+    def __init__(self, metadatas=None, settings=None):
+        if settings is None:
+            settings = CONFIG
         if metadatas:
             settings.update({"metadata": {"inline": metadatas}})
-
         config = Saml2Config()
         config.load(settings)
         config.allow_unknown_attributes = True

auth_saml/tests/test_pysaml.py

@@ -1,11 +1,13 @@
 import base64
 import os
-
+import responses
 from odoo.exceptions import AccessDenied, UserError, ValidationError
 from odoo.tests import HttpCase, tagged
 
-from .fake_idp import FakeIDP
+from .fake_idp import CONFIG, FakeIDP
 
+import os.path as osp
+from copy import deepcopy
 
 @tagged("saml", "post_install", "-at_install")
 class TestPySaml(HttpCase):
@@ -296,3 +298,56 @@ def test_disallow_user_password_when_changing_settings(self):
             self.authenticate(
                 user="[email protected]", password="NesTNSte9340D720te>/-A"
             )
+
+    @responses.activate
+    def test_download_metadata(self):
+        expected_metadata = self.idp.get_metadata()
+        responses.add(
+            responses.GET,
+            "http://localhost:8000/metadata",
+            status=200,
+            content_type="text/xml",
+            body=expected_metadata,
+        )
+        self.saml_provider.idp_metadata_url = "http://localhost:8000/metadata"
+        self.saml_provider.idp_metadata = ""
+        self.saml_provider.action_refresh_metadata_from_url()
+        self.assertEqual(self.saml_provider.idp_metadata, expected_metadata)
+
+    @responses.activate
+    def test_login_with_saml_metadata_empty(self):
+        self.saml_provider.idp_metadata_url = "http://localhost:8000/metadata"
+        self.saml_provider.idp_metadata = ""
+        expected_metadata = self.idp.get_metadata()
+        responses.add(
+            responses.GET,
+            "http://localhost:8000/metadata",
+            status=200,
+            content_type="text/xml",
+            body=expected_metadata,
+        )
+        self.test_login_with_saml()
+        self.assertEqual(self.saml_provider.idp_metadata, expected_metadata)
+
+    @responses.activate
+    def test_login_with_saml_metadata_key_changed(self):
+        settings = deepcopy(CONFIG)
+        settings["key_file"] = osp.join(
+            osp.dirname(__file__), "data", "key_idp_expired.pem"
+        )
+        settings["cert"] = osp.join(
+            osp.dirname(__file__), "data", "key_idp_expired.pem"
+        )
+        expired_idp = FakeIDP(settings=settings)
+        self.saml_provider.idp_metadata = expired_idp.get_metadata()
+        self.saml_provider.idp_metadata_url = "http://localhost:8000/metadata"
+        up_to_date_metadata = self.idp.get_metadata()
+        self.assertNotEqual(self.saml_provider.idp_metadata, up_to_date_metadata)
+        responses.add(
+            responses.GET,
+            "http://localhost:8000/metadata",
+            status=200,
+            content_type="text/xml",
+            body=up_to_date_metadata,
+        )
+        self.test_login_with_saml()

auth_saml/views/auth_saml.xml

@@ -72,8 +72,23 @@
                         </h1>
                     </div>
                     <group name="idp_settings">
-                        <group string="Identity Provider Settings">
-                            <label for="idp_metadata" />
+                         <group string="Identity Provider Settings">
+                            <span>
+                                <label for="idp_metadata_url" />
+                                <div>
+                                    <field name="idp_metadata_url" />
+                                    <p
+                                        class="help small"
+                                    >If you provider gives you a URL, use this field preferably</p>
+                                </div>
+                                <button
+                                    type="object"
+                                    string="Refresh"
+                                    name="action_refresh_metadata_from_url"
+                                    attrs="{'invisible': [('idp_metadata_url', '=', False)]}"
+                                />
+                            </span>
+                             <label for="idp_metadata" />
                             <div>
                                 <field
                                     name="idp_metadata"