refactor: improve error handling for client auth

kapitan/refs/vault_resources.py

@@ -5,9 +5,11 @@
 
 "hashicorp vault resource functions"
 
-import os
 import logging
+import os
+
 import hvac
+from requests.exceptions import SSLError
 
 from kapitan.errors import KapitanError
 
@@ -25,9 +27,9 @@ class VaultClient(hvac.Client):
 
     def __init__(self, vault_parameters):
         self.vault_parameters = vault_parameters
-        self.env = get_env(vault_parameters)
 
-        super().__init__(**{k: v for k, v in self.env.items() if k != "auth"})
+        self.client_parameters = self.get_env()
+        super().__init__(**self.client_parameters)
 
         self.authenticate()
 
@@ -44,7 +46,7 @@ def get_auth_token(self):
                 token_file = os.path.join(os.path.expanduser("~"), ".vault-token")
                 token = self.read_token_from_file(token_file)
 
-        self.env["token"] = token
+        return token
 
     def read_token_from_file(self, token_file):
         try:
@@ -53,83 +55,133 @@ def read_token_from_file(self, token_file):
         except IOError:
             raise VaultError("Cannot read file {}".format(token_file))
 
-        if not token:
+        if not token or len(token) == 0:
             raise VaultError("{} is empty".format(token_file))
 
+        # clean up token of unwanted line endings
+        token = token.replace("\n", "")
+        token = token.replace("\r", "")
+
         return token
 
     def authenticate(self):
-        # DIFFERENT LOGIN METHOD BASED ON AUTHENTICATION TYPE
-        auth_type = self.vault_parameters["auth"]
-        self.get_auth_token()
+        # different login method based on authentication type
+        auth_type = self.vault_parameters.get("auth")
+        if not auth_type:
+            raise VaultError(
+                "No authentication method defined. Please specify it in 'parameters.kapitan.secrets.vaultkv.auth'."
+            )
+
+        token = self.get_auth_token()
         username = os.getenv("VAULT_USERNAME")
         password = os.getenv("VAULT_PASSWORD")
 
+        # token
         if auth_type == "token":
-            self.token = self.env["token"]
+            if token:
+                self.token = token
+            else:
+                raise VaultError(
+                    "token authentication failed: VAULT_TOKEN is empty and '~/.vaulttoken not found"
+                )
+        # github
+        elif auth_type == "github":
+            if token:
+                self.auth.github.login = token
+            else:
+                raise VaultError(
+                    "token authentication failed: VAULT_TOKEN is empty and '~/.vaulttoken not found"
+                )
+        # ldap
         elif auth_type == "ldap":
-            self.auth.ldap.login(username=username, password=password)
+            if username and password:
+                self.auth.ldap.login(username=username, password=password)
+            else:
+                raise VaultError("ldap authentication failed: VAULT_USERNAME or VAULT_PASSWORD is empty")
+        # userpass
         elif auth_type == "userpass":
-            self.auth.userpass.login(username=username, password=password)
+            if username and password:
+                self.auth.userpass.login(username=username, password=password)
+            else:
+                raise VaultError("userpass authentication failed: VAULT_USERNAME or VAULT_PASSWORD is empty")
+        # approle
         elif auth_type == "approle":
-            self.auth.approle.login(os.getenv("VAULT_ROLE_ID"), secret_id=os.getenv("VAULT_SECRET_ID"))
-        elif auth_type == "github":
-            self.auth.github.login(token=self.env["token"])
+            role_id = os.getenv("VAULT_ROLE_ID")
+            secret_id = os.getenv("VAULT_SECRET_ID")
+            if role_id and secret_id:
+                self.auth.approle.login(role_id, secret_id=secret_id)
+            else:
+                raise VaultError("approle authentication failed: VAULT_ROLE_ID or VAULT_SECRET_ID is empty")
         else:
-            raise VaultError("Authentication type '{}' not supported".format(auth_type))
+            raise VaultError(f"Authentication type '{auth_type}' not supported")
+
+        try:
+            authenticated = self.is_authenticated()
+        except SSLError as e:
+            ca_path = os.path.abspath(self.client_parameters["verify"])
+            raise VaultError(f"Error while verifying ca-bundle at {ca_path}: {e.__context__.__context__}")
 
-        if not self.is_authenticated():
+        if not authenticated:
             self.adapter.close()
             raise VaultError("Vault Authentication Error, Environment Variables defined?")
 
