Secrets support for configuration files

homeassistant/util/yaml.py

@@ -5,10 +5,15 @@
 
 import glob
 import yaml
+try:
+    import keyring
+except ImportError:
+    keyring = None
 
 from homeassistant.exceptions import HomeAssistantError
 
 _LOGGER = logging.getLogger(__name__)
+_SECRET_NAMESPACE = 'homeassistant'
 
 
 # pylint: disable=too-many-ancestors
@@ -119,10 +124,49 @@ def _env_var_yaml(loader, node):
         raise HomeAssistantError(node.value)
 
 
+# pylint: disable=protected-access
+def _secret_yaml(loader, node):
+    """Load secrets and embed it into the configuration YAML."""
+    # Create secret cache on loader and load secret.yaml
+    if not hasattr(loader, '_SECRET_CACHE'):
+        loader._SECRET_CACHE = {}
+
+    secret_path = os.path.join(os.path.dirname(loader.name), 'secrets.yaml')
+    if secret_path not in loader._SECRET_CACHE:
+        if os.path.isfile(secret_path):
+            loader._SECRET_CACHE[secret_path] = load_yaml(secret_path)
+            secrets = loader._SECRET_CACHE[secret_path]
+            if 'logger' in secrets:
+                logger = str(secrets['logger']).lower()
+                if logger == 'debug':
+                    _LOGGER.setLevel(logging.DEBUG)
+                else:
+                    _LOGGER.error("secrets.yaml: 'logger: debug' expected,"
+                                  " but 'logger: %s' found", logger)
+                del secrets['logger']
+        else:
+            loader._SECRET_CACHE[secret_path] = None
+    secrets = loader._SECRET_CACHE[secret_path]
+
+    # Retrieve secret, first from secrets.yaml, then from keyring
+    if secrets is not None and node.value in secrets:
+        _LOGGER.debug('Secret %s retrieved from secrets.yaml.', node.value)
+        return secrets[node.value]
+    elif keyring:
+        # do ome keyring stuff
+        pwd = keyring.get_password(_SECRET_NAMESPACE, node.value)
+        if pwd:
+            _LOGGER.debug('Secret %s retrieved from keyring.', node.value)
+            return pwd
+
+    _LOGGER.error('Secret %s not defined.', node.value)
+    raise HomeAssistantError(node.value)
+
 yaml.SafeLoader.add_constructor('!include', _include_yaml)
 yaml.SafeLoader.add_constructor(yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG,
                                 _ordered_dict)
 yaml.SafeLoader.add_constructor('!env_var', _env_var_yaml)
+yaml.SafeLoader.add_constructor('!secret', _secret_yaml)
 yaml.SafeLoader.add_constructor('!include_dir_list', _include_dir_list_yaml)
 yaml.SafeLoader.add_constructor('!include_dir_merge_list',
                                 _include_dir_merge_list_yaml)

tests/util/test_yaml.py

@@ -3,8 +3,9 @@
 import unittest
 import os
 import tempfile
-
 from homeassistant.util import yaml
+import homeassistant.config as config_util
+from tests.common import get_test_config_dir
 
 
 class TestYaml(unittest.TestCase):
@@ -135,3 +136,81 @@ def test_include_dir_merge_named(self):
                     "key2": "two",
                     "key3": "three"
                 }
+
+
+def load_yaml(fname, string):
+    """Write a string to file and return the parsed yaml."""
+    with open(fname, 'w') as file:
+        file.write(string)
+    return config_util.load_yaml_config_file(fname)
+
+
+class FakeKeyring():
+    """Fake a keyring class."""
+
+    def __init__(self, secrets_dict):
+        """Store keyring dictionary."""
+        self._secrets = secrets_dict
+
+    # pylint: disable=protected-access
+    def get_password(self, domain, name):
+        """Retrieve password."""
+        assert domain == yaml._SECRET_NAMESPACE
+        return self._secrets.get(name)
+
+
+class TestSecrets(unittest.TestCase):
+    """Test the secrets parameter in the yaml utility."""
+
+    def setUp(self):  # pylint: disable=invalid-name
+        """Create & load secrets file."""
+        config_dir = get_test_config_dir()
+        self._yaml_path = os.path.join(config_dir,
+                                       config_util.YAML_CONFIG_FILE)
+        self._secret_path = os.path.join(config_dir, 'secrets.yaml')
+
+        load_yaml(self._secret_path,
+                  'http_pw: pwhttp\n'
+                  'comp1_un: un1\n'
+                  'comp1_pw: pw1\n'
+                  'stale_pw: not_used\n'
+                  'logger: debug\n')
+        self._yaml = load_yaml(self._yaml_path,
+                               'http:\n'
+                               '  api_password: !secret http_pw\n'
+                               'component:\n'
+                               '  username: !secret comp1_un\n'
+                               '  password: !secret comp1_pw\n'
+                               '')
+
+    def tearDown(self):  # pylint: disable=invalid-name
+        """Clean up secrets."""
+        for path in [self._yaml_path, self._secret_path]:
+            if os.path.isfile(path):
+                os.remove(path)
+
+    def test_secrets_from_yaml(self):
+        """Did secrets load ok."""
+        expected = {'api_password': 'pwhttp'}
+        self.assertEqual(expected, self._yaml['http'])
+
+        expected = {
+            'username': 'un1',
+            'password': 'pw1'}
+        self.assertEqual(expected, self._yaml['component'])
+
+    def test_secrets_keyring(self):
+        """Test keyring fallback & get_password."""
+        yaml.keyring = None  # Ensure its not there
+        yaml_str = 'http:\n  api_password: !secret http_pw_keyring'
+        with self.assertRaises(yaml.HomeAssistantError):
+            load_yaml(self._yaml_path, yaml_str)
+
+        yaml.keyring = FakeKeyring({'http_pw_keyring': 'yeah'})
+        _yaml = load_yaml(self._yaml_path, yaml_str)
+        self.assertEqual({'http': {'api_password': 'yeah'}}, _yaml)
+
+    def test_secrets_logger_removed(self):
+        """Ensure logger: debug was removed."""
+        with self.assertRaises(yaml.HomeAssistantError):
+            load_yaml(self._yaml_path, 'api_password: !secret logger')