From 17b0dfe974cf1a6caa0cc6e4129d8da5be08b85a Mon Sep 17 00:00:00 2001 From: LIghtJUNction Date: Thu, 19 Mar 2026 19:46:21 +0800 Subject: [PATCH] fix(cli): restore cmd_conf password hashing and validators; use absolute import for check_astrbot_root --- astrbot/cli/commands/cmd_conf.py | 319 +++++++++++++++++++------------ 1 file changed, 194 insertions(+), 125 deletions(-) diff --git a/astrbot/cli/commands/cmd_conf.py b/astrbot/cli/commands/cmd_conf.py index 71bf1aa57..5153653d3 100644 --- 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 = "" + 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,