in export/securedrop_export/disk/cli.py [0:0]
def _mount_volume(self, volume: Volume, full_unlocked_name: str) -> MountedVolume:
"""
Given an unlocked volume, mount volume in /media/user/ by udisksctl and
return MountedVolume object.
Unlocked name could be `/dev/mapper/$id` or `/dev/dm-X`.
Raise ExportException if errors are encountered during mounting.
`pexpect.ExeptionPexpect` can't be try/caught, since it's not a
child of BaseException, but instead, exceptions can be included
in the list of results to check for. (See
https://pexpect.readthedocs.io/en/stable/api/pexpect.html#pexpect.spawn.expect)
"""
info_cmd = "udisksctl"
info_args = ["info", "--block-device", quote(volume.device_name)]
# The terminal output has colours and other formatting. A match is anything
# that includes our device identified as PreferredDevice on one line
# \x1b[37mPreferredDevice:\x1b[0m /dev/sdaX\r\n
expected_info: PexpectList = [
f"PreferredDevice:.*[^\r\n]{volume.device_name}",
"Error looking up object for device",
pexpect.EOF,
pexpect.TIMEOUT,
]
max_retries = 3
mount_cmd = "udisksctl"
mount_args = ["mount", "--block-device", quote(full_unlocked_name)]
# We can't pass {full_unlocked_name} in the match statement since even if we
# pass in /dev/mapper/xxx, udisks2 may refer to the disk as /dev/dm-X.
expected_mount: PexpectList = [
"Mounted .* at (.*)",
"Error mounting .*: GDBus.Error:org.freedesktop.UDisks2.Error.AlreadyMounted: "
"Device (.*) is already mounted at `(.*)'.",
"Error looking up object for device",
pexpect.EOF,
pexpect.TIMEOUT,
]
mountpoint = None
logger.debug(
f"Check to make sure udisks identified {volume.device_name} "
f"(unlocked as {full_unlocked_name})"
)
for _ in range(max_retries):
child = pexpect.spawn(info_cmd, info_args)
index = child.expect(expected_info)
child.close()
if index != 0:
logger.debug(f"udisks can't identify {volume.device_name}, retrying...")
time.sleep(0.5)
else:
logger.debug(f"udisks found {volume.device_name}")
break
logger.info(f"Mount {full_unlocked_name} using udisksctl")
child = pexpect.spawn(mount_cmd, mount_args)
index = child.expect(expected_mount)
if index == 0:
# As above, we know the format.
# Per https://pexpect.readthedocs.io/en/stable/api/pexpect.html#pexpect.spawn.expect,
# `child.match` is a re.Match object
mountpoint = child.match.group(1).decode("utf-8").strip() # type: ignore
logger.info(f"Successfully mounted device at {mountpoint}")
elif index == 1:
# Use udisks unlocked name
logger.debug("Already mounted, get unlocked_name and mountpoint")
full_unlocked_name = child.match.group(1).decode("utf-8").strip() # type: ignore
mountpoint = child.match.group(2).decode("utf-8").strip() # type: ignore
logger.info(f"Device {full_unlocked_name} already mounted at {mountpoint}")
elif index == 2:
logger.debug("Device is not ready")
logger.debug("Close pexpect process")
child.close()
if mountpoint:
return MountedVolume(
device_name=volume.device_name,
unlocked_name=full_unlocked_name,
encryption=volume.encryption,
mountpoint=mountpoint,
)
logger.error("Could not get mountpoint")
raise ExportException(sdstatus=Status.ERROR_MOUNT)