Security advisory GHSA-8prr-286p-4w7j: alerta-server has potential SQL Injection vulnerability in Query String Syntax (q=) API
Package: alerta-server (pypi), vulnerable version: 9.0.4, fixed in: 9.1.0
CWE: CWE-89
Files changed by the fix:
alerta/database/backends/postgres/queryparser.py
alerta/database/backends/postgres/utils.py

Fix diff:
diff --git a/alerta/database/backends/postgres/queryparser.py b/alerta/database/backends/postgres/queryparser.py
index 3f4feae04..346ff01a8 100644
--- a/alerta/database/backends/postgres/queryparser.py
+++ b/alerta/database/backends/postgres/queryparser.py
@@ -5,6 +5,11 @@
 ParserElement.enablePackrat()
 
 
+def _next_param(params):
+    name = f'_qp_{len(params)}'
+    return name
+
+
 class UnaryOperation:
     """takes one operand,e.g. not"""
 
@@ -26,12 +31,18 @@ class SearchModifier(UnaryOperation):
     def __repr__(self):
         return f'{self.op} {self.operands}'
 
+    def to_sql(self, params):
+        return f'{self.op} {self.operands.to_sql(params)}'
+
 
 class SearchAnd(BinaryOperation):
 
     def __repr__(self):
         return f'({self.lhs} AND {self.rhs})'
 
+    def to_sql(self, params):
+        return f'({self.lhs.to_sql(params)} AND {self.rhs.to_sql(params)})'
+
 
 class SearchOr(BinaryOperation):
 
@@ -40,12 +51,20 @@ def __repr__(self):
             return f'({self.lhs} AND {self.rhs})'
         return f'({self.lhs} OR {self.rhs})'
 
+    def to_sql(self, params):
+        if getattr(self.rhs, 'op', None) == 'NOT':
+            return f'({self.lhs.to_sql(params)} AND {self.rhs.to_sql(params)})'
+        return f'({self.lhs.to_sql(params)} OR {self.rhs.to_sql(params)})'
+
 
 class SearchNot(UnaryOperation):
 
     def __repr__(self):
         return f'NOT ({self.operands})'
 
+    def to_sql(self, params):
+        return f'NOT ({self.operands.to_sql(params)})'
+
 
 class SearchTerm:
 
@@ -113,6 +132,89 @@ def __repr__(self):
 
         raise ParseException(f'Search term did not match query syntax: {self.tokens}')
 
