在前一篇測試中利用 Antigravity CLI (agy) 完成 serverless 函式執行平台的重構, 為其加上 API Key 認證機制, 所有請求必須在標頭中攜帶有效的 X-API-Key 屬性值, 否則請求將被拒絕, 使用 curl.exe 驗證新版 serverless 功能確實符合要求, 本篇則要改用 requests 套件來測試, 這是在實際應用中最常見用法.
5. 使用 requests 提出請求 :
使用 requests 套件發送請求時, 要將 API Key 放進 headers 參數中的 X-API-Key 欄位傳送給伺服器, 請求的 HTTP 方法可以使用 GET 或 POST, 分別呼叫 requests.get() 與 requests.post() 函式.
注意, 在前一篇測試中我們已將 API Key 更新如下 :
以下測試將使用 new-test-token 這個 API Key.
(1). 使用 GET 方法 :
GET 方法不帶 Body, 只需要單純將 API Key 放入 headers 中的 X-API-Key 欄位, 然後在呼叫 requests.get() 時傳入 headers 參數即可, 例如 :
# serverless_api_key_get_ok.py
import requests
url="http://localhost:5000/function/hello"
headers={"X-API-Key": "new-test-token"}
response=requests.get(url, headers=headers)
if response.status_code == 200:
print("回應:", response.text)
else:
print(f"錯誤 {response.status_code}:", response.text)
此處因為 hello.py 傳回值是純文字, 所以用 resposne.text 而非 response.json() 取得回應值, 執行結果如下, 因為傳送了有效的 API Key, 所以伺服器回應 200 OK :
>>> %Run serverless_api_key_get_ok.py
回應: Hello World!
如果沒有傳送有效的 API Key, 請求會被拒絕而得到 401 回應, 例如 :
# serverless_api_key_get_ng.py
import requests
url="http://localhost:5000/function/hello"
response=requests.get(url)
if response.status_code == 200:
print("回應:", response.text)
else:
print(f"錯誤 {response.status_code}:", response.text)
執行結果如下 :
>>> %Run serverless_api_key_get_ng.py
錯誤 401: {
"error": "Missing API token",
"hint": "Provide X-API-Key header"
}
(2). 使用 POST 方法 :
呼叫 requests.post() 時傳入有效的 API Key 就可順利取得 200 OK 回應, 例如 :
# serverless_api_key_post_ok.py
import requests
url="http://localhost:5000/function/hello"
headers={"X-API-Key": "new-test-token"}
response=requests.post(url, headers=headers)
if response.status_code == 200:
try:
# 先嘗試用 JSON 解析
print("回應 (JSON):", response.json())
except requests.exceptions.JSONDecodeError:
# 如果不是 JSON 就印出純文字
print("回應 (純文字):", response.text)
else:
print(f"錯誤 {response.status_code}:", response.text)
此例使用 try catch 來判斷回應物件型態是 JSON 格式字串還是純文字, 前者呼叫 response.json() 取得回應內容, 後者則用 response.text, 執行結果如下 :
>>> %Run serverless_api_key_post_ok.py
回應 (純文字): Hello World!
如果呼叫 requests.post() 時沒有傳入有效的 API Key 就會收到 401 錯誤, 例如 :
# serverless_api_key_post_ng.py
import requests
url="http://localhost:5000/function/hello"
response=requests.post(url)
if response.status_code == 200:
try:
# 先嘗試用 JSON 解析
print("回應 (JSON):", response.json())
except requests.exceptions.JSONDecodeError:
# 如果不是 JSON 就印出純文字
print("回應 (純文字):", response.text)
else:
print(f"錯誤 {response.status_code}:", response.text)
執行結果如下 :
>> %Run serverless_api_key_post_ng.py
錯誤 401: {
"error": "Missing API token",
"hint": "Provide X-API-Key header"
}
以上測試說明重構後的 serverless 平台確實已改為 API Key 版本, 任何請求必須攜帶 API Key 才可能取得 200 OK 回應, 否則請求都會被拒絕, 收到 401 錯誤.
6. 主程式原始碼異動摘要 :
經過上面測試, 我們確認了此重購專案已達成了預期目標, 以下做個 Code Review, 將主程式 serverless.py 的異動摘要如下 :
首先是在 init_db() 函式中添加了一個儲存 API Key 的認證資料表 api_tokens :
# API Token 認證表
cursor.execute("""
CREATE TABLE IF NOT EXISTS api_tokens (
id INTEGER PRIMARY KEY AUTOINCREMENT,
token TEXT NOT NULL UNIQUE,
owner TEXT NOT NULL,
is_active INTEGER NOT NULL DEFAULT 1,
created_at TEXT NOT NULL DEFAULT (datetime('now'))
)
""")
其次是添加了一個 TOKEN_PROTECTED_FUNCTIONS 清單 :
# 需要 API Token 驗證的函式列表（儲存/更新 function 及所有一般函式執行）
TOKEN_PROTECTED_FUNCTIONS=['save_function', 'update_function']
這兩個系統函式 save_function 與 update_function 用來將函式內容寫進檔案中, 屬於最高權限的敏感操作, 必須通過 API Key 驗證.
其三是新增了一個 require_api_token() 函式來檢查請求標頭中是否有攜帶有效之 API Key, 有就傳回所呼叫的函式, 否則傳回 401 錯誤 :
def require_api_token(f): # 裝飾器：驗證請求 Header 中的 X-API-Key
@wraps(f)
def decorated(*args, **kwargs):
token=request.headers.get('X-API-Key')
if not token: # Header 遺失
return jsonify({'error': 'Missing API token', 'hint': 'Provide X-API-Key header'}), 401
conn=sqlite3.connect(DB_PATH)
cursor=conn.cursor()
cursor.execute(
'SELECT id FROM api_tokens WHERE token=? AND is_active=1',
(token,)
)
row=cursor.fetchone()
conn.close()
if not row: # Token 不存在或已停用
return jsonify({'error': 'Invalid or inactive API token'}), 401
return f(*args, **kwargs)
return decorated
最後是在 handle_function() 函式中添加雙軌互補驗證 (Session 或 Token 擇一通過) 機制 :
# 2. API Token 驗證：已登入管理者直接放行；否則需提供有效 Token
# 觸發範圍：TOKEN_PROTECTED_FUNCTIONS（save/update）及所有一般函式
if func_name in TOKEN_PROTECTED_FUNCTIONS or func_name not in PROTECTED_FUNCTIONS:
if not check_auth(): # 管理者已登入 → 信任，跳過 Token 驗證
token=request.headers.get('X-API-Key')
if not token:
return jsonify({'error': 'Missing API token', 'hint': 'Provide X-API-Key header'}), 401
conn=sqlite3.connect(DB_PATH)
cursor=conn.cursor()
cursor.execute(
'SELECT id FROM api_tokens WHERE token=? AND is_active=1',
(token,)
)
row=cursor.fetchone()
conn.close()
if not row:
return jsonify({'error': 'Invalid or inactive API token'}), 401
這段程式碼採用了巢狀條件設計, 提供了兩道過濾網 :
- 第一層 (判定範圍) :只要呼叫的函式是敏感的系統功能 (save/update_function) 或任何使用者自訂函式 (不在白名單內) 就會觸發此區塊.
- 第二層 (身份切換) :
- 情境 A (管理員) :
瀏覽器帶著 Cookie 請求, check_auth() 回傳 True. 此時 if not check_auth() 判定不成立, 直接跳過整個 Token 檢查與資料庫查詢, 此安排讓管理員登入後線上操作函式更新與存檔不因為沒帶 API Key 而被 401 擋掉. - 情境 B (外部 API 呼叫/排程腳本) :
本機 Python 腳本或 curl 請求沒有瀏覽器 Session, 呼叫 check_auth() 會回傳 False, 程式進入 if 內部強制執行後續的 X-API-Key 擷取與 SQLite 比對.
if not check_auth() 是檢查使用者是否有登入, 沒有登入就要接受 API Key 查驗, 檢查 HTTP 請求的 Header 有沒有帶 X-API-Key 欄位, 沒有帶就直接拒絕請求, 回傳 401 Unauthorized 狀態碼. 如果已登入, 則程式會直接跳過整個 if not check_auth(): 內部的所有程式碼, 直接放行去執行該函式.
注意此處使用了 SQLite 的參數化查詢 token=? 並把變數包在元組 (token,) 裡傳進去, 這樣可以徹底杜絕駭客在 Header 亂寫 SQL 語法, 利用 SQL Injection 來破解驗證. 如果資料庫回傳 None, 表示 API Key 不存在, 或存在但被停用 (is_active 不是 1), 那就回傳 401 錯誤拒絕請求; 反之就是通過了驗證, 准許呼叫此函式, 放行讓後續的業務邏輯繼續執行.
沒有留言 :
張貼留言