[v7r2] Integration merged with v7r1

.github/workflows/basic.yml

@@ -12,11 +12,11 @@ jobs:
       fail-fast: False
       matrix:
         command:
-          - pytest
-          - DIRAC_USE_M2CRYPTO=Yes DIRAC_M2CRYPTO_SPLIT_HANDSHAKE=Yes pytest
+          - pytest --no-cov
+          - DIRAC_USE_M2CRYPTO=Yes DIRAC_M2CRYPTO_SPLIT_HANDSHAKE=Yes pytest --no-cov
           # Security tests are flakey due to reference counting bugs in pyGSI/src/crypto/asn1.c
-          - pytest Core/Security/test || (echo "Retrying..."; pytest Core/Security/test) || (echo "Retrying again..."; pytest Core/Security/test)
-          - DIRAC_USE_M2CRYPTO=Yes DIRAC_M2CRYPTO_SPLIT_HANDSHAKE=Yes pytest Core/Security/test || (echo "Retrying..."; DIRAC_USE_M2CRYPTO=No pytest Core/Security/test) || (echo "Retrying again..."; DIRAC_USE_M2CRYPTO=No pytest Core/Security/test)
+          - pytest --no-cov Core/Security/test
+          - DIRAC_USE_M2CRYPTO=Yes DIRAC_M2CRYPTO_SPLIT_HANDSHAKE=Yes pytest --no-cov Core/Security/test
           - tests/checkDocs.sh
           # TODO This should cover more than just tests/CI
           # Excluded codes related to sourcing files

.gitignore

@@ -73,6 +73,7 @@ tests/CI/SERVERCONFIG
 
 # VSCode
 .vscode
+.env
 
 # docs
 # this is auto generated

ConfigurationSystem/Client/CSAPI.py

@@ -210,7 +210,7 @@ def listHosts(self):
   def describeUsers(self, users=None):
     """ describe users by nickname
 
-        :param list users: list of users' nickanames
+        :param list users: list of users' nicknames
         :return: a S_OK(description) of the users in input
     """
     if users is None:
@@ -329,14 +329,14 @@ def addUser(self, username, properties):
     """
     Add a user to the cs
 
-      :param str username: group name
-      :param dict properties: dictionary describing user properties:
+    :param str username: username
+    :param dict properties: dictionary describing user properties:
 
-        - DN
-        - groups
-        - <extra params>
+      - DN
+      - groups
+      - <extra params>
 
-      :return: True/False
+    :return: True/False
     """
     if not self.__initialized['OK']:
       return self.__initialized
@@ -368,14 +368,15 @@ def modifyUser(self, username, properties, createIfNonExistant=False):
     """
     Modify a user
 
-      :param str username: group name
-      :param dict properties: dictionary describing user properties:
+    :param str username: group name
+    :param dict properties: dictionary describing user properties:
 
         - DN
         - Groups
         - <extra params>
 
-      :return: S_OK, Value = True/False
+    :param bool createIfNonExistant: if true, registers the users if it did not exist
+    :return: S_OK, Value = True/False
     """
     if not self.__initialized['OK']:
       return self.__initialized
@@ -430,14 +431,14 @@ def addGroup(self, groupname, properties):
     """
     Add a group to the cs
 
-      :param str groupname: group name
-      :param dict properties: dictionary describing group properties:
+    :param str groupname: group name
+    :param dict properties: dictionary describing group properties:
 
         - Users
         - Properties
         - <extra params>
 
-      :return: True/False
+    :return: S_OK, Value = True/False
     """
     if not self.__initialized['OK']:
       return self.__initialized
@@ -455,14 +456,15 @@ def modifyGroup(self, groupname, properties, createIfNonExistant=False):
     """
     Modify a group
 
-      :param str groupname: group name
-      :param dict properties: dictionary describing group properties:
+    :param str groupname: group name
+    :param dict properties: dictionary describing group properties:
 
         - Users
         - Properties
         - <extra params>
 
-      :return: True/False
+    :param bool createIfNonExistant: if true, creates the group if it did not exist
+    :return: S_OK, Value = True/False
     """
     if not self.__initialized['OK']:
       return self.__initialized
