Feat: helper script to read KeyVault secrets

.devcontainer/Dockerfile

@@ -1,5 +1,11 @@
 FROM benefits_client:latest
 
+# install Azure CLI
+# https://learn.microsoft.com/en-us/cli/azure/install-azure-cli-linux?pivots=apt
+USER root
+RUN curl -sL https://aka.ms/InstallAzureCLIDeb | bash
+USER $USER
+
 # install devcontainer requirements
 RUN pip install -e .[dev,test]
 

benefits/secrets.py

@@ -0,0 +1,37 @@
+import sys
+
+from django.conf import settings
+
+from azure.identity import DefaultAzureCredential
+from azure.keyvault.secrets import SecretClient
+
+
+KEY_VAULT_URL = "https://kv-cdt-pub-calitp-{env}-001.vault.azure.net/"
+
+
+def get_secret_by_name(secret_name, client=None):
+    if client is None:
+        # construct the KeyVault URL from the runtime environment
+        # see https://docs.calitp.org/benefits/deployment/infrastructure/#environments
+        # and https://github.com/cal-itp/benefits/blob/dev/terraform/key_vault.tf
+        runtime_env = settings.RUNTIME_ENVIRONMENT()
+        vault_url = KEY_VAULT_URL.format(env=runtime_env[0])
+
+        credential = DefaultAzureCredential()
+        client = SecretClient(vault_url=vault_url, credential=credential)
+
+    secret = client.get_secret(secret_name)
+    return secret.value
+
+
+if __name__ == "__main__":
+    args = sys.argv[1:]
+    if len(args) < 1:
+        print("Provide the name of the secret to read")
+        exit(1)
+
+    secret_name = args[0]
+    secret_value = get_secret_by_name(secret_name)
+
+    print(f"[{settings.RUNTIME_ENVIRONMENT()}] {secret_name}: {secret_value}")
+    exit(0)

benefits/sentry.py

@@ -3,6 +3,7 @@
 import os
 import subprocess
 
+from django.conf import settings
 import sentry_sdk
 from sentry_sdk.integrations.django import DjangoIntegration
 from sentry_sdk.scrubber import EventScrubber, DEFAULT_DENYLIST
@@ -11,7 +12,6 @@
 
 logger = logging.getLogger(__name__)
 
-SENTRY_ENVIRONMENT = os.environ.get("SENTRY_ENVIRONMENT", "local")
 SENTRY_CSP_REPORT_URI = None
 
 
@@ -80,19 +80,21 @@ def get_traces_sample_rate():
 
 
 def configure():
-    SENTRY_DSN = os.environ.get("SENTRY_DSN")
-    if SENTRY_DSN:
+    sentry_dsn = os.environ.get("SENTRY_DSN")
+    sentry_environment = os.environ.get("SENTRY_ENVIRONMENT", settings.RUNTIME_ENVIRONMENT())
+
+    if sentry_dsn:
         release = get_release()