+    def to_sql(self, params):
+        if 'singleterm' in self.tokens:
+            if self.tokens.fieldname == '_exists_':
+                p = _next_param(params)
+                params[p] = self.tokens.singleterm
+                return f'"attributes"::jsonb ? %({p})s'
+            elif self.tokens.fieldname in ['correlate', 'service', 'tags']:
+                p = _next_param(params)
+                params[p] = self.tokens.singleterm
+                return f'%({p})s=ANY("{self.tokens.field[0]}")'
+            elif self.tokens.attr:
+                tokens_attr = self.tokens.attr.replace('_', 'attributes')
+                p_key = _next_param(params)
+                params[p_key] = self.tokens.fieldname
+                p_val = _next_param(params)
+                params[p_val] = f'%{self.tokens.singleterm}%'
+                return f'"{tokens_attr}"::jsonb ->>%({p_key})s ILIKE %({p_val})s'
+            else:
+                p = _next_param(params)
+                params[p] = f'%{self.tokens.singleterm}%'
+                return f'"{self.tokens.field[0]}" ILIKE %({p})s'
+        if 'phrase' in self.tokens:
+            if self.tokens.field[0] == '__default_field__':
+                p = _next_param(params)
+                params[p] = f'\\y{self.tokens.phrase}\\y'
+                return f'"__default_field__" ~* %({p})s'
+            elif self.tokens.field[0] in ['correlate', 'service', 'tags']:
+                p = _next_param(params)
+                params[p] = self.tokens.phrase
+                return f'%({p})s=ANY("{self.tokens.field[0]}")'
+            elif self.tokens.attr:
+                tokens_attr = self.tokens.attr.replace('_', 'attributes')
+                p_key = _next_param(params)
+                params[p_key] = self.tokens.fieldname
+                p_val = _next_param(params)
+                params[p_val] = f'\\y{self.tokens.phrase}\\y'
+                return f'"{tokens_attr}"::jsonb ->>%({p_key})s ~* %({p_val})s'
+            else:
+                p = _next_param(params)
+                params[p] = f'\\y{self.tokens.phrase}\\y'
+                return f'"{self.tokens.field[0]}" ~* %({p})s'
+        if 'wildcard' in self.tokens:
+            p = _next_param(params)
+            params[p] = f'\\y{self.tokens.wildcard}\\y'
+            return f'"{self.tokens.field[0]}" ~* %({p})s'
+        if 'regex' in self.tokens:
+            p = _next_param(params)
+            params[p] = self.tokens.regex
+            return f'"{self.tokens.field[0]}" ~* %({p})s'
+        if 'range' in self.tokens:
+            if self.tokens.range[0].lowerbound == '*':
+                lower_term = '1=1'
+            else:
+                p = _next_param(params)
+                params[p] = self.tokens.range[0].lowerbound
+                op = '>=' if 'inclusive' in self.tokens.range[0] else '>'
+                lower_term = f'"{self.tokens.field[0]}" {op} %({p})s'
+
+            if self.tokens.range[2].upperbound == '*':
+                upper_term = '1=1'
+            else:
+                p = _next_param(params)
+                params[p] = self.tokens.range[2].upperbound
+                op = '<=' if 'inclusive' in self.tokens.range[2] else '<'
+                upper_term = f'"{self.tokens.field[0]}" {op} %({p})s'
+            return f'({lower_term} AND {upper_term})'
+        if 'onesidedrange' in self.tokens:
+            p = _next_param(params)
+            params[p] = self.tokens.onesidedrange.bound
+            return f'("{self.tokens.field[0]}" {self.tokens.onesidedrange.op} %({p})s)'
+        if 'subquery' in self.tokens:
+            subquery_sql = self.tokens.subquery[0].to_sql(params)
+            if self.tokens.attr:
+                tokens_attr = 'attributes' if self.tokens.attr == '_' else self.tokens.attr
+                p = _next_param(params)
+                params[p] = self.tokens.fieldname
+                tokens_fieldname = f'"{tokens_attr}"::jsonb ->>%({p})s'
+            else:
+                tokens_fieldname = f'"{self.tokens.fieldname or self.tokens.field[0]}"'
+            return subquery_sql.replace('"__default_field__"', tokens_fieldname)
+
+        raise ParseException(f'Search term did not match query syntax: {self.tokens}')
+
 
 # BNF for Lucene query syntax
 #
@@ -181,4 +283,7 @@ class QueryParser:
 
     def parse(self, query, default_field=None):
         default_field = default_field or QueryParser.DEFAULT_FIELD
-        return repr(query_expr.parseString(query)[0]).replace('__default_field__', default_field)
+        params = {}
+        parsed = query_expr.parseString(query)[0]
+        sql = parsed.to_sql(params)
+        return sql.replace('__default_field__', default_field), params
diff --git a/alerta/database/backends/postgres/utils.py b/alerta/database/backends/postgres/utils.py
index 7249ee5b3..b3dbd7684 100644
--- a/alerta/database/backends/postgres/utils.py
+++ b/alerta/database/backends/postgres/utils.py
@@ -145,11 +145,12 @@ def from_params(params: MultiDict, customers=None, query_time=None):
         if params.get('q', None):
             try:
                 parser = QueryParser()
-                query = [parser.parse(
+                parsed_query, parsed_vars = parser.parse(
                     query=params['q'],
                     default_field=params.get('q.df')
-                )]
-                qvars = dict()  # type: Dict[str, Any]
+                )
+                query = [parsed_query]
+                qvars = dict(parsed_vars)  # type: Dict[str, Any]
             except ParseException as e:
                 raise ApiError('Failed to parse query string.', 400, [e])
         else:
