This repository has been archived on 2024-05-31. You can view files and clone it, but cannot push or open issues or pull requests.
authentik/passbook/lib/config.py

151 lines
5.3 KiB
Python
Raw Normal View History

2019-09-30 16:04:04 +00:00
"""passbook core config loader"""
2018-11-14 18:14:14 +00:00
import os
2019-10-01 08:48:55 +00:00
from collections.abc import Mapping
2018-11-14 18:14:14 +00:00
from contextlib import contextmanager
from glob import glob
from typing import Any, Dict
2019-09-30 16:04:04 +00:00
from urllib.parse import urlparse
2018-11-14 18:14:14 +00:00
import yaml
from django.conf import ImproperlyConfigured
from django.http import HttpRequest
2019-09-30 16:04:04 +00:00
from structlog import get_logger
2018-11-14 18:14:14 +00:00
2020-05-27 09:26:48 +00:00
SEARCH_PATHS = ["passbook/lib/default.yml", "/etc/passbook/config.yml", ""] + glob(
2019-12-31 11:51:16 +00:00
"/etc/passbook/config.d/*.yml", recursive=True
)
2019-09-30 16:04:04 +00:00
LOGGER = get_logger()
2019-12-31 11:51:16 +00:00
ENV_PREFIX = "PASSBOOK"
ENVIRONMENT = os.getenv(f"{ENV_PREFIX}_ENV", "local")
2018-11-14 18:14:14 +00:00
def context_processor(request: HttpRequest) -> Dict[str, Any]:
"""Context Processor that injects config object into every template"""
kwargs = {"config": CONFIG.raw}
return kwargs
2018-11-14 18:14:14 +00:00
class ConfigLoader:
2019-09-30 16:04:04 +00:00
"""Search through SEARCH_PATHS and load configuration. Environment variables starting with
`ENV_PREFIX` are also applied.
A variable like PASSBOOK_POSTGRESQL__HOST would translate to postgresql.host"""
2018-11-14 18:14:14 +00:00
2019-04-10 16:48:55 +00:00
loaded_file = []
2018-11-14 18:14:14 +00:00
__config = {}
__sub_dicts = []
def __init__(self):
super().__init__()
2019-12-31 11:51:16 +00:00
base_dir = os.path.realpath(os.path.join(os.path.dirname(__file__), "../.."))
2018-11-14 18:14:14 +00:00
for path in SEARCH_PATHS:
# Check if path is relative, and if so join with base_dir
if not os.path.isabs(path):
path = os.path.join(base_dir, path)
if os.path.isfile(path) and os.path.exists(path):
# Path is an existing file, so we just read it and update our config with it
self.update_from_file(path)
elif os.path.isdir(path) and os.path.exists(path):
# Path is an existing dir, so we try to read the env config from it
2019-12-31 11:51:16 +00:00
env_paths = [
os.path.join(path, ENVIRONMENT + ".yml"),
os.path.join(path, ENVIRONMENT + ".env.yml"),
]
2018-11-14 18:14:14 +00:00
for env_file in env_paths:
if os.path.isfile(env_file) and os.path.exists(env_file):
# Update config with env file
self.update_from_file(env_file)
2019-09-30 16:04:04 +00:00
self.update_from_env()
2018-11-14 18:14:14 +00:00
def update(self, root, updatee):
"""Recursively update dictionary"""
for key, value in updatee.items():
if isinstance(value, Mapping):
root[key] = self.update(root.get(key, {}), value)
else:
2019-09-30 16:04:04 +00:00
if isinstance(value, str):
value = self.parse_uri(value)
2018-11-14 18:14:14 +00:00
root[key] = value
return root
2019-09-30 16:04:04 +00:00
def parse_uri(self, value):
"""Parse string values which start with a URI"""
url = urlparse(value)
2019-12-31 11:51:16 +00:00
if url.scheme == "env":
2019-09-30 16:04:04 +00:00
value = os.getenv(url.netloc, url.query)
return value
2018-11-14 18:14:14 +00:00
def update_from_file(self, path: str):
"""Update config from file contents"""
try:
with open(path) as file:
try:
self.update(self.__config, yaml.safe_load(file))
2019-09-30 16:04:04 +00:00
LOGGER.debug("Loaded config", file=path)
2019-04-10 16:48:55 +00:00
self.loaded_file.append(path)
2018-11-14 18:14:14 +00:00
except yaml.YAMLError as exc:
raise ImproperlyConfigured from exc
except PermissionError as exc:
2019-12-31 11:51:16 +00:00
LOGGER.warning("Permission denied while reading file", path=path, error=exc)
2018-11-14 18:14:14 +00:00
def update_from_dict(self, update: dict):
"""Update config from dict"""
self.__config.update(update)
2019-09-30 16:04:04 +00:00
def update_from_env(self):
"""Check environment variables"""
outer = {}
idx = 0
for key, value in os.environ.items():
if not key.startswith(ENV_PREFIX):
continue
2019-12-31 11:51:16 +00:00
relative_key = key.replace(f"{ENV_PREFIX}_", "").replace("__", ".").lower()
2019-09-30 16:04:04 +00:00
# Recursively convert path from a.b.c into outer[a][b][c]
current_obj = outer
2019-12-31 11:51:16 +00:00
dot_parts = relative_key.split(".")
2019-09-30 16:04:04 +00:00
for dot_part in dot_parts[:-1]:
if dot_part not in current_obj:
current_obj[dot_part] = {}
current_obj = current_obj[dot_part]
current_obj[dot_parts[-1]] = value
idx += 1
if idx > 0:
LOGGER.debug("Loaded environment variables", count=idx)
self.update(self.__config, outer)
2018-11-14 18:14:14 +00:00
@contextmanager
# pylint: disable=invalid-name
def cd(self, sub: str):
"""Contextmanager that descends into sub-dict. Can be chained."""
self.__sub_dicts.append(sub)
yield
self.__sub_dicts.pop()
@property
def raw(self) -> dict:
"""Get raw config dictionary"""
return self.__config
# pylint: disable=invalid-name
2019-12-31 11:51:16 +00:00
def y(self, path: str, default=None, sep=".") -> Any:
2018-11-14 18:14:14 +00:00
"""Access attribute by using yaml path"""
# Walk sub_dicts before parsing path
root = self.raw
for sub in self.__sub_dicts:
root = root.get(sub, None)
# Walk each component of the path
for comp in path.split(sep):
if comp in root:
root = root.get(comp)
else:
return default
return root
2019-09-30 16:04:04 +00:00
def y_bool(self, path: str, default=False) -> bool:
"""Wrapper for y that converts value into boolean"""
2019-12-31 11:51:16 +00:00
return str(self.y(path, default)).lower() == "true"
2019-09-30 16:04:04 +00:00
2018-11-14 18:14:14 +00:00
CONFIG = ConfigLoader()