CLI improvements and fixes; v1.3.6

Author: BerndSchuller

CHANGES.md

@@ -3,6 +3,13 @@ Changelog for PyUNICORE
 
 Issue tracker: https://github.com/HumanBrainProject/pyunicore
 
+Version 1.3.6  (Dec 18, 2025)
+-----------------------------
+ - CLI: add bearer-token authentication method
+ - CLI: add sshkey authentication method
+ - CLI: resolve ${...} environment values in properties from config file
+ - fix: using non-EdDSA keys did not correctly set the JWT signing algorithm
+
 Version 1.3.5 (Sep 8, 2025)
 ---------------------------
  - improve forwarder

pyunicore/cli/base.py

@@ -5,6 +5,7 @@
 import json
 import os
 from base64 import b64decode
+from string import Template
 
 import pyunicore.client
 import pyunicore.credentials
@@ -33,7 +34,8 @@ def _value(self, value: str):
             return True
         if value.lower() == "false":
             return False
-        return value
+        t = Template(value)
+        return t.substitute(os.environ)
 
     def load_user_properties(self):
         with open(self.config_file) as f:
@@ -43,7 +45,7 @@ def load_user_properties(self):
                     continue
                 try:
                     key, value = line.split("=", 1)
-                    self.config[key] = self._value(value)
+                    self.config[key.strip()] = self._value(value.strip())
                 except ValueError:
                     pass
 
@@ -82,18 +84,31 @@ def get_group(self):
         return "Other"
 
     def create_credential(self):
-        auth_method = self.config.get("authentication-method", "USERNAME").upper()
-        if "OIDC-AGENT" == auth_method:
+        auth_method = self.config.get("authentication-method", "USERNAME")
+        _m = auth_method.upper()
+        if "OIDC-AGENT" == _m:
             account_name = self.config.get("oidc-agent.account")
             self.credential = pyunicore.credentials.OIDCAgentToken(account_name)
-        elif "OIDC-SERVER" == auth_method:
+        elif "OIDC-SERVER" == _m:
             self.credential = pyunicore.credentials.OIDCServerToken(self.config)
-        elif "USERNAME" == auth_method:
+        elif "USERNAME" == _m:
             username = self.config["username"]
             password = self._get_password()
             self.credential = pyunicore.credentials.create_credential(username, password)
-        elif "ANONYMOUS" == auth_method:
+        elif "BEARER-TOKEN" == _m:
+            token = self.config["token"]
+            self.credential = pyunicore.credentials.BearerToken(token)
+        elif "SSHKEY" == _m:
+            username = self.config["username"]
+            identity = self.config["identity"]
+            password = self._get_password()
+            self.credential = pyunicore.credentials.create_credential(
+                username=username, password=password, identity=identity
+            )
+        elif "ANONYMOUS" == _m or "NONE" == _m:
             self.credential = pyunicore.credentials.Anonymous()
+        else:
+            raise ValueError(f"No such authentication method: {auth_method}")
 
     def _get_password(self, key="password") -> str:
         password = self.config.get(key)

pyunicore/cli/info.py

@@ -8,7 +8,7 @@
 
 class Info(Base):
     def add_command_args(self):
-        self.parser.prog = "unicore system-info"
+        self.parser.prog = "unicore info"
         self.parser.description = self.get_synopsis()
         self.parser.add_argument("URL", help="Endpoint URL(s)", nargs="*")
         self.parser.add_argument(
@@ -52,21 +52,25 @@ def run(self, args):
 
     def show_endpoint_details(self, ep: Resource):
         print(ep.resource_url)
-        if ep.resource_url.endswith("/rest/core"):
+        if re.match(".*/rest/core[/]?$", ep.resource_url):
             self._show_details_core(ep)
         elif re.match(".*/rest/core/storages/.+", ep.resource_url):
             self._show_details_storage(ep)
+        elif re.match(".*/rest/core/factories/.+", ep.resource_url):
+            self._show_details_sitefactory(ep)
         else:
             print(" * no further details available.")
 
     def _show_details_core(self, ep: Resource):
         props = ep.properties
         print(" * type: UNICORE/X base")
         print(f" * server v{props['server']['version']}")
+        dn = {props["client"]["dn"]}
         xlogin = props["client"]["xlogin"]
         role = props["client"]["role"]["selected"]
         uid = xlogin.get("UID", "n/a")
-        print(f" * authenticated as: '{props['client']['dn']}' role='{role}' uid='{uid}'")
+        method = props["client"].get("authenticationMethod", "n/a")
+        print(f" * authenticated (via '{method}') as: '{dn}' role='{role}' uid='{uid}'")
         grps = xlogin.get("availableGroups", [])
         uids = xlogin.get("availableUIDs", [])
         if len(uids) > 0:
@@ -85,3 +89,7 @@ def _show_details_storage(self, ep: Resource):
         print(f" * type: {t}")
         print(f" * mount point: {props['mountPoint']}")
         print(f" * free space : {int(props['freeSpace'] / 1024 / 1024)} MB")
+
+    def _show_details_sitefactory(self, ep: Resource):
+        t = "compute"
+        print(f" * type: {t}")

pyunicore/cli/main.py

@@ -23,7 +23,7 @@
 }
 
 
-def get_command(name):
+def get_command(name) -> pyunicore.cli.base.Base:
     return _commands.get(name)()
 
 
@@ -37,12 +37,12 @@ def show_version():
 
 
 def help():
+    print(_header)
     s = """UNICORE Commandline Client (pyUNICORE) %s, https://www.unicore.eu
 Usage: unicore <command> [OPTIONS] <args>
 The following commands are available:""" % pyunicore._version.get_versions().get(
         "version", "n/a"
     )
-    print(_header)
     print(s)
     for cmd in sorted(_commands):
         print(f" {cmd:20} - {get_command(cmd).get_description()}")

pyunicore/credentials.py

@@ -332,7 +332,7 @@ def create_credential(username=None, password=None, token=None, identity=None):
     username + password : create a UsernamePassword credential
     token               ; create a OIDCToken credential from the given token
     username + identity : create a JWTToken credential which will be signed
-                          with the given private key (ssh key or PEM)
+                          with the given private key (ssh key or X509)
     """
 
     if token is not None:
@@ -343,6 +343,8 @@ def create_credential(username=None, password=None, token=None, identity=None):
         raise AuthenticationFailedException("Not enough info to create user credential")
     try:
         from cryptography.hazmat.primitives import serialization
+        from cryptography.hazmat.primitives.asymmetric.ec import EllipticCurvePrivateKey
+        from cryptography.hazmat.primitives.asymmetric.rsa import RSAPrivateKey
 
         if not isabs(identity):
             if identity.startswith("~"):
@@ -362,9 +364,9 @@ def create_credential(username=None, password=None, token=None, identity=None):
         secret = private_key
         sub = username
         algo = "EdDSA"
-        if "BEGIN RSA" in pem:
+        if isinstance(private_key, RSAPrivateKey):
             algo = "RS256"
-        elif "BEGIN EC" in pem or "PuTTY" in pem:
+        elif isinstance(private_key, EllipticCurvePrivateKey) or "PuTTY" in pem:
             algo = "ES256"
         return JWTToken(sub, sub, secret, algorithm=algo, etd=False)
     except ImportError: