Add firmware extraction functions

README.md

@@ -54,6 +54,8 @@ provided on PyPI.
 
 ### Command line
 
+#### List
+
 ```
 $ reolinkfw info file_or_url
 ```
@@ -83,16 +85,48 @@ $ reolinkfw info RLC-410-5MP_20_20052300.zip -i 2
 argument. If it's a local or remote ZIP file it will be the path of the PAK file
 inside it. If it's a remote PAK file, it will be the value of the `name` query
 parameter or `None` if not found. And finally for a local PAK file it will be
-file name.
+the file name.
+
+#### Extract
+
+```
+usage: reolinkfw extract [-h] [-d DEST] [-f] file_or_url
+
+Extract the file system from a Reolink firmware
+
+positional arguments:
+  file_or_url           URL or on-disk file
+
+optional arguments:
+  -h, --help            show this help message and exit
+  -d DEST, --dest DEST  destination directory. Default: current directory
+  -f, --force           overwrite existing files. Does not apply to UBIFS. Default: False
+```
+
+A firmware's file system can be laid out in two different ways inside a PAK file:
+1. In a single section named `fs` or `rootfs` containing the whole file system
+1. In two sections with the second one named `app` containing the files that go in `/mnt/app`
+
+In the second case, the contents of `app` will be extracted to the appropriate
+location so that the files are organized the same way as they are when the
+camera is running.
+
+Consider the result of this command a one-way operation.
+You should not use it to repack a custom firmware.
 
 ### As a library
 
 ```py
-import reolinkfw
+from pakler import PAK
+from reolinkfw import get_info
+from reolinkfw.extract import extract_pak
+
 url = "https://reolink-storage.s3.amazonaws.com/website/firmware/20200523firmware/RLC-410-5MP_20_20052300.zip"
-print(reolinkfw.get_info(url))
+print(get_info(url))
 file = "/home/ben/RLC-410-5MP_20_20052300.zip"
-print(reolinkfw.get_info(file))
+print(get_info(file))
+with PAK.from_file(file) as pak:
+    extract_pak(pak)
 ```
 
 In most cases where a URL is used, it will be a direct link to the file

reolinkfw/__init__.py

@@ -31,6 +31,9 @@
 FILES = ("version_file", "version.json", "dvr.xml", "dvr", "router")
 INFO_KEYS = ("firmware_version_prefix", "board_type", "board_name", "build_date", "display_type_info", "detail_machine_type", "type")
 
