asahi_firmware: Add cpio file generation

Signed-off-by: Hector Martin <[email protected]>

Author: marcan

asahi_firmware/core.py

@@ -1,6 +1,7 @@
 # SPDX-License-Identifier: MIT
-import tarfile, io, logging
+import tarfile, io, logging, os.path
 from hashlib import sha256
+from . import cpio
 
 class FWFile(object):
     def __init__(self, name, data):
@@ -20,14 +21,31 @@ def __hash__(self):
         return hash(self.sha)
 
 class FWPackage(object):
-    def __init__(self, target):
-        self.path = target
-        self.tarfile = tarfile.open(target, mode="w")
+    def __init__(self, tar_path, cpio_path):
+        self.closed = False
+        self.tar_path = tar_path
+        self.cpio_path = cpio_path
+        self.tarfile = tarfile.open(tar_path, mode="w")
+        self.cpiofile = cpio.CPIO(cpio_path)
         self.hashes = {}
         self.manifest = []
 
     def close(self):
+        if self.closed:
+            return
+
+        ti = tarfile.TarInfo("vendorfw/.vendorfw.manifest")
+        ti.type = tarfile.REGTYPE
+        fd = io.BytesIO()
+        for i in self.manifest:
+            fd.write(i.encode("ascii") + b"\n")
+        ti.size = fd.tell()
+        fd.seek(0)
+        self.cpiofile.addfile(ti, fd)
+
         self.tarfile.close()
+        self.cpiofile.close()
+        self.closed = True
 
     def add_file(self, name, data):
         ti = tarfile.TarInfo(name)
@@ -45,6 +63,12 @@ def add_file(self, name, data):
 
         logging.info(f"+ {self.manifest[-1]}")
         self.tarfile.addfile(ti, fd)
+        if fd is not None:
+            fd.seek(0)
+        ti.name = os.path.join("vendorfw", ti.name)
+        if ti.linkname:
+            ti.linkname = os.path.join("vendorfw", ti.linkname)
+        self.cpiofile.addfile(ti, fd)
 
     def add_files(self, it):
         for name, data in it:
@@ -56,4 +80,4 @@ def save_manifest(self, filename):
                 fd.write(i + "\n")
 
     def __del__(self):
-        self.tarfile.close()
+        self.close()

asahi_firmware/cpio.py

@@ -0,0 +1,88 @@
+# SPDX-License-Identifier: MIT
+import os.path, tarfile
+
+class CPIO:
+    DIR = 0o040755
+    FILE = 0o100644
+
+    def __init__(self, filename):
+        self.fd = open(filename, "wb")
+        self.closed = False
+        self.dirs = set()
+        self.inode = 1
+        self.nlink = {}
+        self.nlinkoff = {}
+        self.inodemap = {}
+
+    def cpio_hdr(self, name, mode, size, target=None):
+        if target is not None:
+            inode = self.inodemap[target]
+            self.nlink[inode] += 1
+            p = self.fd.tell()
+            for i in self.nlinkoff[inode]:
+                self.fd.seek(i)
+                self.fd.write(b"%08x" % self.nlink[inode])
+            self.fd.seek(p)
+        else:
+            inode = self.inode
+            self.inode += 1
+            self.nlink[inode] = 1
+            self.nlinkoff[inode] = []
+
+        self.inodemap[name] = inode
+
+        p = self.fd.tell()
+        if p & 3:
+            self.fd.write(bytes(4 - (p & 3)))
+
+        self.fd.write(b"070701")
+        self.nlinkoff[inode].append(self.fd.tell() + 8 * 4)
+        hdr = [
+            inode,
+            mode,
+            0, # uid
+            0, # gid
+            self.nlink[inode],
+            0, # mtime
+            size,
+            0, # maj
+            0, # min
+            0, # rmaj
+            0, # rmin
+            len(name) + 1,
+            0, # chksum
+        ]
+        self.fd.write(b"".join(b"%08x" % i for i in hdr))
+        self.fd.write(name.encode("ascii") + b"\x00")
+
+        p = self.fd.tell()
+        if p & 3:
+            self.fd.write(bytes(4 - (p & 3)))
+    
+    def addfile(self, ti, fd):
+        path = ""
+        for i in ti.name.split("/")[:-1]:
+            if not i:
+                continue
+            path = os.path.join(path, i)
+            if path not in self.dirs:
+                self.cpio_hdr(path, self.DIR, 0)
+                self.dirs.add(path)
+
+        if ti.type == tarfile.LNKTYPE:
+            self.cpio_hdr(ti.name, self.FILE, 0, ti.linkname)
+        elif ti.type == tarfile.REGTYPE:
+            self.cpio_hdr(ti.name, self.FILE, ti.size)
+            self.fd.write(fd.read())
+        else:
+            raise Exception(f"Unsupported file type {ti.type}")
+    
+    def close(self):
+        if self.closed:
+            return
+        self.closed = True
+        self.cpio_hdr("TRAILER!!!", self.FILE, 0);    
+        self.fd.close()
+    
+    def __del__(self):
+        self.close()

