Merge pull request #6803 from atsareg/dev-ce-tokens-4

[8.0] Pilot submission with tokens

Author: fstagni

dirac.cfg

@@ -187,7 +187,7 @@ Registry
       VOMSName = lhcb
 
       # Registered identity provider associated with VO
-      IdP = CheckIn
+      IdProvider = CheckIn
 
       # Section to describe all the VOMS servers that can be used with the given VOMS VO
       VOMSServers

src/DIRAC/ConfigurationSystem/Client/Helpers/Registry.py

@@ -163,23 +163,23 @@ def findDefaultGroupForDN(dn):
     return findDefaultGroupForUser(result["Value"])
 
 
-def findDefaultGroupForUser(userName):
+def findDefaultGroupForUser(username):
     """Get default group for user
 
-    :param str userName: user name
+    :param str username: user name
 
     :return: S_OK(str)/S_ERROR()
     """
-    defGroups = getUserOption(userName, "DefaultGroup", [])
+    defGroups = getUserOption(username, "DefaultGroup", [])
     defGroups += gConfig.getValue(f"{gBaseRegistrySection}/DefaultGroup", ["user"])
-    result = getGroupsForUser(userName)
+    result = getGroupsForUser(username)
     if not result["OK"]:
         return result
     userGroups = result["Value"]
     for group in defGroups:
         if group in userGroups:
             return S_OK(group)
-    return S_OK(userGroups[0]) if userGroups else S_ERROR(f"User {userName} has no groups")
+    return S_OK(userGroups[0]) if userGroups else S_ERROR(f"User {username} has no groups")
 
 
 def getAllUsers():
@@ -349,16 +349,16 @@ def hostHasProperties(hostName, propList):
     return __matchProps(propList, getPropertiesForHost(hostName))
 
 
-def getUserOption(userName, optName, defaultValue=""):
+def getUserOption(username, optName, defaultValue=""):
     """Get user option
 
-    :param str userName: user name
+    :param str username: user name
     :param str optName: option name
     :param defaultValue: default value
 
     :return: defaultValue or str
     """
-    return gConfig.getValue(f"{gBaseRegistrySection}/Users/{userName}/{optName}", defaultValue)
+    return gConfig.getValue(f"{gBaseRegistrySection}/Users/{username}/{optName}", defaultValue)
 
 
 def getGroupOption(groupName, optName, defaultValue=""):
