icalendar-timeseries-server/icalendar_timeseries_server/config.py

from typing import Any, Dict, List, Optional, Type, Union

import base64
import json
from datetime import timedelta
import ssl
import urllib.request
import sys
import logging

import pytz
import jinja2
from isodate import Duration, parse_duration

from icalendar_timeseries_server import __version__


CONFIG: Optional['Config'] = None

JENV: Optional[jinja2.Environment] = None


__py_version: str = f'{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}'
USER_AGENT: str = f'icalendar-timeseries-server/{__version__} (Python/{__py_version})'


class CalendarConfig:

    def __init__(self, config: Dict[str, Any], config_path: str) -> None:
        self._url: str = _keycheck('url', config, str, config_path)
        self._scrape_interval: Duration = _parse_timedelta('interval', config, config_path, default_value='PT15M')
        self._ca: Optional[str] = _keycheck('ca', config, str, config_path, optional=True)
        auth: Dict[str, Any] = _keycheck('auth', config, dict, config_path, default_value={'type': 'none'})
        self._authtype: str = _keycheck('type', auth, str, f'{config_path}.auth',
                                        default_value='none',
                                        valid_values=['none', 'basic', 'tls'])
        self._request_headers: Dict[str, str] = {
            'User-Agent': USER_AGENT
        }

        if self._authtype == 'none':
            # No auth, nothing to do
            pass

        elif self._authtype == 'basic':
            # Build the Authorization header required for HTTP Basic Auth
            username: str = _keycheck('username', auth, str, f'{config_path}.auth')
            password: str = _keycheck('password', auth, str, f'{config_path}.auth')
            token: str = base64.b64encode(f'{username}:{password}'.encode()).decode()
            self._request_headers['Authorization'] = f'Basic {token}'

        elif self._authtype == 'tls':
            # Store TLS config for later use
            self._tls_keyfile: str = _keycheck('keyfile', auth, str, f'{config_path}.auth')
            self._tls_passphrase: str = _keycheck('passphrase', auth, str, f'{config_path}.auth', optional=True)

    @property
    def url(self) -> str:
        return self._url

    @property
    def interval(self) -> Duration:
        return self._scrape_interval

    def get_url_opener(self) -> urllib.request.OpenerDirector:

        if self._authtype == 'tls':
            # Create an OpenSSL context and load the CA and client certificates
            context: ssl.SSLContext = ssl.create_default_context(cafile=self._ca)
            context.load_cert_chain(certfile=self._tls_keyfile, password=self._tls_passphrase)
            opener: urllib.request.OpenerDirector = \
                urllib.request.build_opener(urllib.request.HTTPSHandler(context=context))

        elif self._ca is not None:
            # No TLS Client Certificate, but load the CA certificate
            context: ssl.SSLContext = ssl.create_default_context(cafile=self._ca)
            opener = urllib.request.build_opener(urllib.request.HTTPSHandler(context=context))

        else:
            # No TLS config, return default opener
            opener = urllib.request.build_opener()

        # Set request headers, at least User-Agent, optionally Authorization
        opener.addheaders = [(k, v) for k, v in self._request_headers.items()]
        return opener


