fix: escape embedded quotes in identifiers to prevent SQL injection by dev-engineer

Author: Coding-Dev-Tools

.github/workflows/ci.yml

@@ -34,3 +34,20 @@ jobs:
         run: pip install ruff && ruff check src/ tests/ --target-version py310
 
       - name: Run tests
+        run: |
+          python -m pytest tests/ -v --cov=src --cov-report=term-missing
+
+  js-wrapper:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5  # v4
+        with:
+          persist-credentials: false
+
+      - name: Set up Node
+        uses: actions/setup-node@1e60f620b9541d16bece96c5465dc8ee9832be0b  # v4
+        with:
+          node-version: "20"
+
+      - name: Test the npm wrapper
+        run: node --test tests/*.test.js

src/json2sql/dialects.py

@@ -47,10 +47,20 @@ def sql_type_for(value: Any, dialect: Dialect) -> str:
 
 
 def quote_identifier(name: str, dialect: Dialect) -> str:
-    """Quote an identifier (table/column name) for the given dialect."""
+    """Quote an identifier (table/column name) for the given dialect.
+
+    Embedded quote characters are escaped by doubling — the standard SQL rule
+    for quoted identifiers — mirroring how ``format_value`` escapes string
+    literals. Without this, a table/column name (which comes straight from
+    untrusted JSON object keys) containing a ``"`` (Postgres/SQLite) or a
+    backtick (MySQL) could break out of the quoted identifier and inject
+    arbitrary SQL into the generated statement.
+    """
     if dialect == Dialect.MYSQL:
-        return f"`{name}`"
-    return f'"{name}"'
+        escaped = name.replace("`", "``")
+        return f"`{escaped}`"
+    escaped = name.replace('"', '""')
+    return f'"{escaped}"'
 
 
 def format_value(value: Any, dialect: Dialect) -> str:

tests/test_converter.py

@@ -174,9 +174,7 @@ def test_flatten_nested_array_column_value_count(self):
                 if cols and vals:
                     c_count = len([c for c in cols.group(1).split(",") if c.strip()])
                     v_count = len([v for v in vals.group(1).split(",") if v.strip()])
-                    assert c_count == v_count, (
-                        f"Column count ({c_count}) != value count ({v_count})"
-                    )
+                    assert c_count == v_count, f"Column count ({c_count}) != value count ({v_count})"
 
     def test_flatten_mixed_nested_array_and_object_count(self):
         """
@@ -201,9 +199,7 @@ def test_flatten_mixed_nested_array_and_object_count(self):
                 if cols and vals:
                     c_count = len([c for c in cols.group(1).split(",") if c.strip()])
                     v_count = len([v for v in vals.group(1).split(",") if v.strip()])
-                    assert c_count == v_count, (
-                        f"Column count ({c_count}) != value count ({v_count})"
-                    )
+                    assert c_count == v_count, f"Column count ({c_count}) != value count ({v_count})"
 
 
 class TestFlattenDetail:
@@ -401,6 +397,28 @@ def test_string_with_quotes(self, converter_postgres):
         result = converter_postgres.convert(data, table_name="profiles")
         assert "It''s a test" in result  # SQL-escaped single quote
 
+    def test_key_with_embedded_double_quote_postgres(self, converter_postgres):
+        # A JSON key containing " must be escaped in the generated identifier
+        # to prevent SQL injection through untrusted object keys.
+        data = json.dumps({'col"evil': 1})
+        result = converter_postgres.convert(data, table_name="profiles")
+        assert '"col""evil"' in result
+        # Ensure the dangerous unescaped form is NOT present
+        assert '"col"evil"' not in result
+
+    def test_key_with_embedded_double_quote_sqlite(self, converter_sqlite):
+        data = json.dumps({'col"evil': 1})
+        result = converter_sqlite.convert(data, table_name="profiles")
+        assert '"col""evil"' in result
+        assert '"col"evil"' not in result
+
+    def test_key_with_embedded_backtick_mysql(self, converter_mysql):
+        # A JSON key containing a backtick must be escaped in MySQL identifiers.
+        data = json.dumps({"col`evil": 1})
+        result = converter_mysql.convert(data, table_name="profiles")
+        assert "`col``evil`" in result
+        assert "`col`evil`" not in result
+
     def test_float_values(self, converter_postgres):
         data = json.dumps({"price": 19.99})
         result = converter_postgres.convert(data, table_name="products")
@@ -515,6 +533,5 @@ def test_version_in_init_matches_pyproject(self):
         with open(pyproject, "rb") as f:
             data = tomllib.load(f)
         assert data["project"]["version"] == __version__, (
-            f"pyproject.toml version ({data['project']['version']}) != "
-            f"__init__.__version__ ({__version__})"
+            f"pyproject.toml version ({data['project']['version']}) != __init__.__version__ ({__version__})"
         )

tests/test_dialects.py

@@ -36,14 +36,9 @@ def test_all_members(self):
 class TestSQLTypeFor:
     """Python type → SQL type mapping per dialect."""
 
-    @pytest.mark.parametrize(
-        "dialect", [Dialect.POSTGRES, Dialect.MYSQL, Dialect.SQLITE]
-    )
+    @pytest.mark.parametrize("dialect", [Dialect.POSTGRES, Dialect.MYSQL, Dialect.SQLITE])
     def test_string_type(self, dialect):
-        assert (
-            "TEXT" in sql_type_for("hello", dialect).upper()
-            or "VARCHAR" in sql_type_for("hello", dialect).upper()
-        )
+        assert "TEXT" in sql_type_for("hello", dialect).upper() or "VARCHAR" in sql_type_for("hello", dialect).upper()
 
     def test_postgres_int(self):
         assert sql_type_for(42, Dialect.POSTGRES) == "INTEGER"
@@ -115,6 +110,26 @@ def test_empty_name(self):
             result = quote_identifier("", dialect)
             assert len(result) >= 2
 
+    def test_embedded_quote_mysql(self):
+        """MySQL: backtick embedded in identifier is escaped by doubling."""
+        assert quote_identifier("my`table", Dialect.MYSQL) == "`my``table`"
+
+    def test_embedded_quote_postgres(self):
+        """Postgres: double-quote embedded in identifier is escaped by doubling."""
+        assert quote_identifier('my"table', Dialect.POSTGRES) == '"my""table"'
+
+    def test_embedded_quote_sqlite(self):
+        """SQLite: double-quote embedded in identifier is escaped by doubling."""
+        assert quote_identifier('my"table', Dialect.SQLITE) == '"my""table"'
+
+    def test_multiple_embedded_quotes_mysql(self):
+        """MySQL: multiple backticks in identifier are all escaped."""
+        assert quote_identifier("a`b`c", Dialect.MYSQL) == "`a``b``c`"
+
+    def test_multiple_embedded_quotes_postgres(self):
+        """Postgres: multiple double-quotes in identifier are all escaped."""
+        assert quote_identifier('a"b"c', Dialect.POSTGRES) == '"a""b""c"'
+
 
 # --- format_value ---