Process URL encoded password in connection string (#356)

Co-authored-by: Lev Gorodetskiy <[email protected]>

CHANGELOG.rst

@@ -32,6 +32,7 @@ Changelog
 
 - 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 processing URL encoded password in connection string
 - 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)

CONTRIBUTORS.rst

@@ -28,6 +28,7 @@ Contributors
 * Sang-Heon Jeon ``@lntuition``
 * Jong-Yeop Park ``@pjongy``
 * ``@sm0k``
+* Lev Gorodetskiy ``@droserasprout``
 
 Special Thanks
 ==============

docs/databases.rst

@@ -23,6 +23,14 @@ The form is:
 
 :samp:`{DB_TYPE}://{USERNAME}:{PASSWORD}@{HOST}:{PORT}/{DB_NAME}?{PARAM1}=value&{PARAM2}=value`
 
+If password contains special characters it need to be URL encoded:
+
+.. code-block::  python3
+
+    >>> import urllib.parse
+    >>> urllib.parse.quote_plus("kx%jj5/g")
+    'kx%25jj5%2Fg'
+
 The supported ``DB_TYPE``:
 
 ``sqlite``:

tests/backends/test_db_url.py

@@ -108,6 +108,22 @@ def test_postgres_basic(self):
             },
         )
 
+    def test_postgres_encoded_password(self):
+        res = expand_db_url("postgres://postgres:kx%25jj5%[email protected]:54321/test")
+        self.assertDictEqual(
+            res,
+            {
+                "engine": "tortoise.backends.asyncpg",
+                "credentials": {
+                    "database": "test",
+                    "host": "127.0.0.1",
+                    "password": "kx%jj5/g",
+                    "port": 54321,
+                    "user": "postgres",
+                },
+            },
+        )
+
     def test_postgres_no_db(self):
         res = expand_db_url("postgres://postgres:[email protected]:54321")
         self.assertDictEqual(
@@ -199,6 +215,24 @@ def test_mysql_basic(self):
             },
         )
 
+    def test_mysql_encoded_password(self):
+        res = expand_db_url("mysql://root:kx%25jj5%[email protected]:33060/test")
+        self.assertEqual(
+            res,
+            {
+                "engine": "tortoise.backends.mysql",
+                "credentials": {
+                    "database": "test",
+                    "host": "127.0.0.1",
+                    "password": "kx%jj5/g",
+                    "port": 33060,
+                    "user": "root",
+                    "charset": "utf8mb4",
+                    "sql_mode": "STRICT_TRANS_TABLES",
+                },
+            },
+        )
+
     def test_mysql_no_db(self):
         res = expand_db_url("mysql://root:@127.0.0.1:33060")
         self.assertEqual(

tortoise/backends/base/config_generator.py

@@ -106,7 +106,9 @@ def expand_db_url(db_url: str, testing: bool = False) -> dict:
     if vmap.get("password"):
         # asyncpg accepts None for password, but aiomysql not
         params[vmap["password"]] = (
-            None if (not url.password and db_backend == "postgres") else str(url.password or "")
+            None
+            if (not url.password and db_backend == "postgres")
+            else urlparse.unquote_plus(url.password or "")
         )
 
     return {"engine": db["engine"], "credentials": params}