Documentation updates

+ parallel download support
+ decryption after download enabled

.github/workflows/python-sphinx.yml

@@ -0,0 +1,34 @@
+name: Sphinx Documentation
+
+on:
+  push:
+    branches: ["master"]
+
+jobs:
+  build-deploy:
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Checkout Repository
+        uses: actions/checkout@v3
+
+      - name: Setup Python
+        uses: actions/setup-python@v2
+        with:
+          python-version: '3.11'
+
+      - name: Install dependencies
+        run: pip install -r docs/requirements.txt && pip install -r requirements.txt
+
+      - name: Build
+        # Create .nojekyll file to disable Jekyll processing
+        run: |
+          cd docs
+          make html
+          touch build/html/.nojekyll
+
+      - name: Deploy to GitHub Pages
+        uses: peaceiris/actions-gh-pages@v3
+        with:
+          github_token: ${{ secrets.GITHUB_TOKEN }}
+          publish_dir: docs/build/html

README.md

@@ -0,0 +1,81 @@
+# samloader3
+
+Cross plattform Firmware downloader and decryptor for Samsung devices with maximum download speed.
+A list of API examples are given in the documentation available at [Github-Pages](https://matrixeditor.github.io/samloader3).
+
+> [!NOTE]
+> This project was formerly hosted at `nlscc/samloader`, and has moved from `samloader/samloader` to an refactored and updated version with enhanced CLI support and an API documentation.
+
+## Installation
+
+You can easily install it by using the `pip` package manager.
+
+```console
+pip install git+https://github.com/MatrixEditor/samloader3.git
+```
+
+## CLI
+
+The interface procided here is separated into two layers. In The first one, one can set basic options, such as the device's country code, model name or a global timeout value. Next, you will
+operate on a shell that takes commands with arguments as input.
+
+```console
+$ python3 -m samloader3 -M "SM-A336B" -R "SFR"
+(sl3)> # type commands here
+```
+
+### List firmware information
+
+Utilizing the `list` command you can list all available firmwares for a specific model within
+the selected region.
+
+> [!NOTE]
+> Make sure to always set the device's model name and region code, otherwise you won't get any
+> valid results. For simplicity, we don't write the model and region code explicitly.
+
+Using this command without any arguments will result in a table view that displays all available
+versions:
+
+<p align="center">
+
+![cmd_list](/docs/cmd_list.png)
+
+</p>
+
+> [!TIP]
+> If you just want to list the latest firmware use `-l` and if you want to print out the version
+> strings only, use `-q`. Using `-v VERSION` you can also view details on one specific version.
+
+
+### Download Firmware
+
+With this updated version of `samloader`, you can download multiple firmware files at one (though, most likely not a real use case) and accelerate to the maximum download speed. Using one version
+string from the output before, simply run the following command:
+
+```console
+(sl3)> download -o "/path/to/destination/" "$version1" "$version2" ...
+```
+
+As these files can be huge, once canceled, the donwload will resume at the current download
+position. You can disable that behaviour using `--no-cache`. With a special version identifier (`*`) you can download all firmware binaries at once.
+
+> [!WARNING]
+> Because of some issues with python.rich, parallel download is disabled by default. It can be
+> enabled using `--parallel`.
+
+To decrypt files directly after downloading them, use `--decrypt`.
+
+
+## Decrypt Firmware
+
+The decryption command (`decrypt`) is designd to operate on one file only. You just have
+to provide a version number and the file path:
+
+```console
+(sl3)> decrypt -v "$version" "/path/to/firmware.zip.enc4"
+```
+
+> [!TIP]
+> If you only want to generate the decryption key, use `--key-only`. Note that the actual
+> key is the MD5 value
+

pyproject.toml

@@ -0,0 +1,29 @@
+[project]
+name = "samloader3"
+version = "1.0.0"
+description="Updated implementation of the Samsung-Firmware Downloader (samloader)"
+authors = [
+  { name="MatrixEditor", email="not@supported.com" },
+]
+readme = "README.md"
+classifiers = [
+    'Intended Audience :: Science/Research',
+    'License :: OSI Approved :: MIT License',
+
+    # These versions are subject to be approved
+    # 'Programming Language :: Python :: 3.8',
+    # 'Programming Language :: Python :: 3.9',
+    # 'Programming Language :: Python :: 3.10',
+    'Programming Language :: Python :: 3.11',
+  ]
+
+[project.urls]
+"Homepage" = "https://github.com/MatrixEditor/samloader3"
+"API-Docs" = "https://matrixeditor.github.io/samloader3"
+
+[project.scripts]
+samloader3 = "samloader3.cli:run_with_args"
+
+[tool.setuptools.packages.find]
+where = ["."]
+include = ["samloader3*"]

samloader3/__init__.py

@@ -0,0 +1,4 @@
+__version__ = "1.0.0"
+__authors__ = ("MatrixEditor", )
+__author__ = ", ".join(__authors__)
+

samloader3/cli.py

@@ -4,10 +4,12 @@
 import typing as t
 import argparse
 
+import shlex
 import cmd
 import signal
 import traceback
 
+from concurrent.futures import ThreadPoolExecutor
 from threading import Event
 
 from rich import print as pprint
@@ -29,7 +31,7 @@
 
 from samloader3.fus import FUSClient, firmware_spec_url, FUS_USER_AGENT, v4_key, v2_key
 from samloader3.firmware import FirmwareSpec, FirmwareInfo
-from samloader3 import crypto
+from samloader3 import crypto, __version__
 
 
 def _print_ok(msg) -> None:
@@ -44,6 +46,10 @@ def _print_info(msg) -> None:
     pprint(r"\[  [bold cyan]Info[/]  ] " + msg)
 
 
+def _print_warn(msg) -> None:
+    pprint(r"\[  [bold yellow]Warn[/]  ] " + msg)
+
+
 class CLIExit(Exception):
     pass
 
@@ -131,26 +137,13 @@ def run(self, path: str, key: bytes, key_version: str) -> None:
             total_size = os.stat(path).st_size
             task = self.progress.add_task("Decrypting File", total=total_size)
 
-            cipher = crypto.get_file_decryptor(key)
-            chunks = total_size // 4096 + 1
-            out = self.argv.out
-            if os.path.isdir(out):
-                name = os.path.basename(path)
-                out = os.path.join(out, name.removesuffix(key_version))
+            def update():
+                self.progress.advance(task, block_size)
 
-            with open(path, "rb") as fp, open(out, "wb") as ostream:
-                for i in range(chunks):
-                    block = fp.read(4096)
-                    if not block:
-                        print(block)
-                        break
-
-                    decrypted = cipher.update(block)
-                    if i == chunks - 1:
-                        ostream.write(crypto.unpad(decrypted))
-                    else:
-                        ostream.write(decrypted)
-                    self.progress.update(task, advance=4096)
+            try:
+                crypto.file_decrypt(path, self.argv.out, key, block_size, key_version, update)
+            except ValueError:
+                _print_error("[bold]Invalid Padding:[/] Most likely due to a wrong input file!")
 
 
 class Download(Task):
@@ -168,8 +161,9 @@ def __init__(self, client: FUSClient, argv) -> None:
         self.names = []
 
     def handle_sigint(self, signum, frame):
+        if not self.done_event.is_set():
+            _print_info("Download canceled!")
         super().handle_sigint(signum, frame)
-        _print_info("Download canceled!")
 
     def _progress(self) -> Progress:
         """
@@ -210,18 +204,38 @@ def do_download(self, task_id: TaskID, info: FirmwareInfo, path: str) -> None:
             start = os.stat(path).st_size
 
         result = self.client.start_download(info, start)
-        self.progress.update(
-            task_id, total=int(result.headers["Content-Length"]), advance=start
-        )
-        with open(path, "wb") as dest_fp:
-            self.progress.start_task(task_id)
-            for chunk in result.iter_content(self.argv.chunk_size):
-                if self.done_event.is_set():
-                    return
+        total = int(result.headers["Content-Length"])
+        self.progress.update(task_id, total=total, advance=start)
+        self.progress.start_task(task_id)
+        if start < total:
+            with open(path, "wb") as dest_fp:
+                for chunk in result.iter_content(self.argv.chunk_size):
+                    if self.done_event.is_set():
+                        return
+
+                    if chunk:
+                        dest_fp.write(chunk)
+                        self.progress.update(task_id, advance=len(chunk))
+
+        if self.argv.decrypt:
+            self.progress.update(task_id, completed=0, filename="Decrypting file...")
+            if path.endswith(".enc4"):
+                _, dkey = v4_key(info)
+                key_version = "enc4"
+            elif path.endswith(".enc2"):
+                _, dkey = v2_key(info.version, info.model_name, info.local_code)
+                key_version = "enc2"
+            else:
+                _print_error(f"Could not find a suitable decryptor for {path}")
+                return
+
+            def update():
+                self.progress.update(task_id, advance=self.argv.block_size)
 
-                if chunk:
-                    dest_fp.write(chunk)
-                    self.progress.update(task_id, advance=len(chunk))
+            out = path.removesuffix(key_version)
+            crypto.file_decrypt(
+                path, out, dkey, self.argv.block_size, key_version, update
+            )
 
     def run(self, specs: t.List[FirmwareSpec], model, region) -> None:
         """
@@ -239,6 +253,10 @@ def run(self, specs: t.List[FirmwareSpec], model, region) -> None:
             data.append(self.client.fw_info(spec.normalized_version, model, region))
 
         with self.progress:
+            pool = None
+            if self.argv.parallel:
+                pool = ThreadPoolExecutor(max_workers=10)
+
             for info in data:
                 if info.binary_name in self.names:
                     continue
@@ -247,10 +265,16 @@ def run(self, specs: t.List[FirmwareSpec], model, region) -> None:
                 task_id = self.progress.add_task(
                     "download", filename=info.binary_name, start=False
                 )
-                self.do_download(task_id, info, self.argv.dest)
+                if pool is not None:
+                    pool.submit(self.do_download, task_id, info, self.argv.dest)
+                else:
+                    self.do_download(task_id, info, self.argv.dest)
                 if self.done_event.is_set():
                     return
 
+            if pool:
+                pool.shutdown(wait=True)
+
 
 class SamLoader3CLI(cmd.Cmd):
     """
@@ -327,6 +351,9 @@ def setup_parsers(self) -> None:
         download_mod.add_argument("--chunk-size", type=int, default=32768)
         download_mod.add_argument("-o", "--out", dest="dest", type=str, required=True)
         download_mod.add_argument("--no-cache", action="store_true")
+        download_mod.add_argument("--decrypt", action="store_true")
+        download_mod.add_argument("--parallel", action="store_true")
+        download_mod.add_argument("--block-size", type=int, default=4096)
         download_mod.set_defaults(fn=self._download)
         self.parsers["download"] = download_mod
 
@@ -350,8 +377,7 @@ def _run(self, name: str, args) -> None:
         :param args: The command arguments.
         :type args: str
         """
-        # REVISIT: split(...) may not be applicable here
-        argv = self.parsers[name].parse_args(args.split(" "))
+        argv = self.parsers[name].parse_args(shlex.split(args))
         argv.fn(argv)
 
     def do_exit(self, args) -> None:
@@ -360,6 +386,21 @@ def do_exit(self, args) -> None:
         """
         raise CLIExit
 
+    def do_version(self, args) -> None:
+        """Prints the library's version"""
+        pprint(f"[bold]Version:[/] {__version__}")
+
+    def do_setmodel(self, args) -> None:
+        """Sets the model name to use"""
+        self.model = args
+
+    def do_setregion(self, args) -> None:
+        """Sets the region code to use"""
+        self.region = args
+
+    def default(self, line) -> None:
+        _print_warn(f"[bold]Unknown syntax:[/] {line}")
+
     def get_names(self):
         """
         Retrieves all command names.
@@ -426,7 +467,7 @@ def _connect(self) -> bool:
 
     def _verify_device_info(self, argv) -> bool:
         model = self.get_model(argv)
-        region = self.get_model(argv)
+        region = self.get_region(argv)
         if not model or not region:
             _print_error("[bold]Device:[/] No device specified!")
             return False

samloader3/crypto.py

@@ -136,6 +136,7 @@ def file_decrypt(
     key: bytes,
     block_size: int = 4096,
     key_version: t.Optional[str] = None,
+    cb: t.Callable[[], None] = None
 ) -> None:
     """
     Decrypts a file using a given key.
@@ -171,7 +172,8 @@ def file_decrypt(
 
             decrypted = cipher.update(block)
             if i == chunks - 1:
-                # we actually don't need .finalize() here
-                ostream.write(unpad(decrypted))
+                ostream.write(unpad(decrypted + cipher.finalize()))
             else:
                 ostream.write(decrypted)
+            if cb:
+                cb()