CVE-2020-11010: Improper Neutralization of Special Elements used in an SQL Command ('SQL Injection') (CWE-89)
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).
CWE description: 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.
CVSS v3 base score: 8.8
Published: 2020-04-20T22:15Z
Repository: https://github.com/tortoise/tortoise-orm
Language: Python
Files changed: CHANGELOG.rst, tests/test_fuzz.py, tortoise/backends/mysql/executor.py, tortoise/filters.py
Commit date: 2020-04-18 16:58:20 +0200
Commit message: Fixed various SQL generation issues

Diff: diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index 3877c0c..ebe4455 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -30,10 +30,13 @@ Changelog
   * If you do a ``<model>.save(update_fields=[...])`` and one of the fields in ``update_fields`` was not in the ``.only(...)``,
     then ``IncompleteInstanceError`` as that field is not available to be updated.
 
 - Fixed bad SQL generation when doing a ``.values()`` query over a Foreign Key
 - Added `<model>.update_from_dict({...})` that will mass update values safely from a dictionary
+- 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)
 
 0.16.5
 ------
 * Moved ``Tortoise.describe_model(<MODEL>, ...)`` to ``<MODEL>.describe(...)``
 * Deprecated ``Tortoise.describe_model()``
diff --git a/tests/test_fuzz.py b/tests/test_fuzz.py
new file mode 100644
index 0000000..58f0923
--- /dev/null
+++ b/tests/test_fuzz.py
@@ -0,0 +1,127 @@
+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)
diff --git a/tortoise/backends/mysql/executor.py b/tortoise/backends/mysql/executor.py
index d03f295..dfd8049 100644
--- a/tortoise/backends/mysql/executor.py
+++ b/tortoise/backends/mysql/executor.py
@@ -2,48 +2,85 @@ from pypika import Parameter, functions
 from pypika.enums import SqlTypes
 from pypika.terms import Criterion
 
 from tortoise import Model
 from tortoise.backends.base.executor import BaseExecutor
-from tortoise.fields import BigIntField, Field, IntField, SmallIntField
+from tortoise.fields import BigIntField, IntField, SmallIntField
 from tortoise.filters import (
+    Like,
+    Term,
+    ValueWrapper,
     contains,
     ends_with,
+    format_quotes,
     insensitive_contains,
     insensitive_ends_with,
     insensitive_exact,
     insensitive_starts_with,
     starts_with,
 )
 
 
-def mysql_contains(field: Field, value: str) -> Criterion:
-    return functions.Cast(field, SqlTypes.CHAR).like(f"%{value}%")
+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 mysql_starts_with(field: Field, value: str) -> Criterion:
-    return functions.Cast(field, SqlTypes.CHAR).like(f"{value}%")
 
+def escape_like(val: str) -> str:
+    return val.replace("\\", "\\\\\\\\").replace("%", "\\%").replace("_", "\\_")
 
-def mysql_ends_with(field: Field, value: str) -> Criterion:
-    return functions.Cast(field, SqlTypes.CHAR).like(f"%{value}")
 
+def mysql_contains(field: Term, value: str) -> Criterion:
+    return Like(
+        functions.Cast(field, SqlTypes.CHAR), StrWrapper(f"%{escape_like(value)}%"), escape=""
+    )
 
-def mysql_insensitive_exact(field: Field, value: str) -> Criterion:
-    return functions.Upper(functions.Cast(field, SqlTypes.CHAR)).eq(functions.Upper(f"{value}"))
 
+def mysql_starts_with(field: Term, value: str) -> Criterion:
+    return Like(
+        functions.Cast(field, SqlTypes.CHAR), StrWrapper(f"{escape_like(value)}%"), escape=""
+    )
 
-def mysql_insensitive_contains(field: Field, value: str) -> Criterion:
-    return functions.Upper(functions.Cast(field, SqlTypes.CHAR)).like(functions.Upper(f"%{value}%"))
 
+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_starts_with(field: Field, value: str) -> Criterion:
-    return functions.Upper(functions.Cast(field, SqlTypes.CHAR)).like(functions.Upper(f"{value}%"))
 
+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_ends_with(field: Field, value: str) -> Criterion:
-    return functions.Upper(functions.Cast(field, SqlTypes.CHAR)).like(functions.Upper(f"%{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="",
+    )
 
 
 class MySQLExecutor(BaseExecutor):
     FILTER_FUNC_OVERRIDE = {
         contains: mysql_contains,
diff --git a/tortoise/filters.py b/tortoise/filters.py
index 79857f8..c07acea 100644
--- a/tortoise/filters.py
+++ b/tortoise/filters.py
@@ -2,18 +2,93 @@ import operator
 from functools import partial
 from typing import TYPE_CHECKING, Any, Dict, Iterable, Optional, Tuple
 
 from pypika import Table
 from pypika.functions import Upper
-from pypika.terms import BasicCriterion, Criterion, Equality, Term, ValueWrapper
+from pypika.terms import (
+    BasicCriterion,
+    Criterion,
+    Enum,
+    Equality,
+    Term,
+    ValueWrapper,
+    basestring,
+    date,
+    format_quotes,
+)
 
 from tortoise.fields import Field
 from tortoise.fields.relational import BackwardFKRelation, ManyToManyFieldInstance
 
 if TYPE_CHECKING:  # pragma: nocoverage
     from tortoise.models import Model
 
+##############################################################################
+# 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("_", "\\_")
+
+
 ##############################################################################
 # Encoders
 # Should be type: (Any, instance: "Model", field: Field) -> type:
 ##############################################################################
 
@@ -77,35 +152,35 @@ def not_null(field: Term, value: Any) -> Criterion:
         return field.notnull()
     return field.isnull()
 
 
 def contains(field: Term, value: str) -> Criterion:
-    return field.like(f"%{value}%")
+    return Like(field, field.wrap_constant(f"%{escape_like(value)}%"))
 
 
 def starts_with(field: Term, value: str) -> Criterion:
-    return field.like(f"{value}%")
+    return Like(field, field.wrap_constant(f"{escape_like(value)}%"))
 
 
 def ends_with(field: Term, value: str) -> Criterion:
-    return field.like(f"%{value}")
+    return Like(field, field.wrap_constant(f"%{escape_like(value)}"))
 
 
 def insensitive_exact(field: Term, value: str) -> Criterion:
-    return Upper(field).eq(Upper(f"{value}"))
+    return Upper(field).eq(Upper(str(value)))
 
 
 def insensitive_contains(field: Term, value: str) -> Criterion:
-    return Upper(field).like(Upper(f"%{value}%"))
+    return Like(Upper(field), field.wrap_constant(Upper(f"%{escape_like(value)}%")))
 
 
 def insensitive_starts_with(field: Term, value: str) -> Criterion:
-    return Upper(field).like(Upper(f"{value}%"))
+    return Like(Upper(field), field.wrap_constant(Upper(f"{escape_like(value)}%")))
 
 
 def insensitive_ends_with(field: Term, value: str) -> Criterion:
-    return Upper(field).like(Upper(f"%{value}"))
+    return Like(Upper(field), field.wrap_constant(Upper(f"%{escape_like(value)}")))
 
 
 ##############################################################################
 # Filter resolvers
 ##############################################################################

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)}")))