From f364327f4c5be54a0a1e342b7f044fa58832e1a3 Mon Sep 17 00:00:00 2001
From: Waket Zheng <waketzheng@gmail.com>
Date: Fri, 20 Dec 2024 11:19:30 +0800
Subject: [PATCH] refactor: only replace password by star when logging level is
 debug

---
 tortoise/__init__.py               | 47 +++++++++++++++++-------------
 tortoise/backends/sqlite/client.py |  2 +-
 2 files changed, 28 insertions(+), 21 deletions(-)

diff --git a/tortoise/__init__.py b/tortoise/__init__.py
index 2aace9a16..07dad88f8 100644
--- a/tortoise/__init__.py
+++ b/tortoise/__init__.py
@@ -1,15 +1,18 @@
 from __future__ import annotations
 
 import asyncio
+import copy
 import importlib
 import importlib.metadata as importlib_metadata
 import json
+import logging
 import os
 import warnings
 from copy import deepcopy
 from inspect import isclass
 from types import ModuleType
 from typing import Any, Callable, Coroutine, Iterable, Type, cast
+from urllib.parse import quote_plus
 
 from pypika import Query, Table
 
@@ -495,28 +498,32 @@ async def init(
 
         cls.table_name_generator = table_name_generator
 
-        # Mask passwords in logs output
-        passwords = []
-        for name, info in connections_config.items():
-            if isinstance(info, str):
-                info = expand_db_url(info)
-            if password := info.get("credentials", {}).get("password"):
-                passwords.append(password)
-
-        str_connection_config = str(connections_config)
-        for password in passwords:
-            str_connection_config = str_connection_config.replace(
-                password,
-                # Show one third of the password at beginning (may be better for debugging purposes)
-                f"{password[0:len(password) // 3]}***",
+        if logger.isEnabledFor(logging.DEBUG):
+            # Mask passwords in logs output
+            connections_config_copied = copy.deepcopy(connections_config)
+            for name, info in connections_config_copied.items():
+                if is_string := isinstance(info, str):
+                    info_dict = expand_db_url(info)
+                else:
+                    info_dict = info
+                if password := info_dict.get("credentials", {}).get("password"):
+                    # Show one third of the password at beginning (may be better for debugging purposes)
+                    password_star = f"{password[0:len(password) // 3]}***"
+                    if is_string:
+                        if (passwd := ":" + password) in info:
+                            info = info.replace(passwd, ":" + password_star)
+                        else:
+                            # password in db_url may be unquoted
+                            info = info.replace(":" + quote_plus(password), ":" + password_star)
+                        connections_config_copied[name] = info
+                    else:
+                        info["credentials"]["password"] = password_star
+            logger.debug(
+                "Tortoise-ORM startup\n    connections: %s\n    apps: %s",
+                str(connections_config_copied),
+                str(apps_config),
             )
 
-        logger.debug(
-            "Tortoise-ORM startup\n    connections: %s\n    apps: %s",
-            str_connection_config,
-            str(apps_config),
-        )
-
         cls._init_timezone(use_tz, timezone)
         await connections._init(connections_config, _create_db)
         cls._init_apps(apps_config)
diff --git a/tortoise/backends/sqlite/client.py b/tortoise/backends/sqlite/client.py
index ed39c0663..9829671af 100644
--- a/tortoise/backends/sqlite/client.py
+++ b/tortoise/backends/sqlite/client.py
@@ -23,8 +23,8 @@
     Capabilities,
     ConnectionWrapper,
     NestedTransactionContext,
-    TransactionContext,
     T_conn,
+    TransactionContext,
 )
 from tortoise.backends.sqlite.executor import SqliteExecutor
 from tortoise.backends.sqlite.schema_generator import SqliteSchemaGenerator