完成階段二：偵測觸發、5分鐘計時器與驗證邏輯輯

api.py

@@ -7,11 +7,9 @@ from fastapi.responses import JSONResponse
 from pydantic import BaseModel
 from typing import Optional, List
 
-from config import SYSTEM_WEIGHTS, ACTION_MAP
-from models import global_stats, log_history, baseline_df, AI_READY
-from utils import _extract_gr_update_value, update_system_weights
-from auth import simulate_click, USER_DB
+from auth import simulate_click, USER_DB, check_session_timeouts, get_user_status
 from security import test_send_security_alert, verify_identity
+from config import SYSTEM_WEIGHTS, ACTION_MAP, FIXED_TEST_EMAIL
 from plotting import get_plot_data_for_api
 import pandas as pd
 
@@ -148,6 +146,23 @@ def test_send_security_alert_endpoint():
 # --- Student Actions ---
 
 def _run_student_action_helper(ip, c, acc, action_code, tm, ct):
+    # 每次請求時檢查超時
+    check_session_timeouts(timeout_seconds=300)
+    
+    # 檢查用戶當前狀態 (目前統一使用 FIXED_TEST_EMAIL 作為識別)
+    current_status = get_user_status(FIXED_TEST_EMAIL)
+    
+    if current_status == "verifying":
+        return {
+            "status": "pending_verify",
+            "message": "偵測到異常，請於5分鐘內確認 Email 以繼續使用"
+        }
+    elif current_status == "blocked":
+        return {
+            "status": "blocked",
+            "message": "帳號已被封鎖，請聯絡系統管理員或寄驗證信進行解封"
+        }
+
     df_all, df_abn, inp_upd = simulate_click(ip, c, acc, action_code, tm, ct)
     val = _extract_gr_update_value(inp_upd)
     return {
@@ -189,6 +204,23 @@ def student_sql_injection(req: StudentActionExtendedRequest):
 @api_app.post("/api/student/run_action")
 def student_run_action(req: GenericActionRequest):
     """通用學生操作端點"""
+    # 每次請求時檢查超時
+    check_session_timeouts(timeout_seconds=300)
+    
+    # 檢查用戶當前狀態
+    current_status = get_user_status(FIXED_TEST_EMAIL)
+    
+    if current_status == "verifying":
+        return {
+            "status": "pending_verify",
+            "message": "偵測到異常，請於5分鐘內確認 Email 以繼續使用"
+        }
+    elif current_status == "blocked":
+        return {
+            "status": "blocked",
+            "message": "帳號已被封鎖，請聯絡系統管理員或寄驗證信進行解封"
+        }
+
     df_all, df_abn, inp_upd = simulate_click(req.ip, req.cookie, req.account, req.action_code, req.time_mode, req.custom_time)
     val = _extract_gr_update_value(inp_upd)
     return {

auth.py

@@ -18,6 +18,7 @@ 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] = {
@@ -27,8 +28,51 @@ def _get_or_create_session_entry(user_email: str):
             }
         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):
+    """手動解封用戶"""
+    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
+    return False
+
 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:
@@ -79,10 +123,20 @@ def logout(ip, cookie, account, time_mode, custom_time):
 
 def simulate_click(ip, cookie, account, button_name, time_mode, custom_time):
     """模擬點擊操作"""
+    # 檢查用戶狀態
+    status = get_user_status(FIXED_TEST_EMAIL)
+    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)

models.py

