[IMP] auth_saml: user provisioning on login

- custom message when response is too old
- avoid using werkzeug.urls method, they are deprecated
- add missing ondelete cascade when user is deleted
- attribute mapping is now also duplicated when the provider is duplicated
- factorize getting SAML attribute value, allowing using subject.nameId in mapping attributes too

auth_saml/controllers/main.py

@@ -5,10 +5,11 @@
 import functools
 import json
 import logging
+from urllib.parse import quote_plus, unquote_plus, urlencode
 
 import werkzeug.utils
+from saml2.validate import ResponseLifetimeExceed
 from werkzeug.exceptions import BadRequest
-from werkzeug.urls import url_quote_plus
 
 from odoo import (
     SUPERUSER_ID,
@@ -100,7 +101,7 @@ def _auth_saml_request_link(self, provider: models.Model):
         redirect = request.params.get("redirect")
         if redirect:
             params["redirect"] = redirect
-        return "/auth_saml/get_auth_request?%s" % werkzeug.urls.url_encode(params)
+        return "/auth_saml/get_auth_request?%s" % urlencode(params)
 
     @http.route()
     def web_client(self, s_action=None, **kw):
@@ -132,10 +133,13 @@ def web_login(self, *args, **kw):
         response = super().web_login(*args, **kw)
         if response.is_qweb:
             error = request.params.get("saml_error")
+            # TODO c’est par là qu’il faut changer des trucs
             if error == "no-signup":
                 error = _("Sign up is not allowed on this database.")
             elif error == "access-denied":
                 error = _("Access Denied")
+            elif error == "response-lifetime-exceed":
+                error = _("Response Lifetime Exceeded")
             elif error == "expired":
                 error = _(
                     "You do not have access to this database. Please contact"
@@ -169,7 +173,7 @@ def _get_saml_extra_relaystate(self):
             )
 
         state = {
-            "r": url_quote_plus(redirect),
+            "r": quote_plus(redirect),
         }
         return state
 
@@ -231,9 +235,7 @@ def signin(self, **kw):
             )
             action = state.get("a")
             menu = state.get("m")
-            redirect = (
-                werkzeug.urls.url_unquote_plus(state["r"]) if state.get("r") else False
-            )
+            redirect = unquote_plus(state["r"]) if state.get("r") else False
             url = "/web"
             if redirect:
                 url = redirect
@@ -255,6 +257,9 @@ def signin(self, **kw):
             redirect = werkzeug.utils.redirect(url, 303)
             redirect.autocorrect_location_header = False
             return redirect
+        except ResponseLifetimeExceed as e:
+            _logger.debug("Response Lifetime Exceed - %s", str(e))
+            url = "/web/login?saml_error=response-lifetime-exceed"
 
         except Exception as e:
             # signup error

auth_saml/models/auth_saml_attribute_mapping.py

@@ -13,6 +13,7 @@ class AuthSamlAttributeMapping(models.Model):
         "auth.saml.provider",
         index=True,
         required=True,
+        ondelete="cascade",
     )
     attribute_name = fields.Char(
         string="IDP Response Attribute",

auth_saml/models/auth_saml_provider.py

@@ -81,6 +81,7 @@ class AuthSamlProvider(models.Model):
         "auth.saml.attribute.mapping",
         "provider_id",
         string="Attribute Mapping",
+        copy=True,
     )
     active = fields.Boolean(default=True)
     sequence = fields.Integer(index=True)
@@ -136,6 +137,20 @@ class AuthSamlProvider(models.Model):
         default=True,
         help="Whether metadata should be signed or not",
     )
+    # User creation fields
+    create_user = fields.Boolean(
+        default=False,
+        help="Create user if not found. The login and name will defaults to the SAML "
+        "user matching attribute. Use the mapping attributes to change the value "
+        "used.",
+    )
+    create_user_template_id = fields.Many2one(
+        comodel_name="res.users",
+        # Template users, like base.default_user, are disabled by default so allow them
+        domain="[('active', 'in', (True, False))]",
+        default=lambda self: self.env.ref("base.default_user"),
+        help="When creating user, this user is used as a template",
+    )
 
     @api.model
     def _sig_alg_selection(self):
@@ -256,9 +271,7 @@ def _get_auth_request(self, extra_state=None, url_root=None):
         }
         state.update(extra_state)
 
