Merge 00a9ef9 into 151d31d

pylint_odoo/checkers/no_modules.py

@@ -441,6 +441,11 @@ def _sqli_allowable(self, node):
         # sql.SQL or sql.Identifier is OK
         if self._is_psycopg2_sql(node):
             return True
+        if isinstance(node, astroid.FormattedValue):
+            if hasattr(node, 'value'):
+                return self._sqli_allowable(node.value)
+            if hasattr(node, 'values'):
+                return all(self._sqli_allowable(v) for v in node.values)
         if isinstance(node, astroid.Call):
             node = node.func
         # self._thing is OK (mostly self._table), self._thing() also because
@@ -521,6 +526,13 @@ def _check_node_for_sqli_risk(self, node):
             ):
                 return True
 
+        # Check fstrings (PEP 498). Only Python >= 3.6
+        if isinstance(node, astroid.JoinedStr):
+            if hasattr(node, 'value'):
+                return self._sqli_allowable(node.value)
+            elif hasattr(node, 'values'):
+                return not all(self._sqli_allowable(v) for v in node.values)
+
         return False
 
     def _check_sql_injection_risky(self, node):

pylint_odoo/test/main.py

@@ -1,8 +1,8 @@
-
 import os
 import stat
 import sys
-from tempfile import gettempdir
+from tempfile import gettempdir, NamedTemporaryFile
+from textwrap import dedent
 import six
 
 import unittest
@@ -127,7 +127,7 @@ def tearDown(self):
         sys.path = list(self.sys_path_origin)
         test = self._testMethodName
         prefix = os.path.expanduser(os.environ.get('PYLINT_ODOO_STATS',
-                                    '~/pylint_odoo_cprofile'))
+                                                   '~/pylint_odoo_cprofile'))
         fstats = prefix + '_' + test + '.stats'
         if test != 'test_10_path_dont_exist':
             self.profile.dump_stats(fstats)
@@ -229,6 +229,7 @@ def my_which(bin_name, *args, **kwargs):
             if bin_name == 'eslint':
                 return None
             return which_original(bin_name)
+
         misc.which = my_which
         my_which("noeslint")
         pylint_res = self.run_pylint(self.paths_modules)
@@ -307,14 +308,14 @@ def test_90_valid_odoo_versions(self):
         self.assertDictEqual(real_errors, expected_errors)
 
     @unittest.skipIf(not six.PY3, "unnecessary-utf8-coding-comment "
-                     "disabled directly from py2")
+                                  "disabled directly from py2")
     def test_100_read_version_from_manifest(self):
         """Test the functionality to get the version from the file manifest
         to avoid the parameter --valid_odoo_versions"""
         modules = [mod for mod in self.paths_modules if
                    'eleven_module' in mod or 'twelve_module' in mod]
         extra_params = ['--disable=all', '--enable=no-utf8-coding-comment,'
-                        'unnecessary-utf8-coding-comment']
+                                         'unnecessary-utf8-coding-comment']
         pylint_res = self.run_pylint(modules, extra_params)
         real_errors = pylint_res.linter.stats['by_msg']
         self.assertListEqual(list(real_errors.items()),
@@ -432,6 +433,42 @@ def test_140_check_migrations_is_not_odoo_module(self):
         expected_errors = {}
         self.assertDictEqual(real_errors, expected_errors)
 
+    @unittest.skipUnless(
+        sys.version_info[0:2] >= (3, 6),
+        "Fstrings (PEP498) are only supported since 3.6"
+    )
+    def test_145_check_fstring_sqli(self):
+        """Verify the linter is capable of finding SQL Injection vulnerabilities
+        when using fstrings.
+        Related to https://github.com/OCA/pylint-odoo/issues/363"""
+        extra_params = [
+            '--disable=all',
+            '--enable=sql-injection'
+        ]
+        queries = '''
+            def fstring_sqli(self):
+                self.env.cr.execute(f"SELECT * FROM TABLE WHERE SQLI = {self.table}")
+                self.env.cr.execute(
+                    f"SELECT * FROM TABLE WHERE SQLI = {'hello' + self.name}"
+                )
+                self.env.cr.execute(f"SELECT * FROM {self.name} WHERE SQLI = {'hello'}")
+                death_wish = f"SELECT * FROM TABLE WHERE SQLI = {self.name}"
+                self.env.cr.execute(death_wish)
+            def fstring_no_sqli(self):
+                self.env.cr.execute(f"SELECT * FROM TABLE WHERE SQLI = {'hello'}")
+                self.env.cr.execute(
+                    f"CREATE VIEW {self._table} AS (SELECT * FROM res_partner)"
+                )
+                self.env.cr.execute(f"SELECT NAME FROM res_partner LIMIT 10")
+            '''
+        with NamedTemporaryFile(mode='w', encoding='utf-8', delete=False) as f:
+            self.addCleanup(os.remove, f.name)
+            f.write(dedent(queries).lstrip())
+
+        pylint_res = self.run_pylint([f.name], extra_params)
+        real_errors = pylint_res.linter.stats['by_msg']
+        self.assertDictEqual(real_errors, {'sql-injection': 4})
+
 
 if __name__ == '__main__':
     unittest.main()