Initial Version of SQL Parameterizer Codemod

integration_tests/test_sql_parameterization.py

@@ -0,0 +1,41 @@
+from core_codemods.sql_parameterization import SQLQueryParameterization
+from integration_tests.base_test import (
+    BaseIntegrationTest,
+    original_and_expected_from_code_path,
+)
+
+
+class TestSQLQueryParameterization(BaseIntegrationTest):
+    codemod = SQLQueryParameterization
+    code_path = "tests/samples/sql_injection.py"
+    original_code, expected_new_code = original_and_expected_from_code_path(
+        code_path,
+        [
+            (9, """    b = " WHERE name =?"\n"""),
+            (10, """    c = " AND phone = ?"\n"""),
+            (11, """    r = cursor.execute(a + b + c, (name, phone, ))\n"""),
+        ],
+    )
+
+    # fmt: off
+    expected_diff =(
+    """--- \n"""
+    """+++ \n"""
+    """@@ -7,9 +7,9 @@\n"""
+    """ \n"""
+    """ def foo(cursor: sqlite3.Cursor, name: str, phone: str):\n"""
+    """     a = "SELECT * FROM Users"\n"""
+    """-    b = " WHERE name ='" + name\n"""
+    """-    c = "' AND phone = '" + phone + "'"\n"""
+    """-    r = cursor.execute(a + b + c)\n"""
+    """+    b = " WHERE name =?"\n"""
+    """+    c = " AND phone = ?"\n"""
+    """+    r = cursor.execute(a + b + c, (name, phone, ))\n"""
+    """     print(r.fetchone())\n"""
+    """ \n"""
+    """ \n""")
+    # fmt: on
+
+    expected_line_change = "12"
+    change_description = SQLQueryParameterization.CHANGE_DESCRIPTION
+    num_changed_files = 1

src/codemodder/codemods/transformations/remove_empty_string_concatenation.py

@@ -0,0 +1,40 @@
+import libcst as cst
+from libcst import CSTTransformer
+
+
+class RemoveEmptyStringConcatenation(CSTTransformer):
+    """
+    Removes concatenation with empty strings (e.g. "hello " + "") or "hello" ""
+    """
+
+    def leave_BinaryOperation(
+        self, original_node: cst.BinaryOperation, updated_node: cst.BinaryOperation
+    ) -> cst.BaseExpression:
+        return self.handle_node(updated_node)
+
+    def leave_ConcatenatedString(
+        self,
+        original_node: cst.ConcatenatedString,
+        updated_node: cst.ConcatenatedString,
+    ) -> cst.BaseExpression:
+        return self.handle_node(updated_node)
+
+    def handle_node(
+        self, updated_node: cst.BinaryOperation | cst.ConcatenatedString
+    ) -> cst.BaseExpression:
+        match updated_node.left:
+            # TODO f-string cases
+            case cst.SimpleString() if updated_node.left.raw_value == "":
+                match updated_node.right:
+                    case cst.SimpleString() if updated_node.right.raw_value == "":
+                        return cst.SimpleString(value='""')
+                    case _:
+                        return updated_node.right
+        match updated_node.right:
+            case cst.SimpleString() if updated_node.right.raw_value == "":
+                match updated_node.left:
+                    case cst.SimpleString() if updated_node.left.raw_value == "":
+                        return cst.SimpleString(value='""')
+                    case _:
+                        return updated_node.left
+        return updated_node

src/codemodder/codemods/utils.py

@@ -1,27 +1,65 @@
 from pathlib import Path
-from typing import Union
+from typing import Optional, Any
 
 from libcst import matchers
 import libcst as cst
 
 
+class SequenceExtension:
+    def __init__(self, sequence: list[cst.CSTNode]) -> None:
+        self.sequence = sequence
+
+
+class Append(SequenceExtension):
+    pass
+
+
+class Prepend(SequenceExtension):
+    pass
+
+
 class ReplaceNodes(cst.CSTTransformer):
     """
-    Replace nodes with their corresponding values in a given dict.
+    Replace nodes with their corresponding values in a given dict. The replacements dictionary should either contain a mapping from a node to another node, RemovalSentinel, or FlattenSentinel to be replaced, or a dict mapping each attribute, by name, to a new value. Additionally if the attribute is a sequence, you may pass Append(l)/Prepend(l), where l is a list of nodes, to append or prepend, respectively.
     """
 
     def __init__(
         self,
         replacements: dict[
             cst.CSTNode,
-            Union[cst.CSTNode, cst.FlattenSentinel[cst.CSTNode], cst.RemovalSentinel],
+            dict[
+                str,
+                cst.CSTNode
+                | cst.FlattenSentinel
+                | cst.RemovalSentinel
+                | dict[str, Any],
+            ],
         ],
     ):
         self.replacements = replacements
 
     def on_leave(self, original_node, updated_node):
         if original_node in self.replacements.keys():
-            return self.replacements[original_node]
+            replacement = self.replacements[original_node]
+            match replacement:
+                case dict():
+                    changes_dict = {}
+                    for key, value in replacement.items():
+                        match value:
+                            case Prepend():
+                                changes_dict[key] = value.sequence + [
+                                    *getattr(updated_node, key)
+                                ]
+
+                            case Append():
+                                changes_dict[key] = [
+                                    *getattr(updated_node, key)
+                                ] + value.sequence
+                            case _:
+                                changes_dict[key] = value
+                    return updated_node.with_changes(**changes_dict)
+                case cst.CSTNode() | cst.RemovalSentinel() | cst.FlattenSentinel():
+                    return replacement
         return updated_node
 
 
@@ -44,3 +82,12 @@
         return call.func.attr.value
     # It's a simple Name
     return call.func.value
+
+
+def get_function_name_node(call: cst.Call) -> Optional[cst.Name]:
+    match call.func:
+        case cst.Name():
+            return call.func
+        case cst.Attribute():
+            return call.func.attr
+    return None

src/core_codemods/__init__.py

@@ -1,4 +1,5 @@
 from codemodder.registry import CodemodCollection
+from core_codemods.sql_parameterization import SQLQueryParameterization
 
 from .django_debug_flag_on import DjangoDebugFlagOn
 from .django_session_cookie_secure_off import DjangoSessionCookieSecureOff
@@ -52,5 +53,6 @@
         UrlSandbox,
         UseWalrusIf,
         WithThreadingLock,
+        SQLQueryParameterization,
     ],
 )

src/core_codemods/docs/pixee_python_sql-parameterization.md

@@ -0,0 +1,15 @@
+This codemod refactors SQL statements to be parameterized, rather than built by hand.
+
+Without parameterization, developers must remember to escape string inputs using the rules for that column type and database. This usually results in bugs -- and sometimes vulnerability. Although it's not clear if this code is exploitable today, this change will make the code more robust in case the conditions which prevent exploitation today ever go away.
+
+Our changes look something like this:
+
+```diff
+import sqlite3
+
+name = input()
+connection = sqlite3.connect("my_db.db")
+cursor = connection.cursor()
+- cursor.execute("SELECT * from USERS WHERE name ='" + name + "'")
++ cursor.execute("SELECT * from USERS WHERE name =?", (name, ))
+```