-
-def get_env(parameter):
-    """
-    The following variables need to be exported to the environment or defined in inventory.
-        * VAULT_ADDR: url for vault
-        * VAULT_SKIP_VERIFY=true: if set, do not verify presented TLS certificate before communicating with Vault server.
-        * VAULT_CLIENT_KEY: path to an unencrypted PEM-encoded private key matching the client certificate
-        * VAULT_CLIENT_CERT: path to a PEM-encoded client certificate for TLS authentication to the Vault server
-        * VAULT_CACERT: path to a PEM-encoded CA cert file to use to verify the Vault server TLS certificate
-        * VAULT_CAPATH: path to a directory of PEM-encoded CA cert files to verify the Vault server TLS certificate
-        * VAULT_NAMESPACE: specify the Vault Namespace, if you have one
-    Following keys are used to creates a new hvac client instance.
-        :param url: Base URL for the Vault instance being addressed.
-        :type url: str
-        :param cert: Certificates for use in requests sent to the Vault instance. This should be a tuple with the
-            certificate and then key.
-        :type cert: tuple
-        :param verify: Either a boolean to indicate whether TLS verification should be performed when sending requests to Vault,
-            or a string pointing at the CA bundle to use for verification.
-            ### ssl-cert-verification.
-            See http://docs.python-requests.org/en/master/user/advanced/
-        :type verify: Union[bool,str]
-        :param namespace: Optional Vault Namespace.
-        :type namespace: str
-    """
-    client_parameters = {}
-    client_parameters["url"] = parameter.get("VAULT_ADDR", os.getenv("VAULT_ADDR", default=""))
-    client_parameters["namespace"] = parameter.get(
-        "VAULT_NAMESPACE", os.getenv("VAULT_NAMESPACE", default="")
-    )
-    # VERIFY VAULT SERVER TLS CERTIFICATE
-    skip_verify = str(parameter.get("VAULT_SKIP_VERIFY", os.getenv("VAULT_SKIP_VERIFY", default="")))
-
-    if skip_verify.lower() == "false":
-        cert = parameter.get("VAULT_CACERT", os.getenv("VAULT_CACERT", default=""))
-        if not cert:
-            cert_path = parameter.get("VAULT_CAPATH", os.getenv("VAULT_CAPATH", default=""))
-            if not cert_path:
-                raise VaultError("Neither VAULT_CACERT nor VAULT_CAPATH specified")
-            client_parameters["verify"] = cert_path
+    def get_env(self):
+        """
+        The following variables need to be exported to the environment or defined in inventory.
+            * VAULT_ADDR: url for vault
+            * VAULT_SKIP_VERIFY: if set, do not verify presented TLS certificate before communicating with Vault server.
+            * VAULT_CLIENT_KEY: path to an unencrypted PEM-encoded private key matching the client certificate
+            * VAULT_CLIENT_CERT: path to a PEM-encoded client certificate for TLS authentication to the Vault server
+            * VAULT_CACERT: path to a PEM-encoded CA cert file to use to verify the Vault server TLS certificate
+            * VAULT_CAPATH: path to a directory of PEM-encoded CA cert files to verify the Vault server TLS certificate
+            * VAULT_NAMESPACE: specify the Vault Namespace, if you have one
+        """
+        client_parameters = {}
+
+        # fetch missing values from env
+        variables = ["ADDR", "NAMESPACE", "SKIP_VERIFY", "CACERT", "CAPATH", "CLIENT_KEY", "CLIENT_CERT"]
+        for var in variables:
+            var = "VAULT_" + var
+            if self.vault_parameters.get(var) is None and os.getenv(var) is not None:
+                self.vault_parameters[var] = os.getenv(var)
+
+        # set vault adress
+        vault_address = self.vault_parameters.get("VAULT_ADDR")
+        if not vault_address:
+            raise VaultError("VAULT_ADDR has to be specified in inventory or env")
+        else:
+            client_parameters["url"] = vault_address
+
+        # set vault namespace
+        vault_namespace = self.vault_parameters.get("VAULT_NAMESPACE")
+        if vault_namespace:
+            client_parameters["namespace"] = vault_namespace
+
+        # set ca cert/path or skip verification
+        skip_verify_value = self.vault_parameters.get("VAULT_SKIP_VERIFY", False)
+        skip_verify_bool = not (str(skip_verify_value).lower() == "false")
+        if skip_verify_bool:
+            # disable ssl warnings
+            import urllib3
+
+            urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
+            client_parameters["verify"] = False
         else:
