Passkey bug fix in gunicorn
This commit is contained in:
+23
-2
@@ -544,7 +544,7 @@ def api_webauthn_register_complete():
|
|||||||
uid = _cur()
|
uid = _cur()
|
||||||
|
|
||||||
try:
|
try:
|
||||||
result = complete_registration(request, credential)
|
result = complete_registration(request, credential, user_id=uid)
|
||||||
if result.get("user_id") != uid:
|
if result.get("user_id") != uid:
|
||||||
return jsonify({"error": "注册上下文不匹配,请重试"}), 400
|
return jsonify({"error": "注册上下文不匹配,请重试"}), 400
|
||||||
update_auth_config(
|
update_auth_config(
|
||||||
@@ -569,6 +569,27 @@ def api_webauthn_register_complete():
|
|||||||
return jsonify({"error": f"Passkey 注册失败: {e}"}), 500
|
return jsonify({"error": f"Passkey 注册失败: {e}"}), 500
|
||||||
|
|
||||||
|
|
||||||
|
@api.route("/api/auth/passkey", methods=["DELETE"])
|
||||||
|
@login_required
|
||||||
|
def api_remove_passkey():
|
||||||
|
"""Remove the current user's Passkey (re-auth required).
|
||||||
|
|
||||||
|
If this leaves the account with no MFA method at all, the account is
|
||||||
|
flagged must_configure_mfa so the next login forces MFA setup again
|
||||||
|
(same guard as a fresh temp account).
|
||||||
|
"""
|
||||||
|
if not require_reauth():
|
||||||
|
return jsonify({"error": "需要重新验证身份", "reauth_required": True}), 403
|
||||||
|
uid = _cur()
|
||||||
|
cfg = get_auth_config(user_id=uid)
|
||||||
|
if not cfg.get("passkey_credential_id"):
|
||||||
|
return jsonify({"error": "未配置 Passkey"}), 400
|
||||||
|
update_auth_config(uid, passkey_credential_id="", passkey_public_key="")
|
||||||
|
if not has_mfa_configured(user_id=uid):
|
||||||
|
update_auth_config(uid, must_configure_mfa=1)
|
||||||
|
return jsonify({"ok": True})
|
||||||
|
|
||||||
|
|
||||||
# ─── TOTP Setup ──────────────────────────────
|
# ─── TOTP Setup ──────────────────────────────
|
||||||
|
|
||||||
@api.route("/api/auth/totp/setup", methods=["POST"])
|
@api.route("/api/auth/totp/setup", methods=["POST"])
|
||||||
@@ -974,7 +995,7 @@ def admin_provision_passkey_complete(user_id):
|
|||||||
return jsonify({"error": "用户不存在"}), 404
|
return jsonify({"error": "用户不存在"}), 404
|
||||||
|
|
||||||
try:
|
try:
|
||||||
result = complete_registration(request, credential)
|
result = complete_registration(request, credential, user_id=user_id)
|
||||||
if result.get("user_id") != user_id:
|
if result.get("user_id") != user_id:
|
||||||
return jsonify({"error": "注册上下文不匹配,请重试"}), 400
|
return jsonify({"error": "注册上下文不匹配,请重试"}), 400
|
||||||
update_auth_config(
|
update_auth_config(
|
||||||
|
|||||||
+23
-12
@@ -153,32 +153,43 @@ def begin_registration(request, user_id: int, display_name: str = "",
|
|||||||
return _serialize_options(options)
|
return _serialize_options(options)
|
||||||
|
|
||||||
|
|
||||||
def complete_registration(request, credential: dict) -> dict:
|
def complete_registration(request, credential: dict, user_id: int = None) -> dict:
|
||||||
"""Complete WebAuthn registration.
|
"""Complete WebAuthn registration.
|
||||||
|
|
||||||
Returns {'ok': True, 'user_id': int, 'credential_id': str, 'public_key': str}
|
Returns {'ok': True, 'user_id': int, 'credential_id': str, 'public_key': str}
|
||||||
or raises an exception on failure.
|
or raises an exception on failure.
|
||||||
|
|
||||||
|
user_id: explicit target user. Always pass it from the route when known —
|
||||||
|
admin provisioning runs inside the ADMIN's session (its uid differs from
|
||||||
|
the target user, so deriving the challenge key from the session would never
|
||||||
|
match). When omitted, it is derived from credential.user_name ("user_5" → 5),
|
||||||
|
then from the authenticated request context.
|
||||||
"""
|
"""
|
||||||
rp_id = _get_rp_id(request)
|
rp_id = _get_rp_id(request)
|
||||||
origin = _get_origin(request)
|
origin = _get_origin(request)
|
||||||
|
|
||||||
user_name = credential.get("user_name", "")
|
uid = user_id
|
||||||
# Derive the target user id from the registration challenge key.
|
if uid is None:
|
||||||
expected_challenge = pop_webauthn_challenge(f"reg_{user_name}", db_path=_challenge_db())
|
uname = credential.get("user_name", "")
|
||||||
if not expected_challenge:
|
if isinstance(uname, str) and uname.startswith("user_"):
|
||||||
# Self-registration fallback: the browser may not echo the user handle,
|
try:
|
||||||
# so resolve the target from the authenticated request context.
|
uid = int(uname.split("_")[-1])
|
||||||
|
except (ValueError, IndexError):
|
||||||
|
uid = None
|
||||||
|
if uid is None:
|
||||||
try:
|
try:
|
||||||
from backend.auth import current_user_id
|
from backend.auth import current_user_id
|
||||||
uid = current_user_id()
|
uid = current_user_id()
|
||||||
if uid is not None:
|
|
||||||
expected_challenge = pop_webauthn_challenge(f"reg_{uid}", db_path=_challenge_db())
|
|
||||||
user_name = f"user_{uid}"
|
|
||||||
except Exception:
|
except Exception:
|
||||||
pass
|
uid = None
|
||||||
|
if uid is None:
|
||||||
|
raise ValueError("No registration challenge found. Please restart registration.")
|
||||||
|
|
||||||
|
# Challenge key must match begin_registration's f"reg_{user_id}" exactly.
|
||||||
|
expected_challenge = pop_webauthn_challenge(f"reg_{uid}", db_path=_challenge_db())
|
||||||
if not expected_challenge:
|
if not expected_challenge:
|
||||||
raise ValueError("No registration challenge found. Please restart registration.")
|
raise ValueError("No registration challenge found. Please restart registration.")
|
||||||
user_id = int(user_name.split("_")[-1])
|
user_id = uid
|
||||||
|
|
||||||
try:
|
try:
|
||||||
# Library expects rawId field (browser-native camelCase, base64url string)
|
# Library expects rawId field (browser-native camelCase, base64url string)
|
||||||
|
|||||||
@@ -930,6 +930,23 @@
|
|||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
// ─── Passkey Removal (from settings) ──────
|
||||||
|
var removePasskeyBtn = $('mfa-remove-passkey');
|
||||||
|
if (removePasskeyBtn) {
|
||||||
|
removePasskeyBtn.addEventListener('click', function () {
|
||||||
|
_reauthThen(function () {
|
||||||
|
api('DELETE', '/api/auth/passkey', null, function (res) {
|
||||||
|
if (res.ok) {
|
||||||
|
toast('✅ Passkey 已移除');
|
||||||
|
loadMFAStatus();
|
||||||
|
} else {
|
||||||
|
toast('❌ ' + (res.error || '移除失败'));
|
||||||
|
}
|
||||||
|
});
|
||||||
|
});
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
// ─── TOTP Setup ───────────────────────────
|
// ─── TOTP Setup ───────────────────────────
|
||||||
$('mfa-setup-totp').addEventListener('click', function () {
|
$('mfa-setup-totp').addEventListener('click', function () {
|
||||||
_reauthThen(function () {
|
_reauthThen(function () {
|
||||||
|
|||||||
@@ -655,8 +655,10 @@ orig_complete_auth = routes_mod.complete_authentication
|
|||||||
routes_mod.begin_registration = lambda request, user_id, display_name="", existing_credential_id=None: {
|
routes_mod.begin_registration = lambda request, user_id, display_name="", existing_credential_id=None: {
|
||||||
"challenge": "AA", "user": {"id": b64url(f"user_{user_id}".encode().hex()), "name": f"user_{user_id}", "displayName": display_name},
|
"challenge": "AA", "user": {"id": b64url(f"user_{user_id}".encode().hex()), "name": f"user_{user_id}", "displayName": display_name},
|
||||||
}
|
}
|
||||||
routes_mod.complete_registration = lambda request, credential: {
|
routes_mod.complete_registration = lambda request, credential, user_id=None: {
|
||||||
"ok": True, "user_id": int(credential.get("user_name", "user_0").split("_")[-1]),
|
"ok": True,
|
||||||
|
"user_id": user_id if user_id is not None
|
||||||
|
else int(credential.get("user_name", "user_0").split("_")[-1]),
|
||||||
"credential_id": CRED_HEX, "public_key": "abcdef",
|
"credential_id": CRED_HEX, "public_key": "abcdef",
|
||||||
}
|
}
|
||||||
routes_mod.begin_authentication = lambda request, allow_credential_ids=None, challenge_key="login": {"challenge": "BB"}
|
routes_mod.begin_authentication = lambda request, allow_credential_ids=None, challenge_key="login": {"challenge": "BB"}
|
||||||
@@ -723,6 +725,20 @@ routes_mod.complete_registration = orig_complete_reg
|
|||||||
routes_mod.begin_authentication = orig_begin_auth
|
routes_mod.begin_authentication = orig_begin_auth
|
||||||
routes_mod.complete_authentication = orig_complete_auth
|
routes_mod.complete_authentication = orig_complete_auth
|
||||||
|
|
||||||
|
# 移除 Passkey(Root 自助,TOTP reauth)
|
||||||
|
rr = reauth_totp(root15, root15_secret)
|
||||||
|
r = root15.delete("/api/auth/passkey")
|
||||||
|
check("移除 Passkey 成功", r.status_code == 200, str(r.get_json()))
|
||||||
|
cfg = get_auth_config(db15, 0)
|
||||||
|
check("移除后凭据清空", cfg["passkey_credential_id"] == "" and cfg["passkey_public_key"] == "")
|
||||||
|
check("仍有 TOTP 不强制重配 MFA", cfg["must_configure_mfa"] == 0)
|
||||||
|
r = root15.delete("/api/auth/passkey")
|
||||||
|
check("未配置时重复移除 → 400", r.status_code == 400)
|
||||||
|
# 未 reauth 的会话移除 → 403
|
||||||
|
fresh15, _ = do_login(app15.test_client(), pyotp.TOTP(root15_secret).now())
|
||||||
|
r = fresh15.delete("/api/auth/passkey")
|
||||||
|
check("移除未 reauth → 403", r.status_code == 403 and r.get_json().get("reauth_required"))
|
||||||
|
|
||||||
print()
|
print()
|
||||||
print("=" * 60)
|
print("=" * 60)
|
||||||
print("TEST 14: Scheduler thread context (per-user timezone, no request context)")
|
print("TEST 14: Scheduler thread context (per-user timezone, no request context)")
|
||||||
@@ -1158,6 +1174,56 @@ check("下一个请求立即读到新时区(无陈旧缓存)", r.get_json()[
|
|||||||
st = c22.get("/api/display/state").get_json()
|
st = c22.get("/api/display/state").get_json()
|
||||||
check("display/state 返回新时区", st.get("timezone") == "America/New_York")
|
check("display/state 返回新时区", st.get("timezone") == "America/New_York")
|
||||||
|
|
||||||
|
print()
|
||||||
|
print("=" * 60)
|
||||||
|
print("TEST 23: Passkey 注册 challenge key 匹配(admin 代配非 Root 用户,真实 webauthn_mgr)")
|
||||||
|
print("=" * 60)
|
||||||
|
from backend import webauthn_mgr as wm
|
||||||
|
import base64 as _b64
|
||||||
|
|
||||||
|
app23, db23 = fresh_app("t23")
|
||||||
|
root23 = root_client(app23, db23)
|
||||||
|
root23_secret = get_auth_config(db23, 0)["totp_secret"]
|
||||||
|
rr = reauth_totp(root23, root23_secret)
|
||||||
|
r = root23.post("/api/admin/users", json={"remark": "PK Target", "mode": "admin_fills"})
|
||||||
|
target_id = r.get_json()["user"]["id"]
|
||||||
|
|
||||||
|
# 真实 begin_registration(admin 代配,目标用户非 Root)
|
||||||
|
with app23.test_request_context():
|
||||||
|
from flask import request as _req
|
||||||
|
opts = wm.begin_registration(_req, user_id=target_id, display_name="PK Target")
|
||||||
|
challenge_b64 = opts["challenge"]
|
||||||
|
got = pop_webauthn_challenge(f"reg_{target_id}", db_path=db23)
|
||||||
|
check("challenge 存于 reg_<uid>(key 拼装一致)", got == challenge_b64)
|
||||||
|
|
||||||
|
# 用显式 user_id 的 complete_registration(mock 掉 verify,验证 key 匹配路径)
|
||||||
|
with app23.test_request_context():
|
||||||
|
_req2 = _req
|
||||||
|
wm.begin_registration(_req2, user_id=target_id, display_name="PK Target")
|
||||||
|
class _FakeVer:
|
||||||
|
credential_id = bytes.fromhex("aabb" * 8)
|
||||||
|
credential_public_key = bytes.fromhex("ccdd" * 8)
|
||||||
|
_orig_verify = wm.verify_registration_response
|
||||||
|
wm.verify_registration_response = lambda **kw: _FakeVer()
|
||||||
|
try:
|
||||||
|
with app23.test_request_context():
|
||||||
|
res = wm.complete_registration(_req2, {
|
||||||
|
"id": "qqq", "rawId": "qqq", "user_name": f"user_{target_id}",
|
||||||
|
}, user_id=target_id)
|
||||||
|
check("admin 代配 complete 显式 user_id 命中 challenge", res["user_id"] == target_id)
|
||||||
|
finally:
|
||||||
|
wm.verify_registration_response = _orig_verify
|
||||||
|
|
||||||
|
# 反向验证:若 key 不匹配(如用错 user_id)→ 明确报错
|
||||||
|
with app23.test_request_context():
|
||||||
|
wm.begin_registration(_req2, user_id=target_id, display_name="PK Target")
|
||||||
|
try:
|
||||||
|
with app23.test_request_context():
|
||||||
|
wm.complete_registration(_req2, {"id": "qqq", "rawId": "qqq"}, user_id=0)
|
||||||
|
check("错误 user_id 被拒绝", False, "没有抛错")
|
||||||
|
except ValueError as e:
|
||||||
|
check("错误 user_id 被拒绝(challenge 不匹配)", "No registration challenge found" in str(e))
|
||||||
|
|
||||||
print()
|
print()
|
||||||
print("=" * 60)
|
print("=" * 60)
|
||||||
print(f"RESULT: {PASS} passed, {FAIL} failed")
|
print(f"RESULT: {PASS} passed, {FAIL} failed")
|
||||||
|
|||||||
Reference in New Issue
Block a user