refactor(p1-01d): routes/ 移除 safe_read_sql/validate_table_name/find_col 三份重複定義
All checks were successful
CD Pipeline / deploy (push) Successful in 1m6s

- routes/sales_routes.py 移除 find_col、validate_table_name、safe_read_sql 各自實作(-40 行)
- routes/daily_sales_routes.py 移除 validate_table_name、safe_read_sql 各自實作(-26 行)
- 兩檔改為 from utils.security import ... 的 re-export,行為對齊單一權威來源

注意:原本 routes 自己的 validate_table_name 較寬鬆(只 regex),
改用 utils.security 後升級為「白名單 + SQL 關鍵字」雙重防護。
所有 call site 都用 'realtime_sales_monthly' 或 'daily_sales_snapshot',皆在白名單內,行為相容。
This commit is contained in:
ooo
2026-04-28 15:50:21 +08:00
parent 17cb012be7
commit dea94d2e0f
2 changed files with 5 additions and 65 deletions

View File

@@ -74,34 +74,8 @@ def _is_cache_valid(cache_key):
# 輔助函數
# ==========================================
def validate_table_name(table_name):
"""驗證表名(防止 SQL Injection"""
import re
if not re.match(r'^[a-zA-Z_][a-zA-Z0-9_]*$', table_name):
raise ValueError(f"Invalid table name: {table_name}")
return table_name
def safe_read_sql(table_name, columns=None, engine=None, where_clause=None, limit=None, params=None):
"""安全的 SQL 查詢函數,防止 SQL Injection"""
table_name = validate_table_name(table_name)
if 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:
sys_log.error(f"[Security] SQL 查詢失敗: {e}")
raise
# 共用工具改 import 自 utils去重原本檔案內定義已移除
from utils.security import validate_table_name, safe_read_sql # noqa: E402, F401
def preprocess_daily_sales_data(df):

View File

@@ -38,43 +38,9 @@ _SALES_OPTIONS_CACHE = {}
# 輔助函數
# ==========================================
def find_col(df_cols, keywords):
"""從欄位列表中,根據關鍵字列表找出最匹配的欄位名稱"""
for k in keywords:
for col in df_cols:
if k in str(col):
return col
return None
def validate_table_name(table_name):
"""驗證表名(防止 SQL Injection"""
import re
if not re.match(r'^[a-zA-Z_][a-zA-Z0-9_]*$', table_name):
raise ValueError(f"Invalid table name: {table_name}")
return table_name
def safe_read_sql(table_name, columns=None, engine=None, where_clause=None, limit=None, params=None):
"""安全的 SQL 查詢函數,防止 SQL Injection"""
table_name = validate_table_name(table_name)
if 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:
sys_log.error(f"[Security] SQL 查詢失敗: {e}")
raise
# 共用工具改 import 自 utils去重原各 routes 各自定義已移除)
from utils.df_helpers import find_col # noqa: E402, F401
from utils.security import validate_table_name, safe_read_sql # noqa: E402, F401
def _get_filtered_sales_data(cache_key):