Repository: osbuild/osbuild
Pull request #2123 (2025-06-17 09:48:10)

stages: Add org.osbuild.write-device stage
This stage writes a file to a device using dd. This is a rewrite/backport of one of the stages in osbuild-auto.

The osbuild-auto stage is used in automotive-image-builder to write the aboot image to the "boot_a" partition, to allow android boot systems to boot. We will want similar functionallity in bootc-image-builder, so it is important to upstream this.

alexlarsson: Added a commit to fix the nits

Patch:
diff --git a/stages/org.osbuild.write-device b/stages/org.osbuild.write-device
new file mode 100755
index 00000000..d8016848
--- /dev/null
+++ b/stages/org.osbuild.write-device
@@ -0,0 +1,49 @@
+#!/usr/bin/python3
+"""
+Write to device
+
+"""
+
+import fcntl
+import os
+import struct
+import subprocess
+import sys
+
+import osbuild.api
+from osbuild.util import parsing
+
+
+def blksize(device_path):
+    req = 0x80081272  # BLKGETSIZE64, result is bytes as unsigned 64-bit integer (uint64)
+    buf = b' ' * 8
+    fmt = 'L'
+
+    with open(device_path, "rb") as dev:
+        buf = fcntl.ioctl(dev.fileno(), req, buf)
+    return struct.unpack(fmt, buf)[0]
+
+
+def main(args, devices, options):
+    src = parsing.parse_location(options.get("from"), args)
+    device = devices["device"]["path"]
+
+    src_size = os.path.getsize(src)
+    print(f"src: '{src}' src_size: {src_size}")
+    device_size = blksize(device)
+    print(f"device: '{device}' device_size: {device_size}")
+    if src_size > device_size:
+        raise ValueError(
+            f"File too large ({src_size / (1024 * 1024):.1f} mb) for device ({device_size / (1024 * 1024):.1f} mb)")
+
+    cmd = ["dd", f"if={src}", f"of={device}", "status=progress", "conv=fsync"]
+    print("Running {cmd}")
+    subprocess.run(cmd, check=True)
+
+    return 0
+
+
+if __name__ == '__main__':
+    _args = osbuild.api.arguments()
+    r = main(_args, _args["devices"], _args["options"])
+    sys.exit(r)
diff --git a/stages/org.osbuild.write-device.meta.json b/stages/org.osbuild.write-device.meta.json
new file mode 100644
index 00000000..8c51c07e
--- /dev/null
+++ b/stages/org.osbuild.write-device.meta.json
@@ -0,0 +1,39 @@
+{
+  "summary": "Write a file to a device",
+  "description": [
+    "This allows writing a file to a raw device. This is useful for example",
+    "to write bootloader or firmware files to custom partitions."
+  ],
+  "schema_2": {
+    "devices": {
+      "type": "object",
+      "additionalProperties": true,
+      "required": [
+        "device"
+      ],
+      "properties": {
+        "device": {
+          "type": "object",
+          "additionalProperties": true
+        }
+      }
+    },
+    "inputs": {
+      "type": "object",
+      "additionalProperties": true
+    },
+    "options": {
+      "additionalProperties": false,
+      "required": [
+        "from"
+      ],
+      "properties": {
+        "from": {
+          "type": "string",
+          "description": "The source",
+          "pattern": "^input:\\/\\/[^\\/]+\\/"
+        }
+      }
+    }
+  }
+}


Test patch:
diff --git a/stages/test/test_write_device.py b/stages/test/test_write_device.py
new file mode 100644
index 00000000..53805da5
--- /dev/null
+++ b/stages/test/test_write_device.py
@@ -0,0 +1,89 @@
+#!/usr/bin/python3
+
+import contextlib
+import os
+import struct
+import subprocess
+from unittest.mock import call, mock_open, patch
+
+import pytest  # type: ignore
+
+STAGE_NAME = "org.osbuild.write-device"
+
+
+def create_args(tree):
+    tree.mkdir()
+    a_file = tree / "a-file"
+    a_file.write_text("Some content")
+
+    options = {
+        "from": "input://tree/a-file"
+    }
+    devices = {
+        "device": {
+            "path": "/dev/loop2",
+        },
+    }
+    args = {
+        "inputs": {
+            "tree": {
+                "path": f"{tree}",
+            }
+        },
+        "devices": devices,
+        "options": options,
+    }
+    return args
+
+
+def block_device_size(size):
+    return struct.pack('L', size)
+
+
+@patch("subprocess.run")
+@patch("builtins.open", mock_open(read_data="data"))
+def test_write_device(mocked_run, tmp_path, stage_module):
+    tree = tmp_path / "tree"
+    args = create_args(tree)
+
+    with patch("fcntl.ioctl", return_value=block_device_size(1024)):
+        stage_module.main(args, args["devices"], args["options"])
+
+    assert mocked_run.call_args_list == [
+        call(["dd", f"if={tree}/a-file", "of=/dev/loop2", "status=progress", "conv=fsync"], check=True)]
+
+
+@patch("builtins.open", mock_open(read_data="data"))
+def test_write_device_size_check(tmp_path, stage_module):
+    tree = tmp_path / "tree"
+    args = create_args(tree)
+
+    with pytest.raises(ValueError, match=r"File too large \(0.0 mb\) for device \(0.0 mb\)"):
+        with patch("fcntl.ioctl", return_value=block_device_size(4)):
+            stage_module.main(args, args["devices"], args["options"])
+
+
+@pytest.mark.skipif(os.getuid() != 0, reason="test must run as root")
+def test_write_device_integration(tmp_path, stage_module):
+    tree = tmp_path / "tree"
+    args = create_args(tree)
+    # ensure we have a ramdisk
+    subprocess.check_call(["modprobe", "brd"])
+    test_blk_device = "/dev/ram7"
+    if open(test_blk_device, "br").read(512) != b'\x00' * 512:
+        pytest.skip(f"block device {test_blk_device} not empty")
+    args["devices"]["device"]["path"] = test_blk_device
+    # ensure we have a test file
+    test_input_file = tree / "test-input.txt"
+    test_input_file.write_text("test-data")
+    args["options"]["from"] = "input://tree/test-input.txt"
+
+    with contextlib.ExitStack() as cm:
+        cm.callback(
+            subprocess.check_call,
+            ["dd", "if=/dev/zero", f"of={test_blk_device}", "bs=512", "count=1"],
+        )
+
+        stage_module.main(args, args["devices"], args["options"])
+
+        assert open(test_blk_device, "rb").read(128) == b"test-data" + b'\x00' * 119