Auth OIDC

qiita_core/configuration_manager.py

@@ -126,6 +126,32 @@ class ConfigurationManager(object):
         The email address a user should write to when asking for help
     sysadmin_email : str
         The email address, Qiita sends internal notifications to a sys admin
+    None (=internal user authentication) or one or several 'oidc_' sections
+    to use external identity providers (IdP) with following values:
+    client_id : str
+        The name you registered Qiita with at the external IdP
+    client_secret : str
+        A secret string with which Qiita identifies at the external IdP (not
+        all IdPs need a secret)
+    redirect_endpoint : str
+        The internal Qiita endpoint the IdP shall redirect the user after
+        logging in
+    wellknown_uri : str
+        The URL of the well-known json document, specifying how API end points
+        like 'authorize', 'token' or 'userinfo' are defined. See e.g.
+        https://swagger.io/docs/specification/authentication/
+            openid-connect-discovery/
+    label : str
+        A speaking label for the Identity Provider
+    scope : str
+        The scope, i.e. fields about a user, which Qiita requests from the
+        Identity Provider, e.g. "profile email eduperson_orcid".
+        Will be automatically extended by the scope "openid", to enable the
+        "authorize_code" OIDC flow.
+    logo : str
+        Optional. Name of a file in qiita_pet/static/img that shall be
+        displayed for login through Service Provider, instead of a plain
+        button
 
     Raises
     ------
@@ -160,6 +186,7 @@ def __init__(self):
         self._get_vamps(config)
         self._get_portal(config)
         self._iframe(config)
+        self._get_oidc(config)
 
     def _get_main(self, config):
         """Get the configuration of the main section"""
@@ -413,3 +440,43 @@ def _get_portal(self, config):
 
     def _iframe(self, config):
         self.iframe_qiimp = config.get("iframe", "QIIMP", fallback=None)
+
+    def _get_oidc(self, config):
+        """Get the configuration of the open ID connect section(s)
+           User can provide multiple sections with naming schema oidc_foo where
+           foo is the name of an Identity Provider - Qiita can handle multiple
+           Identity Providers simultaneously.
+           """
+        PREFIX = 'oidc_'
+        self.oidc = dict()
+        for section_name in config.sections():
+            if section_name.startswith(PREFIX):
+                provider = dict()
+                provider['client_id'] = config.get(
+                    section_name, 'CLIENT_ID', fallback=None)
+                provider['client_secret'] = config.get(
+                    section_name, 'CLIENT_SECRET', fallback=None)
+                provider['redirect_endpoint'] = config.get(
+                    section_name, 'REDIRECT_ENDPOINT')
+                if provider['redirect_endpoint']:
+                    if not provider['redirect_endpoint'].startswith('/'):
+                        provider['redirect_endpoint'] = '/%s' % provider[
+                            'redirect_endpoint']
+                    if provider['redirect_endpoint'].endswith('/'):
+                        provider['redirect_endpoint'] = provider[
+                            'redirect_endpoint'][:-1]
+                provider['wellknown_uri'] = config.get(
+                    section_name, 'WELLKNOWN_URI')
+                provider['label'] = config.get(section_name, 'LABEL')
+                if not provider['label']:
+                    # fallback, if no label is provided
+                    provider['label'] = section_name[len(PREFIX):]
+                self.oidc[section_name[len(PREFIX):]] = provider
+                provider['scope'] = config.get(
+                    section_name, 'SCOPE', fallback=None)
+                if not provider['scope']:
+                    provider['scope'] = 'openid'
+                if 'openid' not in provider['scope']:
+                    provider['scope'] = 'openid %s' % provider['scope']
+                provider['logo'] = config.get(
+                    section_name, 'LOGO', fallback=None)

qiita_core/support_files/config_test.cfg

@@ -193,3 +193,68 @@ STATS_MAP_CENTER_LONGITUDE =
 # On May 2024, we removed QIIMP from the code base but we will leave this
 # section in case we need to add access to another iframe in the future; note
 # that the qiita-terms are also accessed via iframe but this is internal
