大家好！歡迎來到「Build on Google AI」工程挑戰的第 27 天。

昨天我們風光地完成了 Telegram 的「一鍵核准」機制，PM 在高鐵上點擊按鈕，系統就會自動把貼文發布出去。這個功能看起來非常酷，直到你們公司的資安長 (CISO) 走到你背後，冷冷地問了一句：

「如果競爭對手搜到這隻 Telegram 機器人，是不是也能幫我們『核准』一堆亂七八糟的假消息發布到粉專？」

不僅如此，如果你把 Day 21 寫好的 FastAPI 戰情室端點直接推上雲端，等於是把公司內部的 AI 產線赤裸裸地暴露在公網上。在企業級 SaaS 開發中，「沒有身分驗證的自動化，就是一顆隨時會引爆的炸彈」。

今天，我們要為這台強大的 AI 跑車裝上防盜鎖。我們將實作雙重資安防護：Telegram 主管白名單 與 Web 戰情室登入驗證。

第一步：實作 Telegram 權限白名單 (TG Whitelist)
在 Telegram Bot 的世界裡，只要有人知道你的 Bot Username，就能發送訊息給它。當我們發送帶有 Inline Keyboard 的審核卡片時，群組裡的「任何人」都可以點擊那顆按鈕。

為了解決這個問題，我們必須在 callback_query_handler 中，攔截點擊者的 user.id，並與我們的「授權主管名單」進行比對。如果不在名單內，就直接拒絕執行並彈出警告。

第二步：為 ADK 戰情室加上 HTTP Basic Auth
在開發階段我們使用 adk web，但在邁向 SaaS 化時，我們通常會用 FastAPI 來封裝 ADK 工作流（如同 Day 21 所述），以便提供客製化的戰情室儀表板或 API。

我們可以使用 FastAPI 內建的 HTTPBasic 與 Depends 依賴注入機制，強制要求存取戰情室 API 的使用者必須輸入帳號密碼。

第三步：完整實作程式碼 (main.py)
以下是結合了 ADK 狀態恢復 (Resume)、Telegram 白名單攔截 以及 FastAPI 登入驗證 的完整企業級安全實作。

import os
import secrets
import logging
import telebot
from telebot.types import InlineKeyboardMarkup, InlineKeyboardButton
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBasic, HTTPBasicCredentials
from pydantic import BaseModel
from google.adk import Agent, Workflow

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(message)s')
logger = logging.getLogger(__name__)

# ==========================================
# [資安設定] 環境變數與白名單
# ==========================================
# 1. Telegram 設定
TELEGRAM_BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN", "你的_BOT_TOKEN")
TELEGRAM_CHAT_ID = os.getenv("TELEGRAM_CHAT_ID", "你的_CHAT_ID")
# 【資安關鍵】：只有這個 List 裡面的 User ID 點擊按鈕才有效！
ALLOWED_TG_USERS = [123456789, 987654321] 
bot = telebot.TeleBot(TELEGRAM_BOT_TOKEN)

# 2. Web API 登入設定
SECURITY_USERNAME = os.getenv("ADMIN_USER", "admin")
SECURITY_PASSWORD = os.getenv("ADMIN_PASS", "supersecret")
security = HTTPBasic()

def verify_credentials(credentials: HTTPBasicCredentials = Depends(security)):
    """FastAPI 的 Basic Auth 依賴注入，驗證帳號密碼"""
    correct_username = secrets.compare_digest(credentials.username, SECURITY_USERNAME)
    correct_password = secrets.compare_digest(credentials.password, SECURITY_PASSWORD)
    if not (correct_username and correct_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="無效的帳號或密碼",
            headers={"WWW-Authenticate": "Basic"},
        )
    return credentials.username

# ==========================================
# [ADK 核心] 具備中斷點的 Workflow (精簡版)
# ==========================================
class CampaignState(BaseModel):
    marketing_copy: str = ""
    is_approved_by_human: bool = False

marketing_agent = Agent(name="marketing_agent", instruction="寫一句 AI 課程推廣文案", state_mapping={"outputs": ["marketing_copy"]})

def publish_node(state: CampaignState) -> CampaignState:
    if state.is_approved_by_human:
        logger.info(f"🚀 【正式發布】內容：{state.marketing_copy}")
    return state