@@ -230,11 +231,12 @@ def from_params(params: MultiDict, customers=None, query_time=None):
         if params.get('q', None):
             try:
                 parser = QueryParser()
-                query = [parser.parse(
+                parsed_query, parsed_vars = parser.parse(
                     query=params['q'],
                     default_field=params.get('q.df')
-                )]
-                qvars = dict()  # type: Dict[str, Any]
+                )
+                query = [parsed_query]
+                qvars = dict(parsed_vars)  # type: Dict[str, Any]
             except ParseException as e:
                 raise ApiError('Failed to parse query string.', 400, [e])
         else:
diff --git a/tests/test_queryparser.py b/tests/test_queryparser.py
index 6e8439cf3..d5e578897 100644
--- a/tests/test_queryparser.py
+++ b/tests/test_queryparser.py
@@ -23,103 +23,121 @@ def test_word_and_phrase_terms(self):
 
         # default field (ie. "text") contains word
         string = r'''quick'''
-        r = self.parser.parse(string)
-        self.assertEqual(r, '"text" ILIKE \'%%quick%%\'')
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '"text" ILIKE %(_qp_0)s')
+        self.assertEqual(params, {'_qp_0': '%quick%'})
 
         # default field (ie. "text") contains either words
         string = r'''quick OR brown'''
-        r = self.parser.parse(string)
-        self.assertEqual(r, '("text" ILIKE \'%%quick%%\' OR "text" ILIKE \'%%brown%%\')')
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '("text" ILIKE %(_qp_0)s OR "text" ILIKE %(_qp_1)s)')
+        self.assertEqual(params, {'_qp_0': '%quick%', '_qp_1': '%brown%'})
 
         # default field (ie. "text") contains either words (default operator)
         string = r'''quick brown'''
-        r = self.parser.parse(string)
-        self.assertEqual(r, '("text" ILIKE \'%%quick%%\' OR "text" ILIKE \'%%brown%%\')')
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '("text" ILIKE %(_qp_0)s OR "text" ILIKE %(_qp_1)s)')
+        self.assertEqual(params, {'_qp_0': '%quick%', '_qp_1': '%brown%'})
 
         # default field (ie. "text") contains exact phrase
         string = r'''"quick brown"'''
-        r = self.parser.parse(string)
-        self.assertEqual(r, '"text" ~* \'\\yquick brown\\y\'')
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '"text" ~* %(_qp_0)s')
+        self.assertEqual(params, {'_qp_0': '\\yquick brown\\y'})
 
     def test_field_names(self):
 
         # field contains word
         string = r'''status:active'''
-        r = self.parser.parse(string)
-        self.assertEqual(r, '"status" ILIKE \'%%active%%\'')
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '"status" ILIKE %(_qp_0)s')
+        self.assertEqual(params, {'_qp_0': '%active%'})
 
         # field contains either words
         string = r'''title:(quick OR brown)'''
-        r = self.parser.parse(string)
-        self.assertEqual(r, '("title" ILIKE \'%%quick%%\' OR "title" ILIKE \'%%brown%%\')')
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '("title" ILIKE %(_qp_0)s OR "title" ILIKE %(_qp_1)s)')
+        self.assertEqual(params, {'_qp_0': '%quick%', '_qp_1': '%brown%'})
 
         # field contains either words (default operator)
         string = r'''title:(quick brown)'''
-        r = self.parser.parse(string)
-        self.assertEqual(r, '("title" ILIKE \'%%quick%%\' OR "title" ILIKE \'%%brown%%\')')
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '("title" ILIKE %(_qp_0)s OR "title" ILIKE %(_qp_1)s)')
+        self.assertEqual(params, {'_qp_0': '%quick%', '_qp_1': '%brown%'})
 
         # field contains exact phrase
         string = r'''author:"John Smith"'''