+
+# --------------------- External Identity Provider settings --------------------
+# user authentication happens per default within Qiita, i.e. when a user logs in,
+# the stored password hash and email address is compared against what a user
+# just provided. You might however, use an external identity provider (IdP) to
+# authenticate the user like
+#    google: https://developers.google.com/identity/protocols/oauth2 or
+#    github: https://docs.github.com/en/apps/oauth-apps/building-oauth-apps/authorizing-oauth-apps or
+#    self hosted keycloak: https://www.keycloak.org/
+# Thus, you don't have to deal with user verification, reset passwords, ...
+# Authorization (i.e. if the authorized user is allowed to use Qiita or which
+# user level he/she gets assigned is an independent process. You can even use
+# multiple independent external identity providers!
+# Qiita currently only support the "open ID connect" protocol with the implicit flow.
+# Each identity provider comes as its own config section [oidc_foo] and needs
+# to specify the following five fields:
+#
+# Typical identity provider manage multiple "realms" and specific "clients" per realm
+# You need to contact your IdP and register Qiita as a new "client". The IdP will
+# provide you with the correct values.
+#
+# The authorization protocol requires three steps to obtain user information:
+# 1) you identify as the correct client and ask the IdP for a request code
+#    You have to forward the user to the login page of your IdP. To let the IdP
+#    know how to come back to Qiita, you need to provide a redirect URL
+# 2) you exchange the code for a user token
+# 3) you obtain information about the user for the obtaines user token
+# Typically, each step is implemented as a separate URL endpoint
+#
+# To activate IdP: comment out the following config section
+
+#[oidc_academicid]
+#
+## client ID for Qiita as registered at your Identity Provider of choice
+#CLIENT_ID = gi-qiita-prod
+#
+## client secret to verify Qiita as the correct client. Not all IdPs require
+## a client secret!
+#CLIENT_SECRET = verySecretString
+#
+## redirect URL (end point in your Qiita instance), to which the IdP redirects
+## after user types in his/her credentials. If you don't want to change code in
+## qiita_pet/webserver.py the URL must follow the pattern:
+## base_URL/auth/login_OIDC/foo where foo is the name of this config section
+## without the oidc_ prefix!
+#REDIRECT_ENDPOINT = /auth/login_OIDC/localkeycloak
+#
+## The URL of the well-known json document, specifying how API end points
+## like 'authorize', 'token' or 'userinfo' are defined. See e.g.
+## https://swagger.io/docs/specification/authentication/
+##    openid-connect-discovery/
+#WELLKNOWN_URI = https://keycloak.sso.gwdg.de/.well-known/openid-configuration
+#
+## a speaking label for the Identity Provider. Section name is used if empty.
+#LABEL = GWDG Academic Cloud
+#
+## The scope, i.e. fields about a user, which Qiita requests from the
+## Identity Provider, e.g. "profile email eduperson_orcid".
+## Will be automatically extended by the scope "openid", to enable the
+## "authorize_code" OIDC flow.
+#SCOPE = openid
+#
+##Optional. Name of a file in qiita_pet/static/img that shall be
+##displayed for login through Service Provider, instead of a plain button
+#LOGO = oidc_lifescienceAAI.png

qiita_core/tests/test_configuration_manager.py

@@ -139,7 +139,7 @@ def test_get_main(self):
         with warnings.catch_warnings(record=True) as warns:
             obs._get_main(self.conf)
 
