#!/usr/bin/env python3 """ Ender-3 V3 KE firmware extractor/repacker. This script handles the stock Creality 7z-wrapped `.img` package, reconstructs the split payload files, and can rebuild a fresh package from an extracted rootfs tree. """ from __future__ import annotations import argparse import hashlib import os import shlex import re import shutil import subprocess import sys import tempfile import urllib.request import zipfile from pathlib import Path from typing import List, Sequence, Tuple import py7zr PASSWORD = "$1$cxswfile$ZFd0RWFYkJQugbtKVGL9y0" CHUNK_SIZE = 1024 * 1024 UIMAGE_HEADER_SIZE = 64 DEFAULT_STOCK_ARCHIVE = Path("Ender-3_V3_KE_F005_ota_img_V1.1.0.17.img") DEFAULT_OVERLAY_DIR = Path("overlay_rootfs") DEFAULT_VERSION = "11034" SQUASHFS_TOOLS_URL = "https://infraroot.at/pub/squashfs/windows/squashfs-tools-ng-1.3.2-mingw64.zip" SQUASHFS_TOOLS_DIR = Path("build") / "_toolcache" / "squashfs-tools-ng-1.3.2-mingw64" def md5_file(path: Path) -> str: digest = hashlib.md5() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(4 * 1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def ensure_clean_dir(path: Path) -> None: if path.exists(): shutil.rmtree(path) path.mkdir(parents=True, exist_ok=True) def reconstruct_split_file(src_dir: Path, prefix: str, out_path: Path) -> List[Path]: pattern = re.compile(rf"^{re.escape(prefix)}\.(\d{{4}})\.[0-9a-f]{{32}}$") parts: List[Tuple[int, Path]] = [] for entry in src_dir.iterdir(): if not entry.is_file(): continue match = pattern.match(entry.name) if match: parts.append((int(match.group(1)), entry)) if not parts: raise FileNotFoundError(f"no split files found for {prefix} in {src_dir}") parts.sort(key=lambda item: item[0]) with out_path.open("wb") as out_handle: for _, part in parts: out_handle.write(part.read_bytes()) return [path for _, path in parts] def write_lines(path: Path, lines: Sequence[str]) -> None: path.write_text("\n".join(lines) + "\n", encoding="utf-8", newline="\n") def split_file(src_path: Path, dest_dir: Path, base_name: str) -> Tuple[str, List[Path], List[str]]: full_md5 = md5_file(src_path) chunk_md5s: List[str] = [] chunk_paths: List[Path] = [] with src_path.open("rb") as handle: index = 0 previous_md5 = full_md5 while True: data = handle.read(CHUNK_SIZE) if not data: break chunk_md5 = hashlib.md5(data).hexdigest() chunk_name = f"{base_name}.{index:04d}.{previous_md5}" chunk_path = dest_dir / chunk_name chunk_path.write_bytes(data) chunk_paths.append(chunk_path) chunk_md5s.append(chunk_md5) previous_md5 = chunk_md5 index += 1 return full_md5, chunk_paths, chunk_md5s def download_file(url: str, dest: Path) -> None: dest.parent.mkdir(parents=True, exist_ok=True) temp_dest = dest.with_name(dest.name + ".download") if temp_dest.exists(): temp_dest.unlink() with urllib.request.urlopen(url) as response, temp_dest.open("wb") as out_handle: shutil.copyfileobj(response, out_handle) temp_dest.replace(dest) def resolve_squashfs_tool(tool_name: str) -> Path: """ Resolve a squashfs-tools-ng executable for the current host. On Windows we use the cached MinGW bundle that ships with this repo. On other platforms we require a native binary from PATH. """ if os.name == "nt": tools_bin = ensure_squashfs_tools(SQUASHFS_TOOLS_DIR, SQUASHFS_TOOLS_URL) return tools_bin / f"{tool_name}.exe" resolved = shutil.which(tool_name) if resolved: return Path(resolved) raise FileNotFoundError( f"{tool_name} not found on PATH. Install squashfs-tools-ng for your Linux distribution " f"or run the helper on Windows where the bundled MinGW tools are available." ) def extract_archive(archive: Path, outdir: Path, password: str) -> Path: ensure_clean_dir(outdir) with py7zr.SevenZipFile(archive, mode="r", password=password) as zf: zf.extractall(path=outdir) top_entries = [entry for entry in outdir.iterdir() if entry.is_dir()] if len(top_entries) != 1: raise RuntimeError(f"expected one top-level folder after extraction, found {len(top_entries)}") return top_entries[0] def ensure_stock_archive(stock_archive: Path, stock_url: str | None, password: str) -> Path: if stock_archive.exists(): return stock_archive if not stock_url: raise FileNotFoundError( f"stock archive not found at {stock_archive}; supply --stock-url to download it first" ) download_file(stock_url, stock_archive) with py7zr.SevenZipFile(stock_archive, mode="r", password=password) as zf: zf.test() return stock_archive def ensure_squashfs_tools(tools_dir: Path, tools_url: str) -> Path: bin_dir = tools_dir / "bin" gensquashfs = bin_dir / "gensquashfs.exe" rdsquashfs = bin_dir / "rdsquashfs.exe" if gensquashfs.exists() and rdsquashfs.exists(): return bin_dir tools_dir.parent.mkdir(parents=True, exist_ok=True) zip_path = tools_dir.with_suffix(".zip") if not zip_path.exists(): download_file(tools_url, zip_path) if tools_dir.exists(): shutil.rmtree(tools_dir) with zipfile.ZipFile(zip_path) as zf: zf.extractall(path=tools_dir.parent) if not gensquashfs.exists() or not rdsquashfs.exists(): raise FileNotFoundError(f"failed to install squashfs tools into {bin_dir}") return bin_dir def reconstruct_from_archive_tree(tree_root: Path, workspace: Path) -> dict[str, Path]: version_file = tree_root / "ota_config.in" if version_file.exists(): version = read_version(version_file) ota_dir = tree_root / f"ota_v{version}" else: ota_dir = tree_root / f"ota_v{DEFAULT_VERSION}" if not ota_dir.exists(): matches = sorted([entry for entry in tree_root.iterdir() if entry.is_dir() and entry.name.startswith("ota_v")]) if not matches: raise FileNotFoundError(f"could not locate ota_v* directory under {tree_root}") ota_dir = matches[0] rootfs_full = workspace / "rootfs.squashfs.full" ximage_full = workspace / "xImage.full" zero_full = workspace / "zero.bin.full" rootfs_full.parent.mkdir(parents=True, exist_ok=True) ximage_full.parent.mkdir(parents=True, exist_ok=True) zero_full.parent.mkdir(parents=True, exist_ok=True) reconstruct_split_file(ota_dir, "rootfs.squashfs", rootfs_full) reconstruct_split_file(ota_dir, "xImage", ximage_full) reconstruct_split_file(ota_dir, "zero.bin", zero_full) ximage_payload = workspace / "xImage.payload" ximage_payload.write_bytes(ximage_full.read_bytes()[UIMAGE_HEADER_SIZE:]) return { "ota_dir": ota_dir, "rootfs_full": rootfs_full, "ximage_full": ximage_full, "ximage_payload": ximage_payload, "zero_full": zero_full, } def extract_rootfs_to_dir(rootfs_full: Path, outdir: Path) -> None: ensure_clean_dir(outdir) rdsquashfs = resolve_squashfs_tool("rdsquashfs") subprocess.run( [ str(rdsquashfs), "--unpack-path", "/", "--unpack-root", str(outdir), "--chmod", "--chown", "--set-times", "--quiet", str(rootfs_full), ], check=True, ) def describe_rootfs(rootfs_full: Path) -> List[str]: rdsquashfs = resolve_squashfs_tool("rdsquashfs") result = subprocess.run( [str(rdsquashfs), "--describe", str(rootfs_full)], check=True, capture_output=True, text=True, encoding="utf-8", errors="replace", ) return [line.strip() for line in result.stdout.splitlines() if line.strip()] def describe_rootfs_map(rootfs_full: Path) -> dict[str, str]: root_entries: dict[str, str] = {} for line in describe_rootfs(rootfs_full): parts = shlex.split(line) if len(parts) < 2: continue root_entries[parts[1].lstrip("/")] = line return root_entries def packfile_entry(line: str) -> tuple[str, str]: parts = shlex.split(line) if len(parts) < 5: raise ValueError(f"invalid squashfs describe line: {line}") kind = parts[0] path = parts[1] mode = parts[2] uid = parts[3] gid = parts[4] extra = " ".join(parts[5:]) pack_path = path if path.startswith("/") else f"/{path}" if kind in {"dir", "file"}: entry = f"{kind} {quote_pack_path(pack_path)} {mode} {uid} {gid}" elif kind in {"slink", "link"}: if not extra: raise ValueError(f"missing link target for {line}") entry = f"{kind} {quote_pack_path(pack_path)} {mode} {uid} {gid} {extra}" elif kind == "nod": if not extra: raise ValueError(f"missing device metadata for {line}") entry = f"{kind} {quote_pack_path(pack_path)} {mode} {uid} {gid} {extra}" else: raise ValueError(f"unsupported squashfs entry type: {kind}") return kind, entry def quote_pack_path(path: str) -> str: if not any(ch.isspace() for ch in path) and '"' not in path: return path return '"' + path.replace("\\", "\\\\").replace('"', '\\"') + '"' def list_overlay_entries(rootfs_dir: Path, stock_paths: set[str]) -> List[str]: additions: List[tuple[int, str]] = [] for entry in rootfs_dir.rglob("*"): relative = entry.relative_to(rootfs_dir).as_posix() pack_path = f"/{relative}" if relative in stock_paths: continue if entry.is_dir() and not entry.is_symlink(): additions.append((0, f"dir {quote_pack_path(pack_path)} 0755 0 0")) elif entry.is_symlink(): additions.append((2, f"slink {quote_pack_path(pack_path)} 0777 0 0 {os.readlink(entry)}")) else: mode_bits = entry.stat().st_mode & 0o7777 if mode_bits & 0o111 == 0 and relative.startswith("etc/init.d/"): mode_bits = (mode_bits & ~0o666) | 0o755 mode = oct(mode_bits)[2:] additions.append((1, f"file {quote_pack_path(pack_path)} {mode} 0 0")) additions.sort(key=lambda item: (item[0], item[1])) return [entry for _, entry in additions] def collect_intentional_overlay_paths(overlay_dir: Path) -> set[str]: intentional: set[str] = set() delete_file = overlay_dir / ".delete" if delete_file.exists(): intentional.update( str(path).replace("\\", "/") for path in read_delete_list(overlay_dir) ) for entry in overlay_dir.rglob("*"): relative = entry.relative_to(overlay_dir).as_posix() intentional.add(relative) return intentional def should_skip_compare_path(path: str) -> bool: return any(ord(ch) > 127 for ch in path) def compare_rootfs_metadata(stock_rootfs_full: Path, rebuilt_rootfs_full: Path, overlay_dir: Path) -> List[str]: stock_map = describe_rootfs_map(stock_rootfs_full) rebuilt_map = describe_rootfs_map(rebuilt_rootfs_full) intentional = collect_intentional_overlay_paths(overlay_dir) issues: List[str] = [] for path, stock_line in stock_map.items(): if should_skip_compare_path(path): continue if path in intentional: continue rebuilt_line = rebuilt_map.get(path) if rebuilt_line is None: issues.append(f"missing from rebuilt rootfs: {path}") continue if rebuilt_line != stock_line: issues.append(f"metadata mismatch: {path}") for path, rebuilt_line in rebuilt_map.items(): if should_skip_compare_path(path): continue if path in stock_map or path in intentional: continue issues.append(f"unexpected extra path in rebuilt rootfs: {path}") return issues def write_pack_file(rootfs_dir: Path, stock_rootfs_full: Path, packfile_path: Path) -> None: lines = describe_rootfs(stock_rootfs_full) stock_entries: List[str] = [] stock_paths: set[str] = set() for line in lines: kind, entry = packfile_entry(line) path = shlex.split(line)[1] stock_paths.add(path) target_path = rootfs_dir / path if kind in {"file", "dir"}: if not target_path.exists(): continue elif kind in {"slink", "link", "nod"}: if kind in {"slink", "link"} and target_path.exists(): remove_path(target_path) stock_entries.append(entry) overlay_entries = list_overlay_entries(rootfs_dir, stock_paths) packfile_path.write_text("\n".join(stock_entries + overlay_entries) + "\n", encoding="utf-8", newline="\n") def remove_path(path: Path) -> None: if not path.exists() and not path.is_symlink(): return if path.is_dir() and not path.is_symlink(): shutil.rmtree(path) else: path.unlink() def safe_rmtree(path: Path) -> None: if not path.exists() and not path.is_symlink(): return if path.is_symlink() or path.is_file(): path.unlink(missing_ok=True) return for entry in path.iterdir(): safe_rmtree(entry) path.rmdir() def read_delete_list(overlay_dir: Path) -> List[Path]: delete_file = overlay_dir / ".delete" if not delete_file.exists(): return [] deletions: List[Path] = [] for raw_line in delete_file.read_text(encoding="utf-8").splitlines(): line = raw_line.strip() if not line or line.startswith("#"): continue deletions.append(Path(line)) return deletions def apply_overlay_tree(rootfs_dir: Path, overlay_dir: Path) -> None: if not overlay_dir.exists(): raise FileNotFoundError(f"overlay directory not found: {overlay_dir}") for relative in read_delete_list(overlay_dir): target = rootfs_dir / relative remove_path(target) for entry in overlay_dir.rglob("*"): if entry.is_dir(): continue if entry.name == ".delete" and entry.parent == overlay_dir: continue relative = entry.relative_to(overlay_dir) target = rootfs_dir / relative target.parent.mkdir(parents=True, exist_ok=True) remove_path(target) shutil.copy2(entry, target) def build_rootfs_from_dir(rootfs_dir: Path, stock_rootfs_full: Path, output_path: Path, workspace: Path) -> None: gensquashfs = resolve_squashfs_tool("gensquashfs") packfile_path = workspace / "rootfs.packfile" write_pack_file(rootfs_dir, stock_rootfs_full, packfile_path) if output_path.exists(): output_path.unlink() subprocess.run( [ str(gensquashfs), "--pack-file", str(packfile_path), "--pack-dir", str(rootfs_dir), "--keep-time", "--block-size", "131072", "--compressor", "gzip", "--force", "--quiet", str(output_path), ], check=True, ) def read_version(config_path: Path) -> str: for line in config_path.read_text(encoding="utf-8").splitlines(): line = line.strip() if line.startswith("current_version="): return line.split("=", 1)[1].strip() raise ValueError(f"current_version not found in {config_path}") def write_ota_manifest( staging_root: Path, version: str, items: Sequence[Tuple[str, str, Path, Path, List[str]]], ) -> None: archive_root = staging_root.name version_dir = staging_root / f"ota_v{version}" version_dir.mkdir(parents=True, exist_ok=True) (staging_root / "ota_config.in").write_text(f"current_version={version}\n", encoding="utf-8") (version_dir / f"ota_v{version}.ok").write_bytes(b"") ota_update_lines = [f"ota_version={version}", ""] for img_type, img_name, full_path, md5_path, chunk_md5s in items: ota_update_lines.extend( [ f"img_type={img_type}", f"img_name={img_name}", f"img_size={full_path.stat().st_size}", f"img_md5={md5_file(full_path)}", "", ] ) sidecar_name = f"ota_md5_{img_name}.{md5_file(full_path)}" write_lines(version_dir / sidecar_name, chunk_md5s) write_lines(version_dir / "ota_update.in", ota_update_lines) def stage_and_pack( workspace: Path, version: str, rootfs_full: Path, ximage_full: Path, zero_full: Path, output_img: Path, password: str, ) -> Path: archive_root = f"DarKE_KE_F005_ota_img_V{version}" staging_root = workspace / "_firmware_stage" / archive_root ensure_clean_dir(staging_root.parent) staging_root.mkdir(parents=True, exist_ok=True) version_dir = staging_root / f"ota_v{version}" version_dir.mkdir(parents=True, exist_ok=True) rootfs_md5, rootfs_chunks, rootfs_chunk_md5s = split_file(rootfs_full, version_dir, "rootfs.squashfs") ximage_md5, ximage_chunks, ximage_chunk_md5s = split_file(ximage_full, version_dir, "xImage") zero_md5, zero_chunks, zero_chunk_md5s = split_file(zero_full, version_dir, "zero.bin") (staging_root / "ota_config.in").write_text(f"current_version={version}\n", encoding="utf-8", newline="\n") (version_dir / f"ota_v{version}.ok").write_bytes(b"") ota_update_lines = [f"ota_version={version}", ""] for img_type, img_name, full_path, chunk_md5s, full_md5 in [ ("kernel", "xImage", ximage_full, ximage_chunk_md5s, ximage_md5), ("rootfs", "rootfs.squashfs", rootfs_full, rootfs_chunk_md5s, rootfs_md5), ("rtos", "zero.bin", zero_full, zero_chunk_md5s, zero_md5), ]: ota_update_lines.extend( [ f"img_type={img_type}", f"img_name={img_name}", f"img_size={full_path.stat().st_size}", f"img_md5={full_md5}", "", ] ) sidecar_name = f"ota_md5_{img_name}.{full_md5}" write_lines(version_dir / sidecar_name, chunk_md5s) write_lines(version_dir / "ota_update.in", ota_update_lines) if output_img.exists(): output_img.unlink() with py7zr.SevenZipFile(output_img, mode="w", password=password) as zf: zf.writeall(staging_root, arcname=archive_root) return output_img def validate_archive(archive: Path, password: str) -> None: with tempfile.TemporaryDirectory(dir=str(archive.parent)) as tmpdir: tmp_path = Path(tmpdir) with py7zr.SevenZipFile(archive, mode="r", password=password) as zf: zf.extractall(path=tmp_path) top = next(tmp_path.iterdir(), None) if top is None or not top.is_dir(): raise RuntimeError("archive validation failed: missing top-level directory") ota_config = top / "ota_config.in" if not ota_config.exists(): raise RuntimeError("archive validation failed: missing ota_config.in") def cmd_extract(args: argparse.Namespace) -> None: archive = Path(args.archive).expanduser().resolve() outdir = Path(args.outdir).expanduser().resolve() workspace = Path(args.workspace).expanduser().resolve() tree_root = extract_archive(archive, outdir, args.password) files = reconstruct_from_archive_tree(tree_root, workspace) extract_rootfs_to_dir(files["rootfs_full"], workspace / "rootfs_extract") print(f"extracted archive to {outdir}") print(f"reconstructed rootfs: {files['rootfs_full']}") print(f"reconstructed kernel: {files['ximage_full']}") print(f"reconstructed rtos: {files['zero_full']}") print(f"rootfs unpacked to: {workspace / 'rootfs_extract'}") def cmd_build(args: argparse.Namespace) -> None: version = args.version if not version: version = DEFAULT_VERSION build_dir = Path(args.build_dir).expanduser().resolve() build_dir.mkdir(parents=True, exist_ok=True) output_img = Path(args.output).expanduser().resolve() stock_archive = ensure_stock_archive( Path(args.stock_archive).expanduser().resolve(), args.stock_url, args.password, ) overlay_dir = Path(args.overlay_dir).expanduser().resolve() workdir = Path(tempfile.mkdtemp(prefix="_firmware_build_", dir=str(build_dir))) try: stock_extract_dir = workdir / "stock_extract" stock_tree = extract_archive(stock_archive, stock_extract_dir, args.password) stock_files = reconstruct_from_archive_tree(stock_tree, workdir) rootfs_dir = workdir / "rootfs_work" extract_rootfs_to_dir(stock_files["rootfs_full"], rootfs_dir) apply_overlay_tree(rootfs_dir, overlay_dir) rootfs_full = workdir / "rootfs.squashfs.full" build_rootfs_from_dir(rootfs_dir, stock_files["rootfs_full"], rootfs_full, workdir) stage_and_pack( workspace=workdir, version=version, rootfs_full=rootfs_full, ximage_full=stock_files["ximage_full"], zero_full=stock_files["zero_full"], output_img=output_img, password=args.password, ) validate_archive(output_img, args.password) finally: safe_rmtree(workdir) print(f"built firmware image: {output_img}") def cmd_compare(args: argparse.Namespace) -> None: stock_archive = Path(args.stock_archive).expanduser().resolve() rebuilt_archive = Path(args.rebuilt_archive).expanduser().resolve() overlay_dir = Path(args.overlay_dir).expanduser().resolve() compare_root = Path(args.work_dir).expanduser().resolve() compare_root.mkdir(parents=True, exist_ok=True) workdir = Path(tempfile.mkdtemp(prefix="_firmware_compare_", dir=str(compare_root))) try: stock_tree = extract_archive(stock_archive, workdir / "stock_extract", args.password) rebuilt_tree = extract_archive(rebuilt_archive, workdir / "rebuilt_extract", args.password) stock_files = reconstruct_from_archive_tree(stock_tree, workdir / "stock") rebuilt_files = reconstruct_from_archive_tree(rebuilt_tree, workdir / "rebuilt") stock_rootfs_full = stock_files["rootfs_full"] rebuilt_rootfs_full = rebuilt_files["rootfs_full"] issues = compare_rootfs_metadata(stock_rootfs_full, rebuilt_rootfs_full, overlay_dir) if issues: for issue in issues: print(issue) raise SystemExit(1) finally: safe_rmtree(workdir) print("rootfs metadata matches stock outside the overlay") def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Extract and rebuild Ender-3 V3 KE firmware images.") parser.add_argument("--password", default=PASSWORD, help="7z archive password") subparsers = parser.add_subparsers(dest="command", required=True) extract = subparsers.add_parser("extract", help="extract an official firmware archive") extract.add_argument("--archive", required=True, help="path to the stock .img archive") extract.add_argument("--outdir", default="ke_firmware_extract", help="archive extraction directory") extract.add_argument("--workspace", default=".", help="workspace for reconstructed artifacts") extract.set_defaults(func=cmd_extract) build = subparsers.add_parser("build", help="build a repacked firmware archive") build.add_argument("--workspace", default=".", help="workspace root") build.add_argument("--build-dir", default="build", help="directory for intermediate build artifacts") build.add_argument( "--stock-archive", default=str(DEFAULT_STOCK_ARCHIVE), help="path to the stock OTA image to use as the build base", ) build.add_argument( "--stock-url", default=None, help="download URL used when the stock OTA image is missing", ) build.add_argument( "--overlay-dir", default=str(DEFAULT_OVERLAY_DIR), help="directory containing only our custom rootfs changes", ) build.add_argument( "--output", default=str(Path("build") / f"DarKE_KE_F005_ota_img_V{DEFAULT_VERSION}.img"), help="output .img archive path", ) build.add_argument("--version", default=DEFAULT_VERSION, help="firmware version to encode in manifests") build.set_defaults(func=cmd_build) compare = subparsers.add_parser("compare", help="compare a rebuilt image's rootfs metadata against stock") compare.add_argument("--stock-archive", required=True, help="path to the stock .img archive") compare.add_argument("--rebuilt-archive", required=True, help="path to the rebuilt .img archive") compare.add_argument( "--overlay-dir", default=str(DEFAULT_OVERLAY_DIR), help="overlay tree used to produce the rebuilt image", ) compare.add_argument("--work-dir", default="build", help="directory for temporary compare artifacts") compare.set_defaults(func=cmd_compare) return parser def main(argv: Sequence[str] | None = None) -> int: for stream in (sys.stdout, sys.stderr): try: stream.reconfigure(encoding="utf-8", errors="replace") except Exception: pass parser = build_parser() args = parser.parse_args(argv) args.func(args) return 0 if __name__ == "__main__": raise SystemExit(main())