Fixed #29199 -- Fixed crash when database user password contains @ sign on Oracle.

Thanks Shane Allgeier for the report and Tim Graham for the review.

2 files changed, 26 insertions, 10 deletions

diff --git a/django/db/backends/oracle/base.py b/django/db/backends/oracle/base.py
index 488a1b08ab..2d953af2e4 100644
--- a/django/db/backends/oracle/base.py
+++ b/django/db/backends/oracle/base.py
@@ -185,18 +185,16 @@ class DatabaseWrapper(BaseDatabaseWrapper):
         use_returning_into = self.settings_dict["OPTIONS"].get('use_returning_into', True)
         self.features.can_return_id_from_insert = use_returning_into
 
-    def _connect_string(self):
+    def _dsn(self):
         settings_dict = self.settings_dict
         if not settings_dict['HOST'].strip():
             settings_dict['HOST'] = 'localhost'
         if settings_dict['PORT']:
-            dsn = Database.makedsn(settings_dict['HOST'],
-                                   int(settings_dict['PORT']),
-                                   settings_dict['NAME'])
-        else:
-            dsn = settings_dict['NAME']
-        return "%s/%s@%s" % (settings_dict['USER'],
-                             settings_dict['PASSWORD'], dsn)
+            return Database.makedsn(settings_dict['HOST'], int(settings_dict['PORT']), settings_dict['NAME'])
+        return settings_dict['NAME']
+
+    def _connect_string(self):
+        return '%s/\\"%s\\"@%s' % (self.settings_dict['USER'], self.settings_dict['PASSWORD'], self._dsn())
 
     def get_connection_params(self):
         conn_params = self.settings_dict['OPTIONS'].copy()
@@ -205,7 +203,12 @@ class DatabaseWrapper(BaseDatabaseWrapper):
         return conn_params
 
     def get_new_connection(self, conn_params):
-        return Database.connect(self._connect_string(), **conn_params)
+        return Database.connect(
+            user=self.settings_dict['USER'],
+            password=self.settings_dict['PASSWORD'],
+            dsn=self._dsn(),
+            **conn_params,
+        )
 
     def init_connection_state(self):
         cursor = self.create_cursor()
diff --git a/tests/backends/oracle/tests.py b/tests/backends/oracle/tests.py
index 6896419778..77de9cf279 100644
--- a/tests/backends/oracle/tests.py
+++ b/tests/backends/oracle/tests.py
@@ -57,7 +57,7 @@ class Tests(unittest.TestCase):
 
 
 @unittest.skipUnless(connection.vendor == 'oracle', 'Oracle tests')
-class HiddenNoDataFoundExceptionTest(TransactionTestCase):
+class TransactionalTests(TransactionTestCase):
     available_apps = ['backends']
 
     def test_hidden_no_data_found_exception(self):
@@ -83,3 +83,16 @@ class HiddenNoDataFoundExceptionTest(TransactionTestCase):
         finally:
             with connection.cursor() as cursor:
                 cursor.execute('DROP TRIGGER "TRG_NO_DATA_FOUND"')
+
+    def test_password_with_at_sign(self):
+        old_password = connection.settings_dict['PASSWORD']
+        connection.settings_dict['PASSWORD'] = 'p@ssword'
+        try:
+            self.assertIn('/\\"p@ssword\\"@', connection._connect_string())
+            with self.assertRaises(DatabaseError) as context:
+                connection.cursor()
+            # Database exception: "ORA-01017: invalid username/password" is
+            # expected.
+            self.assertIn('ORA-01017', context.exception.args[0].message)
+        finally:
+            connection.settings_dict['PASSWORD'] = old_password