+ROOTFS_SECTIONS = ["fs", "rootfs"]
+FS_SECTIONS = ROOTFS_SECTIONS + ["app"]
+
 
 async def download(url):
     """Return resource as bytes.
@@ -45,7 +48,7 @@ async def download(url):
 def extract_fs(pakbytes):
     """Return the fs.bin, app.bin or rootfs.bin file as bytes."""
     with PAK.from_bytes(pakbytes) as pak:
-        sections = {s.name: s for s in pak.sections if s.name in ("fs", "app", "rootfs")}
+        sections = {s.name: s for s in pak.sections if s.name in FS_SECTIONS}
         if len(sections) == 2:
             return pak.extract_section(sections["app"])
         elif len(sections) == 1:

reolinkfw/__main__.py

@@ -3,26 +3,57 @@
 import argparse
 import asyncio
 import json
+import sys
+from pathlib import Path, PurePath
 
-from reolinkfw import get_info, __version__
+from pakler import PAK
+from reolinkfw import __version__, get_info, get_paks
+from reolinkfw.extract import extract_pak
+from reolinkfw.util import sha256
 
 
-def info(args):
+def info(args: argparse.Namespace) -> None:
     info = asyncio.run(get_info(args.file_or_url))
     print(json.dumps(info, indent=args.indent, default=str))
 
 
+async def extract(args: argparse.Namespace) -> None:
+    paks = await get_paks(args.file_or_url)
+    if not paks:
+        raise Exception("No PAKs found in ZIP file")
+    dest = Path.cwd() if args.dest is None else args.dest
+    for pakname, pakbytes in paks:
+        name = sha256(pakbytes) if pakname is None else PurePath(pakname).stem
+        with PAK.from_bytes(pakbytes) as pak:
+            await asyncio.to_thread(extract_pak, pak, dest / name, args.force)
+
+
 def main():
-    parser = argparse.ArgumentParser(description="Extract information from Reolink firmware files")
-    parser.add_argument("-V", "--version", action="version", version=f"%(prog)s {__version__}", help="print version")
+    parser = argparse.ArgumentParser(description="Extract information and files from Reolink firmwares")
+    parser.add_argument("-V", "--version", action="version", version=f"%(prog)s {__version__}")
     subparsers = parser.add_subparsers(required=True)
+
     parser_i = subparsers.add_parser("info")
     parser_i.add_argument("file_or_url", help="URL or on-disk file")
     parser_i.add_argument("-i", "--indent", type=int, help="indent level for pretty print")
     parser_i.set_defaults(func=info)
+
+    descex = "Extract the file system from a Reolink firmware"
+    parser_e = subparsers.add_parser("extract", help=descex.lower(), description=descex)
+    parser_e.add_argument("file_or_url", help="URL or on-disk file")
+    parser_e.add_argument("-d", "--dest", type=Path, help="destination directory. Default: current directory")
+    parser_e.add_argument("-f", "--force", action="store_true", help="overwrite existing files. Does not apply to UBIFS. Default: %(default)s")
+    parser_e.set_defaults(func=extract)
+
     args = parser.parse_args()
-    args.func(args)    
+    try:
+        if asyncio.iscoroutinefunction(args.func):
+            asyncio.run(args.func(args))
+        else:
+            args.func(args)
+    except Exception as e:
+        sys.exit(f"error: {e}")
 
 
 if __name__ == "__main__":
-    main()
+    main()

reolinkfw/extract.py

@@ -0,0 +1,54 @@
+from contextlib import redirect_stdout
+from io import StringIO
+from pathlib import Path
+
+from pakler import PAK
+from pycramfs import Cramfs
+from pycramfs.extract import extract_dir as extract_cramfs
+from PySquashfsImage import SquashFsImage
+from PySquashfsImage.extract import extract_dir as extract_squashfs
+from ubireader.ubifs import ubifs
+from ubireader.ubifs.output import extract_files as extract_ubifs
+
+from reolinkfw import FS_SECTIONS, ROOTFS_SECTIONS
+from reolinkfw.util import (
+    DummyLEB,
+    get_fs_from_ubi,
+    is_cramfs,
+    is_squashfs,
+    is_ubi,
+    is_ubifs
+)
+
+
+def extract_file_system(fs_bytes, dest: Path = None):
+    dest = (Path.cwd() / "reolink_fs") if dest is None else dest
+    dest.mkdir(parents=True, exist_ok=True)
+    if is_ubi(fs_bytes):
+        extract_file_system(get_fs_from_ubi(fs_bytes), dest)
+    elif is_ubifs(fs_bytes):
+        with DummyLEB.from_bytes(fs_bytes) as leb:
+            with redirect_stdout(StringIO()):
+                # If files already exist they are not written again.
+                extract_ubifs(ubifs(leb), dest)
+    elif is_squashfs(fs_bytes):
+        with SquashFsImage.from_bytes(fs_bytes) as image:
+            extract_squashfs(image.root, dest, True)
+    elif is_cramfs(fs_bytes):
+        with Cramfs.from_bytes(fs_bytes) as image:
+            extract_cramfs(image.rootdir, dest, True)
+    else:
+        raise Exception("Unknown file system")
+
+
+def extract_pak(pak: PAK, dest: Path = None, force: bool = False):
+    dest = (Path.cwd() / "reolink_firmware") if dest is None else dest
+    dest.mkdir(parents=True, exist_ok=force)
+    rootfsdir = [s.name for s in pak.sections if s.name in ROOTFS_SECTIONS][0]
+    for section in pak.sections:
+        if section.name in FS_SECTIONS:
+            if section.name == "app":
+                outpath = dest / rootfsdir / "mnt" / "app"
+            else:
+                outpath = dest / rootfsdir
+            extract_file_system(pak.extract_section(section), outpath)