vmspawn,machinectl: integrate io.systemd.StorageProvider (#41910)

PR #41776 introduced the io.systemd.StorageProvider Varlink interface and two backends ('block' exposes host block devices, 'fs' exposes regular files / dirs / subvolumes under /var/lib/storage), plus the storagectl(1) CLI to enumerate them. The only consumer so far was mount.storage. This series wires up the first of the three integrations called out in TODO.md: systemd-vmspawn --bind-volume=PROVIDER:VOLUME[:CONFIG][:K=V,...] Boot-time attach. Drives added this way are immutable at runtime. io.systemd.MachineInstance.AddStorage / .RemoveStorage Two new generic methods on the per-machine control socket. vmspawn implements them (this series); systemd-nspawn will reuse the same methods later. machinectl bind-volume MACHINE PROVIDER:VOLUME[:CONFIG][:K=V,...] machinectl unbind-volume MACHINE PROVIDER:VOLUME Runtime hotplug front-end: machinectl Acquire()s the fd locally and pushes it across to the target machine's MachineInstance socket. Volumes are identified by a user-visible name "<provider>:<volume>" (e.g. "block:/dev/sda"). The 3rd 'config' field is opaque to the shared layer and interpreted per backend — vmspawn maps it to a DiskType from disk_type_table[] (virtio-blk default, virtio-scsi, nvme, scsi-cd; same vocabulary as --extra-drive); future nspawn will read it as a mount path.
2026-08-13 17:08:14 +00:00 · 2026-05-06 12:20:26 +02:00
parent 9d2df9daf6 8a4451d2f0
commit 4ddfd04a48
29 changed files with 1371 additions and 183 deletions
--- a/TODO.md
+++ b/TODO.md
@@ -138,7 +138,6 @@ SPDX-License-Identifier: LGPL-2.1-or-later

 - StorageProvider interface + storagectl
  - hook-up in systemd-nspawn
-  - hook-up in systemd-vmspawn
  - hook-up in service manager (BindVolume=)
  - introduce a locking concept: right now all access to volumes is fully
    shared. Let's add a basic locking concept: supporting backends can take an
--- a/man/machinectl.xml
+++ b/man/machinectl.xml
@@ -357,6 +357,38 @@
        <xi:include href="version-info.xml" xpointer="v219"/></listitem>
      </varlistentry>

+      <varlistentry>
+        <term><command>bind-volume</command> <replaceable>NAME</replaceable> <replaceable>SPEC</replaceable></term>
+
+        <listitem><para>Acquire a storage volume from a
+        <citerefentry><refentrytitle>storagectl</refentrytitle><manvolnum>1</manvolnum></citerefentry>
+        provider and attach it to the running machine. <replaceable>SPEC</replaceable> is a string of the form
+        <literal><replaceable>PROVIDER</replaceable>:<replaceable>VOLUME</replaceable>[:<replaceable>CONFIG</replaceable>][:<replaceable>K=V</replaceable>,…]</literal>,
+        identical in grammar to the <option>--bind-volume=</option> argument of
+        <citerefentry><refentrytitle>systemd-vmspawn</refentrytitle><manvolnum>1</manvolnum></citerefentry>.</para>
+
+        <para>The attached volume is identified by the name <literal><replaceable>PROVIDER</replaceable>:<replaceable>VOLUME</replaceable></literal>
+        and may be detached at runtime via <command>unbind-volume</command>. Currently only supported for
+        <command>systemd-vmspawn</command> machines that expose an
+        <constant>io.systemd.MachineInstance</constant> control socket.</para>
+
+        <xi:include href="version-info.xml" xpointer="v261"/></listitem>
+      </varlistentry>
+
+      <varlistentry>
+        <term><command>unbind-volume</command> <replaceable>NAME</replaceable> <replaceable>STORAGE-NAME</replaceable></term>
+
+        <listitem><para>Detach a storage volume from the running machine. <replaceable>STORAGE-NAME</replaceable>
+        is the <literal><replaceable>PROVIDER</replaceable>:<replaceable>VOLUME</replaceable></literal>
+        identifier that was specified at <command>bind-volume</command> time. Volumes that were attached at machine
+        startup (e.g. via <option>--bind-volume=</option> on
+        <citerefentry><refentrytitle>systemd-vmspawn</refentrytitle><manvolnum>1</manvolnum></citerefentry>)
+        cannot be detached and will fail with
+        <constant>io.systemd.MachineInstance.StorageImmutable</constant>.</para>
+
+        <xi:include href="version-info.xml" xpointer="v261"/></listitem>
+      </varlistentry>
+
      <varlistentry>
        <term><command>copy-to</command> <replaceable>NAME</replaceable> <replaceable>PATH</replaceable> [<replaceable>PATH</replaceable>] <option>--force</option></term>

--- a/man/systemd-vmspawn.xml
+++ b/man/systemd-vmspawn.xml
@@ -566,6 +566,39 @@
          <xi:include href="version-info.xml" xpointer="v256"/></listitem>
        </varlistentry>

+        <varlistentry>
+          <term><option>--bind-volume=<replaceable>PROVIDER</replaceable>:<replaceable>VOLUME</replaceable>[:<replaceable>CONFIG</replaceable>][:<replaceable>K=V</replaceable>,…]</option></term>
+
+          <listitem><para>Acquire a storage volume from a
+          <citerefentry><refentrytitle>storagectl</refentrytitle><manvolnum>1</manvolnum></citerefentry>
+          provider and attach it to the virtual machine. <replaceable>PROVIDER</replaceable> is the
+          provider name (typically <literal>block</literal> or <literal>fs</literal>). <replaceable>VOLUME</replaceable>
+          is the volume name passed to the provider's <function>Acquire()</function> method.
+          <replaceable>CONFIG</replaceable> selects the guest device type and takes one of
+          <literal>virtio-blk</literal>, <literal>virtio-scsi</literal>, <literal>nvme</literal>, or
+          <literal>scsi-cd</literal>. If empty or omitted, defaults to <literal>virtio-blk</literal>.</para>
+
+          <para>The trailing comma-separated <replaceable>K=V</replaceable> list passes parameters to
+          <function>io.systemd.StorageProvider.Acquire()</function>: <varname>template=</varname>,
+          <varname>create=</varname> (one of <literal>any</literal>, <literal>new</literal>, <literal>open</literal>),
+          <varname>read-only=</varname> (or <varname>ro=</varname>; takes a boolean or <literal>auto</literal>),
+          <varname>size=</varname> / <varname>create-size=</varname> (size for created volumes),
+          <varname>request-as=</varname> (one of <literal>blk</literal>, <literal>reg</literal>,
+          <literal>dir</literal>; <literal>dir</literal> is rejected by vmspawn).</para>
+
+          <para>Each attached volume is identified by the name <literal><replaceable>PROVIDER</replaceable>:<replaceable>VOLUME</replaceable></literal>.
+          Volumes attached at startup via this option cannot be detached at runtime via
+          <command>machinectl unbind-volume</command>; only volumes added at runtime via
+          <command>machinectl bind-volume</command> are removable.</para>
+
+          <para>The provider is looked up under
+          <filename>/run/systemd/io.systemd.StorageProvider/</filename> for system mode (or
+          <filename>$XDG_RUNTIME_DIR/systemd/io.systemd.StorageProvider/</filename> for user mode), matching
+          the runtime scope chosen via <option>--user</option> / <option>--system</option>.</para>
+
+          <xi:include href="version-info.xml" xpointer="v261"/></listitem>
+        </varlistentry>
+
        <varlistentry>
          <term><option>--bind-user=</option></term>

--- a/shell-completion/bash/machinectl
+++ b/shell-completion/bash/machinectl
@@ -48,7 +48,7 @@ _machinectl() {
        [MACHINES]='status show start stop login shell enable disable poweroff reboot pause resume terminate kill
                    image-status show-image remove export-tar export-raw'
        [MACHINES_OR_FILES]='edit cat'
-        [MACHINE_ONLY]='clone rename set-limit'
+        [MACHINE_ONLY]='clone rename set-limit bind-volume unbind-volume'
        [READONLY]='read-only'
        [FILE]='import-tar import-raw'
        [MACHINES_AND_FILES]='copy-to copy-from bind'
--- a/shell-completion/bash/systemd-vmspawn
+++ b/shell-completion/bash/systemd-vmspawn
@@ -38,7 +38,7 @@ _systemd_vmspawn() {
        [BIND]='--bind --bind-ro'
        [SSH_KEY]='--ssh-key'
        [CONSOLE]='--console'
-        [ARG]='--cpus --ram --vsock-cid -M --machine --uuid --private-users --background --set-credential --load-credential --forward-journal-max-use --forward-journal-keep-free --forward-journal-max-file-size --forward-journal-max-files'
+        [ARG]='--cpus --ram --vsock-cid -M --machine --uuid --private-users --background --set-credential --load-credential --forward-journal-max-use --forward-journal-keep-free --forward-journal-max-file-size --forward-journal-max-files --bind-volume'
        [IMAGE_FORMAT]='--image-format'
        [IMAGE_DISK_TYPE]='--image-disk-type'
    )
--- a/shell-completion/zsh/_machinectl
+++ b/shell-completion/zsh/_machinectl
@@ -45,6 +45,8 @@
        "copy-to:Copy files from the host to a container"
        "copy-from:Copy files from a container to the host"
        "bind:Bind mount a path from the host into a container"
+        "bind-volume:Attach a storage volume to a running machine"
+        "unbind-volume:Detach a storage volume from a running machine"

        "list-images:Show available container and VM images"
        "image-status:Show image details"
@@ -115,6 +117,12 @@
            else stop=1
            fi ;;

+        bind-volume|unbind-volume)
+            if (( CURRENT == 2 )); then _sd_machines
+            elif (( CURRENT == 3 )); then _message "volume spec"
+            else stop=1
+            fi ;;
+
        read-only)
            if (( CURRENT == 2 )); then _machinectl_images
            elif (( CURRENT == 3 )); then _values 'read-only flag' 'true' 'false'
--- a/src/basic/user-util.h
+++ b/src/basic/user-util.h
@@ -90,6 +90,10 @@ int take_etc_passwd_lock(const char *root);
 #define UID_NOBODY ((uid_t) 65534U)
 #define GID_NOBODY ((gid_t) 65534U)

