add configurable attribute to use as username from OIDC tokens instea…

…d of hardcoding to email

docs/configuration/index.rst

@@ -56,6 +56,7 @@ Define parameters for the authentication:
 - oidc_conf_well_known_url: OIDC well-known URL for metadata
 - oidc_client_secret: The client secret for OIDC
 - oidc_client_id: The client_id for OIDC
+- oidc_username_attribute: What attribute in access token or userinfo endpoint to use for username, defaults to "email"
 - frontend_callback_url: The frontend URL that the OIDC client should redirect to after the login process
 - oidc_enabled: Set True to enabled OIDC login. Defaults to False
 - audience: The string to verify the aud attribute in the access token with

src/cnaas_nms/api/app.py

@@ -190,7 +190,7 @@ def socketio_on_connect():
     if auth_settings.OIDC_ENABLED:
         try:
             token = oauth_required.get_token_validator("bearer").authenticate_token(token_string)
-            user = get_oauth_token_info(token)["email"]
+            user = get_oauth_token_info(token)[auth_settings.OIDC_USERNAME_ATTRIBUTE]
         except InvalidTokenError as e:
             logger.debug("InvalidTokenError: " + format(e))
             return False
@@ -239,8 +239,11 @@ def log_request(response):
                 token_string = request.headers.get("Authorization").split(" ")[-1]
                 token = oauth_required.get_token_validator("bearer").authenticate_token(token_string)
                 token_info = get_oauth_token_info(token)
-                if "email" in token_info:
-                    user = "User: {} (email), ".format(get_oauth_token_info(token)["email"])
+                if auth_settings.OIDC_USERNAME_ATTRIBUTE in token_info:
+                    user = "User: {} ({}), ".format(
+                        get_oauth_token_info(token)[auth_settings.OIDC_USERNAME_ATTRIBUTE],
+                        auth_settings.OIDC_USERNAME_ATTRIBUTE,
+                    )
                 elif "client_id" in token_info:
                     user = "User: {} (client_id), ".format(get_oauth_token_info(token)["client_id"])
                 else:

src/cnaas_nms/api/auth.py

@@ -83,14 +83,13 @@ def get(self):
         url = auth_settings.FRONTEND_CALLBACK_URL
         parameters = {"token": token["access_token"]}
 
-        if "userinfo" in token and "email" in token["userinfo"]:
-            parameters["email"] = token["userinfo"]["email"]
+        if "userinfo" in token and auth_settings.OIDC_USERNAME_ATTRIBUTE in token["userinfo"]:
+            parameters["username"] = token["userinfo"][auth_settings.OIDC_USERNAME_ATTRIBUTE]
 
         req = PreparedRequest()
         req.prepare_url(url, parameters)
         resp = redirect(req.url, code=302)
         if "refresh_token" in token:
-            # TODO: set secure true when in production
             resp.set_cookie(
                 "REFRESH_TOKEN",
                 token["refresh_token"],

src/cnaas_nms/app_settings.py

@@ -73,6 +73,7 @@ class AuthSettings(BaseSettings):
     OIDC_CLIENT_SECRET: str = "xxx"
     OIDC_CLIENT_ID: str = "client-id"
     OIDC_ENABLED: bool = False
+    OIDC_USERNAME_ATTRIBUTE: str = "email"
     PERMISSIONS: Optional[PermissionsModel] = None
     PERMISSIONS_DISABLED: bool = False
     OIDC_CLIENT_SCOPE: str = "openid"
@@ -172,6 +173,9 @@ def construct_auth_settings() -> AuthSettings:
         auth_settings.OIDC_CLIENT_SECRET = config.get("oidc_client_secret", AuthSettings().OIDC_CLIENT_SECRET)
         auth_settings.OIDC_CLIENT_ID = config.get("oidc_client_id", AuthSettings().OIDC_CLIENT_ID)
         auth_settings.OIDC_CLIENT_SCOPE = config.get("oidc_client_scope", AuthSettings().OIDC_CLIENT_SCOPE)
+        auth_settings.OIDC_USERNAME_ATTRIBUTE = config.get(
+            "oidc_username_attribute", AuthSettings().OIDC_USERNAME_ATTRIBUTE
+        )
         auth_settings.PERMISSIONS_DISABLED = config.get("permissions_disabled", AuthSettings().PERMISSIONS_DISABLED)
         auth_settings.AUDIENCE = config.get("audience", AuthSettings().AUDIENCE)
         auth_settings.VERIFY_AUDIENCE = config.get("verify_audience", AuthSettings().VERIFY_AUDIENCE)

src/cnaas_nms/tools/security.py

@@ -254,28 +254,29 @@ def validate_token(self, token, scopes, request: OAuth2Request):
 
 
 def get_oauth_identity() -> str:
-    """Give back the email address of the OAUTH account
+    """Give back the username of the OAUTH account
 
     If JWT is disabled, we return "admin".
 
     We do an api call to request userinfo. This gives back all the userinfo.
     We get the right info from there and return this to the user.
 
     Returns:
-        email(str): Email of the logged in user
+        username(str): Username of the logged in user
 
     """
     # For now unnecersary, useful when we only use one log in method
     if not auth_settings.OIDC_ENABLED:
         return "Admin"
     token_info = get_oauth_token_info(current_token)
-    if "email" in token_info:
-        return token_info["email"]
+    if auth_settings.OIDC_USERNAME_ATTRIBUTE in token_info:
+        return token_info[auth_settings.OIDC_USERNAME_ATTRIBUTE]
     elif "client_id" in token_info:
         return token_info["client_id"]
     else:
-        logger.error("Email or client_id is a required claim for oauth")
-        raise KeyError("Email or client_id is a required claim for oauth")
+        error_message = "{} or client_id is a required claim for oauth".format(auth_settings.OIDC_USERNAME_ATTRIBUTE)
+        logger.error(error_message)
+        raise KeyError(error_message)
 
 
 # check which method we use to log in and load vars needed for that