[v7r2] HTTPs services within DIRAC

.github/workflows/integration.yml

@@ -12,8 +12,10 @@ jobs:
       MATRIX_DEFAULT_MYSQL_VER: 5.7
       MATRIX_DEFAULT_HOST_OS: cc7
       MATRIX_DEFAULT_USE_NEWTHREADPOOL: default
+      MATRIX_DEFAULT_DIRACOSVER: default
       MATRIX_DEFAULT_SERVER_USE_M2CRYPTO: Yes
       MATRIX_DEFAULT_CLIENT_USE_M2CRYPTO: Yes
+      MATRIX_DEFAULT_TEST_HTTPS: No
 
     strategy:
       fail-fast: False
@@ -38,6 +40,12 @@ jobs:
             CLIENT_USE_M2CRYPTO: Yes
           - SERVER_USE_M2CRYPTO: No
             CLIENT_USE_M2CRYPTO: No
+          ###### HTTPS tests
+          # IMPLICIT: - DIRACOSVER: default
+          #             TEST_HTTPS: No
+          - TEST_HTTPS: Yes
+            DIRACOSVER: https
+
 
     steps:
     - uses: actions/checkout@v2
@@ -59,6 +67,8 @@ jobs:
         echo -n "-e MYSQL_VER=${{ matrix.MYSQL_VER || env.MATRIX_DEFAULT_MYSQL_VER }} " >> run_in_container
         echo -n "-e ES_VER=${{ matrix.ES_VER || env.MATRIX_DEFAULT_ES_VER }} " >> run_in_container
         if [[ "${{ matrix.USE_NEWTHREADPOOL || env.MATRIX_DEFAULT_USE_NEWTHREADPOOL }}" != "default" ]]; then echo -n "-e DIRAC_USE_NEWTHREADPOOL=${{ matrix.USE_NEWTHREADPOOL || env.MATRIX_DEFAULT_USE_NEWTHREADPOOL }} " >> run_in_container; fi
+        if [[ "${{ matrix.DIRACOSVER || env.MATRIX_DEFAULT_DIRACOSVER }}" != "default" ]]; then echo -n "-e DIRACOSVER=${{ matrix.DIRACOSVER || env.MATRIX_DEFAULT_DIRACOSVER }} " >> run_in_container; fi
+        echo -n "-e TEST_HTTPS=${{ matrix.TEST_HTTPS || env.MATRIX_DEFAULT_TEST_HTTPS }} " >> run_in_container
         echo -n "-e SERVER_USE_M2CRYPTO=${{ matrix.SERVER_USE_M2CRYPTO || env.MATRIX_DEFAULT_SERVER_USE_M2CRYPTO }} " >> run_in_container
         echo -n "-e CLIENT_USE_M2CRYPTO=${{ matrix.CLIENT_USE_M2CRYPTO || env.MATRIX_DEFAULT_CLIENT_USE_M2CRYPTO }} " >> run_in_container
         # Finish wrapper script

ConfigurationSystem/Client/CSAPI.py

@@ -8,7 +8,7 @@
 
 import six
 from DIRAC import gLogger, gConfig, S_OK, S_ERROR
-from DIRAC.Core.DISET.RPCClient import RPCClient
+from DIRAC.ConfigurationSystem.Client.ConfigurationClient import ConfigurationClient
 from DIRAC.Core.Utilities import List, Time
 from DIRAC.Core.Security.X509Chain import X509Chain  # pylint: disable=import-error
 from DIRAC.Core.Security import Locations
@@ -90,7 +90,7 @@ def initialize(self):
     if not retVal['OK']:
       self.__initialized = S_ERROR("Master server is not known. Is everything initialized?")
       return self.__initialized
-    self.__rpcClient = RPCClient(gConfig.getValue("/DIRAC/Configuration/MasterServer", ""))
+    self.__rpcClient = ConfigurationClient(url=gConfig.getValue("/DIRAC/Configuration/MasterServer", ""))
     self.__csMod = Modificator(self.__rpcClient, "%s - %s - %s" %
                                (self.__userGroup, self.__userDN, Time.dateTime().strftime("%Y-%m-%d %H:%M:%S")))
     retVal = self.downloadCSData()

