refactor: remove passkey login

routes/admin_api/core.py

@@ -2,7 +2,6 @@
 # -*- coding: utf-8 -*-
 from __future__ import annotations
 
-import json
 import os
 import stat
 import tempfile
@@ -17,19 +16,6 @@ from routes.admin_api import admin_api_bp
 from routes.decorators import admin_required
 from services.accounts_service import load_user_accounts
 from services.checkpoints import get_checkpoint_mgr
-from services.passkeys import (
-    MAX_PASSKEYS_PER_OWNER,
-    encode_credential_id,
-    get_credential_transports,
-    get_expected_origins,
-    get_rp_id,
-    is_challenge_valid,
-    make_authentication_options,
-    make_registration_options,
-    normalize_device_name,
-    verify_authentication,
-    verify_registration,
-)
 from services.state import (
     safe_get_user_accounts_snapshot,
     safe_verify_and_consume_captcha,
@@ -46,8 +32,6 @@ from services.tasks import submit_account_task
 
 logger = get_logger("app")
 config = get_config()
-_ADMIN_PASSKEY_LOGIN_SESSION_KEY = "admin_passkey_login_state"
-_ADMIN_PASSKEY_REGISTER_SESSION_KEY = "admin_passkey_register_state"
 
 
 def _admin_reauth_required() -> bool:
@@ -63,26 +47,6 @@ def _require_admin_reauth():
     return None
 
 
-def _parse_credential_payload(data: dict) -> dict | None:
-    credential = data.get("credential")
-    if isinstance(credential, dict):
-        return credential
-    if isinstance(credential, str):
-        try:
-            parsed = json.loads(credential)
-            return parsed if isinstance(parsed, dict) else None
-        except Exception:
-            return None
-    return None
-
-
-def _truncate_text(value, max_len: int = 300) -> str:
-    text = str(value or "").strip()
-    if len(text) > max_len:
-        return f"{text[:max_len]}..."
-    return text
-
-
 @admin_api_bp.route("/debug-config", methods=["GET"])
 @admin_required
 def debug_config():
@@ -107,169 +71,6 @@ def debug_config():
     )
 
 