-        r = self.parser.parse(string)
-        self.assertEqual(r, '"author" ~* \'\\yJohn Smith\\y\'')
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '"author" ~* %(_qp_0)s')
+        self.assertEqual(params, {'_qp_0': '\\yJohn Smith\\y'})
 
         # # any attribute contains word or phrase
         # string = r'''attributes.\*:(quick brown)'''
-        # r = self.parser.parse(string)
-        # self.assertEqual(r, '??')
+        # sql, params = self.parser.parse(string)
+        # self.assertEqual(sql, '??')
 
         # attribute field has non-null value
         string = r'''_exists_:title'''
-        r = self.parser.parse(string)
-        self.assertEqual(r, '"attributes"::jsonb ? \'title\'')
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '"attributes"::jsonb ? %(_qp_0)s')
+        self.assertEqual(params, {'_qp_0': 'title'})
 
         # attribute contains word
         string = r'''foo.vendor:cisco'''
-        r = self.parser.parse(string)
-        self.assertEqual(r, '"foo"::jsonb ->>\'vendor\' ILIKE \'%%cisco%%\'')
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '"foo"::jsonb ->>%(_qp_0)s ILIKE %(_qp_1)s')
+        self.assertEqual(params, {'_qp_0': 'vendor', '_qp_1': '%cisco%'})
 
         # attribute contains word ("_" shortcut)
         string = r'''_.vendor:cisco'''
-        r = self.parser.parse(string)
-        self.assertEqual(r, '"attributes"::jsonb ->>\'vendor\' ILIKE \'%%cisco%%\'')
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '"attributes"::jsonb ->>%(_qp_0)s ILIKE %(_qp_1)s')
+        self.assertEqual(params, {'_qp_0': 'vendor', '_qp_1': '%cisco%'})
 
         # attribute contains either words
         string = r'''attributes.vendor:(cisco OR juniper)'''
-        r = self.parser.parse(string)
-        self.assertEqual(r, '("attributes"::jsonb ->>\'vendor\' ILIKE \'%%cisco%%\' OR "attributes"::jsonb ->>\'vendor\' ILIKE \'%%juniper%%\')')
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '("attributes"::jsonb ->>%(_qp_2)s ILIKE %(_qp_0)s OR "attributes"::jsonb ->>%(_qp_2)s ILIKE %(_qp_1)s)')
+        self.assertEqual(params, {'_qp_0': '%cisco%', '_qp_1': '%juniper%', '_qp_2': 'vendor'})
 
         # attribute contains either words (default operator)
         string = r'''attributes.vendor:(cisco juniper)'''
-        r = self.parser.parse(string)
-        self.assertEqual(r, '("attributes"::jsonb ->>\'vendor\' ILIKE \'%%cisco%%\' OR "attributes"::jsonb ->>\'vendor\' ILIKE \'%%juniper%%\')')
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '("attributes"::jsonb ->>%(_qp_2)s ILIKE %(_qp_0)s OR "attributes"::jsonb ->>%(_qp_2)s ILIKE %(_qp_1)s)')
+        self.assertEqual(params, {'_qp_0': '%cisco%', '_qp_1': '%juniper%', '_qp_2': 'vendor'})
 
         # attribute contains either words ("_" shortcut, default operator)
         string = r'''_.vendor:(cisco juniper)'''
-        r = self.parser.parse(string)
-        self.assertEqual(r, '("attributes"::jsonb ->>\'vendor\' ILIKE \'%%cisco%%\' OR "attributes"::jsonb ->>\'vendor\' ILIKE \'%%juniper%%\')')
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '("attributes"::jsonb ->>%(_qp_2)s ILIKE %(_qp_0)s OR "attributes"::jsonb ->>%(_qp_2)s ILIKE %(_qp_1)s)')
+        self.assertEqual(params, {'_qp_0': '%cisco%', '_qp_1': '%juniper%', '_qp_2': 'vendor'})
 
         # attribute contains exact phrase
         string = r'''foo.vendor:"quick brown"'''
