2025年8月6日 星期三

Mapleboard MP510-50 測試 (三十八) : HTTP API 函式執行平台 (4)

今天在評估利用 serverless 函式執行平台作為 LINE Bot 的 webhook 的後台時, 發現一個擴展性問題, 在主程式目錄 ~/flask_apps/serverless 下有一個環境變數檔 .env 目前只用來儲存管理功能所需之 SECRET 與 TOKEN, 將來其他函式模組可能也會用到別的網路服務的金鑰或權杖, 這當然可以在模組目錄 ~/flask_apps/serverless/functions 下另建一個 .env 檔, 但我覺得整個系統的金鑰或權杖應該集中在主程式目錄下統一管理才對, 這必須修改主程式 serverless 第六步呼叫被載入模組的 main() 函式時的傳入參數, 從原本只傳入 request 參數改為增加傳入 config 參數 :

       # 6. 執行模組中的函式 :  (原來做法)      
        result=module.main(request)  # 將 request 傳給被載入模組由其自行處理參數

改成如下 :

       # 6. 執行模組中的函式 :  (新的做法)       
        result=module.main(request, config=config)  # 加入關鍵字參數 config (以後可擴充)

那麼函式模組的基本架構中, main() 函式也要添加一個 **kwargs 參數接收所有關鍵字參數 (即使此函式用不到任何關鍵字參數也是要接收) :

# func_module.py
def main(request, **kwargs):
    config=kwargs.get('config', {})  # 預設為空 dict
    result='do something'
    return result

kwargs 為一個鍵為關鍵字的字典, 因此可以用 kwargs.get() 取出指定鍵之參數, 例如目前只有 config 一個關鍵字參數, 則可用 kwargs.get('config', {}) 取出 config 參數. 

本系列全部測試文章參考 :



1. 主程式 serverless.py :

首先修改主程式 serverless.py (只需修改第六步那一行) : 

# serverless.py
from flask import Flask, request, jsonify, session  
import importlib.util
import os
import logging
from dotenv import dotenv_values  

def check_auth():  # 檢查使用者是否已登入
    return session.get('authenticated') == True  

app=Flask(__name__)
# 從 .env 讀取權杖 (密碼) 與金鑰
config=dotenv_values('.env')
SECRET_TOKEN=config.get('SECRET_TOKEN')  # 易記的令牌 (類似密碼)
SECRET_KEY=config.get('SECRET_KEY')  # 簽章加密用的金鑰
app.secret_key=SECRET_KEY  # 用來簽章與驗證 session cookie
# 指定函式模組所在的資料夾
FUNCTIONS_DIR=os.path.expanduser('./functions')
# 指定錯誤日誌檔 (在目前工作目錄下)
logging.basicConfig(filename='serverless_error.log', level=logging.ERROR)
# 需要驗證的函式列表
PROTECTED_FUNCTIONS=['list_functions',
                     'add_function',
                     'save_function',
                     'edit_function',
                     'update_function',
                     'delete_function'
                     ]
# 登入管理功能
@app.route('/login', methods=['GET', 'POST'])
def login():
    # GET 請求 : 顯示登入頁面
    if request.method == 'GET':  
        return '''
        <!DOCTYPE html>
        <html>
        <head><title>系統登入</title></head>
        <body>
            <h2>系統登入</h2>
            <form method="post">
                <input type="password" name="token" placeholder="請輸入密碼" required>
                <button type="submit">登入</button>
            </form>
        </body>
        </html>
        '''
    # POST 請求 : 處理登入請求
    if request.is_json:  # JSON 登入 
        token=request.json.get('token')
    else:   # 表單登入
        token=request.form.get('token')    
    if token == SECRET_TOKEN:  # 驗證登入密碼
        # 將登入狀態儲存在瀏覽器的 session cookie 中 (以明碼方式儲存)
        # Flask 會用金鑰對資料進行簽章 (非加密) 確保內容未被竄改
        session['authenticated']=True    
        # 回應登入成功
        if request.is_json:  
            return jsonify({'message': '登入成功'})
        else:
            return '<p>登入成功！<a href="/function/list_functions">查看函式列表</a></p>'
    else:  # 密碼錯誤 : 回應登入失敗訊息
        if request.is_json:
            return jsonify({'message': '登入失敗'}), 401
        else:
            return '<p>登入失敗！<a href="/login">重新登入</a></p>', 401

