libervia-backend/libervia/backend/tools/config.py

#!/usr/bin/env python3

# Libervia XMPP Client
# Copyright (C) 2009-2025 Jérôme Poisson (goffi@goffi.org)
# Copyright (C) 2013-2016 Adrien Cossa (souliane@mailoo.org)

# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.

# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.

# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

"""Configuration related useful methods"""

from collections import defaultdict
from pathlib import Path
from socket import getfqdn
from typing import Annotated, Any, Literal, TypeVar, cast

from pydantic import (
    AfterValidator,
    BaseModel,
    ConfigDict,
    Field,
    ValidationError,
    create_model,
    field_validator,
)
from pydantic.types import PathType
from pydantic_settings import (
    BaseSettings,
    PydanticBaseSettingsSource,
    SettingsConfigDict,
    TomlConfigSettingsSource,
)

from libervia.backend.core import exceptions
from libervia.backend.core.constants import Const as C, xdg_data_home
from libervia.backend.core.i18n import _
from libervia.backend.core.log import getLogger
from libervia.backend.tools.common.files_utils import get_downloads_dir

log = getLogger(__name__)

expanduser_validator = AfterValidator(lambda p: p.expanduser().resolve())

cached_config: "Settings|None" = None
DynamicSettingsRegister = dict[str, "type[DynamicSettingsBase]"]
DynamicSettings: dict[str, DynamicSettingsRegister | type["DynamicSettingsBase"]] = (
    defaultdict(DynamicSettingsRegister)
)


def autocreatedir(p: Path) -> Path:
    """If directory doesn't exists, create it.

    Will also create parents, permission are set for user-only access.
    """
    p.mkdir(mode=0o700, parents=True, exist_ok=True)
    return p


autocreatedir_validator = AfterValidator(autocreatedir)


def _validate_executable_file(path: Path) -> Path:
    """Validate that a file is executable."""
    stat_result = path.stat()
    if not (stat_result.st_mode & 0o100):
        raise ValueError(f"File {path} is not executable.")
    return path


FilePath = Annotated[Path, expanduser_validator, PathType("file")]
ExecutableFilePath = Annotated[
    Path,
    expanduser_validator,
    PathType("file"),
    AfterValidator(_validate_executable_file),
]
DirectoryPathAuto = Annotated[
    Path, expanduser_validator, autocreatedir_validator, PathType("dir")
]
DirectoryPath = Annotated[Path, expanduser_validator, PathType("dir")]


class ConfigOptions(BaseModel):
    """Configuration options."""

    # We need to validate default to be sure that autocreatedir_validator is run.
    model_config = ConfigDict(extra="forbid", validate_default=True)


class DynamicSettingsBase(ConfigOptions):
    """Settings which are loaded at runtime."""

    _key: str = ""
    _sub_key: str | None = None

    @classmethod
    def _init_subclass(cls, /, key: str, plugin_info: dict | None = None) -> None:
        """Initialize the sublass, and register it in DynamicSettings.

        @param key: name of the field which will be used to store those settings.
        @param info: ``PLUGIN_INFO`` data of the plugin adding settings.
            Use None when the data is from a frontend and not a plugin.
        """
        cls._key = key
        if plugin_info is not None:
            sub_key = plugin_info.get(C.PI_CONFIG_KEY) or plugin_info[C.PI_IMPORT_NAME]
            assert sub_key
            # FIXME: We transform key here, but in the future plugins key name must be
            #   normalized on plugin import.
            sub_key = sub_key.replace("-", "_").lower()
            cls._sub_key = sub_key
            register = DynamicSettings[key]
            assert isinstance(register, dict)
            if sub_key in register:
                raise exceptions.ConflictError(
                    f"There is already a plugin registered for {key!r}.{sub_key!r}."
                )
            register[sub_key] = cls
        else:
            cls._sub_key = None
            DynamicSettings[key] = cls


