Merge pull request #7913 from sfayer/fix_arcproxy

[8.0] Add option to include proxy on AREX token submission

src/DIRAC/Resources/Computing/AREXComputingElement.py

@@ -20,6 +20,11 @@
 RESTVersion:
    Version of the REST interface to use.
 
+AlwaysIncludeProxy:
+    A boolean, set to true to include the proxy in job submission even
+    in cases where tokens are the primary authentication method.
+    (Recommended for ARC6 tokens, deprecated for ARC7)
+
 **Code Documentation**
 """
 
@@ -56,6 +61,8 @@ def __init__(self, ceUniqueID):
         }
         # URL used to communicate with the REST interface
         self.base_url = ""
+        # A flag to always include a proxy, even if a token is the primary auth method
+        self.alwaysIncludeProxy = False
 
     #############################################################################
 
@@ -88,6 +95,10 @@ def _reset(self):
         service_url = os.path.join("https://", f"{self.ceName}:{self.port}")
         self.base_url = os.path.join(service_url, "arex", "rest", self.restVersion)
 
+        self.alwaysIncludeProxy = False
+        if self.ceParameters.get("AlwaysIncludeProxy", "false").lower() in ("true", "yes"):
+            self.alwaysIncludeProxy = True
+
         # Set up the request framework
         self.session = requests.Session()
         self.session.verify = Locations.getCAsLocation()
@@ -187,13 +198,16 @@ def _checkSession(self):
         if not (self.token or self.proxy):
             self.log.error("Proxy or token not set")
             return S_ERROR("Proxy or token not set")
+        if not self.proxy and self.alwaysIncludeProxy:
+            self.log.error("Proxy required but not set")
+            return S_ERROR("Proxy required but not set")
 
         # If a token is set, we use it
         if self.token:
             # Attach the token to the headers if present
             self.headers["Authorization"] = f"Bearer {self.token['access_token']}"
             self.log.verbose("A token is attached to the header of the request(s)")
-        else:
+        if not self.token or self.alwaysIncludeProxy:
             # Prepare the proxy in X509_USER_PROXY
             if not (result := self._prepareProxy())["OK"]:
                 self.log.error("Failed to set up proxy", result["Message"])
@@ -433,7 +447,7 @@ def submitJob(self, executableFile, proxy, numberOfJobs=1, inputs=None, outputs=
 
         # Delegation cannot be used with a token
         delegation = ""
-        if not self.token:
+        if not self.token or self.alwaysIncludeProxy:
             # Get existing delegations
             result = self._getDelegationIDs()
             if not result["OK"]:
@@ -770,7 +784,7 @@ def getJobStatus(self, jobIDList):
                 self.log.debug(f"Killing held job {jobReference}")
 
         # Renew delegations to renew the proxies of the jobs
-        if not self.token:
+        if not self.token or self.alwaysIncludeProxy:
             result = self._getDelegationIDs()
             if not result["OK"]:
                 return result