@@ -490,14 +492,15 @@ def modifyGroup(self, groupname, properties, createIfNonExistant=False):
   def addHost(self, hostname, properties):
     """
     Add a host to the cs
-      :param str hostname: hostname name
-      :param dict properties: dictionary describing host properties:
+
+    :param str hostname: host name
+    :param dict properties: dictionary describing host properties:
 
         - DN
         - Properties
         - <extra params>
 
-      :return: True/False
+    :return: S_OK, Value = True/False
     """
     if not self.__initialized['OK']:
       return self.__initialized
@@ -617,14 +620,16 @@ def getOpsSection():
   def modifyHost(self, hostname, properties, createIfNonExistant=False):
     """
     Modify a host
-      :param str hostname: hostname name
-      :param dict properties: dictionary describing host properties:
+
+    :param str hostname: hostname name
+    :param dict properties: dictionary describing host properties:
 
         - DN
         - Properties
         - <extra params>
 
-      :return: True/False
+    :param bool createIfNonExistant: if true, creates the host if it did not exist
+    :return: S_OK, Value = True/False
     """
     if not self.__initialized['OK']:
       return self.__initialized

ConfigurationSystem/Client/Config.py

@@ -6,8 +6,13 @@
 
 from DIRAC.ConfigurationSystem.private.ConfigurationClient import ConfigurationClient
 
+#: Global gConfig object of type :class:`~DIRAC.ConfigurationSystem.private.ConfigurationClient.ConfigurationClient`
 gConfig = ConfigurationClient()
 
 
 def getConfig():
+  """
+  :returns: gConfig
+  :rtype: ~DIRAC.ConfigurationSystem.private.ConfigurationClient.ConfigurationClient
+  """
   return gConfig

ConfigurationSystem/Client/Helpers/Registry.py

@@ -23,6 +23,7 @@ def getUsernameForDN(dn, usersList=None):
 
       :return: S_OK(str)/S_ERROR()
   """
+  dn = dn.strip()
   if not usersList:
     result = gConfig.getSections("%s/Users" % gBaseRegistrySection)
     if not result['OK']:
@@ -63,6 +64,7 @@ def getGroupsForDN(dn):
 
       :return: S_OK(list)/S_ERROR() -- contain list of groups
   """
+  dn = dn.strip()
   result = getUsernameForDN(dn)
   if not result['OK']:
     return result
@@ -128,6 +130,7 @@ def getHostnameForDN(dn):
 
       :return: S_OK()/S_ERROR()
   """
+  dn = dn.strip()
   result = gConfig.getSections("%s/Hosts" % gBaseRegistrySection)
   if not result['OK']:
     return result
@@ -153,6 +156,7 @@ def findDefaultGroupForDN(dn):
 
       :return: S_OK()/S_ERROR()
   """
+  dn = dn.strip()
   result = getUsernameForDN(dn)
   if not result['OK']:
     return result

Core/DISET/private/Transports/M2SSLTransport.py