-@admin_api_bp.route("/passkeys/login/options", methods=["POST"])
-@require_ip_not_locked
-def admin_passkey_login_options():
-    """管理员 Passkey 登录：获取 assertion challenge。"""
-    data = request.get_json(silent=True) or {}
-    username = str(data.get("username", "") or "").strip()
-
-    client_ip = get_rate_limit_ip()
-    mode = "named" if username else "discoverable"
-    username_key = f"admin-passkey:{username}" if username else "admin-passkey:discoverable"
-
-    is_locked, remaining = check_login_ip_user_locked(client_ip, username_key)
-    if is_locked:
-        wait_hint = f"{remaining // 60 + 1}分钟" if remaining >= 60 else f"{remaining}秒"
-        return jsonify({"error": f"账号短时锁定，请{wait_hint}后再试"}), 429
-
-    allowed, error_msg = check_ip_request_rate(client_ip, "login")
-    if not allowed:
-        return jsonify({"error": error_msg}), 429
-
-    allowed, error_msg = check_login_rate_limits(client_ip, username_key)
-    if not allowed:
-        return jsonify({"error": error_msg}), 429
-
-    admin_id = 0
-    allow_credential_ids = []
-    if mode == "named":
-        admin_row = database.get_admin_by_username(username)
-        if not admin_row:
-            record_login_failure(client_ip, username_key)
-            return jsonify({"error": "账号或Passkey不可用"}), 400
-
-        admin_id = int(admin_row["id"])
-        passkeys = database.list_passkeys("admin", admin_id)
-        if not passkeys:
-            record_login_failure(client_ip, username_key)
-            return jsonify({"error": "该管理员尚未绑定Passkey"}), 400
-        allow_credential_ids = [str(item.get("credential_id") or "").strip() for item in passkeys if item.get("credential_id")]
-
-    try:
-        rp_id = get_rp_id(request)
-        expected_origins = get_expected_origins(request)
-    except Exception as e:
-        logger.warning(f"[passkey] 管理员登录 options 失败(mode={mode}, username={username or '-'}) : {e}")
-        return jsonify({"error": "Passkey配置异常，请联系管理员"}), 500
-
-    options = make_authentication_options(rp_id=rp_id, allow_credential_ids=allow_credential_ids)
-    challenge = str(options.get("challenge") or "").strip()
-    if not challenge:
-        return jsonify({"error": "生成Passkey挑战失败"}), 500
-
-    session[_ADMIN_PASSKEY_LOGIN_SESSION_KEY] = {
-        "mode": mode,
-        "username": username,
-        "admin_id": int(admin_id),
-        "challenge": challenge,
-        "rp_id": rp_id,
-        "expected_origins": expected_origins,
-        "username_key": username_key,
-        "created_at": time.time(),
-    }
-    session.modified = True
-    return jsonify({"publicKey": options})
-
-
-@admin_api_bp.route("/passkeys/login/verify", methods=["POST"])
-@require_ip_not_locked
-def admin_passkey_login_verify():
-    """管理员 Passkey 登录：校验 assertion 并登录。"""
-    data = request.get_json(silent=True) or {}
-    request_username = str(data.get("username", "") or "").strip()
-    credential = _parse_credential_payload(data)
-    if not credential:
-        return jsonify({"error": "Passkey参数缺失"}), 400
-
-    state = session.get(_ADMIN_PASSKEY_LOGIN_SESSION_KEY) or {}
-    if not state:
-        return jsonify({"error": "Passkey挑战不存在或已过期，请重试"}), 400
-    if not is_challenge_valid(state.get("created_at")):
-        session.pop(_ADMIN_PASSKEY_LOGIN_SESSION_KEY, None)
-        return jsonify({"error": "Passkey挑战已过期，请重试"}), 400
-
-    mode = str(state.get("mode") or "named")
-    if mode not in {"named", "discoverable"}:
-        session.pop(_ADMIN_PASSKEY_LOGIN_SESSION_KEY, None)
-        return jsonify({"error": "Passkey状态异常，请重试"}), 400
-
-    expected_username = str(state.get("username") or "").strip()
-    username = expected_username
-    if mode == "named":
-        if not expected_username:
-            session.pop(_ADMIN_PASSKEY_LOGIN_SESSION_KEY, None)
-            return jsonify({"error": "Passkey状态异常，请重试"}), 400
-        if request_username and request_username != expected_username:
-            return jsonify({"error": "用户名与挑战不匹配，请重试"}), 400
-    else:
-        username = request_username
-
-    client_ip = get_rate_limit_ip()
-    username_key = str(state.get("username_key") or "").strip() or (
-        f"admin-passkey:{expected_username}" if mode == "named" else "admin-passkey:discoverable"
-    )
-
-    is_locked, remaining = check_login_ip_user_locked(client_ip, username_key)
-    if is_locked:
-        wait_hint = f"{remaining // 60 + 1}分钟" if remaining >= 60 else f"{remaining}秒"
-        return jsonify({"error": f"账号短时锁定，请{wait_hint}后再试"}), 429
-
-    credential_id = str(credential.get("id") or credential.get("rawId") or "").strip()
-    if not credential_id:
-        return jsonify({"error": "Passkey参数无效"}), 400
-
-    passkey = database.get_passkey_by_credential_id(credential_id)
-    if not passkey:
-        record_login_failure(client_ip, username_key)
-        return jsonify({"error": "Passkey不存在或已删除"}), 401
-    if str(passkey.get("owner_type") or "") != "admin":
-        record_login_failure(client_ip, username_key)
-        return jsonify({"error": "Passkey不属于管理员账号"}), 401
-    if mode == "named" and int(passkey.get("owner_id") or 0) != int(state.get("admin_id") or 0):
-        record_login_failure(client_ip, username_key)
-        return jsonify({"error": "Passkey与管理员账号不匹配"}), 401
-
-    try:
-        _, verified = verify_authentication(
-            credential=credential,
-            expected_challenge=str(state.get("challenge") or ""),
-            expected_rp_id=str(state.get("rp_id") or ""),
-            expected_origins=list(state.get("expected_origins") or []),
-            credential_public_key=str(passkey.get("public_key") or ""),
-            credential_current_sign_count=int(passkey.get("sign_count") or 0),
-        )
-        verified_credential_id = encode_credential_id(verified.credential_id)
-        if verified_credential_id != str(passkey.get("credential_id") or ""):
-            raise ValueError("credential_id mismatch")
-    except Exception as e:
-        logger.warning(f"[passkey] 管理员登录验签失败(mode={mode}, username={expected_username or request_username or '-'}) : {e}")
-        record_login_failure(client_ip, username_key)
-        return jsonify({"error": "Passkey验证失败"}), 401
-
-    admin_id = int(passkey.get("owner_id") or 0)
-    admin_row = database.get_admin_by_id(admin_id)
-    if not admin_row:
-        return jsonify({"error": "管理员账号不存在"}), 401
-    admin_username = str(admin_row.get("username") or "").strip() or username or f"admin-{admin_id}"
-
-    database.update_passkey_usage(int(passkey["id"]), int(verified.new_sign_count))
-    clear_login_failures(client_ip, username_key)
-    admin_login_key = f"admin-passkey:{admin_username}"
-    if admin_login_key and admin_login_key != username_key:
-        clear_login_failures(client_ip, admin_login_key)
-
-    session.pop(_ADMIN_PASSKEY_LOGIN_SESSION_KEY, None)
-    session.pop("admin_id", None)
-    session.pop("admin_username", None)
-    session["admin_id"] = admin_id
-    session["admin_username"] = admin_username
-    session["admin_reauth_until"] = time.time() + int(config.ADMIN_REAUTH_WINDOW_SECONDS)
-    session.permanent = True
-    session.modified = True
-    return jsonify({"success": True, "redirect": "/yuyx/admin", "username": admin_username})
-
-
 @admin_api_bp.route("/login", methods=["POST"])
 @require_ip_not_locked
 def admin_login():