-        r = self.parser.parse(string)
-        self.assertEqual(r, '"foo"::jsonb ->>\'vendor\' ~* \'\\yquick brown\\y\'')
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '"foo"::jsonb ->>%(_qp_0)s ~* %(_qp_1)s')
+        self.assertEqual(params, {'_qp_0': 'vendor', '_qp_1': '\\yquick brown\\y'})
 
         # attribute contains exact phrase ("_" shortcut)
         string = r'''_.vendor:"quick brown"'''
-        r = self.parser.parse(string)
-        self.assertEqual(r, '"attributes"::jsonb ->>\'vendor\' ~* \'\\yquick brown\\y\'')
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '"attributes"::jsonb ->>%(_qp_0)s ~* %(_qp_1)s')
+        self.assertEqual(params, {'_qp_0': 'vendor', '_qp_1': '\\yquick brown\\y'})
 
     def test_wildcards(self):
 
         # ? = single character, * = one or more characters
         string = r'''text:qu?ck bro*'''
-        r = self.parser.parse(string)
-        self.assertEqual(r, '("text" ~* \'\\yqu.?ck\\y\' OR "text" ~* \'\\ybro.*\\y\')')
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '("text" ~* %(_qp_0)s OR "text" ~* %(_qp_1)s)')
+        self.assertEqual(params, {'_qp_0': '\\yqu.?ck\\y', '_qp_1': '\\ybro.*\\y'})
 
     def test_regular_expressions(self):
 
         string = r'''name:/joh?n(ath[oa]n)/'''
-        r = self.parser.parse(string)
-        self.assertEqual(r, '"name" ~* \'joh?n(ath[oa]n)\'')
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '"name" ~* %(_qp_0)s')
+        self.assertEqual(params, {'_qp_0': 'joh?n(ath[oa]n)'})
 
     def test_fuzziness(self):
         pass
@@ -130,46 +148,56 @@ def test_proximity_searches(self):
     def test_ranges(self):
 
         string = r'''date:[2012-01-01 TO 2012-12-31]'''
-        r = self.parser.parse(string)
-        self.assertEqual(r, '("date" >= \'2012-01-01\' AND "date" <= \'2012-12-31\')')
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '("date" >= %(_qp_0)s AND "date" <= %(_qp_1)s)')
+        self.assertEqual(params, {'_qp_0': '2012-01-01', '_qp_1': '2012-12-31'})
 
         string = r'''count:[1 TO 5]'''
-        r = self.parser.parse(string)
-        self.assertEqual(r, '("count" >= \'1\' AND "count" <= \'5\')')
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '("count" >= %(_qp_0)s AND "count" <= %(_qp_1)s)')
+        self.assertEqual(params, {'_qp_0': '1', '_qp_1': '5'})
 
         string = r'''tag:{alpha TO omega}'''
-        r = self.parser.parse(string)
-        self.assertEqual(r, '("tag" > \'alpha\' AND "tag" < \'omega\')')
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '("tag" > %(_qp_0)s AND "tag" < %(_qp_1)s)')
+        self.assertEqual(params, {'_qp_0': 'alpha', '_qp_1': 'omega'})
 
         string = r'''count:[10 TO *]'''
-        r = self.parser.parse(string)
-        self.assertEqual(r, '("count" >= \'10\' AND 1=1)')
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '("count" >= %(_qp_0)s AND 1=1)')
+        self.assertEqual(params, {'_qp_0': '10'})
 
         string = r'''date:{* TO 2012-01-01}'''
-        r = self.parser.parse(string)
-        self.assertEqual(r, '(1=1 AND "date" < \'2012-01-01\')')
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '(1=1 AND "date" < %(_qp_0)s)')
+        self.assertEqual(params, {'_qp_0': '2012-01-01'})
 
         string = r'''count:[1 TO 5}'''
