CVE-2020-11010 (Python, published 2020-04-20T22:15Z)
Description: value: In Tortoise ORM before versions 0.15.23 and 0.16.6, various forms of SQL injection have been found for MySQL and when filtering or doing mass-updates on char/text fields. SQLite & PostgreSQL are only affected when filtering with contains, starts_with, or ends_with filters (and their case-insensitive counterparts).
Severity: MEDIUM (CVSS v3 base score: 8.8)
Weakness: CWE-89 \u2014 Improper Neutralization of Special Elements used in an SQL Command ('SQL Injection')
The product constructs all or part of an SQL command using externally-influenced input from an upstream component, but it does not neutralize or incorrectly neutralizes special elements that could modify the intended SQL command when it is sent to a downstream component. Without sufficient removal or quoting of SQL syntax in user-controllable inputs, the generated SQL query can cause those inputs to be interpreted as SQL instead of ordinary user data.
Repository: https://github.com/tortoise/tortoise-orm
Files changed:
CHANGELOG.rst
tests/test_fuzz.py
tortoise/backends/mysql/executor.py
tortoise/filters.py
Commit message: Fixed various SQL generation issues


Vulnerable code:
from tortoise.fields import BigIntField, Field, IntField, SmallIntField
def mysql_contains(field: Field, value: str) -> Criterion:
    return functions.Cast(field, SqlTypes.CHAR).like(f"%{value}%")
def mysql_starts_with(field: Field, value: str) -> Criterion:
    return functions.Cast(field, SqlTypes.CHAR).like(f"{value}%")
def mysql_ends_with(field: Field, value: str) -> Criterion:
    return functions.Cast(field, SqlTypes.CHAR).like(f"%{value}")
def mysql_insensitive_exact(field: Field, value: str) -> Criterion:
    return functions.Upper(functions.Cast(field, SqlTypes.CHAR)).eq(functions.Upper(f"{value}"))
def mysql_insensitive_contains(field: Field, value: str) -> Criterion:
    return functions.Upper(functions.Cast(field, SqlTypes.CHAR)).like(functions.Upper(f"%{value}%"))
def mysql_insensitive_starts_with(field: Field, value: str) -> Criterion:
    return functions.Upper(functions.Cast(field, SqlTypes.CHAR)).like(functions.Upper(f"{value}%"))
def mysql_insensitive_ends_with(field: Field, value: str) -> Criterion:
    return functions.Upper(functions.Cast(field, SqlTypes.CHAR)).like(functions.Upper(f"%{value}"))
from pypika.terms import BasicCriterion, Criterion, Equality, Term, ValueWrapper
    return field.like(f"%{value}%")
    return field.like(f"{value}%")
    return field.like(f"%{value}")
    return Upper(field).eq(Upper(f"{value}"))
    return Upper(field).like(Upper(f"%{value}%"))
    return Upper(field).like(Upper(f"{value}%"))
    return Upper(field).like(Upper(f"%{value}"))

Fixed code:
- Fixed SQL injection issue in MySQL
- Fixed SQL injection issues in MySQL when using ``contains``, ``starts_with`` or ``ends_with`` filters (and their case-insensitive counterparts)
- Fixed malformed SQL for PostgreSQL and SQLite when using ``contains``, ``starts_with`` or ``ends_with`` filters (and their case-insensitive counterparts)
from tests.testmodels import CharFields
from tortoise.contrib import test