-            obs_warns = [str(w.message) for w in warns]
+            obs_warns = [str(w.message) for w in warns if w.category is UserWarning]
             exp_warns = [
                 "Random cookie secret generated.",
                 "Random JWT secret generated.  Non Public Artifact "
@@ -293,6 +293,55 @@ def test_get_portal_latlong(self):
         obs._get_portal(self.conf)
         self.assertEqual(obs.stats_map_center_longitude, -105.24827)
 
+    def test_get_oidc(self):
+        SECTION_NAME = 'oidc_academicid'
+        obs = ConfigurationManager()
+        self.assertTrue(len(obs.oidc), 1)
+        self.assertTrue(obs.oidc.keys(), [SECTION_NAME])
+
+        # assert endpoint starts with /
+        self.conf.set(SECTION_NAME, 'REDIRECT_ENDPOINT', 'auth/something')
+        obs._get_oidc(self.conf)
+        self.assertEqual(obs.oidc['academicid']['redirect_endpoint'],
+                         '/auth/something')
+
+        # assert endpoint does not end with /
+        self.conf.set(SECTION_NAME, 'REDIRECT_ENDPOINT', 'auth/something/')
+        obs._get_oidc(self.conf)
+        self.assertEqual(obs.oidc['academicid']['redirect_endpoint'],
+                         '/auth/something')
+
+        self.conf.set(SECTION_NAME, 'CLIENT_ID', 'foo')
+        obs._get_oidc(self.conf)
+        self.assertEqual(obs.oidc['academicid']['client_id'], "foo")
+
+        self.assertTrue('gwdg.de' in obs.oidc['academicid']['wellknown_uri'])
+
+        self.assertEqual(obs.oidc['academicid']['label'],
+                         'GWDG Academic Cloud')
+        # test fallback, if no label is provided
+        self.conf.set(SECTION_NAME, 'LABEL', '')
+        obs._get_oidc(self.conf)
+        self.assertEqual(obs.oidc['academicid']['label'], 'academicid')
+
+        self.assertEqual(obs.oidc['academicid']['scope'], 'openid')
+        # test fallback, if no scope is provided
+        self.conf.set(SECTION_NAME, 'SCOPE', '')
+        obs._get_oidc(self.conf)
+        self.assertEqual(obs.oidc['academicid']['scope'], 'openid')
+
+        # test if scope will be automatically extended with 'openid'
+        self.conf.set(SECTION_NAME, 'SCOPE', 'email affiliation')
+        obs._get_oidc(self.conf)
+        self.assertTrue('openid' in obs.oidc['academicid']['scope'].split())
+
+        self.assertEqual(obs.oidc['academicid']['logo'],
+                         'oidc_lifescienceAAI.png')
+        # test fallback, if no scope is provided
+        self.conf.remove_option(SECTION_NAME, 'LOGO')
+        obs._get_oidc(self.conf)
+        self.assertEqual(obs.oidc['academicid']['logo'], None)
+
 
 CONF = """
 # ------------------------------ Main settings --------------------------------
@@ -472,6 +521,43 @@ def test_get_portal_latlong(self):
 
 # ----------------------------- iframes settings ---------------------------
 [iframe]
+
+
+# ------------------- External Identity Provider settings ------------------
+[oidc_academicid]
+
+# client ID for Qiita as registered at your Identity Provider of choice
+CLIENT_ID = gi-qiita-prod
+
+# client secret to verify Qiita as the correct client. Not all IdPs require
+# a client secret.
+CLIENT_SECRET = verySecretString
+
+# redirect URL (end point in your Qiita instance), to which the IdP redirects
+# after user types in his/her credentials. If you don't want to change code in
+# qiita_pet/webserver.py the URL must follow the pattern:
+# base_URL/auth/login_OIDC/foo where foo is the name of this config section
+# without the oidc_ prefix!
+REDIRECT_ENDPOINT = /auth/login_OIDC/academicid
+
+# The URL of the well-known json document, specifying how API end points
+# like 'authorize', 'token' or 'userinfo' are defined. See e.g.
+# https://swagger.io/docs/specification/authentication/
+#    openid-connect-discovery/
+WELLKNOWN_URI = https://keycloak.sso.gwdg.de/.well-known/openid-configuration
+
+# a speaking label for the Identity Provider. Section name is used if empty.
+LABEL = GWDG Academic Cloud
+
+# The scope, i.e. fields about a user, which Qiita requests from the
+# Identity Provider, e.g. "profile email eduperson_orcid".
+# Will be automatically extended by the scope "openid", to enable the
+# "authorize_code" OIDC flow.
+SCOPE = openid
+
+# Optional. Name of a file in qiita_pet/static/img that shall be
+# displayed for login through Service Provider, instead of a plain button
+LOGO = oidc_lifescienceAAI.png
 """
 
 if __name__ == "__main__":

qiita_db/user.py

@@ -249,6 +249,54 @@ def create(cls, email, password, info=None):
 
             return cls(email)
 
+    @classmethod
+    def create_oidc(cls, email, user_info, idp):
+        """Creates a new user with information obtained from an external
+           identity provider
+
+        Parameters
+        ----------
+        email : str
+            The user's email fetched from the User Info of the Identity
+            Provider upon successful authentication.
+        user_info : dict
+            User information provided by the external identity provider
+        idp : str
+            The label of the external identity provider as set in config file
+
+        Raises
+        ------
+        IncorrectEmailError
+            Email string given is not a valid email
+        """
+        if not validate_email(email):
+            raise IncorrectEmailError("Bad email given: %s" % email)
+        info = {}
+        # email and password are minimal needed information, password indicates
+        # OIDC user registration purely for admins
+        info['email'] = email
+        info['password'] = "not_necessary_due_to_OIDC"
+        # verify code is necessary to manually authorize users on the admin
+        # page
+        info['user_verify_code'] = idp
+        # check if we got useful information from the IdP
+        if 'name' in user_info.keys():
+            info['name'] = user_info['name']
+
+        qdb.util.check_table_cols(info, cls._table)
+        columns = info.keys()
+        values = [info[col] for col in columns]
+
+        # create user
+        sql = "INSERT INTO qiita.{0} ({1}) VALUES ({2})".format(
+            cls._table, ','.join(columns), ','.join(['%s'] * len(values)))
+
+        qdb.sql_connection.TRN.add(sql, values)
+
+        qdb.sql_connection.TRN.execute()
+
+        return cls(email)
+
     @classmethod
     def verify_code(cls, email, code, code_type):
         """Verify that a code and email match