-        logger.info(f"Enabling Sentry for environment '{SENTRY_ENVIRONMENT}', release '{release}'...")
+        logger.info(f"Enabling Sentry for environment '{sentry_environment}', release '{release}'...")
 
         # https://docs.sentry.io/platforms/python/configuration/
         sentry_sdk.init(
-            dsn=SENTRY_DSN,
+            dsn=sentry_dsn,
             integrations=[
                 DjangoIntegration(),
             ],
             traces_sample_rate=get_traces_sample_rate(),
-            environment=SENTRY_ENVIRONMENT,
+            environment=sentry_environment,
             release=release,
             in_app_include=["benefits"],
             # send_default_pii must be False (the default) for a custom EventScrubber/denylist

benefits/settings.py

@@ -4,6 +4,8 @@
 
 import os
 
+from django.conf import settings
+
 from benefits import sentry
 
 
@@ -24,6 +26,20 @@ def _filter_empty(ls):
 
 ALLOWED_HOSTS = _filter_empty(os.environ.get("DJANGO_ALLOWED_HOSTS", "localhost,127.0.0.1").split(","))
 
+
+def RUNTIME_ENVIRONMENT():
+    """Helper calculates the current runtime environment from ALLOWED_HOSTS."""
+
+    # usage of django.conf.settings.ALLOWED_HOSTS here (rather than the module variable directly)
+    # is to ensure dynamic calculation, e.g. for unit tests and elsewhere this setting is needed
+    env = "dev"
+    if "test-benefits.calitp.org" in settings.ALLOWED_HOSTS:
+        env = "test"
+    elif "benefits.calitp.org" in settings.ALLOWED_HOSTS:
+        env = "prod"
+    return env
+
+
 # Application definition
 
 INSTALLED_APPS = [

benefits/urls.py

@@ -8,6 +8,7 @@
 import logging
 
 from django.conf import settings
+from django.http import HttpResponse
 from django.urls import include, path
 
 logger = logging.getLogger(__name__)
@@ -34,6 +35,18 @@ def trigger_error(request):
 
     urlpatterns.append(path("error/", trigger_error))
 
+    # simple route to read a pre-defined "secret"
+    # this "secret" does not contain sensitive information
+    # and is only configured in the dev environment for testing/debugging
+
+    def test_secret(request):
+        from benefits.secrets import get_secret_by_name
+
+        return HttpResponse(get_secret_by_name("testsecret"))
+
+    urlpatterns.append(path("testsecret/", test_secret))
+
+
 if settings.ADMIN:
     from django.contrib import admin
 

docs/configuration/environment-variables.md

@@ -219,7 +219,9 @@ Enables [sending events to Sentry](../../deployment/troubleshooting/#error-monit
 
     [`environment` config value](https://docs.sentry.io/platforms/python/configuration/options/#environment)
 
-Segments errors by which deployment they occur in. This defaults to `local`, and can be set to match one of the [environment names](../../deployment/infrastructure/#environments).
+Segments errors by which deployment they occur in. This defaults to `dev`, and can be set to match one of the [environment names](../../deployment/infrastructure/#environments).
+
+`local` may also be used for local testing of the Sentry integration.
 
 ### `SENTRY_REPORT_URI`
 

pyproject.toml

@@ -8,6 +8,8 @@ classifiers = ["Programming Language :: Python :: 3 :: Only"]
 requires-python = ">=3.9"
 dependencies = [
     "Authlib==1.3.0",
+    "azure-keyvault-secrets==4.7.0",
+    "azure-identity==1.15.0",
     "Django==5.0.1",
     "django-csp==3.7",
     "eligibility-api==2023.9.1",

tests/pytest/test_secrets.py

@@ -0,0 +1,45 @@
+import pytest
+
+from benefits.secrets import KEY_VAULT_URL, get_secret_by_name
+
+
+@pytest.fixture(autouse=True)
+def mock_DefaultAzureCredential(mocker):
+    # patching the class to ensure new instances always return the same mock
+    credential_cls = mocker.patch("benefits.secrets.DefaultAzureCredential")
+    credential_cls.return_value = mocker.Mock()
+    return credential_cls
+
+
+def test_get_secret_by_name__with_client__returns_value(mocker):
+    secret_name = "the secret name"
+    secret_value = "the secret value"
+    client = mocker.patch("benefits.secrets.SecretClient")
+    client.get_secret.return_value = mocker.Mock(value=secret_value)
+
+    actual_value = get_secret_by_name(secret_name, client)
+
+    client.get_secret.assert_called_once_with(secret_name)
+    assert actual_value == secret_value
+
+
+def test_get_secret_by_name__None_client__returns_value(mocker, settings, mock_DefaultAzureCredential):
+    secret_name = "the secret name"
+    secret_value = "the secret value"
+
+    # override runtime to dev
+    settings.RUNTIME_ENVIRONMENT = lambda: "dev"
+    expected_keyvault_url = KEY_VAULT_URL.format(env="d")
+
+    # set up the mock client class and expected return values
+    # this test does not pass in a known client, instead checking that a client is constructed as expected
+    mock_credential = mock_DefaultAzureCredential.return_value
+    client_cls = mocker.patch("benefits.secrets.SecretClient")
+    client = client_cls.return_value
+    client.get_secret.return_value = mocker.Mock(value=secret_value)
+
+    actual_value = get_secret_by_name(secret_name)
+
+    client_cls.assert_called_once_with(vault_url=expected_keyvault_url, credential=mock_credential)
+    client.get_secret.assert_called_once_with(secret_name)
+    assert actual_value == secret_value

tests/pytest/test_settings.py

@@ -0,0 +1,46 @@
+def test_runtime_environment__default(settings):
+    assert settings.RUNTIME_ENVIRONMENT() == "dev"
+
+
+def test_runtime_environment__dev(settings):
+    settings.ALLOWED_HOSTS = ["dev-benefits.calitp.org"]
+    assert settings.RUNTIME_ENVIRONMENT() == "dev"
+
+
+def test_runtime_environment__local(settings):
+    settings.ALLOWED_HOSTS = ["localhost", "127.0.0.1"]
+    assert settings.RUNTIME_ENVIRONMENT() == "dev"
+
+
+def test_runtime_environment__nonmatching(settings):
+    # with only nonmatching hosts, return dev
+    settings.ALLOWED_HOSTS = ["example.com", "example2.org"]
+    assert settings.RUNTIME_ENVIRONMENT() == "dev"
+
+
+def test_runtime_environment__test(settings):
+    settings.ALLOWED_HOSTS = ["test-benefits.calitp.org"]
+    assert settings.RUNTIME_ENVIRONMENT() == "test"
+
+
+def test_runtime_environment__test_and_nonmatching(settings):
+    # when test is specified with other nonmatching hosts, assume test
+    settings.ALLOWED_HOSTS = ["test-benefits.calitp.org", "example.com"]
+    assert settings.RUNTIME_ENVIRONMENT() == "test"
+
+
+def test_runtime_environment__test_and_prod(settings):
+    # if both test and prod are specified (edge case/error in configuration), assume test
+    settings.ALLOWED_HOSTS = ["benefits.calitp.org", "test-benefits.calitp.org"]
+    assert settings.RUNTIME_ENVIRONMENT() == "test"
+
+
+def test_runtime_environment__prod(settings):
+    settings.ALLOWED_HOSTS = ["benefits.calitp.org"]
+    assert settings.RUNTIME_ENVIRONMENT() == "prod"
+
+
+def test_runtime_environment__prod_and_nonmatching(settings):
+    # when prod is specified with other nonmatching hosts, assume prod
+    settings.ALLOWED_HOSTS = ["benefits.calitp.org", "https://example.com"]
+    assert settings.RUNTIME_ENVIRONMENT() == "prod"