feat(auth): implement secure login challenge and proof verification

astrbot/core/utils/auth_password.py

@@ -4,6 +4,7 @@ import hashlib
 import hmac
 import re
 import secrets
+from typing import Any
 
 _PBKDF2_ITERATIONS = 600_000
 _PBKDF2_SALT_BYTES = 16
@@ -65,6 +66,59 @@ def _is_pbkdf2_hash(stored: str) -> bool:
     return isinstance(stored, str) and stored.startswith(_PBKDF2_FORMAT)
 
 
+def get_dashboard_login_challenge(stored_hash: str) -> dict[str, Any]:
+    """Return the public challenge parameters needed for proof-based login."""
+    if _is_legacy_md5_hash(stored_hash):
+        return {"algorithm": "legacy_md5"}
+
+    if _is_pbkdf2_hash(stored_hash):
+        parts: list[str] = stored_hash.split("$")
+        if len(parts) != 4:
+            raise ValueError("Invalid dashboard password hash")
+        _, iterations_s, salt, _ = parts
+        return {
+            "algorithm": _PBKDF2_ALGORITHM,
+            "iterations": int(iterations_s),
+            "salt": salt,
+        }
+
+    raise ValueError("Unsupported dashboard password hash")
+
+
+def verify_dashboard_login_proof(
+    stored_hash: str, challenge_nonce: str, proof: str
+) -> bool:
+    """Verify an HMAC-SHA256 login proof generated from the stored password secret."""
+    if (
+        not isinstance(stored_hash, str)
+        or not isinstance(challenge_nonce, str)
+        or not isinstance(proof, str)
+    ):
+        return False
+
+    proof_key: bytes
+    if _is_legacy_md5_hash(stored_hash):
+        proof_key = stored_hash.lower().encode("utf-8")
+    elif _is_pbkdf2_hash(stored_hash):
+        parts: list[str] = stored_hash.split("$")
+        if len(parts) != 4:
+            return False
+        _, _, _, digest = parts
+        try:
+            proof_key = bytes.fromhex(digest)
+        except ValueError:
+            return False
+    else:
+        return False
+
+    expected = hmac.new(
+        proof_key,
+        challenge_nonce.encode("utf-8"),
+        hashlib.sha256,
+    ).hexdigest()
+    return hmac.compare_digest(expected.lower(), proof.lower())
+
+
 def verify_dashboard_password(stored_hash: str, candidate_password: str) -> bool:
     """Verify password against legacy md5 or new PBKDF2-SHA256 format."""
     if not isinstance(stored_hash, str) or not isinstance(candidate_password, str):
@@ -72,7 +126,7 @@ def verify_dashboard_password(stored_hash: str, candidate_password: str) -> bool
 
     if _is_legacy_md5_hash(stored_hash):
         # Keep compatibility with existing md5-based deployments:
