Close leaked sqlite3 connections (ResourceWarning on Python 3.13+) (#6181)

Python 3.13+ reports sqlite3 connections that are garbage collected
without being closed. The unit tests print a run of
"ResourceWarning: unclosed database" lines on Linux/CPython 3.14:

- SQLAlchemy.close() only returned its connection to the engine's
  pool, so the underlying DBAPI connection (e.g. sqlite3 for
  '-d sqlite://...') stayed open. Dispose the engine on close.
- vulnTest() used "with sqlite3.connect(...)", which commits but does
  not close the connection. Use contextlib.closing().
- test_multibit and test_sqlalchemy never closed what they opened.

Co-authored-by: PiyushSatyarthi <244159432+PiyushSatyarthi@users.noreply.github.com>

lib/core/testing.py

@@ -5,6 +5,7 @@ Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
 See the file 'LICENSE' for copying permission
 """
 
+import contextlib
 import doctest
 import json
 import logging
@@ -206,9 +207,11 @@ def vulnTest(tests=None, label="vuln"):
     os.close(handle)
     cleanups.append(database)
 
-    with sqlite3.connect(database) as conn:
+    # Note: sqlite3's own context manager only commits (it does not close the connection)
+    with contextlib.closing(sqlite3.connect(database)) as conn:
         c = conn.cursor()
         c.executescript(vulnserver.SCHEMA)
+        conn.commit()
 
     handle, request = tempfile.mkstemp(suffix=".req")
     os.close(handle)

lib/utils/sqlalchemy.py

@@ -50,6 +50,7 @@ class SQLAlchemy(GenericConnector):
 
         self.dialect = dialect
         self.address = conf.direct
+        self.engine = None
 
         if conf.dbmsUser:
             self.address = self.address.replace("'%s':" % conf.dbmsUser, "%s:" % _urllib.parse.quote(conf.dbmsUser))
@@ -84,6 +85,7 @@ class SQLAlchemy(GenericConnector):
                 else:
                     engine = _sqlalchemy.create_engine(self.address, connect_args={})
 
+                self.engine = engine
                 self.connector = engine.connect()
             except (TypeError, ValueError) as ex:
                 if "_get_server_version_info" in traceback.format_exc():
@@ -110,6 +112,19 @@ class SQLAlchemy(GenericConnector):
         else:
             raise SqlmapMissingDependence("SQLAlchemy not available (e.g. 'pip%s install SQLAlchemy')" % ('3' if six.PY3 else ""))
 
+    def close(self):
+        try:
+            GenericConnector.close(self)
+        finally:
+            # Note: closing the connection only returns it to the engine's pool (the underlying DBAPI
+            # connection, e.g. sqlite3, stays open) - disposing the engine is what actually releases it
+            if self.engine is not None:
+                try:
+                    self.engine.dispose()
+                except Exception as ex:
+                    logger.debug(getSafeExString(ex))
+                self.engine = None
+
     def fetchall(self):
         try:
             retVal = []

tests/test_multibit.py

@@ -140,8 +140,14 @@ class MultibitTest(unittest.TestCase):
         multibit._page = self._savedPage
         multibit.singleTimeWarnMessage, multibit.singleTimeLogMessage = _SAID
 
-    def _attempt(self, secret, length=None, **kwargs):
+    def _target(self, secret, **kwargs):
         target = _Target(secret, **kwargs)
+        self.addCleanup(target.db.close)
+
+        return target
+
+    def _attempt(self, secret, length=None, **kwargs):
+        target = self._target(secret, **kwargs)
         multibit._page = target.page
 
         return self._read(target, secret, length), target
@@ -231,7 +237,7 @@ class MultibitTest(unittest.TestCase):
         # non-ASCII one, and one whose codepoint has an ASCII low byte (U+0141 -> 0x41 'A') all settle
         # nothing, and none of them may condemn a channel that calibrated cleanly
         for first in ("", u"\u017eaba pod vodom", u"\u0141ukasz Nowak here"):
-            target = _Target(first)
+            target = self._target(first)
             self.assertIsNone(self._read(target, first, len(first) or None))
 
             good = target.secret("plain ascii value that must still come back")
@@ -242,7 +248,7 @@ class MultibitTest(unittest.TestCase):
         # what the channel did - i.e. exactly the way a channel dropping rows fails it. That is the
         # data's doing, and a handful of NULLs in one dump must not retire a working channel
         first = "first value is perfectly readable ascii"
-        target = _Target(first)
+        target = self._target(first)
         self.assertEqual(self._read(target, first, len(first)), first)
 
         for _ in range(MULTIBIT_MAX_FAILURES + 1):
@@ -253,7 +259,7 @@ class MultibitTest(unittest.TestCase):
         self.assertEqual(self._read(target, good, len(good)), good)
 
     def test_unreadable_value_mid_run_keeps_the_channel(self):
-        target = _Target("first value is perfectly readable ascii")
+        target = self._target("first value is perfectly readable ascii")
         self.assertEqual(self._read(target, "first value is perfectly readable ascii", 39), "first value is perfectly readable ascii")
 
         bad = target.secret(u"\u0141ukasz")          # reads back as "Aukasz" - right length, wrong char
@@ -295,7 +301,7 @@ class MultibitTest(unittest.TestCase):
     def test_resumed_partial_value_is_left_to_bisection(self):
         # a resumed value starts mid-string, and multi-bit reads from the first character: it would
         # re-read what is already cached and then fail its own length read-back
-        target = _Target("resuming this one is bisection's job")
+        target = self._target("resuming this one is bisection's job")
         multibit._page = target.page
 
         self.assertIsNone(multibit.attempt("SELECT s FROM v", lambda idx: 'x', 35, None, True))

tests/test_sqlalchemy.py

@@ -58,6 +58,7 @@ class TestSQLAlchemySqlitePath(unittest.TestCase):
 
         sa = SQLAlchemy(dialect="sqlite")
         sa.connect()
+        self.addCleanup(sa.close)
 
         # the reformatted URL must resolve to the exact absolute file (not '//...path')
         self.assertEqual(_sa.engine.url.make_url(sa.address).database, os.path.abspath(dbfile))