-        r = self.parser.parse(string)
-        self.assertEqual(r, '("count" >= \'1\' AND "count" < \'5\')')
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '("count" >= %(_qp_0)s AND "count" < %(_qp_1)s)')
+        self.assertEqual(params, {'_qp_0': '1', '_qp_1': '5'})
 
     def test_unbounded_ranges(self):
 
         string = r'''age:>10'''
-        r = self.parser.parse(string)
-        self.assertEqual(r, '("age" > \'10\')')
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '("age" > %(_qp_0)s)')
+        self.assertEqual(params, {'_qp_0': '10'})
 
         string = r'''age:>=10'''
-        r = self.parser.parse(string)
-        self.assertEqual(r, '("age" >= \'10\')')
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '("age" >= %(_qp_0)s)')
+        self.assertEqual(params, {'_qp_0': '10'})
 
         string = r'''age:<10'''
-        r = self.parser.parse(string)
-        self.assertEqual(r, '("age" < \'10\')')
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '("age" < %(_qp_0)s)')
+        self.assertEqual(params, {'_qp_0': '10'})
 
         string = r'''age:<=10'''
-        r = self.parser.parse(string)
-        self.assertEqual(r, '("age" <= \'10\')')
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '("age" <= %(_qp_0)s)')
+        self.assertEqual(params, {'_qp_0': '10'})
 
     def test_boosting(self):
         pass
@@ -178,45 +206,54 @@ def test_boolean_operators(self):
 
         # OR (||)
         string = r'''"jakarta apache" jakarta'''
-        r = self.parser.parse(string)
-        self.assertEqual(r, '("text" ~* \'\\yjakarta apache\\y\' OR "text" ILIKE \'%%jakarta%%\')')
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '("text" ~* %(_qp_0)s OR "text" ILIKE %(_qp_1)s)')
+        self.assertEqual(params, {'_qp_0': '\\yjakarta apache\\y', '_qp_1': '%jakarta%'})
 
         string = r'''"jakarta apache" OR jakarta'''
-        r = self.parser.parse(string)
-        self.assertEqual(r, '("text" ~* \'\\yjakarta apache\\y\' OR "text" ILIKE \'%%jakarta%%\')')
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '("text" ~* %(_qp_0)s OR "text" ILIKE %(_qp_1)s)')
+        self.assertEqual(params, {'_qp_0': '\\yjakarta apache\\y', '_qp_1': '%jakarta%'})
 
         string = r'''"jakarta apache" || jakarta'''
-        r = self.parser.parse(string)
-        self.assertEqual(r, '("text" ~* \'\\yjakarta apache\\y\' OR "text" ILIKE \'%%jakarta%%\')')
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '("text" ~* %(_qp_0)s OR "text" ILIKE %(_qp_1)s)')
+        self.assertEqual(params, {'_qp_0': '\\yjakarta apache\\y', '_qp_1': '%jakarta%'})
 
         # AND (&&)
         string = r'''"jakarta apache" AND "Apache Lucene"'''
-        r = self.parser.parse(string)
-        self.assertEqual(r, '("text" ~* \'\\yjakarta apache\\y\' AND "text" ~* \'\\yApache Lucene\\y\')')
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '("text" ~* %(_qp_0)s AND "text" ~* %(_qp_1)s)')
+        self.assertEqual(params, {'_qp_0': '\\yjakarta apache\\y', '_qp_1': '\\yApache Lucene\\y'})
 
         string = r'''"jakarta apache" && "Apache Lucene"'''
-        r = self.parser.parse(string)
-        self.assertEqual(r, '("text" ~* \'\\yjakarta apache\\y\' AND "text" ~* \'\\yApache Lucene\\y\')')
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '("text" ~* %(_qp_0)s AND "text" ~* %(_qp_1)s)')
+        self.assertEqual(params, {'_qp_0': '\\yjakarta apache\\y', '_qp_1': '\\yApache Lucene\\y'})
 
         # + (required)
         pass
 
         # NOT (!)
         string = r'''"jakarta apache" NOT "Apache Lucene"'''