@@ -430,7 +430,7 @@ def getIdPForGroup(group):
 
     :return: str
     """
-    return getVOOption(getVOForGroup(group), "IdP")
+    return getVOOption(getVOForGroup(group), "IdProvider")
 
 
 def getDefaultVOMSAttribute():

src/DIRAC/ConfigurationSystem/Client/Helpers/Resources.py

@@ -228,8 +228,17 @@ def getQueue(site, ce, queue):
     return S_OK(resultDict)
 
 
-def getQueues(siteList=None, ceList=None, ceTypeList=None, community=None):
-    """Get CE/queue options according to the specified selection"""
+def getQueues(siteList=None, ceList=None, ceTypeList=None, community=None, tags=None):
+    """Get CE/queue options according to the specified selection
+
+    :param str list siteList: sites to be selected
+    :param str list ceList: CEs to be selected
+    :param str list ceTypeList: CE types to be selected
+    :param str community: selected VO
+    :param str list tags: tags required for selection
+
+    :return: S_OK/S_ERROR with Value - dictionary of selected Site/CE/Queue parameters
+    """
 
     result = gConfig.getSections("/Resources/Sites")
     if not result["OK"]:
@@ -259,6 +268,7 @@ def getQueues(siteList=None, ceList=None, ceTypeList=None, community=None):
                 continue
             ces = result["Value"]
             for ce in ces:
+                ceTags = gConfig.getValue(f"/Resources/Sites/{grid}/{site}/CEs/{ce}/Tag", [])
                 if ceTypeList:
                     ceType = gConfig.getValue(f"/Resources/Sites/{grid}/{site}/CEs/{ce}/CEType", "")
                     if not ceType or ceType not in ceTypeList:
@@ -283,6 +293,11 @@ def getQueues(siteList=None, ceList=None, ceTypeList=None, community=None):
                         comList = gConfig.getValue(f"/Resources/Sites/{grid}/{site}/CEs/{ce}/Queues/{queue}/VO", [])
                         if comList and community.lower() not in [cl.lower() for cl in comList]:
                             continue
+                    if tags:
+                        queueTags = gConfig.getValue(f"/Resources/Sites/{grid}/{site}/CEs/{ce}/Queues/{queue}/Tag", [])
+                        queueTags = set(ceTags + queueTags)
+                        if not queueTags or not set(tags).issubset(queueTags):
+                            continue
                     resultDict.setdefault(site, {})
                     resultDict[site].setdefault(ce, ceOptionsDict)
                     resultDict[site][ce].setdefault("Queues", {})

src/DIRAC/Core/Tornado/Server/private/BaseRequestHandler.py

@@ -14,7 +14,6 @@
 from functools import partial
 
 import jwt
-import tornado
 from tornado.web import RequestHandler, HTTPError
 from tornado.ioloop import IOLoop
 

src/DIRAC/Core/Utilities/Grid.py

@@ -11,7 +11,7 @@
 from DIRAC.Core.Utilities.Subprocess import systemCall, shellCall
 
 
-def executeGridCommand(proxy, cmd, gridEnvScript=None):
+def executeGridCommand(proxy, cmd, gridEnvScript=None, gridEnvDict=None):
     """
     Execute cmd tuple after sourcing GridEnv
     """
@@ -53,6 +53,9 @@ def executeGridCommand(proxy, cmd, gridEnvScript=None):
             return ret
         gridEnv["X509_USER_PROXY"] = ret["Value"]
 
+    if gridEnvDict:
+        gridEnv.update(gridEnvDict)
+
     result = systemCall(120, cmd, env=gridEnv)
     return result
 

src/DIRAC/FrameworkSystem/API/AuthHandler.py

@@ -420,12 +420,12 @@ def post_token(self):
         return self.server.create_token_response(self.request)
 
     def __researchDIRACGroup(self, extSession, chooseScope, state):
-        """Research DIRAC groups for authorized user
+        """Look for DIRAC groups of a user already authorized by an Identity Provider
 
         :param dict extSession: ended authorized external IdP session
 
         :return: -- will return (None, response) to provide error or group selector
-                    will return (grant_user, request) to contionue authorization with choosed group
+                    will return (grant_user, request) to continue authorization with chosen group
         """
         # Base DIRAC client auth session
         firstRequest = createOAuth2Request(extSession["firstRequest"])
@@ -435,9 +435,13 @@ def __researchDIRACGroup(self, extSession, chooseScope, state):
         username = extSession["authed"]["username"]
         # Requested arguments in first request
         provider = firstRequest.provider
-        self.log.debug(f"Next groups has been found for {username}:", ", ".join(firstRequest.groups))
+        self.log.debug("The following groups found", f"for {username}: {', '.join(firstRequest.groups)}")
 