class Config:

    def __init__(self, config: Dict[str, Any] = None) -> None:
        if config is None:
            config = dict()
        self._addr: str = _keycheck('addr', config, str, '', default_value='127.0.0.1')
        self._port: int = _keycheck('port', config, int, '', default_value=8090)
        self._tz: pytz.tzinfo = _parse_timezone('tz', config, '', default_value='UTC')
        self._start_delta: Duration = _parse_timedelta('start_delta', config, '', default_value='PT')
        self._end_delta: Duration = _parse_timedelta('end_delta', config, '', default_value='P30D')
        self._calendars: Dict[str, CalendarConfig] = self._parse_calendars_config('calendars', config, '')
        self._key_replace = _parse_key_replace('key_replace', config, '')
        self._value_replace = _parse_value_replace('value_replace', config, '')

    @staticmethod
    def _parse_calendars_config(key: str,
                                config: Dict[str, Any],
                                path: str) -> Dict[str, CalendarConfig]:
        cdef: Dict[str, Any] = _keycheck(key, config, dict, path, default_value={})
        calendars: Dict[str, CalendarConfig] = dict()
        for name, c in cdef.items():
            calendar: CalendarConfig = CalendarConfig(c, f'{path}.{name}')
            calendars[name] = calendar
        return calendars

    @property
    def addr(self) -> str:
        return self._addr

    @property
    def port(self) -> int:
        return self._port

    @property
    def tz(self) -> pytz.tzinfo:
        return self._tz

    @property
    def start_delta(self) -> Duration:
        return self._start_delta

    @property
    def end_delta(self) -> Duration:
        return self._end_delta

    @property
    def calendars(self) -> Dict[str, CalendarConfig]:
        return self._calendars

    @property
    def key_replace(self) -> Dict[str, str]:
        return self._key_replace

    @property
    def value_replace(self) -> Dict[str, str]:
        return self._value_replace


def _keycheck(key: str,
              config: Dict[str, Any],
              typ: Type,
              path: str,
              default_value: Any = None,
              optional: bool = False,
              valid_values: Optional[List[Any]] = None) -> Any:
    if key not in config:
        if default_value is not None or optional:
            return default_value
        raise KeyError(f'Expected member "{key}" not found at path {path}')
    value: Any = config[key]
    if not isinstance(value, typ):
        raise TypeError(f'Expected {typ.__name__}, not {type(value).__name__} for path {path}.{key}')
    if valid_values is not None:
        if value not in valid_values:
            raise ValueError(f'Expected one of {", ".join(valid_values)} ({typ}), not {value} for path {path}.{key}')
    return value


def _parse_timedelta(key: str,
                     config: Dict[str, Any],
                     path: str,
                     default_value: Any = None,
                     force_positive: bool = False) -> Duration:
    tdstring: str = _keycheck(key, config, str, path, default_value=default_value)
    duration: Union[Duration, timedelta] = parse_duration(tdstring)
    if isinstance(duration, timedelta):
        duration = Duration(days=duration.days,
                            seconds=duration.seconds,
                            microseconds=duration.microseconds)
    if force_positive and duration.total_seconds() < 0:
        raise ValueError(f'Duration must be positive for path {path}.{key}')
    return duration


def _parse_timezone(key: str,
                    config: Dict[str, Any],
                    path: str,
                    default_value: Any = None) -> Any:
    zonename: str = _keycheck(key, config, str, path, default_value=default_value)
    return pytz.timezone(zonename)


def _parse_key_replace(key: str,
                       config: Dict[str, Any],
                       path: str) -> Dict[str, str]:
    key_replace: Dict[str, Any] = _keycheck(key, config, dict, path, default_value={})
    for label in key_replace.keys():
        _keycheck(label, key_replace, str, f'{path}.{key}')
    return key_replace


def _parse_value_replace(key: str,
                         config: Dict[str, Any],
                         path: str) -> Dict[str, str]:
    value_replace: Dict[str, Any] = _keycheck(key, config, dict, path, default_value={})
    for label in value_replace.keys():
        _keycheck(label, value_replace, str, f'{path}.{key}')
    return value_replace


def get_config() -> Config:
    global CONFIG
    return CONFIG


def get_jenv() -> jinja2.Environment:
    global JENV
    return JENV


def load_config(filename: str):
    global CONFIG, JENV
    try:
        with open(filename, 'r') as f:
            json_config = json.loads(f.read())
        CONFIG = Config(json_config)
        JENV = jinja2.Environment()
    except json.JSONDecodeError as e:
        logging.exception('Cannot parse config JSON')
        raise e
    except Exception as e:
        logging.error(e)
        raise e


def load_default_config():
    global CONFIG, JENV
    CONFIG = Config({})
    JENV = jinja2.Environment()