@@ -45,6 +45,9 @@ def __init__(self, *args, **kwargs):
         as for other transports. If ctx is specified (as an instance of
         SSL.Context) then use that rather than creating a new context.
     """
+    # The thread init of M2Crypto is not really thread safe.
+    # So we put it a second time
+    M2Threading.init()
     self.remoteAddress = None
     self.peerCredentials = {}
     self.__timeout = 1
@@ -98,9 +101,7 @@ def initAsClient(self):
         error = "%s:%s" % (e, repr(e))
 
         if self.oSocket is not None:
-          self.oSocket.close()
-          self.oSocket.socket.close()
-          self.oSocket = None
+          self.close()
 
     return S_ERROR(error)
 
@@ -125,20 +126,85 @@ def initAsServer(self):
     return S_OK()
 
   def close(self):
+    # pylint: disable=line-too-long
     """ Close this socket. """
+
     if self.oSocket:
+      # TL;DR:
+      # Do NOT touch that method
+      #
       # Surprisingly (to me at least), M2Crypto does not close
-      # the socket when calling SSL.Connection.close
-      # It only does it when the garbage collector kicks in
-      # We have to manually close it here, otherwise the connections
-      # will hang forever
+      # the underlying socket when calling SSL.Connection.close
+      # It only does it when the garbage collector kicks in (see ~M2Crypto.SSL.Connection.Connection.__del__)
+      # If the socket is not closed, the connection may hang forever.
+      #
+      # Thus, we are setting self.oSocket to None to allow the GC to do the work, but since we are not sure
+      # that it will run, we anyway force the connection to be closed
+      #
+      # However, we should close the underlying socket only after SSL was shutdown properly.
+      # This is because OpenSSL `ssl3_shutdown` (see callstack below) may still read some data
+      # (see https://github.com/openssl/openssl/blob/master/ssl/s3_lib.c#L4509)::
+      #
+      #
+      # 1  0x00007fffe9d48fc0 in sock_read () from /lib/libcrypto.so.1.0.0
+      # 2  0x00007fffe9d46e83 in BIO_read () from /lib/libcrypto.so.1.0.0
+      # 3  0x00007fffe9eab9dd in ssl3_read_n () from /lib/libssl.so.1.0.0
+      # 4  0x00007fffe9ead216 in ssl3_read_bytes () from /lib/libssl.so.1.0.0
+      # 5  0x00007fffe9ea999c in ssl3_shutdown () from /lib/libssl.so.1.0.0
+      # 6  0x00007fffe9ed4f93 in ssl_free () from /lib/libssl.so.1.0.0
+      # 7  0x00007fffe9d46d5b in BIO_free () from /lib/libcrypto.so.1.0.0
+      # 8  0x00007fffe9f30a96 in bio_free (bio=0x5555556f3200) at SWIG/_m2crypto_wrap.c:5008
+      # 9  0x00007fffe9f30b1e in _wrap_bio_free (self=<optimized out>, args=<optimized out>) at SWIG/_m2crypto_wrap.c
+      #
+      # We unfortunately have no way to force that order, and there is a risk of deadlock
+      # when running in a multi threaded environment like the agents::
+      #
+      # Thread A opens socket, gets FD = 111
+      # Thread A works on it
+      # Thread A closes FD 111 (underlying socket.close())
+      # Thread B opens socket, gets FD = 111
+      # Thread A calls read on FD=111 from ssl3_shutdown
+      #
+      # This is illustrated on the strace below::
+      #
+      # 26461 14:25:15.266692 write(111<TCPv6:[[<srcAddressV6>]:42688->[<dstAddressV6>]:9140]>,
+      #                            "blabla", 37 <unfinished ...>
+      # 26464 14:25:15.266857 <... connect resumed>) = 0 <0.000195>
+      # 26464 14:25:15.267023 getsockname(120<UDP:[<srcAddress>:44252->188.185.84.86:9140]>, <unfinished ...>
+      # 26461 14:25:15.267176 <... write resumed>) = 37 <0.000453>
+      # 26464 14:25:15.267425 <... getsockname resumed>{sa_family=AF_INET, sin_port=htons(44252),
+      #                        sin_addr=inet_addr("<srcAddress>")}, [28->16]) = 0 <0.000292>
+      # 26461 14:25:15.267466 close(111<TCPv6:[[<srcAddressV6>]:42688->[<dstAddressV6>]:9140]> <unfinished ...>
+      # 26464 14:25:15.267637 close(120<UDP:[<srcAddress>:44252->188.185.84.86:9140]> <unfinished ...>
+      # 26464 14:25:15.267738 <... close resumed>) = 0 <0.000086>
+      # 26461 14:25:15.267768 <... close resumed>) = 0 <0.000285>
+      # 26464 14:25:15.267827 socket(AF_INET6, SOCK_DGRAM|SOCK_CLOEXEC, IPPROTO_IP <unfinished ...>
+      # 26461 14:25:15.267888 futex(0x21f8620, FUTEX_WAKE_PRIVATE, 1 <unfinished ...>
+      # 26464 14:25:15.267976 <... socket resumed>) = 111<UDPv6:[1207802822]> <0.000138>
+      # 26461 14:25:15.268092 <... futex resumed>) = 1 <0.000196>
+      # 26464 14:25:15.268195 connect(111<UDPv6:[1207802822]>,
+      #                      {sa_family=AF_INET6, sin6_port=htons(9140),
+      #                       inet_pton(AF_INET6, "<dstAddressV6>", &sin6_addr),
+      #                      sin6_flowinfo=htonl(0), sin6_scope_id=0
+      #                      }, 28 <unfinished ...>
+      # 26461 14:25:15.268294 read(111<UDPv6:[[<srcAddressV6>]:42480->[<dstAddressV6>]:9140]>, <unfinished ...>
+      # 26464 14:25:15.268503 <... connect resumed>) = 0 <0.000217>
+      # 26464 14:25:15.268673 getsockname(111<UDPv6:[[<srcAddressV6>]:42480->[<dstAddressV6>]:9140]>, <unfinished ...>
+      # 26464 14:25:15.268862 <... getsockname resumed>{sa_family=AF_INET6, sin6_port=htons(42480),
+      #                        inet_pton(AF_INET6, "<srcAddressV6>", &sin6_addr), sin6_flowinfo=htonl(0), sin6_scope_id=
+      # 0}, [28]) = 0 <0.000168>
+      # 26464 14:25:15.269048
+      # close(111<UDPv6:[[<srcAddressV6>]:42480->[<dstAddressV6>]:9140]>
+      # <unfinished ...>
+
       self.oSocket.close()
-      self.oSocket.socket.close()
-      # del self.oSocket
+      underlyingSocket = self.oSocket.socket
       self.oSocket = None
+      underlyingSocket.close()
     return S_OK()
 
   def renewServerContext(self):
+    # pylint: disable=line-too-long
     """ Renews the server context.
         This reloads the certificates and re-initialises the SSL context.
 

