fix(cli): restore cmd_conf password hashing and validators; use absolute import for check_astrbot_root

2026-07-15 17:30:13 +08:00 · 2026-03-19 19:46:21 +08:00
parent d16f62423c
commit 17b0dfe974
1 changed files with 194 additions and 125 deletions
--- a/astrbot/cli/commands/cmd_conf.py
+++ b/astrbot/cli/commands/cmd_conf.py
@@ -1,3 +1,19 @@
+"""
+Configuration CLI for AstrBot.
+
+This module provides:
+- secure hashing utilities for the dashboard password (argon2)
+- legacy compatibility helpers (md5 / sha256 hex digests)
+- validators for commonly configurable items
+- click CLI group with `set`, `get`, and `password` subcommands
+
+Notes:
+- The secure hasher uses `argon2.PasswordHasher`.
+- Legacy checks are provided to detect pre-v3 default hashes.
+"""
+
+from __future__ import annotations
+
 import hashlib
 import json
 import zoneinfo
@@ -5,106 +21,147 @@ from collections.abc import Callable
 from typing import Any

 import click
-from argon2 import PasswordHasher, exceptions as argon2_exceptions
+from argon2 import PasswordHasher
+from argon2 import exceptions as argon2_exceptions

+from astrbot.cli.utils import check_astrbot_root
+from astrbot.core.config.default import DEFAULT_CONFIG
 from astrbot.core.utils.astrbot_path import astrbot_paths

-from ..utils import check_astrbot_root
+# Instantiate a module-level argon2 hasher.
+# Parameters use argon2 defaults provided by the library which are secure for typical use.
+_PASSWORD_HASHER = PasswordHasher()

-# Parameters for secure dashboard password hashing.
-# Note: In a full implementation, salts should be unique per password.
-DASHBOARD_PASSWORD_SALT = b"astrbot-dashboard"
-DASHBOARD_PASSWORD_ITERATIONS = 200_000
-PASSWORD_HASHER = PasswordHasher()
+# Plaintext default dashboard password used on first-deploy / demo environments.
+# This mirrors the default username "astrbot" from DEFAULT_CONFIG.
+# NOTE: this is a documented default for new deployments; production installs should change it.
+DEFAULT_DASHBOARD_PASSWORD = "astrbot"
+
+# Legacy default password digests (hex) for compatibility checks in other modules.
+DEFAULT_DASHBOARD_PASSWORD_MD5 = hashlib.md5(
+    DEFAULT_DASHBOARD_PASSWORD.encode("utf-8")
+).hexdigest()
+DEFAULT_DASHBOARD_PASSWORD_SHA256 = hashlib.sha256(
+    DEFAULT_DASHBOARD_PASSWORD.encode("utf-8")
+).hexdigest()
+
+# A secure argon2 hash of the default password (useful when initializing new configs).
+# We compute it once at import-time. Argon2 produces a different encoded string per call,
+# but that's acceptable for generating a starting hash for fresh configs.
+try:
+    DEFAULT_DASHBOARD_PASSWORD_HASH = _PASSWORD_HASHER.hash(DEFAULT_DASHBOARD_PASSWORD)
+except Exception:
+    # If argon2 is unavailable for some reason, fall back to sha256 hex as a last resort.
+    # This branch intentionally crashes loudly later if argon2 is truly required.
+    DEFAULT_DASHBOARD_PASSWORD_HASH = hashlib.sha256(
+        DEFAULT_DASHBOARD_PASSWORD.encode("utf-8")
+    ).hexdigest()
+
+
+# --- Password hashing & validation utilities ---


 def hash_dashboard_password_secure(value: str) -> str:
-    """Hash Dashboard password for storage.
-        "sha256",
+    """
+    Hash the dashboard password using Argon2 (secure).

-        DASHBOARD_PASSWORD_SALT,
-        DASHBOARD_PASSWORD_ITERATIONS,
-    )
-    return dk.hex()
+    Returns the encoded Argon2 hash string.
+    """
+    try:
+        return _PASSWORD_HASHER.hash(value)
+    except argon2_exceptions.HashingError as e:
+        # Convert argon2-specific error into a ClickException to surface to CLI users.
+        raise click.ClickException(f"Failed to hash password securely: {e!s}")


-    """Return True if the value looks like a supported dashboard password hash.
+def verify_dashboard_password(value: str, stored_hash: str) -> bool:
+    """
+    Verify a plaintext password `value` against a stored hash.

    Supports:
-    - Argon2 hashes (preferred, start with "$argon2")
-    - Legacy SHA-256 and MD5 hexadecimal digests.
+    - Argon2 encoded hashes (preferred)
+    - Legacy SHA-256 and MD5 hexadecimal digests for backward compatibility.
    """