-        # new clients send plain password, old clients may send md5 of it.
+        # challenge-based clients send proof, fallback clients may send plain password or md5.
         candidate_md5 = hashlib.md5(candidate_password.encode("utf-8")).hexdigest()
         return hmac.compare_digest(
             stored_hash.lower(), candidate_md5.lower()

astrbot/dashboard/routes/auth.py

@@ -1,5 +1,6 @@
 import asyncio
 import datetime
+import secrets
 
 import jwt
 from quart import request
@@ -7,10 +8,12 @@ from quart import request
 from astrbot import logger
 from astrbot.core import DEMO_MODE
 from astrbot.core.utils.auth_password import (
+    get_dashboard_login_challenge,
     hash_dashboard_password,
     is_default_dashboard_password,
     is_legacy_dashboard_password,
     validate_dashboard_password,
+    verify_dashboard_login_proof,
     verify_dashboard_password,
 )
 
@@ -20,12 +23,48 @@ from .route import Response, Route, RouteContext
 class AuthRoute(Route):
     def __init__(self, context: RouteContext) -> None:
         super().__init__(context)
+        self._login_challenges: dict[str, dict[str, object]] = {}
         self.routes = {
+            "/auth/login/challenge": ("POST", self.login_challenge),
             "/auth/login": ("POST", self.login),
             "/auth/account/edit": ("POST", self.edit_account),
         }
         self.register_routes()
 
+    async def login_challenge(self):
+        password = self.config["dashboard"]["password"]
+        self._prune_login_challenges()
+
+        try:
+            challenge = get_dashboard_login_challenge(password)
+        except ValueError as exc:
+            logger.error("Failed to create dashboard login challenge: %s", exc)
+            return (
+                Response()
+                .error("Unsupported dashboard password configuration")
+                .__dict__
+            )
+
+        challenge_id = secrets.token_hex(16)
+        nonce = secrets.token_hex(32)
+        self._login_challenges[challenge_id] = {
+            "nonce": nonce,
+            "expires_at": datetime.datetime.now(datetime.timezone.utc)
+            + datetime.timedelta(minutes=1),
+        }
+
+        return (
+            Response()
+            .ok(
+                {
+                    "challenge_id": challenge_id,
+                    "nonce": nonce,
+                    **challenge,
+                }
+            )
+            .__dict__
+        )
+
     async def login(self):
         username = self.config["dashboard"]["username"]
         password = self.config["dashboard"]["password"]
@@ -37,12 +76,33 @@ class AuthRoute(Route):
         req_password = (
             post_data.get("password") if isinstance(post_data, dict) else None
         )
-        if not isinstance(req_username, str) or not isinstance(req_password, str):
+        req_challenge_id = (
+            post_data.get("challenge_id") if isinstance(post_data, dict) else None
+        )
+        req_password_proof = (
+            post_data.get("password_proof") if isinstance(post_data, dict) else None
+        )
+        if not isinstance(req_username, str):
             return Response().error("Invalid request payload").__dict__
 
-        if req_username == username and verify_dashboard_password(
-            password, req_password
-        ):
+        login_verified = False
+        if isinstance(req_password, str):
+            login_verified = req_username == username and verify_dashboard_password(
+                password, req_password
+            )
+        elif isinstance(req_challenge_id, str) and isinstance(req_password_proof, str):
+            challenge_nonce = self._consume_login_challenge(req_challenge_id)
+            login_verified = (
+                req_username == username
+                and isinstance(challenge_nonce, str)
+                and verify_dashboard_login_proof(
+                    password, challenge_nonce, req_password_proof
+                )
+            )
+        else:
+            return Response().error("Invalid request payload").__dict__
+
+        if login_verified:
             change_pwd_hint = False
             legacy_pwd_hint = is_legacy_dashboard_password(password)
             if (
@@ -51,7 +111,9 @@ class AuthRoute(Route):
                 and not DEMO_MODE
             ):
                 change_pwd_hint = True
-                logger.warning("检测到默认管理员凭据，请尽快修改密码。")
+                logger.warning(
+                    "The dashboard is using the default password, please change it immediately to ensure security."
+                )
                 legacy_pwd_hint = True
 
             return (
@@ -66,8 +128,25 @@ class AuthRoute(Route):
                 )
                 .__dict__
             )
-        await asyncio.sleep(3)
-        return Response().error("用户名或密码错误").__dict__
+        return Response().error("User not found or incorrect password").__dict__
+
+    def _prune_login_challenges(self) -> None:
+        now = datetime.datetime.now(datetime.timezone.utc)
+        expired_ids = [
+            challenge_id
+            for challenge_id, challenge in self._login_challenges.items()
+            if challenge.get("expires_at") <= now
+        ]
+        for challenge_id in expired_ids:
+            self._login_challenges.pop(challenge_id, None)
+
+    def _consume_login_challenge(self, challenge_id: str) -> str | None:
+        self._prune_login_challenges()
+        challenge = self._login_challenges.pop(challenge_id, None)
+        if not isinstance(challenge, dict):
+            return None
+        nonce = challenge.get("nonce")
+        return nonce if isinstance(nonce, str) else None
 
     async def edit_account(self):
         if DEMO_MODE:
@@ -111,7 +190,7 @@ class AuthRoute(Route):
 
         self.config.save_config()
 
