Merge pull request #24 from color/int-2190-clrenv-secrets-repr

Keep secrets out of ClrEnv repr

Author: rarredon

clrenv/evaluate.py

@@ -41,7 +41,7 @@
 
 from .path import environment_paths
 from .read import EnvReader
-from .types import LeafValue, NestedMapping, check_valid_leaf_value
+from .types import LeafValue, NestedMapping, Secret, check_valid_leaf_value
 
 logger = logging.getLogger(__name__)
 
@@ -185,6 +185,9 @@ def _evaluate_key(self, key: str) -> Union[LeafValue, Mapping, None]:
         if value is None and key in self._sub_keys:
             return {}
 
+        if isinstance(value, Secret):
+            return value.value
+
         return value
 
     def _sub_key_path(self, key: str) -> Tuple[str, ...]:

clrenv/read.py

@@ -15,14 +15,14 @@
 import os
 from collections import abc, deque
 from pathlib import Path
-from typing import Any, Deque, Iterable, Mapping, Optional, Tuple
+from typing import Any, Deque, Iterable, Mapping, Optional, Tuple, Union
 
 import boto3
 from botocore.exceptions import EndpointConnectionError  # type: ignore
 
 from .deepmerge import deepmerge
 from .path import environment_paths
-from .types import MutableNestedMapping, NestedMapping, check_valid_leaf_value
+from .types import MutableNestedMapping, NestedMapping, Secret, check_valid_leaf_value
 
 logger = logging.getLogger(__name__)
 
@@ -108,19 +108,19 @@ def read(self) -> NestedMapping:
 
         return result
 
-    def postprocess_str(self, value: str) -> str:
+    def postprocess_str(self, value: str) -> Union[str, Secret]:
         """Post process string values."""
         # Expand environmental variables in the form of $FOO or ${FOO}.
         value = os.path.expandvars(value)
         # If value is a path starting with ~, expand.
         if value.startswith("~"):
-            value = os.path.expanduser(value)
+            return os.path.expanduser(value)
         # Substitute from clrypt keyfile.
         elif value.startswith("^keyfile "):
-            value = self.evaluate_clrypt_key(value[9:])
+            return Secret(source=value, value=self.evaluate_clrypt_key(value[9:]))
         # Substitute from aws ssm parameter store.
         elif value.startswith("^parameter "):
-            value = self.evaluate_ssm_parameter(value[11:])
+            return Secret(source=value, value=self.evaluate_ssm_parameter(value[11:]))
 
         return value
 

clrenv/tests/test_evaluate.py

@@ -1,5 +1,7 @@
 import functools
+from types import SimpleNamespace
 
+import botocore.exceptions  # type: ignore
 import pytest
 import yaml
 
@@ -182,3 +184,67 @@ def test_underscored_keys(default_env):
         default_env["__env"]  # pylint: disable=pointless-statement
     with pytest.raises(AttributeError):
         getattr(default_env.__unknown)
+
+
+def test_clrypt_secret(tmp_path, monkeypatch):
+    """
+    A keyfile secret should be accessible via clrenv.
+    """
+    env_path = tmp_path / "env"
+    env_path.write_text(yaml.dump({"base": {"foo": "^keyfile aaa"}}))
+
+    import clrypt
+
+    def mock_read(group, name):
+        assert group == "keys"
+        assert name == "keys"
+        return {"aaa": "bbb"}
+
+    monkeypatch.setattr(clrypt, "read_file_as_dict", mock_read)
+
+    env = clrenv.evaluate.RootClrEnv([env_path])
+
+    # Check that the repr does not leak the secret, but the secret is still accessible
+    assert '^keyfile aaa' in repr(env)
+    assert 'bbb' not in repr(env)
+    assert env.foo == "bbb"    
+
+def test_ssm_secret(tmp_path, monkeypatch):
+    """
+    A SSM secret should be accessible via clrenv.
+    """
+    env_path = tmp_path / "env"
+    env_path.write_text(yaml.dump({"base": {"foo": "^parameter aaa"}}))
+
+    monkeypatch.setenv("CLRENV_OFFLINE_DEV", "")
+    import boto3
+
+    class MockClient:
+        exceptions = SimpleNamespace()
+        exceptions.ParameterNotFound = botocore.exceptions.ClientError
+
+        def get_parameter(self, Name=None, WithDecryption=None):
+            assert WithDecryption is True
+            assert Name in ("aaa", "endpoint_error", "param_error")
+            if Name == "aaa":
+                return {"Parameter": {"Value": "bbb"}}
+            elif Name == "endpoint_error":
+                raise botocore.exceptions.EndpointConnectionError(endpoint_url="url")
+            elif Name == "param_error":
+                raise MockClient.exceptions.ParameterNotFound(
+                    error_response={}, operation_name=""
+                )
+
+    def mock_client(name):
+        assert name == "ssm"
+        return MockClient()
+
+    monkeypatch.setattr(boto3, "client", mock_client)
+
+    env = clrenv.evaluate.RootClrEnv([env_path])
+
+    # Check that the repr does not leak the secret, but the secret is still accessible
+    assert '^parameter aaa' in repr(env)
+    assert 'bbb' not in repr(env)
+    assert env.foo == "bbb"    
+

