diff --git a/README.md b/README.md
index 488d973c..77b75619 100644
--- a/README.md
+++ b/README.md
@@ -61,6 +61,57 @@ python3 -m pip install git+https://github.com/zmc/ceph-devstack.git
 
 `ceph-devstack config dump` will output the current configuration.
 
+### Shared block pool (optional)
+
+To back loop devices with slices of a real NVMe device instead of sparse files,
+configure `[block_pool]` in your config. The pool is shared across all
+`testnode_*` containers. Each loop device uses that container's
+`loop_device_size` for its backing region on the pool parent.
+
+```toml
+[block_pool]
+parent = "/dev/nvme0n1p1"   # or a dedicated whole disk
+allow_enroll = true         # first run only; unset afterward
+
+[containers.testnode]
+loop_device_size = "5G"
+```
+
+Requirements:
+
+* Your user must be in the `disk` group and able to read/write the parent
+  device directly (enrollment writes a tail marker to the raw device).
+* On a Fedora CoreOS podman machine (macOS), the `disk` group is defined in
+  `/usr/lib/group`, so `usermod -aG disk core` silently does nothing until the
+  group line is copied into `/etc/group`. `ceph-devstack doctor --fix` does
+  both; by hand, inside `podman machine ssh`:
+
+  ```bash
+  grep -q '^disk:' /etc/group || grep '^disk:' /usr/lib/group | sudo tee -a /etc/group
+  sudo usermod -aG disk core
+  ```
+
+  New `podman machine ssh` sessions then have the group.
+* The parent must be empty, or already enrolled by ceph-devstack (tail marker
+  at the end of the device).
+
+`ceph-devstack block-pool status` shows current allocations.
+
+All pool device access (sysfs, `findmnt`, `blkid`, and the marker and probe
+reads/writes via `dd`) runs on the container host, so a pool inside a podman
+machine (macOS, `RemoteHost`) should work, but that setup is unverified. The
+pool state file and its lock always stay on the local machine.
+
+Each container picks host loop devices at create time: it needs
+`loop_device_count` devices of `loop_device_size`, reuses any existing backing
+files for that container name, then takes the lowest free loop numbers after
+scanning what is already attached or claimed in the `disk_images` directory.
+
+If `block_pool.json` is lost but the on-disk marker remains, start again to
+reclaim the marker; the device is not reformatted. The old allocations are
+forgotten, though, so regions that still hold data are refused as not empty
+until they are wiped.
+
 As an example, the following configuration will use a local image for paddles with the tag `TEST`; it will also create ten testnode containers; and will build its teuthology container from the git repo at `~/src/teuthology`:
 ```
 containers:
diff --git a/ceph_devstack/__init__.py b/ceph_devstack/__init__.py
index d71474cd..71612111 100644
--- a/ceph_devstack/__init__.py
+++ b/ceph_devstack/__init__.py
@@ -15,6 +15,7 @@
 
 PROJECT_ROOT = Path(__file__).parent
 DEFAULT_CONFIG_PATH = Path("~/.config/ceph-devstack/config.toml")
+DEFAULT_DATA_DIR = "~/.local/share/ceph-devstack"
 
 
 def parse_args(args: List[str]) -> argparse.Namespace:
@@ -61,6 +62,11 @@ def parse_args(args: List[str]) -> argparse.Namespace:
         default=False,
         help="Apply suggested fixes for issues found",
     )
+    parser_block_pool = subparsers.add_parser(
+        "block-pool", help="Inspect the shared block device pool"
+    )
+    block_pool_subparsers = parser_block_pool.add_subparsers(dest="block_pool_op")
+    block_pool_subparsers.add_parser("status", help="Show pool allocations and state")
     parser_pull = subparsers.add_parser("pull", help="Pull container images")
     parser_pull.add_argument(
         "image",
diff --git a/ceph_devstack/block_pool.py b/ceph_devstack/block_pool.py
new file mode 100644
index 00000000..f4890dac
--- /dev/null
+++ b/ceph_devstack/block_pool.py
@@ -0,0 +1,474 @@
+import contextlib
+import fcntl
+import functools
+import json
+import os
+import re
+import uuid
+from pathlib import Path
+
+from ceph_devstack import DEFAULT_DATA_DIR, config
+from ceph_devstack.host import host
+
+
+_SIZE_RE = re.compile(r"^(\d+(?:\.\d+)?)([KMGTP]?)$", re.IGNORECASE)
+_UNITS = {
+    "": 1,
+    "K": 1024,
+    "M": 1024**2,
+    "G": 1024**3,
+    "T": 1024**4,
+    "P": 1024**5,
+}
+_SECTOR_SIZE = 512
+_SYSFS_BLOCK = Path("/sys/class/block")
+
+# Written to the last 4 KiB of an enrolled parent so we can recognize our own
+# partitions even if block_pool.json is removed.
+_POOL_TAIL_MAGIC = b"CEPH-DEVSTACK-BLOCK-POOL-TAIL-v1\x00"
+_TAIL_MARKER_SIZE = 4096
+_PROBE_SIZE = 4096
+# BlueStore writes its label at the start of a device and copies of it at
+# these offsets; ceph-volume refuses (silently, for raw prepare) a device
+# that carries any of them. Region probes cover each one inside the region.
+_BLUESTORE_LABEL_OFFSETS = (0, 1024**3, 10 * 1024**3, 100 * 1024**3, 1000 * 1024**3)
+
+# Whole disks and partitions are allowed; device-mapper/loop types are not.
+# NVMe and MMC partitions add "p<N>" to the disk name; SCSI/virtio add "<N>".
+_NVME_DISK = r"nvme\d+n\d+"
+_MMC_DISK = r"mmcblk\d+"
+_SCSI_DISK = r"(?:sd|[xv]d)[a-z]+"
+_WHOLE_DISK_NAME = re.compile(rf"^(?:{_NVME_DISK}|{_MMC_DISK}|{_SCSI_DISK})$")
+_PARTITION_NAME = re.compile(rf"^(?:(?:{_NVME_DISK}|{_MMC_DISK})p|{_SCSI_DISK})\d+$")
+_UNSAFE_NAME = re.compile(r"^(?:dm-\d+|md\d+|loop\d+|ram\d+|fd\d+)$")
+
+
+class BlockPoolError(ValueError):
+    """Raised when a block pool operation would be unsafe."""
+
+
+def parse_size(value: str | int) -> int:
+    text = str(value).strip().upper()
+    match = _SIZE_RE.match(text)
+    if not match:
+        raise BlockPoolError(f"invalid size: {value!r}")
+    number, unit = match.groups()
+    size = int(float(number) * _UNITS[unit.upper()])
+    if size <= 0:
+        raise BlockPoolError(f"size must be positive: {value!r}")
+    return size
+
+
+def format_size(size_bytes: int) -> str:
+    for unit in ("P", "T", "G", "M", "K"):
+        multiplier = _UNITS[unit]
+        if size_bytes >= multiplier and size_bytes % multiplier == 0:
+            return f"{size_bytes // multiplier}{unit}"
+    return f"{size_bytes}B"
+
+
+def validate_parent_name(parent: str) -> str:
+    """Return canonical parent path or raise if the device is not a safe block device."""
+    parent = host.realpath(parent)
+    if not parent.startswith("/dev/"):
+        raise BlockPoolError(f"block pool parent must be a /dev path, not {parent!r}")
+    name = os.path.basename(parent)
+    if _UNSAFE_NAME.match(name):
+        raise BlockPoolError(
+            f"block pool parent {parent!r} is not allowed "
+            "(device-mapper, md, loop, and ram devices are rejected)"
+        )
+    if not (_PARTITION_NAME.match(name) or _WHOLE_DISK_NAME.match(name)):
+        raise BlockPoolError(
+            f"block pool parent {parent!r} must be a block device or partition "
+            "(e.g. /dev/nvme0n1, /dev/nvme0n1p1, or /dev/sda1)"
+        )
+    if not host.path_exists(parent):
+        raise BlockPoolError(f"block pool parent {parent!r} does not exist")
+    if not host.path_exists(_SYSFS_BLOCK / name):
+        raise BlockPoolError(f"block pool parent {parent!r} is not a block device")
+    return parent
+
+
+class BlockPool:
+    """Allocate sized regions from a shared parent block device.
+
+    Device access (sysfs, findmnt, blkid, marker and probe I/O) goes through
+    the Host abstraction; the state file and its lock stay local.
+    """
+
+    @staticmethod
+    def _has_partitions(name: str) -> bool:
+        """True when a whole-disk parent already has partitions.
+
+        The kernel lists a disk's partitions as subdirectories of its sysfs
+        directory, so this needs no name-prefix guessing.
+        """
+        if not _WHOLE_DISK_NAME.match(name):
+            return False
+        return any(_PARTITION_NAME.match(e) for e in host.list_dir(_SYSFS_BLOCK / name))
+
+    @functools.cached_property
+    def _device_size(self) -> int:
+        name = os.path.basename(self.parent)
+        sectors = host.read_text(_SYSFS_BLOCK / name / "size")
+        try:
+            return int(sectors or "") * _SECTOR_SIZE
+        except ValueError:
+            raise BlockPoolError(f"cannot read size of {self.parent!r}") from None
+
+    def _read_parent(self, offset: int, length: int) -> bytes:
+        try:
+            return host.read_bytes(self.parent, offset, length)
+        except OSError as e:
+            raise BlockPoolError(str(e)) from e
+
+    def _write_parent(self, offset: int, data: bytes) -> None:
+        try:
+            host.write_bytes(self.parent, offset, data)
+        except OSError as e:
+            raise BlockPoolError(str(e)) from e
+
+    @staticmethod
+    def _is_region_empty(data: bytes) -> bool:
+        return not data or data == b"\x00" * len(data)
+
+    def _device_has_blkid_signature(self) -> bool:
+        proc = host.output(["blkid", "-p", "-o", "export", self.parent])
+        if proc.returncode != 0:
+            return False
+        for line in proc.stdout.decode().splitlines():
+            key, _, value = line.partition("=")
+            if key in {"TYPE", "PTTYPE"} and value:
+                return True
+        return False
+
+    def _tail_marker_offset(self) -> int:
+        size = self._device_size
+        if size < _TAIL_MARKER_SIZE:
+            raise BlockPoolError(f"block pool parent {self.parent!r} is too small")
+        return size - _TAIL_MARKER_SIZE
+
+    def _read_tail_marker(self) -> dict | None:
+        offset = self._tail_marker_offset()
+        raw = self._read_parent(offset, _TAIL_MARKER_SIZE)
+        if not raw.startswith(_POOL_TAIL_MAGIC):
+            return None
+        payload = raw[len(_POOL_TAIL_MAGIC) :].split(b"\x00", 1)[0]
+        if not payload:
+            return None
+        try:
+            return json.loads(payload.decode())
+        except json.JSONDecodeError:
+            return None
+
+    def _write_tail_marker(self, payload: dict) -> None:
+        offset = self._tail_marker_offset()
+        encoded = json.dumps(payload, sort_keys=True).encode()
+        if len(_POOL_TAIL_MAGIC) + len(encoded) + 1 > _TAIL_MARKER_SIZE:
+            raise BlockPoolError("block pool tail marker payload is too large")
+        raw = _POOL_TAIL_MAGIC + encoded + b"\x00"
+        raw = raw.ljust(_TAIL_MARKER_SIZE, b"\x00")
+        self._write_parent(offset, raw)
+
+    def _probe_parent_empty(self) -> None:
+        size = self._device_size
+        offsets = [0]
+        if size > _SECTOR_SIZE:
+            offsets.append(_SECTOR_SIZE)
+        if size > _PROBE_SIZE:
+            offsets.append(size // 2)
+        tail = self._tail_marker_offset()
+        if tail not in offsets and tail >= _PROBE_SIZE:
+            offsets.append(tail)
+        for offset in offsets:
+            data = self._read_parent(offset, _PROBE_SIZE)
+            if not self._is_region_empty(data):
+                raise BlockPoolError(
+                    f"block pool parent {self.parent!r} is not empty at offset "
+                    f"{offset}; refusing to enroll a partition that may contain data"
+                )
+
+    def __init__(
+        self,
+        state_path: Path,
+        parent: str,
+        *,
+        allow_enroll: bool = False,
+    ):
+        self.state_path = state_path
+        self.parent = validate_parent_name(parent)
+        self.allow_enroll = allow_enroll
+        self._state = self._load()
+
+    @classmethod
+    def from_config(cls, cfg: dict) -> "BlockPool | None":
+        pool_cfg = cfg.get("block_pool") or {}
+        parent = pool_cfg.get("parent")
+        if not parent:
+            return None
+        state_dir = pool_cfg.get("state_dir") or cfg.get("data_dir", DEFAULT_DATA_DIR)
+        state_path = Path(state_dir).expanduser().absolute() / "block_pool.json"
+        allow_enroll = pool_cfg.get("allow_enroll", False) is True
+        return cls(state_path, parent, allow_enroll=allow_enroll)
+
+    @property
+    def enrolled(self) -> bool:
+        return bool(self._state.get("enrolled"))
+
+    @property
+    def pool_id(self) -> str:
+        return self._state["pool_id"]
+
+    @property
+    def allocations(self) -> dict[str, dict]:
+        """Allocated regions keyed by "<owner>:<index>"."""
+        return dict(self._state["allocations"])
+
+    def owner_offsets(self, owner: str) -> set[int]:
+        """Offsets of the regions currently allocated to owner (fresh from disk)."""
+        with self._state_lock():
+            return {
+                region["offset"]
+                for region in self._state["allocations"].values()
+                if region["owner"] == owner
+            }
+
+    @property
+    def free_regions(self) -> list[dict]:
+        return list(self._state["free_regions"])
+
+    @property
+    def next_offset(self) -> int:
+        return self._state["next_offset"]
+
+    def _load(self) -> dict:
+        if self.state_path.exists():
+            state = json.loads(self.state_path.read_text())
+            if state.get("parent") != self.parent:
+                raise BlockPoolError(
+                    f"block pool parent changed ({state.get('parent')!r} -> "
+                    f"{self.parent!r}); remove {self.state_path} to reconfigure"
+                )
+            return state
+        return self._fresh_state()
+
+    def _fresh_state(self) -> dict:
+        return {
+            "parent": self.parent,
+            "pool_id": str(uuid.uuid4()),
+            "enrolled": False,
+            "next_offset": 0,
+            "free_regions": [],
+            "allocations": {},
+        }
+
+    @contextlib.contextmanager
+    def _state_lock(self):
+        self.state_path.parent.mkdir(parents=True, exist_ok=True)
+        lock_path = self.state_path.with_suffix(".lock")
+        with open(lock_path, "w") as lock_file:
+            fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX)
+            try:
+                if self.state_path.exists():
+                    self._state = self._load()
+                yield
+            finally:
+                fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN)
+
+    def _atomic_save(self) -> None:
+        tmp_path = self.state_path.with_suffix(".tmp")
+        tmp_path.write_text(json.dumps(self._state, indent=2, sort_keys=True) + "\n")
+        os.replace(tmp_path, self.state_path)
+
+    def ensure_ready(self) -> None:
+        """Verify the parent partition is safe to use before any slice I/O."""
+        with self._state_lock():
+            if self._ensure_ready_unlocked():
+                self._atomic_save()
+
+    def _ensure_ready_unlocked(self) -> bool:
+        if self.enrolled:
+            self._verify_enrolled_parent()
+            return False
+        self._enroll_new_parent()
+        return True
+
+    def _verify_enrolled_parent(self) -> None:
+        marker = self._read_tail_marker()
+        if marker is None:
+            raise BlockPoolError(
+                f"block pool parent {self.parent!r} is missing the devstack "
+                f"enrollment marker; refusing to use it. If {self.state_path} "
+                "was removed accidentally, start again to reclaim the marker. "
+                "If you wiped the device, set block_pool.allow_enroll = true "
+                "for a fresh enrollment."
+            )
+        if marker.get("pool_id") != self._state.get("pool_id"):
+            raise BlockPoolError(
+                f"block pool parent {self.parent!r} belongs to another pool "
+                f"({marker.get('pool_id')!r}); remove {self.state_path} only "
+                "if you intentionally reset the pool"
+            )
+        if marker.get("parent") != self.parent:
+            raise BlockPoolError(
+                f"block pool parent path mismatch for enrolled partition "
+                f"({marker.get('parent')!r} != {self.parent!r})"
+            )
+
+    def _enroll_new_parent(self) -> None:
+        marker = self._read_tail_marker()
+        if marker is not None:
+            if marker.get("parent") != self.parent:
+                raise BlockPoolError(
+                    f"block pool parent {self.parent!r} has a devstack marker "
+                    f"for a different path ({marker.get('parent')!r})"
+                )
+            self._state["pool_id"] = marker.get("pool_id")
+            self._state["enrolled"] = True
+            return
+
+        if not self.allow_enroll:
+            raise BlockPoolError(
+                f"block pool parent {self.parent!r} is not enrolled. "
+                "Use a dedicated empty partition and set "
+                "block_pool.allow_enroll = true for the first run only."
+            )
+        if host.device_mounted(self.parent):
+            raise BlockPoolError(
+                f"block pool parent {self.parent!r} is mounted; refusing to enroll"
+            )
+        if self._device_has_blkid_signature():
+            raise BlockPoolError(
+                f"block pool parent {self.parent!r} has a filesystem or partition "
+                "table signature; refusing to enroll"
+            )
+
+        if self._has_partitions(os.path.basename(self.parent)):
+            raise BlockPoolError(
+                f"block pool parent {self.parent!r} is a whole disk with "
+                "existing partitions; use a dedicated empty disk or a single "
+                "partition instead"
+            )
+        self._probe_parent_empty()
+
+        payload = {
+            "pool_id": self._state["pool_id"],
+            "parent": self.parent,
+        }
+        self._write_tail_marker(payload)
+        self._state["enrolled"] = True
+
+    def _validate_region(self, offset: int, size: int, owner: str) -> None:
+        parent_size = self._device_size
+        if offset + size > parent_size:
+            raise BlockPoolError(
+                f"block pool region for {owner!r} at offset {offset} with size "
+                f"{size} exceeds parent size {parent_size} for {self.parent!r}"
+            )
+        if offset >= self._tail_marker_offset():
+            raise BlockPoolError(
+                f"block pool region for {owner!r} would overlap the enrollment "
+                f"marker on {self.parent!r}"
+            )
+        probe = min(_PROBE_SIZE, size)
+        for probe_offset in self._region_probe_offsets(offset, size):
+            data = self._read_parent(probe_offset, probe)
+            if not self._is_region_empty(data):
+                raise BlockPoolError(
+                    f"block pool region for {owner!r} on {self.parent!r} is not "
+                    f"empty at offset {probe_offset}; refusing to allocate over "
+                    "existing data"
+                )
+
+    @staticmethod
+    def _region_probe_offsets(offset: int, size: int) -> list[int]:
+        """Parent offsets of the region's probe areas (see _BLUESTORE_LABEL_OFFSETS)."""
+        probe = min(_PROBE_SIZE, size)
+        return [
+            offset + label
+            for label in _BLUESTORE_LABEL_OFFSETS
+            if label + probe <= size
+        ]
+
+    def _pop_free_region(self, size: int) -> dict | None:
+        free_regions = self._state["free_regions"]
+        for index, region in enumerate(free_regions):
+            if region["size"] == size:
+                return free_regions.pop(index)
+        return None
+
+    def get_or_allocate(self, owner: str, index: int, size: int) -> int:
+        """Offset of the region for owner's index-th device, allocating if new."""
+        with self._state_lock():
+            changed = self._ensure_ready_unlocked()
+            key = f"{owner}:{index}"
+            allocations = self._state["allocations"]
+            if key in allocations:
+                if changed:
+                    self._atomic_save()
+                return allocations[key]["offset"]
+
+            if region := self._pop_free_region(size):
+                offset = region["offset"]
+            else:
+                offset = self._state["next_offset"]
+                self._state["next_offset"] = offset + size
+
+            self._validate_region(offset, size, owner)
+            allocations[key] = {"offset": offset, "size": size, "owner": owner}
+            self._atomic_save()
+            return offset
+
+    def release_owner(self, owner: str) -> list[str]:
+        """Return owner's regions to the free list; list the ones kept back.
+
+        Each region's probe areas are zeroed first, so that _validate_region
+        accepts it when it is handed out again, and no BlueStore label copy
+        left by its owner's OSD survives for ceph-volume to trip over;
+        anything else found there later is foreign data and still refused. A region that cannot be
+        zeroed stays allocated to owner (releasing again retries it), and
+        its "<owner>:<index>" key is returned. Nothing is written, and every
+        region is kept, unless the parent still carries this pool's
+        enrollment marker: the device name may now point at another disk.
+        """
+        kept = []
+        with self._state_lock():
+            prefix = f"{owner}:"
+            allocations = self._state["allocations"]
+            free_regions = self._state["free_regions"]
+            keys = [key for key in allocations if key.startswith(prefix)]
+            if not self.enrolled:
+                return keys
+            try:
+                self._verify_enrolled_parent()
+            except BlockPoolError:
+                return keys
+            for key in keys:
+                region = allocations[key]
+                probe = min(_PROBE_SIZE, region["size"])
+                try:
+                    for probe_offset in self._region_probe_offsets(
+                        region["offset"], region["size"]
+                    ):
+                        self._write_parent(probe_offset, b"\x00" * probe)
+                except BlockPoolError:
+                    kept.append(key)
+                    continue
+                del allocations[key]
+                free_regions.append(
+                    {"offset": region["offset"], "size": region["size"]}
+                )
+            free_regions.sort(key=lambda region: region["offset"])
+            self._atomic_save()
+        return kept
+
+
+@functools.cache
+def shared_pool() -> BlockPool | None:
+    """The block pool for the loaded config, built once per process and shared.
+
+    State is re-read under the lock for every operation, so one instance can
+    serve every requirement check and provisioner in a run.
+    """
+    return BlockPool.from_config(config)
diff --git a/ceph_devstack/cli.py b/ceph_devstack/cli.py
index 8658bcbc..760b5490 100644
--- a/ceph_devstack/cli.py
+++ b/ceph_devstack/cli.py
@@ -5,6 +5,7 @@
 from pathlib import Path
 
 from ceph_devstack import config, logger, parse_args, VERBOSE
+from ceph_devstack.block_pool import BlockPoolError, format_size, shared_pool
 from ceph_devstack.requirements import check_requirements
 from ceph_devstack.resources.ceph import CephDevStack
 
@@ -32,16 +33,52 @@
 }
 
 
+def _configure_console_logging(verbose: bool) -> None:
+    if not verbose:
+        return
+    for handler in logging.getLogger("root").handlers:
+        if not isinstance(handler, logging.FileHandler):
+            handler.setLevel(VERBOSE)
+
+
+def block_pool_status() -> int:
+    try:
+        pool = shared_pool()
+    except BlockPoolError as e:
+        logger.error(f"block pool: {e}")
+        return 1
+    if pool is None:
+        logger.info("block pool: disabled (no parent configured)")
+        return 0
+    logger.info(f"block pool parent: {pool.parent}")
+    logger.info(f"enrolled: {pool.enrolled}")
+    logger.info(f"pool id: {pool.pool_id}")
+    logger.info(f"state file: {pool.state_path}")
+    logger.info("allocations:" if pool.allocations else "allocations: (none)")
+    for key, region in sorted(pool.allocations.items()):
+        logger.info(
+            f"  {key}: offset {region['offset']}, "
+            f"size {format_size(region['size'])}, last owner {region['owner']}"
+        )
+    logger.info("free regions:" if pool.free_regions else "free regions: (none)")
+    for region in pool.free_regions:
+        logger.info(f"  offset {region['offset']}, size {format_size(region['size'])}")
+    logger.info(f"next offset: {pool.next_offset}")
+    return 0
+
+
 def main() -> int:
     args = parse_args(sys.argv[1:])
     config.load(args.config_file)
-    if args.verbose:
-        for handler in logging.getLogger("root").handlers:
-            if not isinstance(handler, logging.FileHandler):
-                handler.setLevel(VERBOSE)
+    _configure_console_logging(args.verbose)
     if args.command == "config":
         CONFIG_HANDLERS[args.config_op](config, args)
         return 0
+    if args.command == "block-pool":
+        if args.block_pool_op != "status":
+            logger.error("Usage: ceph-devstack block-pool status")
+            return 2
+        return block_pool_status()
     config["args"] = vars(args)
     data_path = Path(config["data_dir"]).expanduser()
     data_path.mkdir(parents=True, exist_ok=True)
diff --git a/ceph_devstack/config.toml b/ceph_devstack/config.toml
index 147a729e..41f24b1d 100644
--- a/ceph_devstack/config.toml
+++ b/ceph_devstack/config.toml
@@ -1,5 +1,19 @@
 data_dir = "~/.local/share/ceph-devstack"
 
+[block_pool]
+# Dedicated empty block device or partition. Whole disks are allowed when
+# empty; disks with existing partitions must use a partition path instead.
+# Enrollment reads/writes the parent directly: your user must be able to
+# open the device for read/write (typically membership in the disk group).
+# First run requires allow_enroll = true; the tool writes a tail marker and
+# refuses non-empty or foreign devices. If block_pool.json is lost but the
+# marker remains, start again to reclaim the marker; the old allocations are
+# forgotten, so regions that still hold data are refused until wiped.
+# parent = "/dev/nvme0n1"
+# parent = "/dev/nvme0n1p1"
+# allow_enroll = true
+# state_dir = "/path"  # where block_pool.json lives; defaults to data_dir
+
 [containers.archive]
 image = "python:alpine"
 
@@ -17,6 +31,7 @@ image = "quay.io/ceph-infra/pulpito:main"
 
 [containers.testnode]
 count = 3
+loop_device_count = 1
 loop_device_size = "5G"
 image = "quay.io/ceph-infra/teuthology-testnode:main"
 
diff --git a/ceph_devstack/exec.py b/ceph_devstack/exec.py
index 1bddb991..fded6a50 100644
--- a/ceph_devstack/exec.py
+++ b/ceph_devstack/exec.py
@@ -124,6 +124,13 @@ def run(self) -> subprocess.Popen:
         proc.wait()
         return proc
 
+    def capture(self, input: Optional[bytes] = None) -> subprocess.CompletedProcess:
+        """Run to completion, feeding input on stdin and capturing output."""
+        logger.log(VERBOSE, self._make_log_msg())
+        return subprocess.run(
+            self.args, env=self.env, input=input, capture_output=True, check=False
+        )
+
     async def arun(self) -> Subprocess:
         logger.log(VERBOSE, self._make_log_msg())
         loop = asyncio.get_running_loop()
diff --git a/ceph_devstack/host.py b/ceph_devstack/host.py
index 8c744acc..8dae701b 100644
--- a/ceph_devstack/host.py
+++ b/ceph_devstack/host.py
@@ -1,14 +1,17 @@
+import functools
 import json
 import logging
+import math
 import os
 import pathlib
 import shlex
 import socket
+import subprocess
 import sys
 import yaml
 
 from packaging.version import parse as parse_version, Version
-from typing import Dict, List, Optional, Union
+from typing import Dict, List, Optional, Tuple, Union
 
 from .exec import Command, Subprocess
 
@@ -56,6 +59,57 @@ def path_exists(self, path: Union[str, pathlib.Path]):
             path = pathlib.Path(path)
         return path.expanduser().exists()
 
+    def output(
+        self, args: List[str], input: Optional[bytes] = None
+    ) -> subprocess.CompletedProcess:
+        """Run args to completion on this host; capture stdout/stderr as bytes.
+
+        This blocks the caller (and any running event loop) until the command
+        exits; it is meant for short queries and small device reads/writes.
+        """
+        return self.cmd(args).capture(input)
+
+    def read_text(self, path: Union[str, pathlib.Path]) -> Optional[str]:
+        """Contents of a file on this host, or None if it cannot be read."""
+        proc = self.output(["cat", str(path)])
+        return proc.stdout.decode() if proc.returncode == 0 else None
+
+    def list_dir(self, path: Union[str, pathlib.Path]) -> List[str]:
+        """Entry names in a directory on this host; empty if it is unreadable."""
+        proc = self.output(["ls", "-1", str(path)])
+        return proc.stdout.decode().splitlines() if proc.returncode == 0 else []
+
+    def realpath(self, path: Union[str, pathlib.Path]) -> str:
+        """Canonical path on this host; the input unchanged if it can't resolve."""
+        proc = self.output(["realpath", str(path)])
+        return proc.stdout.decode().strip() if proc.returncode == 0 else str(path)
+
+    def device_mounted(self, device: str) -> bool:
+        """True if a filesystem on this host is mounted from device."""
+        proc = self.output(["findmnt", "-n", "-o", "TARGET", "--source", device])
+        return proc.returncode == 0 and bool(proc.stdout.strip())
+
+    def read_bytes(
+        self, path: Union[str, pathlib.Path], offset: int, length: int
+    ) -> bytes:
+        """Read length (> 0) bytes at offset from a file or device on this host."""
+        bs = math.gcd(offset, length)
+        args = ["dd", f"if={path}", f"bs={bs}", f"skip={offset // bs}"]
+        proc = self.output(args + [f"count={length // bs}"])
+        if proc.returncode != 0:
+            raise OSError(f"cannot read {path}: {proc.stderr.decode().strip()}")
+        return proc.stdout
+
+    def write_bytes(
+        self, path: Union[str, pathlib.Path], offset: int, data: bytes
+    ) -> None:
+        """Write data (non-empty) at offset into a file or device on this host."""
+        bs = math.gcd(offset, len(data))
+        args = ["dd", f"of={path}", f"bs={bs}", f"seek={offset // bs}"]
+        proc = self.output(args + ["conv=notrunc,fsync"], input=data)
+        if proc.returncode != 0:
+            raise OSError(f"cannot write {path}: {proc.stderr.decode().strip()}")
+
     def hostname(self) -> str:
         name = socket.getfqdn()
         try:
@@ -138,6 +192,10 @@ async def apparmor_enabled(self) -> bool:
             return False
         return await proc.wait() == 0
 
+    def user_ids(self) -> Tuple[int, int]:
+        """Return (uid, gid) of the user running this host."""
+        return (os.getuid(), os.getgid())
+
 
 class LocalHost(Host):
     pass
@@ -148,9 +206,9 @@ class RemoteHost(Host):
     base_args = ["podman", "machine", "ssh"]
 
     def _remote_args(self, args: List[str], stream_output: bool) -> List[str]:
+        # Quote each arg so compound arguments like "bash -c '<cmd>'" stay as one shell word
         remote = list(self.base_args)
         remote.append("--")
-        # Quote each arg so compound arguments like "bash -c '<cmd>'" stay as one shell word
         remote.extend(shlex.quote(arg) for arg in args)
         return remote
 
@@ -175,6 +233,21 @@ def hostname(self) -> str:
         assert proc.stdout is not None
         return proc.stdout.read().decode().strip()
 
+    def user_ids(self) -> Tuple[int, int]:
+        """Return (uid, gid) of the user running podman on the container host."""
+        return self._user_ids
+
+    @functools.cached_property
+    def _user_ids(self) -> Tuple[int, int]:
+        return self._id("-u"), self._id("-g")
+
+    def _id(self, flag: str) -> int:
+        out = self.output(["id", flag]).stdout.decode().strip()
+        try:
+            return int(out)
+        except ValueError as e:
+            raise RuntimeError(f"Failed to parse output of 'id {flag}': {out!r}") from e
+
 
 local_host = LocalHost()
 
diff --git a/ceph_devstack/resources/ceph/__init__.py b/ceph_devstack/resources/ceph/__init__.py
index 5864c1b6..bbe10451 100644
--- a/ceph_devstack/resources/ceph/__init__.py
+++ b/ceph_devstack/resources/ceph/__init__.py
@@ -18,10 +18,13 @@
     Teuthology,
     Archive,
 )
+from ceph_devstack.resources.ceph.host_loops import LoopDeviceMixin
 from ceph_devstack.resources.ceph.requirements import (
     HasSudo,
     LoopControlDeviceExists,
     LoopControlDeviceWriteable,
+    BlockPoolDiskGroup,
+    BlockPoolParentAccessible,
     SELinuxModule,
 )
 from ceph_devstack.resources.ceph.utils import get_runs, get_jobs
@@ -128,6 +131,8 @@ async def check_requirements(self):
         result = has_sudo = await HasSudo().evaluate()
         result = result and await LoopControlDeviceExists().evaluate()
         result = result and await LoopControlDeviceWriteable().evaluate()
+        result = result and await BlockPoolDiskGroup().evaluate()
+        result = result and await BlockPoolParentAccessible().evaluate()
 
         # Check for SELinux being enabled and Enforcing; then check for the
         # presence of our module. If necessary, inform the user and instruct
@@ -158,6 +163,12 @@ async def create(self):
         logger.info("Creating containers...")
         await CephDevStackNetwork().create()
         await SSHKeyPair().create()
+        # Pre-allocate loop device numbers sequentially so concurrent creates
+        # don't race on the same device numbers.
+        for spec in self.service_specs.values():
+            for obj in spec["objects"]:
+                if isinstance(obj, LoopDeviceMixin):
+                    obj.allocate_devices()
         containers = []
         for spec in self.service_specs.values():
             for object in spec["objects"]:
diff --git a/ceph_devstack/resources/ceph/block_devices.py b/ceph_devstack/resources/ceph/block_devices.py
new file mode 100644
index 00000000..9cf2e3ef
--- /dev/null
+++ b/ceph_devstack/resources/ceph/block_devices.py
@@ -0,0 +1,254 @@
+import asyncio
+import json
+from pathlib import Path
+from typing import Protocol
+
+from ceph_devstack import logger
+from ceph_devstack.block_pool import BlockPool, BlockPoolError, format_size, parse_size
+from ceph_devstack.exec import Subprocess
+from ceph_devstack.host import host
+
+# After detaching, wait up to _DETACH_CHECKS * _DETACH_RETRY_DELAY seconds for
+# busy loops (e.g. still held by udev) to let go of their pool regions.
+_DETACH_CHECKS = 5
+_DETACH_RETRY_DELAY = 0.5
+
+
+def backing_name(owner: str, loop_number: int) -> str:
+    """Backing file name for a loop device owned by a container."""
+    return f"{owner}-{loop_number}"
+
+
+def parse_backing_name(name: str) -> tuple[str, int] | None:
+    """Parse backing file name into (owner, loop_number) or return None."""
+    owner, sep, loop_id = name.rpartition("-")
+    if not sep or not loop_id.isdigit():
+        return None
+    return owner, int(loop_id)
+
+
+def loop_path(number: int) -> str:
+    """Host /dev/loop<N> path for a loop number."""
+    return f"/dev/loop{number}"
+
+
+def loop_number(path: str) -> int:
+    """Extract loop number from /dev/loop<N> path."""
+    return int(path.removeprefix("/dev/loop"))
+
+
+def attached_loops() -> list[tuple[int, int, str]]:
+    """(loop number, offset, backing file) of every attached loop on the host.
+
+    One ``losetup --list`` call reads all of it from sysfs; no root needed.
+    Raises BlockPoolError if the loops can't be listed: callers decide what is
+    free or safe to release from this, so a failure must never look empty.
+    """
+    proc = host.output(
+        ["losetup", "--list", "--json", "--output", "NAME,OFFSET,BACK-FILE"]
+    )
+    if proc.returncode != 0:
+        raise BlockPoolError(
+            f"cannot list loop devices: {proc.stderr.decode().strip()}"
+        )
+    if not proc.stdout.strip():
+        return []  # losetup prints nothing when no loops are attached
+    try:
+        return [
+            (loop_number(loop["name"]), int(loop["offset"]), loop["back-file"])
+            for loop in json.loads(proc.stdout)["loopdevices"]
+        ]
+    except (ValueError, KeyError, TypeError) as e:
+        raise BlockPoolError(f"cannot parse losetup output: {e}") from e
+
+
+def backing_files(image_dir: Path) -> list[tuple[str, int]]:
+    """(owner, loop number) for each backing file in image_dir on the host."""
+    return [
+        parsed
+        for name in host.list_dir(image_dir)
+        if (parsed := parse_backing_name(name)) is not None
+    ]
+
+
+class CommandRunner(Protocol):
+    async def __call__(
+        self,
+        args: list[str],
+        *,
+        check: bool = False,
+    ) -> Subprocess: ...
+
+
+class BlockDeviceProvisioner:
+    """Create loop devices backed by sparse files or NVMe pool slices.
+
+    All commands, including backing-file management in image_dir, run on the
+    container host: losetup resolves image paths there.
+    """
+
+    def __init__(
+        self,
+        owner: str,
+        *,
+        image_dir: Path,
+        file_size: str,
+        cmd: CommandRunner,
+        trigger_udev: bool = False,
+        pool: BlockPool | None,
+    ):
+        self.owner = owner
+        self.image_dir = Path(image_dir).expanduser().absolute()
+        self.device_size = parse_size(file_size)
+        self.trigger_udev = trigger_udev
+        self._cmd = cmd
+        self.pool = pool
+
+    def _image_path(self, device: str) -> Path:
+        num = loop_number(device)
+        return self.image_dir / backing_name(self.owner, num)
+
+    async def create_devices(self, devices: list[str]) -> None:
+        for index, device in enumerate(devices):
+            await self.create_device(device, index)
+
+    def owned_devices(self) -> list[str]:
+        """Loop devices this owner holds on the host, found from host state.
+
+        Pool mode: loops attached to the pool parent at one of the owner's
+        region offsets. Sparse mode: loops named by the owner's backing files.
+        """
+        if (pool := self.pool) is not None:
+            offsets = pool.owner_offsets(self.owner)
+            numbers = [
+                number
+                for number, offset, backing in attached_loops()
+                if backing == pool.parent and offset in offsets
+            ]
+        else:
+            numbers = [
+                number
+                for owner, number in backing_files(self.image_dir)
+                if owner == self.owner
+            ]
+        return [loop_path(number) for number in sorted(numbers)]
+
+    async def remove_devices(self, devices: list[str]) -> None:
+        """Detach devices; then release the owner's pool regions once detached.
+
+        If loops stay attached (e.g. busy), or a region's start can't be
+        zeroed, the regions stay allocated and a warning says so; removing
+        again later releases them.
+        """
+        for device in devices:
+            await self.remove_device(device)
+        if (pool := self.pool) is None:
+            return
+        await self._cmd(["sudo", "udevadm", "settle"], check=False)
+        for attempt in range(_DETACH_CHECKS):
+            if not (still_attached := self.owned_devices()):
+                if kept := pool.release_owner(self.owner):
+                    logger.warning(
+                        f"{self.owner}: could not wipe pool regions "
+                        f"{', '.join(kept)} on {pool.parent}; keeping them "
+                        "allocated until a later remove succeeds"
+                    )
+                return
+            if attempt < _DETACH_CHECKS - 1:
+                await asyncio.sleep(_DETACH_RETRY_DELAY)
+        logger.warning(
+            f"{self.owner}: {', '.join(still_attached)} still attached to "
+            f"{pool.parent}; keeping its pool regions allocated until a later "
+            "remove succeeds"
+        )
+
+    async def create_device(self, device: str, index: int) -> None:
+        await self._ensure_loop_module()
+        await self.remove_device(device)
+        try:
+            await self._make_device(device, index)
+        except Exception:
+            # Removal finds devices from attached loops (pool) or backing
+            # files (sparse), so a node left behind here would never be found.
+            try:
+                await self.remove_device(device)
+            except Exception:
+                logger.exception(f"{self.owner}: could not clean up {device}")
+            raise
+
+    async def _make_device(self, device: str, index: int) -> None:
+        """Create the device node and attach it to its pool region or file."""
+        device_pos = str(loop_number(device))
+        await self._cmd(
+            ["sudo", "mknod", "-m700", device, "b", "7", device_pos],
+            check=True,
+        )
+        # Use host.user_ids() to get the container host's uid:gid (not the local process uid:gid)
+        uid, gid = host.user_ids()
+        await self._cmd(
+            ["sudo", "chown", f"{uid}:{gid}", device],
+            check=True,
+        )
+        if (pool := self.pool) is not None:
+            offset = pool.get_or_allocate(self.owner, index, self.device_size)
+            logger.info(
+                f"{self.owner}: region on {pool.parent} "
+                f"(offset={offset}, size={format_size(self.device_size)}) -> {device}"
+            )
+            await self._cmd(
+                [
+                    "sudo",
+                    "losetup",
+                    "--offset",
+                    str(offset),
+                    "--sizelimit",
+                    str(self.device_size),
+                    device,
+                    pool.parent,
+                ],
+                check=True,
+            )
+        else:
+            image_path = self._image_path(device)
+            await self._cmd(["mkdir", "-p", str(self.image_dir)], check=True)
+            await self._cmd(
+                [
+                    "sudo",
+                    "dd",
+                    "if=/dev/null",
+                    f"of={image_path}",
+                    "bs=1",
+                    "count=0",
+                    f"seek={self.device_size}",
+                ],
+                check=True,
+            )
+            await self._cmd(["sudo", "losetup", device, str(image_path)], check=True)
+
+        await self._cmd(["chcon", "-t", "fixed_disk_device_t", device], check=False)
+        if self.trigger_udev:
+            await self._cmd(
+                [
+                    "sudo",
+                    "udevadm",
+                    "trigger",
+                    "--action=add",
+                    f"--name=block/loop{device_pos}",
+                ],
+                check=False,
+            )
+            await self._cmd(["sudo", "udevadm", "settle"], check=False)
+
+    async def remove_device(self, device: str) -> None:
+        if host.device_mounted(device):
+            await self._cmd(["umount", device], check=True)
+        if host.path_exists(device):
+            await self._cmd(["sudo", "losetup", "-d", device], check=False)
+            await self._cmd(["sudo", "rm", "-f", device], check=False)
+        if self.pool is None:
+            await self._cmd(["rm", "-f", str(self._image_path(device))], check=False)
+
+    async def _ensure_loop_module(self) -> None:
+        proc = await self._cmd(["bash", "-c", "lsmod | grep -q loop"], check=False)
+        if await proc.wait() != 0:
+            await self._cmd(["sudo", "modprobe", "loop"])
diff --git a/ceph_devstack/resources/ceph/containers.py b/ceph_devstack/resources/ceph/containers.py
index 1d3807c7..6a5f3ac3 100644
--- a/ceph_devstack/resources/ceph/containers.py
+++ b/ceph_devstack/resources/ceph/containers.py
@@ -6,6 +6,7 @@
 
 from ceph_devstack import config, DEFAULT_CONFIG_PATH, logger
 from ceph_devstack.host import host
+from ceph_devstack.resources.ceph.host_loops import LoopDeviceMixin
 from ceph_devstack.resources.container import Container
 
 
@@ -154,7 +155,7 @@ class Pulpito(Container):
     }
 
 
-class TestNode(Container):
+class TestNode(LoopDeviceMixin, Container):
     _image_name = "teuthology-testnode"
     capabilities = [
         "SYS_ADMIN",
@@ -178,13 +179,7 @@ class TestNode(Container):
 
     def __init__(self, name: str = ""):
         super().__init__(name=name)
-        self.index = 0
-        if "_" in self.name:
-            self.index = int(self.name.split("_")[-1])
-        self.loop_device_count = config["containers"]["testnode"].get(
-            "loop_device_count", 1
-        )
-        self.devices = [self.device_name(i) for i in range(self.loop_device_count)]
+        self._init_loop_devices()
 
     @property
     def loop_img_dir(self):
@@ -260,70 +255,6 @@ async def remove(self):
         await super().remove()
         await self.remove_loop_devices()
 
-    async def create_loop_devices(self):
-        for device in self.devices:
-            await self.create_loop_device(device)
-
-    async def remove_loop_devices(self):
-        for device in self.devices:
-            await self.remove_loop_device(device)
-
-    async def create_loop_device(self, device: str):
-        size = config["containers"]["testnode"]["loop_device_size"]
-        os.makedirs(self.loop_img_dir, exist_ok=True)
-        proc = await self.cmd(["lsmod", "|", "grep", "loop"])
-        if proc and await proc.wait() != 0:
-            await self.cmd(["sudo", "modprobe", "loop"])
-        loop_img_name = os.path.join(self.loop_img_dir, self.device_image(device))
-        await self.remove_loop_device(device)
-        device_pos = device.removeprefix("/dev/loop")
-        await self.cmd(
-            [
-                "sudo",
-                "mknod",
-                "-m700",
-                device,
-                "b",
-                "7",
-                device_pos,
-            ],
-            check=True,
-        )
-        await self.cmd(
-            ["sudo", "chown", f"{os.getuid()}:{os.getgid()}", device],
-            check=True,
-        )
-        await self.cmd(
-            [
-                "sudo",
-                "dd",
-                "if=/dev/null",
-                f"of={loop_img_name}",
-                "bs=1",
-                "count=0",
-                f"seek={size}",
-            ],
-            check=True,
-        )
-        await self.cmd(["sudo", "losetup", device, loop_img_name], check=True)
-        await self.cmd(["chcon", "-t", "fixed_disk_device_t", device])
-
-    async def remove_loop_device(self, device: str):
-        loop_img_name = os.path.join(self.loop_img_dir, self.device_image(device))
-        if os.path.ismount(device):
-            await self.cmd(["umount", device], check=True)
-        if host.path_exists(device):
-            await self.cmd(["sudo", "losetup", "-d", device])
-            await self.cmd(["sudo", "rm", "-f", device], check=True)
-        if host.path_exists(loop_img_name):
-            os.remove(loop_img_name)
-
-    def device_name(self, index: int):
-        return f"/dev/loop{self.loop_device_count * self.index + index}"
-
-    def device_image(self, device: str):
-        return f"{self.name}-{device.removeprefix('/dev/loop')}"
-
     def _get_available_dmi_mounts(self) -> List[str]:
         """Return volume mount args for DMI files that exist on the host.
 
diff --git a/ceph_devstack/resources/ceph/host_loops.py b/ceph_devstack/resources/ceph/host_loops.py
new file mode 100644
index 00000000..51f1e4b7
--- /dev/null
+++ b/ceph_devstack/resources/ceph/host_loops.py
@@ -0,0 +1,123 @@
+"""Host loop device discovery, allocation, and provisioning mixin."""
+
+from pathlib import Path
+from typing import Any, ClassVar
+
+from ceph_devstack import logger
+from ceph_devstack.block_pool import shared_pool
+from ceph_devstack.resources.ceph.block_devices import (
+    BlockDeviceProvisioner,
+    attached_loops,
+    backing_files,
+    loop_number,
+    loop_path,
+)
+
+
+class LoopDeviceMixin:
+    """Mixin for Container subclasses that manage host loop devices.
+
+    Subclasses must define a ``loop_img_dir`` property returning a ``Path``.
+    Set ``trigger_udev = True`` on the class to trigger udev after device creation.
+    """
+
+    name: str
+    config: dict[str, Any]
+    cmd: Any
+    trigger_udev: bool = False
+    _allocated: ClassVar[set[int]] = set()
+
+    @property
+    def loop_img_dir(self) -> Path:
+        raise NotImplementedError
+
+    @classmethod
+    def discover_used_loop_numbers(cls, image_dir: Path) -> set[int]:
+        """Loop numbers already attached on the host or claimed by backing files."""
+        attached = {number for number, _, _ in attached_loops()}
+        return attached | {number for _, number in backing_files(image_dir)}
+
+    @classmethod
+    def owner_loop_numbers(cls, owner: str, image_dir: Path) -> list[int]:
+        """Existing backing-file loop numbers for an owner, lowest first."""
+        return sorted(n for o, n in backing_files(image_dir) if o == owner)
+
+    @classmethod
+    def allocate_loop_numbers(
+        cls, owner: str, count: int, image_dir: Path
+    ) -> list[int]:
+        """Pick loop numbers for an owner from config count and host availability."""
+        used = cls.discover_used_loop_numbers(image_dir) | cls._allocated
+        reclaimed = cls.owner_loop_numbers(owner, image_dir)
+        numbers: list[int] = []
+        for index in range(count):
+            if index < len(reclaimed):
+                numbers.append(reclaimed[index])
+                continue
+            candidate = 0
+            while candidate in used or candidate in numbers:
+                candidate += 1
+            numbers.append(candidate)
+            used.add(candidate)
+        cls._allocated.update(numbers)
+        return numbers
+
+    @classmethod
+    def allocate_loop_devices(
+        cls, owner: str, count: int, image_dir: Path
+    ) -> list[str]:
+        return [
+            loop_path(number)
+            for number in cls.allocate_loop_numbers(owner, count, image_dir)
+        ]
+
+    def _init_loop_devices(self) -> None:
+        self.loop_device_count: int = self.config["loop_device_count"]
+        self._devices: list[str] | None = None
+        self._block_device_provisioner: BlockDeviceProvisioner | None = None
+
+    def allocate_devices(self) -> list[str]:
+        """Pick this container's host loop devices now, if not chosen already."""
+        if self._devices is None:
+            self._devices = self.allocate_loop_devices(
+                self.name, self.loop_device_count, self.loop_img_dir
+            )
+        return self._devices
+
+    @property
+    def devices(self) -> list[str]:
+        return self.allocate_devices()
+
+    def _block_provisioner(self) -> BlockDeviceProvisioner:
+        if self._block_device_provisioner is None:
+            self._block_device_provisioner = BlockDeviceProvisioner(
+                self.name,
+                image_dir=self.loop_img_dir,
+                file_size=self.config["loop_device_size"],
+                cmd=self.cmd,
+                trigger_udev=self.trigger_udev,
+                pool=shared_pool(),
+            )
+        return self._block_device_provisioner
+
+    async def create_loop_devices(self):
+        if self.devices:
+            numbers = ", ".join(str(loop_number(d)) for d in self.devices)
+            logger.info(
+                f"{self.name}: host loop devices {numbers} "
+                f"({self.config['loop_device_size']} each)"
+            )
+        await self._block_provisioner().create_devices(self.devices)
+
+    async def remove_loop_devices(self):
+        """Detach the devices this container holds on the host.
+
+        Devices are found from host state (plus any picked earlier in this
+        process), so removal never allocates fresh numbers that might belong
+        to someone else. Their numbers are then released for reallocation.
+        """
+        provisioner = self._block_provisioner()
+        devices = set(provisioner.owned_devices()) | set(self._devices or [])
+        await provisioner.remove_devices(sorted(devices, key=loop_number))
+        self._allocated.difference_update(loop_number(d) for d in devices)
+        self._devices = None
diff --git a/ceph_devstack/resources/ceph/requirements.py b/ceph_devstack/resources/ceph/requirements.py
index 7a19d3e9..aee63c49 100644
--- a/ceph_devstack/resources/ceph/requirements.py
+++ b/ceph_devstack/resources/ceph/requirements.py
@@ -1,12 +1,19 @@
 import hashlib
 import re
+import shlex
 
 from subprocess import CalledProcessError
 
 from ceph_devstack import logger, PROJECT_ROOT
+from ceph_devstack.block_pool import BlockPool, BlockPoolError, shared_pool
+from ceph_devstack.host import Host
 from ceph_devstack.requirements import Requirement, FixableRequirement
 
 
+def _host_user(host: Host) -> str:
+    return host.output(["whoami"]).stdout.decode().strip()
+
+
 class HasSudo(Requirement):
     check_cmd = ["sudo", "true"]
     suggest_msg = "sudo access is required"
@@ -31,7 +38,7 @@ async def check(self):
                 .communicate()[0]
                 .decode()
             )
-            user = self.host.run(["whoami"]).communicate()[0].strip().decode()
+            user = _host_user(self.host)
             if self.host.type == "local":
                 self.fix_cmd = ["sudo", "usermod", "-a", "-G", group, user]
             else:
@@ -47,6 +54,73 @@ async def suggest(self):
             )
 
 
+class BlockPoolRequirement(Requirement):
+    """Passes without a configured pool; fails (logged) if its config is invalid.
+
+    An invalid config fails in evaluate(), before any suggest or --fix step,
+    since there is nothing to fix on the host for it.
+    """
+
+    async def evaluate(self) -> bool:
+        try:
+            pool = shared_pool()
+        except BlockPoolError as e:
+            logger.error(f"block pool: {e}")
+            return False
+        return pool is None or await super().evaluate()
+
+    async def check(self) -> bool:
+        pool = shared_pool()
+        return pool is None or await self.check_pool(pool)
+
+    async def check_pool(self, pool: BlockPool) -> bool:
+        raise NotImplementedError
+
+
+# Fedora CoreOS (e.g. a podman machine) defines the disk group in
+# /usr/lib/group, where usermod silently fails to add members; copy the group
+# into /etc/group first. Elsewhere it is already there and this does nothing.
+DISK_GROUP_TO_ETC = (
+    "grep -q '^disk:' /etc/group || grep '^disk:' /usr/lib/group >> /etc/group"
+)
+
+
+class BlockPoolDiskGroup(BlockPoolRequirement, FixableRequirement):
+    check_cmd = ["bash", "-c", "id -nG | grep -qw disk"]
+    suggest_msg = (
+        "block pool parent requires membership in the disk group "
+        "(log out and back in after joining)"
+    )
+
+    async def check_pool(self, pool: BlockPool) -> bool:
+        if await Requirement.check(self):
+            return True
+        user = shlex.quote(_host_user(self.host))
+        self.fix_cmd = [
+            "sudo",
+            "bash",
+            "-c",
+            f"{DISK_GROUP_TO_ETC}; usermod -a -G disk {user}",
+        ]
+        return False
+
+
+class BlockPoolParentAccessible(BlockPoolRequirement):
+    suggest_msg = "block pool parent must be readable and writable by this user"
+
+    async def check_pool(self, pool: BlockPool) -> bool:
+        proc = await self.host.arun(
+            ["test", "-r", pool.parent, "-a", "-w", pool.parent]
+        )
+        if await proc.wait() == 0:
+            return True
+        logger.error(
+            f"{self.suggest_msg} ({pool.parent}). "
+            "Join the disk group and re-login, or fix device permissions."
+        )
+        return False
+
+
 class SELinuxModule(FixableRequirement):
     MODULE_NAME = "ceph_devstack"
     SHIPPED_PP = PROJECT_ROOT / f"{MODULE_NAME}.pp"
diff --git a/tests/conftest.py b/tests/conftest.py
index 6dfea423..40fc5b91 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -1,19 +1,165 @@
+import json
 import os
 import pathlib
-import pytest
 import random
-
+import subprocess
+from collections.abc import Callable
 from datetime import datetime, timedelta
+from unittest.mock import AsyncMock, patch
+
+import pytest
 
 from ceph_devstack import config
+from ceph_devstack.block_pool import BlockPool, shared_pool
+from ceph_devstack.host import Host
+from ceph_devstack.resources.ceph.host_loops import LoopDeviceMixin
 
 
 @pytest.fixture(autouse=True)
 def reset_config():
     config.load()
+    shared_pool.cache_clear()
+    LoopDeviceMixin._allocated.clear()
     yield
 
 
+class FakeHost(Host):
+    """In-memory host for block-pool and loop-device commands.
+
+    ``files`` maps absolute paths to contents; directories are implied by path
+    prefixes. ``loops`` lists attached loop devices as (number, offset,
+    backing file) for ``losetup --list``. ``responses`` gives (returncode,
+    stdout) by argv[0] and overrides everything else; the defaults describe an
+    unmounted device without signatures. Any other command fails the test.
+    Every command run through ``output`` is recorded in ``commands``.
+    """
+
+    def __init__(self):
+        self.files: dict[str, bytearray] = {}
+        self.loops: list[tuple[int, int, str]] = []
+        self.responses: dict[str, tuple[int, bytes]] = {
+            "findmnt": (1, b""),
+            "blkid": (2, b""),
+        }
+        self.commands: list[list[str]] = []
+
+    def add_file(self, path: str | pathlib.Path, data: bytes | str = b"") -> None:
+        if isinstance(data, str):
+            data = data.encode()
+        self.files[str(path)] = bytearray(data)
+
+    def add_block_device(
+        self, name: str, size: int = 20 * 1024**2, disk: str | None = None
+    ) -> str:
+        """Add /dev/<name> of size zero bytes and its sysfs entry; return the path.
+
+        For a partition, ``disk`` also lists it under the disk's sysfs directory.
+        """
+        self.add_file(f"/dev/{name}", bytes(size))
+        self.add_file(f"/sys/class/block/{name}/size", f"{size // 512}\n")
+        if disk:
+            self.add_file(f"/sys/class/block/{disk}/{name}/size", f"{size // 512}\n")
+        return f"/dev/{name}"
+
+    def _children(self, path: str) -> list[str]:
+        prefix = path.rstrip("/") + "/"
+        return sorted(
+            {p[len(prefix) :].split("/")[0] for p in self.files if p.startswith(prefix)}
+        )
+
+    def path_exists(self, path) -> bool:
+        return str(path) in self.files or bool(self._children(str(path)))
+
+    def output(self, args, input=None) -> subprocess.CompletedProcess:
+        self.commands.append(args)
+        name, *rest = args
+        if name in self.responses:
+            rc, out = self.responses[name]
+        elif name == "cat":
+            data = self.files.get(rest[0])
+            rc, out = (1, b"") if data is None else (0, bytes(data))
+        elif name == "ls":
+            children = self._children(rest[-1])
+            rc, out = (0, "".join(f"{c}\n" for c in children).encode())
+        elif name == "realpath":
+            rc, out = 0, f"{rest[0]}\n".encode()
+        elif name == "losetup" and rest[0] == "--list":
+            loops = [
+                {"name": f"/dev/loop{n}", "offset": off, "back-file": back}
+                for n, off, back in self.loops
+            ]
+            rc, out = 0, (json.dumps({"loopdevices": loops}).encode() if loops else b"")
+        elif name == "dd":
+            opts = dict(arg.split("=", 1) for arg in rest)
+            bs = int(opts["bs"])
+            rc, out = 0, b""
+            if "if" in opts:
+                start = int(opts["skip"]) * bs
+                out = bytes(
+                    self.files[opts["if"]][start : start + int(opts["count"]) * bs]
+                )
+            else:
+                assert input is not None and opts.get("conv") == "notrunc,fsync"
+                start = int(opts["seek"]) * bs
+                self.files[opts["of"]][start : start + len(input)] = input
+        else:
+            raise AssertionError(f"unexpected host command {args}")
+        return subprocess.CompletedProcess(args, rc, out, b"")
+
+
+@pytest.fixture(autouse=True)
+def fake_host():
+    """Isolate block-pool and loop-device code from the real host.
+
+    Nothing reads the real /sys/class/block or block devices; tests that need
+    sysfs entries, devices, or backing files add them to the fake.
+    """
+    fake = FakeHost()
+    with (
+        patch("ceph_devstack.block_pool.host", fake),
+        patch("ceph_devstack.resources.ceph.block_devices.host", fake),
+    ):
+        yield fake
+
+
+@pytest.fixture
+def mock_cmd() -> AsyncMock:
+    return AsyncMock()
+
+
+@pytest.fixture
+def pool_parent(fake_host: FakeHost) -> str:
+    return fake_host.add_block_device("nvme0n1p1")
+
+
+@pytest.fixture
+def block_pool_factory(
+    tmp_path: pathlib.Path, pool_parent: str
+) -> Callable[..., BlockPool]:
+    def factory(*, allow_enroll: bool = True) -> BlockPool:
+        return BlockPool(
+            tmp_path / "block_pool.json", pool_parent, allow_enroll=allow_enroll
+        )
+
+    return factory
+
+
+@pytest.fixture
+def block_pool(block_pool_factory: Callable[..., BlockPool]) -> BlockPool:
+    return block_pool_factory()
+
+
+@pytest.fixture
+def enrolled_pool(block_pool: BlockPool) -> BlockPool:
+    block_pool.ensure_ready()
+    return block_pool
+
+
+@pytest.fixture
+def disallow_enroll_pool(block_pool_factory: Callable[..., BlockPool]) -> BlockPool:
+    return block_pool_factory(allow_enroll=False)
+
+
 @pytest.fixture(scope="function")
 def create_log_file():
     def _create_log_file(data_dir: pathlib.Path, **kwargs) -> pathlib.Path:
diff --git a/tests/resources/ceph/conftest.py b/tests/resources/ceph/conftest.py
new file mode 100644
index 00000000..d25b9da4
--- /dev/null
+++ b/tests/resources/ceph/conftest.py
@@ -0,0 +1,50 @@
+from collections.abc import Iterator
+from pathlib import Path
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from ceph_devstack.resources.ceph.block_devices import BlockDeviceProvisioner
+
+
+@pytest.fixture
+def pool_provisioner_factory(tmp_path: Path, block_pool, mock_cmd: AsyncMock):
+    def factory(file_size: str = "1M") -> BlockDeviceProvisioner:
+        return BlockDeviceProvisioner(
+            "ceph_node",
+            image_dir=tmp_path / "disk_images",
+            file_size=file_size,
+            cmd=mock_cmd,
+            pool=block_pool,
+        )
+
+    return factory
+
+
+@pytest.fixture
+def pool_provisioner(pool_provisioner_factory) -> BlockDeviceProvisioner:
+    return pool_provisioner_factory()
+
+
+@pytest.fixture
+def ready_pool_provisioner(
+    pool_provisioner: BlockDeviceProvisioner,
+) -> Iterator[BlockDeviceProvisioner]:
+    with patch.object(pool_provisioner, "remove_device", new=AsyncMock()):
+        yield pool_provisioner
+
+
+@pytest.fixture
+def sparse_provisioner(tmp_path: Path, mock_cmd: AsyncMock) -> BlockDeviceProvisioner:
+    return BlockDeviceProvisioner(
+        "testnode_0",
+        image_dir=tmp_path / "disk_images",
+        file_size="1M",
+        cmd=mock_cmd,
+        pool=None,
+    )
+
+
+@pytest.fixture
+def large_pool_provisioner(pool_provisioner_factory) -> BlockDeviceProvisioner:
+    return pool_provisioner_factory(file_size="32M")
diff --git a/tests/resources/ceph/test_block_devices.py b/tests/resources/ceph/test_block_devices.py
new file mode 100644
index 00000000..992602f7
--- /dev/null
+++ b/tests/resources/ceph/test_block_devices.py
@@ -0,0 +1,226 @@
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from ceph_devstack.block_pool import BlockPoolError
+from ceph_devstack.resources.ceph.block_devices import (
+    BlockDeviceProvisioner,
+    attached_loops,
+)
+
+MiB = 1024**2
+
+
+class TestBlockDeviceProvisioner:
+    async def test_create_device_uses_pool_slice(
+        self,
+        ready_pool_provisioner: BlockDeviceProvisioner,
+        mock_cmd: AsyncMock,
+        pool_parent,
+    ):
+        await ready_pool_provisioner.create_device("/dev/loop9", 0)
+        losetup = [
+            call.args[0]
+            for call in mock_cmd.await_args_list
+            if "losetup" in call.args[0]
+        ][0]
+        offset_idx = losetup.index("--offset")
+        assert losetup[offset_idx + 1] == "0"
+        sizelimit_idx = losetup.index("--sizelimit")
+        assert losetup[sizelimit_idx + 1] == str(ready_pool_provisioner.device_size)
+        assert losetup[-2:] == ["/dev/loop9", pool_parent]
+
+    async def test_remove_devices_releases_pool_owner(
+        self,
+        ready_pool_provisioner: BlockDeviceProvisioner,
+    ):
+        await ready_pool_provisioner.create_device("/dev/loop9", 0)
+        pool = ready_pool_provisioner.pool
+        assert pool is not None
+        assert list(pool.allocations) == ["ceph_node:0"]
+        await ready_pool_provisioner.remove_devices(["/dev/loop9"])
+        assert pool.allocations == {}
+        assert pool.free_regions == [{"offset": 0, "size": 1024**2}]
+
+    async def test_failed_create_removes_the_device_node(
+        self,
+        pool_provisioner: BlockDeviceProvisioner,
+        mock_cmd: AsyncMock,
+    ):
+        pool = pool_provisioner.pool
+        assert pool is not None
+        with (
+            patch.object(pool_provisioner, "remove_device", new=AsyncMock()) as remove,
+            patch.object(
+                pool, "get_or_allocate", side_effect=BlockPoolError("not empty")
+            ),
+            pytest.raises(BlockPoolError),
+        ):
+            await pool_provisioner.create_device("/dev/loop9", 0)
+        assert ["sudo", "mknod", "-m700", "/dev/loop9", "b", "7", "9"] in [
+            call.args[0] for call in mock_cmd.await_args_list
+        ]
+        assert remove.await_args_list == [(("/dev/loop9",),)] * 2
+
+    async def test_failed_cleanup_does_not_hide_the_create_error(
+        self,
+        pool_provisioner: BlockDeviceProvisioner,
+        caplog,
+    ):
+        pool = pool_provisioner.pool
+        assert pool is not None
+        cleanup = AsyncMock(side_effect=[None, RuntimeError("umount failed")])
+        with (
+            patch.object(pool_provisioner, "remove_device", new=cleanup),
+            patch.object(
+                pool, "get_or_allocate", side_effect=BlockPoolError("not empty")
+            ),
+            pytest.raises(BlockPoolError, match="not empty"),
+        ):
+            await pool_provisioner.create_device("/dev/loop9", 0)
+        assert "could not clean up /dev/loop9" in caplog.text
+
+    async def test_remove_devices_warns_about_regions_it_cannot_wipe(
+        self,
+        ready_pool_provisioner: BlockDeviceProvisioner,
+        caplog,
+    ):
+        pool = ready_pool_provisioner.pool
+        assert pool is not None
+        with patch.object(pool, "release_owner", return_value=["ceph_node:0"]):
+            await ready_pool_provisioner.remove_devices([])
+        assert "could not wipe pool regions ceph_node:0" in caplog.text
+
+    async def test_create_device_uses_sparse_file_without_pool(
+        self,
+        sparse_provisioner: BlockDeviceProvisioner,
+        mock_cmd: AsyncMock,
+        tmp_path,
+    ):
+        with (
+            patch.object(sparse_provisioner, "remove_device", new=AsyncMock()),
+            patch.object(sparse_provisioner, "_ensure_loop_module", new=AsyncMock()),
+        ):
+            await sparse_provisioner.create_device("/dev/loop1", 0)
+        image_dir = tmp_path / "disk_images"
+        image_path = str(image_dir / "testnode_0-1")
+        commands = [call.args[0] for call in mock_cmd.await_args_list]
+        assert ["mkdir", "-p", str(image_dir)] in commands
+        dd = next(c for c in commands if "dd" in c)
+        assert f"of={image_path}" in dd
+        assert f"seek={sparse_provisioner.device_size}" in dd
+        assert ["sudo", "losetup", "/dev/loop1", image_path] in commands
+
+    async def test_remove_device_deletes_backing_file_on_host(
+        self,
+        sparse_provisioner: BlockDeviceProvisioner,
+        mock_cmd: AsyncMock,
+        tmp_path,
+    ):
+        await sparse_provisioner.remove_device("/dev/loop1")
+        image_path = str(tmp_path / "disk_images" / "testnode_0-1")
+        mock_cmd.assert_awaited_once_with(["rm", "-f", image_path], check=False)
+
+    async def test_rejects_region_larger_than_parent(
+        self,
+        large_pool_provisioner: BlockDeviceProvisioner,
+    ):
+        with (
+            patch.object(large_pool_provisioner, "remove_device", new=AsyncMock()),
+            pytest.raises(BlockPoolError, match="exceeds"),
+        ):
+            await large_pool_provisioner.create_device("/dev/loop9", 0)
+
+    async def test_ensure_loop_module_loads_when_missing(
+        self,
+        sparse_provisioner: BlockDeviceProvisioner,
+        mock_cmd: AsyncMock,
+    ):
+        mock_proc = AsyncMock()
+        mock_proc.wait = AsyncMock(return_value=1)
+        mock_cmd.return_value = mock_proc
+        await sparse_provisioner._ensure_loop_module()
+        mock_cmd.assert_any_await(["sudo", "modprobe", "loop"])
+
+
+class TestHostLoopDiscovery:
+    def test_attached_loops_reads_losetup_list(self, fake_host):
+        fake_host.responses["losetup"] = (
+            0,
+            b'{"loopdevices": [{"name": "/dev/loop7", "offset": "4096",'
+            b' "back-file": "/dev/sda1"}]}',
+        )
+        assert attached_loops() == [(7, 4096, "/dev/sda1")]
+        assert fake_host.commands == [
+            ["losetup", "--list", "--json", "--output", "NAME,OFFSET,BACK-FILE"]
+        ]
+
+    def test_pool_owner_devices_are_its_loops_on_its_regions(
+        self, fake_host, pool_provisioner
+    ):
+        pool = pool_provisioner.pool
+        assert pool is not None
+        mine = [pool.get_or_allocate("ceph_node", i, MiB) for i in (0, 1)]
+        other = pool.get_or_allocate("testnode_0", 0, MiB)
+        fake_host.loops = [
+            (9, mine[1], pool.parent),
+            (7, mine[0], pool.parent),
+            (3, other, pool.parent),
+            (4, mine[0], "/var/lib/other.img"),
+        ]
+        assert pool_provisioner.owned_devices() == ["/dev/loop7", "/dev/loop9"]
+
+    def test_sparse_owner_devices_come_from_backing_files(
+        self, fake_host, sparse_provisioner, tmp_path
+    ):
+        for name in ("testnode_0-5", "testnode_0-2", "testnode_01-3"):
+            fake_host.add_file(tmp_path / "disk_images" / name)
+        assert sparse_provisioner.owned_devices() == ["/dev/loop2", "/dev/loop5"]
+
+    async def test_regions_stay_allocated_while_loops_remain_attached(
+        self, fake_host, ready_pool_provisioner, mock_cmd, caplog
+    ):
+        pool = ready_pool_provisioner.pool
+        assert pool is not None
+        await ready_pool_provisioner.create_device("/dev/loop9", 0)
+        fake_host.loops = [(9, 0, pool.parent)]  # detach did not take effect
+        with patch("ceph_devstack.resources.ceph.block_devices.asyncio.sleep") as sleep:
+            await ready_pool_provisioner.remove_devices(["/dev/loop9"])
+        mock_cmd.assert_any_await(["sudo", "udevadm", "settle"], check=False)
+        assert sleep.await_count == 4
+        assert "/dev/loop9 still attached" in caplog.text
+        assert list(pool.allocations) == ["ceph_node:0"]
+
+    async def test_regions_released_once_busy_loop_lets_go(
+        self, fake_host, ready_pool_provisioner
+    ):
+        pool = ready_pool_provisioner.pool
+        assert pool is not None
+        await ready_pool_provisioner.create_device("/dev/loop9", 0)
+        fake_host.loops = [(9, 0, pool.parent)]
+
+        async def detach_later(delay):
+            fake_host.loops = []
+
+        with patch(
+            "ceph_devstack.resources.ceph.block_devices.asyncio.sleep",
+            side_effect=detach_later,
+        ):
+            await ready_pool_provisioner.remove_devices(["/dev/loop9"])
+        assert pool.allocations == {}
+
+    @pytest.mark.parametrize(
+        "response",
+        [(1, b""), (0, b"not json"), (0, b'{"other": []}')],
+        ids=["losetup-fails", "garbage", "unexpected-json"],
+    )
+    async def test_failed_loop_discovery_aborts_removal(
+        self, fake_host, ready_pool_provisioner, response
+    ):
+        pool = ready_pool_provisioner.pool
+        assert pool is not None
+        await ready_pool_provisioner.create_device("/dev/loop9", 0)
+        fake_host.responses["losetup"] = response
+        with pytest.raises(BlockPoolError, match="losetup|loop devices"):
+            await ready_pool_provisioner.remove_devices(["/dev/loop9"])
+        assert list(pool.allocations) == ["ceph_node:0"]
diff --git a/tests/resources/ceph/test_cephdevstack_core.py b/tests/resources/ceph/test_cephdevstack_core.py
index fa22590e..ddf2d889 100644
--- a/tests/resources/ceph/test_cephdevstack_core.py
+++ b/tests/resources/ceph/test_cephdevstack_core.py
@@ -61,6 +61,17 @@ def test_service_specs_named_objects_when_count_greater_than_one(self):
         assert testnode_objects[1].name == "testnode_1"
         assert testnode_objects[2].name == "testnode_2"
 
+    def test_multiple_testnodes_get_distinct_loop_devices(self, tmp_path):
+        config["data_dir"] = str(tmp_path)
+        devstack = CephDevStack()
+        testnode_objects = devstack.service_specs["testnode"]["objects"]
+        all_devices = []
+        for node in testnode_objects:
+            all_devices.extend(node.devices)
+        assert len(all_devices) == len(set(all_devices)), (
+            f"Loop device collision: {all_devices}"
+        )
+
     def test_service_specs_sets_postgres_paddles_url(self):
         devstack = CephDevStack()
         paddles_obj = devstack.service_specs["paddles"]["objects"][0]
@@ -439,7 +450,11 @@ def test_init_without_postgres(self):
             "paddles": {"image": "paddles:latest", "count": 1},
             "beanstalk": {"image": "beanstalk:latest", "count": 1},
             "pulpito": {"image": "pulpito:latest", "count": 1},
-            "testnode": {"image": "testnode:latest", "count": 3},
+            "testnode": {
+                "image": "testnode:latest",
+                "count": 3,
+                "loop_device_count": 1,
+            },
             "teuthology": {"image": "teuthology:latest", "count": 1},
             "archive": {"image": "archive:latest", "count": 1},
         }
diff --git a/tests/resources/ceph/test_host_loops.py b/tests/resources/ceph/test_host_loops.py
new file mode 100644
index 00000000..d2b46966
--- /dev/null
+++ b/tests/resources/ceph/test_host_loops.py
@@ -0,0 +1,117 @@
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from ceph_devstack.block_pool import BlockPoolError
+from ceph_devstack.resources.ceph.block_devices import (
+    BlockDeviceProvisioner,
+    loop_number,
+    loop_path,
+)
+from ceph_devstack.resources.ceph.host_loops import LoopDeviceMixin
+
+
+def make_backing(fake_host, image_dir, *names):
+    """Create empty backing files for loop devices on the (fake) host."""
+    for name in names:
+        fake_host.add_file(image_dir / name)
+
+
+def make_node(name, image_dir, count=2):
+    class Node(LoopDeviceMixin):
+        loop_img_dir = image_dir
+        cmd = AsyncMock()
+
+        def __init__(self):
+            self.name = name
+            self.config = {"loop_device_count": count, "loop_device_size": "1M"}
+            self._init_loop_devices()
+
+    return Node()
+
+
+class TestHostLoops:
+    def test_loop_path_round_trips(self):
+        assert loop_path(21) == "/dev/loop21"
+        assert loop_number(loop_path(21)) == 21
+
+    def test_discover_used_loop_numbers_from_backing_files(self, fake_host, tmp_path):
+        make_backing(fake_host, tmp_path, "testnode_0-1", "ceph_node-4", "ignore-me")
+        assert LoopDeviceMixin.discover_used_loop_numbers(tmp_path) == {1, 4}
+
+    def test_discover_used_loop_numbers_includes_attached_loops(
+        self, fake_host, tmp_path
+    ):
+        fake_host.loops = [(0, 0, "/img"), (5, 1048576, "/dev/nvme0n1p1")]
+        make_backing(fake_host, tmp_path, "testnode_0-3")
+        assert LoopDeviceMixin.discover_used_loop_numbers(tmp_path) == {0, 3, 5}
+
+    def test_allocation_fails_when_attached_loops_are_unknown(
+        self, fake_host, tmp_path
+    ):
+        fake_host.responses["losetup"] = (1, b"")
+        with pytest.raises(BlockPoolError, match="cannot list loop devices"):
+            LoopDeviceMixin.allocate_loop_numbers("testnode_0", 1, tmp_path)
+
+    def test_owner_loop_numbers(self, fake_host, tmp_path):
+        make_backing(
+            fake_host, tmp_path, "testnode_1-2", "testnode_1-0", "testnode_10-1"
+        )
+        assert LoopDeviceMixin.owner_loop_numbers("testnode_1", tmp_path) == [0, 2]
+
+    def test_allocate_loop_numbers_uses_lowest_free(self, fake_host, tmp_path):
+        make_backing(fake_host, tmp_path, "other-0")
+        assert LoopDeviceMixin.allocate_loop_numbers("testnode_0", 2, tmp_path) == [
+            1,
+            2,
+        ]
+
+    def test_allocate_loop_numbers_reuses_owner_backing_files(
+        self, fake_host, tmp_path
+    ):
+        make_backing(fake_host, tmp_path, "testnode_1-4", "testnode_1-1", "other-0")
+        assert LoopDeviceMixin.allocate_loop_numbers("testnode_1", 3, tmp_path) == [
+            1,
+            4,
+            2,
+        ]
+
+    async def test_remove_targets_held_devices_without_allocating(
+        self, fake_host, tmp_path
+    ):
+        """A fresh process (e.g. `remove`) acts on what the host shows it holds."""
+        make_backing(fake_host, tmp_path, "testnode_0-4", "other-0")
+        node = make_node("testnode_0", tmp_path)
+        with (
+            patch.object(LoopDeviceMixin, "allocate_loop_devices") as allocate,
+            patch.object(BlockDeviceProvisioner, "remove_devices") as remove,
+        ):
+            await node.remove_loop_devices()
+        allocate.assert_not_called()
+        remove.assert_awaited_once_with(["/dev/loop4"])
+
+    def test_allocate_loop_numbers_avoids_cross_call_collisions(
+        self, fake_host, tmp_path
+    ):
+        """Multiple owners allocating from the same dir get distinct numbers."""
+        make_backing(fake_host, tmp_path, "existing-0")
+        a = LoopDeviceMixin.allocate_loop_numbers("testnode_0", 1, tmp_path)
+        b = LoopDeviceMixin.allocate_loop_numbers("testnode_1", 1, tmp_path)
+        c = LoopDeviceMixin.allocate_loop_numbers("testnode_2", 1, tmp_path)
+        assert a == [1]
+        assert b == [2]
+        assert c == [3]
+
+    async def test_removed_devices_are_released_for_reallocation(self, tmp_path):
+        first = make_node("testnode_0", tmp_path)
+        assert first.allocate_devices() == ["/dev/loop0", "/dev/loop1"]
+        provisioner = MagicMock(
+            owned_devices=MagicMock(return_value=[]), remove_devices=AsyncMock()
+        )
+        with patch.object(first, "_block_provisioner", return_value=provisioner):
+            await first.remove_loop_devices()
+        provisioner.remove_devices.assert_awaited_once_with(
+            ["/dev/loop0", "/dev/loop1"]
+        )
+        second = make_node("testnode_1", tmp_path)
+        assert second.allocate_devices() == ["/dev/loop0", "/dev/loop1"]
diff --git a/tests/resources/ceph/test_requirements_ceph.py b/tests/resources/ceph/test_requirements_ceph.py
index 3e93261f..756b5aeb 100644
--- a/tests/resources/ceph/test_requirements_ceph.py
+++ b/tests/resources/ceph/test_requirements_ceph.py
@@ -1,4 +1,6 @@
+import contextlib
 import hashlib
+import subprocess
 import pytest
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -9,6 +11,8 @@
     HasSudo,
     LoopControlDeviceExists,
     LoopControlDeviceWriteable,
+    BlockPoolDiskGroup,
+    BlockPoolParentAccessible,
     SELinuxModule,
 )
 
@@ -77,55 +81,25 @@ async def test_loop_control_writeable_check_true(self):
             result = await self.req.check()
             assert result is True
 
-    async def test_loop_control_writeable_check_false_local(self):
-        mock_check_proc = AsyncMock()
-        mock_check_proc.wait = AsyncMock(return_value=1)
-
-        mock_stat_proc = MagicMock()
-        mock_stat_proc.communicate = MagicMock(return_value=(b"disk", 0))
-
-        mock_whoami_proc = MagicMock()
-        mock_whoami_proc.communicate = MagicMock(return_value=(b"testuser", 0))
-
-        async def side_effect_arun(args):
-            if "stat" in args:
-                return mock_stat_proc
-            if "whoami" in args:
-                return mock_whoami_proc
-            return mock_check_proc
-
-        with (
-            patch.object(self.req.host, "arun", side_effect=side_effect_arun),
-            patch.object(self.req.host, "type", "local"),
-        ):
-            result = await self.req.check()
-            assert result is False
-            assert "usermod" in " ".join(self.req.fix_cmd)
-
-    async def test_loop_control_writeable_check_false_remote(self):
-        mock_check_proc = AsyncMock()
-        mock_check_proc.wait = AsyncMock(return_value=1)
-
-        mock_stat_proc = MagicMock()
-        mock_stat_proc.communicate = MagicMock(return_value=(b"disk", 0))
-
-        mock_whoami_proc = MagicMock()
-        mock_whoami_proc.communicate = MagicMock(return_value=(b"testuser", 0))
-
-        async def side_effect_arun(args):
-            if "stat" in args:
-                return mock_stat_proc
-            if "whoami" in args:
-                return mock_whoami_proc
-            return mock_check_proc
-
+    @pytest.mark.parametrize(
+        "host_type,fix_cmd",
+        [
+            ("local", ["sudo", "usermod", "-a", "-G", "disk", "testuser"]),
+            ("remote", ["sudo", "chgrp", "testuser", "/dev/loop-control"]),
+        ],
+    )
+    async def test_loop_control_writeable_check_false(self, host_type, fix_cmd):
+        mock_check_proc = AsyncMock(wait=AsyncMock(return_value=1))
+        mock_stat_proc = MagicMock(communicate=MagicMock(return_value=(b"disk", 0)))
+        whoami = subprocess.CompletedProcess([], 0, b"testuser\n", b"")
         with (
-            patch.object(self.req.host, "arun", side_effect=side_effect_arun),
-            patch.object(self.req.host, "type", "remote"),
+            patch.object(self.req.host, "arun", return_value=mock_check_proc),
+            patch.object(self.req.host, "run", return_value=mock_stat_proc),
+            patch.object(self.req.host, "output", return_value=whoami),
+            patch.object(self.req.host, "type", host_type),
         ):
-            result = await self.req.check()
-            assert result is False
-            assert "chgrp" in " ".join(self.req.fix_cmd)
+            assert await self.req.check() is False
+        assert self.req.fix_cmd == fix_cmd
 
 
 SHIPPED = b"shipped module"
@@ -226,55 +200,113 @@ class MockRemoteHost:
             assert req.fix_cmd[7].endswith("ceph_devstack.pp")
 
 
-class TestCephDevStackCheckRequirements:
-    async def test_check_requirements_returns_true_when_all_pass(self):
-        devstack = CephDevStack()
-        devstack.service_specs = {}
-        config["containers"] = {}
+class TestBlockPoolRequirements:
+    @pytest.fixture
+    def configured_pool(self, tmp_path, pool_parent):
+        config["block_pool"] = {"parent": pool_parent, "state_dir": str(tmp_path)}
+        return pool_parent
+
+    @pytest.mark.parametrize("cls", [BlockPoolDiskGroup, BlockPoolParentAccessible])
+    async def test_skipped_when_pool_disabled(self, cls):
+        req = cls()
+        with patch.object(req.host, "arun") as arun:
+            assert await req.check() is True
+        arun.assert_not_called()
+
+    async def test_disk_group_member_passes(self, configured_pool):
+        req = BlockPoolDiskGroup()
+        mock_proc = AsyncMock(wait=AsyncMock(return_value=0))
+        with patch.object(req.host, "arun", return_value=mock_proc) as arun:
+            assert await req.check() is True
+        arun.assert_awaited_once_with(["bash", "-c", "id -nG | grep -qw disk"])
 
+    async def test_disk_group_non_member_gets_fix_cmd(self, configured_pool):
+        req = BlockPoolDiskGroup()
+        mock_proc = AsyncMock(wait=AsyncMock(return_value=1))
+        whoami = subprocess.CompletedProcess([], 0, b"zack\n", b"")
         with (
-            patch("ceph_devstack.resources.ceph.HasSudo") as MockHasSudo,
-            patch(
-                "ceph_devstack.resources.ceph.LoopControlDeviceExists"
-            ) as MockLoopCtrl,
-            patch(
-                "ceph_devstack.resources.ceph.LoopControlDeviceWriteable"
-            ) as MockLoopCtrlWrite,
-            patch("ceph_devstack.host.host.selinux_enforcing") as mock_selinux,
+            patch.object(req.host, "arun", return_value=mock_proc),
+            patch.object(req.host, "output", return_value=whoami),
         ):
-            MockHasSudo.return_value = AsyncMock(evaluate=AsyncMock(return_value=True))
-            MockLoopCtrl.return_value = AsyncMock(evaluate=AsyncMock(return_value=True))
-            MockLoopCtrlWrite.return_value = AsyncMock(
-                evaluate=AsyncMock(return_value=True)
+            assert await req.check() is False
+        assert req.fix_cmd == [
+            "sudo",
+            "bash",
+            "-c",
+            "grep -q '^disk:' /etc/group || grep '^disk:' /usr/lib/group"
+            " >> /etc/group; usermod -a -G disk zack",
+        ]
+
+    @pytest.mark.parametrize("fix", [False, True])
+    @pytest.mark.parametrize("cls", [BlockPoolDiskGroup, BlockPoolParentAccessible])
+    async def test_invalid_pool_config_fails_without_traceback(self, cls, fix, caplog):
+        config["block_pool"] = {"parent": "/dev/dm-0"}
+        req = cls()
+        with (
+            patch.dict(config, {"args": {"fix": fix}}),
+            patch.object(req.host, "arun") as arun,
+        ):
+            assert await req.evaluate() is False
+        arun.assert_not_called()
+        assert "not allowed" in caplog.text
+
+    @pytest.mark.parametrize("returncode,expected", [(0, True), (1, False)])
+    async def test_parent_access_checked_on_host(
+        self, configured_pool, returncode, expected
+    ):
+        req = BlockPoolParentAccessible()
+        mock_proc = AsyncMock(wait=AsyncMock(return_value=returncode))
+        with patch.object(req.host, "arun", return_value=mock_proc) as arun:
+            assert await req.check() is expected
+        arun.assert_awaited_once_with(
+            ["test", "-r", configured_pool, "-a", "-w", configured_pool]
+        )
+
+
+CORE_REQUIREMENTS = [
+    "HasSudo",
+    "LoopControlDeviceExists",
+    "LoopControlDeviceWriteable",
+    "BlockPoolDiskGroup",
+    "BlockPoolParentAccessible",
+]
+
+
+@pytest.fixture
+def requirement_results():
+    """Stub CephDevStack's requirement classes; set an entry False to fail it."""
+    results = dict.fromkeys(CORE_REQUIREMENTS, True)
+    with contextlib.ExitStack() as stack:
+        for name in CORE_REQUIREMENTS:
+            mock = stack.enter_context(patch(f"ceph_devstack.resources.ceph.{name}"))
+            mock.return_value.evaluate = AsyncMock(
+                side_effect=lambda name=name: results[name]
             )
-            mock_selinux.return_value = False
-            result = await devstack.check_requirements()
-            assert result is True
+        stack.enter_context(
+            patch("ceph_devstack.host.host.selinux_enforcing", return_value=False)
+        )
+        yield results
+
 
-    async def test_check_requirements_returns_false_when_repo_missing(self):
+class TestCephDevStackCheckRequirements:
+    @pytest.mark.parametrize("failing", [None, *CORE_REQUIREMENTS])
+    async def test_check_requirements_fails_if_any_requirement_fails(
+        self, requirement_results, failing
+    ):
+        devstack = CephDevStack()
+        devstack.service_specs = {}
+        config["containers"] = {}
+        if failing:
+            requirement_results[failing] = False
+        assert await devstack.check_requirements() is (failing is None)
+
+    async def test_check_requirements_returns_false_when_repo_missing(
+        self, requirement_results
+    ):
         devstack = CephDevStack()
         devstack.service_specs = {}
         config["containers"] = {
             "custom": {"repo": "/nonexistent/path"},
         }
-
-        with (
-            patch("ceph_devstack.resources.ceph.HasSudo") as MockHasSudo,
-            patch(
-                "ceph_devstack.resources.ceph.LoopControlDeviceExists"
-            ) as MockLoopCtrl,
-            patch(
-                "ceph_devstack.resources.ceph.LoopControlDeviceWriteable"
-            ) as MockLoopCtrlWrite,
-            patch("ceph_devstack.host.host.selinux_enforcing") as mock_selinux,
-            patch("ceph_devstack.host.host.path_exists") as mock_path_exists,
-        ):
-            MockHasSudo.return_value = AsyncMock(evaluate=AsyncMock(return_value=True))
-            MockLoopCtrl.return_value = AsyncMock(evaluate=AsyncMock(return_value=True))
-            MockLoopCtrlWrite.return_value = AsyncMock(
-                evaluate=AsyncMock(return_value=True)
-            )
-            mock_selinux.return_value = False
-            mock_path_exists.return_value = False
-            result = await devstack.check_requirements()
-            assert result is False
+        with patch("ceph_devstack.host.host.path_exists", return_value=False):
+            assert await devstack.check_requirements() is False
diff --git a/tests/resources/ceph/test_testnode.py b/tests/resources/ceph/test_testnode.py
index 6971afc5..53bfe2c7 100644
--- a/tests/resources/ceph/test_testnode.py
+++ b/tests/resources/ceph/test_testnode.py
@@ -1,54 +1,60 @@
 from pathlib import Path
 from unittest.mock import patch
 
-import pytest
-
-from ceph_devstack.resources.ceph import TestNode as _TestNode
 from ceph_devstack import config
+from ceph_devstack.resources.ceph import TestNode as _TestNode
 
 
 class TestTestnode:
-    @pytest.fixture(scope="class")
-    @classmethod
-    def cls(self) -> type[_TestNode]:
-        return _TestNode
-
-    def test_testnode_loop_device_count_default_to_one(self, cls):
-        testnode = cls("testnode_1")
+    def test_testnode_loop_device_count_default_to_one(self):
+        testnode = _TestNode("testnode_1")
         assert testnode.loop_device_count == 1
 
-    def test_testnode_create_cmd_includes_related_devices(self, cls):
+    def test_testnode_create_cmd_includes_related_devices(self, tmp_path):
         config.load(Path(__file__).parent.joinpath("fixtures", "testnode-config.toml"))
-        testnode = cls("testnode_1")
+        config["data_dir"] = str(tmp_path)
+        testnode = _TestNode("testnode_1")
         create_cmd = testnode.create_cmd
-        assert "--device=/dev/loop4" in create_cmd
-        assert "--device=/dev/loop5" in create_cmd
-        assert "--device=/dev/loop6" in create_cmd
-        assert "--device=/dev/loop7" in create_cmd
+        assert "--device=/dev/loop0" in create_cmd
+        assert "--device=/dev/loop1" in create_cmd
+        assert "--device=/dev/loop2" in create_cmd
+        assert "--device=/dev/loop3" in create_cmd
 
-    def test_testnode_create_cmd_mounts_run_udev_read_only(self, cls):
-        create_cmd = cls("testnode_1").create_cmd
+    def test_testnode_create_cmd_mounts_run_udev_read_only(self, tmp_path):
+        config["data_dir"] = str(tmp_path)
+        create_cmd = _TestNode("testnode_1").create_cmd
         index = create_cmd.index("/run/udev:/run/udev:ro")
         assert create_cmd[index - 1] == "-v"
 
-    def test_testnode_create_cmd_unmasks_proc_for_nested_containers(self, cls):
-        create_cmd = cls("testnode_1").create_cmd
+    def test_testnode_create_cmd_unmasks_proc_for_nested_containers(self, tmp_path):
+        config["data_dir"] = str(tmp_path)
+        create_cmd = _TestNode("testnode_1").create_cmd
         index = create_cmd.index("unmask=/proc/*")
         assert create_cmd[index - 1] == "--security-opt"
 
-    def test_testnode_devices_is_based_on_loop_device_count_config(self, cls):
+    def test_testnode_devices_is_based_on_loop_device_count_config(self, tmp_path):
         config.load(Path(__file__).parent.joinpath("fixtures", "testnode-config.toml"))
-        testnode = cls("testnode_1")
+        config["data_dir"] = str(tmp_path)
+        testnode = _TestNode("testnode_1")
         assert testnode.loop_device_count == 4
         assert testnode.devices == [
-            "/dev/loop4",
-            "/dev/loop5",
-            "/dev/loop6",
-            "/dev/loop7",
+            "/dev/loop0",
+            "/dev/loop1",
+            "/dev/loop2",
+            "/dev/loop3",
         ]
 
-    def test_get_available_dmi_mounts_with_all_files_present(self, cls):
-        testnode = cls("testnode_1")
+    def test_testnode_reuses_existing_backing_files(self, tmp_path, fake_host):
+        config["data_dir"] = str(tmp_path)
+        image_dir = tmp_path / "disk_images"
+        fake_host.add_file(image_dir / "testnode_1-4")
+        fake_host.add_file(image_dir / "testnode_1-1")
+        config["containers"]["testnode"]["loop_device_count"] = 3
+        testnode = _TestNode("testnode_1")
+        assert testnode.devices == ["/dev/loop1", "/dev/loop4", "/dev/loop0"]
+
+    def test_get_available_dmi_mounts_with_all_files_present(self):
+        testnode = _TestNode("testnode_1")
         with patch(
             "ceph_devstack.resources.ceph.containers.host.path_exists"
         ) as mock_exists:
@@ -63,8 +69,8 @@ def test_get_available_dmi_mounts_with_all_files_present(self, cls):
                 "/dev/null:/sys/class/dmi/id/product_serial",
             ]
 
-    def test_get_available_dmi_mounts_with_no_files_present(self, cls):
-        testnode = cls("testnode_1")
+    def test_get_available_dmi_mounts_with_no_files_present(self):
+        testnode = _TestNode("testnode_1")
         with patch(
             "ceph_devstack.resources.ceph.containers.host.path_exists"
         ) as mock_exists:
@@ -72,8 +78,8 @@ def test_get_available_dmi_mounts_with_no_files_present(self, cls):
             mounts = testnode._get_available_dmi_mounts()
             assert mounts == []
 
-    def test_get_available_dmi_mounts_with_partial_files_present(self, cls):
-        testnode = cls("testnode_1")
+    def test_get_available_dmi_mounts_with_partial_files_present(self):
+        testnode = _TestNode("testnode_1")
         with patch(
             "ceph_devstack.resources.ceph.containers.host.path_exists"
         ) as mock_exists:
@@ -85,9 +91,10 @@ def test_get_available_dmi_mounts_with_partial_files_present(self, cls):
                 "/dev/null:/sys/class/dmi/id/board_serial",
             ]
 
-    def test_create_cmd_includes_dmi_mounts_when_available(self, cls):
+    def test_create_cmd_includes_dmi_mounts_when_available(self, tmp_path):
         config.load(Path(__file__).parent.joinpath("fixtures", "testnode-config.toml"))
-        testnode = cls("testnode_1")
+        config["data_dir"] = str(tmp_path)
+        testnode = _TestNode("testnode_1")
         with patch(
             "ceph_devstack.resources.ceph.containers.host.path_exists"
         ) as mock_exists:
@@ -97,9 +104,10 @@ def test_create_cmd_includes_dmi_mounts_when_available(self, cls):
             assert "/dev/null:/sys/class/dmi/id/chassis_serial" in " ".join(create_cmd)
             assert "/dev/null:/sys/class/dmi/id/product_serial" in " ".join(create_cmd)
 
-    def test_create_cmd_excludes_dmi_mounts_when_unavailable(self, cls):
+    def test_create_cmd_excludes_dmi_mounts_when_unavailable(self, tmp_path):
         config.load(Path(__file__).parent.joinpath("fixtures", "testnode-config.toml"))
-        testnode = cls("testnode_1")
+        config["data_dir"] = str(tmp_path)
+        testnode = _TestNode("testnode_1")
         with patch(
             "ceph_devstack.resources.ceph.containers.host.path_exists"
         ) as mock_exists:
diff --git a/tests/test_block_pool.py b/tests/test_block_pool.py
new file mode 100644
index 00000000..b7e35331
--- /dev/null
+++ b/tests/test_block_pool.py
@@ -0,0 +1,330 @@
+import json
+from unittest.mock import patch
+
+import pytest
+
+from ceph_devstack import config
+from ceph_devstack.block_pool import (
+    BlockPool,
+    BlockPoolError,
+    format_size,
+    parse_size,
+    shared_pool,
+    validate_parent_name,
+)
+
+MiB = 1024**2
+TAIL = 20 * MiB - 4096  # tail marker offset on the default 20 MiB fake parent
+
+
+def write(fake_host, path: str, offset: int, data: bytes) -> None:
+    fake_host.files[path][offset : offset + len(data)] = data
+
+
+class TestParseSize:
+    def test_gigabytes(self):
+        assert parse_size("50G") == 50 * 1024**3
+
+    def test_integer(self):
+        assert parse_size(1024) == 1024
+
+    @pytest.mark.parametrize("value", ["not-a-size", "0", 0, "0G"])
+    def test_invalid(self, value):
+        with pytest.raises(BlockPoolError):
+            parse_size(value)
+
+    def test_format_size_human_units(self):
+        assert format_size(5 * 1024**3) == "5G"
+        assert format_size(12345) == "12345B"
+
+
+class TestValidateParentName:
+    @pytest.mark.parametrize("name", ["nvme0n1p1", "nvme0n1", "sda1"])
+    def test_accepts_disks_and_partitions(self, fake_host, name):
+        parent = fake_host.add_block_device(name)
+        assert validate_parent_name(parent) == parent
+
+    def test_resolves_symlinks_on_host(self, fake_host):
+        fake_host.add_block_device("nvme0n1p1")
+        fake_host.responses["realpath"] = (0, b"/dev/nvme0n1p1\n")
+        assert validate_parent_name("/dev/disk/by-id/x") == "/dev/nvme0n1p1"
+        assert ["realpath", "/dev/disk/by-id/x"] in fake_host.commands
+
+    def test_rejects_device_mapper(self):
+        with pytest.raises(BlockPoolError, match="not allowed"):
+            validate_parent_name("/dev/dm-0")
+
+    def test_rejects_non_dev_path(self):
+        with pytest.raises(BlockPoolError, match="must be a /dev path"):
+            validate_parent_name("/tmp/not-a-device")
+
+    def test_rejects_missing_device(self):
+        with pytest.raises(BlockPoolError, match="does not exist"):
+            validate_parent_name("/dev/nvme0n1p1")
+
+    def test_rejects_non_block_device(self, fake_host):
+        fake_host.add_file("/dev/nvme0n1p1")
+        with pytest.raises(BlockPoolError, match="not a block device"):
+            validate_parent_name("/dev/nvme0n1p1")
+
+    def test_rejects_unknown_block_name(self, fake_host):
+        fake_host.add_file("/dev/foo")
+        with pytest.raises(BlockPoolError, match="block device or partition"):
+            validate_parent_name("/dev/foo")
+
+
+class TestBlockPool:
+    def test_allocate_and_reuse_stable_owner(self, enrolled_pool):
+        first_offset = enrolled_pool.get_or_allocate("ceph_node", 0, MiB)
+        second_offset = enrolled_pool.get_or_allocate("ceph_node", 1, MiB)
+        again_offset = enrolled_pool.get_or_allocate("ceph_node", 0, MiB)
+        assert (first_offset, second_offset, again_offset) == (0, MiB, 0)
+
+    def test_release_and_reuse_slice(self, enrolled_pool):
+        enrolled_pool.get_or_allocate("testnode_0", 0, MiB)
+        enrolled_pool.get_or_allocate("testnode_0", 1, MiB)
+        enrolled_pool.release_owner("testnode_0")
+        reused_offset = enrolled_pool.get_or_allocate("ceph_node", 0, MiB)
+        assert reused_offset == 0
+
+    def test_different_sizes_pack_sequentially(self, enrolled_pool):
+        first_offset = enrolled_pool.get_or_allocate("ceph_node", 0, MiB)
+        second_offset = enrolled_pool.get_or_allocate("testnode_0", 0, 2 * MiB)
+        assert (first_offset, second_offset) == (0, MiB)
+
+    def test_parent_change_rejected(self, tmp_path, fake_host, enrolled_pool):
+        other = fake_host.add_block_device("nvme1n1p1")
+        with pytest.raises(BlockPoolError, match="parent changed"):
+            BlockPool(tmp_path / "block_pool.json", other, allow_enroll=True)
+
+    def test_enroll_requires_allow_enroll(self, disallow_enroll_pool):
+        with pytest.raises(BlockPoolError, match="allow_enroll"):
+            disallow_enroll_pool.ensure_ready()
+
+    @pytest.mark.parametrize(
+        "cfg_allow_enroll,expected",
+        [
+            ({}, False),  # defaults to False when omitted
+            ({"allow_enroll": False}, False),
+            ({"allow_enroll": True}, True),
+        ],
+    )
+    def test_from_config_allow_enroll(
+        self, cfg_allow_enroll, expected, tmp_path, pool_parent
+    ):
+        pool = BlockPool.from_config(
+            {
+                "block_pool": {
+                    "parent": pool_parent,
+                    "state_dir": str(tmp_path),
+                    **cfg_allow_enroll,
+                }
+            }
+        )
+        assert pool is not None
+        assert pool.allow_enroll is expected
+
+    def test_state_dir_defaults_to_data_dir(self, tmp_path, pool_parent):
+        pool = BlockPool.from_config(
+            {"data_dir": str(tmp_path), "block_pool": {"parent": pool_parent}}
+        )
+        assert pool is not None
+        assert pool.state_path == tmp_path / "block_pool.json"
+
+    def test_shared_pool_is_built_once(self, tmp_path, pool_parent):
+        config["block_pool"] = {"parent": pool_parent, "state_dir": str(tmp_path)}
+        pool = shared_pool()
+        assert pool is not None and pool.parent == pool_parent
+        assert shared_pool() is pool
+
+    def test_enroll_rejects_non_empty_parent(self, fake_host, block_pool):
+        write(fake_host, block_pool.parent, 0, b"DATA")
+        with pytest.raises(BlockPoolError, match="not empty"):
+            block_pool.ensure_ready()
+
+    def test_enroll_writes_marker_and_state(self, block_pool):
+        block_pool.ensure_ready()
+        assert block_pool.enrolled
+        assert block_pool._read_tail_marker() == {
+            "pool_id": block_pool.pool_id,
+            "parent": block_pool.parent,
+        }
+
+    def test_new_slice_rejects_existing_data(self, fake_host, enrolled_pool):
+        write(fake_host, enrolled_pool.parent, 0, b"NOTZEROS")
+        with pytest.raises(BlockPoolError, match="not empty"):
+            enrolled_pool.get_or_allocate("ceph_node", 0, MiB)
+
+    def test_enroll_rejects_whole_disk_with_partitions(self, tmp_path, fake_host):
+        parent = fake_host.add_block_device("nvme0n1")
+        fake_host.add_block_device("nvme0n1p1", MiB, disk="nvme0n1")
+        pool = BlockPool(tmp_path / "block_pool.json", parent, allow_enroll=True)
+        with pytest.raises(BlockPoolError, match="existing partitions"):
+            pool.ensure_ready()
+
+    def test_reclaims_marker_after_state_loss(self, disallow_enroll_pool):
+        payload = {"pool_id": "saved-pool-id", "parent": disallow_enroll_pool.parent}
+        disallow_enroll_pool._write_tail_marker(payload)
+        disallow_enroll_pool.ensure_ready()
+        assert disallow_enroll_pool.enrolled
+        assert disallow_enroll_pool.pool_id == "saved-pool-id"
+
+    def test_reused_slice_must_be_empty(self, fake_host, enrolled_pool):
+        enrolled_pool.get_or_allocate("testnode_0", 0, MiB)
+        enrolled_pool.release_owner("testnode_0")
+        write(fake_host, enrolled_pool.parent, 0, b"STALE")
+        with pytest.raises(BlockPoolError, match="not empty"):
+            enrolled_pool.get_or_allocate("ceph_node", 0, MiB)
+
+    def test_released_region_is_zeroed_and_reusable(self, fake_host, enrolled_pool):
+        parent = enrolled_pool.parent
+        assert enrolled_pool.get_or_allocate("testnode_0", 0, MiB) == 0
+        write(fake_host, parent, 0, b"BLUESTORE LABEL")  # the owner used it
+        fake_host.commands.clear()
+        assert enrolled_pool.release_owner("testnode_0") == []
+        writes = [c for c in fake_host.commands if any("of=" in a for a in c)]
+        assert writes == [
+            ["dd", f"of={parent}", "bs=4096", "seek=0", "conv=notrunc,fsync"]
+        ]
+        assert bytes(fake_host.files[parent][:4096]) == bytes(4096)
+        assert enrolled_pool.get_or_allocate("ceph_node", 0, MiB) == 0
+
+    def test_release_zeroes_bluestore_label_copies(self, fake_host, enrolled_pool):
+        parent = enrolled_pool.parent
+        # Label copies at 256 KiB and 512 KiB fall inside a 1 MiB region; the
+        # one at 1 MiB is the next region's start and must not be touched.
+        labels = (0, 256 * 1024, 512 * 1024, MiB)
+        with patch("ceph_devstack.block_pool._BLUESTORE_LABEL_OFFSETS", labels):
+            assert enrolled_pool.get_or_allocate("testnode_0", 0, MiB) == 0
+            assert enrolled_pool.get_or_allocate("testnode_1", 0, MiB) == MiB
+            for label in labels:
+                write(fake_host, parent, label, b"bluestore block device")
+            fake_host.commands.clear()
+            assert enrolled_pool.release_owner("testnode_0") == []
+            seeks = [
+                a
+                for c in fake_host.commands
+                if any("of=" in a for a in c)
+                for a in c
+                if a.startswith("seek=")
+            ]
+            assert seeks == ["seek=0", "seek=64", "seek=128"]
+            for label in labels[:3]:
+                assert bytes(fake_host.files[parent][label : label + 4096]) == bytes(
+                    4096
+                )
+            assert fake_host.files[parent][MiB : MiB + 9] == b"bluestore"
+
+    def test_new_slice_rejects_bluestore_label_copy(self, fake_host, enrolled_pool):
+        with patch(
+            "ceph_devstack.block_pool._BLUESTORE_LABEL_OFFSETS", (0, 512 * 1024)
+        ):
+            write(fake_host, enrolled_pool.parent, 512 * 1024, b"bluestore")
+            with pytest.raises(BlockPoolError, match=f"at offset {512 * 1024}"):
+                enrolled_pool.get_or_allocate("ceph_node", 0, MiB)
+
+    def test_region_that_cannot_be_zeroed_stays_allocated(
+        self, fake_host, enrolled_pool
+    ):
+        enrolled_pool.get_or_allocate("testnode_0", 0, MiB)
+        with patch.object(
+            enrolled_pool, "_write_parent", side_effect=BlockPoolError("EIO")
+        ):
+            assert enrolled_pool.release_owner("testnode_0") == ["testnode_0:0"]
+        assert list(enrolled_pool.allocations) == ["testnode_0:0"]
+        assert enrolled_pool.free_regions == []
+
+    @pytest.mark.parametrize("marker", ["missing", "foreign", "unenrolled"])
+    def test_release_writes_nothing_unless_parent_is_ours(
+        self, fake_host, enrolled_pool, marker
+    ):
+        enrolled_pool.get_or_allocate("testnode_0", 0, MiB)
+        enrolled_pool.get_or_allocate("testnode_0", 1, MiB)
+        if marker == "missing":
+            write(fake_host, enrolled_pool.parent, TAIL, bytes(4096))
+        elif marker == "foreign":
+            enrolled_pool._write_tail_marker(
+                {"parent": enrolled_pool.parent, "pool_id": "other-pool"}
+            )
+        else:
+            state = json.loads(enrolled_pool.state_path.read_text())
+            state["enrolled"] = False
+            enrolled_pool.state_path.write_text(json.dumps(state))
+        write(fake_host, enrolled_pool.parent, 0, b"FOREIGN")
+        fake_host.commands.clear()
+        kept = enrolled_pool.release_owner("testnode_0")
+        assert kept == ["testnode_0:0", "testnode_0:1"]
+        assert not [c for c in fake_host.commands if any("of=" in a for a in c)]
+        assert bytes(fake_host.files[enrolled_pool.parent][:7]) == b"FOREIGN"
+        assert sorted(enrolled_pool.allocations) == kept
+
+
+class TestBlockPoolEnrollmentGuards:
+    def test_enroll_rejects_mounted_parent(self, fake_host, block_pool):
+        fake_host.responses["findmnt"] = (0, b"/mnt\n")
+        with pytest.raises(BlockPoolError, match="mounted"):
+            block_pool.ensure_ready()
+        assert [
+            "findmnt",
+            "-n",
+            "-o",
+            "TARGET",
+            "--source",
+            block_pool.parent,
+        ] in fake_host.commands
+
+    def test_enroll_rejects_blkid_signature(self, fake_host, block_pool):
+        fake_host.responses["blkid"] = (0, b"DEVNAME=x\nTYPE=xfs\n")
+        with pytest.raises(BlockPoolError, match="filesystem or partition"):
+            block_pool.ensure_ready()
+
+    def test_verify_rejects_missing_marker(self, fake_host, enrolled_pool):
+        write(fake_host, enrolled_pool.parent, TAIL, bytes(4096))
+        with pytest.raises(BlockPoolError, match="missing the devstack"):
+            enrolled_pool.ensure_ready()
+
+    def test_verify_rejects_foreign_pool_id(self, enrolled_pool):
+        enrolled_pool._write_tail_marker(
+            {"parent": enrolled_pool.parent, "pool_id": "other-pool"}
+        )
+        with pytest.raises(BlockPoolError, match="belongs to another pool"):
+            enrolled_pool.ensure_ready()
+
+    def test_region_exceeds_parent_size(self, enrolled_pool):
+        with pytest.raises(BlockPoolError, match="exceeds"):
+            enrolled_pool.get_or_allocate("ceph_node", 0, 32 * MiB)
+
+
+class TestHasPartitions:
+    def test_whole_disk_with_partitions(self, fake_host):
+        fake_host.add_block_device("nvme0n1")
+        fake_host.add_block_device("nvme0n1p1", disk="nvme0n1")
+        assert BlockPool._has_partitions("nvme0n1") is True
+
+    @pytest.mark.parametrize(
+        "disk,other_disk,other_partition",
+        [("nvme0n1", "nvme0n11", "nvme0n11p1"), ("sda", "sdaa", "sdaa1")],
+    )
+    def test_partitions_of_similarly_named_disks_do_not_count(
+        self, fake_host, disk, other_disk, other_partition
+    ):
+        fake_host.add_block_device(disk)
+        fake_host.add_block_device(other_disk)
+        fake_host.add_block_device(other_partition, disk=other_disk)
+        assert BlockPool._has_partitions(disk) is False
+
+    def test_partition_has_no_partitions(self):
+        assert BlockPool._has_partitions("nvme0n1p1") is False
+
+
+class TestTailMarker:
+    def test_roundtrip(self, block_pool):
+        payload = {"pool_id": "abc", "parent": block_pool.parent}
+        block_pool._write_tail_marker(payload)
+        assert block_pool._read_tail_marker() == payload
+
+    def test_corrupt_marker_returns_none(self, fake_host, block_pool):
+        write(
+            fake_host, block_pool.parent, TAIL, b"CEPH-DEVSTACK-BLOCK-POOL-TAIL-v1\x00x"
+        )
+        assert block_pool._read_tail_marker() is None
diff --git a/tests/test_cli.py b/tests/test_cli.py
new file mode 100644
index 00000000..e925f59e
--- /dev/null
+++ b/tests/test_cli.py
@@ -0,0 +1,49 @@
+import sys
+from unittest.mock import patch
+
+import pytest
+
+from ceph_devstack import config, parse_args
+from ceph_devstack.cli import block_pool_status, main
+
+
+class TestBlockPoolCLI:
+    def test_parse_args_block_pool_status(self):
+        args = parse_args(["block-pool", "status"])
+        assert args.command == "block-pool"
+        assert args.block_pool_op == "status"
+
+    @pytest.mark.parametrize("argv,rc", [(["status"], 0), ([], 2)])
+    def test_main_dispatches_block_pool_ops(self, argv, rc):
+        with (
+            patch.object(sys, "argv", ["ceph-devstack", "block-pool", *argv]),
+            patch.object(type(config), "load"),
+            patch("ceph_devstack.cli.block_pool_status", return_value=0) as status,
+        ):
+            assert main() == rc
+        assert status.call_count == (rc == 0)
+
+    def test_status_disabled(self, caplog):
+        with caplog.at_level("INFO", logger="ceph-devstack"):
+            assert block_pool_status() == 0
+        assert "disabled" in caplog.text
+
+    def test_status_invalid_parent_fails(self, caplog):
+        config["block_pool"] = {"parent": "/dev/dm-0"}
+        assert block_pool_status() == 1
+        assert "not allowed" in caplog.text
+
+    def test_status_shows_allocations(self, tmp_path, enrolled_pool, caplog):
+        enrolled_pool.get_or_allocate("ceph_node", 0, 1024**2)
+        enrolled_pool.get_or_allocate("testnode_0", 0, 1024**2)
+        enrolled_pool.release_owner("testnode_0")
+        config["block_pool"] = {
+            "parent": enrolled_pool.parent,
+            "state_dir": str(tmp_path),
+        }
+        with caplog.at_level("INFO", logger="ceph-devstack"):
+            assert block_pool_status() == 0
+        assert f"pool id: {enrolled_pool.pool_id}" in caplog.text
+        assert "ceph_node:0: offset 0, size 1M, last owner ceph_node" in caplog.text
+        assert "offset 1048576, size 1M" in caplog.text
+        assert "next offset: 2097152" in caplog.text
diff --git a/tests/test_config.py b/tests/test_config.py
index 5d11c1dc..b399c430 100644
--- a/tests/test_config.py
+++ b/tests/test_config.py
@@ -104,6 +104,7 @@ class TestConfigDefaults:
     def test_config_defaults(self):
         assert config == {
             "data_dir": "~/.local/share/ceph-devstack",
+            "block_pool": {},
             "containers": {
                 "archive": {"image": "python:alpine"},
                 "beanstalk": {"image": "quay.io/ceph-infra/teuthology-beanstalkd:main"},
@@ -114,6 +115,7 @@ def test_config_defaults(self):
                 "pulpito": {"image": "quay.io/ceph-infra/pulpito:main"},
                 "testnode": {
                     "count": 3,
+                    "loop_device_count": 1,
                     "loop_device_size": "5G",
                     "image": "quay.io/ceph-infra/teuthology-testnode:main",
                 },
diff --git a/tests/test_host.py b/tests/test_host.py
index 6588576e..246d09eb 100644
--- a/tests/test_host.py
+++ b/tests/test_host.py
@@ -1,3 +1,7 @@
+import os
+import subprocess
+from unittest.mock import MagicMock, patch
+
 import pytest
 
 from ceph_devstack.host import LocalHost, RemoteHost
@@ -12,11 +16,64 @@ def test_path_exists_expands_tilde(self, host):
         assert host.path_exists("~") is True
         assert host.path_exists("~/nonexistent_dir_12345") is False
 
+    def test_file_helpers_run_commands_on_host(self, host, tmp_path):
+        (tmp_path / "a").write_text("hello\n")
+        (tmp_path / "link").symlink_to(tmp_path / "a")
+        assert host.read_text(tmp_path / "a") == "hello\n"
+        assert host.read_text(tmp_path / "missing") is None
+        assert host.list_dir(tmp_path) == ["a", "link"]
+        assert host.list_dir(tmp_path / "missing") == []
+        assert host.realpath(tmp_path / "link") == os.path.realpath(tmp_path / "a")
+
+    def test_byte_io_at_unaligned_offsets(self, host, tmp_path):
+        disk = tmp_path / "disk"
+        disk.write_bytes(bytes(8192))
+        host.write_bytes(disk, 4096 + 256, b"x" * 512)
+        data = disk.read_bytes()
+        assert len(data) == 8192
+        assert data[4352:4864] == b"x" * 512 and data.count(b"x") == 512
+        assert host.read_bytes(disk, 4096, 1024) == bytes(256) + b"x" * 512 + bytes(256)
+
+    @pytest.mark.parametrize("stdout,mounted", [(b"/mnt\n", True), (b"", False)])
+    def test_device_mounted_asks_findmnt(self, host, stdout, mounted):
+        done = subprocess.CompletedProcess([], 0 if stdout else 1, stdout, b"")
+        with patch.object(host, "output", return_value=done) as output:
+            assert host.device_mounted("/dev/loop3") is mounted
+        output.assert_called_once_with(
+            ["findmnt", "-n", "-o", "TARGET", "--source", "/dev/loop3"]
+        )
+
+    def test_user_ids_returns_os_ids(self, host):
+        uid, gid = host.user_ids()
+        assert uid == os.getuid()
+        assert gid == os.getgid()
+        assert isinstance(uid, int)
+        assert isinstance(gid, int)
+
 
 SSH = ["podman", "machine", "ssh", "--"]
 
 
 class TestRemoteHost:
+    def test_write_bytes_pipes_data_through_machine_ssh(self):
+        remote = RemoteHost()
+        done = MagicMock(returncode=0)
+        with patch("ceph_devstack.exec.subprocess.run", return_value=done) as run:
+            remote.write_bytes("/dev/sda1", 8192, b"m" * 4096)
+        args, kwargs = run.call_args
+        assert args[0] == [
+            "podman",
+            "machine",
+            "ssh",
+            "--",
+            "dd",
+            "of=/dev/sda1",
+            "bs=4096",
+            "seek=2",
+            "conv=notrunc,fsync",
+        ]
+        assert kwargs["input"] == b"m" * 4096
+
     @pytest.mark.parametrize(
         "args,expected",
         [
@@ -30,3 +87,27 @@ class TestRemoteHost:
     )
     def test_cmd_runs_non_podman_commands_in_machine(self, args, expected):
         assert RemoteHost().cmd(args).args == expected
+
+    def test_user_ids_parses_and_caches(self):
+        remote = RemoteHost()
+        outputs = {"-u": b"501\n", "-g": b"1000\n"}
+
+        def output(args):
+            return subprocess.CompletedProcess(args, 0, outputs[args[1]], b"")
+
+        with patch.object(remote, "output", side_effect=output) as mock_output:
+            assert remote.user_ids() == (501, 1000)
+            assert remote.user_ids() == (501, 1000)
+        assert [c.args[0] for c in mock_output.call_args_list] == [
+            ["id", "-u"],
+            ["id", "-g"],
+        ]
+
+    def test_user_ids_raises_on_invalid_output(self):
+        remote = RemoteHost()
+        bad = subprocess.CompletedProcess([], 0, b"not-a-number\n", b"")
+        with (
+            patch.object(remote, "output", return_value=bad),
+            pytest.raises(RuntimeError, match="'id -u'"),
+        ):
+            remote.user_ids()