# auth.py # 此檔案負責身份驗證相關功能 # 包含登入處理、會話管理、角色切換等 import datetime import threading import secrets from config import USER_DB, FIXED_TEST_EMAIL from models import record_log, global_stats from utils import get_current_time import gradio as gr # --- Identity verification session cache --- # key: user_email # value: { status(active/verifying/blocked), pending_token, anomaly_timestamp } session_cache = {} session_cache_lock = threading.Lock() def _get_or_create_session_entry(user_email: str): """獲取或創建會話條目""" user_email = user_email.strip().replace("\xa0", " ") with session_cache_lock: if user_email not in session_cache: session_cache[user_email] = { "status": "active", "pending_token": None, "anomaly_timestamp": None, } return session_cache[user_email] def check_session_timeouts(timeout_seconds=300): """檢查會話超時，如果 verifying 超過 timeout_seconds 則改為 blocked""" now = datetime.datetime.now() with session_cache_lock: for user_email, entry in session_cache.items(): if entry.get("status") == "verifying" and entry.get("anomaly_timestamp"): try: # 嘗試解析多種可能的格式 ts_str = entry["anomaly_timestamp"] try: ts = datetime.datetime.strptime(ts_str, "%Y-%m-%d %H:%M:%S") except ValueError: # 處理可能帶有時區或其他格式的情況 import pandas as pd ts = pd.to_datetime(ts_str).to_pydatetime().replace(tzinfo=None) if (now - ts).total_seconds() > timeout_seconds: entry["status"] = "blocked" entry["pending_token"] = None print(f"⏰ [Timeout] User {user_email} status changed to blocked due to verification timeout.") except Exception as e: print(f"❌ [Timeout Check Error] {e}") def get_user_status(user_email: str): """獲取用戶當前狀態""" user_email = user_email.strip().replace("\xa0", " ") with session_cache_lock: if user_email not in session_cache: return "active" return session_cache[user_email].get("status", "active") def unblock_user(user_email: str): """手動解封用戶（以帳戶名稱作為 key）""" user_email = user_email.strip().replace("\xa0", " ") with session_cache_lock: if user_email in session_cache: session_cache[user_email]["status"] = "active" session_cache[user_email]["pending_token"] = None session_cache[user_email]["anomaly_timestamp"] = None return True else: # 就算 session_cache 中沒有此 key，也視為解鎖成功（帳戶原本就是 active） return True def _verify_token_to_user(token: str): """根據 token 驗證用戶""" token = token.strip() with session_cache_lock: for user_email, entry in session_cache.items(): if entry.get("pending_token") == token: return user_email return None def process_login(ip, cookie, account, password, captcha, time_mode, custom_time): """處理登入邏輯""" def fail_response(msg, df_all, df_abn, inp_upd): return gr.update(selected="login_route"), gr.update(), msg, gr.update(), df_all, df_abn, inp_upd if account not in USER_DB: df_all, df_abn, inp_upd = record_log(ip, cookie, account, "Unknown", "login_attempt", "failed_no_account", time_mode, custom_time) return fail_response("帳號不存在！", df_all, df_abn, inp_upd) role, name = USER_DB[account][1], USER_DB[account][2] if captcha.upper() != "FEIK": df_all, df_abn, inp_upd = record_log(ip, cookie, account, role, "login_attempt", "failed_captcha", time_mode, custom_time) return fail_response("驗證碼錯誤！", df_all, df_abn, inp_upd) if USER_DB[account][0] == password: df_all, df_abn, inp_upd = record_log(ip, cookie, account, role, "login_attempt", "success", time_mode, custom_time) return gr.update(selected="dashboard_route"), gr.update(selected="student_tab" if role == "學生" else "admin_tab"), "", f"**{name}** 歡迎您 | 線上人數: 938", df_all, df_abn, inp_upd else: df_all, df_abn, inp_upd = record_log(ip, cookie, account, role, "login_attempt", "failed_password", time_mode, custom_time) return fail_response("密碼錯誤！", df_all, df_abn, inp_upd) def internal_process_login(ip, cookie, account, password, captcha, time_mode, custom_time): """內部處理登入邏輯，回傳 JSON 友善格式""" if account not in USER_DB: df_all, df_abn, inp_upd = record_log(ip, cookie, account, "Unknown", "login_attempt", "failed_no_account", time_mode, custom_time) return {"status": "error", "message": "帳號不存在！"} if get_user_status(account) == "blocked": df_all, df_abn, inp_upd = record_log(ip, cookie, account, USER_DB[account][1], "login_attempt", "failed_blocked", time_mode, custom_time) return {"status": "error", "message": "此帳號已被封鎖，無法登入！"} role, name = USER_DB[account][1], USER_DB[account][2] if captcha.upper() != "FEIK": df_all, df_abn, inp_upd = record_log(ip, cookie, account, role, "login_attempt", "failed_captcha", time_mode, custom_time) return {"status": "error", "message": "驗證碼錯誤！"} if USER_DB[account][0] == password: df_all, df_abn, inp_upd = record_log(ip, cookie, account, role, "login_attempt", "success", time_mode, custom_time) return { "status": "success", "role": "admin" if role == "管理" else "student", "name": name, "message": "登入成功" } else: df_all, df_abn, inp_upd = record_log(ip, cookie, account, role, "login_attempt", "failed_password", time_mode, custom_time) return {"status": "error", "message": "密碼錯誤！"} def internal_logout(ip, cookie, account, time_mode, custom_time): """內部處理登出邏輯""" role = USER_DB.get(account, ["", "Unknown"])[1] record_log(ip, cookie, account, role, "logout", "success", time_mode, custom_time) return {"status": "success", "message": "已登出"} def fast_login(target_account, ip, cookie, time_mode, custom_time): """快速登入""" role, name = USER_DB[target_account][1], USER_DB[target_account][2] df_all, df_abn, inp_upd = record_log(ip, cookie, target_account, role, "login_attempt", "success", time_mode, custom_time) target_tab = "student_tab" if role == "學生" else "admin_tab" return gr.update(selected="dashboard_route"), gr.update(selected=target_tab), "", f"**{name}** 歡迎您 | 線上人數: 938 (⚡快速登入模式)", df_all, df_abn, inp_upd, target_account def toggle_role(current_account, ip, cookie, time_mode, custom_time): """切換角色""" target_account = "student" if current_account == "admin" else "admin" role, name = USER_DB[target_account][1], USER_DB[target_account][2] df_all, df_abn, inp_upd = record_log(ip, cookie, target_account, role, "login_attempt", "success", time_mode, custom_time) return gr.update(selected="student_tab" if role == "學生" else "admin_tab"), f"**{name}** 歡迎您 | 線上人數: 938 (⚡單鍵切換模式)", df_all, df_abn, inp_upd, target_account def logout(ip, cookie, account, time_mode, custom_time): """登出""" role = USER_DB.get(account, ["", "Unknown"])[1] df_all, df_abn, inp_upd = record_log(ip, cookie, account, role, "logout", "success", time_mode, custom_time) return gr.update(selected="login_route"), "", "", "", "", df_all, df_abn, inp_upd def simulate_click(ip, cookie, account, button_name, time_mode, custom_time): """模擬點擊操作""" # 以帳戶名稱查詢封鎖狀態（不再使用全域 FIXED_TEST_EMAIL 作為 key） status = get_user_status(account) if status == "verifying": gr.Info("偵測到異常，請於5分鐘內確認 Email 以繼續使用") return gr.update(), gr.update(), gr.update() elif status == "blocked": gr.Warning("帳號已被封鎖，請聯絡系統管理員或寄驗證信進行解封") return gr.update(), gr.update(), gr.update() role = USER_DB.get(account, ["", "Unknown"])[1] return record_log(ip, cookie, account, role, button_name, "success", time_mode, custom_time) def test_anomaly(ip, cookie, account, action_name, test_time_str): """測試異常偵測""" # 測試異常時不阻擋，以便觀察結果 role = USER_DB.get(account, ["", "Unknown"])[1] return record_log(ip, cookie, account, role, action_name, "success", "自訂時間 (模擬過去/未來)", test_time_str)