devices/loopback: add read-only option

It's sometimes useful to set up a loop device for an already formatted
disk/filesystem image to derive new artifacts from it. In that case, we
want to make sure it's impossible to modify its contents in any way in
that process, both for our own purposes and for other stages operating
on it.

Notably, mounting some filesystems read-only still seem to touch the
disk (like XFS).
This commit is contained in:
Jonathan Lebon 2024-08-13 11:57:59 -04:00 committed by Michael Vogt
parent 478fee2876
commit 3c3be92016
2 changed files with 25 additions and 7 deletions

View file

@ -58,6 +58,10 @@ SCHEMA = """
"lock": {
"type": "boolean",
"description": "Lock the device after opening it"
},
"read-only": {
"type": "boolean",
"description": "Set up the device as read-only"
}
}
"""
@ -77,7 +81,7 @@ class LoopbackService(devices.DeviceService):
lock = UdevInhibitor.for_device(lo.LOOP_MAJOR, lo.minor)
lo.on_close = lambda _l: lock.release()
def make_loop(self, fd: int, offset, sizelimit, lock, partscan):
def make_loop(self, fd: int, offset, sizelimit, lock, partscan, read_only):
if not sizelimit:
sizelimit = os.fstat(fd).st_size - offset
else:
@ -89,6 +93,7 @@ class LoopbackService(devices.DeviceService):
sizelimit=sizelimit,
blocksize=self.sector_size,
partscan=partscan,
read_only=read_only,
autoclear=True)
return lo
@ -100,12 +105,13 @@ class LoopbackService(devices.DeviceService):
size = options.get("size")
lock = options.get("lock", False)
partscan = options.get("partscan", False)
read_only = options.get("read-only", False)
path = os.path.join(tree, filename.lstrip("/"))
self.fd = os.open(path, os.O_RDWR | os.O_CLOEXEC)
try:
self.lo = self.make_loop(self.fd, start, size, lock, partscan)
self.lo = self.make_loop(self.fd, start, size, lock, partscan, read_only)
except Exception as error: # pylint: disable: broad-except
self.close()
raise error from None

View file

@ -317,7 +317,7 @@ class Loop:
# it is bound, check if it is bound by `fd`
return loop_info.is_bound_to(file_info)
def _config_info(self, info, offset, sizelimit, autoclear, partscan):
def _config_info(self, info, offset, sizelimit, autoclear, partscan, read_only):
# pylint: disable=attribute-defined-outside-init
if offset:
info.lo_offset = offset
@ -333,9 +333,14 @@ class Loop:
info.lo_flags |= self.LO_FLAGS_PARTSCAN
else:
info.lo_flags &= ~self.LO_FLAGS_PARTSCAN
if read_only is not None:
if read_only:
info.lo_flags |= self.LO_FLAGS_READ_ONLY
else:
info.lo_flags &= ~self.LO_FLAGS_READ_ONLY
return info
def set_status(self, offset=None, sizelimit=None, autoclear=None, partscan=None):
def set_status(self, offset=None, sizelimit=None, autoclear=None, partscan=None, read_only=None):
"""Set properties of the loopback device
The loopback device must be bound, and the properties will be
@ -371,12 +376,16 @@ class Loop:
partscan : bool, optional
Whether or not to enable partition scanning, or None to leave
unchanged (default is None)
read_only : bool, optional
Whether or not to setup the loopback device as read-only (default
is None).
"""
info = self._config_info(self.get_status(), offset, sizelimit, autoclear, partscan)
info = self._config_info(self.get_status(), offset, sizelimit, autoclear, partscan, read_only)
fcntl.ioctl(self.fd, self.LOOP_SET_STATUS64, info)
def configure(self, fd: int, offset=None, sizelimit=None, blocksize=0, autoclear=None, partscan=None):
def configure(self, fd: int, offset=None, sizelimit=None, blocksize=0, autoclear=None, partscan=None,
read_only=None):
"""
Configure the loopback device
Bind and configure in a single operation a file descriptor to the
@ -426,12 +435,15 @@ class Loop:
partscan : bool, optional
Whether or not to enable partition scanning, or None to leave
unchanged (default is None)
read_only : bool, optional
Whether or not to setup the loopback device as read-only (default
is None).
"""
# pylint: disable=attribute-defined-outside-init
config = LoopConfig()
config.fd = fd
config.block_size = int(blocksize)
config.info = self._config_info(LoopInfo(), offset, sizelimit, autoclear, partscan)
config.info = self._config_info(LoopInfo(), offset, sizelimit, autoclear, partscan, read_only)
try:
fcntl.ioctl(self.fd, self.LOOP_CONFIGURE, config)
except OSError as e: