Initial commit for vaultpy package

LICENSE

@@ -0,0 +1 @@
+Copyright (c) 2021, DirectEmployers Association -  https://directemployers.org

README.md

@@ -0,0 +1,24 @@
+# Vaultpy
+A module to parse injected [Vault](https://www.vaultproject.io/) secrets and track their usage with Datadog.
+
+## Requirements
+- Local Datadog agent
+- Environment variables to access it
+- Set [container annotations](https://www.vaultproject.io/docs/platform/k8s/injector/annotations) to inject Vault secrets in K8s
+
+## Setup
+For production, use the `VAULT_SECRETS_PATH` environment variable to set the path to the secrets that are injected by the Vault Agent in Kubernetes.
+
+For local development, a `de_secrets.py` can be used to load secrets in a format not unlike Django settings.
+
+## Usage
+Import `vault.secrets` and then access the loaded secrets using by accessing dynamic properties loaded into the `secrets` object (i.e. `secrets.FOO`).
+
+Example of usage in a settings file:
+```python
+from vault import secrets
+
+FOO = secrets.FOO
+BAR = getattr(secrets, "BAR", "")
+BAZ = getattr(secrets, "BAZ")
+```

build/lib/vault/__init__.py

@@ -0,0 +1,85 @@
+import logging
+from base64 import b64decode
+from importlib import import_module
+from json import loads
+from os import environ
+from typing import Dict
+
+from datadog import statsd
+
+logger = logging.getLogger(__name__)
+
+
+def _load_de_secrets() -> Dict:
+    """
+    Imports de_secrets module and returns a dictionary of its attributes.
+    """
+    de_secrets = import_module("de_secrets")
+    return {k: getattr(de_secrets, k) for k in dir(de_secrets) if not k.startswith("_")}
+
+
+def _load_vault_secrets() -> Dict:
+    """
+    Load Vault injected secrets file located at VAULT_SECRETS_PATH, then perform
+    base 64 decode followed by JSON decode on file contents. This function
+    should not be called anywhere except within this module!
+    """
+    with open(environ["VAULT_SECRETS_PATH"]) as file:
+        contents = file.read()
+
+    json_secrets = b64decode(contents)
+    return loads(json_secrets)
+
+
+def _get_secrets() -> Dict:
+    """
+    Get secrets from de_secrets.py in local dev, or from Vault injected secrets file
+    located at path in VAULT_SECRETS_PATH. Performs base 64 decode followed by JSON
+    decode on file contents.
+    """
+    if not environ.get("USE_VAULT"):
+        # Use dev secrets when available.
+        return _load_de_secrets()
+
+    return _load_vault_secrets()
+
+
+class VaultSecretsWrapper:
+    """
+    Provide access to secrets as attributes and send secret-usage analytics to Datadog.
+    """
+
+    def __init__(self, secrets: Dict):
+        self._keys = secrets.keys()
+        self._env = environ.get("DD_ENV")
+
+        for key, value in secrets.items():
+            statsd.increment(
+                "vault.secrets.access_count",
+                value=1,
+                tags=[f"env:{self._env}", f"secret_key:{key}"],
+            )
+            setattr(self, key, value)
+
+    def __getattribute__(self, key: str):
+        """
+        Override the default getattribute method so that we can track secret key
+        usage with Datadog. Non-secret attributes are passed on to the default method.
+        """
+        if key not in ["_keys", "_env"] and key in self._keys:
+            try:
+                statsd.increment(
+                    "vault.secrets.access_count",
+                    value=1,
+                    tags=[f"env:{self._env}", f"secret_key:{key}"],
+                )
+                return super().__getattribute__(key)
+            except AttributeError as error:
+                logger.error(f"Requested secret could not be loaded: {key}")
+                raise error
+
+        return super().__getattribute__(key)
+
+
+secrets = VaultSecretsWrapper(_get_secrets())
+__all__ = ("secrets",)

pyproject.toml

@@ -0,0 +1,6 @@
+[build-system]
+requires = [
+    "setuptools>=42",
+    "wheel"
+]
+build-backend = "setuptools.build_meta"

setup.cfg

@@ -0,0 +1,13 @@
+[metadata]
+name = vaultpy
+version = 0.0.1
+author = Tim Loyer
+author_email = [email protected]
+description = A module to parse injected Vault secrets and track their usage with Datadog.
+long_description = file: README.md
+long_description_content_type = text/markdown
+url = https://github.com/DirectEmployers/vaultpy
+
+[options]
+packages = find:
+python_requires = >=3.6

vault/__init__.py

@@ -0,0 +1,85 @@
+import logging
+from base64 import b64decode
+from importlib import import_module
+from json import loads
+from os import environ
+from typing import Dict
+
+from datadog import statsd
+
+logger = logging.getLogger(__name__)
+
+
+def _load_de_secrets() -> Dict:
+    """
+    Imports de_secrets module and returns a dictionary of its attributes.
+    """
+    de_secrets = import_module("de_secrets")
+    return {k: getattr(de_secrets, k) for k in dir(de_secrets) if not k.startswith("_")}
+
+
+def _load_vault_secrets() -> Dict:
+    """
+    Load Vault injected secrets file located at VAULT_SECRETS_PATH, then perform
+    base 64 decode followed by JSON decode on file contents. This function
+    should not be called anywhere except within this module!
+    """
+    with open(environ["VAULT_SECRETS_PATH"]) as file:
+        contents = file.read()
+
+    json_secrets = b64decode(contents)
+    return loads(json_secrets)
+
+
+def _get_secrets() -> Dict:
+    """
+    Get secrets from de_secrets.py in local dev, or from Vault injected secrets file
+    located at path in VAULT_SECRETS_PATH. Performs base 64 decode followed by JSON
+    decode on file contents.
+    """
+    if not environ.get("USE_VAULT"):
+        # Use dev secrets when available.
+        return _load_de_secrets()
+
+    return _load_vault_secrets()
+
+
+class VaultSecretsWrapper:
+    """
+    Provide access to secrets as attributes and send secret-usage analytics to Datadog.
+    """
+
+    def __init__(self, secrets: Dict):
+        self._keys = secrets.keys()
+        self._env = environ.get("DD_ENV")
+
+        for key, value in secrets.items():
+            statsd.increment(
+                "vault.secrets.access_count",
+                value=1,
+                tags=[f"env:{self._env}", f"secret_key:{key}"],
+            )
+            setattr(self, key, value)
+
+    def __getattribute__(self, key: str):
+        """
+        Override the default getattribute method so that we can track secret key
+        usage with Datadog. Non-secret attributes are passed on to the default method.
+        """
+        if key not in ["_keys", "_env"] and key in self._keys:
+            try:
+                statsd.increment(
+                    "vault.secrets.access_count",
+                    value=1,
+                    tags=[f"env:{self._env}", f"secret_key:{key}"],
+                )
+                return super().__getattribute__(key)
+            except AttributeError as error:
+                logger.error(f"Requested secret could not be loaded: {key}")
+                raise error
+
+        return super().__getattribute__(key)
+
+
+secrets = VaultSecretsWrapper(_get_secrets())
+__all__ = ("secrets",)

vaultpy.egg-info/PKG-INFO

@@ -0,0 +1,21 @@
+Metadata-Version: 2.1
+Name: vaultpy
+Version: 0.0.1
+Summary: A module to parse injected Vault secrets and track their usage with Datadog.
+Home-page: https://github.com/DirectEmployers/vaultpy
+Author: Tim Loyer
+Author-email: [email protected]
+License: UNKNOWN
+Description: # Vaultpy
+        A module to parse injected [Vault](https://www.vaultproject.io/) secrets and track their usage with Datadog.
+
+        ## Requirements
+        - Local Datadog agent
+        - Environment variables to access it
+
+        ## Usage
+
+
+Platform: UNKNOWN
+Requires-Python: >=3.6
+Description-Content-Type: text/markdown

vaultpy.egg-info/SOURCES.txt

@@ -0,0 +1,8 @@
+README.md
+pyproject.toml
+setup.cfg
+vault/__init__.py
+vaultpy.egg-info/PKG-INFO
+vaultpy.egg-info/SOURCES.txt
+vaultpy.egg-info/dependency_links.txt
+vaultpy.egg-info/top_level.txt

vaultpy.egg-info/dependency_links.txt

@@ -0,0 +1 @@
+

vaultpy.egg-info/top_level.txt

@@ -0,0 +1 @@
+vault