-        # Researche Group
+        # If group is already defined in the first request, just return it as it was already validated
+        if firstRequest.groups:
+            return extSession["authed"], firstRequest
+
+        # Look for DIRAC groups valid for the user
         result = getGroupsForUser(username)
         if not result["OK"]:
             return None, self.server.handle_response(
@@ -460,16 +464,12 @@ def __researchDIRACGroup(self, extSession, chooseScope, state):
 
         self.log.debug(f"The state of {username} user groups has been checked:", pprint.pformat(validGroups))
 
-        # If group already defined in first request, just return it
-        if firstRequest.groups:
-            return extSession["authed"], firstRequest
-
         # If not and we found only one valid group, apply this group
         if len(validGroups) == 1:
             firstRequest.addScopes([f"g:{validGroups[0]}"])
             return extSession["authed"], firstRequest
 
-        # Else give user chanse to choose group in browser
+        # Else give user a chance to choose a group in the browser
         with dom.div(cls="row mt-5 justify-content-md-center align-items-center") as tag:
             for group in sorted(validGroups):
                 vo, gr = group.split("_")

src/DIRAC/FrameworkSystem/Client/TokenManagerClient.py

@@ -7,6 +7,7 @@
 from DIRAC.Core.Base.Client import Client, createClient
 from DIRAC.ConfigurationSystem.Client.Helpers import Registry
 from DIRAC.Resources.IdProvider.IdProviderFactory import IdProviderFactory
+from DIRAC.FrameworkSystem.private.authorization.utils.Tokens import OAuth2Token
 
 gTokensSync = ThreadSafe.Synchronizer()
 
@@ -33,7 +34,7 @@ def getToken(
         identityProvider: str = None,
         requiredTimeLeft: int = 0,
     ):
-        """Get an access token for a user/group.
+        """Get an access token for a user/group keeping the local cache
 
         :param username: user name
         :param userGroup: group name
@@ -55,9 +56,9 @@ def getToken(
             return result
         idpObj = result["Value"]
 
-        if userGroup and (result := idpObj.getGroupScopes(userGroup))["OK"]:
+        if userGroup and (result := idpObj.getGroupScopes(userGroup)):
             # What scope correspond to the requested group?
-            scope = list(set((scope or []) + result["Value"]))
+            scope = list(set((scope or []) + result))
 
         # Set the scope
         idpObj.scope = " ".join(scope)
@@ -70,21 +71,20 @@ def getToken(
             # Let's check if the access token is fresh
             if not token.is_expired(requiredTimeLeft):
                 return S_OK(token)
-            # It seems that it is no longer valid for us, but whether there is a refresh token?
-            if token.get("refresh_token"):
-                # Okay, so we can try to refresh tokens
-                if (result := idpObj.refreshToken(token["refresh_token"]))["OK"]:
-                    # caching new tokens
-                    self.__tokensCache.add(
-                        cacheKey,
-                        token.get_claim("exp", "refresh_token") or self.DEFAULT_RT_EXPIRATION_TIME,
-                        result["Value"],
-                    )
-                    return result
-                self.log.verbose(f"Failed to get token on client's side: {result['Message']}")
-                # Let's try to revoke broken token
-                idpObj.revokeToken(token["refresh_token"])
-
-        return self.executeRPC(
+
+        result = self.executeRPC(
             username, userGroup, scope, audience, identityProvider, requiredTimeLeft, call="getToken"
         )
+
+        if result["OK"]:
+            token = OAuth2Token(dict(result["Value"]))
+            self.__tokensCache.add(
+                cacheKey,
+                token.get_claim("exp", "refresh_token") or self.DEFAULT_RT_EXPIRATION_TIME,
+                token,
+            )
+
+        return result
+
+
+gTokenManager = TokenManagerClient()

src/DIRAC/FrameworkSystem/DB/TokenDB.py

@@ -21,7 +21,7 @@
 
 
 class Token(Model, OAuth2TokenMixin):
-    """This class describe token fields"""
+    """This class describes token fields"""
 
     __tablename__ = "Token"
     __table_args__ = {"mysql_engine": "InnoDB", "mysql_charset": "utf8"}
@@ -30,8 +30,8 @@ class Token(Model, OAuth2TokenMixin):
     # https://stackoverflow.com/questions/1827063/mysql-error-key-specification-without-a-key-length
     id = Column(Integer, autoincrement=True, primary_key=True)  # Unique token ID
     kid = Column(String(255))  # Unique secret key ID for token encryption
-    user_id = Column(String(255))  # User identificator that registred in an identity provider, token owner
-    provider = Column(String(255))  # Provider name registred in DIRAC
+    user_id = Column(String(255))  # User identifier registered in an identity provider, token owner
+    provider = Column(String(255))  # Provider name registered in DIRAC
     expires_at = Column(Integer, nullable=False, default=0)  # When the access token is expired
     access_token = Column(Text, nullable=False)
     refresh_token = Column(Text, nullable=False)
@@ -101,7 +101,7 @@ def updateToken(self, token: dict, userID: str, provider: str, rt_expired_in: in
         :return: S_OK(list)/S_ERROR() -- return old tokens that should be revoked.
         """
         if not token["refresh_token"]:
-            return S_ERROR("Cannot store absent refresh token.")
+            return S_ERROR("Cannot store token without a refresh token.")
 
         # Let's collect the necessary attributes of the token
         token["user_id"] = userID

src/DIRAC/FrameworkSystem/Service/TokenManagerHandler.py

@@ -190,7 +190,7 @@ def export_getToken(
             * LimitedDelegation <- permits downloading only limited tokens
             * PrivateLimitedDelegation <- permits downloading only limited tokens for one self
 
-        :paarm username: user name
+        :param username: user name
         :param userGroup: user group
         :param scope: requested scope
         :param audience: requested audience
@@ -210,38 +210,24 @@ def export_getToken(
             return result
         idpObj = result["Value"]
 
-        if userGroup and (result := idpObj.getGroupScopes(userGroup))["OK"]:
+        if userGroup and (result := idpObj.getGroupScopes(userGroup)):
             # What scope correspond to the requested group?
-            scope = list(set((scope or []) + result["Value"]))
+            scope = list(set((scope or []) + result))
 
         # Set the scope
         idpObj.scope = " ".join(scope)
 
-        # Let's check if there are corresponding tokens in the cache
+        # Let's check if there is a corresponding token in the cache
         cacheKey = (username, idpObj.scope, audience, identityProvider)
         if self.__tokensCache.exists(cacheKey, requiredTimeLeft):
             # Well we have a fresh record containing a Token object
             token = self.__tokensCache.get(cacheKey)
             # Let's check if the access token is fresh
             if not token.is_expired(requiredTimeLeft):
                 return S_OK(token)
-            # It seems that it is no longer valid for us, but whether there is a refresh token?
-            if token.get("refresh_token"):
-                # Okay, so we can try to refresh tokens
-                if (result := idpObj.refreshToken(token["refresh_token"]))["OK"]:
-                    # caching new tokens
-                    self.__tokensCache.add(
-                        cacheKey,
-                        result["Value"].get_claim("exp", "refresh_token") or self.DEFAULT_RT_EXPIRATION_TIME,
-                        result["Value"],
-                    )
-                    return result
-                self.log.verbose(f"Failed to get token with cached tokens: {result['Message']}")
-                # Let's try to revoke broken token
-                idpObj.revokeToken(token["refresh_token"])
 
         err = []
-        # The cache did not help, so let's make an exchange token
+        # No luck so far, let's refresh the token stored in the database
         result = Registry.getDNForUsername(username)
         if not result["OK"]:
             return result
@@ -256,8 +242,8 @@ def export_getToken(
                     idpObj.token = result["Value"]
                     result = self.__checkProperties(dn, userGroup)
                     if result["OK"]:
-                        # exchange token with requested scope
-                        result = idpObj.exchangeToken()
+                        # refresh token with requested scope
+                        result = idpObj.refreshToken()
                         if result["OK"]:
                             # caching new tokens
                             self.__tokensCache.add(
@@ -266,8 +252,8 @@ def export_getToken(
                                 result["Value"],
                             )
                             return result
-                # Not find any token associated with the found user ID
-                err.append(result.get("Message", f"No token found for {uid}."))
+                # Did not find any token associated with the found user ID
+                err.append(result.get("Message", f"No token found for {uid}"))
         # Collect all errors when trying to get a token, or if no user ID is registered
         return S_ERROR("; ".join(err or [f"No user ID found for {username}"]))