Merge branch 'conf-class' into 'main'

use pydantic to validate the configuration

See merge request yaal/canaille!170

canaille/__init__.py

@@ -11,7 +11,7 @@
 
 
 def setup_sentry(app):  # pragma: no cover
-    if not app.config.get("SENTRY_DSN"):
+    if not app.config["CANAILLE"]["SENTRY_DSN"]:
         return None
 
     try:
@@ -21,12 +21,14 @@ def setup_sentry(app):  # pragma: no cover
     except Exception:
         return None
 
-    sentry_sdk.init(dsn=app.config["SENTRY_DSN"], integrations=[FlaskIntegration()])
+    sentry_sdk.init(
+        dsn=app.config["CANAILLE"]["SENTRY_DSN"], integrations=[FlaskIntegration()]
+    )
     return sentry_sdk
 
 
 def setup_logging(app):
-    conf = app.config.get("LOGGING")
+    conf = app.config["CANAILLE"]["LOGGING"]
     if conf is None:
         log_level = "DEBUG" if app.debug else "INFO"
         dictConfig(
@@ -72,7 +74,7 @@ def setup_blueprints(app):
 
     app.register_blueprint(canaille.core.endpoints.bp)
 
-    if "OIDC" in app.config:
+    if "CANAILLE_OIDC" in app.config:
         import canaille.oidc.endpoints
 
         app.register_blueprint(canaille.oidc.endpoints.bp)
@@ -92,19 +94,24 @@ def global_processor():
 
         return {
             "debug": app.debug or app.config.get("TESTING", False),
-            "has_smtp": "SMTP" in app.config,
-            "has_oidc": "OIDC" in app.config,
-            "has_password_recovery": app.config.get("ENABLE_PASSWORD_RECOVERY", True),
-            "has_registration": app.config.get("ENABLE_REGISTRATION", False),
+            "has_smtp": "SMTP" in app.config["CANAILLE"],
+            "has_oidc": "CANAILLE_OIDC" in app.config["CANAILLE"],
+            "has_password_recovery": app.config["CANAILLE"]["ENABLE_PASSWORD_RECOVERY"],
+            "has_registration": app.config["CANAILLE"]["ENABLE_REGISTRATION"],
             "has_account_lockability": app.backend.get().has_account_lockability(),
-            "logo_url": app.config.get("LOGO"),
-            "favicon_url": app.config.get("FAVICON", app.config.get("LOGO")),
-            "website_name": app.config.get("NAME", "Canaille"),
+            "logo_url": app.config["CANAILLE"]["LOGO"],
+            "favicon_url": app.config["CANAILLE"]["FAVICON"]
+            or app.config["CANAILLE"]["LOGO"],
+            "website_name": app.config["CANAILLE"]["NAME"],
             "user": current_user(),
             "menu": True,
             "is_boosted": request.headers.get("HX-Boosted", False),
-            "has_email_confirmation": app.config.get("EMAIL_CONFIRMATION") is True
-            or (app.config.get("EMAIL_CONFIRMATION") is None and "SMTP" in app.config),
+            "has_email_confirmation": app.config["CANAILLE"]["EMAIL_CONFIRMATION"]
+            is True
+            or (
+                app.config["CANAILLE"]["EMAIL_CONFIRMATION"] is None
+                and "SMTP" in app.config["CANAILLE"]
+            ),
         }
 
 
@@ -137,7 +144,7 @@ def create_app(config=None, validate=True, backend=None):
         setup_themer(app)
         setup_flask(app)
 
-        if "OIDC" in app.config:
+        if "CANAILLE_OIDC" in app.config:
             from .oidc.oauth import setup_oauth
 
             setup_oauth(app)

canaille/app/__init__.py

@@ -25,7 +25,7 @@ def build_hash(*args):
 def default_fields():
     read = set()
     write = set()
-    for acl in current_app.config.get("ACL", {}).values():
+    for acl in current_app.config["CANAILLE"]["ACL"].values():
         if not acl.get("FILTER"):
             read |= set(acl.get("READ", []))
             write |= set(acl.get("WRITE", []))
@@ -41,8 +41,8 @@ def get_current_domain():
 
 
 def get_current_mail_domain():
-    if current_app.config["SMTP"].get("FROM_ADDR"):
-        return current_app.config["SMTP"]["FROM_ADDR"].split("@")[-1]
+    if current_app.config["CANAILLE"]["SMTP"]["FROM_ADDR"]:
+        return current_app.config["CANAILLE"]["SMTP"]["FROM_ADDR"].split("@")[-1]
 
     return get_current_domain().split(":")[0]
 

canaille/app/configuration.py

@@ -2,41 +2,101 @@
 import smtplib
 import socket
 import sys
-from collections.abc import Mapping
+from typing import Optional
 
 from flask import current_app
+from pydantic import create_model
+from pydantic_settings import BaseSettings
+from pydantic_settings import SettingsConfigDict
 
-from canaille.app.mails import DEFAULT_SMTP_HOST
-from canaille.app.mails import DEFAULT_SMTP_PORT
+from canaille.core.configuration import CoreSettings
 
-ROOT = os.path.dirname(os.path.abspath(__file__))
+ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 
 
-class ConfigurationException(Exception):
-    pass
+class RootSettings(BaseSettings):
+    """The top-level namespace contains holds the configuration settings
+    unrelated to Canaille. The configuration paramateres from the following
+    libraries can be used:
+
+    - :doc:`Flask <flask:config>`
+    - :doc:`Flask-WTF <flask-wtf:config>`
+    - :doc:`Flask-Babel <flask-babel:index>`
+    - :doc:`Authlib <authlib:flask/2/authorization-server>`
+    """
+
+    model_config = SettingsConfigDict(
+        extra="allow",
+        env_nested_delimiter="__",
+        case_sensitive=True,
+        env_file=".env",
+    )
+
+    SECRET_KEY: str
+    """The Flask :external:py:data:`SECRET_KEY` configuration setting.
+
+    You MUST change this.
+    """
+
+    SERVER_NAME: Optional[str] = None
+    """The Flask :external:py:data:`SERVER_NAME` configuration setting.
+
+    This sets domain name on which canaille will be served.
+    """
+
+    PREFERRED_URL_SCHEME: str = "https"
+    """The Flask :external:py:data:`PREFERRED_URL_SCHEME` configuration
+    setting.
+
+    This sets the url scheme by which canaille will be served.
+    """
+
+    DEBUG: bool = False
+    """The Flask :external:py:data:`DEBUG` configuration setting.
+
+    This enables debug options. This is useful for development but
+    should be absolutely avoided in production environments.
+    """
 
 
-def parse_file_keys(config):
-    """Replaces configuration entries with the '_FILE' suffix with the matching
-    file content."""
+def settings_factory(config):
+    """Overly complicated function that pushes the backend specific
+    configuration into CoreSettings, in the purpose break dependency against
+    backends libraries like python-ldap or sqlalchemy."""
+    attributes = {"CANAILLE": (Optional[CoreSettings], None)}
 
-    SUFFIX = "_FILE"
-    new_config = {}
-    for key, value in config.items():
-        if isinstance(value, Mapping):
-            new_config[key] = parse_file_keys(value)
+    if "CANAILLE_SQL" in config or any(
+        var.startswith("CANAILLE_SQL__") for var in os.environ
+    ):
+        from canaille.backends.sql.configuration import SQLSettings
+
+        attributes["CANAILLE_SQL"] = (Optional[SQLSettings], None)
 
-        elif isinstance(key, str) and key.endswith(SUFFIX) and isinstance(value, str):
-            with open(value) as f:
-                value = f.read().rstrip("\n")
+    if "CANAILLE_LDAP" in config or any(
+        var.startswith("CANAILLE__LDAP__") for var in os.environ
+    ):
+        from canaille.backends.ldap.configuration import LDAPSettings
 
-            root_key = key[: -len(SUFFIX)]
-            new_config[root_key] = value
+        attributes["CANAILLE_LDAP"] = (Optional[LDAPSettings], None)
 
-        else:
-            new_config[key] = value
+    if "CANAILLE_OIDC" in config or any(
+        var.startswith("CANAILLE_OIDC__") for var in os.environ
+    ):
+        from canaille.oidc.configuration import OIDCSettings
+
+        attributes["CANAILLE_OIDC"] = (Optional[OIDCSettings], None)
+
+    Settings = create_model(
+        "Settings",
+        __base__=RootSettings,
+        **attributes,
+    )
 
-    return new_config
+    return Settings(**config, _secrets_dir=os.environ.get("SECRETS_DIR"))
+
+
+class ConfigurationException(Exception):
+    pass
 
 
 def toml_content(file_path):
@@ -55,7 +115,7 @@ def toml_content(file_path):
         raise Exception("toml library not installed. Cannot load configuration.")
 
 
-def setup_config(app, config=None, validate_config=True):
+def setup_config(app, config=None, test_config=True):
     from canaille.oidc.installation import install
 
     app.config.from_mapping(
@@ -65,69 +125,62 @@ def setup_config(app, config=None, validate_config=True):
             "OAUTH2_ACCESS_TOKEN_GENERATOR": "canaille.oidc.oauth.generate_access_token",
         }
     )
