Add option to mount the internal flash as a FUSE filesystem

README.md

@@ -294,6 +294,12 @@ python mtk printgpt
 ```
 
 
+Mount the flash as a filesystem
+
+```
+python mtk fs /mnt/mtk
+```
+
 ### Write flash
 (use --preloader for brom)
 

mtk

@@ -14,6 +14,7 @@ cmds = {
     "rf": "Read whole flash to file",
     "rs": "Read sectors starting at start_sector to filename",
     "ro": "Read flash starting at offset to filename",
+    "fs": "Mount the device as a FUSE filesystem",
     "w": "Write partition from filename",
     "wf": "Write flash from filename",
     "wl": "Write partitions from directory path to flash",
@@ -55,7 +56,7 @@ if __name__ == '__main__':
     parser = argparse.ArgumentParser(description=info)
     subparsers = parser.add_subparsers(dest="cmd",
                                        help='Valid commands are: \n' +
-                                            'printgpt, gpt, r, rl, rf, rs, w, wf, wl, e, es, footer, reset, \n' +
+                                            'printgpt, gpt, r, rl, rf, fs, rs, w, wf, wl, e, es, footer, reset, \n' +
                                             'dumpbrom, dumpsram, dumppreloader, payload, crash, brute, gettargetconfig, \n' +
                                             'peek, stage, plstage, da, script\n')
 
@@ -67,6 +68,7 @@ if __name__ == '__main__':
     parser_rf = subparsers.add_parser("rf", help="Read whole flash to file")
     parser_rs = subparsers.add_parser("rs", help="Read sectors starting at start_sector to filename")
     parser_ro = subparsers.add_parser("ro", help="Read flash starting at offset to filename")
+    parser_fs = subparsers.add_parser("fs", help="Mount the device as a FUSE filesystem")
     parser_w = subparsers.add_parser("w", help="Write partition from filename")
     parser_wf = subparsers.add_parser("wf", help="Write flash from filename")
     parser_wl = subparsers.add_parser("wl", help="Write partitions from directory path to flash")
@@ -405,6 +407,9 @@ if __name__ == '__main__':
     parser_ro.add_argument('--auth', type=str, help="Use auth file (auth_sv5.auth)")
     parser_ro.add_argument('--cert', type=str, help="Use cert file")
 
+    parser_fs.add_argument('mountpoint', help='Directory to mount the FUSE filesystem in')
+    parser_fs.add_argument('--rw', help='Mount the filesystem as writeable', default=False, action='store_true')
+
     parser_w.add_argument('partitionname', help='Partition to write (separate by comma for multiple partitions)')
     parser_w.add_argument('filename', help='Filename for writing (separate by comma for multiple filenames)')
     parser_w.add_argument('--loader', type=str, help='Use specific DA loader, disable autodetection')

mtkclient/Library/DA/mtk_da_handler.py

@@ -9,6 +9,8 @@
 from mtkclient.Library.error import ErrorHandler
 from mtkclient.Library.utils import progress
 from mtkclient.config.brom_config import efuse, damodes
+from mtkclient.Library.Filesystem.mtkdafs import MtkDaFS
+from fuse import FUSE
 
 
 class DA_handler(metaclass=LogBase):
@@ -681,6 +683,9 @@ def handle_da_cmds(self, mtk, cmd: str, args):
                 print(f"Dumped offset {hex(start)} with length {hex(length)} as {filename}.")
             else:
                 print(f"Failed to dump offset {hex(start)} with length {hex(length)} as {filename}.")
+        elif cmd == "fs":
+            print(f'Mounting FUSE fs at: {args.mountpoint}...')
+            fs = FUSE(MtkDaFS(self, rw=args.rw), mountpoint=args.mountpoint, foreground=True, allow_other=True, nothreads=True)
         elif cmd == "footer":
             filename = args.filename
             self.da_footer(filename=filename)

mtkclient/Library/Filesystem/mtkdafs.py

@@ -0,0 +1,78 @@
+from fuse import FuseOSError, Operations, LoggingMixIn
+from pathlib import Path
+import logging
+import os
+from stat import S_IFDIR, S_IFLNK, S_IFREG
+from time import time
+import mmap
+import errno
+from tempfile import NamedTemporaryFile
+
+class MtkDaFS(LoggingMixIn, Operations):
+    def __init__(self, da_handler, rw=False):
+        self.da_handler = da_handler
+        self.rw = rw
+        self.files = {}
+
+        self.files['/'] = dict(
+            st_mode=(S_IFDIR | 0o555),
+            st_ctime=time(),
+            st_mtime=time(),
+            st_atime=time(),
+            st_nlink=2)
+        self.files['/emmc_user.bin'] = dict(
+            st_mode=(S_IFREG | 0o777) if self.rw else (S_IFREG | 0o555),
+            st_ctime=time(),
+            st_mtime=time(),
+            st_atime=time(),
+            st_nlink=2,
+            st_size = self.da_handler.mtk.daloader.daconfig.flashsize)
+        self.files['/partitions'] = dict(
+            st_mode=(S_IFDIR | 0o555),
+            st_ctime=time(),
+            st_mtime=time(),
+            st_atime=time(),
+            st_nlink=2)
+
+        for part in self.da_handler.mtk.daloader.get_partition_data():
+            self.files[f'/partitions/{part.name}'] = dict(
+                st_mode=(S_IFREG | 0o777) if self.rw else (S_IFREG | 0o555),
+                st_ctime=time(),
+                st_mtime=time(),
+                st_atime=time(),
+                st_nlink=2,
+                st_size = part.sectors*self.da_handler.mtk.daloader.daconfig.pagesize,
+                offset=part.sector*self.da_handler.mtk.daloader.daconfig.pagesize)
+
+    def readdir(self, path, fh):
+        return ['.', '..'] + [ x.removeprefix(path).removeprefix('/') for x in self.files if x.startswith(path) and x != path]
+
+    def read(self, path, size, offset, fh):
+        if size+offset > self.files[path]['st_size']:
+            return b''
+        file_offset = 0
+        if 'offset' in self.files[path]:
+            file_offset = self.files[path]['offset']
+        data = self.da_handler.da_ro(start=file_offset+offset, length=size, filename='', parttype=None)
+        return bytes(data)
+
+    def write(self, path, data, offset, fh):
+        if not self.rw:
+            return 0
+
+        if offset+len(data) > self.files[path]['st_size']:
+            return b''
+
+        file_offset = 0
+        if 'offset' in self.files[path]:
+            file_offset = self.files[path]['offset']
+
+        with NamedTemporaryFile('rb+', buffering=0) as f_write:
+            f_write.write(data)
+            self.da_handler.da_wo(start=file_offset+offset, length=len(data), filename=f_write.name, parttype=None)
+        return len(data)
+
+    def getattr(self, path, fh=None):
+        if not self.rw:
+            self.files[path]['st_mode'] &= ~0o222
+        return self.files[path]

requirements.txt

@@ -8,3 +8,4 @@ pyside6 >= 6.4.0.1
 mock >= 4.0.3
 pyserial >= 3.5
 flake8
+fusepy