Add SSO attribute requirements for OIDC providers (#9609)

Allows limiting who can login using OIDC via the claims
made from the IdP.
pull/9631/head
Hubbe 2021-03-16 17:46:07 +02:00 committed by GitHub
parent 8000cf1315
commit dd5e5dc1d6
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 209 additions and 1 deletions

1
changelog.d/9609.feature Normal file
View File

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

View File

@ -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
#

View File

@ -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 @@ class OIDCConfig(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.
#
@ -223,6 +242,9 @@ class OIDCConfig(Config):
# 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 @@ class OIDCConfig(Config):
# client_id: "synapse"
# client_secret: "copy secret generated in Keycloak UI"
# scopes: ["openid", "profile"]
# attribute_requirements:
# - attribute: groups
# value: "admin"
# For use with Github
#
@ -329,6 +354,10 @@ OIDC_PROVIDER_CONFIG_SCHEMA = {
},
"allow_existing_users": {"type": "boolean"},
"user_mapping_provider": {"type": ["object", "null"]},
"attribute_requirements": {
"type": "array",
"items": SsoAttributeRequirement.JSON_SCHEMA,
},
},
}
@ -465,6 +494,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,
@ -488,6 +522,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,
)
@ -577,3 +612,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])

View File

@ -280,6 +280,7 @@ class OidcProvider:
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
@ -859,6 +860,18 @@ class OidcProvider:
)
# 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:

View File

@ -989,6 +989,138 @@ class OidcHandlerTestCase(HomeserverTestCase):
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,