Adds sonar version of SQLParameterizer codemod (#495)

* Fixed issue with multiple class with injections

* Sonar sql parameterization codemod with tests

integration_tests/sonar/test_sonar_sql_parameterization.py

@@ -0,0 +1,28 @@
+from codemodder.codemods.test import SonarIntegrationTest
+from core_codemods.sonar.sonar_sql_parameterization import SonarSQLParameterization
+from core_codemods.sql_parameterization import SQLQueryParameterizationTransformer
+
+
+class TestSonarSQLParameterization(SonarIntegrationTest):
+    codemod = SonarSQLParameterization
+    code_path = "tests/samples/fix_sonar_sql_parameterization.py"
+    replacement_lines = [
+        (11, '    sql = """SELECT user FROM users WHERE user = ?"""\n'),
+        (14, "    conn.cursor().execute(sql, ((user), ))\n"),
+    ]
+    expected_diff = """\
+--- 
++++ 
+@@ -8,7 +8,7 @@
+ @app.route("/example")
+ def f():
+     user = request.args["user"]
+-    sql = \"\"\"SELECT user FROM users WHERE user = \\'%s\\'\"\"\"
++    sql = \"\"\"SELECT user FROM users WHERE user = ?\"\"\"
+ 
+     conn = sqlite3.connect("example")
+-    conn.cursor().execute(sql % (user))
++    conn.cursor().execute(sql, ((user), ))
+"""
+    expected_line_change = "14"
+    change_description = SQLQueryParameterizationTransformer.change_description

src/codemodder/scripts/generate_docs.py

@@ -294,6 +294,7 @@ class DocMetadata:
     "url-sandbox-S5144",
     "fix-float-equality-S1244",
     "fix-math-isclose-S6727",
+    "sql-parameterization-S3649",
 ]
 SONAR_CODEMODS = {
     name: DocMetadata(

src/core_codemods/__init__.py

@@ -67,6 +67,7 @@
     SonarRemoveAssertionInPytestRaises,
 )
 from .sonar.sonar_secure_random import SonarSecureRandom
+from .sonar.sonar_sql_parameterization import SonarSQLParameterization
 from .sonar.sonar_tempfile_mktemp import SonarTempfileMktemp
 from .sonar.sonar_url_sandbox import SonarUrlSandbox
 from .sql_parameterization import SQLQueryParameterization
@@ -166,6 +167,7 @@
         SonarUrlSandbox,
         SonarFixFloatEquality,
         SonarFixMathIsClose,
+        SonarSQLParameterization,
     ],
 )
 

src/core_codemods/sonar/sonar_sql_parameterization.py

@@ -0,0 +1,10 @@
+from core_codemods.sonar.api import SonarCodemod
+from core_codemods.sql_parameterization import SQLQueryParameterization
+
+SonarSQLParameterization = SonarCodemod.from_core_codemod(
+    name="sql-parameterization-S3649",
+    other=SQLQueryParameterization,
+    rule_id="pythonsecurity:S3649",
+    rule_name="Database queries should not be vulnerable to injection attacks",
+    rule_url="https://rules.sonarsource.com/python/RSPEC-3649/",
+)

src/core_codemods/sql_parameterization.py