clrenv/tests/test_read.py

@@ -6,6 +6,7 @@
 import yaml
 
 import clrenv
+from clrenv.types import Secret
 
 
 @pytest.fixture(autouse=True)
@@ -75,7 +76,8 @@ def mock_read(group, name):
     env_path = tmp_path / "env"
     env_path.write_text(yaml.dump({"base": {"foo": "^keyfile aaa"}}))
     env = clrenv.read.EnvReader([env_path]).read()
-    assert env["foo"] == "bbb"
+    assert isinstance(env["foo"], Secret)
+    assert env["foo"].value == "bbb"
 
 
 def test_ssm(tmp_path, monkeypatch):
@@ -107,7 +109,8 @@ def mock_client(name):
     env_path = tmp_path / "env"
     env_path.write_text(yaml.dump({"base": {"foo": "^parameter aaa"}}))
     env = clrenv.read.EnvReader([env_path]).read()
-    assert env["foo"] == "bbb"
+    assert isinstance(env["foo"], Secret)
+    assert env["foo"].value == "bbb"
 
     env_path.write_text(yaml.dump({"base": {"foo": "^parameter endpoint_error"}}))
     with pytest.raises(botocore.exceptions.EndpointConnectionError):
@@ -124,4 +127,5 @@ def test_offline_parameter_flag(tmp_path, monkeypatch):
     env_path = tmp_path / "env"
     env_path.write_text(yaml.dump({"base": {"foo": "^parameter aaa"}}))
     env = clrenv.read.EnvReader([env_path]).read()
-    assert env["foo"] == "CLRENV_OFFLINE_PLACEHOLDER"
+    assert isinstance(env["foo"], Secret)
+    assert env["foo"].value == "CLRENV_OFFLINE_PLACEHOLDER"

clrenv/types.py

@@ -1,13 +1,28 @@
-from typing import Any, List, Mapping, MutableMapping, Union
+from typing import Any, List, Mapping, MutableMapping, NamedTuple, Union
 
 """Defines type annotations for use in clrenv.
 
 Ideally we would only allow str leaf values so that they can be migrated to an env
 var based system. For now we must supports more complex values.
 """
 
+class Secret(NamedTuple):
+    source: str
+    value: str
+
+    def __repr__(self) -> str:
+        """
+        Return a Secret's representation without exposing the secret.
+
+        In the event that a Secret's (or, more generally, a ClrEnv object's)
+        representation is logged, this will prevent us from leaking secrets into
+        plaintext.
+        """
+        return f"Secret(source='{self.source}')"
+
+
 # Type that can be read or set as values of leaf nodes.
-LeafValue = Union[bool, int, float, str, List[Union[bool, int, float, str]]]
+LeafValue = Union[bool, int, float, str, List[Union[bool, int, float, str]], Secret]
 # Type of a non leaf node.
 NestedMapping = Mapping[str, Union[LeafValue, Mapping[str, Any]]]
 MutableNestedMapping = MutableMapping[str, Union[LeafValue, MutableMapping[str, Any]]]