mirror of
https://github.com/veracrypt/VeraCrypt.git
synced 2026-10-05 16:46:35 -05:00
Mounir IDRASSI
41bc8e5f6a
Create a fresh private parent for each FUSE-T auxiliary mount. Elevated parents stay root-owned and grant only the original user read/search access, restricting access through the auxiliary mount path and preventing caller-controlled path replacement during elevated setup. Clear inherited ACLs and reject temporary filesystems that ignore ownership. Remove per-mount parents on setup failure and service teardown, while preserving legacy temporary directories and discovery behavior. Add regression coverage for permissions, unique parent creation, rollback, and cleanup. FUSE-T transport authentication is outside this change.
266 lines
13 KiB
Python
266 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""Test FUSE-T's production private-parent helpers using disposable directories.
|
|
|
|
Requires a matching macOS build, but no mounted volumes or FUSE service. Optional
|
|
--owner UID, when run as root, checks elevated access and real uid isolation
|
|
using a harmless sentinel file. No user accounts are created or modified.
|
|
"""
|
|
import argparse
|
|
import concurrent.futures
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import stat
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def function(source, signature):
|
|
begin = source.index("\t" + signature)
|
|
return source[begin:source.index("\n\t}", begin) + 3]
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--build-dir", type=Path, required=True)
|
|
parser.add_argument("--owner", type=int, default=os.getuid())
|
|
args = parser.parse_args()
|
|
if sys.platform != "darwin":
|
|
parser.error("macOS is required")
|
|
if args.owner != os.getuid() and os.geteuid() != 0:
|
|
parser.error("--owner requires root when it differs from the current uid")
|
|
owner = args.owner
|
|
elevated = os.geteuid() == 0
|
|
prefix = ".veracrypt_aux_root_" if elevated else ".veracrypt_aux_"
|
|
other = 502 if owner != 502 else 503
|
|
with tempfile.TemporaryDirectory(prefix="vc-aux-directory-", dir="/private/tmp") as temporary:
|
|
work = Path(temporary)
|
|
work.chmod(0o755)
|
|
source = (ROOT / "src/Core/Unix/CoreUnix.cpp").read_text()
|
|
helpers = function(source, "static string CreateFuseTAuxiliaryDirectory")
|
|
helpers += function(source, "static bool IsOtherUsersFuseTAuxiliaryMount")
|
|
service = (ROOT / "src/Driver/Fuse/FuseService.cpp").read_text()
|
|
helpers += function(service, "void FuseService::RemoveAuxMountParent")
|
|
begin = service.index("\tclass FuseServiceAuxDirectory")
|
|
helpers += service[begin:service.index("\n\t};", begin) + 4]
|
|
unit = work / "check.cpp"
|
|
unit.write_text(r'''
|
|
#include "Core/CoreException.h"
|
|
#include "Core/Unix/MountedFilesystem.h"
|
|
#include "Platform/StringConverter.h"
|
|
#include <fcntl.h>
|
|
#include <membership.h>
|
|
#include <cstring>
|
|
#include <sys/acl.h>
|
|
#include <sys/mount.h>
|
|
#include <sys/stat.h>
|
|
#include <unistd.h>
|
|
#include <stdio.h>
|
|
#include <iostream>
|
|
namespace VeraCrypt {
|
|
static bool IgnoreOwnership=false, FailAcl=false, Mounted=false;
|
|
static int TestStatfs(int fd, struct statfs *info) {
|
|
int result=fstatfs(fd, info);
|
|
if(result==0 && IgnoreOwnership)info->f_flags |= MNT_IGNORE_OWNERSHIP;
|
|
return result;
|
|
}
|
|
static int TestSetAcl(int fd, acl_t acl, acl_type_t type) {
|
|
if(FailAcl){errno=ENOTSUP;return -1;}
|
|
return acl_set_fd_np(fd, acl, type);
|
|
}
|
|
static bool fuse_service_find_mount(const char *, fsid_t &) { return Mounted; }
|
|
struct FuseService { static void RemoveAuxMountParent(const string &, int=-1); };
|
|
#define fstatfs TestStatfs
|
|
#define acl_set_fd_np TestSetAcl
|
|
''' + helpers + r'''
|
|
}
|
|
using namespace VeraCrypt;
|
|
int main(int argc, char **argv) {
|
|
try {
|
|
const std::string command=argv[1];
|
|
if(command=="create" || command=="noowners" || command=="acl-failure") {
|
|
IgnoreOwnership=command=="noowners"; FailAcl=command=="acl-failure";
|
|
std::cout << CreateFuseTAuxiliaryDirectory(argv[2], std::stoul(argv[3])) << '\n';
|
|
} else if(command=="cleanup") {
|
|
FuseService::RemoveAuxMountParent(argv[2]);
|
|
} else if(command.find("service-")==0) {
|
|
Mounted=command=="service-mounted";
|
|
FuseServiceAuxDirectory directory(argv[2]);
|
|
if(command=="service-removed" && rmdir(argv[2])!=0)throw std::runtime_error("fixture removal failed");
|
|
if(command=="service-replaced-child" || command=="service-replaced-parent") {
|
|
std::string path=argv[2];
|
|
if(command=="service-replaced-parent")path=path.substr(0, path.find_last_of('/'));
|
|
if(rename(path.c_str(), (path+".original").c_str())!=0 || mkdir(path.c_str(), 0700)!=0)
|
|
throw std::runtime_error("fixture replacement failed");
|
|
}
|
|
} else if(command=="filter") {
|
|
MountedFilesystem mount;
|
|
mount.MountPoint = argv[2]; mount.Owner = std::stoul(argv[3]); mount.Type = argv[4];
|
|
std::cout << IsOtherUsersFuseTAuxiliaryMount(mount, std::stoul(argv[5]), std::stoul(argv[6])) << '\n';
|
|
}
|
|
} catch (std::exception &e) { std::cerr << e.what() << '\n'; return 1; }
|
|
}
|
|
''')
|
|
binary = work / "check"
|
|
subprocess.run(["clang++", "-std=c++11", "-DTC_UNIX", "-DTC_MACOSX", "-I" + str(ROOT / "src"),
|
|
str(unit), str(args.build_dir / "Core/Core.a"),
|
|
str(args.build_dir / "Platform/Platform.a"), "-Wl,-dead_strip", "-o", str(binary)], check=True)
|
|
|
|
def create(base, uid=owner, succeeds=True, command="create"):
|
|
result = subprocess.run([str(binary), command, str(base), str(uid)], capture_output=True, text=True)
|
|
assert (result.returncode == 0) == succeeds, result.stdout + result.stderr
|
|
return Path(result.stdout.strip()) if succeeds else None
|
|
|
|
def case(name):
|
|
base = work / name
|
|
base.mkdir(mode=0o755)
|
|
return base
|
|
|
|
base = case("normal")
|
|
directory = create(base)
|
|
info = directory.stat()
|
|
assert info.st_uid == (0 if elevated else owner) and stat.S_IMODE(info.st_mode) == 0o700
|
|
assert create(base) != directory
|
|
sentinel = directory / "sentinel"
|
|
sentinel.write_text("private directory regression fixture\n")
|
|
if os.geteuid() == 0:
|
|
os.chown(sentinel, owner, -1)
|
|
sentinel.chmod(0o644)
|
|
|
|
def foreign(path=directory, user=other, real=other, mount_owner=0, backend="smbfs"):
|
|
result = subprocess.check_output([str(binary), "filter", str(path / ".veracrypt_aux_mnt-unit"),
|
|
str(mount_owner), backend, str(user), str(real)], text=True)
|
|
return result.strip() == "1"
|
|
|
|
assert foreign() == elevated # Only the elevated parent name identifies the user.
|
|
assert not foreign(user=0, real=other)
|
|
assert not foreign(user=owner, real=owner)
|
|
assert not foreign(user=0, real=owner)
|
|
assert not foreign(path=base) # Legacy mounts in a shared temp directory.
|
|
legacy = case("legacy-acl")
|
|
legacy.chmod(0o700)
|
|
subprocess.run(["chmod", "+a", "everyone allow search", str(legacy)], check=True)
|
|
assert not foreign(path=legacy) # ACL access is not reflected in owner/mode bits.
|
|
assert not foreign(backend="macfuse")
|
|
assert not foreign(mount_owner=owner if owner else other)
|
|
assert foreign(path=work / f".veracrypt_aux_root_{owner}")
|
|
assert not foreign(path=work / f".veracrypt_aux_root_{owner}", user=owner, real=owner)
|
|
assert not foreign(path=work / ".veracrypt_aux_root_invalid-ABCDEFGHIJKL")
|
|
|
|
for kind in ("symlink", "file", "directory"):
|
|
base = case(kind)
|
|
path = base / f"{prefix}{owner}"
|
|
if kind == "symlink":
|
|
path.symlink_to(directory, target_is_directory=True)
|
|
elif kind == "file":
|
|
path.write_text("unchanged")
|
|
else:
|
|
path.mkdir(mode=0o700)
|
|
before = path.lstat()
|
|
assert create(base) != path
|
|
after = path.lstat()
|
|
assert (before.st_ino, before.st_uid, before.st_mode) == (after.st_ino, after.st_uid, after.st_mode)
|
|
assert sentinel.read_text() == "private directory regression fixture\n"
|
|
|
|
base = case("inherited-acl")
|
|
subprocess.run(["chmod", "+a", "everyone allow read,search,directory_inherit", str(base)], check=True)
|
|
inherited = create(base)
|
|
acl = subprocess.check_output(["ls", "-lde", str(inherited)], text=True)
|
|
entries = re.findall(r"^\s*\d+: (.+)$", acl, re.M)
|
|
assert len(entries) == (1 if elevated and owner else 0), acl
|
|
if entries:
|
|
assert entries[0].endswith(" allow list,search,readattr,readsecurity"), acl
|
|
|
|
for failure in ("noowners", "acl-failure"):
|
|
base = case(failure)
|
|
create(base, succeeds=False, command=failure)
|
|
assert not list(base.iterdir()), "Failed parent setup left a directory"
|
|
|
|
base = case("concurrent")
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=12) as pool:
|
|
paths = list(pool.map(lambda _: create(base), range(24)))
|
|
assert len(set(paths)) == 24 and set(base.iterdir()) == set(paths)
|
|
for parent in paths:
|
|
child = parent / ".veracrypt_aux_mnt-unit"
|
|
child.mkdir()
|
|
subprocess.run([str(binary), "service-mounted", str(child)], check=True)
|
|
assert child.exists() and parent.exists()
|
|
subprocess.run([str(binary), "service-cleanup", str(child)], check=True)
|
|
assert not child.exists() and not parent.exists()
|
|
# An older client can remove the child before its service exits.
|
|
parent = create(base)
|
|
child = parent / ".veracrypt_aux_mnt-unit"
|
|
child.mkdir()
|
|
subprocess.run([str(binary), "service-removed", str(child)], check=True)
|
|
assert not parent.exists()
|
|
for target in ("child", "parent"):
|
|
parent = create(base)
|
|
child = parent / ".veracrypt_aux_mnt-unit"
|
|
child.mkdir()
|
|
subprocess.run([str(binary), "service-replaced-" + target, str(child)], check=True)
|
|
assert parent.exists(), "Removed a replacement parent"
|
|
if target == "child":
|
|
assert child.exists(), "Removed a replacement child"
|
|
# The fallback removes only empty parents in the per-mount format.
|
|
parent = create(base)
|
|
subprocess.run([str(binary), "cleanup", str(parent / ".veracrypt_aux_mnt-unit")], check=True)
|
|
assert not parent.exists()
|
|
for name in ("legacy-tmp", f"{prefix}{owner}", f"{prefix}{owner}-invalid"):
|
|
parent = base / name
|
|
parent.mkdir()
|
|
child = parent / ".veracrypt_aux_mnt-unit"
|
|
child.mkdir()
|
|
subprocess.run([str(binary), "service-cleanup", str(child)], check=True)
|
|
assert parent.exists() and not child.exists()
|
|
print("PASS: unique parents, owner/mode, ACL normalization, setup rollback, discovery, legacy preservation, service cleanup")
|
|
|
|
if os.geteuid() == 0 and owner != 0:
|
|
for uid, allowed in ((owner, True), (other, False)):
|
|
pid = os.fork()
|
|
if pid == 0:
|
|
try:
|
|
os.setgroups([])
|
|
os.setgid(20)
|
|
os.setuid(uid)
|
|
try:
|
|
with sentinel.open("rb") as stream:
|
|
stream.read(1)
|
|
assert allowed
|
|
except PermissionError:
|
|
assert not allowed
|
|
# Read/search access must not allow path replacement.
|
|
try:
|
|
(directory / "replacement").mkdir()
|
|
raise AssertionError("caller can modify the elevated parent")
|
|
except PermissionError:
|
|
pass
|
|
try:
|
|
directory.chmod(0o777)
|
|
raise AssertionError("caller can change the elevated parent's permissions")
|
|
except PermissionError:
|
|
pass
|
|
try:
|
|
directory.rename(directory.with_name("replacement-parent"))
|
|
raise AssertionError("caller can replace the elevated parent")
|
|
except PermissionError:
|
|
pass
|
|
# Discovery also works without entering the parent.
|
|
expected = "0" if allowed else "1"
|
|
output = subprocess.check_output([str(binary), "filter",
|
|
str(directory / ".veracrypt_aux_mnt-unit"),
|
|
"0", "smbfs", str(uid), str(uid)], text=True)
|
|
assert output.strip() == expected
|
|
os._exit(0)
|
|
except BaseException:
|
|
os._exit(1)
|
|
_, status = os.waitpid(pid, 0)
|
|
assert os.WIFEXITED(status) and os.WEXITSTATUS(status) == 0, (uid, status)
|
|
print("PASS: root owns elevated parent; caller has read/search only; unrelated uid gets EACCES; discovery scope")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|