refactor(p1-01c): 整併 utils/security 與 utils/validators 重複實作
All checks were successful
CD Pipeline / deploy (push) Successful in 1m6s
All checks were successful
CD Pipeline / deploy (push) Successful in 1m6s
發現 utils/validators.py 已存在且完整重複 utils/security.py 的 9 個函數。 不收拾的話會繼續腐爛 — 立刻整併為單一權威來源。 變更: - utils/security.py 增加 safe_read_sql(取自 validators.py 較完整版本,含 limit + params) - utils/security.py ALLOWED_TABLES 取兩份聯集(補上 monthly_summary_analysis, realtime_sales_daily),避免破壞既有呼叫者 - utils/validators.py 改為純 re-export shim(保 from utils.validators import 不破) - app.py 移除原 safe_read_sql 重複定義(35 行),改 import utils.security routes/import_routes.py 不變(它 from utils.validators 走得到 re-export,等下輪統一)。 行數變化: app.py 7,187 → 7,151 (-36)
This commit is contained in:
@@ -1,12 +1,15 @@
|
||||
"""安全相關工具:SQL injection 防護、路徑遍歷防護、檔案上傳驗證。
|
||||
|
||||
從 app.py 抽出,純驗證邏輯,無 Flask 依賴。
|
||||
從 app.py + utils/validators.py 整併的單一權威來源。純驗證邏輯,無 Flask 依賴。
|
||||
舊的 utils/validators.py 已 deprecate,僅保留 re-export 不破壞既有 import。
|
||||
"""
|
||||
import os
|
||||
import re
|
||||
import unicodedata
|
||||
from pathlib import Path
|
||||
|
||||
import pandas as pd
|
||||
|
||||
from utils.logger_manager import SystemLogger
|
||||
|
||||
_log = SystemLogger("Security").get_logger()
|
||||
@@ -16,9 +19,12 @@ _log = SystemLogger("Security").get_logger()
|
||||
# SQL Injection 防護
|
||||
# ────────────────────────────────────────────────────────────────────────
|
||||
|
||||
# 整合 app.py 與 utils/validators.py 的兩份 ALLOWED_TABLES(取聯集,避免破壞既有呼叫者)
|
||||
ALLOWED_TABLES = {
|
||||
'realtime_sales_monthly',
|
||||
'realtime_sales_daily',
|
||||
'daily_sales_snapshot',
|
||||
'monthly_summary_analysis',
|
||||
'products',
|
||||
'price_records',
|
||||
'promo_products',
|
||||
@@ -62,6 +68,40 @@ def validate_column_names(column_names):
|
||||
return validated
|
||||
|
||||
|
||||
def safe_read_sql(table_name, columns=None, engine=None, where_clause=None, limit=None, params=None):
|
||||
"""安全的 SQL 查詢函數,防止 SQL Injection。
|
||||
|
||||
Args:
|
||||
table_name: 資料表名稱(必驗證白名單)
|
||||
columns: 欄位列表,None 表示 *
|
||||
engine: SQLAlchemy engine
|
||||
where_clause: WHERE 子句(呼叫端負責安全)
|
||||
limit: 限制筆數(自動轉 int)
|
||||
params: 參數化查詢的參數字典
|
||||
"""
|
||||
from sqlalchemy import text
|
||||
|
||||
table_name = validate_table_name(table_name)
|
||||
|
||||
if columns:
|
||||
columns = validate_column_names(columns)
|
||||
col_str = ', '.join([f'"{col}"' for col in columns])
|
||||
else:
|
||||
col_str = '*'
|
||||
|
||||
try:
|
||||
query = f'SELECT {col_str} FROM "{table_name}"'
|
||||
if where_clause:
|
||||
query += f' WHERE {where_clause}'
|
||||
if limit:
|
||||
query += f' LIMIT {int(limit)}'
|
||||
|
||||
return pd.read_sql(text(query), engine, params=params)
|
||||
except Exception as e:
|
||||
_log.error(f"[Security] SQL 查詢失敗: {e}")
|
||||
raise
|
||||
|
||||
|
||||
# ────────────────────────────────────────────────────────────────────────
|
||||
# 路徑遍歷防護
|
||||
# ────────────────────────────────────────────────────────────────────────
|
||||
|
||||
Reference in New Issue
Block a user