pengyuthon/odoo_source
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[FIX] test_lint, *: false negative in sql injection linter

Browse Source 
The linter would miss / fail to warn on injection of *local variables*
in some cases.

Try to improve it to be stricter and more reliable, after discussion
with odo, sql which is "correctly" dynamic should use psycopg2's sql
package in order to bypass the linter (bonus: it should also properly
escape & quote identifiers).

closes odoo/odoo#53938

Related: odoo/enterprise#11718
Signed-off-by: Xavier Morel (xmo) <xmo@odoo.com>
This commit is contained in:
Xavier Morel
2020-07-10 07:12:35 +00:00
parent 7be1429656
commit e162e6f714
8 changed files with 111 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
addons/crm_iap_lead_website/models/crm_reveal_rule.py
+3 -3
View File
@@ -262,10 +262,10 @@ class CRMRevealRule(models.Model):
ON v.reveal_rule_id = r.id
WHERE v.reveal_state='to_process'
GROUP BY v.reveal_ip
LIMIT %d
""" % batch_limit
LIMIT %s
"""
self.env.cr.execute(query)
self.env.cr.execute(query, [batch_limit])
return self.env.cr.fetchall()
def _prepare_iap_payload(self, pgv):
addons/fleet/report/fleet_report.py
+6 -1
View File
@@ -1,5 +1,6 @@
# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
from psycopg2 import sql
from odoo import tools
from odoo import api, fields, models
@@ -128,4 +129,8 @@ UNION ALL (
contract_costs cc)
"""
tools.drop_view_if_exists(self.env.cr, self._table)
self.env.cr.execute("""CREATE or REPLACE VIEW %s as (%s)""" % (self._table, query))
self.env.cr.execute(
sql.SQL("""CREATE or REPLACE VIEW {} as ({})""").format(
sql.Identifier(self._table),
sql.SQL(query)
))
addons/website_slides/models/slide_channel.py
+3 -3
View File
@@ -375,9 +375,9 @@ class Channel(models.Model):
else:
query = """
UPDATE %(table_name)s
SET %(column_name)s = md5(md5(random()::varchar || id::varchar) || clock_timestamp()::varchar)::uuid::varchar
WHERE %(column_name)s IS NULL
""" % {'table_name': self._table, 'column_name': column_name}
SET access_token = md5(md5(random()::varchar || id::varchar) || clock_timestamp()::varchar)::uuid::varchar
WHERE access_token IS NULL
""" % {'table_name': self._table}
self.env.cr.execute(query)
@api.model
odoo/addons/base/models/ir_model.py
+26 -11
View File
@@ -11,6 +11,7 @@ from collections.abc import Mapping
from operator import itemgetter
import dateutil
from psycopg2 import sql
from odoo import api, fields, models, tools, _
from odoo.exceptions import AccessError, UserError, ValidationError
@@ -203,7 +204,7 @@ class IrModel(models.Model):
for model in self:
records = self.env[model.model]
if not records._abstract:
cr.execute('SELECT COUNT(*) FROM "%s"' % records._table)
cr.execute(sql.SQL('SELECT COUNT(*) FROM {}').format(records._table))
model.count = cr.fetchone()[0]
@api.constrains('model')
@@ -266,9 +267,9 @@ class IrModel(models.Model):
table = current_model._table
kind = tools.table_kind(self._cr, table)
if kind == 'v':
self._cr.execute('DROP VIEW "%s"' % table)
self._cr.execute(sql.SQL('DROP VIEW {}').format(sql.Identifier(table)))
elif kind == 'r':
self._cr.execute('DROP TABLE "%s" CASCADE' % table)
self._cr.execute(sql.SQL('DROP TABLE {} CASCADE').format(sql.Identifier(table)))
# discard all translations for this model
self._cr.execute("""
DELETE FROM ir_translation
@@ -713,8 +714,8 @@ class IrModelFields(models.Model):
# TODO: Refactor this brol in master
if is_model and tools.column_exists(self._cr, model._table, field.name) and \
tools.table_kind(self._cr, model._table) == 'r':
self._cr.execute('ALTER TABLE "%s" DROP COLUMN "%s" CASCADE' % (
model._table, field.name,
self._cr.execute(sql.SQL('ALTER TABLE {} DROP COLUMN {} CASCADE').format(
sql.Identifier(model._table), sql.Identifier(field.name),
))
if field.state == 'manual' and field.ttype == 'many2many':
rel_name = field.relation_table or (is_model and model._fields[field.name].relation)
@@ -735,7 +736,7 @@ class IrModelFields(models.Model):
(tuple(tables_to_drop), tuple(self.ids)))
tables_to_keep = set(row[0] for row in self._cr.fetchall())
for rel_name in tables_to_drop - tables_to_keep:
self._cr.execute('DROP TABLE "%s"' % rel_name)
self._cr.execute(sql.SQL('DROP TABLE {}').format(rel_name))
return True
@@ -937,9 +938,18 @@ class IrModelFields(models.Model):
# rename column in database, and its corresponding index if present
table, oldname, newname, index, stored = column_rename
if stored:
self._cr.execute('ALTER TABLE "%s" RENAME COLUMN "%s" TO "%s"' % (table, oldname, newname))
self._cr.execute(
sql.SQL('ALTER TABLE {} RENAME COLUMN {} TO {}').format(
sql.Identifier(table),
sql.Identifier(oldname),
sql.Identifier(newname)
))
if index:
self._cr.execute('ALTER INDEX "%s_%s_index" RENAME TO "%s_%s_index"' % (table, oldname, table, newname))
self._cr.execute(
sql.SQL('ALTER INDEX {} RENAME TO {}').format(
sql.Identifier(f'{table}_{oldname}_index'),
sql.Identifier(f'{table}_{newname}_index'),
))
if column_rename or patched_models:
# setup models, this will reload all manual fields in registry
@@ -1450,7 +1460,11 @@ class IrModelConstraint(models.Model):
WHERE cs.contype=%s and cs.conname=%s and cl.relname=%s""",
('f', name, table))
if self._cr.fetchone():
self._cr.execute('ALTER TABLE "%s" DROP CONSTRAINT "%s"' % (table, name),)
self._cr.execute(
sql.SQL('ALTER TABLE {} DROP CONSTRAINT {}').format(
sql.Identifier(table),
sql.Identifier(name)
))
_logger.info('Dropped FK CONSTRAINT %s@%s', name, data.model.model)
if typ == 'u':
@@ -1459,7 +1473,8 @@ class IrModelConstraint(models.Model):
WHERE cs.contype=%s and cs.conname=%s and cl.relname=%s""",
('u', name, table))
if self._cr.fetchone():
self._cr.execute('ALTER TABLE "%s" DROP CONSTRAINT "%s"' % (table, name),)
self._cr.execute(sql.SQL('ALTER TABLE {} DROP CONSTRAINT {}').format(
sql.Identifier(table), sql.Identifier(name)))
_logger.info('Dropped CONSTRAINT %s@%s', name, data.model.model)
self.unlink()
@@ -1581,7 +1596,7 @@ class IrModelRelation(models.Model):
# drop m2m relation tables
for table in to_drop:
self._cr.execute('DROP TABLE "%s" CASCADE' % table,)
self._cr.execute(sql.SQL('DROP TABLE {} CASCADE').format(table))
_logger.info('Dropped table %s', table)
def _reflect_relation(self, model, table, module):
odoo/addons/base/models/ir_sequence.py
+22 -14
View File
@@ -3,6 +3,7 @@
from datetime import datetime, timedelta
import logging
import pytz
from psycopg2 import sql
from odoo import api, fields, models, _
from odoo.exceptions import UserError
@@ -20,10 +21,10 @@ def _create_sequence(cr, seq_name, number_increment, number_next):
def _drop_sequences(cr, seq_names):
""" Drop the PostreSQL sequences if they exist. """
names = ','.join(seq_names)
names = sql.SQL(',').join(map(sql.Identifier, seq_names))
# RESTRICT is the default; it prevents dropping the sequence if an
# object depends on it.
cr.execute("DROP SEQUENCE IF EXISTS %s RESTRICT " % names)
cr.execute(sql.SQL("DROP SEQUENCE IF EXISTS {} RESTRICT").format(names))
def _alter_sequence(cr, seq_name, number_increment=None, number_next=None):
@@ -34,38 +35,45 @@ def _alter_sequence(cr, seq_name, number_increment=None, number_next=None):
if not cr.fetchone():
# sequence is not created yet, we're inside create() so ignore it, will be set later
return
statement = "ALTER SEQUENCE %s" % (seq_name, )
statement = sql.SQL("ALTER SEQUENCE") + sql.Identifier(seq_name)
params = []
if number_increment is not None:
statement += " INCREMENT BY %d" % (number_increment, )
statement += sql.SQL("INCREMENT BY") + sql.Placeholder()
params.append(number_increment)
if number_next is not None:
statement += " RESTART WITH %d" % (number_next, )
cr.execute(statement)
statement += sql.SQL("RESTART WITH") + sql.Placeholder()
params.append(number_next)
cr.execute(statement.join(' '), params)
def _select_nextval(cr, seq_name):
cr.execute("SELECT nextval('%s')" % seq_name)
cr.execute("SELECT nextval(%s)", [seq_name])
return cr.fetchone()
def _update_nogap(self, number_increment):
number_next = self.number_next
self._cr.execute("SELECT number_next FROM %s WHERE id=%s FOR UPDATE NOWAIT" % (self._table, self.id))
self._cr.execute("UPDATE %s SET number_next=number_next+%s WHERE id=%s " % (self._table, number_increment, self.id))
self._cr.execute("SELECT number_next FROM %s WHERE id=%%s FOR UPDATE NOWAIT" % self._table, [self.id])
self._cr.execute("UPDATE %s SET number_next=number_next+%%s WHERE id=%%s " % self._table, (number_increment, self.id))
self.invalidate_cache(['number_next'], [self.id])
return number_next
def _predict_nextval(self, seq_id):
"""Predict next value for PostgreSQL sequence without consuming it"""
# Cannot use currval() as it requires prior call to nextval()
query = """SELECT last_value,
seqname = 'ir_sequence_%s' % seq_id
seqtable = sql.Identifier(seqname)
query = sql.SQL("""SELECT last_value,
(SELECT increment_by
FROM pg_sequences
WHERE sequencename = 'ir_sequence_%(seq_id)s'),
WHERE sequencename = %s),
is_called
FROM ir_sequence_%(seq_id)s"""
FROM {}""")
params = []
if self.env.cr._cnx.server_version < 100000:
query = "SELECT last_value, increment_by, is_called FROM ir_sequence_%(seq_id)s"
self.env.cr.execute(query % {'seq_id': seq_id})
query = sql.SQL("SELECT last_value, increment_by, is_called FROM {}")
params = []
self.env.cr.execute(query.format(seqtable), params)
(last_value, increment_by, is_called) = self.env.cr.fetchone()
if is_called:
return last_value + increment_by
odoo/addons/base/models/ir_translation.py
+2 -2
View File
@@ -137,9 +137,9 @@ class IrTranslationImport(object):
cr.execute(""" INSERT INTO %s(name, lang, res_id, src, type, value, module, state, comments)
SELECT name, lang, res_id, src, type, value, module, state, comments
FROM %s
WHERE %s
WHERE %%s OR noupdate is true
ON CONFLICT DO NOTHING;
""" % (self._model_table, self._table, 'noupdate IS TRUE' if self._overwrite else 'TRUE'))
""" % (self._model_table, self._table), [not self._overwrite])
count += cr.rowcount
if self._debug:
odoo/addons/test_lint/tests/_odoo_checker_sql_injection.py
+38 -29
View File
@@ -34,41 +34,50 @@ class OdooBaseChecker(checkers.BaseChecker):
cursor_name = '.'.join(expr_list)
return cursor_name
def _get_func_name(self, node):
func_name = (
isinstance(node, astroid.Name) and node.name or
isinstance(node, astroid.Attribute) and node.attrname or '')
return func_name
def _allowable(self, node):
"""
:type node: NodeNG
"""
if isinstance(node, astroid.Call):
node = node.func
# self._thing is OK (mostly self._table), self._thing() also because
# it's a common pattern of reports (self._select, self._group_by, ...)
return (isinstance(node, astroid.Attribute)
and isinstance(node.expr, astroid.Name)
and node.attrname.startswith('_')
)
def _check_concatenation(self, node):
is_bin_op = False
if isinstance(node, astroid.BinOp) and node.op in ('%', '+'):
# execute("..." % self._table)
if (isinstance(node.right, astroid.Attribute) and
not node.right.attrname.startswith('_')):
is_bin_op = True
if isinstance(node.right, astroid.Tuple):
for elt in node.right.elts:
if (isinstance(elt, astroid.Call) and
# Ignoring for this case: execute("..." % _foo(...))
not self._get_func_name(elt.func).startswith('_')):
is_bin_op = True
# execute("..." % (self._table, thing))
if not all(map(self._allowable, node.right.elts)):
return True
elif isinstance(node.right, astroid.Dict):
# execute("..." % {'table': self._table}
if not all(self._allowable(v) for _, v in node.right.items):
return True
elif not self._allowable(node.right):
# execute("..." % self._table)
return True
is_format = False
# execute("...".format(self._table, table=self._table))
if (isinstance(node, astroid.Call) and
self._get_func_name(node.func) == 'format'):
for keyword in node.keywords or []:
if (isinstance(keyword.value, astroid.Attribute) and
not keyword.value.attrname.startswith('_')):
is_format = True
break
for argument in node.args or []:
if (isinstance(argument, astroid.Name) and not argument.name.startswith('_')):
is_format = True
break
# check execute("...".format(self._table, table=self._table))
# ignore sql.SQL().format
if isinstance(node, astroid.Call) \
and isinstance(node.func, astroid.Attribute) \
and isinstance(node.func.expr, astroid.Const) \
and node.func.attrname == 'format':
return is_bin_op or is_format
if not all(map(self._allowable, node.args or [])):
return True
if not all(
self._allowable(keyword.value)
for keyword in (node.keywords or [])
):
return True
return False
def _check_sql_injection_risky(self, node):
# Inspired from OCA/pylint-odoo project
odoo/service/db.py
+11 -7
View File
@@ -10,6 +10,7 @@ import traceback
from xml.etree import ElementTree as ET
import zipfile
from psycopg2 import sql
from pytz import country_timezones
from functools import wraps
from contextlib import closing
@@ -105,11 +106,11 @@ def _create_empty_database(name):
cr.autocommit(True) # avoid transaction block
# 'C' collate is only safe with template0, but provides more useful indexes
collate = "LC_COLLATE 'C'" if chosen_template == 'template0' else ""
collate = sql.SQL("LC_COLLATE 'C'" if chosen_template == 'template0' else "")
cr.execute(
"""CREATE DATABASE "%s" ENCODING 'unicode' %s TEMPLATE "%s" """ %
(name, collate, chosen_template)
)
sql.SQL("CREATE DATABASE {} ENCODING 'unicode' {} TEMPLATE {}").format(
sql.Identifier(name), collate, sql.Identifier(chosen_template)
))
if odoo.tools.config['unaccent']:
try:
@@ -136,7 +137,10 @@ def exp_duplicate_database(db_original_name, db_name):
with closing(db.cursor()) as cr:
cr.autocommit(True) # avoid transaction block
_drop_conn(cr, db_original_name)
cr.execute("""CREATE DATABASE "%s" ENCODING 'unicode' TEMPLATE "%s" """ % (db_name, db_original_name))
cr.execute(sql.SQL("CREATE DATABASE {} ENCODING 'unicode' TEMPLATE {}").format(
sql.Identifier(db_name),
sql.Identifier(db_original_name)
))
registry = odoo.modules.registry.Registry.new(db_name)
with registry.cursor() as cr:
@@ -179,7 +183,7 @@ def exp_drop(db_name):
_drop_conn(cr, db_name)
try:
cr.execute('DROP DATABASE "%s"' % db_name)
cr.execute(sql.SQL('DROP DATABASE {}').format(db_name))
except Exception as e:
_logger.info('DROP DB: %s failed:\n%s', db_name, e)
raise Exception("Couldn't drop database %s: %s" % (db_name, e))
@@ -323,7 +327,7 @@ def exp_rename(old_name, new_name):
cr.autocommit(True) # avoid transaction block
_drop_conn(cr, old_name)
try:
cr.execute('ALTER DATABASE "%s" RENAME TO "%s"' % (old_name, new_name))
cr.execute(sql.SQL('ALTER DATABASE {} RENAME TO {}').format(old_name, new_name))
_logger.info('RENAME DB: %s -> %s', old_name, new_name)
except Exception as e:
_logger.info('RENAME DB: %s -> %s failed:\n%s', old_name, new_name, e)