workflow = Workflow(name="Secure_Approval_Workflow", state_schema=CampaignState)
workflow.add_node("marketing_agent", marketing_agent)
workflow.add_node("publish_node", publish_node)
workflow.set_entry_point("marketing_agent")
workflow.add_edge("marketing_agent", "publish_node")

# 進入發布節點前強制暫停
app_workflow = workflow.compile(interrupt_before=["publish_node"])
thread_config = {"configurable": {"thread_id": "secure_campaign_001"}}

# ==========================================
# Telegram 安全回呼處理 (TG Whitelist)
# ==========================================
def request_human_approval(current_copy: str):
    markup = InlineKeyboardMarkup()
    markup.add(InlineKeyboardButton("👍 核准發布", callback_data="approve"), InlineKeyboardButton("👎 退回", callback_data="reject"))
    bot.send_message(TELEGRAM_CHAT_ID, f"🛑 *需要授權*\n\n📝 內容：_{current_copy}_", parse_mode="Markdown", reply_markup=markup)

@bot.callback_query_handler(func=lambda call: True)
def handle_approval_callback(call):
    """攔截按鈕點擊，並執行白名單驗證"""
    clicker_id = call.from_user.id
    clicker_name = call.from_user.first_name
    
    # 【資安關鍵】：攔截未授權的使用者
    if clicker_id not in ALLOWED_TG_USERS:
        logger.warning(f"🚨 資安警報：未授權使用者 {clicker_name} (ID: {clicker_id}) 嘗試點擊核准按鈕！")
        # 彈出 Toast 警告點擊者，並且【不】喚醒 ADK
        bot.answer_callback_query(call.id, "❌ 存取被拒：您不在授權主管白名單內！", show_alert=True)
        return

    # 若驗證通過，執行授權與喚醒流程
    is_approved = (call.data == "approve")
    bot.answer_callback_query(call.id, "✅ 授權成功！系統處理中...")
    bot.edit_message_text(f"{call.message.text}\n\n✅ *{clicker_name} 已完成決策*", call.message.chat.id, call.message.message_id, parse_mode="Markdown")
    
    # 覆寫狀態並恢復 ADK 執行
    app_workflow.update_state(thread_config, {"is_approved_by_human": is_approved})
    app_workflow.run(None, config=thread_config)

# ==========================================
# FastAPI 戰情室端點 (Basic Auth 防護)
# ==========================================
app = FastAPI(title="Secure ADK War Room")

@app.post("/api/trigger-campaign")
async def trigger_campaign(username: str = Depends(verify_credentials)):
    """
    這個 API 端點受到 Basic Auth 保護。
    只有輸入正確帳號密碼的使用者，才能觸發行銷 Agent 產生草稿並傳送 TG 通知。
    """
    logger.info(f"👤 授權使用者 {username} 啟動了工作流...")
    initial_state = CampaignState()
    
    # 執行到中斷點暫停
    frozen_state = app_workflow.run(initial_state, config=thread_config)
    pending_nodes = app_workflow.get_state(thread_config).next
    
    if "publish_node" in pending_nodes:
        request_human_approval(frozen_state.marketing_copy)
        
    return {"status": "success", "message": "草稿已生成，並將 Telegram 審核權限傳送給白名單主管。"}

# 若要在本地執行測試，請使用指令： uvicorn main:app --reload --port 8080
# 並在另一個終端機執行 bot.infinity_polling() 以監聽 Telegram 回呼

驚人的防護體驗！
戰情室端點保護： 如果有駭客試圖用 Postman 呼叫 POST /api/trigger-campaign，只要他沒有提供 admin / supersecret 的 Basic Auth 標頭，系統會立刻回傳 401 Unauthorized，連 Agent 都不會被喚醒，有效防止惡意消耗 API 額度。

Telegram 權限攔截： 當主管將審核訊息轉發給其他同事，或是無關的人在群組裡點擊了「👍 核准發布」，手機畫面上會立刻彈出一個紅色的警告框：「❌ 存取被拒：您不在授權主管白名單內！」。後端的 ADK 工作流會繼續乖乖保持凍結狀態，滴水不漏！

小結
「信任機器的效率，但永遠驗證人類的權限。」

今天，我們利用 FastAPI 的 Depends 依賴注入與 Telegram 的 from_user.id 攔截，為這套強大的人機協作系統裝上了企業級的資安防護網。我們的 Agent 產線不僅聰明、穩定，現在更是絕對安全。