# 登出管理功能 
@app.route('/logout')
def logout():
    session.clear()  # 清除伺服端 Flask session 字典中的所有鍵值對
    return '<p>已登出！<a href="/login">重新登入</a></p>'

# 動態載入 & 執行函式模組 (支援 RESTful) 
@app.route('/function/<func_name>', defaults={'subpath': ''}, methods=['GET', 'POST'])
@app.route('/function/<func_name>/<path:subpath>', methods=['GET', 'POST'])
def handle_function(func_name, subpath):  # 傳入 subpath 支援 RESTful
    # 1. 檢查是否為保護的函式模組且使用者已登入
    if func_name in PROTECTED_FUNCTIONS and not check_auth():
        return jsonify({'error': 'Authentication required', 'login_url': '/login'}), 401    
    # 2. 取得檔案路徑
    func_path=os.path.join(FUNCTIONS_DIR, f'{func_name}.py')
    if not os.path.isfile(func_path):  # 模組檔案不存在 -> 回 404
        return jsonify({'error': f'Function "{func_name}" not found'}), 404
    try:
        # 3. 動態載入模組 (絕對路徑)
        spec=importlib.util.spec_from_file_location(func_name, func_path)
        module=importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)
        # 4. 檢查模組中有無 main() 函式 :
        if not hasattr(module, 'main'):  # 模組中無 main() 函式
            return jsonify({'error': f'Module "{func_name}" has no main()'}), 400
        # 5. 將 subpath 加入 request 中 (支援 RESTful)
        request.view_args['subpath']=subpath  
        # 6. 執行模組中的函式 :        
        result=module.main(request, config=config)  # 將 request, config 傳給被載入模組由其自行處理參數
        # 7. 傳回函式執行結果
        return result 
    except Exception as e:
        logging.exception(f'Error in function {func_name}\n{e}')  # 紀錄錯誤於日誌
        return jsonify({'error': 'Function execution failed'}), 500

if __name__ == '__main__':
    app.run(debug=True)


2. 管理功能的函式模組 (位於 functions 子目錄下) :

共有六支函式模組, 都只是在 def main() 中添加 **kwargs :


(1). 列出全部模組兼主控台 : 

# list_functions.py
import os

def main(request, **kwargs):
    # 取得 functions 目錄絕對路徑
    functions_dir='./functions'
    # 取得所有 .py 檔案（但排除 __init__.py）
    try:
        files=os.listdir(functions_dir)
        py_files=[f[:-3] for f in files if f.endswith('.py') and f != '__init__.py']
        py_files.sort()  # 按字母順序排序
    except FileNotFoundError:
        return '<p>directory ./functions not found!</p>'
    # 產生 HTML 碼
    html = '<h2>函式列表</h2>'
    html += '<table border="1" cellspacing="0" cellpadding="6" style="border-collapse: collapse;">'
    html += '<tr><th>函式名稱</th><th>執行</th><th>編輯</th><th>刪除</th></tr>'
    PROTECTED_FUNCTIONS=['list_functions',
                     'add_function',
                     'save_function',
                     'edit_function',
                     'update_function',
                     'delete_function']
    for func in py_files:
        if func in PROTECTED_FUNCTIONS:
            continue
        html += f'<tr>'
        html += f'<td>{func}</td>'
        html += f'<td><a href="/function/{func}">執行</a></td>'
        html += f'<td><a href="/function/edit_function?module_name={func}">編輯</a></td>'
        html += f'<td><a href="/function/delete_function?module_name={func}">刪除</a></td>'
        html += f'</tr>'
    html += '</table>'
    html += '<br><a href="/function/add_function">新增函式</a> '
    html += '<a href="/logout">登出</a>'
    return html


