# security.py # 此檔案負責安全相關功能 # 包含安全警報發送、身份驗證觸發、異常分析等 import smtplib import ssl import secrets import datetime import requests import json from email.policy import SMTPUTF8 from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from urllib.parse import quote import os from config import SMTP_CONFIG, FIXED_TEST_EMAIL, RESEND_API_KEY, EMAIL_PROVIDER from auth import _get_or_create_session_entry, session_cache_lock, session_cache, _verify_token_to_user from utils import get_current_time from fastapi.responses import JSONResponse def _send_via_resend(recipient_email: str, subject: str, html_body: str) -> bool: """ 使用 Resend HTTP API 發送郵件。 適合部署至 Hugging Face Spaces 等不支援傳統 SMTP 的環境。 收件人固定寄送至 h6696201@gmail.com（測試用途）。 """ if not RESEND_API_KEY: print("⚠️ RESEND_API_KEY 未設定，無法使用 Resend 發送郵件。") return False # Resend 免費方案要求發件人使用已驗證的網域， # 使用 onboarding@resend.dev 可在未驗證自有網域時發送至任意收件人。 from_address = "Digital Bodyguard " payload = { "from": from_address, "to": [recipient_email], "subject": subject, "html": html_body, } headers = { "Authorization": f"Bearer {RESEND_API_KEY}", "Content-Type": "application/json", } try: resp = requests.post( "https://api.resend.com/emails", headers=headers, json=payload, timeout=20, ) if resp.status_code in (200, 201): data = resp.json() print(f"✅ [Resend] 郵件已成功發送至: {recipient_email} (id={data.get('id', 'N/A')})") return True else: print(f"❌ [Resend] 發送失敗: HTTP {resp.status_code} → {resp.text}") return False except Exception as e: print(f"❌ [Resend] 請求異常: {e}") return False def _send_via_smtp(recipient_email: str, subject: str, html_body: str) -> bool: """ 使用傳統 SMTP 發送郵件（保留作為備援 / 本地開發使用）。 """ smtp_host = SMTP_CONFIG["host"].replace("\xa0", " ") smtp_port = SMTP_CONFIG["port"] smtp_username = SMTP_CONFIG["username"].replace("\xa0", " ") smtp_password = SMTP_CONFIG["password"].replace("\xa0", " ") smtp_from = SMTP_CONFIG["from"] if not smtp_host or not smtp_username or not smtp_password: print("⚠️ SMTP 設定未完成（SMTP_HOST/SMTP_USERNAME/SMTP_PASSWORD 缺少），跳過寄信。") return False # 正規化非斷行空格 html_body_clean = html_body.replace("\xa0", " ") msg = MIMEText(html_body_clean, "html", "utf-8", policy=SMTPUTF8) msg["Subject"] = subject msg["From"] = smtp_from msg["To"] = recipient_email try: # 使用 SMTP + STARTTLS 進行安全傳輸 with smtplib.SMTP(smtp_host, smtp_port, timeout=20) as server: server.set_debuglevel(0) server.ehlo() context = ssl.create_default_context() server.starttls(context=context) server.ehlo() server.login(smtp_username, smtp_password) # send_message 處理非 ascii 內容的正確編碼 server.send_message(msg) print(f"✅ [SMTP] 安全警報郵件已成功寄送至: {recipient_email}") return True except Exception as e: print(f"❌ [SMTP] 寄信失敗: {e}") return False def send_security_alert(recipient_email: str, verify_token: str, account_name: str = ""): """ 當偵測到異常時，發送包含兩個鏈接的郵件： - 這是我本人（信任） - 這不是我（封鎖） account_name: 觸發異常的帳戶名稱（用於顯示在信件中） 發送優先順序： 1. Resend（若 RESEND_API_KEY 已設定，適合 HF 部署） 2. SMTP（Resend 失敗或未設定時 fallback，適合本地開發） """ base_url = SMTP_CONFIG["base_url"].replace("\xa0", " ") recipient_email = recipient_email.replace("\xa0", " ") trust_url = f"{base_url}/verify_identity?token={quote(verify_token)}&choice=trust" block_url = f"{base_url}/verify_identity?token={quote(verify_token)}&choice=block" # 始終記錄鏈接以供調試/測試 print(f"🔗 [Security Alert] account={account_name}, trust_url={trust_url}") print(f"🔗 [Security Alert] account={account_name}, block_url={block_url}") print(f"📧 [Security Alert] 使用發送方式: {EMAIL_PROVIDER.upper()}") # 帳號顯示名稱（若有提供則加入說明） account_info = f"

⚠️ 警示帳號：{account_name}

