Move the common configuration functionality from oidcrp/oidcop to here.

Author: rohe

src/oidcmsg/configure.py

@@ -0,0 +1,161 @@
+import importlib
+import json
+import logging
+import os
+from typing import Dict
+from typing import List
+from typing import Optional
+
+from oidcmsg.logging import configure_logging
+from oidcmsg.util import load_yaml_config
+
+DEFAULT_FILE_ATTRIBUTE_NAMES = ['server_key', 'server_cert', 'filename', 'template_dir',
+                                'private_path', 'public_path', 'db_file']
+
+URIS = ["redirect_uris", 'issuer', 'base_url']
+
+
+def lower_or_upper(config, param, default=None):
+    res = config.get(param.lower(), default)
+    if not res:
+        res = config.get(param.upper(), default)
+    return res
+
+
+def add_base_path(conf: dict, base_path: str, file_attributes: List[str]):
+    for key, val in conf.items():
+        if key in file_attributes:
+            if val.startswith("/"):
+                continue
+            elif val == "":
+                conf[key] = "./" + val
+            else:
+                conf[key] = os.path.join(base_path, val)
+        if isinstance(val, dict):
+            conf[key] = add_base_path(val, base_path, file_attributes)
+
+    return conf
+
+
+def set_domain_and_port(conf: dict, uris: List[str], domain: str, port: int):
+    for key, val in conf.items():
+        if key in uris:
+            if not val:
+                continue
+
+            if isinstance(val, list):
+                _new = [v.format(domain=domain, port=port) for v in val]
+            else:
+                _new = val.format(domain=domain, port=port)
+            conf[key] = _new
+        elif isinstance(val, dict):
+            conf[key] = set_domain_and_port(val, uris, domain, port)
+    return conf
+
+
+class Base:
+    """ Configuration base class """
+
+    def __init__(self,
+                 conf: Dict,
+                 base_path: str = '',
+                 file_attributes: Optional[List[str]] = None,
+                 ):
+
+        if file_attributes is None:
+            file_attributes = DEFAULT_FILE_ATTRIBUTE_NAMES
+
+        if base_path and file_attributes:
+            # this adds a base path to all paths in the configuration
+            add_base_path(conf, base_path, file_attributes)
+
+    def __getitem__(self, item):
+        if item in self.__dict__:
+            return self.__dict__[item]
+        else:
+            raise KeyError
+
+    def get(self, item, default=None):
+        return getattr(self, item, default)
+
+    def __contains__(self, item):
+        return item in self.__dict__
+
+    def items(self):
+        for key in self.__dict__:
+            if key.startswith('__') and key.endswith('__'):
+                continue
+            yield key, getattr(self, key)
+
+    def extend(self, entity_conf, conf, base_path, file_attributes, domain, port):
+        for econf in entity_conf:
+            _path = econf.get("path")
+            _cnf = conf
+            if _path:
+                for step in _path:
+                    _cnf = _cnf[step]
+            _attr = econf["attr"]
+            _cls = econf["class"]
+            setattr(self, _attr,
+                    _cls(_cnf, base_path=base_path, file_attributes=file_attributes,
+                         domain=domain, port=port))
+
+
+class Configuration(Base):
+    """Server Configuration"""
+
+    def __init__(self,
+                 conf: Dict,
+                 base_path: str = '',
+                 entity_conf: Optional[List[dict]] = None,
+                 file_attributes: Optional[List[str]] = None,
+                 domain: Optional[str] = "",
+                 port: Optional[int] = 0,
+                 ):
+        Base.__init__(self, conf, base_path=base_path, file_attributes=file_attributes)
+
+        log_conf = conf.get('logging')
+        if log_conf:
+            self.logger = configure_logging(config=log_conf).getChild(__name__)
+        else:
+            self.logger = logging.getLogger('oidcrp')
+
+        self.web_conf = lower_or_upper(conf, "webserver")
+
+        # entity info
+        if not domain:
+            domain = conf.get("domain", "127.0.0.1")
+
+        if not port:
+            port = conf.get("port", 80)
+
+        if entity_conf:
+            self.extend(entity_conf=entity_conf, conf=conf, base_path=base_path,
+                        file_attributes=file_attributes, domain=domain, port=port)
+
+
+def create_from_config_file(cls,
+                            filename: str,
+                            base_path: Optional[str] = '',
+                            entity_conf: Optional[List[dict]] = None,
+                            file_attributes: Optional[List[str]] = None,
+                            domain: Optional[str] = "",
+                            port: Optional[int] = 0):
+    if filename.endswith(".yaml"):
+        """Load configuration as YAML"""
+        _cnf = load_yaml_config(filename)
+    elif filename.endswith(".json"):
+        _str = open(filename).read()
+        _cnf = json.loads(_str)
+    elif filename.endswith(".py"):
+        head, tail = os.path.split(filename)
+        tail = tail[:-3]
+        module = importlib.import_module(tail)
+        _cnf = getattr(module, "CONFIG")
+    else:
+        raise ValueError("Unknown file type")
+
+    return cls(_cnf,
+               entity_conf=entity_conf,
+               base_path=base_path, file_attributes=file_attributes,
+               domain=domain, port=port)

src/oidcmsg/logging.py

@@ -0,0 +1,54 @@
+"""Common logging functions"""
+
+import os
+import logging
+from logging.config import dictConfig
+from typing import Optional
+
+import yaml
+
+
+LOGGING_CONF = 'logging.yaml'
+
+LOGGING_DEFAULT = {
+    'version': 1,
+    'formatters': {
+        'default': {
+            'format': '%(asctime)s %(name)s %(levelname)s %(message)s'
+        }
+    },
+    'handlers': {
+        'default': {
+            'class': 'logging.StreamHandler',
+            'formatter': 'default'
+        }
+    },
+    'root': {
+        'handlers': ['default'],
+        'level': 'INFO'
+    }
+}
+
+
+def configure_logging(debug: Optional[bool] = False,
+                      config: Optional[dict] = None,
+                      filename: Optional[str] = LOGGING_CONF) -> logging.Logger:
+    """Configure logging"""
+
+    if config is not None:
+        config_dict = config
+        config_source = 'dictionary'
+    elif filename is not None and os.path.exists(filename):
+        with open(filename, "rt") as file:
+            config_dict = yaml.load(file)
+        config_source = 'file'
+    else:
+        config_dict = LOGGING_DEFAULT
+        config_source = 'default'
+
+    if debug:
+        config_dict['root']['level'] = 'DEBUG'
+
+    dictConfig(config_dict)
+    logging.debug("Configured logging using %s", config_source)
+    return logging.getLogger()

src/oidcmsg/util.py

@@ -0,0 +1,20 @@
+import secrets
+
+import yaml
+
+
+def rndstr(size=16):
+    """
+    Returns a string of random url safe characters
+
+    :param size: The length of the string
+    :return: string
+    """
+    return secrets.token_urlsafe(size)
+
+
+def load_yaml_config(filename):
+    """Load a YAML configuration file."""
+    with open(filename, "rt", encoding='utf-8') as file:
+        config_dict = yaml.safe_load(file)
+    return config_dict