"""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"] FeedSource = Literal["visualkit", "fixture"] DrawingSource = Literal["mfiles", "fixture"] 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", "archive_root") 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") # The archive share (Gitea issue 7). Unset or blank turns the archive off. The Admin picks a folder under it. archive_root: Path | None = None 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 session_idle_minutes: int = Field(default=15, ge=1) session_lock_minutes: int = Field(default=3, ge=1) # Admin and Clerical sit at their own desks: sign-out and screen lock wait at least this long (issue #45) session_desk_minutes: int = Field(default=60, 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) upload_max_bytes: int = Field(default=25 * 1024 * 1024, ge=1024) upload_max_bytes_bom: int = Field(default=5 * 1024 * 1024, ge=1024) # ClamAV (clamd). Empty host means no virus scan. When set, an unreachable scanner refuses uploads. clamav_host: str = "" clamav_port: int = Field(default=3310, ge=1, le=65535) clamav_timeout_seconds: float = Field(default=30.0, gt=0, le=600) # The storage alert: warn at this percent of the assets volume, or of ASSETS_QUOTA_BYTES when that is set. assets_quota_warn_pct: int = Field(default=85, ge=1, le=99) assets_quota_bytes: int = Field(default=0, ge=0) # Where the worker reads VISUAL and M-Files from. `fixture` reads synthetic files under FIXTURE_DIR (CI and tests). feed_source: FeedSource = "visualkit" drawing_source: DrawingSource = "mfiles" fixture_dir: Path = Path("/fixtures") @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] is None or not str(data[key]).strip()): data[key] = None for key in list(data): if key.lower() in ("stack", "feed_source", "drawing_source") and isinstance(data[key], str): data[key] = data[key].strip().lower() elif key.lower() == "clamav_host" and isinstance(data[key], str): data[key] = data[key].strip() 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 _fixture_sources(self) -> Settings: if self.stack == "prod" and (self.feed_source == "fixture" or self.drawing_source == "fixture"): raise ValueError("FEED_SOURCE=fixture and DRAWING_SOURCE=fixture are 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") def seal_key(self) -> bytes: """The HMAC key for filing seals, from SEAL_KEY_FILE.""" key = read_secret(self.seal_key_file, "SEAL_KEY_FILE").strip() if len(key) < 16: raise SecretFileError(f"SEAL_KEY_FILE file {self.seal_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()