BrianChuan commited on
Commit
97a461c
·
1 Parent(s): fb04509

新增更多使用者與管理者帳號,調整API呼叫行為,mail內文調整

Browse files
__pycache__/api.cpython-311.pyc CHANGED
Binary files a/__pycache__/api.cpython-311.pyc and b/__pycache__/api.cpython-311.pyc differ
 
__pycache__/auth.cpython-311.pyc CHANGED
Binary files a/__pycache__/auth.cpython-311.pyc and b/__pycache__/auth.cpython-311.pyc differ
 
__pycache__/config.cpython-311.pyc CHANGED
Binary files a/__pycache__/config.cpython-311.pyc and b/__pycache__/config.cpython-311.pyc differ
 
__pycache__/models.cpython-311.pyc CHANGED
Binary files a/__pycache__/models.cpython-311.pyc and b/__pycache__/models.cpython-311.pyc differ
 
__pycache__/security.cpython-311.pyc CHANGED
Binary files a/__pycache__/security.cpython-311.pyc and b/__pycache__/security.cpython-311.pyc differ
 
api.py CHANGED
@@ -16,7 +16,7 @@ from config import SYSTEM_WEIGHTS, ACTION_MAP, FIXED_TEST_EMAIL
16
  from plotting import get_plot_data_for_api
17
  import pandas as pd
18
  import models # 使用 models.XXX 存取即時狀態,避免布林值被複製後失效
19
- from utils import get_geo_level
20
 
21
  api_app = FastAPI(title="數位保鏢 API")
22
 
@@ -272,8 +272,8 @@ def _run_student_action_helper(ip, c, acc, action_code, tm, ct):
272
  # 每次請求時檢查超時
273
  check_session_timeouts(timeout_seconds=300)
274
 
275
- # 檢查用戶當前狀態 (目前統一使用 FIXED_TEST_EMAIL 作為識別)
276
- current_status = get_user_status(FIXED_TEST_EMAIL)
277
 
278
  if current_status == "verifying":
279
  return {
@@ -295,14 +295,12 @@ def _run_student_action_helper(ip, c, acc, action_code, tm, ct):
295
  }
296
 
297
  @api_app.post("/api/student/eval")
298
- def student_eval(req: StudentActionRequest):
299
- default_acc = USER_DB["student"][0]
300
- return _run_student_action_helper(req.ip, req.cookie, default_acc, "click_eval_system", req.time_mode, req.custom_time)
301
 
302
  @api_app.post("/api/student/pre_select")
303
- def student_pre_select(req: StudentActionRequest):
304
- default_acc = USER_DB["student"][0]
305
- return _run_student_action_helper(req.ip, req.cookie, default_acc, "click_pre_select_system", req.time_mode, req.custom_time)
306
 
307
  @api_app.post("/api/student/leave")
308
  def student_leave(req: StudentActionExtendedRequest):
@@ -330,8 +328,8 @@ def student_run_action(req: GenericActionRequest):
330
  # 每次請求時檢查超時
331
  check_session_timeouts(timeout_seconds=300)
332
 
333
- # 檢查用戶當前狀態
334
- current_status = get_user_status(FIXED_TEST_EMAIL)
335
 
336
  if current_status == "verifying":
337
  return {
@@ -355,4 +353,53 @@ def student_run_action(req: GenericActionRequest):
355
  "action": req.action_code,
356
  "status": val or "success"
357
  }
358
- }
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
16
  from plotting import get_plot_data_for_api
17
  import pandas as pd
18
  import models # 使用 models.XXX 存取即時狀態,避免布林值被複製後失效
19
+ from utils import get_geo_level, update_system_weights
20
 
21
  api_app = FastAPI(title="數位保鏢 API")
22
 
 
272
  # 每次請求時檢查超時
273
  check_session_timeouts(timeout_seconds=300)
274
 
275
+ # 直接以帳戶名稱作為 key 查詢封鎖狀態
276
+ current_status = get_user_status(acc)
277
 
