shelfmark/shelfmark/config/migrations.py

"""Configuration migration helpers."""

from __future__ import annotations

import json
from pathlib import Path
from typing import TYPE_CHECKING, Any, Protocol

if TYPE_CHECKING:
    from collections.abc import Callable
    from os import PathLike

_DEPRECATED_SETTINGS_RESTRICTION_KEYS = (
    "PROXY_AUTH_RESTRICT_SETTINGS_TO_ADMIN",
    "CWA_RESTRICT_SETTINGS_TO_ADMIN",
    "RESTRICT_SETTINGS_TO_ADMIN",
)


class MigrationLogger(Protocol):
    """Logger surface used by config migration helpers."""

    def info(self, msg: str, *args: object) -> object: ...

    def debug(self, msg: str, *args: object) -> object: ...

    def exception(self, msg: str, *args: object) -> object: ...


def _as_bool(value: object) -> bool:
    if isinstance(value, bool):
        return value
    if isinstance(value, str):
        return value.strip().lower() in {"1", "true", "yes", "on"}
    return bool(value)


def _pick_legacy_settings_restriction(config: dict[str, Any]) -> bool | None:
    """Pick the best legacy admin-restriction value to migrate."""
    auth_method = str(config.get("AUTH_METHOD", "")).strip().lower()

    if auth_method == "proxy" and "PROXY_AUTH_RESTRICT_SETTINGS_TO_ADMIN" in config:
        return _as_bool(config.get("PROXY_AUTH_RESTRICT_SETTINGS_TO_ADMIN"))

    if auth_method == "cwa" and "CWA_RESTRICT_SETTINGS_TO_ADMIN" in config:
        return _as_bool(config.get("CWA_RESTRICT_SETTINGS_TO_ADMIN"))

    if "RESTRICT_SETTINGS_TO_ADMIN" in config:
        return _as_bool(config.get("RESTRICT_SETTINGS_TO_ADMIN"))

    if "PROXY_AUTH_RESTRICT_SETTINGS_TO_ADMIN" in config:
        return _as_bool(config.get("PROXY_AUTH_RESTRICT_SETTINGS_TO_ADMIN"))

    if "CWA_RESTRICT_SETTINGS_TO_ADMIN" in config:
        return _as_bool(config.get("CWA_RESTRICT_SETTINGS_TO_ADMIN"))

    return None


def migrate_security_settings(
    *,
    load_security_config: Callable[[], dict[str, Any]],
    load_users_config: Callable[[], dict[str, Any]],
    save_users_config: Callable[[dict[str, Any]], None],
    ensure_config_dir: Callable[[], None],
    get_config_path: Callable[[], str | PathLike[str]],
    sync_builtin_admin_user: Callable[[str, str], None],
    logger: MigrationLogger,
) -> None:
    """Migrate legacy security keys and sync builtin admin credentials."""
    try:
        config = load_security_config()
        users_config = load_users_config()
        migrated_security = False
        migrated_users = False

        if "USE_CWA_AUTH" in config:
            old_value = config.pop("USE_CWA_AUTH")
            if "AUTH_METHOD" not in config:
                if old_value:
                    config["AUTH_METHOD"] = "cwa"
                    logger.info("Migrated USE_CWA_AUTH=True to AUTH_METHOD='cwa'")
                elif config.get("BUILTIN_USERNAME") and config.get("BUILTIN_PASSWORD_HASH"):
                    config["AUTH_METHOD"] = "builtin"
                    logger.info("Migrated USE_CWA_AUTH=False to AUTH_METHOD='builtin'")
                else:
                    config["AUTH_METHOD"] = "none"
                    logger.info("Migrated USE_CWA_AUTH=False to AUTH_METHOD='none'")
                migrated_security = True
            else:
                logger.info("Removed deprecated USE_CWA_AUTH setting (AUTH_METHOD already exists)")
                migrated_security = True

        # Backfill AUTH_METHOD for configs that have builtin credentials but
        # were never migrated from USE_CWA_AUTH (e.g. dev builds that predated
        # the AUTH_METHOD field).
        if (
            "AUTH_METHOD" not in config
            and config.get("BUILTIN_USERNAME")
            and config.get("BUILTIN_PASSWORD_HASH")
        ):
            config["AUTH_METHOD"] = "builtin"
            migrated_security = True
            logger.info(
                "Backfilled AUTH_METHOD='builtin' from legacy "
                "BUILTIN_USERNAME/BUILTIN_PASSWORD_HASH credentials"
            )

        if "RESTRICT_SETTINGS_TO_ADMIN" not in users_config:
            legacy_restrict = _pick_legacy_settings_restriction(config)
            if legacy_restrict is not None:
                save_users_config({"RESTRICT_SETTINGS_TO_ADMIN": legacy_restrict})
                migrated_users = True
                logger.info(
                    "Migrated legacy settings-admin restriction to users.RESTRICT_SETTINGS_TO_ADMIN=%s",
                    legacy_restrict,
                )

        for deprecated_key in _DEPRECATED_SETTINGS_RESTRICTION_KEYS:
            if deprecated_key in config:
                config.pop(deprecated_key, None)
                migrated_security = True
                logger.info("Removed deprecated security setting: %s", deprecated_key)

        try:
            sync_builtin_admin_user(
                config.get("BUILTIN_USERNAME", ""),
                config.get("BUILTIN_PASSWORD_HASH", ""),
            )
        except Exception:
            logger.exception(
                "Failed to sync builtin credentials to users database during migration"
            )

        if migrated_security:
            ensure_config_dir()
            config_path = Path(get_config_path())
            with config_path.open("w") as f:
                json.dump(config, f, indent=2)
            logger.info("Security settings migration completed successfully")
        elif migrated_users:
            logger.info("Users settings migration completed successfully")
        else:
            logger.debug("No security settings migration needed")

    except FileNotFoundError:
        logger.debug("No existing security config file found - nothing to migrate")
    except Exception:
        logger.exception("Failed to migrate security settings")