efe4ad4b92
Add a new utility that wraps ctypes.CDLL() for the self-embedded libc.so. Initially, it only exposes renameat2(2), but more can be added when needed in the future. The Libc class is very similar to the existing LibCap class, with a similar instantiation logic with singleton access. In the future, the Libc class will allow access to other system calls and libc.so functionality, when needed.
242 lines
7.6 KiB
Python
242 lines
7.6 KiB
Python
#
|
|
# Tests for the `osbuild.util.linux` module.
|
|
#
|
|
|
|
import os
|
|
import subprocess
|
|
import tempfile
|
|
|
|
import pytest
|
|
|
|
from osbuild.util import linux
|
|
|
|
from .. import test
|
|
|
|
|
|
@pytest.fixture(name="tmpdir")
|
|
def tmpdir_fixture():
|
|
with tempfile.TemporaryDirectory(dir="/var/tmp") as tmp:
|
|
yield tmp
|
|
|
|
|
|
@pytest.mark.skipif(not test.TestBase.can_modify_immutable("/var/tmp"), reason="root-only")
|
|
def test_ioctl_get_immutable(tmpdir):
|
|
#
|
|
# Test the `ioctl_get_immutable()` helper and make sure it works
|
|
# as intended.
|
|
#
|
|
|
|
with open(f"{tmpdir}/immutable", "x", encoding="utf8") as f:
|
|
assert not linux.ioctl_get_immutable(f.fileno())
|
|
|
|
|
|
@pytest.mark.skipif(not test.TestBase.can_modify_immutable("/var/tmp"), reason="root-only")
|
|
def test_ioctl_toggle_immutable(tmpdir):
|
|
#
|
|
# Test the `ioctl_toggle_immutable()` helper and make sure it works
|
|
# as intended.
|
|
#
|
|
|
|
with open(f"{tmpdir}/immutable", "x", encoding="utf8") as f:
|
|
# Check the file is mutable by default and if we clear it again.
|
|
assert not linux.ioctl_get_immutable(f.fileno())
|
|
linux.ioctl_toggle_immutable(f.fileno(), False)
|
|
assert not linux.ioctl_get_immutable(f.fileno())
|
|
|
|
# Set immutable and check for it. Try again to verify with flag set.
|
|
linux.ioctl_toggle_immutable(f.fileno(), True)
|
|
assert linux.ioctl_get_immutable(f.fileno())
|
|
linux.ioctl_toggle_immutable(f.fileno(), True)
|
|
assert linux.ioctl_get_immutable(f.fileno())
|
|
|
|
# Verify immutable files cannot be unlinked.
|
|
with pytest.raises(OSError):
|
|
os.unlink(f"{tmpdir}/immutable")
|
|
|
|
# Check again that clearing the flag works.
|
|
linux.ioctl_toggle_immutable(f.fileno(), False)
|
|
assert not linux.ioctl_get_immutable(f.fileno())
|
|
|
|
# This time, check that we actually set the same flag as `chattr`.
|
|
subprocess.run(["chattr", "+i",
|
|
f"{tmpdir}/immutable"], check=True)
|
|
assert linux.ioctl_get_immutable(f.fileno())
|
|
|
|
# Same for clearing it.
|
|
subprocess.run(["chattr", "-i",
|
|
f"{tmpdir}/immutable"], check=True)
|
|
assert not linux.ioctl_get_immutable(f.fileno())
|
|
|
|
# Verify we can unlink the file again, once the flag is cleared.
|
|
os.unlink(f"{tmpdir}/immutable")
|
|
|
|
|
|
@pytest.mark.skipif(not linux.cap_is_supported(), reason="no support for capabilities")
|
|
def test_capabilities():
|
|
#
|
|
# Test the capability related utility functions
|
|
#
|
|
|
|
lib = linux.LibCap.get_default()
|
|
assert lib
|
|
l2 = linux.LibCap.get_default()
|
|
assert lib is l2
|
|
|
|
assert linux.cap_is_supported()
|
|
|
|
assert linux.cap_is_supported("CAP_MAC_ADMIN")
|
|
|
|
val = lib.from_name("CAP_MAC_ADMIN")
|
|
assert val >= 0
|
|
|
|
name = lib.to_name(val)
|
|
assert name == "CAP_MAC_ADMIN"
|
|
|
|
assert not linux.cap_is_supported("CAP_GICMO")
|
|
with pytest.raises(OSError):
|
|
lib.from_name("CAP_GICMO")
|
|
|
|
|
|
def test_fcntl_flock():
|
|
#
|
|
# This tests the `linux.fcntl_flock()` file-locking helper. Note
|
|
# that file-locks are on the open-file-description, so they are shared
|
|
# between dupped file-descriptors. We explicitly create a separate
|
|
# file-description via `/proc/self/fd/`.
|
|
#
|
|
|
|
with tempfile.TemporaryFile() as f:
|
|
fd1 = f.fileno()
|
|
fd2 = os.open(os.path.join("/proc/self/fd/", str(fd1)), os.O_RDWR | os.O_CLOEXEC)
|
|
|
|
# Test: unlock
|
|
linux.fcntl_flock(fd1, linux.fcntl.F_UNLCK)
|
|
|
|
# Test: write-lock + unlock
|
|
linux.fcntl_flock(fd1, linux.fcntl.F_WRLCK)
|
|
linux.fcntl_flock(fd1, linux.fcntl.F_UNLCK)
|
|
|
|
# Test: read-lock1 + read-lock2 + unlock1 + unlock2
|
|
linux.fcntl_flock(fd1, linux.fcntl.F_RDLCK)
|
|
linux.fcntl_flock(fd2, linux.fcntl.F_RDLCK)
|
|
linux.fcntl_flock(fd1, linux.fcntl.F_UNLCK)
|
|
linux.fcntl_flock(fd2, linux.fcntl.F_UNLCK)
|
|
|
|
# Test: write-lock1 + write-lock2 + unlock
|
|
linux.fcntl_flock(fd1, linux.fcntl.F_WRLCK)
|
|
with pytest.raises(BlockingIOError):
|
|
linux.fcntl_flock(fd2, linux.fcntl.F_WRLCK)
|
|
linux.fcntl_flock(fd1, linux.fcntl.F_UNLCK)
|
|
|
|
# Test: write-lock1 + read-lock2 + unlock
|
|
linux.fcntl_flock(fd1, linux.fcntl.F_WRLCK)
|
|
with pytest.raises(BlockingIOError):
|
|
linux.fcntl_flock(fd2, linux.fcntl.F_RDLCK)
|
|
linux.fcntl_flock(fd1, linux.fcntl.F_UNLCK)
|
|
|
|
# Test: read-lock1 + write-lock2 + unlock
|
|
linux.fcntl_flock(fd1, linux.fcntl.F_RDLCK)
|
|
with pytest.raises(BlockingIOError):
|
|
linux.fcntl_flock(fd2, linux.fcntl.F_WRLCK)
|
|
linux.fcntl_flock(fd1, linux.fcntl.F_UNLCK)
|
|
|
|
# Test: write-lock1 + read-lock1 + read-lock2 + unlock
|
|
linux.fcntl_flock(fd1, linux.fcntl.F_WRLCK)
|
|
linux.fcntl_flock(fd1, linux.fcntl.F_RDLCK)
|
|
linux.fcntl_flock(fd2, linux.fcntl.F_RDLCK)
|
|
linux.fcntl_flock(fd1, linux.fcntl.F_UNLCK)
|
|
|
|
# Test: read-lock1 + read-lock2 + write-lock1 + unlock1 + unlock2
|
|
linux.fcntl_flock(fd1, linux.fcntl.F_RDLCK)
|
|
linux.fcntl_flock(fd2, linux.fcntl.F_RDLCK)
|
|
with pytest.raises(BlockingIOError):
|
|
linux.fcntl_flock(fd1, linux.fcntl.F_WRLCK)
|
|
linux.fcntl_flock(fd1, linux.fcntl.F_UNLCK)
|
|
linux.fcntl_flock(fd2, linux.fcntl.F_UNLCK)
|
|
|
|
# Test: write-lock3 + write-lock1 + close3 + write-lock1 + unlock1
|
|
fd3 = os.open(os.path.join("/proc/self/fd/", str(fd1)), os.O_RDWR | os.O_CLOEXEC)
|
|
linux.fcntl_flock(fd3, linux.fcntl.F_WRLCK)
|
|
with pytest.raises(BlockingIOError):
|
|
linux.fcntl_flock(fd1, linux.fcntl.F_WRLCK)
|
|
os.close(fd3)
|
|
linux.fcntl_flock(fd1, linux.fcntl.F_WRLCK)
|
|
linux.fcntl_flock(fd1, linux.fcntl.F_UNLCK)
|
|
|
|
# Cleanup
|
|
os.close(fd2)
|
|
|
|
|
|
def test_libc():
|
|
#
|
|
# Test that the Libc class can be instantiated and provides a suitable
|
|
# default singleton. Verify the expected interfaces exist (though tests
|
|
# for them are separate).
|
|
#
|
|
|
|
libc0 = linux.Libc.make()
|
|
libc1 = linux.Libc.default()
|
|
|
|
assert libc0 is not libc1
|
|
assert libc1 is linux.Libc.default()
|
|
|
|
assert libc0.AT_FDCWD
|
|
assert libc0.RENAME_EXCHANGE
|
|
assert libc0.RENAME_NOREPLACE
|
|
assert libc0.RENAME_WHITEOUT
|
|
assert libc0.renameat2
|
|
|
|
|
|
def test_libc_renameat2_errcheck():
|
|
#
|
|
# Verify the `renameat(2)` system call on `Libc` correctly turns errors into
|
|
# python exceptions.
|
|
#
|
|
|
|
libc = linux.Libc.default()
|
|
|
|
with pytest.raises(OSError):
|
|
libc.renameat2(oldpath=b"", newpath=b"")
|
|
|
|
|
|
def test_libc_renameat2_exchange(tmpdir):
|
|
#
|
|
# Verify the `renameat(2)` system call on `Libc` with the
|
|
# `RENAME_EXCHANGE` flag. This swaps two files atomically.
|
|
#
|
|
|
|
libc = linux.Libc.default()
|
|
|
|
with open(f"{tmpdir}/foo", "x", encoding="utf8") as f:
|
|
f.write("foo")
|
|
with open(f"{tmpdir}/bar", "x", encoding="utf8") as f:
|
|
f.write("bar")
|
|
|
|
libc.renameat2(
|
|
oldpath=f"{tmpdir}/foo".encode(),
|
|
newpath=f"{tmpdir}/bar".encode(),
|
|
flags=linux.Libc.RENAME_EXCHANGE,
|
|
)
|
|
|
|
with open(f"{tmpdir}/foo", "r", encoding="utf8") as f:
|
|
assert f.read() == "bar"
|
|
with open(f"{tmpdir}/bar", "r", encoding="utf8") as f:
|
|
assert f.read() == "foo"
|
|
|
|
|
|
def test_proc_boot_id():
|
|
#
|
|
# Test the `proc_boot_id()` function which reads the current boot-id
|
|
# from the kernel. Make sure it is a valid UUID and also consistent on
|
|
# repeated queries.
|
|
#
|
|
|
|
bootid = linux.proc_boot_id("test")
|
|
assert len(bootid.hex) == 32
|
|
assert bootid.version == 4
|
|
|
|
bootid2 = linux.proc_boot_id("test")
|
|
assert bootid.int == bootid2.int
|
|
|
|
bootid3 = linux.proc_boot_id("foobar")
|
|
assert bootid.int != bootid3.int
|