"""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 email.utils import parseaddr from pathlib import Path from urllib.parse import urlsplit from zoneinfo import ZoneInfo, ZoneInfoNotFoundError SMTP_SECURITY_MODES = ("starttls", "ssl") SMTP_DEFAULT_PORTS = {"starttls": 587, "ssl": 465} 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 OidcConfig: issuer: str # exactly as the provider reports it, including any trailing slash client_id: str client_secret: str = field(repr=False) public_base_url: str # origin the users reach the portal at; no trailing slash @property def redirect_uri(self) -> str: return f"{self.public_base_url}/auth/callback" @dataclass(frozen=True) class SmtpConfig: host: str port: int security: str # "starttls" or "ssl"; mail is never sent unencrypted username: str password: str = field(repr=False) sender: str # the From address of every e-mail @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 oidc: OidcConfig smtp: SmtpConfig 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") issuer = required("OIDC_ISSUER", "issuer address from the provider's OpenID configuration") client_id = required("OIDC_CLIENT_ID", "client ID of the provider's application") client_secret = required("OIDC_CLIENT_SECRET", "client secret of the provider's application") base_url = required("PUBLIC_BASE_URL", "address users reach the portal at, e.g. https://host") base_url = base_url.rstrip("/") if issuer and urlsplit(issuer).scheme != "https": problems.append("OIDC_ISSUER: must be an https address") if base_url: parts = urlsplit(base_url) allowed = ("https", "http") if debug else ("https",) if parts.scheme not in allowed or not parts.netloc or parts.path or parts.query: problems.append("PUBLIC_BASE_URL: must be an https origin such as https://host") smtp_host = required("SMTP_HOST", "host name of the mail server") if "://" in smtp_host or "/" in smtp_host or ":" in smtp_host: problems.append("SMTP_HOST: must be a host name only; set the port in SMTP_PORT") smtp_security = (raw("SMTP_SECURITY") or "starttls").lower() if smtp_security not in SMTP_SECURITY_MODES: problems.append(f"SMTP_SECURITY: must be one of {', '.join(SMTP_SECURITY_MODES)}") smtp_security = "starttls" smtp_port = SMTP_DEFAULT_PORTS[smtp_security] port_raw = raw("SMTP_PORT") if port_raw is not None: if port_raw.isdigit() and 0 < int(port_raw) < 65536: smtp_port = int(port_raw) else: problems.append("SMTP_PORT: must be a port number from 1 to 65535") smtp_username = required("SMTP_USERNAME", "user name of the mailbox") smtp_password = required("SMTP_PASSWORD", "password of the mailbox") sender = required("MAIL_FROM", "sender address of the portal's e-mails") address = parseaddr(sender)[1] if sender and ( address.count("@") != 1 or any(c in sender for c in chr(13) + chr(10)) or address.startswith("@") or address.endswith("@") ): problems.append("MAIL_FROM: must be an e-mail address, optionally with a display name") 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, ), oidc=OidcConfig( issuer=issuer, client_id=client_id, client_secret=client_secret, public_base_url=base_url, ), smtp=SmtpConfig( host=smtp_host, port=smtp_port, security=smtp_security, username=smtp_username, password=smtp_password, sender=sender, ), )