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

#!/usr/bin/env python3

# Libervia XMPP Client
# Copyright (C) 2009-2026 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 tomllib import TOMLDecodeError
from typing import Annotated, Literal, Self, TypeVar, cast

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.models.types import JIDType
from libervia.backend.tools.common.files_utils import get_downloads_dir
from pydantic import (
    AfterValidator,
    BaseModel,
    ConfigDict,
    Field,
    ValidationError,
    create_model,
    field_validator,
    model_validator,
)
from pydantic.types import PathType
from pydantic_settings import (
    BaseSettings,
    PydanticBaseSettingsSource,
    SettingsConfigDict,
    TomlConfigSettingsSource,
)

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
    _import_name: 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
            cls._import_name = plugin_info.get(C.PI_IMPORT_NAME)
        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."""

    jid: JIDType | None = Field(
        default=None,
        description=(
            "JID of the component. If used, the component profile will be created"
            " automatically (with the component entry point used as name), and"
            " auto-connect on backend startup."
        ),
    )
    secret: str | None = Field(
        default=None,
        description=(
            "Shared secret between the component and the XMPP server. To be used with"
            " ``jid`` field for automatic configuration and connetion."
        ),
    )
    autoconnect: bool = Field(
        default=True,
        description=(
            "Whether to automatically connect this component on backend startup. Used"
            " when ``jid`` and ``secret`` are set."
        ),
    )

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

    @model_validator(mode="after")
    def jid_and_secret_together(self) -> Self:
        """Check that "jid" and "secret" are (un)set at the same time"""
        if (self.jid is None) ^ (self.secret is None):
            raise ValueError('You must set "jid" and "secret" together.')
        return self


class FrontendSettings(DynamicSettingsBase):

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


class DynamicSettingsCollection(BaseModel):
    """Base class for collection of DynamicSettings."""


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


class PbBridgeSettings(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.RUNTIME_DIR,
        description=(
            "Directory where the UNIX socket file will be created when using unix_socket"
            " connection type. Default is the user's runtime directory."
        ),
    )
    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"
    )

    @model_validator(mode="after")
    def force_isolated_int_prefix(self) -> Self:
        """Append isolation name to D-Bus service prefix to avoid conflicts."""
        if C.isolation_name is not None:
            # The isolation name is already restricted, so we only need to handle D-Bus
            # specific rules.
            sanitized = C.isolation_name.replace("-", "_")
            if sanitized[0].isdigit():
                sanitized = f"_{sanitized}"
            self.int_prefix = f"{self.int_prefix}.isolated.{sanitized}"
        return self


class AutoBridgeSettings(ConfigOptions):
    """Automatic Bridge.

    Will try D-Bus first, then fall back to Pb if necessary.
    """

    type: Literal["auto"] = "auto"


class EmbeddedBridgeSettings(ConfigOptions):
    """Embedded Bridge.

    Runs frontend and backend in the same process, backend is used like a library.
    The backend is started automatically on first bridge connection.
    """

    type: Literal["embedded"] = "embedded"


BridgeSettings = Annotated[
    DBusBridgeSettings | PbBridgeSettings | AutoBridgeSettings | EmbeddedBridgeSettings,
    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")
    pid_dir: DirectoryPathAuto = Field(
        default=C.RUNTIME_DIR,
        description=(
            "Directory where PID files will be stored. Default is the user's runtime"
            " 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")
    # FIXME: cache is temporarily disabled by default (it's broken)
    pubsub_cache_strategy: Literal["no_cache"] | None = Field(
        default="no_cache", description="PubSub cache strategy"
    )
    hosts: dict[str, dict | str] = Field(
        default_factory=dict,
        description=(
            "Mapping of XMPP domains to their connection host (hostname or IP) and port. "
            " Allows overriding the connection target for specific domains. Each entry"
            " can be either a string (host only) or a dictionary with 'host' and 'port'"
            " keys. Example: {'example.com': 'backend.example.com'} or {'example.com':"
            " {'host': '192.168.1.100', 'port': 5222}}"
        ),
    )
    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.",
    )

    @model_validator(mode="after")
    def force_isolated_local_dir(self) -> Self:
        # In isolated mode, we force local_dir to avoid issues with parent configs.
        if C.ISOLATED_LOCAL_DIR is not None:
            self.local_dir = C.ISOLATED_LOCAL_DIR
        return self


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)
    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."
        ),
    )
    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: DynamicSettingsCollection = Field(
        default_factory=DynamicSettingsCollection, description="Plugin specific settings."
    )
    component: DynamicSettingsCollection = Field(
        default_factory=DynamicSettingsCollection,
        description="Component specific settings.",
    )

    @field_validator("bridges", mode="after")
    @classmethod
    def set_bridges_auto(cls, bridges: list[BridgeSettings]) -> list[BridgeSettings]:
        if not bridges:
            bridges = [AutoBridgeSettings()]
        elif any(isinstance(b, AutoBridgeSettings) for b in bridges) and any(
            isinstance(b, (DBusBridgeSettings, PbBridgeSettings)) for b in bridges
        ):
            raise ValueError(
                '"auto" bridge is mutally exclusive with "dbus" and "pb" bridges.'
            )
        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 {bridge_type}."
        )

    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()}SettingsCollection",
                    __base__=DynamicSettingsCollection,
                    **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:
            raise exceptions.ConfigError(e)
        except TOMLDecodeError as e:
            log.error(f"You configuration file is invalid: {e}")
            raise exceptions.ConfigError from e
        except Exception as e:
            log.exception("Can't parse settings.")
            raise exceptions.InternalError from e

    return cached_config