mirror of
https://github.com/home-assistant/core.git
synced 2025-07-19 19:27:45 +00:00
Secrets support for configuration files (#2312)
* ! secret based on yaml.py * Private Secrets Dict, removed cmdline, fixed log level * Secrets limited to yaml only * Add keyring & debug tests
This commit is contained in:
parent
1c1d18053b
commit
7b02dc434a
@ -5,10 +5,15 @@ from collections import OrderedDict
|
|||||||
|
|
||||||
import glob
|
import glob
|
||||||
import yaml
|
import yaml
|
||||||
|
try:
|
||||||
|
import keyring
|
||||||
|
except ImportError:
|
||||||
|
keyring = None
|
||||||
|
|
||||||
from homeassistant.exceptions import HomeAssistantError
|
from homeassistant.exceptions import HomeAssistantError
|
||||||
|
|
||||||
_LOGGER = logging.getLogger(__name__)
|
_LOGGER = logging.getLogger(__name__)
|
||||||
|
_SECRET_NAMESPACE = 'homeassistant'
|
||||||
|
|
||||||
|
|
||||||
# pylint: disable=too-many-ancestors
|
# pylint: disable=too-many-ancestors
|
||||||
@ -119,10 +124,49 @@ def _env_var_yaml(loader, node):
|
|||||||
raise HomeAssistantError(node.value)
|
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('!include', _include_yaml)
|
||||||
yaml.SafeLoader.add_constructor(yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG,
|
yaml.SafeLoader.add_constructor(yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG,
|
||||||
_ordered_dict)
|
_ordered_dict)
|
||||||
yaml.SafeLoader.add_constructor('!env_var', _env_var_yaml)
|
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_list', _include_dir_list_yaml)
|
||||||
yaml.SafeLoader.add_constructor('!include_dir_merge_list',
|
yaml.SafeLoader.add_constructor('!include_dir_merge_list',
|
||||||
_include_dir_merge_list_yaml)
|
_include_dir_merge_list_yaml)
|
||||||
|
@ -3,8 +3,9 @@ import io
|
|||||||
import unittest
|
import unittest
|
||||||
import os
|
import os
|
||||||
import tempfile
|
import tempfile
|
||||||
|
|
||||||
from homeassistant.util import yaml
|
from homeassistant.util import yaml
|
||||||
|
import homeassistant.config as config_util
|
||||||
|
from tests.common import get_test_config_dir
|
||||||
|
|
||||||
|
|
||||||
class TestYaml(unittest.TestCase):
|
class TestYaml(unittest.TestCase):
|
||||||
@ -135,3 +136,81 @@ class TestYaml(unittest.TestCase):
|
|||||||
"key2": "two",
|
"key2": "two",
|
||||||
"key3": "three"
|
"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')
|
||||||
|
Loading…
x
Reference in New Issue
Block a user