278
  if current_status == "verifying":
279
  return {
 
295
  }
296
 
297
  @api_app.post("/api/student/eval")
298
+ def student_eval(req: StudentActionExtendedRequest):
299
+ return _run_student_action_helper(req.ip, req.cookie, req.account, "click_eval_system", req.time_mode, req.custom_time)
 
300
 
301
  @api_app.post("/api/student/pre_select")
302
+ def student_pre_select(req: StudentActionExtendedRequest):
303
+ return _run_student_action_helper(req.ip, req.cookie, req.account, "click_pre_select_system", req.time_mode, req.custom_time)
 
304
 
305
  @api_app.post("/api/student/leave")
306
  def student_leave(req: StudentActionExtendedRequest):
 
328
  # 每次請求時檢查超時
329
  check_session_timeouts(timeout_seconds=300)
330
 
331
+ # 直接以帳戶名稱作為 key 查詢封鎖狀態
332
+ current_status = get_user_status(req.account)
333
 
334
  if current_status == "verifying":
335
  return {
 
353
  "action": req.action_code,
354
  "status": val or "success"
355
  }
356
+ }
357
+
358
+ # --- 使用者管理 API 端點 ---
359
+ class AccountRequest(BaseModel):
360
+ account: str
361
+
362
+ @api_app.get("/api/users")
363
+ def get_users_endpoint():
364
+ """取得所有使用者帳戶清單與其狀態"""
365
+ users_list = []
366
+ for acc, info in USER_DB.items():
367
+ role_en = "admin" if info[1] == "管理" else "student"
368
+ status = get_user_status(acc)
369
+ users_list.append({
370
+ "account": acc,
371
+ "role": role_en,
372
+ "role_zh": info[1],
373
+ "name": info[2],
374
+ "status": status
375
+ })
376
+ return {"status": "success", "data": users_list}
377
+
378
+ @api_app.post("/api/users/block")
379
+ def block_user_endpoint(req: AccountRequest):
380
+ """封鎖特定使用者"""
381
+ acc = req.account.strip()
382
+ if acc not in USER_DB:
383
+ return {"status": "error", "message": "帳號不存在"}
384
+
385
+ from auth import session_cache_lock, session_cache
386
+ with session_cache_lock:
387
+ if acc not in session_cache:
388
+ session_cache[acc] = {}
389
+ session_cache[acc]["status"] = "blocked"
390
+ session_cache[acc]["pending_token"] = None
391
+ session_cache[acc]["anomaly_timestamp"] = None
392
+
393
+ return {"status": "success", "message": f"帳號 {acc} 已成功封鎖"}
394
+
395
+ @api_app.post("/api/users/unblock")
396
+ def unblock_user_endpoint(req: AccountRequest):
397
+ """解鎖特定使用者"""
398
+ acc = req.account.strip()
399
+ if acc not in USER_DB:
400
+ return {"status": "error", "message": "帳號不存在"}
401
+
402
+ from auth import unblock_user
403
+ unblock_user(acc)
404
+
405
+ return {"status": "success", "message": f"帳號 {acc} 已成功解封"}
auth.py CHANGED
@@ -60,7 +60,7 @@ def get_user_status(user_email: str):
60
  return session_cache[user_email].get("status", "active")
61
 
62
  def unblock_user(user_email: str):
63
- """手動解封用戶"""
64
  user_email = user_email.strip().replace("\xa0", " ")
65
  with session_cache_lock:
66
  if user_email in session_cache:
@@ -68,7 +68,9 @@ def unblock_user(user_email: str):
68
  session_cache[user_email]["pending_token"] = None
69
  session_cache[user_email]["anomaly_timestamp"] = None
70
  return True
71
- return False
 
 
72
 
73
  def _verify_token_to_user(token: str):
74
  """根據 token 驗證用戶"""