ConfigurationSystem/Client/ConfigurationClient.py

@@ -0,0 +1,77 @@
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+__RCSID__ = "$Id$"
+
+
+from base64 import b64encode, b64decode
+
+from DIRAC.Core.Tornado.Client.TornadoClient import TornadoClient
+from DIRAC.Core.Base.Client import Client
+
+
+class CSJSONClient(TornadoClient):
+  """
+    The specific Tornado client for configuration system.
+    To avoid JSON limitation the HTTPS handler encode data in base64
+    before sending them, this class only decode the base64
+
+    An exception is made with CommitNewData which ENCODE in base64
+  """
+
+  def getCompressedData(self):
+    """
+      Transmit request to service and get data in base64,
+      it decode base64 before returning
+
+      :returns: Configuration data, compressed
+      :rtype: str
+    """
+    retVal = self.executeRPC('getCompressedData')
+    if retVal['OK']:
+      retVal['Value'] = b64decode(retVal['Value'])
+    return retVal
+
+  def getCompressedDataIfNewer(self, sClientVersion):
+    """
+      Transmit request to service and get data in base64,
+      it decode base64 before returning.
+
+      :returns: Configuration data, if changed, compressed
+    """
+    retVal = self.executeRPC('getCompressedDataIfNewer', sClientVersion)
+    if retVal['OK'] and 'data' in retVal['Value']:
+      retVal['Value']['data'] = b64decode(retVal['Value']['data'])
+    return retVal
+
+  def commitNewData(self, sData):
+    """
+      Transmit request to service by encoding data in base64.
+
+      :param sData: Data to commit, you may call this method with CSAPI and Modificator
+    """
+    return self.executeRPC('commitNewData', b64encode(sData))
+
+
+class ConfigurationClient(Client):
+  """
+    Specific client to speak with ConfigurationServer.
+
+    This class must contain at least the JSON decoder dedicated to
+    the Configuration Server.
+
+    You can implement more logic or function to the client here if needed,
+    RPCCall can be made inside this class with executeRPC method.
+  """
+
+  # The JSON decoder for Configuration Server
+  httpsClient = CSJSONClient
+
+  def __init__(self, **kwargs):
+    # By default we use Configuration/Server as url, client do the resolution
+    # In some case url has to be static (when a slave register to the master server for example)
+    # It's why we can use 'url' as keyword arguments
+    if 'url' not in kwargs:
+      kwargs['url'] = 'Configuration/Server'
+    super(ConfigurationClient, self).__init__(**kwargs)

ConfigurationSystem/Service/TornadoConfigurationHandler.py

