Scott.Idem/OSIT-AE-API-FastAPI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(crud): strip view-only join columns from order_by_li to prevent ambiguous account_id in WHERE

Browse Source 
Sorting by join-derived columns (e.g. event_presenter_family_name on v_event_file)
caused MariaDB to expand the view's JOIN inline, making the unqualified account_id
clause from sql_and_qry_part ambiguous — resulting in a 400 SQL error. filter_order_by
now accepts raw_table_name and validates ORDER BY columns against the physical table
only; join-only columns are silently stripped. Also switches filter_order_by off the
global db connection to engine.connect() context managers. Updated all four call sites
in api_crud_v3.py and api_crud_v3_nested.py.

Docs: add order_by_li raw-table limitation and direct download link patterns to
GUIDE__AE_API_V3_for_Frontend.md; record fix in TODO__Agents.md.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Scott Idem
2026-06-10 13:33:09 -04:00
parent 22e5a3c3fd
commit 2429a1f731
5 changed files with 90 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
app/lib_api_crud_v3.py
View File

@@ -140,30 +140,37 @@ def apply_forced_account_filter(and_qry_dict: Optional[Dict], account: AccountCo
return forced
def filter_order_by(order_by_li: Any, model: Any, table_name: str = None) -> Optional[Dict[str, str]]:
def filter_order_by(order_by_li: Any, model: Any, table_name: str = None, raw_table_name: str = None) -> Optional[Dict[str, str]]:
"""
Sanitize Sorting Parameters.
Prevents SQL injection and logic errors by validating that requested sort columns
actually exist in the Pydantic model and/or the database table.
When raw_table_name is provided (the physical table, not the view), columns are
validated against it instead of table_name. This prevents view-only join columns
(e.g. event_presenter_family_name on v_event_file) from reaching ORDER BY —
those columns cause MariaDB to expand the view's JOIN inline, making unqualified
WHERE references like `account_id` ambiguous across joined tables.
"""
if not order_by_li or not isinstance(order_by_li, dict) or not model:
return order_by_li
if not hasattr(model, '__fields__'):
return order_by_li
model_fields = set(model.__fields__.keys())
model_fields.update({f.alias for f in model.__fields__.values() if f.alias})
filtered = {k: v for k, v in order_by_li.items() if k in model_fields}
if table_name and filtered:
from app.db_sql import db
check_table = raw_table_name or table_name
if check_table and filtered:
from app import lib_sql_core
from sqlalchemy import text
final_filtered = {}
for column in filtered:
try:
# Lightweight check to see if column exists in SQL
db.execute(text(f"SELECT `{column}` FROM `{table_name}` LIMIT 0"))
with lib_sql_core.engine.connect() as conn:
conn.execute(text(f"SELECT `{column}` FROM `{check_table}` LIMIT 0"))
final_filtered[column] = filtered[column]
except Exception:
pass

6
app/routers/api_crud_v3.py
View File

@@ -226,11 +226,12 @@ async def get_obj_li(
table_name = obj_cfg.get(f'tbl_{view}', obj_cfg.get('tbl_default', obj_cfg.get('tbl')))
base_name = obj_cfg.get(f'mdl_{view}', obj_cfg.get('mdl_default', obj_cfg.get('mdl')))
raw_table_name = obj_cfg.get('tbl_update', obj_cfg.get('tbl'))
if not table_name or not base_name:
return mk_resp(data=False, status_code=500, response=response, status_message=f"Configuration error.")
order_by_li = filter_order_by(order_by_li, base_name, table_name)
order_by_li = filter_order_by(order_by_li, base_name, table_name, raw_table_name=raw_table_name)
status_filter = get_supported_filters(base_name, status_filter)
if not obj_cfg.get('public_read', False):
@@ -335,11 +336,12 @@ async def search_obj_li(
table_name = obj_cfg.get(f'tbl_{view}', obj_cfg.get('tbl_default', obj_cfg.get('tbl')))
base_name = obj_cfg.get(f'mdl_{view}', obj_cfg.get('mdl_default', obj_cfg.get('mdl')))
raw_table_name = obj_cfg.get('tbl_update', obj_cfg.get('tbl'))
if not table_name or not base_name:
return mk_resp(data=False, status_code=500, response=response, status_message=f"Configuration error.")
order_by_li = filter_order_by(order_by_li, base_name, table_name)
order_by_li = filter_order_by(order_by_li, base_name, table_name, raw_table_name=raw_table_name)
status_filter = get_supported_filters(base_name, status_filter)
searchable_fields = obj_cfg.get('searchable_fields')

6
app/routers/api_crud_v3_nested.py
View File

@@ -84,6 +84,7 @@ async def get_child_obj_li(
obj_cfg = obj_type_kv_li[obj_name]
table_name = obj_cfg.get(f'tbl_{view}', obj_cfg.get('tbl_default', obj_cfg.get('tbl')))
base_name = obj_cfg.get(f'mdl_{view}', obj_cfg.get('mdl_default', obj_cfg.get('mdl')))
raw_table_name = obj_cfg.get('tbl_update', obj_cfg.get('tbl'))
# Log parent/child resolution details (use INFO so logs appear in production)
log.info("nested.list start parent=%s parent_table=%s parent_id_random=%s child=%s table=%s allowed_parents=%s", parent_obj_type, parent_table, parent_obj_id, obj_name, table_name, obj_cfg.get('parent_types'))
@@ -91,7 +92,7 @@ async def get_child_obj_li(
if not table_name or not base_name:
return mk_resp(data=False, status_code=500, response=response, status_message=f"Configuration error.")
order_by_li = filter_order_by(order_by_li, base_name, table_name)
order_by_li = filter_order_by(order_by_li, base_name, table_name, raw_table_name=raw_table_name)
status_filter = get_supported_filters(base_name, status_filter)
resolved_parent_id = redis_lookup_id_random(record_id_random=parent_obj_id, table_name=parent_table)
@@ -187,11 +188,12 @@ async def search_child_obj_li(
obj_cfg = obj_type_kv_li[obj_name]
table_name = obj_cfg.get(f'tbl_{view}', obj_cfg.get('tbl_default', obj_cfg.get('tbl')))
base_name = obj_cfg.get(f'mdl_{view}', obj_cfg.get('mdl_default', obj_cfg.get('mdl')))
raw_table_name = obj_cfg.get('tbl_update', obj_cfg.get('tbl'))
if not table_name or not base_name:
return mk_resp(data=False, status_code=500, response=response, status_message=f"Configuration error.")
order_by_li = filter_order_by(order_by_li, base_name, table_name)
order_by_li = filter_order_by(order_by_li, base_name, table_name, raw_table_name=raw_table_name)
status_filter = get_supported_filters(base_name, status_filter)
searchable_fields = obj_cfg.get('searchable_fields')