@@ -107,6 +109,10 @@ def internal_process_login(ip, cookie, account, password, captcha, time_mode, cu
107
  df_all, df_abn, inp_upd = record_log(ip, cookie, account, "Unknown", "login_attempt", "failed_no_account", time_mode, custom_time)
108
  return {"status": "error", "message": "帳號不存在!"}
109
 
 
 
 
 
110
  role, name = USER_DB[account][1], USER_DB[account][2]
111
 
112
  if captcha.upper() != "FEIK":
@@ -153,8 +159,8 @@ def logout(ip, cookie, account, time_mode, custom_time):
153
 
154
  def simulate_click(ip, cookie, account, button_name, time_mode, custom_time):
155
  """模擬點擊操作"""
156
- # 檢查用戶狀態
157
- status = get_user_status(FIXED_TEST_EMAIL)
158
  if status == "verifying":
159
  gr.Info("偵測到異常,請於5分鐘內確認 Email 以繼續使用")
160
  return gr.update(), gr.update(), gr.update()
 
60
  return session_cache[user_email].get("status", "active")
61
 
62
  def unblock_user(user_email: str):
63
+ """手動解封用戶(以帳戶名稱作為 key)"""
64
  user_email = user_email.strip().replace("\xa0", " ")
65
  with session_cache_lock:
66
  if user_email in session_cache:
 
68
  session_cache[user_email]["pending_token"] = None
69
  session_cache[user_email]["anomaly_timestamp"] = None
70
  return True
71
+ else:
72
+ # 就算 session_cache 中沒有此 key,也視為解鎖成功(帳戶原本就是 active)
73
+ return True
74
 
75
  def _verify_token_to_user(token: str):
76
  """根據 token 驗證用戶"""
 
109
  df_all, df_abn, inp_upd = record_log(ip, cookie, account, "Unknown", "login_attempt", "failed_no_account", time_mode, custom_time)
110
  return {"status": "error", "message": "帳號不存在!"}
111
 
112
+ if get_user_status(account) == "blocked":
113
+ df_all, df_abn, inp_upd = record_log(ip, cookie, account, USER_DB[account][1], "login_attempt", "failed_blocked", time_mode, custom_time)
114
+ return {"status": "error", "message": "此帳號已被封鎖,無法登入!"}
115
+
116
  role, name = USER_DB[account][1], USER_DB[account][2]
117
 
118
  if captcha.upper() != "FEIK":
 
159
 
160
  def simulate_click(ip, cookie, account, button_name, time_mode, custom_time):
161
  """模擬點擊操作"""
162
+ # 以帳戶名稱查詢封鎖狀態(不再使用全域 FIXED_TEST_EMAIL 作為 key)
163
+ status = get_user_status(account)
164
  if status == "verifying":
165
  gr.Info("偵測到異常,請於5分鐘內確認 Email 以繼續使用")
166
  return gr.update(), gr.update(), gr.update()
config.py CHANGED
@@ -9,7 +9,7 @@ from dotenv import load_dotenv
9
  load_dotenv()
10
 
11
  # 測試用固定郵箱
12
- FIXED_TEST_EMAIL = os.getenv("FIXED_TEST_EMAIL", "t111360205@ntut.org.tw").strip()
13
 
14
  # SMTP 設定
15
  SMTP_CONFIG = {
@@ -46,8 +46,10 @@ ACTION_MAP = {
46
 
47
  # 用戶資料庫
48
  USER_DB = {
49
- "student": ["1234", "學生", "同學"],
50
- "admin": ["1234", "管理", "管理員"]
 
 
51
  }
52
 
53
  # 伺服器設定
 
9
  load_dotenv()
10
 
11
  # 測試用固定郵箱
12
+ FIXED_TEST_EMAIL = os.getenv("FIXED_TEST_EMAIL", "[EMAIL_ADDRESS]").strip()
13
 
14
  # SMTP 設定
15
  SMTP_CONFIG = {
 
46
 
47
  # 用戶資料庫
48
  USER_DB = {
49
+ "student": ["1234", "學生", "Alex Wang"],
50
+ "student2": ["1234", "學生", "Bella Lin"],
51
+ "admin": ["1234", "管理", "Charles Chen"],
52
+ "admin2": ["1234", "管理", "David Lee"]
53
  }
54
 
55
  # 伺服器設定
models.py CHANGED
@@ -257,13 +257,13 @@ def record_log(ip, cookie, account, role, action, status, time_mode, custom_time
257
  if ai_judgment != "success":
258
  status = ai_judgment
259
 
260
- # 如果是異常行為,觸發安全警報與驗證流程
261
  if "異常" in status or "受限" in status:
262
  try:
263
  from security import trigger_identity_verification
264
  from config import FIXED_TEST_EMAIL
265
- # 我們將 account 映射到測試信箱進行警報寄送
266
- trigger_identity_verification(FIXED_TEST_EMAIL, log_time)
267
  print(f"🚨 [Security Alert Triggered] Account: {account}, Status: {status}")
268
  except Exception as e:
269
  print(f"❌ [Security Alert Error] {e}")
 
257
  if ai_judgment != "success":
258
  status = ai_judgment
259
 
260
+ # 如果是異常行為,觸發安全警報與驗證流程(以帳戶名稱作為 session key)
261
  if "異常" in status or "受限" in status:
262
  try:
263
  from security import trigger_identity_verification
264
  from config import FIXED_TEST_EMAIL
265
+ # 以帳戶名稱作為識別 key,信件收件人統一使用 FIXED_TEST_EMAIL
266
+ trigger_identity_verification(account, FIXED_TEST_EMAIL, log_time)
267
  print(f"🚨 [Security Alert Triggered] Account: {account}, Status: {status}")
268
  except Exception as e:
269
  print(f"❌ [Security Alert Error] {e}")
security.py CHANGED
@@ -18,11 +18,12 @@ from auth import _get_or_create_session_entry, session_cache_lock, session_cache
18
  from utils import get_current_time
19
  from fastapi.responses import JSONResponse
20
 
21
- def send_security_alert(user_email: str, verify_token: str):
22
  """
23
  當偵測到異常時,發送包含兩個鏈接的郵件:
24
  - 這是我本人(信任)
25
  - 這不是我(封鎖)
 
26
  """
27
  smtp_host = SMTP_CONFIG["host"]
28
  smtp_port = SMTP_CONFIG["port"]
@@ -36,35 +37,42 @@ def send_security_alert(user_email: str, verify_token: str):
36
  smtp_username = smtp_username.replace("\xa0", " ")
37
  smtp_password = smtp_password.replace("\xa0", " ")
38
  base_url = base_url.replace("\xa0", " ")
39
- user_email = user_email.replace("\xa0", " ")
40
 
41
  trust_url = f"{base_url}/verify_identity?token={quote(verify_token)}&choice=trust"
42
  block_url = f"{base_url}/verify_identity?token={quote(verify_token)}&choice=block"
43
 
44
  # 始終記錄鏈接以供調試/測試(即使 SMTP 未配置)
45
- print(f"🔗 [Security Alert] trust_url={trust_url}")
46
- print(f"🔗 [Security Alert] block_url={block_url}")
47
 
48
  if not smtp_host or not smtp_username or not smtp_password:
49
  print("⚠️ SMTP 設定未完成(SMTP_HOST/SMTP_USERNAME/SMTP_PASSWORD 缺少),跳過寄信。")
50
  return False
51
 
52
- subject = "安全提醒:請確認是否為本人操作"
 
 
 
53
  html_body = f"""
54
- <p>系統偵測到可能的異常登入/操作。</p>
55
- <p>請你在下方選擇其中一項:</p>
56
- <ul>
57
- <li><a href="{trust_url}">這是我本人(信任)</a></li>
58
- <li><a href="{block_url}">這不是我(封鎖)</a></li>
59
- </ul>
60
- <p>若你未主動發起這次操作,建議選擇「這不是我(封鎖)」。</p>
 
 
 
 
61
  """
62
  # 某些環境可能引入非斷行空格,會觸發 ASCII 編碼錯誤
63
  html_body = html_body.replace("\xa0", " ")
64
  msg = MIMEText(html_body, "html", "utf-8", policy=SMTPUTF8)
65
  msg["Subject"] = subject
66
  msg["From"] = smtp_from
67
- msg["To"] = user_email
68
 
69
  try:
70
  # 使用 SMTP + STARTTLS 進行安全傳輸
@@ -77,16 +85,20 @@ def send_security_alert(user_email: str, verify_token: str):
77
  server.login(smtp_username, smtp_password)
78
  # send_message 處理非 ascii 內容的正確編碼
79
  server.send_message(msg)
80
- print(f"✅ 安全警報郵件已成功寄送至: {user_email}")
81
  return True
82
  except Exception as e:
83
  print(f"❌ 寄信失敗: {e}")
84
  return False
85
 
86
- def trigger_identity_verification(user_email: str, anomaly_timestamp: str):
87
- """觸發身份驗證"""
 
 
 
 
88
  # 如果已經封鎖,保持更強的決定
89
- entry = _get_or_create_session_entry(user_email)
90
  with session_cache_lock:
91
  if entry.get("status") == "blocked":
92
  return None, False
@@ -98,7 +110,8 @@ def trigger_identity_verification(user_email: str, anomaly_timestamp: str):
98
  "anomaly_timestamp": anomaly_timestamp,
99
  })
100
 
101
- success = send_security_alert(user_email, verify_token)
 
102
  return verify_token, success
103
 
104
  def verify_identity(token: str, choice: str):
@@ -145,7 +158,8 @@ def test_send_security_alert():
145
  - 返回生成的 verify token(token 也存儲在 session_cache 中)
146
  """
147
  anomaly_timestamp = datetime.datetime.now(datetime.timezone(datetime.timedelta(hours=8))).strftime("%Y-%m-%d %H:%M:%S")
148
- token, success = trigger_identity_verification(FIXED_TEST_EMAIL, anomaly_timestamp)
 
149
  if not token:
150
  return {"ok": False, "error": "Already blocked or session not available"}
151
  return {"ok": True, "user_email": FIXED_TEST_EMAIL, "token": token, "email_sent": success}
 
18
  from utils import get_current_time
19
  from fastapi.responses import JSONResponse
20
 
21
+ def send_security_alert(recipient_email: str, verify_token: str, account_name: str = ""):
22
  """
23
  當偵測到異常時,發送包含兩個鏈接的郵件:
24
  - 這是我本人(信任)
25
  - 這不是我(封鎖)
26
+ account_name: 觸發異常的帳戶名稱(用於顯示在信件中)
27
  """
28
  smtp_host = SMTP_CONFIG["host"]
29
  smtp_port = SMTP_CONFIG["port"]
 
37
  smtp_username = smtp_username.replace("\xa0", " ")
38
  smtp_password = smtp_password.replace("\xa0", " ")
39
  base_url = base_url.replace("\xa0", " ")
40
+ recipient_email = recipient_email.replace("\xa0", " ")
41
 
42
  trust_url = f"{base_url}/verify_identity?token={quote(verify_token)}&choice=trust"
43
  block_url = f"{base_url}/verify_identity?token={quote(verify_token)}&choice=block"
44
 
45
  # 始終記錄鏈接以供調試/測試(即使 SMTP 未配置)
46
+ print(f"🔗 [Security Alert] account={account_name}, trust_url={trust_url}")
47
+ print(f"🔗 [Security Alert] account={account_name}, block_url={block_url}")
48
 
49
  if not smtp_host or not smtp_username or not smtp_password:
50
  print("⚠️ SMTP 設定未完成(SMTP_HOST/SMTP_USERNAME/SMTP_PASSWORD 缺少),跳過寄信。")
51
  return False
52
 
53
+ # 帳號顯示名稱(若有提供則加入說明)
54
+ account_info = f"<p><strong>⚠️ 警示帳號:{account_name}</strong></p>" if account_name else ""
55
+
56
+ subject = f"[數位保鏢] 安全提醒:帳號 {account_name} 偵測到異常操作" if account_name else "安全提醒:請確認是否為本人操作"
57
  html_body = f"""
58
+ <div style="font-family: Arial, sans-serif; max-width: 600px; margin: 0 auto; padding: 20px; border: 1px solid #e0e0e0; border-radius: 8px;">
59
+ <h2 style="color: #d32f2f;">🔐 數位保鏢安全警示通知</h2>
60
+ {account_info}
61
+ <p>系統偵測到此帳號可能發生異常登入或操作行為,請確認是否為本人所為。</p>
62
+ <p>請於 <strong>5 分鐘內</strong> 點選以下其中一項進行確認:</p>
63
+ <div style="margin: 20px 0;">
64
+ <a href="{trust_url}" style="display: inline-block; padding: 12px 24px; background-color: #2e7d32; color: white; text-decoration: none; border-radius: 6px; margin-right: 12px;">✅ 這是我本人(繼續使用)</a>
65
+ <a href="{block_url}" style="display: inline-block; padding: 12px 24px; background-color: #c62828; color: white; text-decoration: none; border-radius: 6px;">🚫 這不是我(立即封鎖)</a>
66
+ </div>
67
+ <p style="color: #757575; font-size: 12px;">若您未主動進行任何操作,請立即點選「這不是我(立即封鎖)」以保護帳號安全。逾時未確認將自動封鎖帳號。</p>
68
+ </div>
69
  """
70
  # 某些環境可能引入非斷行空格,會觸發 ASCII 編碼錯誤
71
  html_body = html_body.replace("\xa0", " ")
72
  msg = MIMEText(html_body, "html", "utf-8", policy=SMTPUTF8)
73
  msg["Subject"] = subject
74
  msg["From"] = smtp_from
75
+ msg["To"] = recipient_email
76
 
77
  try:
78
  # 使用 SMTP + STARTTLS 進行安全傳輸
 
85
  server.login(smtp_username, smtp_password)
86
  # send_message 處理非 ascii 內容的正確編碼
87
  server.send_message(msg)
88
+ print(f"✅ 安全警報郵件已成功寄送至: {recipient_email}(帳號:{account_name})")
89
  return True
90
  except Exception as e:
91
  print(f"❌ 寄信失敗: {e}")
92
  return False
93
 
94
+ def trigger_identity_verification(account_key: str, recipient_email: str, anomaly_timestamp: str):
95
+ """
96
+ 觸發身份驗證
97
+ account_key: 用於 session_cache 的識別 key(帳戶名稱)
98
+ recipient_email: 安全警報信件的收件人(統一使用 FIXED_TEST_EMAIL)
99
+ """
100
  # 如果已經封鎖,保持更強的決定
101
+ entry = _get_or_create_session_entry(account_key)
102
  with session_cache_lock:
103
  if entry.get("status") == "blocked":
104
  return None, False
 
110
  "anomaly_timestamp": anomaly_timestamp,
111
  })
112
 
113
+ # 寄送信件時傳入帳戶名稱,讓收件人知道是哪個帳號觸發警示
114
+ success = send_security_alert(recipient_email, verify_token, account_name=account_key)
115
  return verify_token, success
116
 
117
  def verify_identity(token: str, choice: str):
 
158
  - 返回生成的 verify token(token 也存儲在 session_cache 中)
159
  """
160
  anomaly_timestamp = datetime.datetime.now(datetime.timezone(datetime.timedelta(hours=8))).strftime("%Y-%m-%d %H:%M:%S")
161
+ # 測試時以 FIXED_TEST_EMAIL 同時作為 account_key 與收件人
162
+ token, success = trigger_identity_verification(FIXED_TEST_EMAIL, FIXED_TEST_EMAIL, anomaly_timestamp)
163
  if not token:
164
  return {"ok": False, "error": "Already blocked or session not available"}
165
  return {"ok": True, "user_email": FIXED_TEST_EMAIL, "token": token, "email_sent": success}