ping-mee-self-service-platform/portal/config/env.py

"""Typed environment configuration that reports every problem at once (AD-13)."""

from __future__ import annotations

import ipaddress
from collections.abc import Mapping
from dataclasses import dataclass, field
from pathlib import Path
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError

LOG_LEVELS = ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL")
MIN_SECRET_KEY_LENGTH = 32
_TRUE = {"1", "true", "yes", "on"}
_FALSE = {"0", "false", "no", "off"}


class ConfigError(Exception):
    """Raised with one message that lists every invalid or missing variable.

    Messages name variables and the rule they break, never their values.
    """

    def __init__(self, problems: list[str]) -> None:
        self.problems = problems
        super().__init__(self.render(problems))

    @staticmethod
    def render(problems: list[str]) -> str:
        lines = ["Invalid configuration. Fix these environment variables:"]
        lines.extend(f"  - {p}" for p in problems)
        return "\n".join(lines)


@dataclass(frozen=True)
class DirectoryConfig:
    server: str  # host name, optionally host:port; always reached over LDAPS
    user_ou_dn: str
    bind_dn: str  # distinguished name or user principal name of the service account
    bind_password: str = field(repr=False)
    ca_file: Path
    publisher_group_dn: str
    admin_group_dn: str


@dataclass(frozen=True)
class Config:
    secret_key: str
    allowed_hosts: list[str]
    data_dir: Path
    debug: bool
    trusted_proxies: list[str]
    log_level: str
    default_timezone: str
    directory: DirectoryConfig


def load_config(environ: Mapping[str, str]) -> Config:
    problems: list[str] = []

    def raw(name: str) -> str | None:
        value = environ.get(name)
        if value is None or value.strip() == "":
            return None
        return value.strip()

    secret_key = raw("SECRET_KEY")
    if secret_key is None:
        problems.append("SECRET_KEY: required")
    elif len(secret_key) < MIN_SECRET_KEY_LENGTH:
        problems.append(f"SECRET_KEY: must be at least {MIN_SECRET_KEY_LENGTH} characters")

    hosts_raw = raw("ALLOWED_HOSTS")
    allowed_hosts: list[str] = []
    if hosts_raw is None:
        problems.append("ALLOWED_HOSTS: required, comma-separated host names")
    else:
        allowed_hosts = [h.strip() for h in hosts_raw.split(",") if h.strip()]
        if not allowed_hosts:
            problems.append("ALLOWED_HOSTS: must contain at least one host name")

    data_dir = Path(raw("DATA_DIR") or "/data")
    if not data_dir.is_dir():
        problems.append("DATA_DIR: must be an existing directory")
    else:
        data_dir = data_dir.resolve()

    debug = False
    debug_raw = raw("DEBUG")
    if debug_raw is not None:
        if debug_raw.lower() in _TRUE:
            debug = True
        elif debug_raw.lower() not in _FALSE:
            problems.append("DEBUG: must be true or false")

    proxies: list[str] = []
    proxies_raw = raw("TRUSTED_PROXIES")
    if proxies_raw is not None:
        for item in proxies_raw.split(","):
            item = item.strip()
            if not item:
                continue
            try:
                ipaddress.ip_network(item, strict=False)
            except ValueError:
                problems.append("TRUSTED_PROXIES: must be comma-separated IP addresses or networks")
                break
            proxies.append(item)

    log_level = (raw("LOG_LEVEL") or "INFO").upper()
    if log_level not in LOG_LEVELS:
        problems.append(f"LOG_LEVEL: must be one of {', '.join(LOG_LEVELS)}")

    timezone = raw("DEFAULT_TIMEZONE") or "Europe/Berlin"
    try:
        ZoneInfo(timezone)
    except (ZoneInfoNotFoundError, ValueError, OSError):
        problems.append("DEFAULT_TIMEZONE: must be a valid IANA time zone name")

    def required(name: str, rule: str) -> str:
        value = raw(name)
        if value is None:
            problems.append(f"{name}: required, {rule}")
            return ""
        return value

    server = required("LDAP_SERVER", "host name of the directory server, with optional :port")
    if "://" in server or "/" in server:
        problems.append("LDAP_SERVER: must be a host name, not a URL (LDAPS is always used)")
    user_ou_dn = required("LDAP_USER_OU_DN", "distinguished name of the OU that holds the users")
    bind_dn = required("LDAP_BIND_DN", "service account distinguished name or user principal name")
    bind_password = required("LDAP_BIND_PASSWORD", "password of the service account")
    publisher_group_dn = required("LDAP_PUBLISHER_GROUP_DN", "distinguished name of the group")
    admin_group_dn = required("LDAP_ADMIN_GROUP_DN", "distinguished name of the group")
    ca_file = Path(required("LDAP_CA_FILE", "path of the PEM file with the CA chain"))
    if raw("LDAP_CA_FILE") is not None and not ca_file.is_file():
        problems.append("LDAP_CA_FILE: must be an existing file")

    if problems:
        raise ConfigError(problems)

    return Config(
        secret_key=secret_key or "",
        allowed_hosts=allowed_hosts,
        data_dir=data_dir,
        debug=debug,
        trusted_proxies=proxies,
        log_level=log_level,
        default_timezone=timezone,
        directory=DirectoryConfig(
            server=server,
            user_ou_dn=user_ou_dn,
            bind_dn=bind_dn,
            bind_password=bind_password,
            ca_file=ca_file,
            publisher_group_dn=publisher_group_dn,
            admin_group_dn=admin_group_dn,
        ),
    )