@@ -358,164 +159,6 @@ def admin_logout():
     return jsonify({"success": True})
 
 
-@admin_api_bp.route("/admin/passkeys", methods=["GET"])
-@admin_required
-def list_admin_passkeys():
-    admin_id = int(session.get("admin_id") or 0)
-    rows = database.list_passkeys("admin", admin_id)
-    items = []
-    for row in rows:
-        credential_id = str(row.get("credential_id") or "")
-        preview = f"{credential_id[:8]}...{credential_id[-6:]}" if len(credential_id) > 16 else credential_id
-        items.append(
-            {
-                "id": int(row.get("id")),
-                "device_name": str(row.get("device_name") or ""),
-                "credential_id_preview": preview,
-                "created_at": row.get("created_at"),
-                "last_used_at": row.get("last_used_at"),
-                "transports": str(row.get("transports") or ""),
-            }
-        )
-    return jsonify({"items": items, "limit": MAX_PASSKEYS_PER_OWNER})
-
-
-@admin_api_bp.route("/admin/passkeys/register/options", methods=["POST"])
-@admin_required
-def admin_passkey_register_options():
-    admin_id = int(session.get("admin_id") or 0)
-    admin_username = str(session.get("admin_username") or "").strip() or f"admin-{admin_id}"
-
-    count = database.count_passkeys("admin", admin_id)
-    if count >= MAX_PASSKEYS_PER_OWNER:
-        return jsonify({"error": f"最多可绑定{MAX_PASSKEYS_PER_OWNER}台设备"}), 400
-
-    data = request.get_json(silent=True) or {}
-    device_name = normalize_device_name(data.get("device_name"))
-
-    existing = database.list_passkeys("admin", admin_id)
-    exclude_credential_ids = [str(item.get("credential_id") or "").strip() for item in existing if item.get("credential_id")]
-
-    try:
-        rp_id = get_rp_id(request)
-        expected_origins = get_expected_origins(request)
-        options = make_registration_options(
-            rp_id=rp_id,
-            rp_name="知识管理平台",
-            user_name=admin_username,
-            user_display_name=admin_username,
-            user_id_bytes=f"admin:{admin_id}".encode("utf-8"),
-            exclude_credential_ids=exclude_credential_ids,
-        )
-    except Exception as e:
-        logger.warning(f"[passkey] 管理员注册 options 失败(admin_id={admin_id}): {e}")
-        return jsonify({"error": "生成Passkey挑战失败"}), 500
-
-    challenge = str(options.get("challenge") or "").strip()
-    if not challenge:
-        return jsonify({"error": "生成Passkey挑战失败"}), 500
-
-    session[_ADMIN_PASSKEY_REGISTER_SESSION_KEY] = {
-        "admin_id": admin_id,
-        "challenge": challenge,
-        "rp_id": rp_id,
-        "expected_origins": expected_origins,
-        "device_name": device_name,
-        "created_at": time.time(),
-    }
-    session.modified = True
-    return jsonify({"publicKey": options, "limit": MAX_PASSKEYS_PER_OWNER})
-
-
-@admin_api_bp.route("/admin/passkeys/register/verify", methods=["POST"])
-@admin_required
-def admin_passkey_register_verify():
-    admin_id = int(session.get("admin_id") or 0)
-    state = session.get(_ADMIN_PASSKEY_REGISTER_SESSION_KEY) or {}
-    if not state:
-        return jsonify({"error": "Passkey挑战不存在或已过期，请重试"}), 400
-    if int(state.get("admin_id") or 0) != admin_id:
-        return jsonify({"error": "Passkey挑战与当前管理员不匹配"}), 400
-    if not is_challenge_valid(state.get("created_at")):
-        session.pop(_ADMIN_PASSKEY_REGISTER_SESSION_KEY, None)
-        return jsonify({"error": "Passkey挑战已过期，请重试"}), 400
-
-    data = request.get_json(silent=True) or {}
-    credential = _parse_credential_payload(data)
-    if not credential:
-        return jsonify({"error": "Passkey参数缺失"}), 400
-
-    count = database.count_passkeys("admin", admin_id)
-    if count >= MAX_PASSKEYS_PER_OWNER:
-        session.pop(_ADMIN_PASSKEY_REGISTER_SESSION_KEY, None)
-        return jsonify({"error": f"最多可绑定{MAX_PASSKEYS_PER_OWNER}台设备"}), 400
-
-    try:
-        verified = verify_registration(
-            credential=credential,
-            expected_challenge=str(state.get("challenge") or ""),
-            expected_rp_id=str(state.get("rp_id") or ""),
-            expected_origins=list(state.get("expected_origins") or []),
-        )
-    except Exception as e:
-        logger.warning(f"[passkey] 管理员注册验签失败(admin_id={admin_id}): {e}")
-        return jsonify({"error": "Passkey验证失败，请重试"}), 400
-
-    device_name = normalize_device_name(data.get("device_name") if "device_name" in data else state.get("device_name"))
-
-    created_id = database.create_passkey(
-        "admin",
-        admin_id,
-        credential_id=encode_credential_id(verified.credential_id),
-        public_key=encode_credential_id(verified.credential_public_key),
-        sign_count=int(verified.sign_count or 0),
-        device_name=device_name,
-        transports=get_credential_transports(credential),
-        aaguid=str(verified.aaguid or ""),
-    )
-    if not created_id:
-        return jsonify({"error": "该Passkey已绑定，或保存失败"}), 400
-
-    session.pop(_ADMIN_PASSKEY_REGISTER_SESSION_KEY, None)
-    return jsonify({"success": True, "id": int(created_id)})
-
-
-@admin_api_bp.route("/admin/passkeys/<int:passkey_id>", methods=["DELETE"])
-@admin_required
-def delete_admin_passkey(passkey_id):
-    admin_id = int(session.get("admin_id") or 0)
-    ok = database.delete_passkey("admin", admin_id, int(passkey_id))
-    if ok:
-        return jsonify({"success": True})
-    return jsonify({"error": "设备不存在或已删除"}), 404
-
-
-@admin_api_bp.route("/admin/passkeys/client-error", methods=["POST"])
-@admin_required
-def report_admin_passkey_client_error():
-    """上报管理员端浏览器 Passkey 失败详情，便于排查兼容性问题。"""
-    data = request.get_json(silent=True) or {}
-    error_name = _truncate_text(data.get("name"), 120)
-    error_message = _truncate_text(data.get("message"), 400)
-    error_code = _truncate_text(data.get("code"), 120)
-    ua = _truncate_text(data.get("user_agent") or request.headers.get("User-Agent", ""), 300)
-    stage = _truncate_text(data.get("stage"), 80)
-    source = _truncate_text(data.get("source"), 80)
-    admin_id = int(session.get("admin_id") or 0)
-
-    logger.warning(
-        "[passkey][client-error][admin] admin_id=%s stage=%s source=%s name=%s code=%s message=%s ua=%s",
-        admin_id,
-        stage or "-",
-        source or "-",
-        error_name or "-",
-        error_code or "-",
-        error_message or "-",
-        ua or "-",
-    )
-    return jsonify({"success": True})
-
-
 @admin_api_bp.route("/admin/reauth", methods=["POST"])
 @admin_required
 def admin_reauth():