class PluginSettings(DynamicSettingsBase):
    """Configuration options for plugins."""

    def __init_subclass__(cls, /, plugin_info: dict, **kwargs) -> None:
        super().__init_subclass__(**kwargs)
        super()._init_subclass(key="plugin", plugin_info=plugin_info)


class ComponentSettings(DynamicSettingsBase):
    """Configuration options for components."""

    def __init_subclass__(cls, /, plugin_info: dict, **kwargs) -> None:
        super().__init_subclass__(**kwargs)
        super()._init_subclass(key="component", plugin_info=plugin_info)


class FrontendSettings(DynamicSettingsBase):

    def __init_subclass__(cls, /, key: str, **kwargs) -> None:
        super().__init_subclass__(**kwargs)
        super()._init_subclass(key=key)


D = TypeVar("D", bound=DynamicSettingsBase)


class PbBridgePbSettings(ConfigOptions):
    type: Literal["pb"] = "pb"
    connection_type: Literal["unix_socket", "socket"] = Field(
        default="unix_socket", description="type of connection to use."
    )
    port: int = Field(
        default=8789, description='TCP port to use when ``connection_type`` is "socket"'
    )
    socket_dir: DirectoryPathAuto = Field(
        default=C.TEMP_DIR,
        description=(
            "Directory where the UNIX socket file will be created when using unix_socket"
            " connection type."
        ),
    )
    host: str = Field(
        default="localhost", description="Host to connect to when using socket connection"
    )


class DBusBridgeSettings(ConfigOptions):
    type: Literal["dbus"] = "dbus"
    int_prefix: str = Field(
        default="org.libervia.Libervia", description="Interface prefix"
    )


BridgeSettings = Annotated[
    DBusBridgeSettings | PbBridgePbSettings, Field(discriminator="type")
]

B = TypeVar("B", bound=BridgeSettings)


class EmailSettings(ConfigOptions):

    server: str = Field(default="localhost", description="Email server address")
    port: int = Field(default=25, description="Email server port")
    username: str | None = Field(default=None, description="Email server username")
    password: str | None = Field(default=None, description="Email server password")
    starttls: bool = Field(default=False, description="Use STARTTLS for email connection")
    auth: bool = Field(
        default=False, description="Enable authentication for email connection"
    )
    from_: str | None = Field(
        default="NOREPLY@example.net",
        description="Sender address for emails",
        alias="from",
    )
    sender_domain: str | None = Field(default=None, description="Email sender domain")


class TLSSettings(ConfigOptions):
    certificate: FilePath | None = Field(
        default=None, description="Path to TLS certificate file"
    )
    private_key: FilePath | None = Field(
        default=None, description="Path to TLS private key file"
    )
    chain: FilePath | None = Field(
        default=None,
        description="Path to TLS certificate chain file (intermediate certificates)",
    )


class TorSettings(ConfigOptions):
    """Tor configuration options."""

    enabled: bool = Field(default=False, description="Use Tor for connections")

    control_path: Path | None = Field(
        default=None, description="Tor control Unix Socket Path"
    )
    control_address: str = Field(default="127.0.0.1", description="Tor control address")
    control_port: int = Field(default=9051, description="Tor control port")


