[Medium] gemc-sqlite CLI builds queries via raw f-strings (breakage + SQL injection)

[zhaozhiwen]

The gemc-sqlite CLI assembles WHERE clauses and the SELECT column list with raw f-string interpolation of user-supplied values. Any value containing a quote breaks the query, and the values are a SQL injection vector.

Affected files

• subprojects/pygemc/src/pygemc/api/gsqlite.py:61-91 (main, filter construction)
• subprojects/pygemc/src/pygemc/api/gsqlite.py:124-141 (show_volumes_from_database, show_materials_from_database)

Details

if args.ef:
	experiment_filter = f" WHERE experiment = '{args.ef}'"
...
if args.rf:
	...
	runno_filter = f' WHERE run = {args.rf}'

all_filters = experiment_filter + variation_filter + system_filter + runno_filter
...
show_volumes_from_database(sqlitedb, what, all_filters)

and:

def show_volumes_from_database(sqlitedb, what, all_filters):
	...
	query = "SELECT {} FROM geometry {};".format(what, all_filters)
	sql.execute(query)

The filter values (args.ef, args.vf, args.sf, args.rf) and the column list (args.what) are interpolated directly into the SQL text. A value like O'Brien breaks the query; a value like ' OR '1'='1 (or worse, a stacked statement on a writable connection) is an injection. what and the table name cannot be bound as parameters, but should be validated against an allowlist of known columns.

Impact

CLI breaks on any value containing a single quote, and accepts arbitrary SQL through the filter and -what flags. Even on a read-mostly tool this is a real injection surface.

Proposed fix

Build the WHERE clause from parameter placeholders and collect a params list, passing it to cursor.execute(sql, params). Validate what against the actual table columns (queryable via PRAGMA_TABLE_INFO, already used elsewhere in this module). Representative patch for the filter-building portion of main and the consumer:

-	if args.ef:
-		experiment_filter = f" WHERE experiment = '{args.ef}'"
-
-	if args.vf:
-		if args.ef:
-			variation_filter = f" and variation = '{args.vf}'"
-		else:
-			variation_filter = f" WHERE variation = '{args.vf}'"
-
-	if args.sf:
-		if args.vf or args.ef:
-			system_filter = f" and system = '{args.sf}'"
-		else:
-			system_filter = f" WHERE system = '{args.sf}'"
-
-	if args.rf:
-		if args.vf or args.sf or args.ef:
-			runno_filter = f" and run = {args.rf}"
-		else:
-			runno_filter = f' WHERE run = {args.rf}'
-
-	all_filters = experiment_filter + variation_filter + system_filter + runno_filter
-
-	if args.what:
-		what = args.what
-
-	if args.sv:
-		show_volumes_from_database(sqlitedb, what, all_filters)
-
-	if args.sm:
-		show_materials_from_database(sqlitedb, what, all_filters)
+	# Collect (column, value) pairs for the WHERE clause as bound parameters.
+	conditions = []
+	params = []
+	for column, value in (("experiment", args.ef), ("variation", args.vf),
+	                      ("system", args.sf), ("run", args.rf)):
+		if value:
+			conditions.append(f"{column} = ?")
+			params.append(value)
+	where_clause = (" WHERE " + " AND ".join(conditions)) if conditions else ""
+
+	if args.what:
+		what = args.what
+
+	if args.sv:
+		show_volumes_from_database(sqlitedb, what, where_clause, params)
+
+	if args.sm:
+		show_materials_from_database(sqlitedb, what, where_clause, params)

-def show_volumes_from_database(sqlitedb, what, all_filters):
-	if sqlitedb is not None:
-		sql = sqlitedb.cursor()
-		query = "SELECT {} FROM geometry {};".format(what, all_filters)
-		print(query)
-		sql.execute(query)
-		for row in sql.fetchall():
-			print(row)
+def _validate_columns(sqlitedb, table, what):
+	# `what` is an identifier list and cannot be bound as a parameter,
+	# so validate it against the table's real columns (allowlist).
+	if what.strip() == "*":
+		return "*"
+	sql = sqlitedb.cursor()
+	sql.execute(f"SELECT name FROM PRAGMA_TABLE_INFO('{table}');")
+	valid = {r[0] for r in sql.fetchall()}
+	requested = [c.strip() for c in what.split(",")]
+	bad = [c for c in requested if c not in valid]
+	if bad:
+		sys.exit(f"{GColors.RED}Error: unknown column(s) {bad} for table '{table}'.{GColors.END}")
+	return ", ".join(requested)
+
+
+def show_volumes_from_database(sqlitedb, what, where_clause, params):
+	if sqlitedb is not None:
+		sql = sqlitedb.cursor()
+		cols = _validate_columns(sqlitedb, "geometry", what)
+		query = f"SELECT {cols} FROM geometry{where_clause};"
+		print(query, params)
+		sql.execute(query, params)
+		for row in sql.fetchall():
+			print(row)

show_materials_from_database gets the same treatment with table name "materials". Note table in _validate_columns is a fixed literal ("geometry"/"materials"), not user input, so its interpolation is safe.

---

Split out from #102 (one issue per bug). Found via an AI-assisted code review with manual verification against commit 5f8ce875.