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 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


class _Reader:
    """Reads variables and collects every problem instead of stopping at the first."""

    def __init__(self, environ: Mapping[str, str]) -> None:
        self._environ = environ
        self.problems: list[str] = []

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

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

    def problem(self, message: str) -> None:
        self.problems.append(message)


def _read_debug(r: _Reader) -> bool:
    value = r.raw("DEBUG")
    if value is None:
        return False
    if value.lower() in _TRUE:
        return True
    if value.lower() not in _FALSE:
        r.problem("DEBUG: must be true or false")
    return False


def _read_trusted_proxies(r: _Reader) -> list[str]:
    proxies: list[str] = []
    value = r.raw("TRUSTED_PROXIES")
    for item in (value or "").split(","):
        item = item.strip()
        if not item:
            continue
        try:
            ipaddress.ip_network(item, strict=False)
        except ValueError:
            r.problem("TRUSTED_PROXIES: must be comma-separated IP addresses or networks")
            break
        proxies.append(item)
    return proxies


def _read_core(r: _Reader) -> dict:
    secret_key = r.raw("SECRET_KEY")
    if secret_key is None:
        r.problem("SECRET_KEY: required")
    elif len(secret_key) < MIN_SECRET_KEY_LENGTH:
        r.problem(f"SECRET_KEY: must be at least {MIN_SECRET_KEY_LENGTH} characters")

    hosts_raw = r.raw("ALLOWED_HOSTS")
    allowed_hosts: list[str] = []
    if hosts_raw is None:
        r.problem("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:
            r.problem("ALLOWED_HOSTS: must contain at least one host name")

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

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

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

    return {
        "secret_key": secret_key or "",
        "allowed_hosts": allowed_hosts,
        "data_dir": data_dir,
        "debug": _read_debug(r),
        "trusted_proxies": _read_trusted_proxies(r),
        "log_level": log_level,
        "default_timezone": timezone,
    }


def _read_directory(r: _Reader) -> DirectoryConfig:
    server = r.required("LDAP_SERVER", "host name of the directory server, with optional :port")
    if "://" in server or "/" in server:
        r.problem("LDAP_SERVER: must be a host name, not a URL (LDAPS is always used)")
    user_ou_dn = r.required("LDAP_USER_OU_DN", "distinguished name of the OU that holds the users")
    bind_dn = r.required("LDAP_BIND_DN", "service account distinguished name or principal name")
    bind_password = r.required("LDAP_BIND_PASSWORD", "password of the service account")
    publisher_group_dn = r.required("LDAP_PUBLISHER_GROUP_DN", "distinguished name of the group")
    admin_group_dn = r.required("LDAP_ADMIN_GROUP_DN", "distinguished name of the group")
    ca_file = Path(r.required("LDAP_CA_FILE", "path of the PEM file with the CA chain"))
    if r.raw("LDAP_CA_FILE") is not None and not ca_file.is_file():
        r.problem("LDAP_CA_FILE: must be an existing file")
    return 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,
    )


def _read_oidc(r: _Reader, *, debug: bool) -> OidcConfig:
    issuer = r.required("OIDC_ISSUER", "issuer address from the provider's OpenID configuration")
    client_id = r.required("OIDC_CLIENT_ID", "client ID of the provider's application")
    client_secret = r.required("OIDC_CLIENT_SECRET", "client secret of the provider's application")
    base_url = r.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":
        r.problem("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:
            r.problem("PUBLIC_BASE_URL: must be an https origin such as https://host")
    return OidcConfig(
        issuer=issuer,
        client_id=client_id,
        client_secret=client_secret,
        public_base_url=base_url,
    )


def _read_smtp(r: _Reader) -> SmtpConfig:
    host = r.required("SMTP_HOST", "host name of the mail server")
    if "://" in host or "/" in host or ":" in host:
        r.problem("SMTP_HOST: must be a host name only; set the port in SMTP_PORT")
    security = (r.raw("SMTP_SECURITY") or "starttls").lower()
    if security not in SMTP_SECURITY_MODES:
        r.problem(f"SMTP_SECURITY: must be one of {', '.join(SMTP_SECURITY_MODES)}")
        security = "starttls"
    port = SMTP_DEFAULT_PORTS[security]
    port_raw = r.raw("SMTP_PORT")
    if port_raw is not None:
        if port_raw.isdigit() and 0 < int(port_raw) < 65536:
            port = int(port_raw)
        else:
            r.problem("SMTP_PORT: must be a port number from 1 to 65535")
    username = r.required("SMTP_USERNAME", "user name of the mailbox")
    password = r.required("SMTP_PASSWORD", "password of the mailbox")
    sender = r.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("@")
    ):
        r.problem("MAIL_FROM: must be an e-mail address, optionally with a display name")
    return SmtpConfig(
        host=host, port=port, security=security, username=username, password=password, sender=sender
    )


def load_config(environ: Mapping[str, str]) -> Config:
    reader = _Reader(environ)
    core = _read_core(reader)
    directory = _read_directory(reader)
    oidc = _read_oidc(reader, debug=core["debug"])
    smtp = _read_smtp(reader)
    if reader.problems:
        raise ConfigError(reader.problems)
    return Config(**core, directory=directory, oidc=oidc, smtp=smtp)