DODGY_STRINGS = [
    "a/",
    "a\\",
    "a\\'",
    "a\\x39",
    "a'",
    '"',
    "‘a",
    "a’",
    "‘a’",
    "a/a",
    "a\\a",
    "0x39",
    "%a%",
    "_a_",
    "WC[R]S123456",
    "\x01!\U00043198",
    "\x02\U0006501c",
    "\x03㊿\U000e90ff\U0007d718\x16'%\U000b712a(\x16",
    "\x03\U000d435e\U000aa4cb)\U000fe59b",
    "\x05\x10\U0009417f\U000f22e3\U000a5932🔈\U000a5e47\x18\U0006c16b\x05",
    "\nꝢ$\x17\r\x17\U00014dc2嵋0\U0010fda8\U00041dfa",
    "\x0c\U000d4858",
    "\r",
    '\r\r\U0006c50e\U000e309aᕫ%"\U00105213\U0007ba4b\x03\x0c',
    "\rꝢ$\x17\r\x17\U00014dc2嵋0\U0010fda8\U00041dfa",
    "\x0e\x0e",
    "\x0f\uf308𡉙\x1f\U0008ceaf\x19\U000f156b(\U0006c5b0\U0003881c\U0004b76a\U0010b7a2*+\x1b\x19$\U000f643f,(\U000b7e06",
    "\x14\x14",
    "\x14\U000b45e4.\x19\x01,\U00058aa5\U0008da94\U000bb53e\x10\U000a0328%\U0008e967",
    "\x14\U000eb331",
    "\x17\x17%\x12\U0008c069\x18\x10(\x1f\x0f",
    "\x17\x17(\U0008c069\x18\x10(\x1f\x0f",
    "\x17\x17\U000a084e\U0008c069\x18\x10(\x1f\x0f",
    "\x17((\U0008c069\x18\x10(\x1f\x0f",
    "\x17\U0006083e\x18",
    "\x17\U000ef05e%\x12\U0008c069\x18\x10(\x1f\x0f",
    '\x17\U00108f29\x18\x1c"\x18',
    "\x19\x19\x19",
    "\x19\x19-",
    "\x19\x19-\U000a0865",
    "\x1b",
    "\x1b\x19-\U000a0865",
    '\x1d\x19+\x1c㈏\U000b0305\U000ffbf2\x1b+\U000ff7bf"\U000557c3\x1c%\n',
    '\x1d\x19+\x1c\U000ff7bf\U000b0305\U000ffbf2\x1b+\U000ff7bf"\U000557c3\x1c%\n',
    '\x1d\x19㈏\x1c㈏\U000b0305\U000ffbf2\x1b+\U000ff7bf"\U000557c3\x1c%\n',
    '\x1d\x19㈏\x1c\U000566bf\U000b0305\U000ffbf2\x1b+\U000ff7bf"\U000557c3\x1c%\n',
    "\x1d\U0005f530",
    "\x1f",
    "\x1f\x18",
    "\x1f\x1f🔈",
    "\x1f\x1f🔈\x1f\U000a5932🔈\U000a5e47\x18\U0006c16b\x05",
    "\x1f\x1f🔈\U000f22e3\U000a5932🔈\U000a5e47\x18\U0006c16b\x05",
    "\x1f\uf308𡉙\x1f\U0008ceaf\x19\U000f156b(\U0006c5b0\U0003881c\U0004b76a\U0010b7a2*+\x1b\x19$\U000f643f,(\U000b7e06",
    "\x1f𛉅\U001086b3\x0b\x1b\U00077711\U00057223\U0005e650婯\x1d0\U000c0272\x02\x15\U000d159c\U0005997e!\x04&\x04",
    "\x1f\U00077711\U001086b3\x0b\x1b\U00077711\U00057223\U0005e650婯\x1d0\U000c0272\x02\x15\U000d159c\U0005997e!\x04&\x04",
    "\x1f\U000a0850\U0009417f\U000f22e3\U000a5932🔈\U000a5e47\x18\U0006c16b\x05",
    "%\U000e6f0c",
    "&",
    "(((\U0008c069\x18\x10(\x1f\x0f",
    "*\x10'\U0001ea89\U0006a5fe\U00097b9b\x1e",
    "+/",
    "/𬍘&\U00059587+\n\U0003a4ef\x06\U0004675f\x12\U000bfa73\x14\x02(",
    "0",
    "滕'\u16faꮭ\U00041a44\U000d04ba\U000d341c\n'$,\U000bac0b\U000446f8\U000ff86e(",
    "𩣂𩣂\uf580\U000508c8𩣂\U00041150\uf580\x1c",
    "𩣂\U0005215e\uf580\U000508c8𩣂\U00041150\uf580\x1c",
    "\U0003b0da",
    "\U0003ffe5*\n\U000f9326,",
    "\U00050c3e''",
    "\U00050c3e'\U00050c3e",
    "\U0005215e",
    "\U0005215e\U0005215e\x18\U000508c8𩣂\U00041150\uf580\x1c",
    "\U0005215e\U0005215e\uf580\U000508c8𩣂\U00041150\uf580\x1c",
    "\U00059504\U000a33bc\x18\x1f\U000b3017\U000643a3\x18\U000ea429\U000af53c!\U000bcc8f\U000606df",
    "\U0005b823\U0007d224",
    "\U0007bf54\U0001e97a\x08\x18\x04\x06\U000c4329淪",
    "\U0008d96d\x02\U0006d816",
    "\U0009601b\U000b210a\U00058370",
    "\U000965f7'",
    "\U000965f7'\U00050c3e",
    "\U000a9760\U00108859\x0c\r\U00019fbb\U00045885靜$!\U00074df5\x1a\U000c9c7d\U0004bb28\x08\x19\U00099df6+\x1c!\U0003d75f\U0003f457\U0001352e/\U000495db\U000b6234(",
    "\U000aee91\x1c\x1f\U0001cac6\x08\x1d",
    "\U000af7bd\x17",
    "\U000e6f0c\U000e6f0c",
    "\U000f01c8\x0e",
]


