|
1 # -*- coding: utf-8 -*- |
|
2 |
|
3 # Copyright (c) 2020 Detlev Offenbach <detlev@die-offenbachs.de> |
|
4 # |
|
5 |
|
6 """ |
|
7 Module implementing a check for SQL injection. |
|
8 """ |
|
9 |
|
10 # |
|
11 # This is a modified version of the one found in the bandit package. |
|
12 # |
|
13 # Original Copyright 2014 Hewlett-Packard Development Company, L.P. |
|
14 # |
|
15 # SPDX-License-Identifier: Apache-2.0 |
|
16 # |
|
17 |
|
18 import ast |
|
19 import re |
|
20 |
|
21 from Security import SecurityUtils |
|
22 |
|
23 |
|
24 def getChecks(): |
|
25 """ |
|
26 Public method to get a dictionary with checks handled by this module. |
|
27 |
|
28 @return dictionary containing checker lists containing checker function and |
|
29 list of codes |
|
30 @rtype dict |
|
31 """ |
|
32 return { |
|
33 "Str": [ |
|
34 (checkHardcodedSqlExpressions, ("S608",)), |
|
35 ], |
|
36 } |
|
37 |
|
38 |
|
39 SIMPLE_SQL_RE = re.compile( |
|
40 r'(select\s.*from\s|' |
|
41 r'delete\s+from\s|' |
|
42 r'insert\s+into\s.*values\s|' |
|
43 r'update\s.*set\s)', |
|
44 re.IGNORECASE | re.DOTALL, |
|
45 ) |
|
46 |
|
47 |
|
48 def _checkString(data): |
|
49 """ |
|
50 Function to check a given string against the list of search patterns. |
|
51 |
|
52 @param data string data to be checked |
|
53 @type str |
|
54 @return flag indicating a match |
|
55 @rtype bool |
|
56 """ |
|
57 return SIMPLE_SQL_RE.search(data) is not None |
|
58 |
|
59 |
|
60 def _evaluateAst(node): |
|
61 """ |
|
62 Function to analyze the given ast node. |
|
63 |
|
64 @param node ast node to be analyzed |
|
65 @type ast.Str |
|
66 @return tuple containing a flag indicating an execute call and |
|
67 the resulting statement |
|
68 @rtype tuple of (bool, str) |
|
69 """ |
|
70 wrapper = None |
|
71 statement = '' |
|
72 |
|
73 if isinstance(node._securityParent, ast.BinOp): |
|
74 out = SecurityUtils.concatString(node, node._securityParent) |
|
75 wrapper = out[0]._securityParent |
|
76 statement = out[1] |
|
77 elif ( |
|
78 isinstance(node._securityParent, ast.Attribute) and |
|
79 node._securityParent.attr == 'format' |
|
80 ): |
|
81 statement = node.s |
|
82 # Hierarchy for "".format() is Wrapper -> Call -> Attribute -> Str |
|
83 wrapper = node._securityParent._securityParent._securityParent |
|
84 elif ( |
|
85 hasattr(ast, 'JoinedStr') and |
|
86 isinstance(node._securityParent, ast.JoinedStr) |
|
87 ): |
|
88 statement = node.s |
|
89 wrapper = node._securityParent._securityParent |
|
90 |
|
91 if isinstance(wrapper, ast.Call): # wrapped in "execute" call? |
|
92 names = ['execute', 'executemany'] |
|
93 name = SecurityUtils.getCalledName(wrapper) |
|
94 return (name in names, statement) |
|
95 else: |
|
96 return (False, statement) |
|
97 |
|
98 |
|
99 def checkHardcodedSqlExpressions(reportError, context, config): |
|
100 """ |
|
101 Function to check for SQL injection. |
|
102 |
|
103 @param reportError function to be used to report errors |
|
104 @type func |
|
105 @param context security context object |
|
106 @type SecurityContext |
|
107 @param config dictionary with configuration data |
|
108 @type dict |
|
109 """ |
|
110 val = _evaluateAst(context.node) |
|
111 if _checkString(val[1]): |
|
112 reportError( |
|
113 context.node.lineno - 1, |
|
114 context.node.col_offset, |
|
115 "S608", |
|
116 "M", |
|
117 "M" if val[0] else "L", |
|
118 ) |