feat: Add function to process the passcmd for API key retrieval.

Author: Gopinath-Mahendiran

zulip/zulip/__init__.py

@@ -5,6 +5,8 @@
 import os
 import platform
 import random
+import subprocess
+import shlex
 import sys
 import time
 import traceback
@@ -366,12 +368,15 @@ class MissingURLError(ZulipError):
 class UnrecoverableNetworkError(ZulipError):
     pass
 
+class APIKeyRetrievalError(ZulipError):
+    pass
 
 class Client:
     def __init__(
         self,
         email: Optional[str] = None,
         api_key: Optional[str] = None,
+        passcmd: Optional[str] = None,
         config_file: Optional[str] = None,
         verbose: bool = False,
         retry_on_errors: bool = True,
@@ -424,8 +429,28 @@ def __init__(
             config = ConfigParser()
             with open(config_file) as f:
                 config.read_file(f, config_file)
-            if api_key is None:
+            if api_key is None and config.has_option("api", "key"):
                 api_key = config.get("api", "key")
+            if passcmd is None and config.has_option("api", "passcmd"):
+                passcmd = config.get("api", "passcmd")
+                malicious_chars = {';', '|', '&', '>', '<', '`', '$', '\\', '\n', '\r'}
+                if any(char in passcmd for char in malicious_chars):
+                    raise APIKeyRetrievalError(
+                        f"Invalid characters detected in passcmd: {passcmd!r}"
+                    )
+                try:
+                    cmd_parts = shlex.split(passcmd)
+                except ValueError as e:
+                    raise APIKeyRetrievalError(
+                        f"Failed to parse passcmd '{passcmd}': {str(e)}")
+                try:
+                    result = subprocess.run(cmd_parts, capture_output=True, check=True)
+                    api_key = result.stdout.decode().strip()
+                except subprocess.CalledProcessError as e:
+                    raise APIKeyRetrievalError(
+                        f"Failed to retrieve API key using passcmd '{passcmd}'. "
+                        f"Command exited with return code {e.returncode}."
+                    )
             if email is None:
                 email = config.get("api", "email")
             if site is None and config.has_option("api", "site"):
@@ -512,6 +537,14 @@ def __init__(
         self.feature_level: int = server_settings.get("zulip_feature_level", 0)
         assert self.zulip_version is not None
 
+    def get_api_key(self, passcmd: str) -> Optional[str]:
+        # run the passcmd command and get the API key
+        result = subprocess.run(passcmd.split(), capture_output=True, check=False)
+        if result.returncode == 0:
+            return result.stdout.decode().strip()
+        else:
+            raise RuntimeError("Error: Unable to retrieve API key.")
+
     def ensure_session(self) -> None:
         # Check if the session has been created already, and return
         # immediately if so.