View source for Python Sample SQLite

It is possible to access [https://sqlite.org/ SQLite] files in Truxton without first exporting them to your local system.
Almost always, you will have to export the SQLite file from Truxton to a temporary folder and run your scripts on it.
This is because it is not worth modifying the Python code to open the SQLite file in Truxton's [[Depot|storage]].
It is just easier to export the file from Truxton, run your code on it, delete the exported file, continue on with life.

When your holdings get large, the performance hit of exporting all those files becomes worth the effort to modify the Python script.
When you get there, here's how to query SQLite files directly from Truxton.

=APSW=
Since the [https://docs.python.org/3/library/sqlite3.html Python DB-API] only supports establishing a connection to a SQLite database by providing a file name, we have to create our own filesystem to read from the file in Truxton.
SQLite has the concept of virtual filesystems but DB-API doesn't support it.
However, another SQLite library called [https://rogerbinns.github.io/apsw/ APSW] does.
Install it using pip:
<source lang="bat">
pip install apsw
</source>

You can verify that it is working by running the following script:
<source lang="python">
import sys
import apsw
import apsw.ext

def main() -> None:

  print("      Using APSW file", apsw.__file__)
  print("         APSW version", apsw.apsw_version())
  print("SQLite header version", apsw.SQLITE_VERSION_NUMBER)
  print("   SQLite lib version", apsw.sqlite_lib_version())
  print("   Using amalgamation", apsw.using_amalgamation)

  return None

if __name__ == "__main__":
  sys.exit(main())
</source>

=SQLite in Truxton=
Here's a script to query a SQLite database that is stored in Truxton.
Instead of opening a file by name, we open it using the file's identifier.
This corresponds to the <code>ID</code> column of the <code><nowiki>[</nowiki>[[File Table|File]]<nowiki>]</nowiki></code> table.

<source lang="python">
import sys
import os
import apsw

from pathlib import Path

sys.path.append('C:/Program Files/Truxton/SDK')
import truxton

def main() -> None:

  truxton_virtual_filesystem = TruxtonVFS()
  truxton_virtual_filesystem.Truxton = truxton.create()

  # Open the SQLite file using it's id
  database = apsw.Connection( "65d9d912-7328-d2e7-d511-f909000000c6", flags = SQLITE_OPEN_READONLY, vfs = "truxton")
  
  for row in database.execute("SELECT Z_UUID FROM Z_METADATA"):
    print(row)
  
  database.close()

  return None

SQLITE_OPEN_READONLY = 0x00000001

class TruxtonVFS(apsw.VFS):
  
  Truxton = None

  def __init__(self, vfsname="truxton", basevfs=""):
    self.vfs_name = vfsname
    self.base_vfs = basevfs
    super().__init__(self.vfs_name, self.base_vfs)

  def xOpen(self, fname, flags):
    # fname is of type apsw.URIFilename object
    truxton_guid = os.path.basename(fname.filename())
    opened_file = self.Truxton.getfileid(truxton_guid)
    return TruxtonVFSFile(f = opened_file, name = truxton_guid, flags = flags)

  def xAccess(self, pathname, flags):
    return False
  
  def xDelete(self, filename, syncdir):
    pass
    
class TruxtonVFSFile(apsw.VFSFile):
    def __init__(self, f: truxton.TruxtonFileIO, name, flags):
        """
        Truxton VFS File object

        Args:
          * f: TruxtonFileIO object returned by Truxton.getfileid
          * name: name of the file
          * flags: SQLite open flags
        """

        self.f = f
        self.flags = flags
        self.name = name
        self.mode = "rb"

    def xRead(self, amount, offset) -> bytes:
        self.f.seek(offset)
        data = self.f.read(amount)
        return data

    def xFileControl(self, *args):
        return True

    def xDeviceCharacteristics(self):
        return 4096

    def xCheckReservedLock(self):
        return False

    def xLock(self, level):
        pass

    def xUnlock(self, level):
        pass

    def xSectorSize(self):
        return self.f.block_size

    def xClose(self):
        self.f.close()
        pass

    def xFileSize(self):
        pos = self.f.tell()
        self.f.seek(0, 2)
        size = self.f.tell()
        self.f.seek(pos)
        return size

    def xSync(self, flags):
        pass

    def xTruncate(self, newsize):
        pass

    def xWrite(self, data, offset):
        pass

if __name__ == "__main__":
  sys.exit(main())
</source>

=Exploiting All SQLite Files in Truxton=
Using [[TruxtonEnumeration|enumeration]], we can find all files of a particular type and query them without ever having to write to a file.
This is the <code>AllSpeedTestDatabases.py</code> sample found here.
<source lang="python">
import sys
import os
import apsw
import sqlite3
from pathlib import Path

sys.path.append('C:/Program Files/Truxton/SDK')
import truxton

# Notice that we keep the database parameter typeless
# This is so we can use the same function with an APSW Connection or a splite3 cursor.
# The exploitation code doesn't know nor care which SQLite technology is used.
def exploit_speed_test_database( database ) -> None:
 for row in database.execute("SELECT [ZDATE],[ZLATITUDE],[ZLONGITUDE],[ZALTITUDE],[ZWIFISSID] FROM [ZSPEEDTESTRESULT]"):
   # Your exploitation code goes here
   print(row)
 return None  

def process_file(file_information: truxton.TruxtonFileIO) -> None: 
 # Separate the opening of the SQLite file from the exploitation of it.

 # Open the SQLite database
 database = apsw.Connection( file_information.id, flags = SQLITE_OPEN_READONLY, vfs = "truxton")

 # Exploit the SQLite database
 exploit_speed_test_database( database )
  
 database.close()
 return None

def search_files(media: truxton.TruxtonMedia) -> None:
 files_in_media = truxton_virtual_filesystem.Truxton.newenumerator()
 files_in_media.scope = truxton.Type_Media
 files_in_media.scopeid = media.id
 files_in_media.target = truxton.Type_File
 files_in_media.filetype = truxton.Type_Speedtest_Database

 for file_information in files_in_media:
  process_file(file_information)
  
 return None
  
def search_media(media: truxton.TruxtonMedia) -> None:
 search_files(media)
 return None
 
def search_investigation(investigation: truxton.TruxtonInvestigation) -> None:
 investigation_media = truxton_virtual_filesystem.Truxton.newenumerator()
 investigation_media.scope = truxton.Type_Investigation
 investigation_media.scopeid = investigation.id
 investigation_media.target = truxton.Type_Media

 for media in investigation_media:
  search_media(media)
 return None
  
SQLITE_OPEN_READONLY = 0x00000001

class TruxtonVFS(apsw.VFS):
  
  Truxton = None

  def __init__(self, vfsname="truxton", basevfs=""):
    self.vfs_name = vfsname
    self.base_vfs = basevfs
    super().__init__(self.vfs_name, self.base_vfs)

  def xOpen(self, fname, flags):
    # fname is of type apsw.URIFilename object
    truxton_guid = os.path.basename(fname.filename())
    opened_file = self.Truxton.getfileid(truxton_guid)
    return TruxtonVFSFile(f = opened_file, name = truxton_guid, flags = flags)

  def xAccess(self, pathname, flags):
    return False
  
  def xDelete(self, filename, syncdir):
    pass
    
class TruxtonVFSFile(apsw.VFSFile):
    def __init__(self, f: truxton.TruxtonFileIO, name, flags):
        """
        Truxton VFS File object

        Args:
          * f: TruxtonFileIO object returned by Truxton.getfileid
          * name: name of the file
          * flags: SQLite open flags
        """

        self.f = f
        self.flags = flags
        self.name = name
        self.mode = "rb"

    def xRead(self, amount, offset) -> bytes:
        self.f.seek(offset)
        data = self.f.read(amount)
        return data

    def xFileControl(self, *args):
        return True

    def xDeviceCharacteristics(self):
        return 4096

    def xCheckReservedLock(self):
        return False

    def xLock(self, level):
        pass

    def xUnlock(self, level):
        pass

    def xSectorSize(self):
        return self.f.block_size

    def xClose(self):
        self.f.close()
        pass

    def xFileSize(self):
        pos = self.f.tell()
        self.f.seek(0, 2)
        size = self.f.tell()
        self.f.seek(pos)
        return size

    def xSync(self, flags):
        pass

    def xTruncate(self, newsize):
        pass

    def xWrite(self, data, offset):
        pass

# Global Variables
truxton_virtual_filesystem = TruxtonVFS()
truxton_virtual_filesystem.Truxton = truxton.create()

def main() -> None:
 investigations = truxton_virtual_filesystem.Truxton.newenumerator()
 investigations.target = truxton.Type_Investigation

 for investigation in investigations:
  search_investigation(investigation)
  
 return None

if __name__ == "__main__":
  sys.exit(main())
</source>