"""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 = "noreply@pescoinc.biz" smtp_starttls: bool = True smtp_ca_bundle: Path | None = None notify_dry_run: bool = False @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 @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") @lru_cache(maxsize=1) def get_settings() -> Settings: return Settings() def clear_settings_cache() -> None: get_settings.cache_clear()