routes/api_auth.py

@@ -3,7 +3,6 @@
 from __future__ import annotations
 
 import base64
-import json
 import random
 import secrets
 import threading
@@ -21,15 +20,6 @@ from flask_login import login_required, login_user, logout_user
 from routes.pages import render_app_spa_or_legacy
 from services.accounts_service import load_user_accounts
 from services.models import User
-from services.passkeys import (
-    encode_credential_id,
-    get_expected_origins,
-    get_rp_id,
-    is_challenge_valid,
-    make_authentication_options,
-    normalize_device_name,
-    verify_authentication,
-)
 from services.social_login import parse_providers, provider_label
 from services.state import (
     check_ip_request_rate,
@@ -61,7 +51,6 @@ _CAPTCHA_FONT_PATHS = [
 ]
 _CAPTCHA_FONT = None
 _CAPTCHA_FONT_LOCK = threading.Lock()
-_USER_PASSKEY_LOGIN_SESSION_KEY = "user_passkey_login_state"
 
 
 def _get_json_payload() -> dict:
@@ -206,19 +195,6 @@ def _send_login_security_alert_if_needed(user: dict, username: str, client_ip: s
         logger.warning(f"发送登录安全提醒失败: user_id={user.get('id')}, error={e}")
 
 
-def _parse_credential_payload(data: dict) -> dict | None:
-    credential = data.get("credential")
-    if isinstance(credential, dict):
-        return credential
-    if isinstance(credential, str):
-        try:
-            parsed = json.loads(credential)
-            return parsed if isinstance(parsed, dict) else None
-        except Exception:
-            return None
-    return None
-
-
 @api_auth_bp.route("/api/register", methods=["POST"])
 @require_ip_not_locked
 def register():
@@ -586,166 +562,6 @@ def generate_captcha():
         return jsonify({"error": "验证码服务暂不可用，请联系管理员安装PIL库"}), 503
 
 
-@api_auth_bp.route("/api/passkeys/login/options", methods=["POST"])
-@require_ip_not_locked
-def user_passkey_login_options():
-    """用户 Passkey 登录：获取 assertion challenge。"""
-    data = _get_json_payload()
-    username = str(data.get("username", "") or "").strip()
-    client_ip = get_rate_limit_ip()
-    mode = "named" if username else "discoverable"
-    username_key = f"passkey:{username}" if username else "passkey:discoverable"
-
-    is_locked, remaining = check_login_ip_user_locked(client_ip, username_key)
-    if is_locked:
-        wait_hint = f"{remaining // 60 + 1}分钟" if remaining >= 60 else f"{remaining}秒"
-        return jsonify({"error": f"账号短时锁定，请{wait_hint}后再试"}), 429
-
-    allowed, error_msg = check_ip_request_rate(client_ip, "login")
-    if not allowed:
-        return jsonify({"error": error_msg}), 429
-
-    allowed, error_msg = check_login_rate_limits(client_ip, username_key)
-    if not allowed:
-        return jsonify({"error": error_msg}), 429
-
-    user_id = 0
-    allow_credential_ids = []
-    if mode == "named":
-        user = database.get_user_by_username(username)
-        if not user or user.get("status") != "approved":
-            record_login_failure(client_ip, username_key)
-            return jsonify({"error": "账号或Passkey不可用"}), 400
-
-        user_id = int(user["id"])
-        passkeys = database.list_passkeys("user", user_id)
-        if not passkeys:
-            record_login_failure(client_ip, username_key)
-            return jsonify({"error": "该账号尚未绑定Passkey"}), 400
-        allow_credential_ids = [str(item.get("credential_id") or "").strip() for item in passkeys if item.get("credential_id")]
-
-    try:
-        rp_id = get_rp_id(request)
-        expected_origins = get_expected_origins(request)
-    except Exception as e:
-        logger.warning(f"[passkey] 生成登录 challenge 失败(mode={mode}, username={username or '-'}) : {e}")
-        return jsonify({"error": "Passkey配置异常，请联系管理员"}), 500
-
-    options = make_authentication_options(rp_id=rp_id, allow_credential_ids=allow_credential_ids)
-    challenge = str(options.get("challenge") or "").strip()
-    if not challenge:
-        return jsonify({"error": "生成Passkey挑战失败"}), 500
-
-    session[_USER_PASSKEY_LOGIN_SESSION_KEY] = {
-        "mode": mode,
-        "username": username,
-        "user_id": int(user_id),
-        "challenge": challenge,
-        "rp_id": rp_id,
-        "expected_origins": expected_origins,
-        "username_key": username_key,
-        "created_at": time.time(),
-    }
-    session.modified = True
-    return jsonify({"publicKey": options})
-
-
-@api_auth_bp.route("/api/passkeys/login/verify", methods=["POST"])
-@require_ip_not_locked
-def user_passkey_login_verify():
-    """用户 Passkey 登录：校验 assertion 并登录。"""
-    data = _get_json_payload()
-    request_username = str(data.get("username", "") or "").strip()
-    credential = _parse_credential_payload(data)
-    if not credential:
-        return jsonify({"error": "Passkey参数缺失"}), 400
-
-    state = session.get(_USER_PASSKEY_LOGIN_SESSION_KEY) or {}
-    if not state:
-        return jsonify({"error": "Passkey挑战不存在或已过期，请重试"}), 400
-    if not is_challenge_valid(state.get("created_at")):
-        session.pop(_USER_PASSKEY_LOGIN_SESSION_KEY, None)
-        return jsonify({"error": "Passkey挑战已过期，请重试"}), 400
-
-    mode = str(state.get("mode") or "named")
-    if mode not in {"named", "discoverable"}:
-        session.pop(_USER_PASSKEY_LOGIN_SESSION_KEY, None)
-        return jsonify({"error": "Passkey状态异常，请重试"}), 400
-
-    expected_username = str(state.get("username") or "").strip()
-    username = expected_username
-    if mode == "named":
-        if not expected_username:
-            session.pop(_USER_PASSKEY_LOGIN_SESSION_KEY, None)
-            return jsonify({"error": "Passkey状态异常，请重试"}), 400
-        if request_username and request_username != expected_username:
-            return jsonify({"error": "用户名与挑战不匹配，请重试"}), 400
-    else:
-        username = request_username
-
-    client_ip = get_rate_limit_ip()
-    username_key = str(state.get("username_key") or "").strip() or (
-        f"passkey:{expected_username}" if mode == "named" else "passkey:discoverable"
-    )
-
-    is_locked, remaining = check_login_ip_user_locked(client_ip, username_key)
-    if is_locked:
-        wait_hint = f"{remaining // 60 + 1}分钟" if remaining >= 60 else f"{remaining}秒"
-        return jsonify({"error": f"账号短时锁定，请{wait_hint}后再试"}), 429
-
-    credential_id = str(credential.get("id") or credential.get("rawId") or "").strip()
-    if not credential_id:
-        return jsonify({"error": "Passkey参数无效"}), 400
-
-    passkey = database.get_passkey_by_credential_id(credential_id)
-    if not passkey:
-        record_login_failure(client_ip, username_key)
-        return jsonify({"error": "Passkey不存在或已删除"}), 401
-    if str(passkey.get("owner_type") or "") != "user":
-        record_login_failure(client_ip, username_key)
-        return jsonify({"error": "Passkey不属于用户账号"}), 401
-    if mode == "named" and int(passkey.get("owner_id") or 0) != int(state.get("user_id") or 0):
-        record_login_failure(client_ip, username_key)
-        return jsonify({"error": "Passkey与账号不匹配"}), 401
-
-    try:
-        parsed_credential, verified = verify_authentication(
-            credential=credential,
-            expected_challenge=str(state.get("challenge") or ""),
-            expected_rp_id=str(state.get("rp_id") or ""),
-            expected_origins=list(state.get("expected_origins") or []),
-            credential_public_key=str(passkey.get("public_key") or ""),
-            credential_current_sign_count=int(passkey.get("sign_count") or 0),
-        )
-        verified_credential_id = encode_credential_id(verified.credential_id)
-        if verified_credential_id != str(passkey.get("credential_id") or ""):
-            raise ValueError("credential_id mismatch")
-    except Exception as e:
-        logger.warning(f"[passkey] 用户登录验签失败(mode={mode}, username={expected_username or request_username or '-'}) : {e}")
-        record_login_failure(client_ip, username_key)
-        return jsonify({"error": "Passkey验证失败"}), 401
-
-    user_id = int(passkey.get("owner_id") or 0)
-    user = database.get_user_by_id(user_id)
-    if not user or user.get("status") != "approved":
-        return jsonify({"error": "账号不可用"}), 401
-
-    database.update_passkey_usage(int(passkey["id"]), int(verified.new_sign_count))
-    clear_login_failures(client_ip, username_key)
-    user_login_key = f"passkey:{str(user.get('username') or '').strip()}"
-    if user_login_key and user_login_key != username_key:
-        clear_login_failures(client_ip, user_login_key)
-    session.pop(_USER_PASSKEY_LOGIN_SESSION_KEY, None)
-
-    user_obj = User(user_id)
-    login_user(user_obj)
-    load_user_accounts(user_id)
-
-    resolved_username = str(user.get("username") or "").strip() or username or f"user-{user_id}"
-    _send_login_security_alert_if_needed(user=user, username=resolved_username, client_ip=client_ip)
-    return jsonify({"success": True, "credential_id": parsed_credential.id, "username": resolved_username})
-
-
 @api_auth_bp.route("/api/login", methods=["POST"])
 @require_ip_not_locked
 def login():

routes/api_user.py

@@ -2,9 +2,6 @@
 # -*- coding: utf-8 -*-
 from __future__ import annotations
 
-import json
-import time
-
 import database
 import email_service
 from app_logger import get_logger
@@ -12,24 +9,12 @@ from app_security import get_rate_limit_ip, require_ip_not_locked, validate_emai
 from flask import Blueprint, jsonify, request, session
 from flask_login import current_user, login_required
 from routes.pages import render_app_spa_or_legacy
-from services.passkeys import (
-    MAX_PASSKEYS_PER_OWNER,
-    encode_credential_id,
-    get_credential_transports,
-    get_expected_origins,
-    get_rp_id,
-    is_challenge_valid,
-    make_registration_options,
-    normalize_device_name,
-    verify_registration,
-)
 from services.state import check_email_rate_limit, check_ip_request_rate, safe_iter_task_status_items
 from services.tasks import get_task_scheduler
 
 logger = get_logger("app")
 
 api_user_bp = Blueprint("api_user", __name__)
-_USER_PASSKEY_REGISTER_SESSION_KEY = "user_passkey_register_state"
 
 
 def _get_current_user_record():
@@ -61,26 +46,6 @@ def _coerce_binary_flag(value, *, field_label: str):
     return value, None
 
 
-def _parse_credential_payload(data: dict) -> dict | None:
-    credential = data.get("credential")
-    if isinstance(credential, dict):
-        return credential
-    if isinstance(credential, str):
-        try:
-            parsed = json.loads(credential)
-            return parsed if isinstance(parsed, dict) else None
-        except Exception:
-            return None
-    return None
-
-
-def _truncate_text(value, max_len: int = 300) -> str:
-    text = str(value or "").strip()
-    if len(text) > max_len:
-        return f"{text[:max_len]}..."
-    return text
-
-
 def _check_bind_email_rate_limits(email: str):
     client_ip = get_rate_limit_ip()
     allowed, error_msg = check_ip_request_rate(client_ip, "email")
@@ -409,176 +374,6 @@ def update_user_email_notify():
     return jsonify({"error": "更新失败"}), 500
 
 
-@api_user_bp.route("/api/user/passkeys", methods=["GET"])
-@login_required
-def list_user_passkeys():
-    """获取当前用户绑定的 Passkey 设备列表。"""
-    rows = database.list_passkeys("user", int(current_user.id))
-    items = []
-    for row in rows:
-        credential_id = str(row.get("credential_id") or "")
-        preview = ""
-        if credential_id:
-            preview = f"{credential_id[:8]}...{credential_id[-6:]}" if len(credential_id) > 16 else credential_id
-        items.append(
-            {
-                "id": int(row.get("id")),
-                "device_name": str(row.get("device_name") or ""),
-                "credential_id_preview": preview,
-                "created_at": row.get("created_at"),
-                "last_used_at": row.get("last_used_at"),
-                "transports": str(row.get("transports") or ""),
-            }
-        )
-    return jsonify({"items": items, "limit": MAX_PASSKEYS_PER_OWNER})
-
-
-@api_user_bp.route("/api/user/passkeys/register/options", methods=["POST"])
-@login_required
-def user_passkey_register_options():
-    """当前登录用户创建 Passkey：下发 registration challenge。"""
-    user, error_response = _get_current_user_or_404()
-    if error_response:
-        return error_response
-
-    count = database.count_passkeys("user", int(current_user.id))
-    if count >= MAX_PASSKEYS_PER_OWNER:
-        return jsonify({"error": f"最多可绑定{MAX_PASSKEYS_PER_OWNER}台设备"}), 400
-
-    data = request.get_json(silent=True) or {}
-    device_name = normalize_device_name(data.get("device_name"))
-
-    existing = database.list_passkeys("user", int(current_user.id))
-    exclude_credential_ids = [str(item.get("credential_id") or "").strip() for item in existing if item.get("credential_id")]
-
-    try:
-        rp_id = get_rp_id(request)
-        expected_origins = get_expected_origins(request)
-    except Exception as e:
-        logger.warning(f"[passkey] 用户注册 options 失败(user_id={current_user.id}): {e}")
-        return jsonify({"error": "Passkey配置异常，请联系管理员"}), 500
-
-    try:
-        options = make_registration_options(
-            rp_id=rp_id,
-            rp_name="知识管理平台",
-            user_name=str(user.get("username") or f"user-{current_user.id}"),
-            user_display_name=str(user.get("username") or f"user-{current_user.id}"),
-            user_id_bytes=f"user:{int(current_user.id)}".encode("utf-8"),
-            exclude_credential_ids=exclude_credential_ids,
-        )
-    except Exception as e:
-        logger.warning(f"[passkey] 用户注册 options 构建失败(user_id={current_user.id}): {e}")
-        return jsonify({"error": "生成Passkey挑战失败"}), 500
-
-    challenge = str(options.get("challenge") or "").strip()
-    if not challenge:
-        return jsonify({"error": "生成Passkey挑战失败"}), 500
-
-    session[_USER_PASSKEY_REGISTER_SESSION_KEY] = {
-        "user_id": int(current_user.id),
-        "challenge": challenge,
-        "rp_id": rp_id,
-        "expected_origins": expected_origins,
-        "device_name": device_name,
-        "created_at": time.time(),
-    }
-    session.modified = True
-    return jsonify({"publicKey": options, "limit": MAX_PASSKEYS_PER_OWNER})
-
-
-@api_user_bp.route("/api/user/passkeys/register/verify", methods=["POST"])
-@login_required
-def user_passkey_register_verify():
-    """当前登录用户创建 Passkey：校验 attestation 并落库。"""
-    state = session.get(_USER_PASSKEY_REGISTER_SESSION_KEY) or {}
-    if not state:
-        return jsonify({"error": "Passkey挑战不存在或已过期，请重试"}), 400
-    if int(state.get("user_id") or 0) != int(current_user.id):
-        return jsonify({"error": "Passkey挑战与当前用户不匹配"}), 400
-    if not is_challenge_valid(state.get("created_at")):
-        session.pop(_USER_PASSKEY_REGISTER_SESSION_KEY, None)
-        return jsonify({"error": "Passkey挑战已过期，请重试"}), 400
-
-    data = request.get_json(silent=True) or {}
-    credential = _parse_credential_payload(data)
-    if not credential:
-        return jsonify({"error": "Passkey参数缺失"}), 400
-
-    count = database.count_passkeys("user", int(current_user.id))
-    if count >= MAX_PASSKEYS_PER_OWNER:
-        session.pop(_USER_PASSKEY_REGISTER_SESSION_KEY, None)
-        return jsonify({"error": f"最多可绑定{MAX_PASSKEYS_PER_OWNER}台设备"}), 400
-
-    try:
-        verified = verify_registration(
-            credential=credential,
-            expected_challenge=str(state.get("challenge") or ""),
-            expected_rp_id=str(state.get("rp_id") or ""),
-            expected_origins=list(state.get("expected_origins") or []),
-        )
-    except Exception as e:
-        logger.warning(f"[passkey] 用户注册验签失败(user_id={current_user.id}): {e}")
-        return jsonify({"error": "Passkey验证失败，请重试"}), 400
-
-    credential_id = encode_credential_id(verified.credential_id)
-    public_key = encode_credential_id(verified.credential_public_key)
-    transports = get_credential_transports(credential)
-    device_name = normalize_device_name(data.get("device_name") if "device_name" in data else state.get("device_name"))
-    aaguid = str(verified.aaguid or "")
-
-    created_id = database.create_passkey(
-        "user",
-        int(current_user.id),
-        credential_id=credential_id,
-        public_key=public_key,
-        sign_count=int(verified.sign_count or 0),
-        device_name=device_name,
-        transports=transports,
-        aaguid=aaguid,
-    )
-    if not created_id:
-        return jsonify({"error": "该Passkey已绑定，或保存失败"}), 400
-
-    session.pop(_USER_PASSKEY_REGISTER_SESSION_KEY, None)
-    return jsonify({"success": True, "id": int(created_id), "device_name": device_name})
-
-
-@api_user_bp.route("/api/user/passkeys/<int:passkey_id>", methods=["DELETE"])
-@login_required
-def delete_user_passkey(passkey_id):
-    """删除当前用户绑定的 Passkey 设备。"""
-    ok = database.delete_passkey("user", int(current_user.id), int(passkey_id))
-    if ok:
-        return jsonify({"success": True})
-    return jsonify({"error": "设备不存在或已删除"}), 404
-
-
-@api_user_bp.route("/api/user/passkeys/client-error", methods=["POST"])
-@login_required
-def report_user_passkey_client_error():
-    """上报浏览器端 Passkey 失败详情，便于排查兼容性问题。"""
-    data = request.get_json(silent=True) or {}
-    error_name = _truncate_text(data.get("name"), 120)
-    error_message = _truncate_text(data.get("message"), 400)
-    error_code = _truncate_text(data.get("code"), 120)
-    ua = _truncate_text(data.get("user_agent") or request.headers.get("User-Agent", ""), 300)
-    stage = _truncate_text(data.get("stage"), 80)
-    source = _truncate_text(data.get("source"), 80)
-
-    logger.warning(
-        "[passkey][client-error][user] user_id=%s stage=%s source=%s name=%s code=%s message=%s ua=%s",
-        current_user.id,
-        stage or "-",
-        source or "-",
-        error_name or "-",
-        error_code or "-",
-        error_message or "-",
-        ua or "-",
-    )
-    return jsonify({"success": True})
-
-
 @api_user_bp.route("/api/run_stats", methods=["GET"])
 @login_required
 def get_run_stats():