PacketStation/server/packetstation/settings.py

"""Settings from environment variables (no .env autoload; PS_ENV_FILE points at one for dev)."""

from __future__ import annotations

import os
from functools import lru_cache
from pathlib import Path
from typing import Any, Literal

from pydantic import Field, SecretStr, field_validator, model_validator
from pydantic_settings import BaseSettings, SettingsConfigDict

ENV_FILE_VAR = "PS_ENV_FILE"
Stack = Literal["prod", "test", "ci", "dev"]


class SecretFileError(Exception):
    """A secret file is missing, unreadable or empty. The message names the file, never its content."""


def read_secret(path: Path | str, what: str = "secret") -> str:
    """Read a password or key file, stripping trailing CR/LF. Empty is an error."""
    p = Path(path)
    try:
        text = p.read_text(encoding="utf-8")
    except OSError as e:
        raise SecretFileError(f"can't read {what} file {p}: {e.strerror or type(e).__name__}") from None
    value = text.rstrip("\r\n")
    if not value:
        raise SecretFileError(f"{what} file {p} is empty")
    return value


_OPTIONAL_PATHS = ("ldap_ca_bundle", "smtp_ca_bundle")


class Settings(BaseSettings):
    model_config = SettingsConfigDict(extra="ignore", case_sensitive=False)

    def __init__(self, **values: Any):
        # Containers set no PS_ENV_FILE, so only the real environment is read there
        values.setdefault("_env_file", os.environ.get(ENV_FILE_VAR) or None)
        super().__init__(**values)

    stack: Stack
    site_id: str = "PESCO"
    site_tz: str = "America/Denver"
    public_url: str = ""
    trusted_proxy_host: str = "npm_app_1"

    pg_host: str = "postgres"
    pg_port: int = 5432
    pg_database: str = "packetstation"
    pg_app_user: str = "ps_app"
    pg_app_password_file: Path = Path("/run/secrets/postgres_app_password")
    pg_migrate_user: str = "ps_migrate"
    pg_migrate_password_file: Path = Path("/run/secrets/postgres_migrate_password")
    pg_superuser_url: SecretStr | None = Field(default=None, description="dev/test only, for --bootstrap-roles")

    assets_dir: Path = Path("/data/assets")
    backup_dir: Path = Path("/backups")
    session_key_file: Path = Path("/run/secrets/session_key")
    seal_key_file: Path = Path("/run/secrets/seal_key")
    static_dir: Path = Path("/app/static")
    worker_beat_file: Path = Path("/tmp/ps-worker.beat")  # noqa: S108 (private /tmp in the container)
    ps_build: str = "dev"
    log_level: str = "INFO"

    ldap_url: str = ""
    ldap_base_dn: str = ""
    ldap_bind_dn: str = ""
    ldap_password_file: Path = Path("/run/secrets/ldap_bind_password")
    ldap_user_filter: str = "(|(sAMAccountName={0})(userPrincipalName={0}))"
    ldap_group_base_dn: str = ""
    ldap_ca_bundle: Path | None = None
    ldap_starttls: bool = False
    ldap_allow_insecure: bool = False

    smtp_host: str = "app-03.pesco.com"
    smtp_port: int = 25
    smtp_from: str = "[email protected]"
    smtp_starttls: bool = True
    smtp_ca_bundle: Path | None = None
    notify_dry_run: bool = False

    session_idle_minutes: int = Field(default=15, ge=1)
    session_lock_minutes: int = Field(default=3, ge=1)
    session_hidden_lock_seconds: int = Field(default=60, ge=0)
    session_password_hours: int = Field(default=12, ge=1)
    session_pin_hours: int = Field(default=10, ge=1)
    roles_refresh_minutes: int = Field(default=15, ge=1)
    signin_account_limit: int = Field(default=5, ge=1)
    signin_account_window_minutes: int = Field(default=5, ge=1)
    signin_account_cooldown_minutes: int = Field(default=10, ge=0)
    signin_ip_limit: int = Field(default=30, ge=1)
    signin_ip_window_minutes: int = Field(default=5, ge=1)
    pin_max_failures: int = Field(default=5, ge=1)
    stepup_max_failures: int = Field(default=3, ge=1)
    device_cookie_days: int = Field(default=365, ge=1)
    pin_min_digits: int = Field(default=6, ge=4)
    pin_max_digits: int = Field(default=12, ge=4)

    @model_validator(mode="before")
    @classmethod
    def _blank_is_unset(cls, data: Any) -> Any:
        if isinstance(data, dict):
            data = dict(data)
            for key in list(data):
                if key.lower() in _OPTIONAL_PATHS and data[key] in ("", None):
                    data[key] = None
            for key in list(data):
                if key.lower() == "stack" and isinstance(data[key], str):
                    data[key] = data[key].strip().lower()
        return data

    @field_validator("log_level")
    @classmethod
    def _level(cls, v: str) -> str:
        v = v.strip().upper()
        if v not in ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"):
            raise ValueError("LOG_LEVEL must be DEBUG, INFO, WARNING, ERROR or CRITICAL")
        return v

    @model_validator(mode="after")
    def _insecure_ldap(self) -> Settings:
        if self.stack == "prod" and self.ldap_allow_insecure:
            raise ValueError("LDAP_ALLOW_INSECURE=true is refused on STACK=prod")
        return self

    @model_validator(mode="after")
    def _pin_digits(self) -> Settings:
        if self.pin_min_digits > self.pin_max_digits:
            raise ValueError("PIN_MIN_DIGITS must not exceed PIN_MAX_DIGITS")
        return self

    @property
    def ldap_urls(self) -> list[str]:
        return self.ldap_url.split()

    @property
    def is_prod(self) -> bool:
        return self.stack == "prod"

    def pg_password(self, role: str) -> str:
        if role == "migrate":
            return read_secret(self.pg_migrate_password_file, "PG_MIGRATE_PASSWORD_FILE")
        return read_secret(self.pg_app_password_file, "PG_APP_PASSWORD_FILE")

    def pg_user(self, role: str) -> str:
        return self.pg_migrate_user if role == "migrate" else self.pg_app_user

    def ldap_password(self) -> str:
        return read_secret(self.ldap_password_file, "LDAP_PASSWORD_FILE")

    def session_key(self) -> bytes:
        """The HMAC key for session and device tokens, from SESSION_KEY_FILE."""
        key = read_secret(self.session_key_file, "SESSION_KEY_FILE").strip()
        if len(key) < 16:
            raise SecretFileError(f"SESSION_KEY_FILE file {self.session_key_file} holds a key that is too short")
        return key.encode("utf-8")


@lru_cache(maxsize=1)
def get_settings() -> Settings:
    return Settings()


def clear_settings_cache() -> None:
    get_settings.cache_clear()