DGS-22404 Add AppRole auth for HC Vault (#2084)

Author: rayokota

src/confluent_kafka/schema_registry/rules/encryption/hcvault/hcvault_client.py

@@ -29,9 +29,10 @@ class HcVaultKmsClient(tink.KmsClient):
     """Basic HashiCorp Vault client for AEAD."""
 
     def __init__(
-        self, key_uri: Optional[str], token: str, ns: Optional[str] = None
+        self, key_uri: Optional[str], token: Optional[str], ns: Optional[str] = None,
+        role_id: Optional[str] = None, secret_id: Optional[str] = None
     ) -> None:
-        """Creates a new GcpKmsClient that is bound to the key specified in 'key_uri'.
+        """Creates a new HcVaultKmsClient that is bound to the key specified in 'key_uri'.
 
         Uses the specified credentials when communicating with the KMS.
 
@@ -59,6 +60,8 @@ def __init__(
             namespace=ns,
             verify=False
         )
+        if role_id and secret_id:
+            self._client.auth.approle.login(role_id=role_id, secret_id=secret_id)
 
     def does_support(self, key_uri: str) -> bool:
         """Returns true iff this client supports KMS key specified in 'key_uri'.

src/confluent_kafka/schema_registry/rules/encryption/hcvault/hcvault_driver.py

@@ -23,6 +23,8 @@
 _PREFIX = "hcvault://"
 _TOKEN_ID = "token.id"
 _NAMESPACE = "namespace"
+_APPROLE_ROLE_ID = "approle.role.id"
+_APPROLE_SECRET_ID = "approle.secret.id"
 
 
 class HcVaultKmsDriver(KmsDriver):
@@ -37,11 +39,18 @@ def new_kms_client(self, conf: dict, key_url: str) -> KmsClient:
         if key_url is not None:
             uri_prefix = key_url
         token = conf.get(_TOKEN_ID)
-        namespace = conf.get(_NAMESPACE)
         if token is None:
             token = os.getenv("VAULT_TOKEN")
+        namespace = conf.get(_NAMESPACE)
+        if namespace is None:
             namespace = os.getenv("VAULT_NAMESPACE")
-        return HcVaultKmsClient(uri_prefix, token, namespace)
+        role_id = conf.get(_APPROLE_ROLE_ID)
+        if role_id is None:
+            role_id = os.getenv("VAULT_APPROLE_ROLE_ID")
+        secret_id = conf.get(_APPROLE_SECRET_ID)
+        if secret_id is None:
+            secret_id = os.getenv("VAULT_APPROLE_SECRET_ID")
+        return HcVaultKmsClient(uri_prefix, token, namespace, role_id, secret_id)
 
     @classmethod
     def register(cls):