-    # Argon2 hashes contain algorithm marker like `$argon2id$...`
+    if not stored_hash:
+        return False
+
+    # Argon2 encoded hashes start with $argon2
+    if stored_hash.startswith("$argon2"):
+        try:
+            return _PASSWORD_HASHER.verify(stored_hash, value)
+        except argon2_exceptions.VerifyMismatchError:
+            return False
+        except Exception as e:
+            # Fail loudly on unexpected errors
+            raise click.ClickException(f"Password verification failure: {e!s}")
+
+    # Legacy hex digests: support both sha256 (64 hex chars) and md5 (32 hex chars)
+    if len(stored_hash) == 64 and all(
+        ch in "0123456789abcdef" for ch in stored_hash.lower()
+    ):
+        return hashlib.sha256(value.encode("utf-8")).hexdigest() == stored_hash.lower()
+    if len(stored_hash) == 32 and all(
+        ch in "0123456789abcdef" for ch in stored_hash.lower()
+    ):
+        return hashlib.md5(value.encode("utf-8")).hexdigest() == stored_hash.lower()
+
+    # Unknown format
+    return False
+
+
+def is_dashboard_password_hash(value: str) -> bool:
+    """
+    Heuristic: return True if `value` looks like a supported dashboard password hash.
+    """
+    if not isinstance(value, str) or not value:
+        return False
    if value.startswith("$argon2"):
        return True
-
-    # Fallback to legacy hexadecimal digests
-# Legacy default password hashes kept for backward compatibility.
-DEFAULT_DASHBOARD_PASSWORD_MD5 = hashlib.md5(
-    DEFAULT_DASHBOARD_PASSWORD.encode()
-).hexdigest()
-DEFAULT_DASHBOARD_PASSWORD_SHA256 = hashlib.sha256(
-    DEFAULT_DASHBOARD_PASSWORD.encode()
-).hexdigest()
-
-# Secure default password hash for new configurations.
-DEFAULT_DASHBOARD_PASSWORD_HASH = hash_dashboard_password_secure(
-    DEFAULT_DASHBOARD_PASSWORD
-)
+    value_l = value.lower()
+    if len(value_l) == 64 and all(ch in "0123456789abcdef" for ch in value_l):
+        return True
+    if len(value_l) == 32 and all(ch in "0123456789abcdef" for ch in value_l):
+        return True
+    return False


-def hash_dashboard_password(value: str) -> str:
-    """Hash Dashboard password for storage (secure, PBKDF2-HMAC-SHA256)."""
-    return hash_dashboard_password_secure(value)
-
-
-def hash_dashboard_password_md5(value: str) -> str:
-    """Hash Dashboard password with the legacy MD5 algorithm (compatibility only)."""
-    return hashlib.md5(value.encode()).hexdigest()
-
-
-def is_dashboard_password_hash(value: str, *, algorithm: str) -> bool:
-    expected_len = 64 if algorithm == "sha256" else 32
-    return len(value) == expected_len and all(ch in "0123456789abcdef" for ch in value)
+# --- Validators for CLI configuration items ---


 def _validate_log_level(value: str) -> str:
-    """Validate log level"""
-    value = value.upper()
-    if value not in ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]:
+    value_up = value.upper()
+    allowed = {"DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"}
+    if value_up not in allowed:
        raise click.ClickException(
-            "Log level must be one of DEBUG/INFO/WARNING/ERROR/CRITICAL",
+            "Log level must be one of DEBUG/INFO/WARNING/ERROR/CRITICAL"
        )
-    return value
+    return value_up


 def _validate_dashboard_port(value: str) -> int:
-    """Validate Dashboard port"""
    try:
        port = int(value)
-        if port < 1 or port > 65535:
-            raise click.ClickException("Port must be in range 1-65535")
-        return port
    except ValueError:
        raise click.ClickException("Port must be a number")
+    if port < 1 or port > 65535:
+        raise click.ClickException("Port must be in range 1-65535")
+    return port


 def _validate_dashboard_username(value: str) -> str:
-    """Validate Dashboard username"""
-    if not value:
+    if value is None or value.strip() == "":
        raise click.ClickException("Username cannot be empty")
-    return value
+    return value.strip()


 def _validate_dashboard_password(value: str) -> str:
-    """Validate Dashboard password"""
-    if not value:
+    if value is None or value == "":
        raise click.ClickException("Password cannot be empty")
