Auth: get_auth_token_ssh: Deprecate string support for signature

lib/rucio/common/utils.py

@@ -1200,7 +1200,7 @@ def detect_client_location():
             'longitude': longitude}
 
 
-def ssh_sign(private_key, message):
+def ssh_sign(private_key: str, message: str) -> str:
     """
     Sign a string message using the private key.
 

lib/rucio/core/authentication.py

@@ -215,22 +215,20 @@ def get_auth_token_gss(account: InternalAccount, gsstoken: str, appid: str, ip:
 
 
 @transactional_session
-def get_auth_token_ssh(account: InternalAccount, signature: Union[str, bytes], appid: str, ip: Optional[str] = None, *, session: "Session") -> Optional[TokenDict]:
+def get_auth_token_ssh(account: InternalAccount, signature: bytes, appid: str, ip: Optional[str] = None, *, session: "Session") -> Optional[TokenDict]:
     """
     Authenticate a Rucio account temporarily via SSH key exchange.
 
     The token lifetime is 1 hour.
 
     :param account: Account identifier as a string.
-    :param signature: Response to server challenge signed with SSH private key as string.
+    :param signature: Response to server challenge signed with SSH private key.
     :param appid: The application identifier as a string.
     :param ip: IP address of the client as a string.
     :param session: The database session in use.
 
     :returns: A dict with token and expires_at entries.
     """
-    if isinstance(signature, str):
-        signature = signature.encode()
 
     # Make sure the account exists
     if not account_exists(account, session=session):

tests/test_authentication.py

@@ -146,7 +146,7 @@ def test_get_auth_token_ssh_fail(self, vo):
         except Duplicate:
             pass  # might already exist, can skip
 
-        signature = ssh_sign(PRIVATE_KEY, 'sign_something_else')
+        signature = base64.b64decode(ssh_sign(PRIVATE_KEY, 'sign_something_else'))
 
         result = get_auth_token_ssh(account='root', signature=signature, appid='test', ip='127.0.0.1', vo=vo)