initial commit

Author: Roguelazer

.gitignore

@@ -0,0 +1,7 @@
+*.pyc
+*.egg-info
+*.egg_info
+build/
+dist/
+sdist/
+env/

LICENSE.txt

@@ -0,0 +1,21 @@
+py-find-injection
+Copyright (c) 2013 Uber Technologies, Inc.
+The MIT License
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to
+deal in the Software without restriction, including without limitation the
+rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
+sell copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
+FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
+IN THE SOFTWARE.

README.md

@@ -0,0 +1 @@
+`py_find_injection` uses various heuristics to look for SQL injection vulnerabilities in python source code.

py_find_injection/__init__.py

@@ -0,0 +1,138 @@
+#!/usr/bin/env python
+
+import argparse
+import ast
+import sys
+
+
+def stringify(node):
+    if isinstance(node, ast.Name):
+        return node.id
+    elif isinstance(node, ast.Attribute):
+        return '%s.%s' % (stringify(node.value), node.attr)
+    elif isinstance(node, ast.Subscript):
+        return '%s[%s]' % (stringify(node.value), stringify(node.slice))
+    elif isinstance(node, ast.Index):
+        return stringify(node.value)
+    elif isinstance(node, ast.Call):
+        return '%s(%s, %s)' % (stringify(node.func), stringify(node.args), stringify(node.keywords))
+    elif isinstance(node, list):
+        return '[%s]' % (', '.join(stringify(n) for n in node))
+    elif isinstance(node, ast.Str):
+        return node.s
+    else:
+        return ast.dump(node)
+
+
+class IllegalLine(object):
+    def __init__(self, reason, node, filename):
+        self.reason = reason
+        self.lineno = node.lineno
+        self.filename = filename
+        self.node = node
+
+    def __str__(self):
+        return "%s:%d\t%s" % (self.filename, self.lineno, self.reason)
+
+    def __repr__(self):
+        return "IllegalLine<%s, %s:%s>" % (self.reason, self.filename, self.lineno)
+
+
+def find_assignment_in_context(variable, context):
+    if isinstance(context, (ast.FunctionDef, ast.Module, ast.For, ast.While, ast.With, ast.If)):
+        for node in reversed(list(ast.iter_child_nodes(context))):
+            if isinstance(node, ast.Assign):
+                if variable in (stringify(c) for c in node.targets):
+                    return node
+    if getattr(context, 'parent', None):
+        return find_assignment_in_context(variable, context.parent)
+    else:
+        return None
+
+
+class Checker(ast.NodeVisitor):
+    def __init__(self, filename, *args, **kwargs):
+        self.filename = filename
+        self.errors = []
+        super(Checker, self).__init__(*args, **kwargs)
+
+    def check_execute(self, node):
+        if isinstance(node, ast.BinOp):
+            if isinstance(node.op, ast.Mod):
+                return IllegalLine('string interpolation of SQL query', node, self.filename)
+            elif isinstance(node.op, ast.Add):
+                return IllegalLine('string concatenation of SQL query', node, self.filename)
+        elif isinstance(node, ast.Call):
+            if isinstance(node.func, ast.Attribute):
+                if node.func.attr == 'format':
+                    return IllegalLine('str.format called on SQL query', node, self.filename)
+        elif isinstance(node, ast.Name):
+            # now we need to figure out where that query is assigned. blargh.
+            assignment = find_assignment_in_context(node.id, node)
+            if assignment is not None:
+                return self.check_execute(assignment.value)
+
+    def visit_Call(self, node):
+        function_name = stringify(node.func)
+        if function_name.lower() in ('session.execute', 'cursor.execute'):
+            node.args[0].parent = node
+            node_error = self.check_execute(node.args[0])
+            if node_error:
+                self.errors.append(node_error)
+        elif function_name.lower() == 'eval':
+            self.errors.append(IllegalLine('eval() is just generally evil', node, self.filename))
+        self.generic_visit(node)
+
+    def visit(self, node):
+        """Visit a node."""
+        method = 'visit_' + node.__class__.__name__
+        visitor = getattr(self, method, self.generic_visit)
+        return visitor(node)
+
+    def generic_visit(self, node):
+        """Called if no explicit visitor function exists for a node."""
+        for field, value in ast.iter_fields(node):
+            if isinstance(value, list):
+                for item in value:
+                    if isinstance(item, ast.AST):
+                        item.parent = node
+                        self.visit(item,)
+            elif isinstance(value, ast.AST):
+                value.parent = node
+                self.visit(value)
+
+
+def check(filename):
+    c = Checker(filename=filename)
+    with open(filename, 'r') as fobj:
+        try:
+            parsed = ast.parse(fobj.read(), filename)
+            c.visit(parsed)
+        except Exception:
+            raise
+    return c.errors
+
+
+def main():
+    parser = argparse.ArgumentParser(
+        description='Look for patterns in python source files that might indicate SQL injection vulnerabilities',
+        epilog='Exit status is 0 if all files are okay, 1 if any files have an error. Errors are printed to stdout'
+    )
+    parser.add_argument('files', nargs='+', help='Files to check')
+    args = parser.parse_args()
+
+    errors = []
+    for fname in args.files:
+        these_errors = check(fname)
+        if these_errors:
+            print '\n'.join(str(e) for e in these_errors)
+            errors.extend(these_errors)
+    if errors:
+        print '%d total errors' % len(errors)
+        return 1
+    else:
+        return 0
+
+
+if __name__ == '__main__':
+    sys.exit(main())