(2). 新增函式模組 : 

# add_function.py
def main(request, **kwargs):
    html='''
    <h2>新增函式模組</h2>
    <form action="/function/save_function" method="post">
        <label>模組名稱（請用英數字，不含副檔名 .py）:</label><br>
        <input type="text" name="module_name" size="50" required><br><br>
        <label>模組內容（請輸入合語法的 Python 程式碼）:</label><br>
        <textarea id="code" name="code" rows="20" cols="100" required></textarea><br><br>
        <button onclick="location.href='/function/list_functions'">取消</button>
        <button type="button" onclick="document.getElementById('code').value = '';">清除</button>
        <input type="submit" value="存檔">
    </form>
    '''
    return html

# save_function.py
import os

def main(request, **kwargs):
    # 僅接受 POST 請求
    if request.method != 'POST':
        return '<p>只接受 POST 請求</p>'
    module_name=request.form.get('module_name', '').strip()
    code=request.form.get('code', '').strip()
    # 檢查輸入合法性
    if not module_name.isidentifier():
        return '<p>錯誤：模組名稱須為合法的 Python 識別字</p>'
    if '/' in module_name or '\\' in module_name or '..' in module_name:
        return '錯誤：無效的函式名稱'  # 避免目錄穿越攻擊(只允許純檔名, 無 / 或 ..)      
    if not code:
        return '<p>錯誤：模組內容不得為空</p>'
    # 組成檔案路徑
    filename=f'./functions/{module_name}.py'
    if os.path.exists(filename):
        return f'<p>錯誤：模組 <b>{module_name}.py</b> 已存在</p>'
    try:
        with open(filename, 'w', encoding='utf-8') as f:
            f.write(code)
    except Exception as e:
        return f'<p>儲存失敗：{e}</p>'
    return f'''
    <p>模組 <b>{module_name}.py</b> 已成功建立</p>
    <a href="/function/list_functions">返回函式列表</a>
    '''


(3). 編輯函式模組 : 

# edit_function.py
# 模組名稱也可更改 -> 相當於新增模組
from flask import render_template_string
import os

def main(request, **kwargs):
    module_name=request.args.get('module_name', '')  # 取得模組名稱
    if not module_name:  # 檢查有無傳入模組名稱
        return '請指定要編輯的模組名稱，格式 ?module_name=hello'
    filename=f'./functions/{module_name}.py'
    if not os.path.isfile(filename):
        return f'找不到函式檔案：{module_name}'
    with open(filename, 'r', encoding='utf-8') as f: 
        content=f.read()  # 讀取模組內容
    html=f'''
    <h2>編輯函式模組：/functions/{module_name}.py</h2>
    <form method="POST" action="/function/update_function">
        <label>模組名稱（請用英數字，不含副檔名 .py）:</label><br>    
        <input type="text" name="module_name" size="50" value="{module_name}"><br><br>
        <label>模組內容（請輸入合語法的 Python 程式碼）:</label><br>        
        <textarea id="code" name="code" rows="20" cols="100" style="font-family:monospace;">{content}</textarea><br><br>
        <button type="button" onclick="location.href='/function/list_functions'">取消</button>
        <button type="button" onclick="document.getElementById('code').value = '';">清除</button>
        <button type="submit">更新</button>
    </form>
    '''
    return render_template_string(html)

# update_function.py
import os