-        sig_alg = ds.SIG_RSA_SHA1
-        if self.sig_alg:
-            sig_alg = getattr(ds, self.sig_alg)
+        sig_alg = getattr(ds, self.sig_alg)
 
         saml_client = self._get_client_for_provider(url_root)
         reqid, info = saml_client.prepare_for_authenticate(
@@ -272,6 +285,7 @@ def _get_auth_request(self, extra_state=None, url_root=None):
         for key, value in info["headers"]:
             if key == "Location":
                 redirect_url = value
+                break
 
         self._store_outstanding_request(reqid)
 
@@ -287,27 +301,15 @@ def _validate_auth_response(self, token: str, base_url: str = None):
             saml2.entity.BINDING_HTTP_POST,
             self._get_outstanding_requests_dict(),
         )
-        matching_value = None
-
-        if self.matching_attribute == "subject.nameId":
-            matching_value = response.name_id.text
-        else:
-            attrs = response.get_identity()
-
-            for k, v in attrs.items():
-                if k == self.matching_attribute:
-                    matching_value = v
-                    break
-
-            if not matching_value:
-                raise Exception(
-                    f"Matching attribute {self.matching_attribute} not found "
-                    f"in user attrs: {attrs}"
-                )
-
-        if matching_value and isinstance(matching_value, list):
-            matching_value = next(iter(matching_value), None)
-
+        try:
+            matching_value = self._get_attribute_value(
+                response, self.matching_attribute
+            )
+        except KeyError:
+            raise KeyError(
+                f"Matching attribute {self.matching_attribute} not found "
+                f"in user attrs: {response.get_identity()}"
+            ) from None
         if isinstance(matching_value, str) and self.matching_attribute_to_lower:
             matching_value = matching_value.lower()
 
@@ -349,24 +351,59 @@ def _metadata_string(self, valid=None, base_url: str = None):
             sign=self.sign_metadata,
         )
 
+    @staticmethod
+    def _get_attribute_value(response, attribute_name: str):
+        """
+
+        :raise: KeyError if attribute is not in the response
+        :param response:
+        :param attribute_name:
+        :return: value of the attribut. if the value is an empty list, return None
+                 otherwise return the first element of the list
+        """
+        if attribute_name == "subject.nameId":
+            return response.name_id.text
+        attrs = response.get_identity()
+        attribute_value = attrs[attribute_name]
+        if isinstance(attribute_value, list):
+            attribute_value = next(iter(attribute_value), None)
+        return attribute_value
+
     def _hook_validate_auth_response(self, response, matching_value):
         self.ensure_one()
         vals = {}
-        attrs = response.get_identity()
 
         for attribute in self.attribute_mapping_ids:
-            if attribute.attribute_name not in attrs:
-                _logger.debug(
+            try:
+                vals[attribute.field_name] = self._get_attribute_value(
+                    response, attribute.attribute_name
+                )
+            except KeyError:
+                _logger.warning(
                     "SAML attribute '%s' found in response %s",
                     attribute.attribute_name,
-                    attrs,
+                    response.get_identity(),
                 )
-                continue
 
-            attribute_value = attrs[attribute.attribute_name]
-            if isinstance(attribute_value, list):
-                attribute_value = attribute_value[0]
+        return {"mapped_attrs": vals}
 
-            vals[attribute.field_name] = attribute_value
+    def _user_copy_defaults(self, validation):
+        """
+        Returns defaults when copying the template user.
 
-        return {"mapped_attrs": vals}
+        Can be overridden with extra information.
+        :param validation: validation result
+        :return: a dictionary for copying template user, empty to avoid copying
+        """
+        self.ensure_one()
+        if not self.create_user:
+            return {}
+        saml_uid = validation["user_id"]
+        return {
+            "name": saml_uid,
+            "login": saml_uid,
+            "active": True,
+            # if signature is not provided by mapped_attrs, it will be computed
+            # due to call to compute method in calling method.
+            "signature": None,
+        } | validation.get("mapped_attrs", {})

auth_saml/models/res_users.py

@@ -7,7 +7,7 @@
 
 import passlib
 
-from odoo import SUPERUSER_ID, _, api, fields, models, registry, tools
+from odoo import SUPERUSER_ID, Command, _, api, fields, models, registry, tools
 from odoo.exceptions import AccessDenied, ValidationError
 
 from .ir_config_parameter import ALLOW_SAML_UID_AND_PASSWORD
@@ -45,19 +45,52 @@ def _auth_saml_signin(self, provider: int, validation: dict, saml_response) -> s
             limit=1,
         )
         user = user_saml.user_id