asahi_firmware/kernel.py

@@ -12,12 +12,15 @@ def __init__(self, source_path):
         self.load(source_path)
 
     def load(self, source_path):
-        for fname in os.listdir(source_path):
-            if fname.startswith("kernelcache"):
-                kern_path = os.path.join(source_path, fname)
-                break
+        if os.path.isdir(source_path):
+            for fname in os.listdir(source_path):
+                if fname.startswith("kernelcache"):
+                    kern_path = os.path.join(source_path, fname)
+                    break
+            else:
+                raise Exception("Could not find kernelcache")
         else:
-            raise Exception("Could not find kernelcache")
+            kern_path = source_path
 
         log.info(f"Extracting firmware from kernel at {kern_path}")
 

asahi_firmware/update.py

@@ -1,18 +1,19 @@
 # SPDX-License-Identifier: MIT
-import pathlib, tempfile, subprocess
+import pathlib, tempfile, subprocess, os.path
 
 from .core import FWPackage
 from .wifi import WiFiFWCollection
 from .bluetooth import BluetoothFWCollection
 from .multitouch import MultitouchFWCollection
 from .kernel import KernelFWCollection
 
-def update_firmware(source, dest, manifest):
+def update_firmware(source, dest):
     raw_fw = source.joinpath("all_firmware.tar.gz")
     if not raw_fw.exists():
         print(f"Could not find {raw_fw}")
 
-    pkg = FWPackage(dest)
+    pkg = FWPackage(os.path.join(dest, "firmware.tar"),
+                    os.path.join(dest, "firmware.cpio"))
 
     with tempfile.TemporaryDirectory() as tmpdir:
         tmpdir = pathlib.Path(tmpdir)
@@ -32,7 +33,7 @@ def update_firmware(source, dest, manifest):
 
     pkg.close()
 
-    pkg.save_manifest(manifest)
+    pkg.save_manifest(os.path.join(dest, "manifest.txt"))
 
 if __name__ == "__main__":
     import argparse
@@ -42,11 +43,9 @@ def update_firmware(source, dest, manifest):
     parser = argparse.ArgumentParser(description='Update vendor firmware tarball')
     parser.add_argument('source', metavar='SOURCE', type=pathlib.Path,
                         help='path containing raw firmware')
-    parser.add_argument('dest', metavar='TARBALL', type=pathlib.Path,
-                        help='output vendor firmware tarball')
-    parser.add_argument('manifest', metavar='MANIFEST', type=pathlib.Path,
-                        help='output vendor firmware manifest')
+    parser.add_argument('dest', metavar='DEST', type=pathlib.Path,
+                        help='output path for vendor firmware')
 
     args = parser.parse_args()
 
-    update_firmware(args.source, args.dest, args.manifest)
+    update_firmware(args.source, args.dest)

src/main.py

@@ -317,7 +317,7 @@ def do_install(self, total_size=None):
 
         pkg = None
         if self.osins.needs_firmware:
-            pkg = asahi_firmware.core.FWPackage("firmware.tar")
+            pkg = asahi_firmware.core.FWPackage("firmware.tar", "firmware.cpio")
             self.ins.collect_firmware(pkg)
             pkg.close()
             self.osins.firmware_package = pkg

src/osinstall.py

@@ -123,6 +123,7 @@ def install(self, boot_obj_path):
                 logging.info(f"Firmware -> {base}")
                 os.makedirs(base, exist_ok=True)
                 shutil.copy(self.firmware_package.path, os.path.join(base, "firmware.tar"))
+                shutil.copy(self.firmware_package.cpio_path, os.path.join(base, "firmware.cpio"))
                 self.firmware_package.save_manifest(os.path.join(base, "manifest.txt"))
             if part.get("copy_installer_data", False):
                 mountpoint = self.dutil.mount(info.name)