def main(request, **kwargs):
    if request.method != 'POST':
        return '請使用 POST 方法'
    module_name=request.form.get('module_name', '').strip()
    code=request.form.get('code', '').strip()
    # 檢查檔案名稱合法性
    if not module_name.isidentifier():
        return '<p>錯誤：模組名稱須為合法的 Python 識別字</p>'
    if '/' in module_name or '\\' in module_name or '..' in module_name:
        return '錯誤：無效的函式名稱'  # 避免目錄穿越攻擊(只允許純檔名, 無 / 或 ..)      
    if not code:
        return '<p>錯誤：模組內容不得為空</p>'
    # 組成檔案路徑
    filename=f'./functions/{module_name}.py'
    try:
        with open(filename, 'w', encoding='utf-8') as f:
            f.write(code)  # 若檔名一樣覆蓋原內容, 否則建立新模組
        return f'模組 {module_name} 已成功更新!<br><br><a href="/function/list_functions">返回函式列表</a>'
    except Exception as e:
        return f'更新失敗：{e}'


(4). 刪除函式模組 : 

# delete_function.py
import os

def main(request, **kwargs):
    module_name=request.args.get('module_name', '').strip()
    # 過濾模組名稱
    if not module_name:
        return '<p>錯誤：未提供模組名稱</p>'
    if not module_name.isidentifier():
        return '<p>錯誤：模組名稱須為合法的 Python 識別字</p>'    
    if '/' in module_name or '\\' in module_name or '..' in module_name:
        return '錯誤：無效的函式名稱'  # 避免目錄穿越攻擊(只允許純檔名, 無 / 或 ..)  
    filename=f'./functions/{module_name}.py'
    if not os.path.exists(filename):
        return f'<p>錯誤：模組 {module_name} 不存在</p>'
    try:
        os.remove(filename)  # 刪除檔案
        return f'''
        <p>已成功刪除模組：{module_name}</p>
        <a href="/function/list_functions">返回函式列表</a>
        '''
    except Exception as e:
        return f'<p>刪除失敗：{e}</p>'

其他測試模組例如 hello.py, hello2.py, 以及 add.py 也是要在 main() 添加 **kwargs 參數. 

由於主程式 serverless.py 有變動, 因此必須重啟服務才會生效 :

tony1966@LX2438:~/flask_apps/serverless$ sudo systemctl restart serverless   


3. 測試模組 test.py :

經過上面的結構翻修後, 我新增了一個測試模組來驗證是否可在函式模組中取得主程式從環境變數檔案 .env 讀取後透過 config 參數傳遞的權杖與金鑰資訊. 

用 nano 編輯 .env : 

tony1966@LX2438:~/flask_apps/serverless$ sudo nano .env   

在管理功能金鑰 SECRET 與權杖 TOKEN 下方添加 LINE Mesaging API 的金鑰 LINE_CHANNEL_SECRET 與權杖 LINE_CHANNEL_ACCESS_TOKEN :




按 Ctrl+O 存檔再按 Ctrl+X 跳出 nano 後, 必須重啟服務讓主程式重新載入 .env :

tony1966@LX2438:~/flask_apps/serverless$ sudo systemctl restart serverless   

然後用管理功能新增一個 test.py 模組 :

# test.py

def main(request, **kwargs):
    # 從 .env 讀取金鑰與權杖
    config=kwargs.get('config', {})  # 預設為空 dict 
    SECRET=config.get('LINE_CHANNEL_SECRET')  
    TOKEN=config.get('LINE_CHANNEL_ACCESS_TOKEN')
    return {"secret": SECRET, "token": TOKEN, "status": "OK"}

訪問 https://flask.tony1966.cc/function/test 顯示確實可讀取主程式傳遞的參數 :




當然測試完畢就已刪除 test.py 啦! 

以上新版 (v2) serverless 平台原始碼放在 Github :


解壓縮後需先修改 .env 環境變數檔, 設定平台管理 TOKEN 與 SECRET, 如果要開發 LINE Bot 的應用, 則要設定 LINE_CHANNEL_SECRET 與 LINE_CHANNEL_ACCESS_TOKEN 變數.  

張貼者： 於
標籤： , , , ,

沒有留言 :

張貼留言

訂閱： 張貼留言 ( Atom )