-        if len(user) != 1:
-            raise AccessDenied()
+        user_copy_defaults = {}
+        if not user:
+            user_copy_defaults = (
+                self.env["auth.saml.provider"]
+                .browse(provider)
+                ._user_copy_defaults(validation)
+            )
+            if not user_copy_defaults:
+                raise AccessDenied()
 
         with registry(self.env.cr.dbname).cursor() as new_cr:
             new_env = api.Environment(new_cr, self.env.uid, self.env.context)
+            if user_copy_defaults:
+                new_user = (
+                    new_env["auth.saml.provider"]
+                    .browse(provider)
+                    .create_user_template_id.with_context(no_reset_password=True)
+                    .copy(
+                        {
+                            **user_copy_defaults,
+                            "saml_ids": [
+                                Command.create(
+                                    {
+                                        "saml_provider_id": provider,
+                                        "saml_uid": saml_uid,
+                                        "saml_access_token": saml_response,
+                                    }
+                                )
+                            ],
+                        }
+                    )
+                )
+                # Update signature as needed.
+                new_user._compute_signature()
+                return new_user.login
+
             # Update the token. Need to be committed, otherwise the token is not visible
             # to other envs, like the one used in login_and_redirect
             user_saml.with_env(new_env).write({"saml_access_token": saml_response})
 
-        if validation.get("mapped_attrs", {}):
-            user.write(validation.get("mapped_attrs", {}))
+            # if a login is changed by a mapped attribute, it needs to be commited too
+            user = user.with_env(new_env)
+            if validation.get("mapped_attrs", {}):
+                user.write(validation.get("mapped_attrs", {}))
 
-        return user.login
+            return user.login
 
     @api.model
     def auth_saml(self, provider: int, saml_response: str, base_url: str = None):

auth_saml/models/res_users_saml.py

@@ -7,7 +7,9 @@ class ResUserSaml(models.Model):
     _name = "res.users.saml"
     _description = "User to SAML Provider Mapping"
 
-    user_id = fields.Many2one("res.users", index=True, required=True)
+    user_id = fields.Many2one(
+        "res.users", index=True, required=True, ondelete="cascade"
+    )
     saml_provider_id = fields.Many2one(
         "auth.saml.provider", string="SAML Provider", index=True
     )

auth_saml/readme/CONFIGURE.md

@@ -2,7 +2,8 @@ To use this module, you need an IDP server, properly set up.
 
 1.  Configure the module according to your IdP’s instructions (Settings
     \> Users & Companies \> SAML Providers).
-2.  Pre-create your users and set the SAML information against the user.
+2.  Pre-create your users and set the SAML information against the user,
+    or use the module ability to create users as they log in.
 
 By default, the module let users have both a password and SAML ids. To
 increase security, disable passwords by using the option in Settings.

auth_saml/readme/HISTORY.md

@@ -1,3 +1,12 @@
-## 16.0.1.0.0
+## 17.0.1.1.0
 
-Initial migration for 16.0.
+- custom message when response is too old
+- avoid using werkzeug.urls method, they are deprecated
+- add missing ondelete cascade when user is deleted
+- attribute mapping is now also duplicated when the provider is duplicated
+- factorize getting SAML attribute value, allowing using subject.nameId in mapping attributes too
+- allow creating user if not found by copying a template user
+
+## 17.0.1.0.0
+
+Initial migration for 17.0.

auth_saml/tests/fake_idp.py

@@ -73,13 +73,21 @@
 }
 
 
+class DummyNameId:
+    """Dummy name id with text value"""
+
+    def __init__(self, text):
+        self.text = text
+
+
 class DummyResponse:
-    def __init__(self, status, data, headers=None):
+    def __init__(self, status, data, headers=None, name_id: str = ""):
         self.status_code = status
         self.text = data
         self.headers = headers or []
         self.content = data
         self._identity = {}
+        self.name_id = DummyNameId(name_id)
 
     def _unpack(self, ver="SAMLResponse"):
         """
@@ -127,6 +135,7 @@ def __init__(self, metadatas=None):
         config.load(settings)
         config.allow_unknown_attributes = True
         Server.__init__(self, config=config)
+        self.mail = "[email protected]"
 
     def get_metadata(self):
         return create_metadata_string(
@@ -163,7 +172,7 @@ def authn_request_endpoint(self, req, binding, relay_state):
             "surName": "Example",
             "givenName": "Test",
             "title": "Ind",
-            "mail": "[email protected]",
+            "mail": self.mail,
         }
 
         resp_args.update({"sign_assertion": True, "sign_response": True})