Add SSO attribute requirements for OIDC providers

changelog.d/9609.feature

@@ -0,0 +1 @@
+Logins using OpenID Connect can require attributes on the `userinfo` response in order to login. Contributed by Hubbe King.

docs/sample_config.yaml

@@ -1873,6 +1873,24 @@ saml2_config:
 #           which is set to the claims returned by the UserInfo Endpoint and/or
 #           in the ID Token.
 #
+#   It is possible to configure Synapse to only allow logins if certain attributes
+#   match particular values in the OIDC userinfo. The requirements can be listed under
+#   `attribute_requirements` as shown below. All of the listed attributes must
+#   match for the login to be permitted. Additional attributes can be added to
+#   userinfo by expanding the `scopes` section of the OIDC config to retrieve
+#   additional information from the OIDC provider.
+#
+#   If the OIDC claim is a list, then the attribute must match any value in the list.
+#   Otherwise, it must exactly match the value of the claim. Using the example
+#   below, the `family_name` claim MUST be "Stephensson", but the `groups`
+#   claim MUST contain "admin".
+#
+#   attribute_requirements:
+#     - attribute: family_name
+#       value: "Stephensson"
+#     - attribute: groups
+#       value: "admin"
+#
 # See https://github.com/matrix-org/synapse/blob/master/docs/openid.md
 # for information on how to configure these options.
 #
@@ -1905,6 +1923,9 @@ oidc_providers:
   #      localpart_template: "{{ user.login }}"
   #      display_name_template: "{{ user.name }}"
   #      email_template: "{{ user.email }}"
+  #  attribute_requirements:
+  #    - attribute: userGroup
+  #      value: "synapseUsers"
 
   # For use with Keycloak
   #
@@ -1914,6 +1935,9 @@ oidc_providers:
   #  client_id: "synapse"
   #  client_secret: "copy secret generated in Keycloak UI"
   #  scopes: ["openid", "profile"]
+  #  attribute_requirements:
+  #    - attribute: groups
+  #      value: "admin"
 
   # For use with Github
   #

synapse/config/oidc_config.py

@@ -15,11 +15,12 @@
 # limitations under the License.
 
 from collections import Counter
-from typing import Iterable, Mapping, Optional, Tuple, Type
+from typing import Iterable, List, Mapping, Optional, Tuple, Type
 
 import attr
 
 from synapse.config._util import validate_config
+from synapse.config.sso import SsoAttributeRequirement
 from synapse.python_dependencies import DependencyException, check_requirements
 from synapse.types import Collection, JsonDict
 from synapse.util.module_loader import load_module
@@ -191,6 +192,24 @@ def generate_config_section(self, config_dir_path, server_name, **kwargs):
         #           which is set to the claims returned by the UserInfo Endpoint and/or
         #           in the ID Token.
         #
+        #   It is possible to configure Synapse to only allow logins if certain attributes
+        #   match particular values in the OIDC userinfo. The requirements can be listed under
+        #   `attribute_requirements` as shown below. All of the listed attributes must
+        #   match for the login to be permitted. Additional attributes can be added to
+        #   userinfo by expanding the `scopes` section of the OIDC config to retrieve
+        #   additional information from the OIDC provider.
+        #
+        #   If the OIDC claim is a list, then the attribute must match any value in the list.
+        #   Otherwise, it must exactly match the value of the claim. Using the example
+        #   below, the `family_name` claim MUST be "Stephensson", but the `groups`
+        #   claim MUST contain "admin".
+        #
+        #   attribute_requirements:
+        #     - attribute: family_name
+        #       value: "Stephensson"
+        #     - attribute: groups
+        #       value: "admin"
+        #
         # See https://github.com/matrix-org/synapse/blob/master/docs/openid.md
         # for information on how to configure these options.
         #
@@ -223,6 +242,9 @@ def generate_config_section(self, config_dir_path, server_name, **kwargs):
           #      localpart_template: "{{{{ user.login }}}}"
           #      display_name_template: "{{{{ user.name }}}}"
           #      email_template: "{{{{ user.email }}}}"