class TestFuzz(test.TestCase):
    async def test_char_fuzz(self):
        for char in DODGY_STRINGS:
            # print(repr(char))

            # Create
            obj1 = await CharFields.create(char=char)

            # Get-by-pk, and confirm that reading is correct
            obj2 = await CharFields.get(pk=obj1.pk)
            self.assertEqual(char, obj2.char)

            # Update data using a queryset, confirm that update is correct
            await CharFields.filter(pk=obj1.pk).update(char="a")
            await CharFields.filter(pk=obj1.pk).update(char=char)
            obj3 = await CharFields.get(pk=obj1.pk)
            self.assertEqual(char, obj3.char)

            # Filter by value in queryset, and confirm that it fetched the right one
            obj4 = await CharFields.get(pk=obj1.pk, char=char)
            self.assertEqual(obj1.pk, obj4.pk)
            self.assertEqual(char, obj4.char)

            # LIKE statements are not strict, so require all of these to match
            obj5 = await CharFields.get(
                pk=obj1.pk,
                char__startswith=char,
                char__endswith=char,
                char__contains=char,
                char__istartswith=char,
                char__iendswith=char,
                char__icontains=char,
            )
            self.assertEqual(obj1.pk, obj5.pk)
            self.assertEqual(char, obj5.char)
from tortoise.fields import BigIntField, IntField, SmallIntField
    Like,
    Term,
    ValueWrapper,
    format_quotes,
class StrWrapper(ValueWrapper):  # type: ignore
    """
    Naive str wrapper that doesn't use the monkey-patched pypika ValueWraper for MySQL
    """
    def get_value_sql(self, **kwargs):
        quote_char = kwargs.get("secondary_quote_char") or ""
        value = self.value.replace(quote_char, quote_char * 2)
        return format_quotes(value, quote_char)
def escape_like(val: str) -> str:
    return val.replace("\\", "\\\\\\\\").replace("%", "\\%").replace("_", "\\_")
def mysql_contains(field: Term, value: str) -> Criterion:
    return Like(
        functions.Cast(field, SqlTypes.CHAR), StrWrapper(f"%{escape_like(value)}%"), escape=""
    )
def mysql_starts_with(field: Term, value: str) -> Criterion:
    return Like(
        functions.Cast(field, SqlTypes.CHAR), StrWrapper(f"{escape_like(value)}%"), escape=""
    )
