refactor: Split voms-proxy-init command for use with DiracX

src/DIRAC/Core/Security/Locations.py

@@ -45,6 +45,12 @@ def getCAsLocation():
         casPath = retVal["Value"]
         if os.path.isdir(casPath):
             return casPath
+    # Other locations
+    return getCAsLocationNoConfig()
+
+
+def getCAsLocationNoConfig():
+    """Retrieve the CA's files location"""
     # Look up the X509_CERT_DIR environment variable
     if "X509_CERT_DIR" in os.environ:
         casPath = os.environ["X509_CERT_DIR"]

src/DIRAC/Core/Security/ProxyFile.py

@@ -135,6 +135,8 @@ def multiProxyArgument(proxy=False):
                 return S_ERROR(DErrno.EPROXYFIND)
         if isinstance(proxy, str):
             proxyLoc = proxy
+        else:
+            raise NotImplementedError(f"Unknown proxy type ({type(proxy)})")
         # Load proxy
         proxy = X509Chain()
         retVal = proxy.loadProxyFromFile(proxyLoc)

src/DIRAC/Core/Security/VOMS.py

@@ -18,6 +18,37 @@
 VOMS_PROXY_INIT_CMD = "voms-proxy-init"
 
 
+def voms_init_cmd(
+    vo: str, attribute: str | None, chain: X509Chain, in_fn: str, out_fn: str, vomsesPath: str | None
+) -> list[str]:
+    secs = chain.getRemainingSecs()["Value"] - 300
+    if secs < 0:
+        return S_ERROR(DErrno.EVOMS, "Proxy length is less that 300 secs")
+    hours = int(secs / 3600)
+    mins = int((secs - hours * 3600) / 60)
+
+    bitStrength = chain.getStrength()["Value"]
+
+    cmd = [VOMS_PROXY_INIT_CMD]
+    if chain.isLimitedProxy()["Value"]:
+        cmd.append("-limited")
+    cmd += ["-cert", in_fn]
+    cmd += ["-key", in_fn]
+    cmd += ["-out", out_fn]
+    cmd += ["-voms"]
+    cmd += [f"{vo}:{attribute}" if attribute and attribute != "NoRole" else vo]
+    cmd += ["-valid", f"{hours}:{mins}"]
+    cmd += ["-bits", str(bitStrength)]
+    if vomsesPath:
+        cmd += ["-vomses", vomsesPath]
+
+    if chain.isRFC().get("Value"):
+        cmd += ["-r"]
+    cmd += ["-timeout", "12"]
+
+    return cmd
+
+
 class VOMS:
     def __init__(self, *args, **kwargs):
         """Create VOMS class, setting specific timeout for VOMS shell commands."""
@@ -225,38 +256,13 @@ def setVOMSAttributes(self, proxy, attribute=None, vo=None):
         chain = proxyDict["chain"]
         proxyLocation = proxyDict["file"]
 
-        secs = chain.getRemainingSecs()["Value"] - 300
-        if secs < 0:
-            return S_ERROR(DErrno.EVOMS, "Proxy length is less that 300 secs")
-        hours = int(secs / 3600)
-        mins = int((secs - hours * 3600) / 60)
-
-        # Ask VOMS a proxy the same strength as the one we already have
-        bitStrength = chain.getStrength()["Value"]
-
         retVal = self._generateTemporalFile()
         if not retVal["OK"]:
             deleteMultiProxy(proxyDict)
             return retVal
         newProxyLocation = retVal["Value"]
 
-        cmd = [VOMS_PROXY_INIT_CMD]
-        if chain.isLimitedProxy()["Value"]:
-            cmd.append("-limited")
-        cmd += ["-cert", proxyLocation]
-        cmd += ["-key", proxyLocation]
-        cmd += ["-out", newProxyLocation]
-        cmd += ["-voms"]
-        cmd += [f"{vo}:{attribute}" if attribute and attribute != "NoRole" else vo]
-        cmd += ["-valid", f"{hours}:{mins}"]
-        cmd += ["-bits", str(bitStrength)]
-        vomsesPath = self.getVOMSESLocation()
-        if vomsesPath:
-            cmd += ["-vomses", vomsesPath]
-
-        if chain.isRFC().get("Value"):
-            cmd += ["-r"]
-        cmd += ["-timeout", "12"]
+        cmd = voms_init_cmd(vo, attribute, chain, proxyLocation, newProxyLocation, self.getVOMSESLocation())
 
         result = shellCall(
             self._secCmdTimeout,