+    if not config and "CONFIG" in os.environ:
+        config = toml_content(os.environ.get("CONFIG"))
 
-    if config:
-        app.config.from_mapping(parse_file_keys(config))
-
-    elif "CONFIG" in os.environ:
-        app.config.from_mapping(parse_file_keys(toml_content(os.environ["CONFIG"])))
-
-    else:
-        raise Exception(
-            "No configuration file found. "
-            "Either create conf/config.toml or set the 'CONFIG' variable environment."
-        )
+    config_obj = settings_factory(config)
+    app.config.from_mapping(config_obj.model_dump())
 
     if app.debug:
         install(app.config, debug=True)
 
-    if validate_config:
+    if test_config:
         validate(app.config)
 
 
 def validate(config, validate_remote=False):
-    validate_keypair(config)
-    validate_theme(config)
+    validate_keypair(config.get("CANAILLE_OIDC"))
+    validate_theme(config["CANAILLE"])
 
     if not validate_remote:
         return
 
     from canaille.backends import BaseBackend
 
     BaseBackend.get().validate(config)
-    validate_smtp_configuration(config)
+    validate_smtp_configuration(config["CANAILLE"]["SMTP"])
 
 
 def validate_keypair(config):
     if (
-        config.get("OIDC")
-        and config["OIDC"].get("JWT")
-        and not config["OIDC"]["JWT"].get("PUBLIC_KEY")
+        config
+        and config["JWT"]
+        and not config["JWT"]["PUBLIC_KEY"]
         and not current_app.debug
     ):
         raise ConfigurationException("No public key has been set")
 
     if (
-        config.get("OIDC")
-        and config["OIDC"].get("JWT")
-        and not config["OIDC"]["JWT"].get("PRIVATE_KEY")
+        config
+        and config["JWT"]
+        and not config["JWT"]["PRIVATE_KEY"]
         and not current_app.debug
     ):
         raise ConfigurationException("No private key has been set")
 
 
 def validate_smtp_configuration(config):
