實戰範例 008：安全模式下的多帳號憑證本地儲存與密鑰輪替系統

簡介與場景

強大的 AI Agent 通常需要以使用者的身分登入多個社群或服務平台（例如：GitHub、Twitter、Notion 等），這意味著 Agent 必須在本地保存大量的 Access Token 甚至是帳號密碼。 如果不加以保護，將這些憑證直接明文 (Plaintext) 寫入 .json 檔案中，一旦伺服器或本地機器遭到惡意讀取，後果不堪設想。 本範例將帶領讀者實作一個輕量級的安全憑證儲存庫 (Secure Credential Vault)。我們將使用 AES-256-GCM 進行對稱加密，並且整合密鑰輪替 (Rotation) 的邏輯，讓 Agent 在存取這些敏感資訊時也能具備極高的安全性與容錯性。

原始代碼：極度危險的明文存儲

許多開發者為了貪圖方便，會直接把 Token 以 JSON 格式存入硬碟。這完全忽視了資安風險。

import json
import os

CRED_FILE = "credentials.json"

def save_token(platform, token):
    if os.path.exists(CRED_FILE):
        with open(CRED_FILE, 'r') as f:
            creds = json.load(f)
    else:
        creds = {}
        
    creds[platform] = token
    # 致命危險：明文保存敏感資料
    with open(CRED_FILE, 'w') as f:
        json.dump(creds, f)

def get_token(platform):
    with open(CRED_FILE, 'r') as f:
        creds = json.load(f)
    return creds.get(platform)

save_token("github", "ghp_SuperSecretToken12345")

開發者與 AI 的對話記錄

Ponytail (極簡主義者): 「用 Python 內建的 json 模組讀寫字典，兩三行解決，簡單明瞭。我把它放在 .gitignore 裡就不會洩漏啦。」

Agent Reach (容錯專家): 「.gitignore 只能防止推送到版本庫。如果你這段代碼跑在公有雲或者被其他惡意腳本讀取到同目錄下的檔案，所有權限就會被瞬間竊取。我們必須進行加密 (Encryption at Rest)。」

Ponytail: 「那我用 Base64 encode 一下？」

Agent Reach: 「Base64 只是編碼，不是加密！我們需要使用標準的 AES 加密算法。此外，Token 會過期，當 API 返回 401 Unauthorized 時，我們的 Agent 必須有能力攔截錯誤並觸發輪替 (Rotation) 程序，自動更新儲存庫中的加密憑證。這才是真正的安全與容錯一體化。」

重構/優化後的代碼：AES-256-GCM 加密與輪替保護

重構後我們引入了 cryptography 套件來執行帶有認證標籤 (Auth Tag) 的 AES-GCM 加密。讀寫檔案時都必須透過環境變數注入的 Master Key 解密。

import json
import os
import base64
import logging
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend

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

CRED_FILE = "secure_vault.enc"
# 實務中，MASTER_KEY 應該從 AWS KMS 或系統安全環境變數注入 (32 bytes)
MASTER_KEY = os.getenv("AGENT_MASTER_KEY", "12345678901234567890123456789012").encode('utf-8')

class SecureVault:
    @staticmethod
    def _encrypt(plaintext: str) -> bytes:
        iv = os.urandom(12) # GCM 標準推薦 12 bytes IV
        encryptor = Cipher(
            algorithms.AES(MASTER_KEY),
            modes.GCM(iv),
            backend=default_backend()
        ).encryptor()
        
        ciphertext = encryptor.update(plaintext.encode('utf-8')) + encryptor.finalize()
        
        # 將 IV, Auth Tag, Ciphertext 組合並 Base64 編碼
        return base64.b64encode(iv + encryptor.tag + ciphertext)

    @staticmethod
    def _decrypt(encrypted_data: bytes) -> str:
        try:
            data = base64.b64decode(encrypted_data)
            iv = data[:12]
            tag = data[12:28]
            ciphertext = data[28:]
            
            decryptor = Cipher(
                algorithms.AES(MASTER_KEY),
                modes.GCM(iv, tag),
                backend=default_backend()
            ).decryptor()
            
            return (decryptor.update(ciphertext) + decryptor.finalize()).decode('utf-8')
        except Exception as e:
            logger.error("解密失敗，金鑰錯誤或檔案已損毀！")
            raise

    @classmethod
    def save(cls, platform: str, token: str, refresh_token: str = None):
        """將憑證加密並安全寫入"""
        vault = {}
        if os.path.exists(CRED_FILE):
            with open(CRED_FILE, 'rb') as f:
                vault = json.loads(cls._decrypt(f.read()))
                
        vault[platform] = {"token": token, "refresh_token": refresh_token}
        
        with open(CRED_FILE, 'wb') as f:
            f.write(cls._encrypt(json.dumps(vault)))
        logger.info(f"[SECURE] 平台 {platform} 的憑證已加密儲存。")

    @classmethod
    def get(cls, platform: str) -> dict:
        if not os.path.exists(CRED_FILE):
            return None
        with open(CRED_FILE, 'rb') as f:
            vault = json.loads(cls._decrypt(f.read()))
        return vault.get(platform)

# 模擬 Agent 的 API 客戶端
class AuthenticatedAgent:
    def __init__(self, platform: str):
        self.platform = platform

    def _refresh_credentials(self) -> str:
        """模擬憑證輪替 (Rotation) 的過程"""
        logger.info(f"正在為 {self.platform} 執行憑證輪替...")
        creds = SecureVault.get(self.platform)
        if not creds or not creds.get('refresh_token'):
            raise Exception("無有效的 Refresh Token，需人工介入重新登入。")
            
        # 呼叫平台 API 更新 Token (此處為模擬)
        new_token = "ghp_NEW_SECURE_TOKEN_888"
        SecureVault.save(self.platform, new_token, creds['refresh_token'])
        return new_token

    def make_api_request(self):
        """帶有憑證過期容錯機制的請求"""
        creds = SecureVault.get(self.platform)
        if not creds:
            logger.error("未找到憑證，請先登入。")
            return

        token = creds['token']
        logger.info(f"使用 Token 嘗試請求 API... (Token 前綴: {token[:4]}...)")
        
        # 模擬 401 Unauthorized (Token 過期)
        api_status = 401 
        
        if api_status == 401:
            logger.warning("API 回傳 401 Unauthorized，觸發密鑰輪替。")
            new_token = self._refresh_credentials()
            logger.info(f"輪替完成，使用新 Token 重新請求... (Token 前綴: {new_token[:4]}...)")
            # 重新請求邏輯...

if __name__ == "__main__":
    # 初始儲存
    SecureVault.save("github", "ghp_EXPIRED_TOKEN_123", "rft_SECRET_REFRESH")
    
    # Agent 開始工作
    agent = AuthenticatedAgent("github")
    agent.make_api_request()

效益分析表格與解讀

解讀： 安全性與容錯性是一體兩面的。Agent Reach 不僅要求連線不要斷，更要求「帶有敏感資訊的連線」絕對安全。 透過極簡的 SecureVault 類別，我們隱藏了複雜的加解密細節（符合 Ponytail 極簡觀念），卻在系統底層加上了堅不可摧的防護罩。配合 API 請求時的 401 攔截與輪替機制，完美展示了一個成熟的 AI Agent 該如何優雅地處理身分驗證生命週期。