" if account_name else "" subject = f"[數位保鏢] 安全提醒：帳號 {account_name} 偵測到異常操作" if account_name else "安全提醒：請確認是否為本人操作" html_body = f"""

🔐 數位保鏢安全警示通知

{account_info}

系統偵測到此帳號可能發生異常登入或操作行為，請確認是否為本人所為。

請於 5 分鐘內 點選以下其中一項進行確認：

✅ 這是我本人（繼續使用） 🚫 這不是我（立即封鎖）

若您未主動進行任何操作，請立即點選「這不是我（立即封鎖）」以保護帳號安全。逾時未確認將自動封鎖帳號。

""" html_body = html_body.replace("\xa0", " ") # 依照 EMAIL_PROVIDER 選擇發送方式，失敗時自動 fallback if EMAIL_PROVIDER == "resend": success = _send_via_resend(recipient_email, subject, html_body) if not success: print("🔄 [Fallback] Resend 失敗，嘗試使用 SMTP 備援..") success = _send_via_smtp(recipient_email, subject, html_body) else: success = _send_via_smtp(recipient_email, subject, html_body) if not success: print("🔄 [Fallback] SMTP 失敗，嘗試使用 Resend 備援..") success = _send_via_resend(recipient_email, subject, html_body) return success def trigger_identity_verification(account_key: str, recipient_email: str, anomaly_timestamp: str): """ 觸發身份驗證 account_key: 用於 session_cache 的識別 key（帳戶名稱） recipient_email: 安全警報信件的收件人（統一使用 FIXED_TEST_EMAIL） """ # 如果已經封鎖，保持更強的決定 entry = _get_or_create_session_entry(account_key) with session_cache_lock: if entry.get("status") == "blocked": return None, False verify_token = secrets.token_urlsafe(32) entry.update({ "status": "verifying", "pending_token": verify_token, "anomaly_timestamp": anomaly_timestamp, }) # 寄送信件時傳入帳戶名稱，讓收件人知道是哪個帳號觸發警示 success = send_security_alert(recipient_email, verify_token, account_name=account_key) 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) # 處理選擇 choice_norm = choice.lower().strip() active_choices = {"trust", "allow", "yes", "這是我本人（信任）"} blocked_choices = {"block", "deny", "blocked", "封鎖", "不是我", "這不是我（封鎖）"} if choice_norm in active_choices: next_status = "active" elif choice_norm in blocked_choices: next_status = "blocked" else: 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) entry["status"] = next_status # Token 是單次使用：決定後清除 entry["pending_token"] = None return { "ok": True, "user_email": user_email, "status": next_status, } def test_send_security_alert(): """ 測試助手： - 直接觸發為 FIXED_TEST_EMAIL 發送安全警報 - 返回生成的 verify token（token 也存儲在 session_cache 中） """ anomaly_timestamp = datetime.datetime.now(datetime.timezone(datetime.timedelta(hours=8))).strftime("%Y-%m-%d %H:%M:%S") # 測試時以 FIXED_TEST_EMAIL 同時作為 account_key 與收件人 token, success = trigger_identity_verification(FIXED_TEST_EMAIL, FIXED_TEST_EMAIL, anomaly_timestamp) if not token: return {"ok": False, "error": "Already blocked or session not available"} return {"ok": True, "user_email": FIXED_TEST_EMAIL, "token": token, "email_sent": success} def explain_abnormal_log(log_text): """使用 Groq AI 解釋異常日誌""" if not log_text or log_text.strip() == "": return "⚠️ 請先提供異常 Log 資訊以進行分析。" api_key = os.getenv("GROQ_API_KEY") if not api_key: return "⚠️ 未偵測到 GROQ_API_KEY，請在 .env 檔案中設定以啟用 AI 分析。" url = "https://api.groq.com/openai/v1/chat/completions" headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json" } prompt = f""" 你是一個資安分析助手。請分析以下這條異常 Log，並給出可能的攻擊類型、風險等級（低/中/高/極高）以及建議的處理動作。 請用繁體中文回答，並保持語氣專業且簡潔。 異常 Log: {log_text} """ payload = { "model": "llama-3.3-70b-versatile", "messages": [ {"role": "system", "content": "你是一位資深資安專家，專長於行為分析與入侵偵測。"}, {"role": "user", "content": prompt} ], "temperature": 0.5, "max_tokens": 1024 } try: response = requests.post(url, headers=headers, json=payload, timeout=15) response.raise_for_status() result = response.json() analysis = result['choices'][0]['message']['content'] return analysis except Exception as e: return f"❌ AI 分析請求失敗: {str(e)}"