Linux: enforce read-only protection on native mount devices

Pass --readonly to dmsetup for every cipher layer using the opened volume's protection, including access-error fallback. Use the same protection for the container loop and filesystem, and keep hidden-volume protection on FUSE.

Remove the losetup retry that silently drops -r. Add regression tests linked to the production backend and disposable-VM coverage for file and virtual-device hosts, cascades, filesystem-free mounts, writable mounts, FUSE, and failed setup cleanup.
This commit is contained in:
Mounir IDRASSI
2026-10-22 10:54:22 +09:00
parent 13bd629733
commit 5ef04ad2d0
4 changed files with 417 additions and 19 deletions
+172
View File
@@ -0,0 +1,172 @@
// Exercise the linked Linux backend with command execution intercepted. No mounts are made.
#include "Core/Unix/Linux/CoreLinux.h"
#include "Platform/Unix/Process.h"
#include "Volume/EncryptionModeXTS.h"
#include <algorithm>
#include <iostream>
using namespace VeraCrypt;
static bool ReadOnly;
static unsigned LoopCalls, LoopFailures, Layers, ExpectedLayers;
static string VolumeFilePath;
struct StopBeforeDeviceIO { };
static void Require (bool condition, const char *message)
{
if (!condition) throw std::runtime_error (message);
}
// The Python runner resolves the mangled names from the matching build archives.
extern "C" string WrapExecute (const string &program, const list<string> &args,
int, ProcessExecFunctor *, const Buffer *input)
{
vector<string> a (args.begin(), args.end());
if (program == "modprobe") return "";
if (program == "losetup")
{
if (a == vector<string> {"-f"}) return "";
++LoopCalls;
vector<string> expected;
if (ReadOnly) expected.push_back ("-r");
expected.push_back ("--");
expected.push_back ("/dev/loop" + StringConverter::ToSingle (LoopCalls - 1));
expected.push_back (VolumeFilePath);
Require (a == expected, "loop retry changed protection or arguments");
if (LoopCalls <= LoopFailures)
throw ExecutedProcessFailed (SRC_POS, "losetup", 1, "injected failure");
return "";
}
Require (program == "dmsetup" && !a.empty() && a[0] == "create", "unexpected command");
string name = "veracrypt64";
if (Layers + 1 != ExpectedLayers)
name += "_" + StringConverter::ToSingle (ExpectedLayers - Layers - 2);
vector<string> expected {"create"};
if (ReadOnly) expected.push_back ("--readonly");
expected.push_back ("--uuid");
expected.push_back ("VERACRYPT-987654321-" + name);
expected.push_back (name);
Require (a == expected, "device-mapper access mode, name or UUID changed");
Require (input && input->Size() > 0, "missing device-mapper table");
if (++Layers == ExpectedLayers) throw StopBeforeDeviceIO();
return "";
}
extern "C" string WrapFindBinary (const char *, string &) { return "/test/dmsetup"; }
extern "C" FilesystemPathType::Enum WrapGetType (const FilesystemPath *)
{
return FilesystemPathType::BlockDevice;
}
extern "C" shared_ptr<VolumeInfo> WrapAuxInfo (const DirectoryPath &)
{
shared_ptr<VolumeInfo> info (new VolumeInfo);
info->SerialInstanceNumber = 987654321;
return info;
}
struct TestLayout : VolumeLayoutV2Normal
{
uint64 GetDataOffset (uint64) const override { return 128 * 1024; }
};
struct TestVolume : Volume
{
TestVolume (bool cascade, VolumeProtection::Enum protection, bool incomplete = false)
{
Protection = protection;
EncryptionNotCompleted = incomplete;
VolumeFile.reset (new File);
VolumeFile->Open (VolumeFilePath, File::OpenRead);
EA.reset (cascade ? static_cast<VeraCrypt::EncryptionAlgorithm *> (new SerpentTwofishAES) : new VeraCrypt::AES);
SecureBuffer key (EA->GetKeySize());
key.Zero();
EA->SetKey (key);
shared_ptr<EncryptionMode> mode (new EncryptionModeXTS);
mode->SetKey (key);
EA->SetMode (mode);
Layout.reset (new TestLayout);
VolumeHostSize = 16 * 1024 * 1024;
VolumeDataSize = VolumeHostSize - 256 * 1024;
SectorSize = 512;
}
};
struct TestCore : CoreLinux
{
using CoreLinux::AttachFileToLoopDevice;
using CoreLinux::MountVolumeNative;
using CoreUnix::MountAuxVolumeImage;
};
int main (int argc, char **argv)
{
try
{
Require (argc == 2, "expected temporary file");
VolumeFilePath = argv[1];
TestCore core;
for (bool readOnly : {false, true})
{
ReadOnly = readOnly;
for (unsigned failures : {0u, 2u, 256u})
{
LoopCalls = 0;
LoopFailures = failures;
try
{
core.AttachFileToLoopDevice (VolumeFilePath, readOnly);
Require (failures < 256, "failed loop setup unexpectedly succeeded");
}
catch (LoopDeviceSetupFailed &) { Require (failures == 256, "early loop failure"); }
Require (LoopCalls == std::min (failures + 1, 256u), "unexpected loop attempts");
}
for (bool cascade : {false, true})
for (bool noFilesystem : {false, true})
{
Layers = 0;
ExpectedLayers = cascade ? 3 : 1;
shared_ptr<Volume> volume (new TestVolume (cascade,
readOnly ? VolumeProtection::ReadOnly : VolumeProtection::None));
MountOptions options;
options.SlotNumber = 64;
options.NoFilesystem = noFilesystem;
// Deliberately differ: the opened volume is authoritative.
options.Protection = readOnly ? VolumeProtection::None : VolumeProtection::ReadOnly;
try { core.MountVolumeNative (volume, options, "/test/aux"); }
catch (StopBeforeDeviceIO &) { }
Require (Layers == ExpectedLayers, "cascade skipped a layer");
}
}
MountOptions options;
for (bool incomplete : {false, true})
{
options.Protection = incomplete ? VolumeProtection::None : VolumeProtection::ReadOnly;
shared_ptr<Volume> protectedVolume (new TestVolume (false,
incomplete ? VolumeProtection::ReadOnly : VolumeProtection::HiddenVolumeReadOnly, incomplete));
Layers = 0;
try
{
core.MountVolumeNative (protectedVolume, options, "/test/aux");
throw std::runtime_error ("protected volume selected native backend");
}
catch (NotApplicable &) { }
Require (Layers == 0, "FUSE fallback created a native mapping");
}
// The FUSE image caller must propagate loop setup failure as well.
options.Protection = VolumeProtection::ReadOnly;
VolumeFilePath = "/test/aux/volume";
LoopCalls = 0;
LoopFailures = 256;
ReadOnly = true;
try
{
core.MountAuxVolumeImage ("/test/aux", options);
throw std::runtime_error ("FUSE loop setup unexpectedly succeeded");
}
catch (LoopDeviceSetupFailed &) { }
Require (LoopCalls == 256, "FUSE loop failure not exercised");
std::cout << "PASS mapping arguments, cascades, effective protection, hidden/incomplete fallback, loop retries and failure\n";
}
catch (const std::exception &e) { std::cerr << e.what() << '\n'; return 1; }
}
+65
View File
@@ -0,0 +1,65 @@
#!/usr/bin/env python3
"""Test the production Linux mount commands without root or devices.
Run after a matching Linux build:
python3 Tests/test_linux_mount_read_only.py --build-dir src
Links CoreLinux/CoreUnix from that build and intercepts external commands,
path type queries and auxiliary metadata using GNU ld --wrap. The native
tests stop before device I/O; real kernel access modes and cleanup require
the separate VM integration tests.
"""
import argparse
import os
from pathlib import Path
import shlex
import subprocess
import tempfile
ROOT = Path(__file__).resolve().parents[1]
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--build-dir", type=Path, default=ROOT / "src")
parser.add_argument("--fuse-package", choices=("fuse", "fuse3"), default="fuse3")
args = parser.parse_args()
archives = [args.build_dir.resolve() / name for name in (
"Core/Core.a", "Driver/Fuse/Driver.a", "Volume/Volume.a", "Platform/Platform.a")]
symbols = subprocess.check_output(["nm", *map(str, archives)], text=True)
wrappers = {
"Process7ExecuteE": "WrapExecute",
"Process16FindSystemBinaryE": "WrapFindBinary",
"FilesystemPath7GetTypeE": "WrapGetType",
"CoreUnix23ReadAuxiliaryVolumeInfoE": "WrapAuxInfo",
}
flags = []
for fragment, wrapper in wrappers.items():
matches = [line.split()[-1] for line in symbols.splitlines()
if " T " in line and fragment in line]
if len(matches) != 1:
raise RuntimeError(f"Expected one definition for {fragment}: {matches}")
symbol = matches[0]
flags += [f"-Wl,--wrap={symbol}", f"-Wl,--defsym=__wrap_{symbol}={wrapper}"]
with tempfile.TemporaryDirectory(prefix="vc-read-only-") as directory:
work = Path(directory)
binary = work / "mount-test"
volume = work / "volume"
volume.touch()
command = shlex.split(os.environ.get("CXX", "c++")) + [
"-std=c++11", "-DTC_UNIX", "-DTC_LINUX", "-D_FILE_OFFSET_BITS=64",
"-I" + str(ROOT / "src"), "-I" + str(ROOT / "src/Core/Unix/Linux"),
"-I" + str(ROOT / "src/Crypto"), "-I" + str(ROOT / "src/Crypto/Argon2/include"),
str(ROOT / "Tests/linux_mount_read_only_test.cpp"), *flags,
"-Wl,--start-group", *map(str, archives), "-Wl,--end-group",
"-pthread", "-ldl", "-o", str(binary),
]
command += shlex.split(subprocess.check_output(
["pkg-config", "--libs", args.fuse_package], text=True))
subprocess.run(command, check=True, timeout=120)
subprocess.run([str(binary), str(volume)], check=True, timeout=30)
if __name__ == "__main__":
main()
+173
View File
@@ -0,0 +1,173 @@
#!/usr/bin/env python3
"""Verify Linux read-only device protection in a dedicated disposable VM.
Requires the same tools as test_linux_loop_identity.py, plus blockdev. Run:
python3 Tests/test_linux_read_only_integration.py --binary /usr/bin/veracrypt
To also test a virtual block device, attach a blank 16 MiB virtio disk with
serial=vc-readonly-test and pass --disposable-device /dev/vdb. That device is
overwritten with test containers. Never run this suite against user storage.
Use test_linux_mount_read_only.py for command/failure injection without root.
"""
import argparse
import os
from pathlib import Path
import shutil
import shlex
import sys
from test_linux_loop_identity import (
LoopIdentityChecks, TOOLS, aux_mounts, dm_devices, isolate_mounts, loop_devices,
)
def check_setup_failure(checks, program, kernel):
label = f"{program}-failure-{'native' if kernel else 'fuse'}"
real = checks.root / (program + ".real")
wrapper = checks.root / (program + ".sh")
calls = checks.root / (label + ".args")
count = checks.root / (label + ".count")
target = shutil.which(program)
shutil.copy2(target, real)
count.write_text("0\n")
script = "#!/bin/sh\n"
script += f"printf '%s\\n' \"$*\" >> {shlex.quote(str(calls))}\n"
if program == "losetup":
# Fail the requested read-only setup. Any weakened retry would
# actually succeed, and is caught by both the status and argument checks.
script += 'case "$1" in -r) exit 1;; esac\n'
else:
script += ('if [ "$1" = create ]; then\n'
f' n=$(cat {shlex.quote(str(count))})\n'
' n=$((n + 1))\n'
f' echo "$n" > {shlex.quote(str(count))}\n'
' [ "$n" != 2 ] || exit 1\nfi\n')
script += f'exec {shlex.quote(str(real))} "$@"\n'
wrapper.write_text(script)
wrapper.chmod(0o755)
checks.run(label + "-inject", ["mount", "--bind", wrapper, target])
try:
path = checks.cascade if program == "dmsetup" else checks.volume
result = checks.vc(label, "--mount", path, checks.mountpoint, f"--slot={checks.slot}",
"--password=" + checks.password, "--pim=1", "--keyfiles=",
"--protect-hidden=no", "--filesystem=none",
"--mount-options=ro" + ("" if kernel else ",nokernelcrypto"), expected=None)
if result.returncode == 0:
checks.mounted = path
checks.expect(label + "-fails", result.returncode != 0, f"exit {result.returncode}")
arguments = calls.read_text().splitlines()
if program == "losetup":
attachments = [line for line in arguments if line != "-f"]
checks.expect(label + "-no-downgrade", len(attachments) == 256 and
all(line.startswith("-r -- ") for line in attachments),
f"{len(attachments)} attachment attempts, all must retain -r")
else:
creates = [line for line in arguments if line.startswith("create ")]
checks.expect(label + "-partial-cascade", len(creates) == 2 and
all("--readonly" in line.split() for line in creates), str(creates))
checks.expect(label + "-released", *checks.released())
finally:
checks.run(label + "-restore", ["umount", target])
checks.cleanup()
def check(checks, path, kernel, readonly, filesystem, layers, effective_readonly=None):
label = f"{path.name}-{'native' if kernel else 'fuse'}-{'ro' if readonly else 'rw'}-{'fs' if filesystem else 'none'}"
expected_readonly = readonly if effective_readonly is None else effective_readonly
if expected_readonly != readonly:
label += "-access-fallback"
before_aux = aux_mounts()
before_loops = loop_devices()
options = ",".join((["ro"] if readonly else []) + ([] if kernel else ["nokernelcrypto"]))
args = ["--mount", path, checks.mountpoint, f"--slot={checks.slot}",
"--password=" + checks.password, "--pim=1", "--keyfiles=", "--protect-hidden=no",
"--mount-options=" + options]
if not filesystem:
args += ["--filesystem=none"]
try:
checks.vc(label, *args)
checks.mounted = path
listed = checks.listed()
device = listed and listed["device"]
if not device or device.startswith("/dev/mapper/") != kernel:
raise AssertionError(f"Wrong backend: {listed}")
names = {name for name in dm_devices() if name == checks.dm_name or name.startswith(checks.dm_name + "_")}
expected = {checks.dm_name} | {f"{checks.dm_name}_{i}" for i in range(layers - 1)} if kernel else set()
checks.expect(label + "-layers", names == expected, str(sorted(names)))
devices = ["/dev/mapper/" + name for name in sorted(names)] if kernel else [device]
new_loops = set(loop_devices()) - set(before_loops)
if kernel and str(path).startswith("/dev/"):
checks.expect(label + "-no-loop", not new_loops, str(new_loops))
else:
checks.expect(label + "-loop", len(new_loops) == 1, str(new_loops))
for index, dev in enumerate(sorted(set(devices) | new_loops)):
mode = checks.run(f"{label}-mode-{index}", ["blockdev", "--getro", dev]).stdout.strip()
checks.expect(f"{label}-protected-{index}", mode == str(int(expected_readonly)), f"{dev}: read-only={mode}")
if filesystem:
flags = next(line.split()[3].split(",") for line in Path("/proc/mounts").read_text().splitlines()
if line.split()[1] == str(checks.mountpoint))
checks.expect(label + "-filesystem", ("ro" if expected_readonly else "rw") in flags, str(flags))
if not expected_readonly:
marker = checks.mountpoint / "writable-check"
marker.write_text(label)
checks.expect(label + "-write", marker.read_text() == label, "ordinary file write succeeded")
marker.unlink()
checks.vc(label + "-unmount", "--unmount", path)
checks.mounted = None
checks.expect(label + "-released", *checks.released())
checks.expect(label + "-aux-released", aux_mounts() == before_aux, "auxiliary mount released")
finally:
checks.cleanup()
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--binary", required=True)
parser.add_argument("--disposable-device", type=Path)
parser.add_argument("--slot", type=int, default=61)
args = parser.parse_args()
args.legacy_binary = None
if os.geteuid() != 0:
parser.error("Run as root in a dedicated disposable VM")
missing = [tool for tool in (*TOOLS, "blockdev") if not shutil.which(tool)]
if missing:
parser.error("Missing tools: " + ", ".join(missing))
if args.disposable_device:
serial = Path("/sys/class/block", args.disposable_device.name, "serial")
if not args.disposable_device.is_block_device() or serial.read_text().strip() != "vc-readonly-test":
parser.error("The disposable device must have virtio serial vc-readonly-test")
isolate_mounts()
checks = LoopIdentityChecks(args)
if checks.listed() or any(n.startswith(checks.dm_name) for n in dm_devices()):
parser.error("Test slot is already in use")
for path, cipher, layers in ((checks.volume, "AES", 1), (checks.cascade, "Serpent-Twofish-AES", 3)):
checks.create("create-" + path.name, path, cipher)
paths = [path]
if args.disposable_device:
checks.run("copy-" + path.name, ["dd", "if=" + str(path), "of=" + str(args.disposable_device), "bs=1M", "conv=fsync"])
paths.append(args.disposable_device)
for source in paths:
for kernel in (True, False):
for readonly in (True, False):
for filesystem in (False, True):
check(checks, source, kernel, readonly, filesystem, layers)
# A read-only host mount makes the initial O_RDWR open fail with
# EROFS. A block device's read-only flag alone need not reject open().
checks.run("host-bind", ["mount", "--bind", path, path])
try:
checks.run("host-readonly", ["mount", "-o", "remount,bind,ro", path])
for kernel in (True, False):
check(checks, path, kernel, False, True, layers, effective_readonly=True)
finally:
checks.run("host-unbind", ["umount", path])
for kernel in (True, False):
check_setup_failure(checks, "losetup", kernel)
check_setup_failure(checks, "dmsetup", True)
if checks.failures:
sys.exit("Failed: " + ", ".join(checks.failures))
print("All read-only integration checks passed")
if __name__ == "__main__":
main()
+7 -19
View File
@@ -649,12 +649,8 @@ namespace VeraCrypt
list <string> args;
list <string>::iterator readOnlyArg;
if (readOnly)
{
args.push_back ("-r");
readOnlyArg = --args.end();
}
args.push_back ("--");
args.push_back (loopDev);
@@ -665,19 +661,7 @@ namespace VeraCrypt
Process::Execute ("losetup", args);
return loopDev;
}
catch (ExecutedProcessFailed&)
{
if (readOnly)
{
try
{
args.erase (readOnlyArg);
Process::Execute ("losetup", args);
return loopDev;
}
catch (ExecutedProcessFailed&) { }
}
}
catch (ExecutedProcessFailed&) { }
}
throw LoopDeviceSetupFailed (SRC_POS, wstring (filePath));
@@ -1036,6 +1020,8 @@ namespace VeraCrypt
void CoreLinux::MountVolumeNative (shared_ptr <Volume> volume, MountOptions &options, const DirectoryPath &auxMountPoint) const
{
// Use the protection of the opened volume, including a read-only open after an access error.
const bool readOnly = volume->GetProtectionType() == VolumeProtection::ReadOnly;
bool xts = (typeid (*volume->GetEncryptionMode()) ==
#ifdef WOLFCRYPT_BACKEND
typeid (EncryptionModeWolfCryptXTS));
@@ -1107,7 +1093,7 @@ namespace VeraCrypt
struct stat containerStat;
throw_sys_sub_if (stat (containerPath.c_str(), &containerStat) != 0, containerPath);
volumePath = AttachFileToLoopDevice (volumePath, options.Protection == VolumeProtection::ReadOnly);
volumePath = AttachFileToLoopDevice (volumePath, readOnly);
loopDevAttached = true;
const string loopDevice = volumePath;
@@ -1194,6 +1180,8 @@ namespace VeraCrypt
execArgs.clear();
execArgs.push_back ("create");
if (readOnly)
execArgs.push_back ("--readonly");
execArgs.push_back ("--uuid");
execArgs.push_back (GetDeviceMapperUuid (serialInstanceNumber, nativeDevName.str()));
execArgs.push_back (nativeDevName.str());
@@ -1248,7 +1236,7 @@ namespace VeraCrypt
MountFilesystemWithFallback (nativeDevPath, *options.MountPoint,
StringConverter::ToSingle (filesystemType),
allowFilesystemTypeFallback,
options.Protection == VolumeProtection::ReadOnly,
readOnly,
StringConverter::ToSingle (options.FilesystemOptions),
internalMountOnly);