DataManagementSystem/Agent/FTS3Agent.py

@@ -369,24 +369,36 @@ def _treatOperation(self, operation):
         continueOperationProcessing = True
 
         # Check the status of the associated RMS Request.
-        # If it is canceled then we will not create new FTS3Jobs, and mark
+        # If it is canceled or does not exist anymore then we will not create new FTS3Jobs, and mark
         # this as FTS3Operation canceled.
 
         if operation.rmsReqID:
           res = ReqClient().getRequestStatus(operation.rmsReqID)
           if not res['OK']:
-            log.error("Could not get request status", res)
-            return operation, res
-          rmsReqStatus = res['Value']
-
-          if rmsReqStatus == 'Canceled':
-            log.info(
-                "The RMS Request is canceled, canceling the FTS3Operation",
-                "rmsReqID: %s, FTS3OperationID: %s" %
-                (operation.rmsReqID,
-                 operation.operationID))
-            operation.status = 'Canceled'
-            continueOperationProcessing = False
+            # If the Request does not exist anymore
+            if cmpError(res, errno.ENOENT):
+              log.info(
+                  "The RMS Request does not exist anymore, canceling the FTS3Operation",
+                  "rmsReqID: %s, FTS3OperationID: %s" %
+                  (operation.rmsReqID,
+                   operation.operationID))
+              operation.status = 'Canceled'
+              continueOperationProcessing = False
+            else:
+              log.error("Could not get request status", res)
+              return operation, res
+
+          else:
+            rmsReqStatus = res['Value']
+
+            if rmsReqStatus == 'Canceled':
+              log.info(
+                  "The RMS Request is canceled, canceling the FTS3Operation",
+                  "rmsReqID: %s, FTS3OperationID: %s" %
+                  (operation.rmsReqID,
+                   operation.operationID))
+              operation.status = 'Canceled'
+              continueOperationProcessing = False
 
         if continueOperationProcessing:
           res = operation.prepareNewJobs(

DataManagementSystem/Client/FTS3Client.py

@@ -11,7 +11,7 @@ class FTS3Client(Client):
   def __init__(self, url=None, **kwargs):
     """ Constructor function.
     """
-    Client.__init__(self, **kwargs)
+    super(FTS3Client, self).__init__(**kwargs)
     self.setServer('DataManagement/FTS3Manager')
     if url:
       self.setServer(url)

DataManagementSystem/Client/S3GatewayClient.py

@@ -11,7 +11,7 @@ class S3GatewayClient(Client):
   def __init__(self, url=None, **kwargs):
     """ Constructor function.
     """
-    Client.__init__(self, **kwargs)
+    super(S3GatewayClient, self).__init__(**kwargs)
     self.setServer('DataManagement/S3Gateway')
     if url:
       self.setServer(url)