def mysql_ends_with(field: Term, value: str) -> Criterion:
    return Like(
        functions.Cast(field, SqlTypes.CHAR), StrWrapper(f"%{escape_like(value)}"), escape=""
    )
def mysql_insensitive_exact(field: Term, value: str) -> Criterion:
    return functions.Upper(functions.Cast(field, SqlTypes.CHAR)).eq(functions.Upper(str(value)))

def mysql_insensitive_contains(field: Term, value: str) -> Criterion:
    return Like(
        functions.Upper(functions.Cast(field, SqlTypes.CHAR)),
        functions.Upper(StrWrapper(f"%{escape_like(value)}%")),
        escape="",
    )


def mysql_insensitive_starts_with(field: Term, value: str) -> Criterion:
    return Like(
        functions.Upper(functions.Cast(field, SqlTypes.CHAR)),
        functions.Upper(StrWrapper(f"{escape_like(value)}%")),
        escape="",
    )


def mysql_insensitive_ends_with(field: Term, value: str) -> Criterion:
    return Like(
        functions.Upper(functions.Cast(field, SqlTypes.CHAR)),
        functions.Upper(StrWrapper(f"%{escape_like(value)}")),
        escape="",
    )
from pypika.terms import (
    BasicCriterion,
    Criterion,
    Enum,
    Equality,
    Term,
    ValueWrapper,
    basestring,
    date,
    format_quotes,
)
##############################################################################
# Here we monkey-patch PyPika Valuewrapper to behave differently for MySQL
##############################################################################


def get_value_sql(self, **kwargs):  # pragma: nocoverage
    quote_char = kwargs.get("secondary_quote_char") or ""
    dialect = kwargs.get("dialect")
    if dialect:
        dialect = dialect.value

    if isinstance(self.value, Term):
        return self.value.get_sql(**kwargs)
    if isinstance(self.value, Enum):
        return self.value.value
    if isinstance(self.value, date):
        value = self.value.isoformat()
        return format_quotes(value, quote_char)
    if isinstance(self.value, basestring):
        value = self.value.replace(quote_char, quote_char * 2)
        if dialect == "mysql":
            value = value.replace("\\", "\\\\")
        return format_quotes(value, quote_char)
    if isinstance(self.value, bool):
        return str.lower(str(self.value))
    if self.value is None:
        return "null"
    return str(self.value)


ValueWrapper.get_value_sql = get_value_sql
##############################################################################


class Like(BasicCriterion):  # type: ignore
    def __init__(self, left, right, alias=None, escape=" ESCAPE '\\'") -> None:
        """
        A Like that supports an ESCAPE clause
        """
        super().__init__(" LIKE ", left, right, alias=alias)
        self.escape = escape

    def get_sql(self, quote_char='"', with_alias=False, **kwargs):
        sql = "{left}{comparator}{right}{escape}".format(
            comparator=self.comparator,
            left=self.left.get_sql(quote_char=quote_char, **kwargs),
            right=self.right.get_sql(quote_char=quote_char, **kwargs),
            escape=self.escape,
        )
        if with_alias and self.alias:  # pragma: nocoverage
            return '{sql} "{alias}"'.format(sql=sql, alias=self.alias)
        return sql


def escape_val(val: Any) -> Any:
    if isinstance(val, str):
        print(val)
        return val.replace("\\", "\\\\")
    return val


def escape_like(val: str) -> str:
    return val.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")


    return Like(field, field.wrap_constant(f"%{escape_like(value)}%"))
    return Like(field, field.wrap_constant(f"{escape_like(value)}%"))
    return Like(field, field.wrap_constant(f"%{escape_like(value)}"))
    return Upper(field).eq(Upper(str(value)))
    return Like(Upper(field), field.wrap_constant(Upper(f"%{escape_like(value)}%")))
    return Like(Upper(field), field.wrap_constant(Upper(f"{escape_like(value)}%")))
    return Like(Upper(field), field.wrap_constant(Upper(f"%{escape_like(value)}")))