class CommonSettings(ConfigOptions):
    local_dir: DirectoryPathAuto = Field(
        default=xdg_data_home / C.APP_NAME_FILE,
        description="Local directory for Libervia data",
    )
    downloads_dir: DirectoryPathAuto = Field(
        default_factory=get_downloads_dir,
        description="Directory where files can be downloaded by default.",
    )
    log_dir: DirectoryPath = Field(
        default_factory=lambda data: data["local_dir"],
        description="Directory where log files will be written. Default to local_dir.",
    )
    log_level: str | None = Field(
        default=None, description="Logging level (debug, info, warning, error)"
    )
    log_fmt: str | None = Field(default=None, description="Log message format")
    bridges: list[BridgeSettings] = Field(
        default_factory=list,
        description=(
            "Bridge is the name of the IPC used to communicate between backend and"
            " frontends. Leave empty to let Libervia determine the best one. If you"
            " indicate several bridges, they will run in parallel."
        ),
    )
    pid_dir: DirectoryPathAuto = Field(
        default=C.TEMP_DIR,
        description=(
            "Directory where PID files will be stored. Default is a subdirectory of the"
            " system's temporary directory."
        ),
    )
    xmpp_domain: str = Field(
        default=getfqdn(),
        description=(
            "XMPP domain of the main server. Local hostname will be used by default."
        ),
    )
    new_account_domain: str | None = Field(
        default_factory=lambda data: data["xmpp_domain"],
        description=(
            "Domain to use when creating new account. Default to ``xmpp_domain``."
        ),
    )
    public_url: str | None = Field(default=None, description="Public URL for the server")
    # FIXME: we don't use JIDType here because config is also parsed by frontend, and
    #   JIDType uses Twisted's JID class.
    pubsub_service: str | None = Field(
        default=None,
        description=(
            "Main pubsub service to use. If not set, it will be discovered automatically."
        ),
    )
    admins_list: list[str] = Field(
        default_factory=list, description="List of profile names of admin users"
    )
    email_admins_list: list[str] = Field(
        default_factory=list, description="List of admin email addresses"
    )
    mutli_users: bool = Field(default=False, description="Enable multi-user support")
    pubsub_cache_strategy: Literal["no_cache"] | None = Field(
        default=None, description="PubSub cache strategy"
    )
    hosts_dict: dict[str, dict] = Field(
        default_factory=dict, description="Dictionary of host configurations"
    )
    ack_timeout: int = Field(
        default=35,
        description=(
            "Acknowledgement timeout in seconds. If the server doesn't acknowledge a"
            " stanza within this time, the connection will be aborted."
        ),
    )
    force_autodisconnect: bool = Field(
        default=False, description="Force automatic disconnection"
    )
    # FIXME: Seems not used anymore, delete?
    use_local_shared_tmp: bool = Field(
        default=False,
        description=(
            "If set, a temporary dir in cache will be used to share files between backend"
            " and frontends, useful when they are separated (e.g. when using containers)."
            " If unset, a temporary dir will be automatically created in os-relevant"
            " location."
        ),
    )
    allow_external_search: bool = Field(
        default=False, description="Allow external search functionality"
    )
    image_max: tuple[int, int] = Field(
        default=(1200, 720), description="Maximum dimensions for images (width, height)"
    )
    init_wait_for_service: str | None = Field(
        default=None,
        description=(
            "Wait for a network service to become available before starting the backend."
            " Format is 'host:port[:timeout][:service_name]' where host can be an"
            " IPv4/IPv6 address or hostname, port is the service port number, timeout is"
            " optional (default 60 seconds) and service_name is optional (defaults to"
            " 'service'). Example: 'localhost:5222:30:XmppServer'"
        ),
    )
    init_script: FilePath | None = Field(
        default=None,
        description="Path to an executable script to be run during backend startup.",
    )
    group_chat_search_default_jid: str = Field(
        default="api@search.jabber.network",
        description="JID of the entity used for group chat searches.",
    )

    @field_validator("bridges", mode="after")
    @classmethod
    def set_bridges_auto(cls, bridges: list[BridgeSettings]) -> list[BridgeSettings]:
        if not bridges:
            # FIXME: We default to DBus for now.
            bridges = [DBusBridgeSettings()]
        return bridges

    def get_bridge_conf(self, bridge_type: type[B]) -> B:
        """Get bridge configuration by type.

        @param type: The type of bridge to find (e.g. ``pb``, ``dbus``)
        @return: The configuration for the specified bridge.
        @raise exceptions.InternalError: If no bridge configuration is found for the given
            type.
        """
        for bridge_conf in self.bridges:
            if isinstance(bridge_conf, bridge_type):
                return bridge_conf
        # We raise internal error because there should not be any request for conf if the
        # bridge is not in "bridges" and thus instanciating.
        raise exceptions.InternalError(
            f"No bridge configuration found for type {type!r}."
        )


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        extra="allow",
        env_prefix="LIBERVIA_",
        env_nested_delimiter="__",
        toml_file=C.CONFIG_FILES,
    )

    common: CommonSettings = Field(default_factory=CommonSettings)
    email: EmailSettings = Field(default_factory=EmailSettings)
    tls: TLSSettings = Field(default_factory=TLSSettings)
    tor: TorSettings = Field(default_factory=TorSettings)
    application: dict[str, dict[str, str]] = Field(
        default_factory=dict,
        description="Application specific settings, used by app manager."
    )
    plugin: dict = Field(default_factory=dict, description="Plugin specific settings.")
    component: dict = Field(
        default_factory=dict, description="Component specific settings."
    )

    def get_settings(self, dyn_settings_cls: type[D]) -> D:
        """Get dynamic settings instance.

        @param dyn_settings_cls: The dynamtic settings class to retrieve.
        @return: The instance of the specified settings.
        @raise exceptions.InternalError: If no settings are found. This should not happen
            as the dynamtic classes are automatically registered.
        """
        try:
            settings = getattr(self, dyn_settings_cls._key)
            if dyn_settings_cls._sub_key is None:
                return settings
            else:
                return getattr(settings, dyn_settings_cls._sub_key)
        except AttributeError:
            raise exceptions.InternalError(
                f"No settings found for {dyn_settings_cls._sub_key!r}, they should"
                " be imported dynamically."
            )

    @classmethod
    def settings_customise_sources(
        cls,
        settings_cls: type[BaseSettings],
        init_settings: PydanticBaseSettingsSource,
        env_settings: PydanticBaseSettingsSource,
        dotenv_settings: PydanticBaseSettingsSource,
        file_secret_settings: PydanticBaseSettingsSource,
    ) -> tuple[PydanticBaseSettingsSource, ...]:
        return (
            init_settings,
            env_settings,
            dotenv_settings,
            file_secret_settings,
            TomlConfigSettingsSource(
                settings_cls,
                # FIXME: deep_merge is not yet available, will be for pydantic-settings
                #   2.13
                # deep_merge=True
            ),
        )