-        r = self.parser.parse(string)
-        self.assertEqual(r, '("text" ~* \'\\yjakarta apache\\y\' AND NOT ("text" ~* \'\\yApache Lucene\\y\'))')
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '("text" ~* %(_qp_0)s AND NOT ("text" ~* %(_qp_1)s))')
+        self.assertEqual(params, {'_qp_0': '\\yjakarta apache\\y', '_qp_1': '\\yApache Lucene\\y'})
 
         string = r'''"jakarta apache" !"Apache Lucene"'''
-        r = self.parser.parse(string)
-        self.assertEqual(r, '("text" ~* \'\\yjakarta apache\\y\' AND NOT ("text" ~* \'\\yApache Lucene\\y\'))')
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '("text" ~* %(_qp_0)s AND NOT ("text" ~* %(_qp_1)s))')
+        self.assertEqual(params, {'_qp_0': '\\yjakarta apache\\y', '_qp_1': '\\yApache Lucene\\y'})
 
         string = r'''NOT "jakarta apache"'''
-        r = self.parser.parse(string)
-        self.assertEqual(r, 'NOT ("text" ~* \'\\yjakarta apache\\y\')')
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, 'NOT ("text" ~* %(_qp_0)s)')
+        self.assertEqual(params, {'_qp_0': '\\yjakarta apache\\y'})
 
         string = r'''group:"jakarta apache" NOT group:"Apache Lucene"'''
-        r = self.parser.parse(string)
-        self.assertEqual(r, '("group" ~* \'\\yjakarta apache\\y\' AND NOT ("group" ~* \'\\yApache Lucene\\y\'))')
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '("group" ~* %(_qp_0)s AND NOT ("group" ~* %(_qp_1)s))')
+        self.assertEqual(params, {'_qp_0': '\\yjakarta apache\\y', '_qp_1': '\\yApache Lucene\\y'})
 
         # - (prohibit)
         pass
@@ -225,14 +262,30 @@ def test_grouping(self):
 
         # field exact match
         string = r'''(quick OR brown) AND fox'''
-        r = self.parser.parse(string)
-        self.assertEqual(r, '(("text" ILIKE \'%%quick%%\' OR "text" ILIKE \'%%brown%%\') AND "text" ILIKE \'%%fox%%\')')
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '(("text" ILIKE %(_qp_0)s OR "text" ILIKE %(_qp_1)s) AND "text" ILIKE %(_qp_2)s)')
+        self.assertEqual(params, {'_qp_0': '%quick%', '_qp_1': '%brown%', '_qp_2': '%fox%'})
 
         # field exact match
         string = r'''status:(active OR pending) title:(full text search)'''
-        r = self.parser.parse(string)
+        sql, params = self.parser.parse(string)
         self.assertEqual(
-            r, '(("status" ILIKE \'%%active%%\' OR "status" ILIKE \'%%pending%%\') OR ("title" ILIKE \'%%full%%\' OR "title" ILIKE \'%%text%%\'))')
+            sql, '(("status" ILIKE %(_qp_0)s OR "status" ILIKE %(_qp_1)s) OR ("title" ILIKE %(_qp_2)s OR "title" ILIKE %(_qp_3)s))')
+        self.assertEqual(params, {'_qp_0': '%active%', '_qp_1': '%pending%', '_qp_2': '%full%', '_qp_3': '%text%'})
+
+    def test_sql_injection_prevention(self):
+
+        # single quotes in search terms should be safely parameterized
+        string = r'''text:O'Brien'''
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '"text" ILIKE %(_qp_0)s')
+        self.assertIn("O'Brien", params['_qp_0'])
+
+        # SQL keywords in search terms should be safely parameterized
+        string = '''"DROP TABLE alerts"'''
+        sql, params = self.parser.parse(string)
+        self.assertEqual(sql, '"text" ~* %(_qp_0)s')
+        self.assertIn('DROP TABLE alerts', params['_qp_0'])
 
 
 def skip_mongodb():