@@ -254,6 +254,17 @@ def record_log(ip, cookie, account, role, action, status, time_mode, custom_time
         ai_judgment = check_anomaly(ip, account, action, log_time)
         if ai_judgment != "success":
             status = ai_judgment
+            
+            # 如果是異常行為，觸發安全警報與驗證流程
+            if "異常" in status or "受限" in status:
+                try:
+                    from security import trigger_identity_verification
+                    from config import FIXED_TEST_EMAIL
+                    # 我們將 account 映射到測試信箱進行警報寄送
+                    trigger_identity_verification(FIXED_TEST_EMAIL, log_time)
+                    print(f"🚨 [Security Alert Triggered] Account: {account}, Status: {status}")
+                except Exception as e:
+                    print(f"❌ [Security Alert Error] {e}")
 
     new_log = {
         "timestamp": log_time,

security.py

@@ -102,9 +102,11 @@ def trigger_identity_verification(user_email: str, anomaly_timestamp: str):
     return verify_token, success
 
 def verify_identity(token: str, choice: str):
-    """郵件鏈接回調"""
+    """郵件鏈接回調處理"""
+    token = token.strip()
     user_email = _verify_token_to_user(token)
     if not user_email:
+        print(f"❌ [Verification Failed] Invalid token: {token}")
         return JSONResponse({"ok": False, "error": "Invalid or expired token"}, status_code=400)
 
     # 處理選擇
@@ -120,6 +122,8 @@ def verify_identity(token: str, choice: str):
         return JSONResponse({"ok": False, "error": "Invalid choice"}, status_code=400)
 
     with session_cache_lock:
+        # 確保 key 經過正規化
+        user_email = user_email.strip().replace("\xa0", " ")
         entry = session_cache.get(user_email)
         if not entry:
             return JSONResponse({"ok": False, "error": "Session not found"}, status_code=404)

ui.py

@@ -6,7 +6,7 @@ import gradio as gr
 import pandas as pd
 from config import ACTION_MAP, USER_DB, SYSTEM_WEIGHTS
 from models import log_history, AI_READY
-from auth import process_login, fast_login, logout, toggle_role, simulate_click, test_anomaly
+from auth import process_login, fast_login, logout, toggle_role, simulate_click, test_anomaly, unblock_user
 from security import explain_abnormal_log
 from plotting import get_dashboard_html, get_plot_data, timer_update
 from utils import update_system_weights, get_current_time
@@ -238,6 +238,12 @@ def create_demo():
                                             abnormal_log_input = gr.Textbox(label="要分析的異常 Log", placeholder="系統若偵測到新異常將會自動填入...", scale=4)
                                             btn_explain_log = gr.Button("🧠 Groq AI 分析", variant="primary", scale=1)
                                         ai_explanation_output = gr.Textbox(label="AI 分析結果與說明", lines=4, interactive=False)
+                                    
+                                    with gr.Accordion("▼ 使用者狀態管理", open=True, elem_classes="weight-card"):
+                                        from config import FIXED_TEST_EMAIL
+                                        gr.Markdown(f"<span style='color:gray; font-size:0.9em;'>手動解除對測試帳號 ({FIXED_TEST_EMAIL}) 的封鎖狀態。</span>")
+                                        btn_unblock = gr.Button("🔓 解除封鎖 (Unblock User)", variant="primary")
+                                        unblock_status = gr.Markdown(value="")
 
                                     with gr.Accordion("▼ HDBSCAN 異常辨識自動化測試區", open=True, elem_classes="weight-card"):
                                         gr.Markdown("<span style='color:gray; font-size:0.9em;'>點擊按鈕將直接以指定時間點執行該系統的操作，結果將顯示於左側 Log 區。</span>")
@@ -298,6 +304,17 @@ def create_demo():
         action_dropdown.change(fn=full_ui_update, inputs=[action_dropdown], outputs=[stat_dashboard_display, action_plot, log_display, abnormal_log_display])
         time_mode.change(fn=toggle_time_input, inputs=time_mode, outputs=custom_time_input)
         btn_explain_log.click(fn=explain_abnormal_log, inputs=[abnormal_log_input], outputs=[ai_explanation_output])
+        
+        def handle_unblock():
+            from config import FIXED_TEST_EMAIL
+            success = unblock_user(FIXED_TEST_EMAIL)
+            if success:
+                gr.Info(f"已成功解除 {FIXED_TEST_EMAIL} 的封鎖狀態。")
+                return "✅ 狀態已重設為 Active"
+            else:
+                return "❌ 找不到該使用者或已是 Active 狀態"
+
+        btn_unblock.click(fn=handle_unblock, inputs=[], outputs=[unblock_status])
 
         update_outputs = [main_router, role_tabs, login_msg, welcome_text, store_df_all, store_df_abn, abnormal_log_input]