-            client_parameters["verify"] = cert
-    else:
-        client_parameters["verify"] = False
-
-    # CLIENT CERTIFICATE FOR TLS AUTHENTICATION
-    client_key = parameter.get("VAULT_CLIENT_KEY", os.getenv("VAULT_CLIENT_KEY", default=""))
-    client_cert = parameter.get("VAULT_CLIENT_CERT", os.getenv("VAULT_CLIENT_CERT", default=""))
-    if client_key != "" and client_cert != "":
-        client_parameters["cert"] = (client_cert, client_key)
-    return client_parameters
+            cert = self.vault_parameters.get("VAULT_CACERT")
+            cert_path = self.vault_parameters.get("VAULT_CAPATH")
+            if cert and os.path.isfile(cert):
+                client_parameters["verify"] = cert
+            elif cert_path and os.path.isdir(cert_path):
+                client_parameters["verify"] = cert_path
+            else:
+                raise VaultError(
+                    "Neither VAULT_CACERT nor VAULT_CAPATH specified or found. If you don't want to verify the requests, set VAULT_SKIP_VERIFY to 'true'."
+                )
+
+        # set client cert for tls authentication
+        client_key = self.vault_parameters.get("VAULT_CLIENT_KEY")
+        client_cert = self.vault_parameters.get("VAULT_CLIENT_CERT")
+        if client_key and client_cert:
+            client_parameters["cert"] = (client_cert, client_key)
+
+        return client_parameters

tests/test_cli.py

@@ -16,7 +16,7 @@
 import unittest
 from unittest.mock import patch
 
-from kapitan.cli import main, build_parser
+from kapitan.cli import build_parser, main
 from kapitan.refs.secrets.vaultkv import VaultSecret
 from tests.vault_server import VaultServer
 
@@ -29,30 +29,6 @@
     os.environ["GNUPGHOME"] = GNUPGHOME
 
 
-@contextlib.contextmanager
-def set_env(**environ):
-    """
-    Temporarily set the process environment variables.
-
-    >>> with set_env(PLUGINS_DIR='test/plugins'):
-    ...   "PLUGINS_DIR" in os.environ
-    True
-
-    >>> "PLUGINS_DIR" in os.environ
-    False
-
-    :type environ: dict[str, unicode]
-    :param environ: Environment variables to set
-    """
-    old_environ = dict(os.environ)
-    os.environ.update(environ)
-    try:
-        yield
-    finally:
-        os.environ.clear()
-        os.environ.update(old_environ)
-
-
 class CliFuncsTest(unittest.TestCase):
     @classmethod
     def setUpClass(cls):
@@ -720,8 +696,9 @@ def test_cli_secret_write_vault(self, mock_reveal):
             "--vault-key",
             "testkey",
         ]
-        with set_env(VAULT_ADDR=self.server.vault_url):
-            main()
+        env = {"VAULT_ADDR": self.server.vault_url, "VAULT_SKIP_VERIFY": "True"}
+        os.environ.update(**env)
+        main()
 
         test_tag_content = "revealing: ?{vaultkv:test_secret}"
         test_tag_file = tempfile.mktemp()
@@ -733,7 +710,7 @@ def test_cli_secret_write_vault(self, mock_reveal):
 
         # set stdout as string
         stdout = io.StringIO()
-        with contextlib.redirect_stdout(stdout), set_env(VAULT_ADDR=self.server.vault_url):
+        with contextlib.redirect_stdout(stdout):
             main()
         self.assertEqual("revealing: {value}".format(value=test_secret_content), stdout.getvalue())
 

tests/vault_server.py

@@ -35,6 +35,9 @@ def __init__(self):
         self.vault_container = self.setup_container()
         self.setup_vault()
 
+        # disable tls verify for all clients
+        self.parameters["VAULT_SKIP_VERIFY"] = True
+
     def setup_container(self):
         env = {
             "VAULT_LOCAL_CONFIG": '{"backend": {"file": {"path": "/vault/file"}}, "listener":{"tcp":{"address":"0.0.0.0:8200","tls_disable":"true"}}}'