@@ -0,0 +1,148 @@
+""" The CS! (Configuration Service)
+
+Modified to work with Tornado
+Encode data in base64 because of JSON limitations
+In client side you must use a specific client
+
+"""
+
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+__RCSID__ = "$Id$"
+
+from base64 import b64encode, b64decode
+
+from DIRAC import S_OK, S_ERROR, gLogger
+from DIRAC.ConfigurationSystem.private.ServiceInterfaceTornado import ServiceInterfaceTornado as ServiceInterface
+from DIRAC.Core.Utilities import DErrno
+from DIRAC.Core.Tornado.Server.TornadoService import TornadoService
+
+sLog = gLogger.getSubLogger(__name__)
+
+
+class TornadoConfigurationHandler(TornadoService):
+  """
+    The CS handler
+  """
+  ServiceInterface = None
+  PilotSynchronizer = None
+
+  @classmethod
+  def initializeHandler(cls, serviceInfo):
+    """
+      Initialize the configuration server
+      Behind it starts thread which refresh configuration
+    """
+    cls.ServiceInterface = ServiceInterface(serviceInfo['URL'])
+    return S_OK()
+
+  def export_getVersion(self):
+    """
+      Returns the version of the configuration
+    """
+    return S_OK(self.ServiceInterface.getVersion())
+
+  def export_getCompressedData(self):
+    """
+      Returns the configuration
+    """
+    sData = self.ServiceInterface.getCompressedConfigurationData()
+    return S_OK(b64encode(sData))
+
+  def export_getCompressedDataIfNewer(self, sClientVersion):
+    """
+      Returns the configuration if a newer configuration exists, if not just returns the version
+
+      :param sClientVersion: Version used by client
+    """
+    sVersion = self.ServiceInterface.getVersion()
+    retDict = {'newestVersion': sVersion}
+    if sClientVersion < sVersion:
+      retDict['data'] = b64encode(self.ServiceInterface.getCompressedConfigurationData())
+    return S_OK(retDict)
+
+  def export_publishSlaveServer(self, sURL):
+    """
+      Used by slave server to register as a slave server.
+
+      :param sURL: The url of the slave server.
+    """
+    self.ServiceInterface.publishSlaveServer(sURL)
+    return S_OK()
+
+  def export_commitNewData(self, sData):
+    """
+      Write the new configuration
+    """
+    credDict = self.getRemoteCredentials()
+    if 'DN' not in credDict or 'username' not in credDict:
+      return S_ERROR("You must be authenticated!")
+    sData = b64decode(sData)
+    res = self.ServiceInterface.updateConfiguration(sData, credDict['username'])
+    if not res['OK']:
+      return res
+
+    # Check the flag for updating the pilot 3 JSON file
+    if self.srv_getCSOption('UpdatePilotCStoJSONFile', False) and self.ServiceInterface.isMaster():
+      if self.PilotSynchronizer is None:
+        try:
+          # This import is only needed for the Master CS service, making it conditional avoids
+          # dependency on the git client preinstalled on all the servers running CS slaves
+          from DIRAC.WorkloadManagementSystem.Utilities.PilotCStoJSONSynchronizer import PilotCStoJSONSynchronizer
+        except ImportError as exc:
+          sLog.exception("Failed to import PilotCStoJSONSynchronizer", repr(exc))
+          return S_ERROR(DErrno.EIMPERR, 'Failed to import PilotCStoJSONSynchronizer')
+        self.PilotSynchronizer = PilotCStoJSONSynchronizer()
+      return self.PilotSynchronizer.sync()
+
+    return res
+
+  def export_writeEnabled(self):
+    """
+      Used to know if we can change the configuration on this server
+    """
+    return S_OK(self.ServiceInterface.isMaster())
+
+  def export_getCommitHistory(self, limit=100):
+    """
+      Get the history of modifications in the configuration
+    """
+    if limit > 100:
+      limit = 100
+    history = self.ServiceInterface.getCommitHistory()
+    if limit:
+      history = history[:limit]
+    return S_OK(history)
+
+  def export_getVersionContents(self, versionList):
+    """
+      Get an old version of the configuration, can also be used to get more than one version.
+
+      :param versionList: List of the version we are trying to get
+    """
+    contentsList = []
+    for version in versionList:
+      retVal = self.ServiceInterface.getVersionContents(version)
+      if retVal['OK']:
+        contentsList.append(retVal['Value'])
+      else:
+        return S_ERROR("Can't get contents for version %s: %s" % (version, retVal['Message']))
+    return S_OK(contentsList)
+
+  def export_rollbackToVersion(self, version):
+    """
+      Rollback to an older version of the configuration
+
+      :param version: The version we want to apply
+    """
+    retVal = self.ServiceInterface.getVersionContents(version)
+    if not retVal['OK']:
+      return S_ERROR("Can't get contents for version %s: %s" % (version, retVal['Message']))
+    credDict = self.getRemoteCredentials()
+    if 'DN' not in credDict or 'username' not in credDict:
+      return S_ERROR("You must be authenticated!")
+    return self.ServiceInterface.updateConfiguration(retVal['Value'],
+                                                     credDict['username'],
+                                                     updateVersionOption=True)