Merge pull request #701 from christianlouis/fix-db-migrate-sqli-17066903272609783485
🔒 Fix potential SQL injection in database migration preview
This commit is contained in:
@@ -12,6 +12,7 @@ The utility:
|
|||||||
"""
|
"""
|
||||||
|
|
||||||
import logging
|
import logging
|
||||||
|
import re
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
from sqlalchemy import MetaData, create_engine, inspect, text
|
from sqlalchemy import MetaData, create_engine, inspect, text
|
||||||
@@ -84,6 +85,9 @@ def preview_migration(source_url: str) -> dict[str, Any]:
|
|||||||
total = 0
|
total = 0
|
||||||
with src_engine.connect() as conn:
|
with src_engine.connect() as conn:
|
||||||
for table_name in tables:
|
for table_name in tables:
|
||||||
|
if not re.match(r"^[a-zA-Z0-9_]+$", table_name):
|
||||||
|
logger.warning(f"Skipping table with invalid name format: {table_name}")
|
||||||
|
continue
|
||||||
# table_name is safe — sourced from inspect().get_table_names(), not user input
|
# table_name is safe — sourced from inspect().get_table_names(), not user input
|
||||||
quoted_table = conn.dialect.identifier_preparer.quote(table_name)
|
quoted_table = conn.dialect.identifier_preparer.quote(table_name)
|
||||||
row = conn.execute(text(f"SELECT COUNT(*) FROM {quoted_table}")).fetchone() # noqa: S608
|
row = conn.execute(text(f"SELECT COUNT(*) FROM {quoted_table}")).fetchone() # noqa: S608
|
||||||
|
|||||||
Reference in New Issue
Block a user