+          #  attribute_requirements:
+          #    - attribute: userGroup
+          #      value: "synapseUsers"
 
           # For use with Keycloak
           #
@@ -232,6 +254,9 @@ def generate_config_section(self, config_dir_path, server_name, **kwargs):
           #  client_id: "synapse"
           #  client_secret: "copy secret generated in Keycloak UI"
           #  scopes: ["openid", "profile"]
+          #  attribute_requirements:
+          #    - attribute: groups
+          #      value: "admin"
 
           # For use with Github
           #
@@ -324,6 +349,10 @@ def generate_config_section(self, config_dir_path, server_name, **kwargs):
         },
         "allow_existing_users": {"type": "boolean"},
         "user_mapping_provider": {"type": ["object", "null"]},
+        "attribute_requirements": {
+            "type": "array",
+            "items": SsoAttributeRequirement.JSON_SCHEMA,
+        },
     },
 }
 
@@ -460,6 +489,11 @@ def _parse_oidc_config_dict(
             jwt_header=client_secret_jwt_key_config["jwt_header"],
             jwt_payload=client_secret_jwt_key_config.get("jwt_payload", {}),
         )
+    # parse attribute_requirements from config (list of dicts) into a list of SsoAttributeRequirement
+    attribute_requirements = [
+        SsoAttributeRequirement(**x)
+        for x in oidc_config.get("attribute_requirements", [])
+    ]
 
     return OidcProviderConfig(
         idp_id=idp_id,
@@ -482,6 +516,7 @@ def _parse_oidc_config_dict(
         allow_existing_users=oidc_config.get("allow_existing_users", False),
         user_mapping_provider_class=user_mapping_provider_class,
         user_mapping_provider_config=user_mapping_provider_config,
+        attribute_requirements=attribute_requirements,
     )
 
 
@@ -568,3 +603,6 @@ class OidcProviderConfig:
 
     # the config of the user mapping provider
     user_mapping_provider_config = attr.ib()
+
+    # required attributes to require in userinfo to allow login/registration
+    attribute_requirements = attr.ib(type=List[SsoAttributeRequirement])

synapse/handlers/oidc_handler.py

@@ -278,6 +278,7 @@ def __init__(
         self._config = provider
         self._callback_url = hs.config.oidc_callback_url  # type: str
 
+        self._oidc_attribute_requirements = provider.attribute_requirements
         self._scopes = provider.scopes
         self._user_profile_method = provider.user_profile_method
 
@@ -854,6 +855,18 @@ async def handle_oidc_callback(
             )
 
         # otherwise, it's a login
+        logger.debug("Userinfo for OIDC login: %s", userinfo)
+
+        # Ensure that the attributes of the logged in user meet the required
+        # attributes by checking the userinfo against attribute_requirements
+        # In order to deal with the fact that OIDC userinfo can contain many
+        # types of data, we wrap non-list values in lists.
+        if not self._sso_handler.check_required_attributes(
+            request,
+            {k: v if isinstance(v, list) else [v] for k, v in userinfo.items()},
+            self._oidc_attribute_requirements,
+        ):
+            return
 
         # Call the mapper to register/login the user
         try:

tests/handlers/test_oidc.py

@@ -989,6 +989,138 @@ def test_null_localpart(self):
         self.get_success(_make_callback_with_userinfo(self.hs, userinfo))
         self.assertRenderedError("mapping_error", "localpart is invalid: ")
 
+    @override_config(
+        {
+            "oidc_config": {
+                **DEFAULT_CONFIG,
+                "attribute_requirements": [{"attribute": "test", "value": "foobar"}],
+            }
+        }
+    )
+    def test_attribute_requirements(self):
+        """The required attributes must be met from the OIDC userinfo response."""
+        auth_handler = self.hs.get_auth_handler()
+        auth_handler.complete_sso_login = simple_async_mock()
+
+        # userinfo lacking "test": "foobar" attribute should fail.
+        userinfo = {
+            "sub": "tester",
+            "username": "tester",
+        }
+        self.get_success(_make_callback_with_userinfo(self.hs, userinfo))
+        auth_handler.complete_sso_login.assert_not_called()
+
+        # userinfo with "test": "foobar" attribute should succeed.
+        userinfo = {
+            "sub": "tester",
+            "username": "tester",
+            "test": "foobar",
+        }
+        self.get_success(_make_callback_with_userinfo(self.hs, userinfo))
+
+        # check that the auth handler got called as expected
+        auth_handler.complete_sso_login.assert_called_once_with(
+            "@tester:test", "oidc", ANY, ANY, None, new_user=True
+        )
+
+    @override_config(
+        {
+            "oidc_config": {
+                **DEFAULT_CONFIG,
+                "attribute_requirements": [{"attribute": "test", "value": "foobar"}],
+            }
+        }
+    )
+    def test_attribute_requirements_contains(self):
+        """Test that auth succeeds if userinfo attribute CONTAINS required value"""
+        auth_handler = self.hs.get_auth_handler()
+        auth_handler.complete_sso_login = simple_async_mock()
+        # userinfo with "test": ["foobar", "foo", "bar"] attribute should succeed.
+        userinfo = {
+            "sub": "tester",
+            "username": "tester",
+            "test": ["foobar", "foo", "bar"],
+        }
+        self.get_success(_make_callback_with_userinfo(self.hs, userinfo))
+
+        # check that the auth handler got called as expected
+        auth_handler.complete_sso_login.assert_called_once_with(
+            "@tester:test", "oidc", ANY, ANY, None, new_user=True
+        )
+
+    @override_config(
+        {
+            "oidc_config": {
+                **DEFAULT_CONFIG,
+                "attribute_requirements": [{"attribute": "test", "value": "foobar"}],
+            }
+        }
+    )
+    def test_attribute_requirements_mismatch(self):
+        """
+        Test that auth fails if attributes exist but don't match,
+        or are non-string values.
+        """
+        auth_handler = self.hs.get_auth_handler()
+        auth_handler.complete_sso_login = simple_async_mock()
+        # userinfo with "test": "not_foobar" attribute should fail
+        userinfo = {
+            "sub": "tester",
+            "username": "tester",
+            "test": "not_foobar",
+        }
+        self.get_success(_make_callback_with_userinfo(self.hs, userinfo))
+        auth_handler.complete_sso_login.assert_not_called()
+
+        # userinfo with "test": ["foo", "bar"] attribute should fail
+        userinfo = {
+            "sub": "tester",
+            "username": "tester",
+            "test": ["foo", "bar"],
+        }
+        self.get_success(_make_callback_with_userinfo(self.hs, userinfo))
+        auth_handler.complete_sso_login.assert_not_called()
+
+        # userinfo with "test": False attribute should fail
+        # this is largely just to ensure we don't crash here
+        userinfo = {
+            "sub": "tester",
+            "username": "tester",
+            "test": False,
+        }
+        self.get_success(_make_callback_with_userinfo(self.hs, userinfo))
+        auth_handler.complete_sso_login.assert_not_called()
+
+        # userinfo with "test": None attribute should fail
+        # a value of None breaks the OIDC spec, but it's important to not crash here
+        userinfo = {
+            "sub": "tester",
+            "username": "tester",
+            "test": None,
+        }
+        self.get_success(_make_callback_with_userinfo(self.hs, userinfo))
+        auth_handler.complete_sso_login.assert_not_called()
+
+        # userinfo with "test": 1 attribute should fail
+        # this is largely just to ensure we don't crash here
+        userinfo = {
+            "sub": "tester",
+            "username": "tester",
+            "test": 1,
+        }
+        self.get_success(_make_callback_with_userinfo(self.hs, userinfo))
+        auth_handler.complete_sso_login.assert_not_called()
+
+        # userinfo with "test": 3.14 attribute should fail
+        # this is largely just to ensure we don't crash here
+        userinfo = {
+            "sub": "tester",
+            "username": "tester",
+            "test": 3.14,
+        }
+        self.get_success(_make_callback_with_userinfo(self.hs, userinfo))
+        auth_handler.complete_sso_login.assert_not_called()
+
     def _generate_oidc_session_token(
         self,
         state: str,