-        return Response().ok(None, "修改成功").__dict__
+        return Response().ok(None, "Updated account successfully").__dict__
 
     def generate_jwt(self, username):
         payload = {

dashboard/src/i18n/locales/en-US/features/auth.json

@@ -2,7 +2,7 @@
   "login": "Login",
   "username": "Username",
   "password": "Password",
-  "defaultHint": "The new AstrBot version has improved security. Please change your password.",
+  "defaultHint": "Default credentials: astrbot / astrbot",
   "logo": {
     "title": "AstrBot Dashboard",
     "subtitle": "Welcome"

dashboard/src/i18n/locales/ru-RU/features/auth.json

@@ -2,7 +2,7 @@
     "login": "Вход",
     "username": "Имя пользователя",
     "password": "Пароль",
-    "defaultHint": "Новая версия AstrBot улучшила безопасность. Пожалуйста, измените пароль.",
+    "defaultHint": "Учетные данные по умолчанию: astrbot / astrbot",
     "logo": {
         "title": "Панель управления AstrBot",
         "subtitle": "Добро пожаловать"

dashboard/src/stores/auth.ts

@@ -1,6 +1,7 @@
 import { defineStore } from 'pinia';
 import { router } from '@/router';
 import axios from 'axios';
+import { createLoginProof, type LoginChallenge } from '@/utils/authLoginProof';
 
 export const useAuthStore = defineStore({
   id: 'auth',
@@ -12,9 +13,17 @@ export const useAuthStore = defineStore({
   actions: {
     async login(username: string, password: string): Promise<void> {
       try {
+        const challengeRes = await axios.post('/api/auth/login/challenge');
+        const challenge = challengeRes.data?.data as LoginChallenge | undefined;
+        if (!challenge) {
+          return Promise.reject('Failed to initialize secure login');
+        }
+
+        const passwordProof = await createLoginProof(password, challenge);
         const res = await axios.post('/api/auth/login', {
           username: username,
-          password: password
+          challenge_id: challenge.challenge_id,
+          password_proof: passwordProof
         });
     
         if (res.data.status === 'error') {

dashboard/src/utils/authLoginProof.ts

@@ -0,0 +1,81 @@
+import md5 from 'js-md5';
+
+export type LoginChallenge = {
+  challenge_id: string;
+  nonce: string;
+  algorithm: 'pbkdf2_sha256' | 'legacy_md5';
+  iterations?: number;
+  salt?: string;
+};
+
+const encoder = new TextEncoder();
+
+function bytesToHex(buffer: ArrayBuffer): string {
+  return Array.from(new Uint8Array(buffer))
+    .map((byte) => byte.toString(16).padStart(2, '0'))
+    .join('');
+}
+
+function hexToBytes(hex: string): Uint8Array {
+  if (hex.length % 2 !== 0) {
+    throw new Error('Invalid hex payload');
+  }
+
+  const bytes = new Uint8Array(hex.length / 2);
+  for (let i = 0; i < hex.length; i += 2) {
+    bytes[i / 2] = Number.parseInt(hex.slice(i, i + 2), 16);
+  }
+  return bytes;
+}
+
+async function signNonce(secret: BufferSource, nonce: string): Promise<string> {
+  const subtle = globalThis.crypto?.subtle;
+  if (!subtle) {
+    throw new Error('Current browser does not support secure login');
+  }
+
+  const key = await subtle.importKey(
+    'raw',
+    secret,
+    { name: 'HMAC', hash: 'SHA-256' },
+    false,
+    ['sign']
+  );
+  const signature = await subtle.sign('HMAC', key, encoder.encode(nonce));
+  return bytesToHex(signature);
+}
+
+async function createPbkdf2Proof(password: string, challenge: LoginChallenge): Promise<string> {
+  const subtle = globalThis.crypto?.subtle;
+  if (!subtle || !challenge.salt || !challenge.iterations) {
+    throw new Error('Invalid secure login challenge');
+  }
+
+  const passwordKey = await subtle.importKey('raw', encoder.encode(password), 'PBKDF2', false, ['deriveBits']);
+  const derivedBits = await subtle.deriveBits(
+    {
+      name: 'PBKDF2',
+      hash: 'SHA-256',
+      salt: hexToBytes(challenge.salt),
+      iterations: challenge.iterations
+    },
+    passwordKey,
+    256
+  );
+
+  return signNonce(derivedBits, challenge.nonce);
+}
+
+async function createLegacyMd5Proof(password: string, challenge: LoginChallenge): Promise<string> {
+  return signNonce(encoder.encode(md5(password).toLowerCase()), challenge.nonce);
+}
+
+export async function createLoginProof(password: string, challenge: LoginChallenge): Promise<string> {
+  if (challenge.algorithm === 'pbkdf2_sha256') {
+    return createPbkdf2Proof(password, challenge);
+  }
+  if (challenge.algorithm === 'legacy_md5') {
+    return createLegacyMd5Proof(password, challenge);
+  }
+  throw new Error('Unsupported secure login algorithm');
+}