"""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, ), )