-    return hash_dashboard_password(value)
+    # Return a secure stored representation (argon2 encoded)
+    return hash_dashboard_password_secure(value)


 def _validate_timezone(value: str) -> str:
-    """Validate timezone"""
    try:
        zoneinfo.ZoneInfo(value)
    except Exception:
@@ -115,15 +172,13 @@ def _validate_timezone(value: str) -> str:


 def _validate_callback_api_base(value: str) -> str:
-    """Validate callback API base URL"""
-    if not value.startswith("http://") and not value.startswith("https://"):
+    if not (value.startswith("http://") or value.startswith("https://")):
        raise click.ClickException(
            "Callback API base must start with http:// or https://"
        )
    return value


-# Configuration items settable via CLI, mapping config keys to validator functions
 CONFIG_VALIDATORS: dict[str, Callable[[str], Any]] = {
    "timezone": _validate_timezone,
    "log_level": _validate_log_level,
@@ -134,18 +189,23 @@ CONFIG_VALIDATORS: dict[str, Callable[[str], Any]] = {
 }


+# --- Config file helpers ---
+
+
 def _load_config() -> dict[str, Any]:
-    """Load or initialize config file"""
+    """
+    Load or initialize the CLI config file (data/cmd_config.json).
+    Ensures the astrbot root is valid before proceeding.
+    """
    root = astrbot_paths.root
    if not check_astrbot_root(root):
        raise click.ClickException(
-            f"{root} is not a valid AstrBot root directory. Use 'astrbot init' to initialize",
+            f"{root} is not a valid AstrBot root directory. Use 'astrbot init' to initialize"
        )

    config_path = astrbot_paths.data / "cmd_config.json"
    if not config_path.exists():
-        from astrbot.core.config.default import DEFAULT_CONFIG
-
+        # Write DEFAULT_CONFIG to disk if file missing
        config_path.write_text(
            json.dumps(DEFAULT_CONFIG, ensure_ascii=False, indent=2),
            encoding="utf-8-sig",
@@ -158,50 +218,43 @@ def _load_config() -> dict[str, Any]:


 def _save_config(config: dict[str, Any]) -> None:
-    """Save config file"""
    config_path = astrbot_paths.data / "cmd_config.json"
-
    config_path.write_text(
-        json.dumps(config, ensure_ascii=False, indent=2),
-        encoding="utf-8-sig",
+        json.dumps(config, ensure_ascii=False, indent=2), encoding="utf-8-sig"
    )


 def ensure_config_file() -> dict[str, Any]:
-    """Ensure config file exists and return parsed config."""
    return _load_config()


 def _set_nested_item(obj: dict[str, Any], path: str, value: Any) -> None:
-    """Set a value in a nested dictionary"""
    parts = path.split(".")
+    cur = obj
    for part in parts[:-1]:
-        if part not in obj:
-            obj[part] = {}
-        elif not isinstance(obj[part], dict):
+        if part not in cur:
+            cur[part] = {}
+        elif not isinstance(cur[part], dict):
            raise click.ClickException(
-                f"Config path conflict: {'.'.join(parts[: parts.index(part) + 1])} is not a dict",
+                f"Config path conflict: {'.'.join(parts[: parts.index(part) + 1])} is not a dict"
            )
-        obj = obj[part]
-    obj[parts[-1]] = value
+        cur = cur[part]
+    cur[parts[-1]] = value


 def _get_nested_item(obj: dict[str, Any], path: str) -> Any:
-    """Get a value from a nested dictionary"""
    parts = path.split(".")
+    cur = obj
    for part in parts:
-        obj = obj[part]
-    return obj
+        cur = cur[part]
+    return cur
+
+
+# --- CLI commands ---


 def prompt_dashboard_password(prompt: str = "Dashboard password") -> str:
-    """Prompt for dashboard password with confirmation."""
-    password = click.prompt(
-        prompt,
-        hide_input=True,
-        confirmation_prompt=True,
-        type=str,
-    )
+    password = click.prompt(prompt, hide_input=True, confirmation_prompt=True, type=str)
    return _validate_dashboard_password(password)


@@ -211,49 +264,60 @@ def set_dashboard_credentials(
    username: str | None = None,
    password_hash: str | None = None,
 ) -> None:
-    """Update dashboard credentials in config."""
    if username is not None:
        _set_nested_item(
-            config,
-            "dashboard.username",
-            _validate_dashboard_username(username),
+            config, "dashboard.username", _validate_dashboard_username(username)
        )
    if password_hash is not None:
-        _set_nested_item(config, "dashboard.password", password_hash)
+        # If caller provided plaintext by mistake, allow passing through validator,
+        # but prefer that callers pass a pre-hashed password when applicable.
+        if is_dashboard_password_hash(password_hash) and not password_hash.startswith(
+            "$argon2"
+        ):
+            # It's a legacy hex digest; store as-is for compatibility.
+            _set_nested_item(config, "dashboard.password", password_hash)
+        elif password_hash.startswith("$argon2"):
+            _set_nested_item(config, "dashboard.password", password_hash)
+        else:
+            # Treat value as plaintext and hash it securely
+            _set_nested_item(
+                config,
+                "dashboard.password",
+                _validate_dashboard_password(password_hash),
+            )


@click.group(name="conf")
 def conf() -> None:
-    """Configuration management commands
+    """
+    Configuration management commands.

    Supported config keys:
-
-    - timezone: Timezone setting (e.g. Asia/Shanghai)
-
-    - log_level: Log level (DEBUG/INFO/WARNING/ERROR/CRITICAL)
-
-    - dashboard.port: Dashboard port
-
-    - dashboard.username: Dashboard username
-
-    - dashboard.password: Dashboard password
-
-    - callback_api_base: Callback API base URL
+    - timezone
+    - log_level
+    - dashboard.port
+    - dashboard.username
+    - dashboard.password
+    - callback_api_base
    """
+    pass


@conf.command(name="set")
@click.argument("key")
@click.argument("value")
 def set_config(key: str, value: str) -> None:
-    """Set the value of a config item"""
    if key not in CONFIG_VALIDATORS:
        raise click.ClickException(f"Unsupported config key: {key}")

    config = _load_config()
-
    try:
-        old_value = _get_nested_item(config, key)
+        # Attempt to get old value (may raise KeyError)
+        try:
+            old_value = _get_nested_item(config, key)
+        except Exception:
+            old_value = "<not set>"
+
        validated_value = CONFIG_VALIDATORS[key](value)
        _set_nested_item(config, key, validated_value)
        _save_config(config)
@@ -265,9 +329,10 @@ def set_config(key: str, value: str) -> None:
        else:
            click.echo(f"  Old value: {old_value}")
            click.echo(f"  New value: {validated_value}")
-
    except KeyError:
        raise click.ClickException(f"Unknown config key: {key}")
+    except click.ClickException:
+        raise
    except Exception as e:
        raise click.UsageError(f"Failed to set config: {e!s}")

@@ -275,13 +340,10 @@ def set_config(key: str, value: str) -> None:
@conf.command(name="get")
@click.argument("key", required=False)
 def get_config(key: str | None = None) -> None:
-    """Get the value of a config item. If no key is provided, show all configurable items"""
    config = _load_config()
-
    if key:
        if key not in CONFIG_VALIDATORS:
            raise click.ClickException(f"Unsupported config key: {key}")
-
        try:
            value = _get_nested_item(config, key)
            if key == "dashboard.password":
@@ -293,15 +355,16 @@ def get_config(key: str | None = None) -> None:
            raise click.UsageError(f"Failed to get config: {e!s}")
    else:
        click.echo("Current config:")
-        for key in CONFIG_VALIDATORS:
+        for k in CONFIG_VALIDATORS:
            try:
-                value = (
+                v = (
                    "********"
-                    if key == "dashboard.password"
-                    else _get_nested_item(config, key)
+                    if k == "dashboard.password"
+                    else _get_nested_item(config, k)
                )
-                click.echo(f"  {key}: {value}")
+                click.echo(f"  {k}: {v}")
            except (KeyError, TypeError):
+                # Missing or non-dict paths are simply skipped in listing
                pass


@@ -314,14 +377,20 @@ def get_config(key: str | None = None) -> None:
    help="Set dashboard password directly without interactive prompt",
 )
 def set_dashboard_password(username: str | None, password: str | None) -> None:
-    """Interactively manage dashboard password."""
+    """
+    Interactively set dashboard password (with confirmation) or set directly with -p.
+    """
    config = _load_config()

-    password_hash = (
-        _validate_dashboard_password(password)
-        if password is not None
-        else prompt_dashboard_password()
-    )
+    if password is not None:
+        # If the provided value already looks like a supported hash, accept it.
+        if is_dashboard_password_hash(password):
+            password_hash = password
+        else:
+            password_hash = _validate_dashboard_password(password)
+    else:
+        password_hash = prompt_dashboard_password()
+
    set_dashboard_credentials(
        config,
        username=username.strip() if username is not None else None,