-    host = config["SMTP"].get("HOST", DEFAULT_SMTP_HOST)
-    port = config["SMTP"].get("PORT", DEFAULT_SMTP_PORT)
+    host = config["HOST"]
+    port = config["PORT"]
     try:
         with smtplib.SMTP(host=host, port=port) as smtp:
-            if config["SMTP"].get("TLS"):
+            if config["TLS"]:
                 smtp.starttls()
 
-            if config["SMTP"].get("LOGIN"):
+            if config["LOGIN"]:
                 smtp.login(
-                    user=config["SMTP"]["LOGIN"],
-                    password=config["SMTP"].get("PASSWORD"),
+                    user=config["LOGIN"],
+                    password=config["PASSWORD"],
                 )
     except (socket.gaierror, ConnectionRefusedError) as exc:
         raise ConfigurationException(
@@ -136,15 +189,15 @@ def validate_smtp_configuration(config):
 
     except smtplib.SMTPAuthenticationError as exc:
         raise ConfigurationException(
-            f'SMTP authentication failed with user \'{config["SMTP"]["LOGIN"]}\''
+            f'SMTP authentication failed with user \'{config["LOGIN"]}\''
         ) from exc
 
     except smtplib.SMTPNotSupportedError as exc:
         raise ConfigurationException(exc) from exc
 
 
 def validate_theme(config):
-    if not config.get("THEME"):
+    if not config or not config["THEME"]:
         return
 
     if not os.path.exists(config["THEME"]) and not os.path.exists(

canaille/app/flask.py

@@ -91,7 +91,7 @@ def smtp_needed():
     def wrapper(view_function):
         @wraps(view_function)
         def decorator(*args, **kwargs):
-            if "SMTP" in current_app.config:
+            if "SMTP" in current_app.config["CANAILLE"]:
                 return view_function(*args, **kwargs)
 
             message = _("No SMTP server has been configured")

canaille/app/i18n.py

@@ -53,8 +53,8 @@ def locale_selector():
     if user is not None and user.preferred_language in available_language_codes:
         return user.preferred_language
 
-    if current_app.config.get("LANGUAGE"):
-        return current_app.config.get("LANGUAGE")
+    if current_app.config["CANAILLE"]["LANGUAGE"]:
+        return current_app.config["CANAILLE"]["LANGUAGE"]
 
     return request.accept_languages.best_match(available_language_codes)
 
@@ -67,7 +67,7 @@ def timezone_selector():
     from babel.dates import LOCALTZ
 
     try:
-        return pytz.timezone(current_app.config.get("TIMEZONE"))
+        return pytz.timezone(current_app.config["CANAILLE"]["TIMEZONE"])
     except pytz.exceptions.UnknownTimeZoneError:
         return LOCALTZ