requirements-tests.txt

@@ -0,0 +1,2 @@
+nose==1.3.0
+mock==1.0.1

setup.py

@@ -0,0 +1,35 @@
+from setuptools import setup, find_packages
+
+
+setup(
+    name="py_find_injection",
+    version="0.1",
+    author="James Brown",
+    author_email="jbrown@uber.com",
+    url="http://github.com/uber/py_find_injection",
+    description="simple python ast consumer which searches for common SQL injection attacks",
+    classifiers=[
+        "Programming Language :: Python",
+        "Operating System :: OS Independent",
+        "Topic :: Security",
+        "Topic :: Security",
+        "Intended Audience :: Developers",
+        "Development Status :: 3 - Alpha",
+        "Programming Language :: Python :: 2.7",
+        "License :: OSI Approved :: MIT License",
+    ],
+    packages=find_packages(exclude=["tests"]),
+    entry_points={
+        "console_scripts": [
+            "py-find-injection = py_find_injection:main",
+        ]
+    },
+    tests_require=["nose==1.3.0", "mock==1.0.1"],
+    test_suite="nose.collector",
+    long_description="""py_find_injection
+
+Walks the AST and looks for arguments to cursor.execute or session.execute; then
+determines whether string interpolation, concatenation or the .format() call is used
+on those arguments. Not at all comprehensive, but better than nothing.
+"""
+)

tests/__init__.py

@@ -0,0 +1,6 @@
+import mock
+
+session = mock.Mock()
+
+
+session.execute("SELECT foo WHERE id=%s" % 2)

tests/samples/bad_script.py

@@ -0,0 +1,9 @@
+import mock
+
+session = mock.Mock()
+
+query = 1 + 2
+query = "SELECT foo FROM bar WHERE id="
+query = query + "2"
+cursor = session.cursor()
+cursor.execute(query)

tests/samples/bad_script_2.py

@@ -0,0 +1,7 @@
+import mock
+
+session = mock.Mock()
+uid = 1
+
+session.execute("SELECT foo FROM bar")
+session.execute("SELECT foo FROM bar WHERE id=%s", uid)

tests/samples/good_script.py

@@ -0,0 +1,25 @@
+import os.path
+from unittest import TestCase
+
+import py_find_injection
+
+
+SAMPLE_PATH = os.path.realpath(os.path.join(os.path.dirname(__file__), 'samples'))
+
+
+class TestSimple(TestCase):
+    def test_good_file(self):
+        errors = py_find_injection.check(os.path.join(SAMPLE_PATH, 'good_script.py'))
+        self.assertEqual(errors, [])
+
+    def test_bad_file(self):
+        errors = py_find_injection.check(os.path.join(SAMPLE_PATH, 'bad_script.py'))
+        self.assertEqual(1, len(errors))
+        self.assertEqual(errors[0].lineno, 6)
+        self.assertEqual(errors[0].reason, 'string interpolation of SQL query')
+
+    def test_interpolation_not_inline(self):
+        errors = py_find_injection.check(os.path.join(SAMPLE_PATH, 'bad_script_2.py'))
+        self.assertEqual(1, len(errors))
+        self.assertEqual(errors[0].lineno, 7)
+        self.assertEqual(errors[0].reason, 'string concatenation of SQL query')