[WIP] Secrets in Nautobot Config

README.md

@@ -11,7 +11,7 @@ This plugin supports the following popular secrets backends:
 | Secrets Backend                                              | Supported Secret Types                                       | Supported Authentication Methods                             |
 | ------------------------------------------------------------ | ------------------------------------------------------------ | ------------------------------------------------------------ |
 | [AWS Secrets Manager](https://aws.amazon.com/secrets-manager/) | [Other: Key/value pairs](https://docs.aws.amazon.com/secretsmanager/latest/userguide/manage_create-basic-secret.html) | [AWS credentials](https://docs.aws.amazon.com/general/latest/gr/aws-sec-cred-types.html) (see Usage section below) |
-| [HashiCorp Vault](https://www.vaultproject.io)               | [K/V Version 2](https://www.vaultproject.io/docs/secrets/kv/kv-v2) | [Token](https://www.vaultproject.io/docs/auth/token)         |
+| [HashiCorp Vault](https://www.vaultproject.io)               | [K/V Version 2](https://www.vaultproject.io/docs/secrets/kv/kv-v2) | [Token](https://www.vaultproject.io/docs/auth/token)<br/>[AppRole](https://www.vaultproject.io/docs/auth/approle)<br/>[AWS](https://www.vaultproject.io/docs/auth/aws)<br/>[Kubernetes](https://www.vaultproject.io/docs/auth/kubernetes)         |
 | [Thycotic Secret Server](https://thycotic.com/)               | [Secret Server Cloud](https://github.com/thycotic/python-tss-sdk#secret-server-cloud)<br/>[Secret Server (on-prem)](https://github.com/thycotic/python-tss-sdk#secret-server)| [Access Token Authorization](https://github.com/thycotic/python-tss-sdk#access-token-authorization)<br/>[Domain Authorization](https://github.com/thycotic/python-tss-sdk#domain-authorization)<br/>[Password Authorization](https://github.com/thycotic/python-tss-sdk#password-authorization)<br/>         |
 
 ## Screenshots
@@ -162,10 +162,14 @@ PLUGINS_CONFIG = {
 ```
 
 - `url` - (required) The URL to the HashiCorp Vault instance (e.g. `http://localhost:8200`).
-- `auth_method` - (optional / defaults to "token") The method used to authenticate against the HashiCorp Vault instance. Either `"token"` or `"approle"`.
+- `auth_method` - (optional / defaults to "token") The method used to authenticate against the HashiCorp Vault instance. Either `"approle"`, `"aws"`, `"kubernetes"` or `"token"`.
+- `ca_cert` - (optional) Path to a PEM formatted CA certificatee to use when verifying the vault connection.  Set to `False` to ignore ssl verification (not recommended) or `True` to use the system certificates.
+- `k8s_token_path` - (optional) Path to the kubernetes service account token file.  Defaults to "/var/run/secrets/kubernetes.io/serviceaccount/token".
 - `token` - (optional) Required when `"auth_method": "token"` or `auth_method` is not supplied. The token for authenticating the client with the HashiCorp Vault instance. As with other sensitive service credentials, we recommend that you provide the token value as an environment variable and retrieve it with `{"token": os.getenv("NAUTOBOT_HASHICORP_VAULT_TOKEN")}` rather than hard-coding it in your `nautobot_config.py`.
+- `role_name` - (optional) Required when `"auth_method": "kubernetes"`.  The Vault Kubernetes role to assume which the pod's service account has access to.
 - `role_id` - (optional) Required when `"auth_method": "approle"`. As with other sensitive service credentials, we recommend that you provide the role_id value as an environment variable and retrieve it with `{"role_id": os.getenv("NAUTOBOT_HASHICORP_VAULT_ROLE_ID")}` rather than hard-coding it in your `nautobot_config.py`.
 - `secret_id` - (optional) Required when `"auth_method": "approle"`.As with other sensitive service credentials, we recommend that you provide the secret_id value as an environment variable and retrieve it with `{"secret_id": os.getenv("NAUTOBOT_HASHICORP_VAULT_SECRET_ID")}` rather than hard-coding it in your `nautobot_config.py`.
+- `login_kwargs` - (optional) Additional optional parameters to pass to the login method for [`approle`](https://hvac.readthedocs.io/en/stable/source/hvac_api_auth_methods.html#hvac.api.auth_methods.AppRole.login), [`aws`](https://hvac.readthedocs.io/en/stable/source/hvac_api_auth_methods.html#hvac.api.auth_methods.Aws.iam_login) and [`kubernetes`](https://hvac.readthedocs.io/en/stable/source/hvac_api_auth_methods.html#hvac.api.auth_methods.Kubernetes.login) authentication methods.
 
 ### Thycotic Secret Server (TSS)
 

development/nautobot_config.py

@@ -10,6 +10,7 @@
 from distutils.util import strtobool
 from django.core.exceptions import ImproperlyConfigured
 from nautobot.core import settings
+from nautobot_secrets_providers.connectors import HashiCorpVaultConnector
 
 # Enforce required configuration parameters
 for key in [
@@ -136,7 +137,7 @@ def is_truthy(arg):
 
 # Optionally display a persistent banner at the top and/or bottom of every page. HTML is allowed. To display the same
 # content in both banners, define BANNER_TOP and set BANNER_BOTTOM = BANNER_TOP.
-BANNER_TOP = os.environ.get("BANNER_TOP", "")
+# BANNER_TOP = os.environ.get("BANNER_TOP", "")
 BANNER_BOTTOM = os.environ.get("BANNER_BOTTOM", "")
 
 # Text to include on the login page above the login form. HTML is allowed.
@@ -285,6 +286,9 @@ def is_truthy(arg):
     },
 }
 
+hvac_connector = HashiCorpVaultConnector(PLUGINS_CONFIG["nautobot_secrets_providers"]["hashicorp_vault"])
+BANNER_TOP = hvac_connector.get_kv_value("banner_top", path="nautobot", mount_point="secret")
+
 # When determining the primary IP address for a device, IPv6 is preferred over IPv4 by default. Set this to True to
 # prefer IPv4 instead.
 PREFER_IPV4 = is_truthy(os.environ.get("PREFER_IPV4", False))

nautobot_secrets_providers/connectors/__init__.py

@@ -0,0 +1,7 @@
+"""Nautobot Secrets Connectors."""
+
+from .hashicorp import HashiCorpVaultConnector
+
+__all__ = (  # type: ignore
+    HashiCorpVaultConnector,  # pylint: disable=invalid-all-object
+)

nautobot_secrets_providers/connectors/exceptions.py

@@ -0,0 +1,29 @@
+"""Exception classes specific to secrets."""
+
+
+class ConnectorError(Exception):
+    """General purpose exception class for failures raised when secret value access fails."""
+
+    def __init__(self, connector_class, message, *args, **kwargs):
+        """Collect some additional data."""
+        super().__init__(message, *args, **kwargs)
+        self.connector_class = connector_class
+        self.message = message
+
+    def __str__(self):
+        """Format the Exception as a string."""
+        return f"{self.__class__.__name__}: " f'(connector "{self.connector_class.__name__}"): {self.message}'
+
+
+class SecretValueNotFoundError(Exception):
+    """General purpose exception class for failures raised when secret value access fails."""
+
+    def __init__(self, connector_class, message, *args, **kwargs):
+        """Collect some additional data."""
+        super().__init__(message, *args, **kwargs)
+        self.connector_class = connector_class
+        self.message = message
+
+    def __str__(self):
+        """Format the Exception as a string."""
+        return f"{self.__class__.__name__}: " f'(connector "{self.connector_class.__name__}"): {self.message}'

nautobot_secrets_providers/connectors/hashicorp.py

@@ -0,0 +1,135 @@
+"""Generic connector for HashiCorp Vault not tied to django."""
+from .exceptions import ConnectorError, SecretValueNotFoundError
+from os.path import exists
+
+try:
+    import boto3
+except ImportError:
+    boto3 = None
+
+import hvac
+
+AUTH_METHOD_CHOICES = ["approle", "aws", "kubernetes", "token"]
+DEFAULT_MOUNT_POINT = hvac.api.secrets_engines.kv_v2.DEFAULT_MOUNT_POINT
+K8S_TOKEN_DEFAULT_PATH = "/var/run/secrets/kubernetes.io/serviceaccount/token"  # nosec B105
+
+
+class HashiCorpVaultConnector:
+    """A generic connector for HashiCorp Vault."""
+
+    vault_settings = {}
+    validated = False
+    auth_method = ""
+    client = None
+
+    @classmethod
+    def __init__(cls, vault_settings):
+        """Set some defaults."""
+        cls.vault_settings = vault_settings
+        cls.auth_method = vault_settings.get("auth_method", "token")
+        cls.k8s_token_path = vault_settings.get("k8s_token_path", K8S_TOKEN_DEFAULT_PATH)
+        if hvac and not cls.validated:
+            cls.validate_vault_settings()
+
+    @classmethod
+    def validate_vault_settings(cls):
+        """Validate the vault settings."""
+        if "url" not in cls.vault_settings:
+            raise ConnectorError(cls, "HashiCorp Vault configuration is missing a url")
+
+        if cls.auth_method not in AUTH_METHOD_CHOICES:
+            raise ConnectorError(cls, f"HashiCorp Vault Auth Method {cls.auth_method} is invalid!")
+
+        if cls.auth_method == "aws":
+            if not boto3:
+                raise ConnectorError(cls, "HashiCorp Vault AWS Auth Method requires the boto3 plugin!")
+        elif cls.auth_method == "token":
+            if "token" not in cls.vault_settings:
+                raise ConnectorError(cls, "HashiCorp Vault configuration is missing a token for token authentication!")
+        elif cls.auth_method == "kubernetes":
+            if "role_name" not in cls.vault_settings:
+                raise ConnectorError(
+                    cls, "HashiCorp Vault configuration is missing a role name for kubernetes authentication!"
+                )
+            if not exists(cls.k8s_token_path):
+                raise ConnectorError(
+                    cls, "HashiCorp Vault configuration is missing a role name for kubernetes authentication!"
+                )
+        elif cls.auth_method == "approle":
+            if "role_id" not in cls.vault_settings or "secret_id" not in cls.vault_settings:
+                raise ConnectorError(cls, "HashiCorp Vault configuration is missing a role_id and/or secret_id!")
+        cls.validated = True
+
+    @classmethod
+    def get_client(cls):
+        """Get a hvac client and login."""
+        if not cls.validated:
+            cls.validate_vault_settings()
+        elif cls.client:
+            return cls.client
+
+        login_kwargs = cls.vault_settings.get("login_kwargs", {})
+
+        # According to the docs (https://hvac.readthedocs.io/en/stable/source/hvac_v1.html?highlight=verify#hvac.v1.Client.__init__)
+        # the client verify parameter is either a boolean or a path to a ca certificate file to verify.  This is non-intuitive
+        # so we use a parameter to specify the path to the ca_cert, if not provided we use the default of None
+        ca_cert = cls.vault_settings.get("ca_cert", None)
+
+        # Get the client and attempt to retrieve the secret.
+        try:
+            if cls.auth_method == "token":
+                cls.client = hvac.Client(
+                    url=cls.vault_settings["url"], token=cls.vault_settings["token"], verify=ca_cert
+                )
+            else:
+                cls.client = hvac.Client(url=cls.vault_settings["url"], verify=ca_cert)
+                if cls.auth_method == "approle":
+                    cls.client.auth.approle.login(
+                        role_id=cls.vault_settings["role_id"],
+                        secret_id=cls.vault_settings["secret_id"],
+                        **login_kwargs,
+                    )
+                elif cls.auth_method == "kubernetes":
+                    with open(cls.k8s_token_path, "r", encoding="utf-8") as token_file:
+                        jwt = token_file.read()
+                    cls.client.auth.kubernetes.login(role=cls.vault_settings["role_name"], jwt=jwt, **login_kwargs)
+                elif cls.auth_method == "aws":
+                    session = boto3.Session()
+                    aws_creds = session.get_credentials()
+                    cls.client.auth.aws.iam_login(
+                        access_key=aws_creds.access_key,
+                        secret_key=aws_creds.secret_key,
+                        session_token=aws_creds.token,
+                        region=session.region_name,
+                        role=cls.vault_settings.get("role_name", None),
+                        **login_kwargs,
+                    )
+        except hvac.exceptions.InvalidRequest as err:
+            raise ConnectorError(
+                cls, f"HashiCorp Vault Login failed (auth_method: {cls.auth_method}). Error: {err}"
+            ) from err
+        except hvac.exceptions.Forbidden as err:
+            raise ConnectorError(
+                cls, f"HashiCorp Vault Access Denied (auth_method: {cls.auth_method}). Error: {err}"
+            ) from err
+
+        return cls.client
+
+    @classmethod
+    def get_kv_value(cls, secret_key, default_value=None, **kwargs):
+        """Get a kv key and return the value stored in secret_key."""
+        client = cls.get_client()
+        try:
+            response = client.secrets.kv.read_secret(**kwargs)
+        except hvac.exceptions.InvalidPath as err:
+            raise ConnectorError(cls, str(err)) from err
+
+        # Retrieve the value using the key or complain loudly.
+        try:
+            return response["data"]["data"][secret_key]
+        except KeyError as err:
+            if default_value:
+                return default_value
+            else:
+                msg = f"The secret value could not be retrieved using key {err}"
+                raise SecretValueNotFoundError(cls, msg) from err

nautobot_secrets_providers/providers/hashicorp.py

@@ -3,6 +3,11 @@
 from django import forms
 from django.conf import settings
 
+try:
+    import boto3
+except ImportError:
+    boto3 = None
+
 try:
     import hvac
 
@@ -13,15 +18,22 @@
 from nautobot.utilities.forms import BootstrapMixin
 from nautobot.extras.secrets import exceptions, SecretsProvider
 
+from nautobot_secrets_providers.connectors import HashiCorpVaultConnector
+from nautobot_secrets_providers.connectors.exceptions import ConnectorError, SecretValueNotFoundError
+
 __all__ = ("HashiCorpVaultSecretsProvider",)
 
+K8S_TOKEN_DEFAULT_PATH = "/var/run/secrets/kubernetes.io/serviceaccount/token"  # nosec B105
+AUTH_METHOD_CHOICES = ["approle", "aws", "kubernetes", "token"]
+
 
 class HashiCorpVaultSecretsProvider(SecretsProvider):
     """A secrets provider for HashiCorp Vault."""
 
     slug = "hashicorp-vault"
     name = "HashiCorp Vault"
     is_available = hvac is not None
+    connector = None
 
     class ParametersForm(BootstrapMixin, forms.Form):
         """Required parameters for HashiCorp Vault."""
@@ -41,19 +53,24 @@ class ParametersForm(BootstrapMixin, forms.Form):
         )
 
     @classmethod
-    def get_value_for_secret(cls, secret, obj=None, **kwargs):
-        """Return the value stored under the secret’s key in the secret’s path."""
+    def validate_vault_settings(cls, secret):
+        """Validate the vault settings."""
         # This is only required for HashiCorp Vault therefore not defined in
         # `required_settings` for the plugin config.
         plugin_settings = settings.PLUGINS_CONFIG["nautobot_secrets_providers"]
         if "hashicorp_vault" not in plugin_settings:
             raise exceptions.SecretProviderError(secret, cls, "HashiCorp Vault is not configured!")
 
-        vault_settings = plugin_settings["hashicorp_vault"]
-
-        if "url" not in vault_settings:
-            raise exceptions.SecretProviderError(secret, cls, "HashiCorp Vault configuration is missing a url")
+        vault_settings = plugin_settings.get("hashicorp_vault")
+        if not cls.connector:
+            try:
+                cls.connector = HashiCorpVaultConnector(vault_settings)
+            except ConnectorError as err:
+                raise exceptions.SecretProviderError(secret, cls, str(err)) from err
 
+    @classmethod
+    def get_value_for_secret(cls, secret, obj=None, **kwargs):
+        """Return the value stored under the secret’s key in the secret’s path."""
         # Try to get parameters and error out early.
         parameters = secret.rendered_parameters(obj=obj)
         try:
@@ -64,47 +81,12 @@ def get_value_for_secret(cls, secret, obj=None, **kwargs):
             msg = f"The secret parameter could not be retrieved for field {err}"
             raise exceptions.SecretParametersError(secret, cls, msg) from err
 
-        # default to token authentication
-        auth_method = vault_settings.get("auth_method", "token")
-
-        # Get the client and attempt to retrieve the secret.
-        if auth_method == "token":
-            try:
-                client = hvac.Client(url=vault_settings["url"], token=vault_settings["token"])
-            except KeyError as err:
-                raise exceptions.SecretProviderError(
-                    secret, cls, "HashiCorp Vault configuration is missing a token"
-                ) from err
-            except hvac.exceptions.InvalidRequest as err:
-                raise exceptions.SecretProviderError(secret, cls, "HashiCorp Vault invalid token") from err
-        elif auth_method == "approle":
-            try:
-                client = hvac.Client(url=vault_settings["url"])
-                client.auth.approle.login(
-                    role_id=vault_settings["role_id"],
-                    secret_id=vault_settings["secret_id"],
-                )
-            except KeyError as err:
-                raise exceptions.SecretProviderError(
-                    secret, cls, "HashiCorp Vault configuration is missing a role_id and/or secret_id"
-                ) from err
-            except hvac.exceptions.InvalidRequest as err:
-                raise exceptions.SecretProviderError(
-                    secret, cls, "HashiCorp Vault invalid role_id and/or secret_id"
-                ) from err
-        else:
-            raise exceptions.SecretProviderError(
-                secret, cls, f'HashiCorp Vault configuration "{auth_method}" is not a valid auth_method'
-            )
+        if not cls.connector:
+            cls.validate_vault_settings(secret)
 
         try:
-            response = client.secrets.kv.read_secret(path=secret_path, mount_point=secret_mount_point)
-        except hvac.exceptions.InvalidPath as err:
+            cls.connector.get_kv_value(secret_key=secret_key, path=secret_path, mount_point=secret_mount_point)
+        except ConnectorError as err:
+            raise exceptions.SecretProviderError(secret, cls, str(err)) from err
+        except SecretValueNotFoundError as err:
             raise exceptions.SecretValueNotFoundError(secret, cls, str(err)) from err
-
-        # Retrieve the value using the key or complain loudly.
-        try:
-            return response["data"]["data"][secret_key]
-        except KeyError as err:
-            msg = f"The secret value could not be retrieved using key {err}"
-            raise exceptions.SecretValueNotFoundError(secret, cls, msg) from err