mirror of
https://github.com/elisspace/core.git
synced 2026-10-08 01:58:41 +00:00
Ville Skyttä
b4bac0f7a0
* Remove unnecessary exception re-wraps * Preserve exception chains on re-raise We slap "from cause" to almost all possible cases here. In some cases it could conceivably be better to do "from None" if we really want to hide the cause. However those should be in the minority, and "from cause" should be an improvement over the corresponding raise without a "from" in all cases anyway. The only case where we raise from None here is in plex, where the exception for an original invalid SSL cert is not the root cause for failure to validate a newly fetched one. Follow local convention on exception variable names if there is a consistent one, otherwise `err` to match with majority of codebase. * Fix mistaken re-wrap in homematicip_cloud/hap.py Missed the difference between HmipConnectionError and HmipcConnectionError. * Do not hide original error on plex new cert validation error Original is not the cause for the new one, but showing old in the traceback is useful nevertheless.
328 lines
11 KiB
Python
328 lines
11 KiB
Python
"""Custom loader."""
|
|
from collections import OrderedDict
|
|
import fnmatch
|
|
import logging
|
|
import os
|
|
import sys
|
|
from typing import Dict, Iterator, List, TypeVar, Union, overload
|
|
|
|
import yaml
|
|
|
|
from homeassistant.exceptions import HomeAssistantError
|
|
|
|
from .const import _SECRET_NAMESPACE, SECRET_YAML
|
|
from .objects import NodeListClass, NodeStrClass
|
|
|
|
try:
|
|
import keyring
|
|
except ImportError:
|
|
keyring = None
|
|
|
|
try:
|
|
import credstash
|
|
except ImportError:
|
|
credstash = None
|
|
|
|
|
|
# mypy: allow-untyped-calls, no-warn-return-any
|
|
|
|
JSON_TYPE = Union[List, Dict, str] # pylint: disable=invalid-name
|
|
DICT_T = TypeVar("DICT_T", bound=Dict) # pylint: disable=invalid-name
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
__SECRET_CACHE: Dict[str, JSON_TYPE] = {}
|
|
|
|
|
|
def clear_secret_cache() -> None:
|
|
"""Clear the secret cache.
|
|
|
|
Async friendly.
|
|
"""
|
|
__SECRET_CACHE.clear()
|
|
|
|
|
|
class SafeLineLoader(yaml.SafeLoader):
|
|
"""Loader class that keeps track of line numbers."""
|
|
|
|
def compose_node(self, parent: yaml.nodes.Node, index: int) -> yaml.nodes.Node:
|
|
"""Annotate a node with the first line it was seen."""
|
|
last_line: int = self.line
|
|
node: yaml.nodes.Node = super().compose_node(parent, index)
|
|
node.__line__ = last_line + 1 # type: ignore
|
|
return node
|
|
|
|
|
|
def load_yaml(fname: str) -> JSON_TYPE:
|
|
"""Load a YAML file."""
|
|
try:
|
|
with open(fname, encoding="utf-8") as conf_file:
|
|
# If configuration file is empty YAML returns None
|
|
# We convert that to an empty dict
|
|
return yaml.load(conf_file, Loader=SafeLineLoader) or OrderedDict()
|
|
except yaml.YAMLError as exc:
|
|
_LOGGER.error(str(exc))
|
|
raise HomeAssistantError(exc) from exc
|
|
except UnicodeDecodeError as exc:
|
|
_LOGGER.error("Unable to read file %s: %s", fname, exc)
|
|
raise HomeAssistantError(exc) from exc
|
|
|
|
|
|
@overload
|
|
def _add_reference(
|
|
obj: Union[list, NodeListClass], loader: yaml.SafeLoader, node: yaml.nodes.Node
|
|
) -> NodeListClass:
|
|
...
|
|
|
|
|
|
@overload
|
|
def _add_reference(
|
|
obj: Union[str, NodeStrClass], loader: yaml.SafeLoader, node: yaml.nodes.Node
|
|
) -> NodeStrClass:
|
|
...
|
|
|
|
|
|
@overload
|
|
def _add_reference(
|
|
obj: DICT_T, loader: yaml.SafeLoader, node: yaml.nodes.Node
|
|
) -> DICT_T:
|
|
...
|
|
|
|
|
|
def _add_reference(obj, loader: SafeLineLoader, node: yaml.nodes.Node): # type: ignore
|
|
"""Add file reference information to an object."""
|
|
if isinstance(obj, list):
|
|
obj = NodeListClass(obj)
|
|
if isinstance(obj, str):
|
|
obj = NodeStrClass(obj)
|
|
setattr(obj, "__config_file__", loader.name)
|
|
setattr(obj, "__line__", node.start_mark.line)
|
|
return obj
|
|
|
|
|
|
def _include_yaml(loader: SafeLineLoader, node: yaml.nodes.Node) -> JSON_TYPE:
|
|
"""Load another YAML file and embeds it using the !include tag.
|
|
|
|
Example:
|
|
device_tracker: !include device_tracker.yaml
|
|
|
|
"""
|
|
fname = os.path.join(os.path.dirname(loader.name), node.value)
|
|
try:
|
|
return _add_reference(load_yaml(fname), loader, node)
|
|
except FileNotFoundError as exc:
|
|
raise HomeAssistantError(
|
|
f"{node.start_mark}: Unable to read file {fname}."
|
|
) from exc
|
|
|
|
|
|
def _is_file_valid(name: str) -> bool:
|
|
"""Decide if a file is valid."""
|
|
return not name.startswith(".")
|
|
|
|
|
|
def _find_files(directory: str, pattern: str) -> Iterator[str]:
|
|
"""Recursively load files in a directory."""
|
|
for root, dirs, files in os.walk(directory, topdown=True):
|
|
dirs[:] = [d for d in dirs if _is_file_valid(d)]
|
|
for basename in sorted(files):
|
|
if _is_file_valid(basename) and fnmatch.fnmatch(basename, pattern):
|
|
filename = os.path.join(root, basename)
|
|
yield filename
|
|
|
|
|
|
def _include_dir_named_yaml(
|
|
loader: SafeLineLoader, node: yaml.nodes.Node
|
|
) -> OrderedDict:
|
|
"""Load multiple files from directory as a dictionary."""
|
|
mapping: OrderedDict = OrderedDict()
|
|
loc = os.path.join(os.path.dirname(loader.name), node.value)
|
|
for fname in _find_files(loc, "*.yaml"):
|
|
filename = os.path.splitext(os.path.basename(fname))[0]
|
|
if os.path.basename(fname) == SECRET_YAML:
|
|
continue
|
|
mapping[filename] = load_yaml(fname)
|
|
return _add_reference(mapping, loader, node)
|
|
|
|
|
|
def _include_dir_merge_named_yaml(
|
|
loader: SafeLineLoader, node: yaml.nodes.Node
|
|
) -> OrderedDict:
|
|
"""Load multiple files from directory as a merged dictionary."""
|
|
mapping: OrderedDict = OrderedDict()
|
|
loc = os.path.join(os.path.dirname(loader.name), node.value)
|
|
for fname in _find_files(loc, "*.yaml"):
|
|
if os.path.basename(fname) == SECRET_YAML:
|
|
continue
|
|
loaded_yaml = load_yaml(fname)
|
|
if isinstance(loaded_yaml, dict):
|
|
mapping.update(loaded_yaml)
|
|
return _add_reference(mapping, loader, node)
|
|
|
|
|
|
def _include_dir_list_yaml(
|
|
loader: SafeLineLoader, node: yaml.nodes.Node
|
|
) -> List[JSON_TYPE]:
|
|
"""Load multiple files from directory as a list."""
|
|
loc = os.path.join(os.path.dirname(loader.name), node.value)
|
|
return [
|
|
load_yaml(f)
|
|
for f in _find_files(loc, "*.yaml")
|
|
if os.path.basename(f) != SECRET_YAML
|
|
]
|
|
|
|
|
|
def _include_dir_merge_list_yaml(
|
|
loader: SafeLineLoader, node: yaml.nodes.Node
|
|
) -> JSON_TYPE:
|
|
"""Load multiple files from directory as a merged list."""
|
|
loc: str = os.path.join(os.path.dirname(loader.name), node.value)
|
|
merged_list: List[JSON_TYPE] = []
|
|
for fname in _find_files(loc, "*.yaml"):
|
|
if os.path.basename(fname) == SECRET_YAML:
|
|
continue
|
|
loaded_yaml = load_yaml(fname)
|
|
if isinstance(loaded_yaml, list):
|
|
merged_list.extend(loaded_yaml)
|
|
return _add_reference(merged_list, loader, node)
|
|
|
|
|
|
def _ordered_dict(loader: SafeLineLoader, node: yaml.nodes.MappingNode) -> OrderedDict:
|
|
"""Load YAML mappings into an ordered dictionary to preserve key order."""
|
|
loader.flatten_mapping(node)
|
|
nodes = loader.construct_pairs(node)
|
|
|
|
seen: Dict = {}
|
|
for (key, _), (child_node, _) in zip(nodes, node.value):
|
|
line = child_node.start_mark.line
|
|
|
|
try:
|
|
hash(key)
|
|
except TypeError as exc:
|
|
fname = getattr(loader.stream, "name", "")
|
|
raise yaml.MarkedYAMLError(
|
|
context=f'invalid key: "{key}"',
|
|
context_mark=yaml.Mark(fname, 0, line, -1, None, None),
|
|
) from exc
|
|
|
|
if key in seen:
|
|
fname = getattr(loader.stream, "name", "")
|
|
_LOGGER.warning(
|
|
'YAML file %s contains duplicate key "%s". Check lines %d and %d',
|
|
fname,
|
|
key,
|
|
seen[key],
|
|
line,
|
|
)
|
|
seen[key] = line
|
|
|
|
return _add_reference(OrderedDict(nodes), loader, node)
|
|
|
|
|
|
def _construct_seq(loader: SafeLineLoader, node: yaml.nodes.Node) -> JSON_TYPE:
|
|
"""Add line number and file name to Load YAML sequence."""
|
|
(obj,) = loader.construct_yaml_seq(node)
|
|
return _add_reference(obj, loader, node)
|
|
|
|
|
|
def _env_var_yaml(loader: SafeLineLoader, node: yaml.nodes.Node) -> str:
|
|
"""Load environment variables and embed it into the configuration YAML."""
|
|
args = node.value.split()
|
|
|
|
# Check for a default value
|
|
if len(args) > 1:
|
|
return os.getenv(args[0], " ".join(args[1:]))
|
|
if args[0] in os.environ:
|
|
return os.environ[args[0]]
|
|
_LOGGER.error("Environment variable %s not defined", node.value)
|
|
raise HomeAssistantError(node.value)
|
|
|
|
|
|
def _load_secret_yaml(secret_path: str) -> JSON_TYPE:
|
|
"""Load the secrets yaml from path."""
|
|
secret_path = os.path.join(secret_path, SECRET_YAML)
|
|
if secret_path in __SECRET_CACHE:
|
|
return __SECRET_CACHE[secret_path]
|
|
|
|
_LOGGER.debug("Loading %s", secret_path)
|
|
try:
|
|
secrets = load_yaml(secret_path)
|
|
if not isinstance(secrets, dict):
|
|
raise HomeAssistantError("Secrets is not a dictionary")
|
|
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"]
|
|
except FileNotFoundError:
|
|
secrets = {}
|
|
__SECRET_CACHE[secret_path] = secrets
|
|
return secrets
|
|
|
|
|
|
def secret_yaml(loader: SafeLineLoader, node: yaml.nodes.Node) -> JSON_TYPE:
|
|
"""Load secrets and embed it into the configuration YAML."""
|
|
secret_path = os.path.dirname(loader.name)
|
|
while True:
|
|
secrets = _load_secret_yaml(secret_path)
|
|
|
|
if node.value in secrets:
|
|
_LOGGER.debug(
|
|
"Secret %s retrieved from secrets.yaml in folder %s",
|
|
node.value,
|
|
secret_path,
|
|
)
|
|
return secrets[node.value]
|
|
|
|
if secret_path == os.path.dirname(sys.path[0]):
|
|
break # sys.path[0] set to config/deps folder by bootstrap
|
|
|
|
secret_path = os.path.dirname(secret_path)
|
|
if not os.path.exists(secret_path) or len(secret_path) < 5:
|
|
break # Somehow we got past the .homeassistant config folder
|
|
|
|
if keyring:
|
|
# do some keyring stuff
|
|
pwd = keyring.get_password(_SECRET_NAMESPACE, node.value)
|
|
if pwd:
|
|
_LOGGER.debug("Secret %s retrieved from keyring", node.value)
|
|
return pwd
|
|
|
|
global credstash # pylint: disable=invalid-name, global-statement
|
|
|
|
if credstash:
|
|
# pylint: disable=no-member
|
|
try:
|
|
pwd = credstash.getSecret(node.value, table=_SECRET_NAMESPACE)
|
|
if pwd:
|
|
_LOGGER.debug("Secret %s retrieved from credstash", node.value)
|
|
return pwd
|
|
except credstash.ItemNotFound:
|
|
pass
|
|
except Exception: # pylint: disable=broad-except
|
|
# Catch if package installed and no config
|
|
credstash = None
|
|
|
|
raise HomeAssistantError(f"Secret {node.value} not defined")
|
|
|
|
|
|
yaml.SafeLoader.add_constructor("!include", _include_yaml)
|
|
yaml.SafeLoader.add_constructor(
|
|
yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, _ordered_dict
|
|
)
|
|
yaml.SafeLoader.add_constructor(
|
|
yaml.resolver.BaseResolver.DEFAULT_SEQUENCE_TAG, _construct_seq
|
|
)
|
|
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)
|
|
yaml.SafeLoader.add_constructor("!include_dir_named", _include_dir_named_yaml)
|
|
yaml.SafeLoader.add_constructor(
|
|
"!include_dir_merge_named", _include_dir_merge_named_yaml
|
|
)
|