def parse_main_conf(log_filenames: bool = False, ignore_cache=False) -> Settings:
    """Look for main .toml configuration files, and parse them.

    @param log_filenames: if True, log filenames of read config files.
    @param ignore_cache: if True, settings will be reparsed even if they are already in
        cache. Notable useful when dynamic setting (e.g.; from plugins) have been added.
    """
    global cached_config

    if log_filenames:
        existing_conf_file = []
        for path_str in C.CONFIG_FILES:
            path = Path(path_str)
            if path.is_file() and path.exists():
                existing_conf_file.append(str(path))

        if existing_conf_file:
            log.info(
                _("Configuration will be read from: {filenames}.").format(
                    filenames=", ".join(existing_conf_file)
                )
            )
        else:
            log.info(_("No configuration file found."))

    if ignore_cache or cached_config is None:
        dynamic_settings_kw = {}
        for key, settings in DynamicSettings.items():
            if isinstance(settings, dict):
                NewSettingsModel = create_model(
                    f"{key.title()}Settings",
                    **cast(
                        dict,
                        {
                            sub_key: (settings_type, Field(default_factory=settings_type))
                            for sub_key, settings_type in settings.items()
                        },
                    ),
                    __config__=ConfigDict(extra="forbid"),
                )
                dynamic_settings_kw[key] = (
                    NewSettingsModel,
                    Field(default_factory=NewSettingsModel),
                )
            else:
                dynamic_settings_kw[key] = (
                    settings,
                    Field(default_factory=settings),
                )

        LiberviaSettings = create_model(
            "LiberviaSettings", __base__=Settings, **dynamic_settings_kw
        )

        try:
            cached_config = LiberviaSettings()
        except ValidationError as e:
            log.error(f"Settings are invalid, please fix your configuration: {e}")
            raise exceptions.ConfigError
        except:
            log.exception("Can't parse settings.")
            raise exceptions.InternalError

    return cached_config