Add checks for SQL alchemy (#111)

* Add new sql alchemy inspection skeleton

* SQL alchemy check

HISTORY.md

@@ -1,5 +1,9 @@
 # Release History
 
+## 1.18.0
+
+* Added [SQL200](doc/checks/SQL200.md) for inspection of insecure SQLalchemy method calls
+
 ## 1.17.0 
 
 * Added a source path argument to the GitHub action
@@ -21,7 +25,7 @@
 
 * Reduced size of Docker image [pull#98](pull/98)
 * Bugfix on TRY100 raising false-positives. Fixes [issues#88](issues/88) - [pull#97](pull/97)
-* Added [STR100] for insecure format strings
+* Added [STR100](doc/checks/STR100.md) for insecure format strings
 
 ## 1.13.0
 

build.gradle

@@ -7,7 +7,7 @@ plugins {
 }
 
 group 'org.tonybaloney.security'
-version '1.16.0'
+version '1.18.0'
 
 repositories {
     mavenCentral()
@@ -39,9 +39,9 @@ intellij.version = project.hasProperty('intellijVersion') ? project.getProperty(
 
 patchPluginXml {
     changeNotes """
-      <h2>1.16.0</h2>
+      <h2>1.18.0</h2>
       <ul>
-        <li>Primary suport for 2020 IDEs.</li>
+        <li>Added [SQL200] for inspection of insecure SQLalchemy method calls</li>
       </ul>
       """
 }

doc/checks/SQL200.md

@@ -0,0 +1,55 @@
+# SQL200
+
+Looks for SQL injection in the SQLalchemy library.
+
+- Use of `text()` function to construct parameters on non-literal input
+- Use of the `.suffix_with()` and `.prefix_with()` methods on a query object with unsafe input
+
+## Examples
+
+Use of the SQLalchemy with a `text()` fragment can expose the constructed query to SQL injection.
+
+For example, this query should generate 
+
+```python
+part = f"age<{age}"  # exploitable, can override the original filter.
+_x = session.query(User).filter(User.username == user).filter(text(part)).all()
+```
+
+With an input of `age = 224`:
+
+```sql
+SELECT users.id AS users_id, users.name AS users_name, users.fullname AS users_fullname FROM users WHERE users.id = ? AND age < 224 OR 1=1
+```
+
+If the `age` argument was `224 OR 1=1`, the query would bypass the filter:
+
+```sql
+SELECT users.id AS users_id, users.name AS users_name, users.fullname AS users_fullname FROM users WHERE users.id = ? AND id<224 OR 1=1
+```
+
+Both the `.suffix_with()` and `.prefix_with()` methods are vulnerable to unsafe input.
+
+```python
+suffix = " OR 1=1"  # Example exploiting suffix to add/change WHERE clause
+prefix = " *,"  # Example exploiting query to get all fields
+stmt = select([users.c.name]).where(users.c.id == 1).suffix_with(suffix, dialect="sqlite")
+conn.execute(stmt)
+
+stmt2 = select([addresses]).prefix_with(prefix)  # can be chained
+conn.execute(stmt2)
+```
+
+Direct execution of vulnerable queries will be caught by SQL100:
+
+```python
+connection.execute("SELECT email_address FROM addresses WHERE email_address = \'{}\'".format(unsafe_input))
+```
+
+## Fixes
+
+Replace with native SQLalchemy queries using the API instead of creating direct SQL.
+
+## See Also
+
+[RealPython.com article on SQL injection](https://realpython.com/prevent-python-sql-injection/)

src/main/java/security/Checks.kt

@@ -42,6 +42,8 @@ object Checks {
     val StrFormatInspectionCheck = CheckType("STR100", "Calling format with insecure string.")
     val ShellInjectionCheck = CheckType("SH100", "Potential shell injection with unescaped input.")
     val SpawnShellInjectionCheck = CheckType("SH101", "Potentially risky call to spawned process. Check inputs for validation.")
+    val SqlAlchemyUnsafeQueryCheck = CheckType("SQL200", "Possible SQL injection through SQLAlchemy API")
+
 
     class CheckType(var Code: String, private var Message: String) {
         private var _staticDescription: String? = null

src/main/java/security/validators/SqlAlchemyUnsafeQueryInspection.kt

@@ -0,0 +1,40 @@
+package security.validators
+
+import com.intellij.codeInspection.LocalInspectionToolSession
+import com.intellij.codeInspection.ProblemsHolder
+import com.intellij.psi.PsiElementVisitor
+import com.jetbrains.python.inspections.PyInspection
+import com.jetbrains.python.psi.PyCallExpression
+import com.jetbrains.python.psi.PyStringLiteralExpression
+import security.Checks
+import security.helpers.SecurityVisitor
+import security.helpers.calleeMatches
+import security.helpers.hasImportedNamespace
+import security.helpers.skipDocstring
+import security.registerProblem
+
+class SqlAlchemyUnsafeQueryInspection : PyInspection() {
+    val check = Checks.SqlAlchemyUnsafeQueryCheck
+
+    override fun getStaticDescription(): String? {
+        return check.getStaticDescription()
+    }
+
+    override fun buildVisitor(holder: ProblemsHolder,
+                              isOnTheFly: Boolean,
+                              session: LocalInspectionToolSession): PsiElementVisitor = Visitor(holder, session)
+
+    private class Visitor(holder: ProblemsHolder, session: LocalInspectionToolSession) : SecurityVisitor(holder, session) {
+
+        override fun visitPyCallExpression(node: PyCallExpression) {
+            if (skipDocstring(node)) return
+            val targetFunctions = arrayOf("text", "prefix_with", "suffix_with")
+            if (!calleeMatches(node, targetFunctions)) return
+            if (!hasImportedNamespace(node.containingFile, "sqlalchemy")) return
+            if (node.arguments.isNullOrEmpty()) return
+            val sqlStatement = node.arguments.first()
+            if (sqlStatement is PyStringLiteralExpression) return
+            holder.registerProblem(node, Checks.SqlAlchemyUnsafeQueryCheck)
+        }
+    }
+}

src/main/resources/META-INF/plugin.xml

@@ -64,6 +64,7 @@
         <localInspection language="Python" enabledByDefault="true" groupName="Python Security" hasStaticDescription="true" displayName="STR100: Calling format with insecure string." shortName="StrFormatInspection" implementationClass="security.validators.StrFormatInspection" />
         <localInspection language="Python" enabledByDefault="true" groupName="Python Security" hasStaticDescription="true" displayName="SH100: Potential shell injection with unescaped input." shortName="StandardShellInjectionInspection" implementationClass="security.validators.StandardShellInjectionInspection" />
         <localInspection language="Python" enabledByDefault="true" groupName="Python Security" hasStaticDescription="true" level="WEAK WARNING" displayName="SH101: Potentially risky call to spawned process." shortName="SpawnShellInjectionInspection" implementationClass="security.validators.SpawnShellInjectionInspection" />
+        <localInspection language="Python" enabledByDefault="true" groupName="Python Security" hasStaticDescription="true" displayName="SQL200: Possible SQL injection through SQLAlchemy API." shortName="SqlAlchemyUnsafeQueryInspection" implementationClass="security.validators.SqlAlchemyUnsafeQueryInspection" />
 
         <applicationConfigurable displayName="Python Security" instance="security.settings.SecuritySettingsConfigurable" id="org.tonybaloney.security.pycharm-security.settings.SecuritySettingsConfigurable" groupId="tools"/>
         <applicationService serviceImplementation="security.settings.SecuritySettings" id="org.tonybaloney.security.pycharm-security.settings.SecuritySettings" />

src/main/resources/docs/SQL200.html

@@ -0,0 +1,34 @@
+<h1>SQL200</h1>
+<p>Looks for SQL injection in the SQLalchemy library.</p>
+<ul>
+  <li>Use of <code>text()</code> function to construct parameters on non-literal input</li>
+  <li>Use of the <code>.suffix_with()</code> and <code>.prefix_with()</code> methods on a query object with unsafe input</li>
+</ul>
+<h2>Examples</h2>
+<p>Use of the SQLalchemy with a <code>text()</code> fragment can expose the constructed query to SQL injection.</p>
+<p>For example, this query should generate </p>
+<pre><code class="python">part = f&quot;age&lt;{age}&quot;  # exploitable, can override the original filter.
+_x = session.query(User).filter(User.username == user).filter(text(part)).all()
+</code></pre>
+<p>With an input of <code>age = 224</code>:</p>
+<pre><code class="sql">SELECT users.id AS users_id, users.name AS users_name, users.fullname AS users_fullname FROM users WHERE users.id = ? AND age &lt; 224 OR 1=1
+</code></pre>
+<p>If the <code>age</code> argument was <code>224 OR 1=1</code>, the query would bypass the filter:</p>
+<pre><code class="sql">SELECT users.id AS users_id, users.name AS users_name, users.fullname AS users_fullname FROM users WHERE users.id = ? AND id&lt;224 OR 1=1
+</code></pre>
+<p>Both the <code>.suffix_with()</code> and <code>.prefix_with()</code> methods are vulnerable to unsafe input.</p>
+<pre><code class="python">suffix = &quot; OR 1=1&quot;  # Example exploiting suffix to add/change WHERE clause
+prefix = &quot; *,&quot;  # Example exploiting query to get all fields
+stmt = select([users.c.name]).where(users.c.id == 1).suffix_with(suffix, dialect=&quot;sqlite&quot;)
+conn.execute(stmt)
+
+stmt2 = select([addresses]).prefix_with(prefix)  # can be chained
+conn.execute(stmt2)
+</code></pre>
+<p>Direct execution of vulnerable queries will be caught by SQL100:</p>
+<pre><code class="python">connection.execute(&quot;SELECT email_address FROM addresses WHERE email_address = \&#39;{}\&#39;&quot;.format(unsafe_input))
+</code></pre>
+<h2>Fixes</h2>
+<p>Replace with native SQLalchemy queries using the API instead of creating direct SQL.</p>
+<h2>See Also</h2>
+<p><a href="https://realpython.com/prevent-python-sql-injection/">RealPython.com article on SQL injection</a></p>

src/main/resources/docs/STR100.html

@@ -1,9 +1,8 @@
 <h1>STR100</h1>
 <p>String format function allows access to protected attributes, is someone are able to manage the format string can access to sensible information.</p>
 <h2>Example</h2>
-<pre><code class="python">
-CONFIG = {
-    'SECRET_KEY': 'super secret key'
+<pre><code class="python">CONFIG = {
+    &#39;SECRET_KEY&#39;: &#39;super secret key&#39;
 }
 
 class Event(object):
@@ -15,14 +14,14 @@ <h2>Example</h2>
 def format_event(format_string, event):
     return format_string.format(event=event)
 </code></pre>
-<p>If <code>format_event</code> is executed with <code>format_string = "{event.__init__.__globals__[CONFIG][SECRET_KEY]}"</code>, the secret_key will be read</p>
+<p>If <code>format_event</code> is executed with <code>format_string = &quot;{event.__init__.__globals__[CONFIG][SECRET_KEY]}&quot;</code>, the secret_key will be read</p>
 <h2>Fixes</h2>
 <ul>
   <li>Replace using string.Template</li>
   <li>Replace using CustomFormatter(string.Formatter) overwriting the get_field function and disable the access to protected attributes (all with _ at the beginning)</li>
 </ul>
 <h2>See Also</h2>
 <ul>
-    <li><a href="https://lucumr.pocoo.org/2016/12/29/careful-with-str-format/">Be Careful with Python's New-Style String Format</a></li>
-    <li><a href="https://palletsprojects.com/blog/jinja-281-released/">Jinja 2.8.1 Security Release</a>
-</ul>
+  <li><a href="https://lucumr.pocoo.org/2016/12/29/careful-with-str-format/">Be Careful with Python's New-Style String Format</a></li>
+  <li><a href="https://palletsprojects.com/blog/jinja-281-released/">Jinja 2.8.1 Security Release</a></li>
+</ul>

src/test/java/security/packaging/SnykCheckerTest.kt

@@ -7,7 +7,7 @@ import org.junit.jupiter.api.Test
 
 internal class SnykCheckerTest {
     val testPackage = PyPackage("rsa", "3.4.0", null, listOf())
-    val testUrl = "https://private-anon-c1f72ebc65-snyk.apiary-mock.com/api/v1"
+    val testUrl = "https://private-anon-fb110a8acb-snyk.apiary-mock.com/api/v1"
 
     @Test
     fun `test rsa package has match on test API`() {

src/test/java/security/validators/SqlAlchemyUnsafeQueryInspectionTest.kt

@@ -0,0 +1,86 @@
+package security.validators
+
+import org.junit.jupiter.api.AfterAll
+import org.junit.jupiter.api.BeforeAll
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.TestInstance
+import security.Checks
+import security.SecurityTestTask
+
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+class SqlAlchemyUnsafeQueryInspectionTest: SecurityTestTask() {
+    @BeforeAll
+    override fun setUp() {
+        super.setUp()
+    }
+
+    @AfterAll
+    override fun tearDown(){
+        super.tearDown()
+    }
+
+    @Test
+    fun `verify description is not empty`(){
+        assertFalse(SqlAlchemyUnsafeQueryInspection().staticDescription.isNullOrEmpty())
+    }
+
+    @Test
+    fun `test unsafe text`() {
+        var code = """
+            import sqlalchemy
+            
+            sqlalchemy.text(data)
+        """.trimIndent()
+        testCodeCallExpression(code, 1, Checks.SqlAlchemyUnsafeQueryCheck, "test.py", SqlAlchemyUnsafeQueryInspection())
+    }
+
+    @Test
+    fun `test safe argument`() {
+        var code = """
+            import sqlalchemy
+            
+            sqlalchemy.text(data)
+        """.trimIndent()
+        testCodeCallExpression(code, 0, Checks.SqlAlchemyUnsafeQueryCheck, "test.py", SqlAlchemyUnsafeQueryInspection())
+    }
+
+    @Test
+    fun `test complex text`() {
+        var code = """
+            import sqlalchemy
+            
+            session.query(User).filter(User.id == 1).filter(text(part)).all()
+        """.trimIndent()
+        testCodeCallExpression(code, 1, Checks.SqlAlchemyUnsafeQueryCheck, "test.py", SqlAlchemyUnsafeQueryInspection())
+    }
+
+    @Test
+    fun `test unsafe suffix`() {
+        var code = """
+            import sqlalchemy
+            
+            select([users.c.name]).where(users.c.id == 1).suffix_with(suffix, dialect="sqlite")
+        """.trimIndent()
+        testCodeCallExpression(code, 1, Checks.SqlAlchemyUnsafeQueryCheck, "test.py", SqlAlchemyUnsafeQueryInspection())
+    }
+
+    @Test
+    fun `test text not sqlalchemy`(){
+        var code = """
+            import bicycle
+            
+            select([users.c.name]).where(users.c.id == 1).suffix_with(suffix, dialect="sqlite")
+        """.trimIndent()
+        testCodeCallExpression(code, 0, Checks.SqlAlchemyUnsafeQueryCheck, "test.py", SqlAlchemyUnsafeQueryInspection())
+    }
+
+    @Test
+    fun `test sqlalchemy not text`(){
+        var code = """
+            import sqlalchemy
+            
+            vext(x)
+        """.trimIndent()
+        testCodeCallExpression(code, 0, Checks.SqlAlchemyUnsafeQueryCheck, "test.py", SqlAlchemyUnsafeQueryInspection())
+    }
+}