+/* Conventional size of a user-namespace UID/GID delegation block (64K).
+ * Untyped so it can be used in both UID and GID contexts without casts. */
+#define USERNS_RANGE_SIZE 0x10000U
+
 /* If REMOUNT_IDMAPPING_HOST_ROOT is set for remount_idmap() we'll include a mapping here that maps the host
 * root user accessing the idmapped mount to the this user ID on the backing fs. This is the last valid UID in
 * the *signed* 32-bit range. You might wonder why precisely use this specific UID for this purpose? Well, we
--- a/src/libsystemd/sd-json/json-util.h
+++ b/src/libsystemd/sd-json/json-util.h
@@ -268,6 +268,8 @@ enum {
        SD_JSON_BUILD_PAIR_CONDITION(condition, name, SD_JSON_BUILD_UNSIGNED(value))
 #define JSON_BUILD_PAIR_CONDITION_BOOLEAN(condition, name, value) \
        SD_JSON_BUILD_PAIR_CONDITION(condition, name, SD_JSON_BUILD_BOOLEAN(value))
+#define JSON_BUILD_PAIR_CONDITION_STRING(condition, name, value) \
+        SD_JSON_BUILD_PAIR_CONDITION(condition, name, SD_JSON_BUILD_STRING(value))
 #define JSON_BUILD_PAIR_CONDITION_STRV(condition, name, value) \
        SD_JSON_BUILD_PAIR_CONDITION(condition, name, SD_JSON_BUILD_STRV(value))

--- a/src/machine/machinectl.c
+++ b/src/machine/machinectl.c
@@ -29,16 +29,19 @@
 #include "cgroup-util.h"
 #include "edit-util.h"
 #include "env-util.h"
+#include "fd-util.h"
 #include "format-ifname.h"
 #include "format-table.h"
 #include "format-util.h"
 #include "hostname-util.h"
 #include "import-util.h"
 #include "in-addr-util.h"
+#include "json-util.h"
 #include "label-util.h"
 #include "log.h"
 #include "logs-show.h"
 #include "machine-dbus.h"
+#include "machine-util.h"
 #include "main-func.h"
 #include "nulstr-util.h"
 #include "osc-context.h"
@@ -55,6 +58,7 @@
 #include "ptyfwd.h"
 #include "runtime-scope.h"
 #include "stdio-util.h"
+#include "storage-util.h"
 #include "string-table.h"
 #include "string-util.h"
 #include "strv.h"
@@ -1338,6 +1342,120 @@ static int verb_copy_files(int argc, char *argv[], uintptr_t _data, void *userda
        return 0;
 }

+static int verb_bind_volume(int argc, char *argv[], uintptr_t _data, void *userdata) {
+        int r;
+
+        if (arg_transport != BUS_TRANSPORT_LOCAL)
+                return log_error_errno(SYNTHETIC_ERRNO(EOPNOTSUPP),
+                                       "bind-volume is only supported on the local transport.");
+
+        _cleanup_(bind_volume_freep) BindVolume *bv = NULL;
+        r = bind_volume_parse(argv[2], &bv);
+        if (r < 0)
+                return log_error_errno(r, "Failed to parse bind-volume argument '%s': %m", argv[2]);
+
+        (void) polkit_agent_open_if_enabled(arg_transport, arg_ask_password);
+
+        /* Locate and connect to the target machine before acquiring storage, so a missing
+         * machine doesn't trigger 'create=new' side effects on the StorageProvider. */
+        _cleanup_free_ char *address = NULL;
+        r = machine_get_control_address(argv[1], &address);
+        if (r == -EOPNOTSUPP)
+                return log_error_errno(r, "Machine '%s' does not expose a varlink control socket.", argv[1]);
+        if (r < 0)
+                return r;
+
+        _cleanup_(sd_varlink_unrefp) sd_varlink *vl = NULL;
+        r = sd_varlink_connect_address(&vl, address);
+        if (r < 0)
+                return log_error_errno(r, "Failed to connect to machine control socket %s: %m", address);
+
+        r = sd_varlink_set_allow_fd_passing_output(vl, true);
+        if (r < 0)
+                return log_error_errno(r, "Failed to enable fd passing on varlink connection: %m");
+
+        _cleanup_(storage_acquire_reply_done) StorageAcquireReply reply = STORAGE_ACQUIRE_REPLY_INIT;
+        _cleanup_free_ char *acquire_error_id = NULL;
+        r = storage_acquire_volume(arg_runtime_scope, bv, arg_ask_password, &acquire_error_id, &reply);
+        if (r < 0) {
+                if (acquire_error_id)
+                        return log_error_errno(r, "Failed to acquire storage volume '%s:%s' from provider: %s",
+                                               bv->provider, bv->volume, acquire_error_id);
+                return log_error_errno(r, "Failed to acquire storage volume '%s:%s' from provider: %m",
+                                       bv->provider, bv->volume);
+        }
+
+        int fd_index = sd_varlink_push_fd(vl, reply.fd);
+        if (fd_index < 0)
+                return log_error_errno(fd_index, "Failed to push storage fd onto varlink connection: %m");
+        TAKE_FD(reply.fd);
+
+        _cleanup_free_ char *name = strjoin(bv->provider, ":", bv->volume);
+        if (!name)
+                return log_oom();
+
+        sd_json_variant *vl_reply = NULL;
+        const char *error_id = NULL;
+        r = sd_varlink_callbo(
+                        vl,
+                        "io.systemd.MachineInstance.AddStorage",
+                        &vl_reply, &error_id,
+                        SD_JSON_BUILD_PAIR_INTEGER("fileDescriptorIndex", fd_index),
+                        SD_JSON_BUILD_PAIR_STRING("name", name),
+                        JSON_BUILD_PAIR_STRING_NON_EMPTY("config", bv->config));
+        if (r < 0)
+                return log_error_errno(r, "Failed to call io.systemd.MachineInstance.AddStorage: %m");
+        if (error_id)
+                return log_error_errno(sd_varlink_error_to_errno(error_id, vl_reply),
+                                       "AddStorage failed for '%s': %s", name, error_id);
+
+        return 0;
+}
+
+static int verb_unbind_volume(int argc, char *argv[], uintptr_t _data, void *userdata) {
+        int r;
+
+        if (arg_transport != BUS_TRANSPORT_LOCAL)
+                return log_error_errno(SYNTHETIC_ERRNO(EOPNOTSUPP),
+                                       "unbind-volume is only supported on the local transport.");
+
+        r = machine_storage_name_split(argv[2], /* ret_provider= */ NULL, /* ret_volume= */ NULL);
+        if (r == -ENOMEM)
+                return log_oom();
+        if (r < 0)
+                return log_error_errno(SYNTHETIC_ERRNO(EINVAL),
+                                       "Invalid unbind-volume name '%s', expected '<provider>:<volume>'.", argv[2]);
+
+        (void) polkit_agent_open_if_enabled(arg_transport, arg_ask_password);
+
+        _cleanup_free_ char *address = NULL;
+        r = machine_get_control_address(argv[1], &address);
+        if (r == -EOPNOTSUPP)
+                return log_error_errno(r, "Machine '%s' does not expose a varlink control socket.", argv[1]);
+        if (r < 0)
+                return r;
+
+        _cleanup_(sd_varlink_unrefp) sd_varlink *vl = NULL;
+        r = sd_varlink_connect_address(&vl, address);
+        if (r < 0)
+                return log_error_errno(r, "Failed to connect to machine control socket %s: %m", address);
+
+        sd_json_variant *reply = NULL;
+        const char *error_id = NULL;
+        r = sd_varlink_callbo(
+                        vl,
+                        "io.systemd.MachineInstance.RemoveStorage",
+                        &reply, &error_id,
+                        SD_JSON_BUILD_PAIR_STRING("name", argv[2]));
+        if (r < 0)
+                return log_error_errno(r, "Failed to call io.systemd.MachineInstance.RemoveStorage: %m");
+        if (error_id)
+                return log_error_errno(sd_varlink_error_to_errno(error_id, reply),
+                                       "RemoveStorage failed for '%s': %s", argv[2], error_id);
+
+        return 0;
+}
+
 static int verb_bind_mount(int argc, char *argv[], uintptr_t _data, void *userdata) {
        _cleanup_(sd_bus_error_free) sd_bus_error error = SD_BUS_ERROR_NULL;
        sd_bus *bus = ASSERT_PTR(userdata);
@@ -2606,6 +2724,8 @@ static int machinectl_main(int argc, char *argv[], sd_bus *bus) {
                { "login",           VERB_ANY, 2,        0,            verb_login_machine     },
                { "shell",           VERB_ANY, VERB_ANY, 0,            verb_shell_machine     },
                { "bind",            3,        4,        0,            verb_bind_mount        },
+                { "bind-volume",     3,        3,        0,            verb_bind_volume       },
+                { "unbind-volume",   3,        3,        0,            verb_unbind_volume     },
                { "edit",            2,        VERB_ANY, 0,            verb_edit_settings     },
                { "cat",             2,        VERB_ANY, 0,            verb_cat_settings      },
                { "copy-to",         3,        4,        0,            verb_copy_files        },
--- a/src/shared/machine-util.c
+++ b/src/shared/machine-util.c
@@ -4,7 +4,11 @@
 #include "extract-word.h"
 #include "machine-util.h"
 #include "parse-argument.h"
+#include "parse-util.h"
+#include "storage-util.h"
 #include "string-table.h"
+#include "string-util.h"
+#include "strv.h"

 static const char *const image_format_table[_IMAGE_FORMAT_MAX] = {
        [IMAGE_FORMAT_RAW]   = "raw",
@@ -13,6 +17,14 @@ static const char *const image_format_table[_IMAGE_FORMAT_MAX] = {

 DEFINE_STRING_TABLE_LOOKUP(image_format, ImageFormat);

+static const char *const read_only_mode_table[_READ_ONLY_MAX] = {
+        [READ_ONLY_NO]   = "no",
+        [READ_ONLY_YES]  = "yes",
+        [READ_ONLY_AUTO] = "auto",
+};
+
+DEFINE_STRING_TABLE_LOOKUP(read_only_mode, ReadOnlyMode);
+
 static const char *const disk_type_table[_DISK_TYPE_MAX] = {
        [DISK_TYPE_VIRTIO_BLK]        = "virtio-blk",
        [DISK_TYPE_VIRTIO_SCSI]       = "virtio-scsi",
@@ -100,3 +112,168 @@ int parse_disk_spec(
        *ret_path = TAKE_PTR(path);
        return 0;
 }
+
+BindVolume* bind_volume_free(BindVolume *v) {
+        if (!v)
+                return NULL;
+
+        free(v->provider);
+        free(v->volume);
+        free(v->config);
+        free(v->template);
+
+        return mfree(v);
+}
+
+static int bind_volume_apply_extra(BindVolume *v, const char *key, const char *value) {
+        int r;
+
+        assert(v);
+        assert(key);
+        assert(value);
+
+        if (streq(key, "template")) {
+                if (v->template)
+                        return -EINVAL;
+                if (!storage_template_name_is_valid(value))
+                        return -EINVAL;
+                r = free_and_strdup(&v->template, value);
+                if (r < 0)
+                        return r;
+                return 0;
+        }
+
+        if (streq(key, "create")) {
+                if (v->create_mode >= 0)
+                        return -EINVAL;
+                CreateMode m = create_mode_from_string(value);
+                if (m < 0)
+                        return m;
+                v->create_mode = m;
+                return 0;
+        }
+
+        if (STR_IN_SET(key, "read-only", "ro")) {
+                if (v->read_only >= 0)
+                        return -EINVAL;
+                ReadOnlyMode m = read_only_mode_from_string(value);
+                if (m < 0) {
+                        r = parse_boolean(value);
+                        if (r < 0)
+                                return r;
+                        m = r ? READ_ONLY_YES : READ_ONLY_NO;
+                }
+                v->read_only = m;
+                return 0;
+        }
+
+        if (STR_IN_SET(key, "size", "create-size")) {
+                if (v->create_size_bytes != UINT64_MAX)
+                        return -EINVAL;
+                uint64_t sz;
+                r = parse_size(value, 1024, &sz);
+                if (r < 0)
+                        return r;
+                if (sz == 0)
+                        return -EINVAL;
+                v->create_size_bytes = sz;
+                return 0;
+        }
+
+        if (streq(key, "request-as")) {
+                if (v->request_as >= 0)
+                        return -EINVAL;
+                VolumeType t = volume_type_from_string(value);
+                if (t < 0)
+                        return t;
+                v->request_as = t;
+                return 0;
+        }
+
+        return -EINVAL;
+}
+
+int bind_volume_parse(const char *arg, BindVolume **ret) {
+        _cleanup_(bind_volume_freep) BindVolume *v = NULL;
+        int r;
+
+        assert(arg);
+        assert(ret);
+
+        v = new(BindVolume, 1);
+        if (!v)
+                return -ENOMEM;
+
+        *v = BIND_VOLUME_INIT;
+
+        const char *p = arg;
+        _cleanup_free_ char *provider = NULL, *volume = NULL, *config = NULL;
+
+        r = extract_first_word(&p, &provider, ":", EXTRACT_DONT_COALESCE_SEPARATORS);
+        if (r < 0)
+                return r;
+        if (r == 0 || isempty(provider) || !storage_provider_name_is_valid(provider))
+                return -EINVAL;
+
+        r = extract_first_word(&p, &volume, ":", EXTRACT_DONT_COALESCE_SEPARATORS);
+        if (r < 0)
+                return r;
+        if (r == 0 || isempty(volume) || !storage_volume_name_is_valid(volume))
+                return -EINVAL;
+
+        r = extract_first_word(&p, &config, ":", EXTRACT_DONT_COALESCE_SEPARATORS);
+        if (r < 0)
+                return r;
+
+        v->provider = TAKE_PTR(provider);
+        v->volume = TAKE_PTR(volume);
+        if (!isempty(config)) {
+                if (!string_is_safe(config, /* flags= */ 0))
+                        return -EINVAL;
+                v->config = TAKE_PTR(config);
+        }
+
+        for (;;) {
+                _cleanup_free_ char *kv = NULL, *key = NULL, *value = NULL;
+
+                r = extract_first_word(&p, &kv, ",", 0);
+                if (r < 0)
+                        return r;
+                if (r == 0)
+                        break;
+
+                r = split_pair(kv, "=", &key, &value);
+                if (r < 0)
+                        return r;
+                if (isempty(key))
+                        return -EINVAL;
+
+                r = bind_volume_apply_extra(v, key, value);
+                if (r < 0)
+                        return r;
+        }
+
+        *ret = TAKE_PTR(v);
+        return 0;
+}
+
+int machine_storage_name_split(const char *s, char **ret_provider, char **ret_volume) {
+        _cleanup_free_ char *p = NULL, *v = NULL;
+        int r;
+
+        if (isempty(s))
+                return -EINVAL;
+
+        r = split_pair(s, ":", &p, &v);
+        if (r < 0)
+                return r;
+
+        if (!storage_provider_name_is_valid(p) || !storage_volume_name_is_valid(v))
+                return -EINVAL;
+
+        if (ret_provider)
+                *ret_provider = TAKE_PTR(p);
+        if (ret_volume)
+                *ret_volume = TAKE_PTR(v);
+        return 0;
+}
--- a/src/shared/machine-util.h
+++ b/src/shared/machine-util.h
@@ -2,6 +2,7 @@
 #pragma once

 #include "shared-forward.h"
+#include "storage-util.h"

 typedef enum ImageFormat {
        IMAGE_FORMAT_RAW,
@@ -30,3 +31,56 @@ int parse_disk_spec(
                ImageFormat *format,
                DiskType *disk_type,
                char **ret_path);
+
+typedef enum ReadOnlyMode {
+        READ_ONLY_NO,
+        READ_ONLY_YES,
+        READ_ONLY_AUTO,
+        _READ_ONLY_MAX,
+        _READ_ONLY_INVALID = -EINVAL,
+} ReadOnlyMode;
+
+DECLARE_STRING_TABLE_LOOKUP(read_only_mode, ReadOnlyMode);
+
+/* Map ReadOnlyMode onto the Acquire() wire tristate (-1 unset/auto, 0 no, 1 yes). */
+static inline int read_only_mode_to_tristate(ReadOnlyMode m) {
+        switch (m) {
+        case READ_ONLY_NO:  return 0;
+        case READ_ONLY_YES: return 1;
+        default:            return -1;
+        }
+}
+
+/* Parsed "PROVIDER:VOLUME[:CONFIG][:K=V,K=V,...]" used by --bind-volume,
+ * machinectl bind-volume, and (future) the BindVolume= unit setting. The 'config'
+ * field is opaque here and interpreted per-backend (vmspawn: a DiskType name;
+ * nspawn: a mount path). */
+typedef struct BindVolume {
+        char *provider;
+        char *volume;
+        char *config;
+
+        /* Acquire() parameters parsed from the trailing key=value list. */
+        char *template;
+        CreateMode create_mode;
+        ReadOnlyMode read_only;
+        uint64_t create_size_bytes;
+        VolumeType request_as;
+} BindVolume;
+
+#define BIND_VOLUME_INIT                                                        \
+        (BindVolume) {                                                          \
+                .create_mode       = _CREATE_MODE_INVALID,                      \
+                .read_only         = _READ_ONLY_INVALID,                        \
+                .create_size_bytes = UINT64_MAX,                                \
+                .request_as        = _VOLUME_TYPE_INVALID,                      \
+        }
+
+BindVolume* bind_volume_free(BindVolume *v);
+DEFINE_TRIVIAL_CLEANUP_FUNC(BindVolume*, bind_volume_free);
+
+int bind_volume_parse(const char *arg, BindVolume **ret);
+
+/* Validate a "<provider>:<volume>" binding name as used by AddStorage/RemoveStorage.
+ * ret_provider/ret_volume may each be NULL when the caller only wants validation. */
+int machine_storage_name_split(const char *s, char **ret_provider, char **ret_volume);
--- a/src/shared/meson.build
+++ b/src/shared/meson.build
@@ -199,6 +199,7 @@ shared_sources = files(
        'socket-netlink.c',
        'specifier.c',
        'ssl-util.c',
+        'storage-util.c',
        'switch-root.c',
        'swtpm-util.c',
        'tar-util.c',
--- a/src/shared/nsresource.h
+++ b/src/shared/nsresource.h
@@ -2,9 +2,10 @@
 #pragma once

 #include "shared-forward.h"
+#include "user-util.h"

 /* Helpful constants for the only numbers of UIDs that can currently be allocated */
-#define NSRESOURCE_UIDS_64K 0x10000U
+#define NSRESOURCE_UIDS_64K USERNS_RANGE_SIZE
 #define NSRESOURCE_UIDS_1 1U

 int nsresource_connect(sd_varlink **ret);
--- a/src/shared/storage-util.c
+++ b/src/shared/storage-util.c
@@ -0,0 +1,130 @@
+/* SPDX-License-Identifier: LGPL-2.1-or-later */
+
+#include "sd-varlink.h"
+
+#include "alloc-util.h"
+#include "fd-util.h"
+#include "json-util.h"
+#include "machine-util.h"
+#include "path-lookup.h"
+#include "path-util.h"
+#include "runtime-scope.h"
+#include "string-table.h"
+#include "storage-util.h"
+
+static const char *volume_type_table[_VOLUME_TYPE_MAX] = {
+        [VOLUME_BLK] = "blk",
+        [VOLUME_REG] = "reg",
+        [VOLUME_DIR] = "dir",
+};
+
+static const char *create_mode_table[_CREATE_MODE_MAX] = {
+        [CREATE_ANY]  = "any",
+        [CREATE_NEW]  = "new",
+        [CREATE_OPEN] = "open",
+};
+
+DEFINE_STRING_TABLE_LOOKUP(volume_type, VolumeType);
+DEFINE_STRING_TABLE_LOOKUP(create_mode, CreateMode);
+
+JSON_DISPATCH_ENUM_DEFINE(json_dispatch_volume_type, VolumeType, volume_type_from_string);
+JSON_DISPATCH_ENUM_DEFINE(json_dispatch_create_mode, CreateMode, create_mode_from_string);
+
+void storage_acquire_reply_done(StorageAcquireReply *reply) {
+        if (!reply)
+                return;
+
+        reply->fd = safe_close(reply->fd);
+}
+
+int storage_acquire_volume(
+                RuntimeScope scope,
+                const BindVolume *bv,
+                bool allow_interactive_auth,
+                char **reterr_error_id,
+                StorageAcquireReply *ret) {
+
+        int r;
+
+        assert(bv);
+        assert(bv->provider);
+        assert(bv->volume);
+        assert(ret);
+
+        /* Defense-in-depth: this is a libshared helper that may grow new callers; reject
+         * provider names that could escape the StorageProvider runtime directory. */
+        if (!storage_provider_name_is_valid(bv->provider))
+                return -EINVAL;
+
+        _cleanup_free_ char *socket_path = NULL;
+        r = runtime_directory_generic(scope, "systemd/io.systemd.StorageProvider", &socket_path);
+        if (r < 0)
+                return r;
+
+        if (!path_extend(&socket_path, bv->provider))
+                return -ENOMEM;
+
+        _cleanup_(sd_varlink_unrefp) sd_varlink *link = NULL;
+        r = sd_varlink_connect_address(&link, socket_path);
+        if (r < 0)
+                return r;
+
+        r = sd_varlink_set_allow_fd_passing_input(link, true);
+        if (r < 0)
+                return r;
+
+        sd_json_variant *mreply = NULL;
+        const char *merror_id = NULL;
+        r = sd_varlink_callbo(
+                        link,
+                        "io.systemd.StorageProvider.Acquire",
+                        &mreply,
+                        &merror_id,
+                        SD_JSON_BUILD_PAIR_STRING("name", bv->volume),
+                        JSON_BUILD_PAIR_CONDITION_STRING(bv->create_mode >= 0, "createMode", create_mode_to_string(bv->create_mode)),
+                        JSON_BUILD_PAIR_STRING_NON_EMPTY("template", bv->template),
+                        JSON_BUILD_PAIR_TRISTATE_NON_NULL("readOnly", read_only_mode_to_tristate(bv->read_only)),
+                        JSON_BUILD_PAIR_CONDITION_STRING(bv->request_as >= 0, "requestAs", volume_type_to_string(bv->request_as)),
+                        JSON_BUILD_PAIR_UNSIGNED_NOT_EQUAL("createSizeBytes", bv->create_size_bytes, UINT64_MAX),
+                        SD_JSON_BUILD_PAIR_BOOLEAN("allowInteractiveAuthentication", allow_interactive_auth));
+        if (r < 0)
+                return r;
+
+        if (merror_id) {
+                if (reterr_error_id) {
+                        char *copy = strdup(merror_id);
+                        if (!copy)
+                                return -ENOMEM;
+                        *reterr_error_id = copy;
+                }
+
+                r = sd_varlink_error_to_errno(merror_id, mreply);
+                return r == -EBADR ? -EPROTO : r;
+        }
+
+        /* tmp.fd holds the JSON fd index until sd_varlink_take_fd() swaps it for the real fd. */
+        StorageAcquireReply tmp = STORAGE_ACQUIRE_REPLY_INIT;
+
+        static const sd_json_dispatch_field dispatch_table[] = {
+                { "fileDescriptorIndex", _SD_JSON_VARIANT_TYPE_INVALID, sd_json_dispatch_int,         voffsetof(StorageAcquireReply, fd),        SD_JSON_MANDATORY },
+                { "readOnly",            SD_JSON_VARIANT_BOOLEAN,       sd_json_dispatch_tristate,    voffsetof(StorageAcquireReply, read_only), 0                 },
+                { "type",                SD_JSON_VARIANT_STRING,        json_dispatch_volume_type,    voffsetof(StorageAcquireReply, type),      SD_JSON_MANDATORY },
+                { "baseUID",             _SD_JSON_VARIANT_TYPE_INVALID, sd_json_dispatch_uid_gid,     voffsetof(StorageAcquireReply, base_uid),  0                 },
+                { "baseGID",             _SD_JSON_VARIANT_TYPE_INVALID, sd_json_dispatch_uid_gid,     voffsetof(StorageAcquireReply, base_gid),  0                 },
+                {}
+        };
+
+        r = sd_json_dispatch(mreply, dispatch_table, SD_JSON_ALLOW_EXTENSIONS, &tmp);
+        if (r < 0)
+                return r;
+        if (tmp.fd < 0)
+                return -EBADMSG;
+
+        _cleanup_close_ int fd = sd_varlink_take_fd(link, tmp.fd);
+        if (fd < 0)
+                return fd;
+
+        tmp.fd = TAKE_FD(fd);
+        *ret = tmp;
+        return 0;
+}
--- a/src/storage/storage-util.h
+++ b/src/storage/storage-util.h
@@ -41,3 +41,32 @@ static inline bool storage_template_name_is_valid(const char *n) {
 static inline bool storage_provider_name_is_valid(const char *n) {
        return string_is_safe(n, STRING_FILENAME);
 }
+
+typedef struct StorageAcquireReply {
+        int fd;
+        VolumeType type;
+        int read_only;
+        uid_t base_uid;
+        gid_t base_gid;
+} StorageAcquireReply;
+
+#define STORAGE_ACQUIRE_REPLY_INIT                                              \
+        (StorageAcquireReply) {                                                 \
+                .fd        = -EBADF,                                            \
+                .type      = _VOLUME_TYPE_INVALID,                              \
+                .read_only = -1,                                                \
+                .base_uid  = UID_INVALID,                                       \
+                .base_gid  = GID_INVALID,                                       \
+        }
+
+void storage_acquire_reply_done(StorageAcquireReply *reply);
+
+/* On varlink failure, reterr_error_id (if non-NULL) is set to the io.systemd.StorageProvider.*
+ * error name. The reply is untouched on any error. */
+typedef struct BindVolume BindVolume;
+int storage_acquire_volume(
+                RuntimeScope scope,
+                const BindVolume *bv,
+                bool allow_interactive_auth,
+                char **reterr_error_id,
+                StorageAcquireReply *ret);
--- a/src/shared/varlink-io.systemd.MachineInstance.c
+++ b/src/shared/varlink-io.systemd.MachineInstance.c
@@ -25,8 +25,27 @@ static SD_VARLINK_DEFINE_METHOD_FULL(
                SD_VARLINK_FIELD_COMMENT("Event-specific payload"),
                SD_VARLINK_DEFINE_OUTPUT(data, SD_VARLINK_OBJECT, SD_VARLINK_NULLABLE));

+static SD_VARLINK_DEFINE_METHOD(
+                AddStorage,
+                SD_VARLINK_FIELD_COMMENT("Index of the attached file descriptor for the storage volume"),
+                SD_VARLINK_DEFINE_INPUT(fileDescriptorIndex, SD_VARLINK_INT, 0),
+                SD_VARLINK_FIELD_COMMENT("Unique storage name of the form '<provider>:<volume>' identifying this binding for later removal"),
+                SD_VARLINK_DEFINE_INPUT(name, SD_VARLINK_STRING, 0),
+                SD_VARLINK_FIELD_COMMENT("Backend-specific configuration"),
+                SD_VARLINK_DEFINE_INPUT(config, SD_VARLINK_STRING, SD_VARLINK_NULLABLE));
+
+static SD_VARLINK_DEFINE_METHOD(
+                RemoveStorage,
+                SD_VARLINK_FIELD_COMMENT("Unique storage name '<provider>:<volume>' to detach"),
+                SD_VARLINK_DEFINE_INPUT(name, SD_VARLINK_STRING, 0));
+
 static SD_VARLINK_DEFINE_ERROR(NotConnected);
 static SD_VARLINK_DEFINE_ERROR(NotSupported);
+static SD_VARLINK_DEFINE_ERROR(NoSuchStorage);
+static SD_VARLINK_DEFINE_ERROR(StorageExists);
+static SD_VARLINK_DEFINE_ERROR(StorageImmutable);
+static SD_VARLINK_DEFINE_ERROR(BadConfig);
+static SD_VARLINK_DEFINE_ERROR(ConfigNotSupported);

 SD_VARLINK_DEFINE_INTERFACE(
                io_systemd_MachineInstance,
@@ -45,7 +64,21 @@ SD_VARLINK_DEFINE_INTERFACE(
                &vl_method_Describe,
                SD_VARLINK_SYMBOL_COMMENT("Subscribe to machine events. Returns a stream of events as they occur."),
                &vl_method_SubscribeEvents,
+                SD_VARLINK_SYMBOL_COMMENT("Attach a storage volume (passed via file descriptor) to the running machine"),
+                &vl_method_AddStorage,
+                SD_VARLINK_SYMBOL_COMMENT("Detach a previously-attached storage volume from the running machine"),
+                &vl_method_RemoveStorage,
                SD_VARLINK_SYMBOL_COMMENT("The connection to the machine backend is not available"),
                &vl_error_NotConnected,
                SD_VARLINK_SYMBOL_COMMENT("The requested operation is not supported"),
-                &vl_error_NotSupported);
+                &vl_error_NotSupported,
+                SD_VARLINK_SYMBOL_COMMENT("The named storage binding does not exist"),
+                &vl_error_NoSuchStorage,
+                SD_VARLINK_SYMBOL_COMMENT("A storage binding with this name already exists"),
+                &vl_error_StorageExists,
+                SD_VARLINK_SYMBOL_COMMENT("The storage binding cannot be detached at runtime (e.g. attached at boot)"),
+                &vl_error_StorageImmutable,
+                SD_VARLINK_SYMBOL_COMMENT("The supplied 'config' value is not valid for this backend"),
+                &vl_error_BadConfig,
+                SD_VARLINK_SYMBOL_COMMENT("The supplied 'config' value is recognized but not supported by this backend"),
+                &vl_error_ConfigNotSupported);
--- a/src/storage/meson.build
+++ b/src/storage/meson.build
@@ -4,18 +4,15 @@ executables += [
        libexec_template + {
                'name' : 'systemd-storage-block',
                'sources' : files('storage-block.c'),
-                'extract' : files('storage-util.c')
        },
        libexec_template + {
                'name' : 'systemd-storage-fs',
                'sources' : files('storage-fs.c'),
-                'objects' : ['systemd-storage-block'],
        },
        executable_template + {
                'name' : 'storagectl',
                'public' : true,
                'sources' : files('storagectl.c'),
-                'objects' : ['systemd-storage-block'],
        },
 ]

--- a/src/storage/storage-util.c
+++ b/src/storage/storage-util.c
@@ -1,23 +0,0 @@
-/* SPDX-License-Identifier: LGPL-2.1-or-later */
-
-#include "json-util.h"
-#include "string-table.h"
-#include "storage-util.h"
-
-static const char *volume_type_table[_VOLUME_TYPE_MAX] = {
-        [VOLUME_BLK] = "blk",
-        [VOLUME_REG] = "reg",
-        [VOLUME_DIR] = "dir",
-};
-
-static const char *create_mode_table[_CREATE_MODE_MAX] = {
-        [CREATE_ANY]  = "any",
-        [CREATE_NEW]  = "new",
-        [CREATE_OPEN] = "open",
-};
-
-DEFINE_STRING_TABLE_LOOKUP(volume_type, VolumeType);
-DEFINE_STRING_TABLE_LOOKUP(create_mode, CreateMode);
-
-JSON_DISPATCH_ENUM_DEFINE(json_dispatch_volume_type, VolumeType, volume_type_from_string);
-JSON_DISPATCH_ENUM_DEFINE(json_dispatch_create_mode, CreateMode, create_mode_from_string);
--- a/src/storage/storagectl.c
+++ b/src/storage/storagectl.c
@@ -19,7 +19,7 @@
 #include "format-table.h"
 #include "format-util.h"
 #include "help-util.h"
-#include "json-util.h"
+#include "machine-util.h"
 #include "main-func.h"
 #include "mount-util.h"
 #include "namespace-util.h"
@@ -524,7 +524,7 @@ static int run_as_mount_helper(int argc, char *argv[]) {
        _cleanup_free_ char *filtered = NULL, *template = NULL;
        CreateMode create_mode = _CREATE_MODE_INVALID;
        uint64_t create_size = UINT64_MAX;
-        int read_only = -1;
+        ReadOnlyMode read_only = _READ_ONLY_INVALID;
        for (const char *p = options;;) {
                _cleanup_free_ char *word = NULL;

@@ -555,9 +555,9 @@ static int run_as_mount_helper(int argc, char *argv[]) {
                        } else
                                return log_error_errno(SYNTHETIC_ERRNO(EINVAL), "Unknown mount option '%s', refusing.", word);
                } else if (streq(word, "ro"))
-                        read_only = true;
+                        read_only = READ_ONLY_YES;
                else if (streq(word, "rw"))
-                        read_only = false;
+                        read_only = READ_ONLY_NO;
                else if (!strextend_with_separator(&filtered, ",", word))
                        return log_oom();
        }
@@ -565,141 +565,69 @@ static int run_as_mount_helper(int argc, char *argv[]) {
        if (fake)
                return 0;

-        _cleanup_free_ char *socket_path = NULL;
-        r = runtime_directory_generic(arg_runtime_scope, "systemd/io.systemd.StorageProvider", &socket_path);
-        if (r < 0)
-                return log_error_errno(r, "Failed to determine socket directory: %m");
-
-        if (!path_extend(&socket_path, provider))
-                return log_oom();
-
-        _cleanup_(sd_varlink_unrefp) sd_varlink *link = NULL;
-        r = sd_varlink_connect_address(&link, socket_path);
-        if (r < 0)
-                return log_error_errno(r, "Failed to connect to '%s': %m", socket_path);
-
-        r = sd_varlink_set_allow_fd_passing_input(link, true);
-        if (r < 0)
-                return log_error_errno(r, "Failed to enable file descriptor passing: %m");
-
        (void) polkit_agent_open_if_enabled(BUS_TRANSPORT_LOCAL, arg_ask_password);

-        sd_json_variant *mreply = NULL;
-        const char *merror_id = NULL, *vtype = fstype ? "reg" : "dir";
-        r = sd_varlink_callbo(
-                        link,
-                        "io.systemd.StorageProvider.Acquire",
-                        &mreply,
-                        &merror_id,
-                        SD_JSON_BUILD_PAIR_STRING("name", name),
-                        SD_JSON_BUILD_PAIR_CONDITION(create_mode >= 0, "createMode", SD_JSON_BUILD_STRING(create_mode_to_string(create_mode))),
-                        JSON_BUILD_PAIR_STRING_NON_EMPTY("template", template),
-                        SD_JSON_BUILD_PAIR_CONDITION(read_only >= 0, "readOnly", SD_JSON_BUILD_BOOLEAN(read_only)),
-                        SD_JSON_BUILD_PAIR_STRING("requestAs", vtype),
-                        SD_JSON_BUILD_PAIR_CONDITION(create_size != UINT64_MAX, "createSizeBytes", SD_JSON_BUILD_UNSIGNED(create_size)),
-                        SD_JSON_BUILD_PAIR_BOOLEAN("allowInteractiveAuthentication", arg_ask_password));
-        if (r < 0)
-                return log_error_errno(r, "Failed to issue io.systemd.StorageProvider.Acquire() varlink call: %m");
-        _cleanup_(sd_json_variant_unrefp) sd_json_variant *reply = sd_json_variant_ref(mreply);
-        if (merror_id) {
-                /* Copy out the error ID, as the follow-up call will invalidate it */
-                _cleanup_free_ char *error_id = strdup(merror_id);
-                if (!error_id)
-                        return log_oom();
+        VolumeType requested_type = fstype ? VOLUME_REG : VOLUME_DIR;

-                /* Hmm, the type might not have been right for the backend or the volume? then try
-                 * again, and switch from "reg" to "blk", maybe it works then. (We keep the original
-                 * reply referenced, since we prefer generating an error for the first error.) */
-                if (streq(vtype, "reg") && STR_IN_SET(error_id,
-                                         "io.systemd.StorageProvider.TypeNotSupported",
-                                         "io.systemd.StorageProvider.WrongType")) {
+        BindVolume bv = BIND_VOLUME_INIT;
+        bv.provider = provider;
+        bv.volume = name;
+        bv.create_mode = create_mode;
+        bv.template = template;
+        bv.read_only = read_only;
+        bv.request_as = requested_type;
+        bv.create_size_bytes = create_size;

-                        sd_json_variant *freply = NULL;
-                        const char *ferror_id = NULL;
-                        r = sd_varlink_callbo(
-                                        link,
-                                        "io.systemd.StorageProvider.Acquire",
-                                        &freply,
-                                        &ferror_id,
-                                        SD_JSON_BUILD_PAIR_STRING("name", name),
-                                        SD_JSON_BUILD_PAIR_CONDITION(create_mode >= 0, "createMode", SD_JSON_BUILD_STRING(create_mode_to_string(create_mode))),
-                                        JSON_BUILD_PAIR_STRING_NON_EMPTY("template", template),
-                                        SD_JSON_BUILD_PAIR_CONDITION(read_only >= 0, "readOnly", SD_JSON_BUILD_BOOLEAN(read_only)),
-                                        SD_JSON_BUILD_PAIR_STRING("requestAs", "blk"),
-                                        SD_JSON_BUILD_PAIR_CONDITION(create_size != UINT64_MAX, "createSizeBytes", SD_JSON_BUILD_UNSIGNED(create_size)),
-                                        SD_JSON_BUILD_PAIR_BOOLEAN("allowInteractiveAuthentication", arg_ask_password));
-                        if (r < 0)
-                                return log_error_errno(r, "Failed to issue io.systemd.StorageProvider.Acquire() varlink call: %m");
-                        if (!ferror_id) {
-                                /* The 2nd call worked? then let's forget about the first failure */
-                                sd_json_variant_unref(reply);
-                                reply = sd_json_variant_ref(freply);
-                                error_id = mfree(error_id);
-                        }
-
-                        /* NB: if both fail we show the Varlink error of the first call here, i.e. of the preferred type */
-                }
-
-                if (error_id) {
-                        if (streq(error_id, "io.systemd.StorageProvider.NoSuchVolume"))
-                                return log_error_errno(SYNTHETIC_ERRNO(ENOENT), "Volume '%s' not known.", name);
-                        if (streq(error_id, "io.systemd.StorageProvider.NoSuchTemplate"))
-                                return log_error_errno(SYNTHETIC_ERRNO(ENOENT), "Template '%s' not known.", template);
-                        if (streq(error_id, "io.systemd.StorageProvider.VolumeExists"))
-                                return log_error_errno(SYNTHETIC_ERRNO(EEXIST), "Volume '%s' exists already.", name);
-                        if (streq(error_id, "io.systemd.StorageProvider.TypeNotSupported"))
-                                return log_error_errno(SYNTHETIC_ERRNO(EOPNOTSUPP), "Storage provider does not support the specified volume type '%s'.", vtype);
-                        if (streq(error_id, "io.systemd.StorageProvider.WrongType"))
-                                return log_error_errno(SYNTHETIC_ERRNO(EADDRNOTAVAIL), "Volume '%s' is not of type '%s'.", name, vtype);
-                        if (streq(error_id, "io.systemd.StorageProvider.CreateNotSupported"))
-                                return log_error_errno(SYNTHETIC_ERRNO(EOPNOTSUPP), "Storage provider does not support creating volumes.");
-                        if (streq(error_id, "io.systemd.StorageProvider.CreateSizeRequired"))
-                                return log_error_errno(SYNTHETIC_ERRNO(ENODATA), "Storage provider requires a create size to be provided when creating volumes on-the-fly. Use 'storage.create-size=' mount option.");
-                        if (streq(error_id, "io.systemd.StorageProvider.ReadOnlyVolume"))
-                                return log_error_errno(SYNTHETIC_ERRNO(EROFS), "Volume '%s' is read-only.", name);
-                        if (streq(error_id, "io.systemd.StorageProvider.BadTemplate"))
-                                return log_error_errno(SYNTHETIC_ERRNO(EADDRNOTAVAIL), "Template does not apply to this volume type.");
-
-                        r = sd_varlink_error_to_errno(error_id, reply); /* If this is a system errno style error, output it with %m */
-                        if (r != -EBADR)
-                                return log_error_errno(r, "Failed to issue io.systemd.StorageProvider.Acquire() varlink call: %m");
-
-                        return log_error_errno(r, "Failed to issue io.systemd.StorageProvider.Acquire() varlink call: %s", error_id);
+        _cleanup_(storage_acquire_reply_done) StorageAcquireReply reply = STORAGE_ACQUIRE_REPLY_INIT;
+        _cleanup_free_ char *acquire_error_id = NULL;
+        r = storage_acquire_volume(arg_runtime_scope, &bv, arg_ask_password, &acquire_error_id, &reply);
+        if (r < 0 && fstype &&
+            STR_IN_SET(strna(acquire_error_id),
+                       "io.systemd.StorageProvider.TypeNotSupported",
+                       "io.systemd.StorageProvider.WrongType")) {
+                _cleanup_(storage_acquire_reply_done) StorageAcquireReply retry = STORAGE_ACQUIRE_REPLY_INIT;
+                assert(bv.request_as == VOLUME_REG);
+                bv.request_as = VOLUME_BLK;
+                int k = storage_acquire_volume(arg_runtime_scope, &bv, arg_ask_password, /* reterr_error_id= */ NULL, &retry);
+                if (k >= 0) {
+                        storage_acquire_reply_done(&reply);
+                        reply = retry;
+                        retry = STORAGE_ACQUIRE_REPLY_INIT;
+                        acquire_error_id = mfree(acquire_error_id);
+                        requested_type = VOLUME_BLK;
+                        r = 0;
                }
        }

-        struct {
-                unsigned fd_idx;
-                int read_only;
-                const char *type;
-                uid_t base_uid;
-                gid_t base_gid;
-        } p = {
-                .fd_idx = UINT_MAX,
-                .read_only = -1,
-                .base_uid = UID_INVALID,
-                .base_gid = GID_INVALID,
-        };
+        if (r < 0) {
+                const char *eid = acquire_error_id;

-        static const sd_json_dispatch_field dispatch_table[] = {
-                { "fileDescriptorIndex", _SD_JSON_VARIANT_TYPE_INVALID, sd_json_dispatch_uint,         voffsetof(p, fd_idx),     SD_JSON_MANDATORY },
-                { "readOnly",            SD_JSON_VARIANT_BOOLEAN,       sd_json_dispatch_tristate,     voffsetof(p, read_only),  0                 },
-                { "type",                SD_JSON_VARIANT_STRING,        sd_json_dispatch_const_string, voffsetof(p, type),       SD_JSON_MANDATORY },
-                { "baseUID",             _SD_JSON_VARIANT_TYPE_INVALID, sd_json_dispatch_uid_gid,      voffsetof(p, base_uid),   0                 },
-                { "baseGID",             _SD_JSON_VARIANT_TYPE_INVALID, sd_json_dispatch_uid_gid,      voffsetof(p, base_gid),   0                 },
-                {}
-        };
+                if (streq_ptr(eid, "io.systemd.StorageProvider.NoSuchVolume"))
+                        return log_error_errno(SYNTHETIC_ERRNO(ENOENT), "Volume '%s' not known.", name);
+                if (streq_ptr(eid, "io.systemd.StorageProvider.NoSuchTemplate"))
+                        return log_error_errno(SYNTHETIC_ERRNO(ENOENT), "Template '%s' not known.", template);
+                if (streq_ptr(eid, "io.systemd.StorageProvider.VolumeExists"))
+                        return log_error_errno(SYNTHETIC_ERRNO(EEXIST), "Volume '%s' exists already.", name);
+                if (streq_ptr(eid, "io.systemd.StorageProvider.TypeNotSupported"))
+                        return log_error_errno(SYNTHETIC_ERRNO(EOPNOTSUPP), "Storage provider does not support the specified volume type '%s'.", volume_type_to_string(requested_type));
+                if (streq_ptr(eid, "io.systemd.StorageProvider.WrongType"))
+                        return log_error_errno(SYNTHETIC_ERRNO(EADDRNOTAVAIL), "Volume '%s' is not of type '%s'.", name, volume_type_to_string(requested_type));
+                if (streq_ptr(eid, "io.systemd.StorageProvider.CreateNotSupported"))
+                        return log_error_errno(SYNTHETIC_ERRNO(EOPNOTSUPP), "Storage provider does not support creating volumes.");
+                if (streq_ptr(eid, "io.systemd.StorageProvider.CreateSizeRequired"))
+                        return log_error_errno(SYNTHETIC_ERRNO(ENODATA), "Storage provider requires a create size to be provided when creating volumes on-the-fly. Use 'storage.create-size=' mount option.");
+                if (streq_ptr(eid, "io.systemd.StorageProvider.ReadOnlyVolume"))
+                        return log_error_errno(SYNTHETIC_ERRNO(EROFS), "Volume '%s' is read-only.", name);
+                if (streq_ptr(eid, "io.systemd.StorageProvider.BadTemplate"))
+                        return log_error_errno(SYNTHETIC_ERRNO(EADDRNOTAVAIL), "Template does not apply to this volume type.");

-        r = sd_json_dispatch(reply, dispatch_table, SD_JSON_ALLOW_EXTENSIONS, &p);
-        if (r < 0)
-                return log_error_errno(r, "Failed to decode Acquire() reply: %m");
-
-        _cleanup_close_ int fd = sd_varlink_take_fd(link, p.fd_idx);
-        if (fd < 0)
-                return log_error_errno(fd, "Failed to acquire fd from Varlink connection: %m");
+                if (eid)
+                        return log_error_errno(r, "Failed to issue io.systemd.StorageProvider.Acquire() varlink call (%s): %m", eid);
+                return log_error_errno(r, "Failed to issue io.systemd.StorageProvider.Acquire() varlink call: %m");
+        }

        struct stat st;
-        if (fstat(fd, &st) < 0)
+        if (fstat(reply.fd, &st) < 0)
                return log_error_errno(errno, "Failed to stat returned file descriptor: %m");

        _cleanup_strv_free_ char **cmdline = strv_new("mount", "-c");
@@ -707,7 +635,7 @@ static int run_as_mount_helper(int argc, char *argv[]) {
                return log_oom();

        if (fstype) {
-                if (!STR_IN_SET(p.type, "reg", "blk"))
+                if (!IN_SET(reply.type, VOLUME_REG, VOLUME_BLK))
                        return log_error_errno(SYNTHETIC_ERRNO(EINVAL), "Mounting as file system type '%s' requested, but volume is not a block device or regular file.", fstype);

                r = stat_verify_regular_or_block(&st);
@@ -717,31 +645,31 @@ static int run_as_mount_helper(int argc, char *argv[]) {
                if (strv_extend_strv(&cmdline, STRV_MAKE("-t", fstype), /* filter_duplicates= */ false) < 0)
                        return log_oom();
        } else {
-                if (!streq(p.type, "dir"))
+                if (reply.type != VOLUME_DIR)
                        return log_error_errno(SYNTHETIC_ERRNO(EINVAL), "Mount as directory requested, but volume is not a directory.");

-                if (!uid_is_valid(p.base_uid) || !gid_is_valid(p.base_gid))
+                if (!uid_is_valid(reply.base_uid) || !gid_is_valid(reply.base_gid))
                        return log_error_errno(SYNTHETIC_ERRNO(EINVAL), "Provider did not report base UID/GID, cannot mount.");

-                if (p.base_uid > UINT32_MAX - 0x10000U ||
-                    p.base_gid > UINT32_MAX - 0x10000U)
+                if (reply.base_uid > UINT32_MAX - USERNS_RANGE_SIZE ||
+                    reply.base_gid > UINT32_MAX - USERNS_RANGE_SIZE)
                        return log_error_errno(SYNTHETIC_ERRNO(EINVAL), "Returned base UID/GID out of range.");

                r = stat_verify_directory(&st);
                if (r < 0)
                        return log_error_errno(r, "File descriptor for directory volume is not a directory inode: %m");

-                if (st.st_uid < p.base_uid || st.st_uid >= p.base_uid + 0x10000 ||
-                    st.st_gid < p.base_gid || st.st_gid >= p.base_gid + 0x10000)
+                if (st.st_uid < reply.base_uid || st.st_uid >= reply.base_uid + USERNS_RANGE_SIZE ||
+                    st.st_gid < reply.base_gid || st.st_gid >= reply.base_gid + USERNS_RANGE_SIZE)
                        return log_error_errno(SYNTHETIC_ERRNO(EPERM), "File descriptor for directory volume is not owned by base UID/GID range, refusing.");

                /* Now move the mount into our own UID/GID range */
                _cleanup_free_ char *uid_line = asprintf_safe(
                                UID_FMT " " UID_FMT " " UID_FMT "\n",
-                                p.base_uid, (uid_t) 0, (uid_t) 0x10000);
+                                reply.base_uid, (uid_t) 0, USERNS_RANGE_SIZE);
                _cleanup_free_ char *gid_line = asprintf_safe(
                                GID_FMT " " GID_FMT " " GID_FMT "\n",
-                                p.base_gid, (gid_t) 0, (gid_t) 0x10000);
+                                reply.base_gid, (gid_t) 0, USERNS_RANGE_SIZE);
                if (!uid_line || !gid_line)
                        return log_oom();

@@ -750,7 +678,7 @@ static int run_as_mount_helper(int argc, char *argv[]) {
                        return log_error_errno(userns_fd, "Failed to acquire new user namespace: %m");

                _cleanup_close_ int remapped_fd = open_tree_attr_with_fallback(
-                                fd,
+                                reply.fd,
                                /* path= */ NULL,
                                OPEN_TREE_CLONE | OPEN_TREE_CLOEXEC,
                                &(struct mount_attr) {
@@ -760,25 +688,25 @@ static int run_as_mount_helper(int argc, char *argv[]) {
                if (remapped_fd < 0)
                        return log_error_errno(remapped_fd, "Failed to set ID mapping on returned mount: %m");

-                close_and_replace(fd, remapped_fd);
+                close_and_replace(reply.fd, remapped_fd);

                if (strv_extend(&cmdline, "--bind") < 0)
                        return log_oom();
        }

-        if (p.read_only > 0)
-                read_only = true;
+        if (reply.read_only > 0)
+                read_only = READ_ONLY_YES;

-        if (!strextend_with_separator(&filtered, ",", read_only > 0 ? "ro" : "rw"))
+        if (!strextend_with_separator(&filtered, ",", read_only == READ_ONLY_YES ? "ro" : "rw"))
                return log_oom();

        if (strv_extend_strv(&cmdline, STRV_MAKE("-o", filtered), /* filter_duplicates= */ false) < 0)
                return log_oom();

-        if (strv_extend_strv(&cmdline, STRV_MAKE(FORMAT_PROC_FD_PATH(fd), path), /* filter_duplicates= */ false) < 0)
+        if (strv_extend_strv(&cmdline, STRV_MAKE(FORMAT_PROC_FD_PATH(reply.fd), path), /* filter_duplicates= */ false) < 0)
                return log_oom();

-        r = fd_cloexec(fd, false);
+        r = fd_cloexec(reply.fd, false);
        if (r < 0)
                return log_error_errno(r, "Failed to disable O_CLOEXEC for mount fd: %m");

--- a/src/test/meson.build
+++ b/src/test/meson.build
@@ -141,6 +141,7 @@ simple_tests += files(
        'test-log.c',
        'test-logarithm.c',
        'test-login-util.c',
+        'test-machine-util.c',
        'test-macro.c',
        'test-memfd-util.c',
        'test-memory-util.c',
--- a/src/test/test-machine-util.c
+++ b/src/test/test-machine-util.c
@@ -0,0 +1,144 @@
+/* SPDX-License-Identifier: LGPL-2.1-or-later */
+
+#include "machine-util.h"
+#include "tests.h"
+
+TEST(bind_volume_parse_minimal) {
+        _cleanup_(bind_volume_freep) BindVolume *v = NULL;
+
+        ASSERT_OK(bind_volume_parse("block:/dev/sda", &v));
+        ASSERT_STREQ(v->provider, "block");
+        ASSERT_STREQ(v->volume, "/dev/sda");
+        ASSERT_NULL(v->config);
+        ASSERT_NULL(v->template);
+        ASSERT_EQ(v->create_mode, _CREATE_MODE_INVALID);
+        ASSERT_EQ(v->request_as, _VOLUME_TYPE_INVALID);
+        ASSERT_EQ(v->read_only, _READ_ONLY_INVALID);
+        ASSERT_EQ(v->create_size_bytes, UINT64_MAX);
+}
+
+TEST(bind_volume_parse_with_config) {
+        _cleanup_(bind_volume_freep) BindVolume *v = NULL;
+
+        ASSERT_OK(bind_volume_parse("block:/dev/sda:virtio-scsi", &v));
+        ASSERT_STREQ(v->provider, "block");
+        ASSERT_STREQ(v->volume, "/dev/sda");
+        ASSERT_STREQ(v->config, "virtio-scsi");
+}
+
+TEST(bind_volume_parse_empty_config) {
+        _cleanup_(bind_volume_freep) BindVolume *v = NULL;
+
+        ASSERT_OK(bind_volume_parse("fs:vol-1::create=new,size=64M,template=sparse-file", &v));
+        ASSERT_STREQ(v->provider, "fs");
+        ASSERT_STREQ(v->volume, "vol-1");
+        ASSERT_NULL(v->config);
+        ASSERT_EQ(v->create_mode, CREATE_NEW);
+        ASSERT_STREQ(v->template, "sparse-file");
+        ASSERT_EQ(v->create_size_bytes, UINT64_C(64) * 1024 * 1024);
+}
+
+TEST(bind_volume_parse_full) {
+        _cleanup_(bind_volume_freep) BindVolume *v = NULL;
+
+        ASSERT_OK(bind_volume_parse(
+                          "fs:vol-2:nvme:create=any,template=allocated-file,size=128M,ro=auto,request-as=blk",
+                          &v));
+        ASSERT_STREQ(v->provider, "fs");
+        ASSERT_STREQ(v->volume, "vol-2");
+        ASSERT_STREQ(v->config, "nvme");
+        ASSERT_EQ(v->create_mode, CREATE_ANY);
+        ASSERT_STREQ(v->template, "allocated-file");
+        ASSERT_EQ(v->request_as, VOLUME_BLK);
+        ASSERT_EQ(v->create_size_bytes, UINT64_C(128) * 1024 * 1024);
+        ASSERT_EQ(v->read_only, READ_ONLY_AUTO);
+}
+
+TEST(bind_volume_parse_read_only) {
+        _cleanup_(bind_volume_freep) BindVolume *v = NULL;
+
+        ASSERT_OK(bind_volume_parse("block:/dev/sdb:scsi-cd:read-only=yes", &v));
+        ASSERT_EQ(v->read_only, READ_ONLY_YES);
+
+        v = bind_volume_free(v);
+        ASSERT_OK(bind_volume_parse("block:/dev/sdb:scsi-cd:ro=no", &v));
+        ASSERT_EQ(v->read_only, READ_ONLY_NO);
+}
+
+TEST(bind_volume_parse_invalid) {
+        BindVolume *v = NULL;
+
+        /* Missing provider */
+        ASSERT_ERROR(bind_volume_parse(":vol", &v), EINVAL);
+        ASSERT_NULL(v);
+
+        /* Missing volume */
+        ASSERT_ERROR(bind_volume_parse("block:", &v), EINVAL);
+        ASSERT_NULL(v);
+
+        /* Provider with control char */
+        ASSERT_ERROR(bind_volume_parse("bl\x01ock:vol", &v), EINVAL);
+        ASSERT_NULL(v);
+
+        /* Config with control char */
+        ASSERT_ERROR(bind_volume_parse("block:vol:nv\x01me", &v), EINVAL);
+        ASSERT_NULL(v);
+
+        /* Unknown extras key */
+        ASSERT_ERROR(bind_volume_parse("block:vol::bogus=foo", &v), EINVAL);
+        ASSERT_NULL(v);
+
+        /* Bogus create mode */
+        ASSERT_ERROR(bind_volume_parse("block:vol::create=bogus", &v), EINVAL);
+        ASSERT_NULL(v);
+
+        /* Bogus request-as */
+        ASSERT_ERROR(bind_volume_parse("block:vol::request-as=bogus", &v), EINVAL);
+        ASSERT_NULL(v);
+
+        /* Extras entry without '=' */
+        ASSERT_ERROR(bind_volume_parse("block:vol::nokey", &v), EINVAL);
+        ASSERT_NULL(v);
+
+        /* Empty key (=value with no key) */
+        ASSERT_ERROR(bind_volume_parse("block:vol::=value", &v), EINVAL);
+        ASSERT_NULL(v);
+
+        /* Duplicate key */
+        ASSERT_ERROR(bind_volume_parse("block:vol::create=new,create=any", &v), EINVAL);
+        ASSERT_NULL(v);
+
+        /* Aliased duplicate (size / create-size) */
+        ASSERT_ERROR(bind_volume_parse("block:vol::size=64M,create-size=128M", &v), EINVAL);
+        ASSERT_NULL(v);
+
+        /* Zero-byte size */
+        ASSERT_ERROR(bind_volume_parse("block:vol::size=0", &v), EINVAL);
+        ASSERT_NULL(v);
+
+        /* Duplicate read-only with explicit yes/no values */
+        ASSERT_ERROR(bind_volume_parse("block:vol::read-only=yes,read-only=no", &v), EINVAL);
+        ASSERT_NULL(v);
+        ASSERT_ERROR(bind_volume_parse("block:vol::read-only=yes,ro=auto", &v), EINVAL);
+        ASSERT_NULL(v);
+}
+
+TEST(machine_storage_name_split) {
+        _cleanup_free_ char *p = NULL, *v = NULL;
+
+        ASSERT_OK(machine_storage_name_split("block:/dev/sda", &p, &v));
+        ASSERT_STREQ(p, "block");
+        ASSERT_STREQ(v, "/dev/sda");
+
+        /* NULL outputs — validate-only mode */
+        ASSERT_OK(machine_storage_name_split("fs:vol-1", NULL, NULL));
+
+        ASSERT_ERROR(machine_storage_name_split(NULL, NULL, NULL), EINVAL);
+        ASSERT_ERROR(machine_storage_name_split("", NULL, NULL), EINVAL);
+        ASSERT_ERROR(machine_storage_name_split("no-colon", NULL, NULL), EINVAL);
+        ASSERT_ERROR(machine_storage_name_split(":vol", NULL, NULL), EINVAL);
+        ASSERT_ERROR(machine_storage_name_split("block:", NULL, NULL), EINVAL);
+        ASSERT_ERROR(machine_storage_name_split("bl\x01ock:vol", NULL, NULL), EINVAL);
+}
+
+DEFINE_TEST_MAIN(LOG_INFO);
--- a/src/vmspawn/meson.build
+++ b/src/vmspawn/meson.build
@@ -6,6 +6,7 @@ endif

 vmspawn_sources = files(
        'vmspawn.c',
+        'vmspawn-bind-volume.c',
        'vmspawn-qemu-config.c',
        'vmspawn-qmp.c',
        'vmspawn-varlink.c',
--- a/src/vmspawn/vmspawn-bind-volume.c
+++ b/src/vmspawn/vmspawn-bind-volume.c
@@ -0,0 +1,202 @@
+/* SPDX-License-Identifier: LGPL-2.1-or-later */
+
+#include <fcntl.h>
+#include <sys/stat.h>
+
+#include "sd-varlink.h"
+
+#include "alloc-util.h"
+#include "fd-util.h"
+#include "log.h"
+#include "runtime-scope.h"
+#include "stat-util.h"
+#include "storage-util.h"
+#include "string-util.h"
+#include "vmspawn-bind-volume.h"
+#include "vmspawn-qmp.h"
+
+DiskType disk_type_from_bind_volume_config(const char *config) {
+        if (isempty(config))
+                return DISK_TYPE_VIRTIO_BLK;
+        return disk_type_from_string(config);
+}
+
+int vmspawn_bind_volume_acquire(
+                RuntimeScope scope,
+                const BindVolume *v,
+                bool removable,
+                sd_varlink *link,
+                DriveInfo **ret,
+                char **reterr_error_id) {
+
+        _cleanup_(storage_acquire_reply_done) StorageAcquireReply reply = STORAGE_ACQUIRE_REPLY_INIT;
+        _cleanup_(drive_info_unrefp) DriveInfo *d = NULL;
+        _cleanup_free_ char *err = NULL;
+        int r;
+
+        assert(v);
+        assert(ret);
+
+        DiskType dt = disk_type_from_bind_volume_config(v->config);
+        if (dt < 0) {
+                r = dt;
+                goto fail;
+        }
+
+        r = storage_acquire_volume(scope, v, /* allow_interactive_auth= */ false, &err, &reply);
+        if (r < 0)
+                goto fail;
+
+        if (reply.type == VOLUME_DIR) {
+                r = log_debug_errno(SYNTHETIC_ERRNO(EOPNOTSUPP),
+                                    "Directory volumes are not supported for vmspawn block devices.");
+                goto fail;
+        }
+
+        struct stat st;
+        if (fstat(reply.fd, &st) < 0) {
+                r = -errno;
+                goto fail;
+        }
+        r = stat_verify_regular_or_block(&st);
+        if (r < 0)
+                goto fail;
+
+        d = drive_info_new();
+        if (!d) {
+                r = -ENOMEM;
+                goto fail;
+        }
+
+        d->id = strjoin(v->provider, ":", v->volume);
+        d->disk_driver = strdup(ASSERT_PTR(qemu_device_driver_to_string(dt)));
+        d->format = strdup("raw");
+        d->path = strdup(v->volume);
+        if (!d->id || !d->disk_driver || !d->format || !d->path) {
+                r = -ENOMEM;
+                goto fail;
+        }
+
+        d->disk_type = dt;
+        d->fd = TAKE_FD(reply.fd);
+
+        if (reply.type == VOLUME_BLK || S_ISBLK(st.st_mode))
+                d->flags |= QMP_DRIVE_BLOCK_DEVICE;
+        if (reply.read_only > 0 || dt == DISK_TYPE_VIRTIO_SCSI_CDROM)
+                d->flags |= QMP_DRIVE_READ_ONLY;
+        if (removable)
+                d->flags |= QMP_DRIVE_REMOVABLE;
+        d->link = sd_varlink_ref(link);
+
+        *ret = TAKE_PTR(d);
+        return 0;
+
+fail:
+        if (reterr_error_id)
+                *reterr_error_id = TAKE_PTR(err);
+        return r;
+}
+
+/* Takes ownership of fd unconditionally — it is closed on every error path too. */
+int vmspawn_bind_volume_attach_fd(
+                VmspawnQmpBridge *bridge,
+                sd_varlink *link,
+                int fd,
+                const char *name,
+                const char *config) {
+
+        _cleanup_close_ int owned_fd = fd;
+        int r;
+
+        assert(bridge);
+        assert(link);
+        assert(fd >= 0);
+        assert(name);
+
+        DiskType dt = disk_type_from_bind_volume_config(config);
+        if (dt < 0)
+                return dt;
+
+        struct stat st;
+        if (fstat(owned_fd, &st) < 0)
+                return -errno;
+        r = stat_verify_regular_or_block(&st);
+        if (r < 0)
+                return r;
+
+        _cleanup_(drive_info_unrefp) DriveInfo *d = drive_info_new();
+        if (!d)
+                return -ENOMEM;
+
+        d->id = strdup(name);
+        d->disk_driver = strdup(ASSERT_PTR(qemu_device_driver_to_string(dt)));
+        d->format = strdup("raw");
+        d->path = strdup(name);
+        if (!d->id || !d->disk_driver || !d->format || !d->path)
+                return -ENOMEM;
+
+        int oflags = fcntl(owned_fd, F_GETFL);
+        if (oflags < 0)
+                return -errno;
+
+        d->disk_type = dt;
+        d->fd = TAKE_FD(owned_fd);
+        if (S_ISBLK(st.st_mode))
+                d->flags |= QMP_DRIVE_BLOCK_DEVICE;
+        if (dt == DISK_TYPE_VIRTIO_SCSI_CDROM || (oflags & O_ACCMODE_STRICT) == O_RDONLY)
+                d->flags |= QMP_DRIVE_READ_ONLY;
+        d->flags |= QMP_DRIVE_REMOVABLE;
+        d->link = sd_varlink_ref(link);
+
+        return vmspawn_qmp_add_block_device(bridge, TAKE_PTR(d));
+}
+
+void bind_volumes_done(BindVolumes *bv) {
+        assert(bv);
+        FOREACH_ARRAY(v, bv->items, bv->n_items)
+                bind_volume_free(*v);
+        bv->items = mfree(bv->items);
+        bv->n_items = 0;
+}
+
+int vmspawn_bind_volume_prepare_boot(
+                RuntimeScope scope,
+                const BindVolumes *bv,
+                DriveInfos *drives) {
+
+        int r;
+
+        assert(bv);
+        assert(drives);
+
+        if (bv->n_items == 0)
+                return 0;
+
+        if (!GREEDY_REALLOC(drives->drives, drives->n_drives + bv->n_items))
+                return log_oom();
+
+        FOREACH_ARRAY(it, bv->items, bv->n_items) {
+                BindVolume *v = *it;
+                _cleanup_(drive_info_unrefp) DriveInfo *d = NULL;
+                _cleanup_free_ char *error_id = NULL;
+
+                r = vmspawn_bind_volume_acquire(
+                                scope, v,
+                                /* removable= */ false,
+                                /* link= */ NULL,
+                                &d, &error_id);
+                if (r < 0) {
+                        if (error_id)
+                                return log_error_errno(r,
+                                                       "Failed to acquire storage volume '%s:%s' (%s): %m",
+                                                       v->provider, v->volume, error_id);
+                        return log_error_errno(r,
+                                               "Failed to acquire storage volume '%s:%s': %m",
+                                               v->provider, v->volume);
+                }
+
+                drives->drives[drives->n_drives++] = TAKE_PTR(d);
+        }
+
+        return 0;
+}
--- a/src/vmspawn/vmspawn-bind-volume.h
+++ b/src/vmspawn/vmspawn-bind-volume.h
@@ -0,0 +1,39 @@
+/* SPDX-License-Identifier: LGPL-2.1-or-later */
+#pragma once
+
+#include "machine-util.h"
+#include "shared-forward.h"
+#include "vmspawn-qmp.h"
+
+/* Empty/NULL defaults to virtio-blk; otherwise delegates to disk_type_from_string(). */
+DiskType disk_type_from_bind_volume_config(const char *config);
+
+/* Acquires the volume and builds a DriveInfo with id="<provider>:<volume>" (the
+ * bridge-visible name; QMP-side names are still allocated by add_block_device). */
+int vmspawn_bind_volume_acquire(
+                RuntimeScope scope,
+                const BindVolume *v,
+                bool removable,
+                sd_varlink *link,
+                DriveInfo **ret,
+                char **reterr_error_id);
+
+typedef struct BindVolumes {
+        BindVolume **items;
+        size_t n_items;
+} BindVolumes;
+
+void bind_volumes_done(BindVolumes *bv);
+
+int vmspawn_bind_volume_prepare_boot(
+                RuntimeScope scope,
+                const BindVolumes *bv,
+                DriveInfos *drives);
+
+/* Takes ownership of fd unconditionally. */
+int vmspawn_bind_volume_attach_fd(
+                VmspawnQmpBridge *bridge,
+                sd_varlink *link,
+                int fd,
+                const char *name,
+                const char *config);
--- a/src/vmspawn/vmspawn-qmp.c
+++ b/src/vmspawn/vmspawn-qmp.c
@@ -826,7 +826,7 @@ static int on_add_device_add_complete(
                return 0;

        if (d->link) {
-                (void) sd_varlink_replybo(d->link, SD_JSON_BUILD_PAIR_STRING("id", d->id));
+                (void) sd_varlink_reply(d->link, NULL);
                d->link = sd_varlink_unref(d->link);
        }

@@ -882,7 +882,7 @@ static int qmp_setup_scsi_controller(VmspawnQmpBridge *bridge, const char *pcie_
        return 0;
 }

-static int vmspawn_qmp_add_block_device(VmspawnQmpBridge *bridge, DriveInfo *drive) {
+int vmspawn_qmp_add_block_device(VmspawnQmpBridge *bridge, DriveInfo *drive) {
        int r;

        assert(bridge);
@@ -989,7 +989,6 @@ static int qmp_setup_regular_drive(VmspawnQmpBridge *bridge, DriveInfo *drive) {
        assert(bridge);
        assert(drive);
        assert(drive->fd >= 0);
-        assert(!drive->id);

        return vmspawn_qmp_add_block_device(bridge, drive);
 }
@@ -1028,7 +1027,9 @@ int vmspawn_qmp_remove_block_device(VmspawnQmpBridge *bridge, sd_varlink *link,

        DriveInfo *drive = hashmap_get(bridge->block_devices, id);
        if (!drive)
-                return reply_qmp_error(link, "Unknown block device id", -ENOENT);
+                return sd_varlink_error(link, "io.systemd.MachineInstance.NoSuchStorage", NULL);
+        if (!FLAGS_SET(drive->flags, QMP_DRIVE_REMOVABLE))
+                return sd_varlink_error(link, "io.systemd.MachineInstance.StorageImmutable", NULL);
        if (!FLAGS_SET(drive->state, BLOCK_DEVICE_STATE_BLOCKDEV_ADDED))
                return reply_qmp_error(link, "Block device add pending", -EBUSY);
        if (FLAGS_SET(drive->state, BLOCK_DEVICE_STATE_REMOVE_PENDING))
--- a/src/vmspawn/vmspawn-qmp.h
+++ b/src/vmspawn/vmspawn-qmp.h
@@ -72,6 +72,7 @@ typedef enum QmpDriveFlags {
        QMP_DRIVE_BOOT             = 1u << 4,
        QMP_DRIVE_IO_URING         = 1u << 5,
        QMP_DRIVE_DISCARD_NO_UNREF = 1u << 6,  /* qcow2 only */
+        QMP_DRIVE_REMOVABLE        = 1u << 7,  /* may be detached at runtime via RemoveStorage */
 } QmpDriveFlags;

 typedef enum BlockDeviceStateFlags {
@@ -177,5 +178,6 @@ int vmspawn_qmp_setup_drives(VmspawnQmpBridge *bridge, DriveInfos *drives);
 int vmspawn_qmp_setup_network(VmspawnQmpBridge *bridge, NetworkInfo *network);
 int vmspawn_qmp_setup_virtiofs(VmspawnQmpBridge *bridge, const VirtiofsInfos *virtiofs);
 int vmspawn_qmp_setup_vsock(VmspawnQmpBridge *bridge, VsockInfo *vsock);
+int vmspawn_qmp_add_block_device(VmspawnQmpBridge *bridge, DriveInfo *drive);
 int vmspawn_qmp_remove_block_device(VmspawnQmpBridge *bridge, sd_varlink *link, const char *id);
 int vmspawn_qmp_dispatch_device_deleted(VmspawnQmpBridge *bridge, sd_json_variant *data);
--- a/src/vmspawn/vmspawn-varlink.c
+++ b/src/vmspawn/vmspawn-varlink.c
@@ -2,6 +2,7 @@

 #include "alloc-util.h"
 #include "errno-util.h"
+#include "fd-util.h"
 #include "hashmap.h"
 #include "log.h"
 #include "path-util.h"
@@ -11,6 +12,7 @@
 #include "varlink-io.systemd.MachineInstance.h"
 #include "varlink-io.systemd.VirtualMachineInstance.h"
 #include "varlink-util.h"
+#include "vmspawn-bind-volume.h"
 #include "vmspawn-qmp.h"
 #include "vmspawn-varlink.h"

@@ -168,6 +170,77 @@ static int vl_method_describe(sd_varlink *link, sd_json_variant *parameters, sd_
        return qmp_execute_varlink_async(ctx, link, "query-status", /* arguments= */ NULL, on_qmp_describe_complete);
 }

+static int vl_method_add_storage(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata) {
+        VmspawnVarlinkContext *ctx = ASSERT_PTR(userdata);
+        int r;
+
+        struct {
+                int fd_index;
+                const char *name;
+                const char *config;
+        } p = {
+                .fd_index = -1,
+        };
+
+        static const sd_json_dispatch_field dispatch_table[] = {
+                { "fileDescriptorIndex", _SD_JSON_VARIANT_TYPE_INVALID, sd_json_dispatch_int,          voffsetof(p, fd_index), SD_JSON_MANDATORY },
+                { "name",                SD_JSON_VARIANT_STRING,        sd_json_dispatch_const_string, voffsetof(p, name),     SD_JSON_MANDATORY },
+                { "config",              SD_JSON_VARIANT_STRING,        sd_json_dispatch_const_string, voffsetof(p, config),   0                 },
+                {}
+        };
+
+        r = sd_varlink_dispatch(link, parameters, dispatch_table, &p);
+        if (r != 0)
+                return r;
+
+        if (machine_storage_name_split(p.name, /* ret_provider= */ NULL, /* ret_volume= */ NULL) < 0)
+                return sd_varlink_error_invalid_parameter_name(link, "name");
+
+        if (disk_type_from_bind_volume_config(p.config) < 0)
+                return sd_varlink_error(link, "io.systemd.MachineInstance.BadConfig", NULL);
+
+        if (p.fd_index < 0)
+                return sd_varlink_error_invalid_parameter_name(link, "fileDescriptorIndex");
+
+        _cleanup_close_ int fd = sd_varlink_take_fd(link, p.fd_index);
+        if (fd < 0)
+                return sd_varlink_error_errno(link, fd);
+
+        r = vmspawn_bind_volume_attach_fd(ctx->bridge, link, TAKE_FD(fd), p.name, p.config);
+        if (r == -EEXIST)
+                return sd_varlink_error(link, "io.systemd.MachineInstance.StorageExists", NULL);
+        if (r == -EOPNOTSUPP)
+                return sd_varlink_error(link, "io.systemd.MachineInstance.ConfigNotSupported", NULL);
+        if (r < 0)
+                return sd_varlink_error_errno(link, r);
+
+        /* Async reply via on_add_device_add_complete. */
+        return 0;
+}
+
+static int vl_method_remove_storage(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata) {
+        VmspawnVarlinkContext *ctx = ASSERT_PTR(userdata);
+        int r;
+
+        struct {
+                const char *name;
+        } p = {};
+
+        static const sd_json_dispatch_field dispatch_table[] = {
+                { "name", SD_JSON_VARIANT_STRING, sd_json_dispatch_const_string, voffsetof(p, name), SD_JSON_MANDATORY },
+                {}
+        };
+
+        r = sd_varlink_dispatch(link, parameters, dispatch_table, &p);
+        if (r != 0)
+                return r;
+
+        if (machine_storage_name_split(p.name, /* ret_provider= */ NULL, /* ret_volume= */ NULL) < 0)
+                return sd_varlink_error_invalid_parameter_name(link, "name");
+
+        return vmspawn_qmp_remove_block_device(ctx->bridge, link, p.name);
+}
+
 static int vl_method_subscribe_events(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata) {
        VmspawnVarlinkContext *ctx = ASSERT_PTR(userdata);
        _cleanup_strv_free_ char **filter = NULL;
@@ -380,9 +453,10 @@ int vmspawn_varlink_setup(
        if (!ctx)
                return log_oom();

-        /* Create varlink server for VM control */
+        /* AddStorage receives an fd from the caller. */
        r = varlink_server_new(&ctx->varlink_server,
-                               SD_VARLINK_SERVER_INHERIT_USERDATA,
+                               SD_VARLINK_SERVER_INHERIT_USERDATA |
+                               SD_VARLINK_SERVER_ALLOW_FD_PASSING_INPUT,
                               ctx);
        if (r < 0)
                return log_error_errno(r, "Failed to create varlink server: %m");
@@ -402,7 +476,9 @@ int vmspawn_varlink_setup(
                        "io.systemd.MachineInstance.Resume",            vl_method_resume,
                        "io.systemd.MachineInstance.Reboot",            vl_method_reboot,
                        "io.systemd.MachineInstance.Describe",          vl_method_describe,
-                        "io.systemd.MachineInstance.SubscribeEvents",   vl_method_subscribe_events);
+                        "io.systemd.MachineInstance.SubscribeEvents",   vl_method_subscribe_events,
+                        "io.systemd.MachineInstance.AddStorage",        vl_method_add_storage,
+                        "io.systemd.MachineInstance.RemoveStorage",     vl_method_remove_storage);
        if (r < 0)
                return log_error_errno(r, "Failed to bind varlink methods: %m");

--- a/src/vmspawn/vmspawn.c
+++ b/src/vmspawn/vmspawn.c
@@ -88,6 +88,7 @@
 #include "user-record.h"
 #include "user-util.h"
 #include "utf8.h"
+#include "vmspawn-bind-volume.h"
 #include "vmspawn-mount.h"
 #include "vmspawn-qemu-config.h"
 #include "vmspawn-qmp.h"
@@ -163,6 +164,7 @@ static bool arg_keep_unit = false;
 static sd_id128_t arg_uuid = {};
 static char **arg_kernel_cmdline_extra = NULL;
 static ExtraDriveContext arg_extra_drives = {};
+static BindVolumes arg_bind_volumes = {};
 static char *arg_background = NULL;
 static bool arg_pass_ssh_key = true;
 static char *arg_ssh_key_type = NULL;
@@ -200,6 +202,7 @@ STATIC_DESTRUCTOR_REGISTER(arg_runtime_mounts, runtime_mount_context_done);
 STATIC_DESTRUCTOR_REGISTER(arg_forward_journal, freep);
 STATIC_DESTRUCTOR_REGISTER(arg_kernel_cmdline_extra, strv_freep);
 STATIC_DESTRUCTOR_REGISTER(arg_extra_drives, extra_drive_context_done);
+STATIC_DESTRUCTOR_REGISTER(arg_bind_volumes, bind_volumes_done);
 STATIC_DESTRUCTOR_REGISTER(arg_background, freep);
 STATIC_DESTRUCTOR_REGISTER(arg_ssh_key_type, freep);
 STATIC_DESTRUCTOR_REGISTER(arg_smbios11, strv_freep);
@@ -766,6 +769,30 @@ static int parse_argv(int argc, char *argv[]) {
                        break;
                }

+                OPTION_LONG("bind-volume", "PROVIDER:VOLUME[:CONFIG][:KEY=VALUE,...]",
+                            "Acquire a storage volume from a StorageProvider and attach it to the VM"): {
+                        _cleanup_(bind_volume_freep) BindVolume *bv = NULL;
+
+                        r = bind_volume_parse(opts.arg, &bv);
+                        if (r < 0)
+                                return log_error_errno(r, "Failed to parse --bind-volume= argument '%s': %m", opts.arg);
+
+                        if (disk_type_from_bind_volume_config(bv->config) < 0) {
+                                _cleanup_free_ char *valid = NULL;
+                                for (DiskType t = 0; t < _DISK_TYPE_MAX; t++)
+                                        if (!strextend_with_separator(&valid, ", ", disk_type_to_string(t)))
+                                                return log_oom();
+                                return log_error_errno(SYNTHETIC_ERRNO(EINVAL),
+                                                       "Unknown device type '%s' for --bind-volume=. Valid values: %s.",
+                                                       bv->config, valid);
+                        }
+
+                        if (!GREEDY_REALLOC(arg_bind_volumes.items, arg_bind_volumes.n_items + 1))
+                                return log_oom();
+                        arg_bind_volumes.items[arg_bind_volumes.n_items++] = TAKE_PTR(bv);
+                        break;
+                }
+
                OPTION_LONG("bind-user", "NAME", "Bind user from host to virtual machine"):
                        if (!valid_user_group_name(opts.arg, /* flags= */ 0))
                                return log_error_errno(SYNTHETIC_ERRNO(EINVAL), "Invalid user name to bind: %s", opts.arg);
@@ -2475,7 +2502,7 @@ static int prepare_device_info(const char *runtime_dir, MachineConfig *c) {

        /* Build drive info for QMP-based setup. vmspawn opens all image files and
         * passes fds to QEMU via add-fd — QEMU never needs filesystem access. */
-        drives->drives = new0(DriveInfo*, 1 + arg_extra_drives.n_drives);
+        drives->drives = new0(DriveInfo*, 1 + arg_extra_drives.n_drives + arg_bind_volumes.n_items);
        if (!drives->drives)
                return log_oom();

@@ -2487,6 +2514,10 @@ static int prepare_device_info(const char *runtime_dir, MachineConfig *c) {
        if (r < 0)
                return r;

+        r = vmspawn_bind_volume_prepare_boot(arg_runtime_scope, &arg_bind_volumes, drives);
+        if (r < 0)
+                return r;
+
        return assign_pcie_ports(c);
 }

--- a/test/units/TEST-87-AUX-UTILS-VM.bind-volume.sh
+++ b/test/units/TEST-87-AUX-UTILS-VM.bind-volume.sh
@@ -0,0 +1,166 @@
+#!/usr/bin/env bash
+# SPDX-License-Identifier: LGPL-2.1-or-later
+# Test --bind-volume / machinectl bind-volume / unbind-volume integration with the
+# StorageProvider Varlink interface.
+#
+# Exercises:
+#  - --bind-volume parser + runtime_directory_generic + Acquire round-trip
+#  - boot-time attach via DriveInfo (non-removable)
+#  - runtime hotplug via io.systemd.MachineInstance.AddStorage (removable)
+#  - runtime hot-remove via io.systemd.MachineInstance.RemoveStorage
+#  - StorageImmutable rejection for boot-time attached volumes
+set -eux
+set -o pipefail
+
+# shellcheck source=test/units/util.sh
+. "$(dirname "$0")"/util.sh
+
+if [[ -v ASAN_OPTIONS ]]; then
+    echo "vmspawn launches QEMU which doesn't work under ASan, skipping"
+    exit 0
+fi
+
+if ! command -v systemd-vmspawn >/dev/null 2>&1; then
+    echo "systemd-vmspawn not found, skipping"
+    exit 0
+fi
+
+if ! command -v storagectl >/dev/null 2>&1; then
+    echo "storagectl not found, skipping"
+    exit 0
+fi
+
+if ! find_qemu_binary; then
+    echo "QEMU not found, skipping"
+    exit 0
+fi
+
+if ! command -v mke2fs >/dev/null 2>&1; then
+    echo "mke2fs not found, skipping"
+    exit 0
+fi
+
+# Storage providers are socket-activated; skip if the fs provider socket isn't present.
+if ! test -S /run/systemd/io.systemd.StorageProvider/fs; then
+    echo "StorageProvider fs socket not found, skipping"
+    exit 0
+fi
+
+# Find a kernel for direct boot
+KERNEL=""
+for k in /usr/lib/modules/"$(uname -r)"/vmlinuz /boot/vmlinuz-"$(uname -r)" /boot/vmlinuz; do
+    if [[ -f "$k" ]]; then
+        KERNEL="$k"
+        break
+    fi
+done
+
+if [[ -z "$KERNEL" ]]; then
+    echo "No kernel found for direct VM boot, skipping"
+    exit 0
+fi
+
+WORKDIR="$(mktemp -d /tmp/test-bind-volume.XXXXXXXXXX)"
+
+at_exit() {
+    set +e
+    if [[ -n "${MACHINE:-}" ]]; then
+        if machinectl status "$MACHINE" &>/dev/null; then
+            machinectl terminate "$MACHINE" 2>/dev/null
+            timeout 10 bash -c "while machinectl status '$MACHINE' &>/dev/null; do sleep .5; done" 2>/dev/null
+        fi
+    fi
+    [[ -n "${VMSPAWN_PID:-}" ]] && { kill "$VMSPAWN_PID" 2>/dev/null; wait "$VMSPAWN_PID" 2>/dev/null; }
+    rm -rf "$WORKDIR"
+    rm -f /var/lib/storage/test-bind-volume-*.volume
+}
+trap at_exit EXIT
+
+# Build a minimal root for direct boot — guest just sleeps.
+mkdir -p "$WORKDIR/rootfs/sbin"
+cat >"$WORKDIR/rootfs/sbin/init" <<'INITEOF'
+#!/bin/sh
+exec sleep infinity
+INITEOF
+chmod +x "$WORKDIR/rootfs/sbin/init"
+
+truncate -s 256M "$WORKDIR/root.raw"
+mke2fs -t ext4 -q -d "$WORKDIR/rootfs" "$WORKDIR/root.raw"
+
+BOOT_VOL="test-bind-volume-boot-$$"
+RUNTIME_VOL="test-bind-volume-runtime-$$"
+
+wait_for_machine() {
+    local machine="$1" pid="$2" log="$3"
+    timeout 30 bash -c "
+        while ! machinectl list --no-legend 2>/dev/null | grep >/dev/null '$machine'; do
+            if ! kill -0 $pid 2>/dev/null; then
+                echo 'vmspawn exited before machine registration'
+                cat '$log'
+                exit 77
+            fi
+            sleep .5
+        done
+    " || {
+        local rc=$?
+        if [[ $rc -eq 77 ]]; then exit 0; fi
+        exit "$rc"
+    }
+}
+
+# --- Boot the VM with one boot-time bind-volume ---
+MACHINE="test-bind-volume-$$"
+systemd-vmspawn \
+    --machine="$MACHINE" \
+    --ram=256M \
+    --image="$WORKDIR/root.raw" \
+    --bind-volume="fs:${BOOT_VOL}::create=new,size=64M,template=sparse-file" \
+    --linux="$KERNEL" \
+    --tpm=no \
+    --console=headless \
+    root=/dev/vda rw \
+    &>"$WORKDIR/vmspawn.log" &
+VMSPAWN_PID=$!
+
+wait_for_machine "$MACHINE" "$VMSPAWN_PID" "$WORKDIR/vmspawn.log"
+echo "Machine '$MACHINE' registered"
+
+VARLINK_ADDR=$(varlinkctl call /run/systemd/machine/io.systemd.Machine \
+    io.systemd.Machine.List "{\"name\":\"$MACHINE\"}" | jq -r '.controlAddress')
+assert_neq "$VARLINK_ADDR" "null"
+
+varlinkctl call "$VARLINK_ADDR" io.systemd.MachineInstance.Describe '{}' \
+    | jq -e '.running == true' >/dev/null
+echo "VM running with boot-time bind-volume attached"
+
+# --- Hot-add a second volume via machinectl bind-volume (must succeed) ---
+machinectl bind-volume "$MACHINE" \
+    "fs:${RUNTIME_VOL}:virtio-scsi:create=new,size=32M,template=sparse-file"
+echo "Hot-added runtime bind-volume succeeded"
+
+# --- Hot-remove the runtime-added volume (must succeed) ---
+machinectl unbind-volume "$MACHINE" "fs:${RUNTIME_VOL}"
+echo "Hot-removed runtime bind-volume succeeded"
+
+# --- Removing the boot-time volume must fail with StorageImmutable ---
+if machinectl unbind-volume "$MACHINE" "fs:${BOOT_VOL}" 2>"$WORKDIR/unbind.err"; then
+    echo "ERROR: unbind-volume of boot-time volume should have failed"
+    cat "$WORKDIR/unbind.err"
+    exit 1
+fi
+grep StorageImmutable "$WORKDIR/unbind.err" >/dev/null
+echo "Boot-time bind-volume correctly rejected with StorageImmutable"
+
+# --- Removing a non-existent name must fail with NoSuchStorage ---
+if machinectl unbind-volume "$MACHINE" "fs:no-such-volume-$$" 2>"$WORKDIR/unbind-noexist.err"; then
+    echo "ERROR: unbind-volume of non-existent name should have failed"
+    cat "$WORKDIR/unbind-noexist.err"
+    exit 1
+fi
+grep NoSuchStorage "$WORKDIR/unbind-noexist.err" >/dev/null
+echo "Non-existent unbind-volume correctly rejected with NoSuchStorage"
+
+machinectl terminate "$MACHINE"
+timeout 10 bash -c "while machinectl status '$MACHINE' &>/dev/null; do sleep .5; done"
+timeout 10 bash -c "while kill -0 '$VMSPAWN_PID' 2>/dev/null; do sleep .5; done"
+echo "All bind-volume tests passed"