@@ -126,7 +126,7 @@ def __init__(
         LibcstResultTransformer.__init__(self, *codemod_args, **codemod_kwargs)
         UtilsMixin.__init__(
             self,
-            [],
+            codemod_args[1],
             line_exclude=self.file_context.line_exclude,
             line_include=self.file_context.line_include,
         )
@@ -165,25 +165,25 @@ def _build_param_element(self, prepend, middle, append, linearized_query):
 
     def transform_module_impl(self, tree: cst.Module) -> cst.Module:
         """
-        The transformation is composed of 3 steps, each step is done by a codemod/visitor: (1) FindQueryCalls, (2) ExtractParameters, and (3) SQLQueryParameterization.
+        The transformation is composed of 3 steps, each step is done by a codemod/visitor: (1) FindQueryCalls, (2) ExtractParameters, and (3) _fix_injection
         Step (1) finds the `execute` calls and linearizing the query argument. Step (2) extracts the expressions that are parameters to the query.
         Step (3) swaps the parameters in the query for `?` tokens and passes them as an arguments for the `execute` call. At the end of the transformation, the `CleanCode` codemod is executed to remove leftover empty strings and unused variables.
         """
+
+        # Step (1)
         find_queries = FindQueryCalls(self.context)
         tree.visit(find_queries)
 
-        result = tree
         for call, linearized_query in find_queries.calls.items():
-            # filter by line includes/excludes
-            call_pos = self.node_position(call)
-            if not self.filter_by_path_includes_or_excludes(call_pos):
-                break
+            # filter node
+            if not self.node_is_selected(call):
+                return tree
 
-            # Step (3)
+            # Step (2)
             ep = ExtractParameters(self.context, linearized_query)
             tree.visit(ep)
 
-            # Step (4) - build tuple elements and fix injection
+            # Step (3)
             params_elements: list[cst.Element] = []
             for start, middle, end in ep.injection_patterns:
                 prepend, append = self._fix_injection(
@@ -246,7 +246,7 @@ def transform_module_impl(self, tree: cst.Module) -> cst.Module:
                                 value=new_raw_value
                             )
 
-                result = result.visit(ReplaceNodes(new_changed_nodes))
+                result = tree.visit(ReplaceNodes(new_changed_nodes))
                 self.changed_nodes = {}
                 line_number = self.get_metadata(PositionProvider, call).start.line
                 self.file_context.codemod_changes.append(
@@ -258,7 +258,12 @@ def transform_module_impl(self, tree: cst.Module) -> cst.Module:
                 # Normalization and cleanup
                 result = CleanCode(self.context).transform_module(result)
 
-        return result
+                # return after a single change
+                return result
+        return tree
+
+    def should_allow_multiple_passes(self) -> bool:
+        return True
 
     def _fix_injection(
         self,

tests/codemods/sonar/test_sonar_sql_parameterization.py

@@ -0,0 +1,60 @@
+import json
+
+from codemodder.codemods.test import BaseSASTCodemodTest
+from core_codemods.sonar.sonar_sql_parameterization import SonarSQLParameterization
+
+
+class TestSonarSQLParameterization(BaseSASTCodemodTest):
+    codemod = SonarSQLParameterization
+    tool = "sonar"
+
+    def test_name(self):
+        assert self.codemod.name == "sql-parameterization-S3649"
+
+    def test_simple(self, tmpdir):
+        input_code = """
+        import sqlite3
+
+        from flask import Flask, request
+
+        app = Flask(__name__)
+
+        @app.route("/example")
+        def f():
+            user = request.args["user"]
+            sql = '''SELECT user FROM users WHERE user = \'%s\' '''
+
+            conn = sqlite3.connect("example")
+            conn.cursor().execute(sql % (user))
+        """
+        expected = """
+        import sqlite3
+
+        from flask import Flask, request
+
+        app = Flask(__name__)
+
+        @app.route("/example")
+        def f():
+            user = request.args["user"]
+            sql = '''SELECT user FROM users WHERE user = ? '''
+
+            conn = sqlite3.connect("example")
+            conn.cursor().execute(sql, ((user), ))
+        """
+        issues = {
+            "issues": [
+                {
+                    "rule": "pythonsecurity:S3649",
+                    "status": "OPEN",
+                    "component": "code.py",
+                    "textRange": {
+                        "startLine": 14,
+                        "endLine": 14,
+                        "startOffset": 4,
+                        "endOffset": 39,
+                    },
+                }
+            ]
+        }
+        self.run_and_assert(tmpdir, input_code, expected, results=json.dumps(issues))

tests/codemods/test_sql_parameterization.py

@@ -48,6 +48,39 @@ def test_multiple(self, tmpdir):
         """
         self.run_and_assert(tmpdir, input_code, expected)
 
+    def test_multiple_injectable_calls(self, tmpdir):
+        input_code = """
+        import sqlite3
+
+        def f(user):
+            sql = '''SELECT user FROM users WHERE user = \'%s\' '''
+
+            conn = sqlite3.connect('example')
+            conn.cursor().execute(sql % (user))
+
+        def g(user):
+            sql = "SELECT user FROM users WHERE user = '" + user + "'"
+
+            conn = sqlite3.connect('example')
+            conn.cursor().execute(sql)
+        """
+        expected = """
+        import sqlite3
+
+        def f(user):
+            sql = '''SELECT user FROM users WHERE user = ? '''
+
+            conn = sqlite3.connect('example')
+            conn.cursor().execute(sql, ((user), ))
+
+        def g(user):
+            sql = "SELECT user FROM users WHERE user = ?"
+
+            conn = sqlite3.connect('example')
+            conn.cursor().execute(sql, (user, ))
+        """
+        self.run_and_assert(tmpdir, input_code, expected, num_changes=2)
+
     def test_simple_with_quotes_in_middle(self, tmpdir):
         input_code = """
         import sqlite3

tests/samples/fix_sonar_sql_parameterization.py

@@ -0,0 +1,14 @@
+import sqlite3
+
+from flask import Flask, request
+
+app = Flask(__name__)
+
+
+@app.route("/example")
+def f():
+    user = request.args["user"]
+    sql = """SELECT user FROM users WHERE user = \'%s\'"""
+
+    conn = sqlite3.connect("example")
+    conn.cursor().execute(sql % (user))