Merge pull request #123 from ClarkSource/secret-config-override

feat(secrets): allow overriding secret config in get_secret helper

README.md

@@ -296,6 +296,12 @@ secrets:
 >
 > Be careful to follow this format when setting up the provider `prefix` and `get_secret(key)`.
 
+Global secrets config can be overridden in `get_secret` helper function call by specifying `config_override` argument.
+
+```yaml
+foobar: "{{ get_secret('/my-key', config_override={'prefix': '/dev'}) }}"
+```
+
 ###### role assumption
 
 You can optionally assume an IAM role to retrieve secrets by specyfing `role_arn` in the config:

k8t/filters.py

@@ -80,7 +80,7 @@ def hashf(value, method="sha256"):
     return hash_method.hexdigest()
 
 
-def get_secret(key: str, length: Optional[int] = None) -> str:
+def get_secret(key: str, length: Optional[int] = None, config_override: Optional[dict] = {}) -> str:
     provider_name = config.CONFIG.get("secrets", {}).get("provider")
 
     if not provider_name:
@@ -92,7 +92,7 @@ def get_secret(key: str, length: Optional[int] = None) -> str:
     except AttributeError as no_secret_provider:
         raise NotImplementedError(f"secret provider {provider_name} does not exist.") from no_secret_provider
 
-    return provider(key, length)
+    return provider(key, length, config_override)
 
 
 def to_bool(value: Any) -> Optional[bool]:

k8t/secret_providers.py

@@ -28,8 +28,10 @@
 DEFAULT_SSM_REGION = "eu-central-1"
 
 
-def ssm(key: str, length: Optional[int] = None) -> str:
+def ssm(key: str, length: Optional[int] = None, config_override: Optional[dict] = {}) -> str:
+    # Merge the config given as an argument with default config.
     secrets_config = config.CONFIG.get("secrets", {})
+    secrets_config.update(config_override)
 
     prefix = str(secrets_config.get("prefix", DEFAULT_SSM_PREFIX))
     region = str(secrets_config.get("region", DEFAULT_SSM_REGION))
@@ -89,7 +91,7 @@ def _assume_role(role_arn: str, region: str) -> dict:
         raise RuntimeError(f"Failed to assume role {role_arn}: {exc}") from exc
 
 
-def random(key: str, length: Optional[int] = None) -> str:
+def random(key: str, length: Optional[int] = None, config_override: Optional[dict] = {}) -> str:
     LOGGER.debug("Requesting secret from %s", key)
 
     if key not in RANDOM_STORE:
@@ -106,7 +108,7 @@ def random(key: str, length: Optional[int] = None) -> str:
     return RANDOM_STORE[key]
 
 
-def hash(key: str, length: Optional[int] = None) -> str:
+def hash(key: str, length: Optional[int] = None, config_override: Optional[dict] = {}) -> str:
     LOGGER.debug("Requesting secret from %s", key)
 
     if key not in RANDOM_STORE:

tests/filters.py

@@ -59,16 +59,20 @@ def test_get_secret():
     config.CONFIG = {"secrets": {"provider": "random"}}
     with patch.object(secret_providers, "random") as mock:
         get_secret("any")
-        mock.assert_called_with("any", None)
+        mock.assert_called_with("any", None, {})
         get_secret("any", 99)
-        mock.assert_called_with("any", 99)
+        mock.assert_called_with("any", 99, {})
+        get_secret("any", 99, {"foo": "bar"})
+        mock.assert_called_with("any", 99, {"foo": "bar"})
 
     config.CONFIG = {"secrets": {"provider": "ssm"}}
     with patch.object(secret_providers, "ssm") as mock:
         get_secret("any")
-        mock.assert_called_with("any", None)
+        mock.assert_called_with("any", None, {})
         get_secret("any", 99)
-        mock.assert_called_with("any", 99)
+        mock.assert_called_with("any", 99, {})
+        get_secret("any", 99, {"foo": "bar"})
+        mock.assert_called_with("any", 99, {"foo": "bar"})
 
     config.CONFIG = {"secrets": {"provider": "nothing"}}
     with pytest.raises(NotImplementedError, match=r"secret provider nothing does not exist."):

tests/secret_providers.py

@@ -78,6 +78,11 @@ def test_ssm():
     }
     assert ssm("/password"), "my_secret_value"
 
+    config.CONFIG = {
+        "secrets": {"provider": "ssm", "region": region, "prefix": "/app/dev"}
+    }
+    assert ssm("/test1", config_override={"prefix": "/dev"}), "string_value"
+
     config.CONFIG = {"secrets": {"provider": "ssm", "region": "eu-central-1"}}
     with pytest.raises(RuntimeError, match=r"Failed to retrieve secret foo: ..."):
         ssm("foo")