From: Daan De Meyer <daan.j.demeyer@gmail.com>
Date: Sat, 5 Aug 2023 15:14:08 +0000 (+0200)
Subject: Use umask to control new file/directory permissions
X-Git-Tag: v15~30^2
X-Git-Url: http://git.ipfire.org/gitweb/gitweb.cgi?a=commitdiff_plain;h=refs%2Fpull%2F1738%2Fhead;p=thirdparty%2Fmkosi.git

Use umask to control new file/directory permissions

The primary reason to use umask is that python's mkdir() functions
and methods only apply the given mode to the final component of the
path, and not to its parent paths if parents=True is specified. Aside
from that, it's also just nicer to make sure the file/directory has
the right mode from the start instead of having to modify it later with
chmod().

We also clean up permissions in general, making sure we set umask
explicitly whenever we create a file or directory in state.root and
remove explicit permissions when we're not writing files in state.root.
---

diff --git a/mkosi/__init__.py b/mkosi/__init__.py
index 17bad72af..1ef2ca1da 100644
--- a/mkosi/__init__.py
+++ b/mkosi/__init__.py
@@ -60,6 +60,7 @@ from mkosi.util import (
     one_zero,
     scopedenv,
     try_import,
+    umask,
 )
 
 
@@ -117,14 +118,15 @@ def install_distribution(state: MkosiState) -> None:
 
             if not (state.root / "etc/machine-id").exists():
                 # Uninitialized means we want it to get initialized on first boot.
-                (state.root / "etc/machine-id").write_text("uninitialized\n")
-                (state.root / "etc/machine-id").chmod(0o0444)
+                with umask(~0o444):
+                    (state.root / "etc/machine-id").write_text("uninitialized\n")
 
             # Ensure /efi exists so that the ESP is mounted there, as recommended by
             # https://0pointer.net/blog/linux-boot-partitions.html. Use the most restrictive access mode we
             # can without tripping up mkfs tools since this directory is only meant to be overmounted and
             # should not be read from or written to.
-            state.root.joinpath("efi").mkdir(mode=0o500, exist_ok=True)
+            with umask(~0o500):
+                (state.root / "efi").mkdir(exist_ok=True)
 
             if state.config.packages:
                 state.config.distribution.install_packages(state, state.config.packages)
@@ -178,7 +180,8 @@ def mount_cache_overlay(state: MkosiState) -> Iterator[None]:
         return
 
     d = state.workspace / "cache-overlay"
-    d.mkdir(mode=0o755, exist_ok=True)
+    with umask(~0o755):
+        d.mkdir(exist_ok=True)
 
     with mount_overlay([state.root], d, state.root, read_only=False):
         yield
@@ -187,7 +190,8 @@ def mount_cache_overlay(state: MkosiState) -> Iterator[None]:
 def mount_build_overlay(state: MkosiState, read_only: bool = False) -> ContextManager[Path]:
     d = state.workspace / "build-overlay"
     if not d.is_symlink():
-        d.mkdir(mode=0o755, exist_ok=True)
+        with umask(~0o755):
+            d.mkdir(exist_ok=True)
     return mount_overlay([state.root], state.workspace / "build-overlay", state.root, read_only)
 
 
@@ -497,7 +501,8 @@ def install_boot_loader(state: MkosiState) -> None:
 
         with complete_step("Setting up secure boot auto-enrollment…"):
             keys = state.root / "efi/loader/keys/auto"
-            keys.mkdir(parents=True, exist_ok=True)
+            with umask(~0o700):
+                keys.mkdir(parents=True, exist_ok=True)
 
             # sbsiglist expects a DER certificate.
             run(["openssl",
@@ -935,7 +940,8 @@ def install_unified_kernel(state: MkosiState, roothash: Optional[str]) -> None:
                 cmd += ["--initrd", gen_kernel_modules_initrd(state, kver)]
 
             # Make sure the parent directory where we'll be writing the UKI exists.
-            boot_binary.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
+            with umask(~0o700):
+                boot_binary.parent.mkdir(parents=True, exist_ok=True)
 
             run(cmd)
 
@@ -983,7 +989,7 @@ def copy_nspawn_settings(state: MkosiState) -> None:
         return None
 
     with complete_step("Copying nspawn settings file…"):
-        shutil.copy(state.config.nspawn_settings, state.staging / state.config.output_nspawn_settings)
+        shutil.copy2(state.config.nspawn_settings, state.staging / state.config.output_nspawn_settings)
 
 
 def hash_file(of: TextIO, path: Path) -> None:
@@ -1189,46 +1195,47 @@ def configure_ssh(state: MkosiState) -> None:
     if not state.config.ssh:
         return
 
-    state.root.joinpath("usr/lib/systemd/system/ssh.socket").write_text(
-        dedent(
-            """\
-            [Unit]
-            Description=Mkosi SSH Server VSock Socket
-            ConditionVirtualization=!container
-            Wants=sshd-keygen.target
-
-            [Socket]
-            ListenStream=vsock::22
-            Accept=yes
-
-            [Install]
-            WantedBy=sockets.target
-            """
+    with umask(~0o644):
+        (state.root / "usr/lib/systemd/system/ssh.socket").write_text(
+            dedent(
+                """\
+                [Unit]
+                Description=Mkosi SSH Server VSock Socket
+                ConditionVirtualization=!container
+                Wants=sshd-keygen.target
+
+                [Socket]
+                ListenStream=vsock::22
+                Accept=yes
+
+                [Install]
+                WantedBy=sockets.target
+                """
+            )
         )
-    )
 
-    state.root.joinpath("usr/lib/systemd/system/ssh@.service").write_text(
-        dedent(
-            """\
-            [Unit]
-            Description=Mkosi SSH Server
-            After=sshd-keygen.target
-
-            [Service]
-            # We disable PAM because of an openssh-server bug where it sets PAM_RHOST=UNKNOWN when -i is used
-            # causing a very slow reverse DNS lookup by pam.
-            ExecStart=sshd -i -o UsePAM=no
-            StandardInput=socket
-            RuntimeDirectoryPreserve=yes
-            # ssh always exits with 255 even on normal disconnect, so let's mark that as success so we don't
-            # get noisy logs about SSH service failures.
-            SuccessExitStatus=255
-            """
+        (state.root / "usr/lib/systemd/system/ssh@.service").write_text(
+            dedent(
+                """\
+                [Unit]
+                Description=Mkosi SSH Server
+                After=sshd-keygen.target
+
+                [Service]
+                # We disable PAM because of an openssh-server bug where it sets PAM_RHOST=UNKNOWN when -i is
+                # used causing a very slow reverse DNS lookup by pam.
+                ExecStart=sshd -i -o UsePAM=no
+                StandardInput=socket
+                RuntimeDirectoryPreserve=yes
+                # ssh always exits with 255 even on normal disconnect, so let's mark that as success so we
+                # don't get noisy logs about SSH service failures.
+                SuccessExitStatus=255
+                """
+            )
         )
-    )
 
-    presetdir = state.root / "usr/lib/systemd/system-preset"
-    presetdir.joinpath("80-mkosi-ssh.preset").write_text("enable ssh.socket\n")
+        presetdir = state.root / "usr/lib/systemd/system-preset"
+        (presetdir / "80-mkosi-ssh.preset").write_text("enable ssh.socket\n")
 
 
 def configure_initrd(state: MkosiState) -> None:
@@ -1347,13 +1354,12 @@ def run_firstboot(state: MkosiState) -> None:
         # Initrds generally don't ship with only /usr so there's not much point in putting the credentials in
         # /usr/lib/credstore.
         if state.config.output_format != OutputFormat.cpio or not state.config.make_initrd:
-            (state.root / "usr/lib/credstore").mkdir(mode=0o755, exist_ok=True)
+            with umask(~0o755):
+                (state.root / "usr/lib/credstore").mkdir(exist_ok=True)
 
             for cred, value in creds:
-                (state.root / "usr/lib/credstore" / cred).write_text(value)
-
-                if "password" in cred:
-                    (state.root / "usr/lib/credstore" / cred).chmod(0o600)
+                with umask(~0o600 if "password" in cred else ~0o644):
+                    (state.root / "usr/lib/credstore" / cred).write_text(value)
 
 
 def run_selinux_relabel(state: MkosiState) -> None:
diff --git a/mkosi/distributions/debian.py b/mkosi/distributions/debian.py
index 0fb09f990..b28000ad9 100644
--- a/mkosi/distributions/debian.py
+++ b/mkosi/distributions/debian.py
@@ -12,6 +12,7 @@ from mkosi.log import die
 from mkosi.run import run
 from mkosi.state import MkosiState
 from mkosi.tree import extract_tree
+from mkosi.util import umask
 
 
 class DebianInstaller(DistributionInstaller):
@@ -88,10 +89,10 @@ class DebianInstaller(DistributionInstaller):
             "x32"         : ["lib32", "lib64", "libx32"],
         }.get(state.config.distribution.architecture(state.config.architecture), [])
 
-        state.root.joinpath("usr").mkdir(mode=0o755, exist_ok=True)
-        for d in subdirs:
-            state.root.joinpath(d).symlink_to(f"usr/{d}")
-            state.root.joinpath(f"usr/{d}").mkdir(mode=0o755, exist_ok=True)
+        with umask(~0o755):
+            for d in subdirs:
+                (state.root / d).symlink_to(f"usr/{d}")
+                (state.root / f"usr/{d}").mkdir(parents=True, exist_ok=True)
 
         # Next, we invoke apt-get install to download all the essential packages. With DPkg::Pre-Install-Pkgs,
         # we specify a shell command that will receive the list of packages that will be installed on stdin.
@@ -128,8 +129,8 @@ class DebianInstaller(DistributionInstaller):
         # Note: despite writing in /usr/sbin, this file is not shipped by the OS and instead should be managed by
         # the admin.
         policyrcd = state.root / "usr/sbin/policy-rc.d"
-        policyrcd.write_text("#!/bin/sh\nexit 101\n")
-        policyrcd.chmod(0o755)
+        with umask(~0o644):
+            policyrcd.write_text("#!/bin/sh\nexit 101\n")
 
         invoke_apt(state, "apt-get", "update", apivfs=False)
         invoke_apt(state, "apt-get", "install", packages, apivfs=apivfs)
diff --git a/mkosi/install.py b/mkosi/install.py
index 856c453ba..36a095d78 100644
--- a/mkosi/install.py
+++ b/mkosi/install.py
@@ -2,19 +2,14 @@
 
 import importlib.resources
 from pathlib import Path
-from typing import Optional
 
-from mkosi.util import make_executable
+from mkosi.util import make_executable, umask
 
 
-def write_resource(
-    where: Path, resource: str, key: str, *, executable: bool = False, mode: Optional[int] = None
-) -> None:
+def write_resource(where: Path, resource: str, key: str, *, executable: bool = False) -> None:
     text = importlib.resources.read_text(resource, key)
     where.write_text(text)
-    if mode is not None:
-        where.chmod(mode)
-    elif executable:
+    if executable:
         make_executable(where)
 
 
@@ -22,6 +17,7 @@ def add_dropin_config_from_resource(
     root: Path, unit: str, name: str, resource: str, key: str
 ) -> None:
     dropin = root / f"usr/lib/systemd/system/{unit}.d/{name}.conf"
-    dropin.parent.mkdir(mode=0o755, parents=True, exist_ok=True)
-    write_resource(dropin, resource, key, mode=0o644)
+    with umask(~0o755):
+        dropin.parent.mkdir(parents=True, exist_ok=True)
+        write_resource(dropin, resource, key)
 
diff --git a/mkosi/installer/apt.py b/mkosi/installer/apt.py
index 4e4bf23b8..da36365a0 100644
--- a/mkosi/installer/apt.py
+++ b/mkosi/installer/apt.py
@@ -6,7 +6,7 @@ from collections.abc import Sequence
 from mkosi.run import apivfs_cmd, bwrap
 from mkosi.state import MkosiState
 from mkosi.types import PathString
-from mkosi.util import flatten, sort_packages
+from mkosi.util import flatten, sort_packages, umask
 
 
 def setup_apt(state: MkosiState, repos: Sequence[str]) -> None:
@@ -18,10 +18,9 @@ def setup_apt(state: MkosiState, repos: Sequence[str]) -> None:
     state.pkgmngr.joinpath("var/lib/apt").mkdir(exist_ok=True, parents=True)
 
     # TODO: Drop once apt 2.5.4 is widely available.
-    state.root.joinpath("var").mkdir(mode=0o755, exist_ok=True)
-    state.root.joinpath("var/lib").mkdir(mode=0o755, exist_ok=True)
-    state.root.joinpath("var/lib/dpkg").mkdir(mode=0o755, exist_ok=True)
-    state.root.joinpath("var/lib/dpkg/status").touch()
+    with umask(~0o755):
+        (state.root / "var/lib/dpkg").mkdir(parents=True, exist_ok=True)
+        (state.root / "var/lib/dpkg/status").touch()
 
     # We have a special apt.conf outside of pkgmngr dir that only configures "Dir" that we pass to
     # APT_CONFIG to tell apt it should read config files in pkgmngr instead of in its usual locations. This
diff --git a/mkosi/installer/pacman.py b/mkosi/installer/pacman.py
index ece74c38e..952cf3785 100644
--- a/mkosi/installer/pacman.py
+++ b/mkosi/installer/pacman.py
@@ -8,7 +8,7 @@ from mkosi.config import ConfigFeature
 from mkosi.run import apivfs_cmd, bwrap
 from mkosi.state import MkosiState
 from mkosi.types import PathString
-from mkosi.util import flatten, sort_packages
+from mkosi.util import flatten, sort_packages, umask
 
 
 def setup_pacman(state: MkosiState) -> None:
@@ -30,13 +30,14 @@ def setup_pacman(state: MkosiState) -> None:
         sig_level = "Never"
 
     # Create base layout for pacman and pacman-key
-    state.root.joinpath("var/lib/pacman").mkdir(mode=0o755, exist_ok=True, parents=True)
+    with umask(~0o755):
+        state.root.joinpath("var/lib/pacman").mkdir(exist_ok=True, parents=True)
 
     config = state.pkgmngr / "etc/pacman.conf"
     if config.exists():
         return
 
-    config.parent.mkdir(mode=0o755, exist_ok=True, parents=True)
+    config.parent.mkdir(exist_ok=True, parents=True)
 
     repos = []
 
diff --git a/mkosi/mounts.py b/mkosi/mounts.py
index fca99d3a5..df53caf5c 100644
--- a/mkosi/mounts.py
+++ b/mkosi/mounts.py
@@ -13,6 +13,7 @@ from mkosi.config import GenericVersion, MkosiConfig
 from mkosi.log import complete_step
 from mkosi.run import run
 from mkosi.types import PathString
+from mkosi.util import umask
 
 T = TypeVar("T")
 
@@ -47,7 +48,8 @@ def mount(
     lazy: bool = False,
 ) -> Iterator[Path]:
     if not where.exists():
-        where.mkdir(mode=0o755, parents=True)
+        with umask(~0o755):
+            where.mkdir(parents=True)
 
     if read_only:
         options = ["ro", *options]
diff --git a/mkosi/qemu.py b/mkosi/qemu.py
index 001364dfe..a80aa2dd4 100644
--- a/mkosi/qemu.py
+++ b/mkosi/qemu.py
@@ -259,7 +259,7 @@ def run_qemu(args: MkosiArgs, config: MkosiConfig) -> None:
     with contextlib.ExitStack() as stack:
         if fw_supports_sb:
             ovmf_vars = stack.enter_context(tempfile.NamedTemporaryFile(prefix=".mkosi-"))
-            shutil.copy(find_ovmf_vars(config), Path(ovmf_vars.name))
+            shutil.copy2(find_ovmf_vars(config), Path(ovmf_vars.name))
             cmdline += [
                 "-global", "ICH9-LPC.disable_s3=1",
                 "-global", "driver=cfi.pflash01,property=secure,value=on",
diff --git a/mkosi/state.py b/mkosi/state.py
index ee55e1506..7ea730684 100644
--- a/mkosi/state.py
+++ b/mkosi/state.py
@@ -7,6 +7,7 @@ from typing import Optional, Type
 
 from mkosi.config import MkosiArgs, MkosiConfig
 from mkosi.tree import make_tree
+from mkosi.util import umask
 
 
 class MkosiState:
@@ -18,7 +19,8 @@ class MkosiState:
 
     def __enter__(self) -> "MkosiState":
         self._workspace = tempfile.TemporaryDirectory(dir=self.config.workspace_dir or Path.cwd(), prefix=".mkosi-tmp")
-        make_tree(self.config, self.root, mode=0o755)
+        with umask(~0o755):
+            make_tree(self.config, self.root)
         self.staging.mkdir()
         self.pkgmngr.mkdir()
         self.install_dir.mkdir(exist_ok=True)
diff --git a/mkosi/tree.py b/mkosi/tree.py
index 1d416f9d8..2c1ddea3b 100644
--- a/mkosi/tree.py
+++ b/mkosi/tree.py
@@ -10,7 +10,7 @@ from mkosi.config import ConfigFeature, MkosiConfig
 from mkosi.log import die
 from mkosi.run import bwrap, finalize_passwd_mounts, run
 from mkosi.types import PathString
-from mkosi.util import tar_binary
+from mkosi.util import tar_binary, umask
 
 
 def statfs(path: Path) -> str:
@@ -22,7 +22,7 @@ def is_subvolume(path: Path) -> bool:
     return path.is_dir() and statfs(path) == "btrfs" and path.stat().st_ino == 256
 
 
-def make_tree(config: MkosiConfig, path: Path, mode: int) -> None:
+def make_tree(config: MkosiConfig, path: Path) -> None:
     if config.use_subvolumes == ConfigFeature.enabled and not shutil.which("btrfs"):
         die("Subvolumes requested but the btrfs command was not found")
 
@@ -30,7 +30,7 @@ def make_tree(config: MkosiConfig, path: Path, mode: int) -> None:
         if config.use_subvolumes == ConfigFeature.enabled:
             die(f"Subvolumes requested but {path} is not located on a btrfs filesystem")
 
-        path.mkdir(mode)
+        path.mkdir()
         return
 
     if config.use_subvolumes != ConfigFeature.disabled and shutil.which("btrfs") is not None:
@@ -39,10 +39,8 @@ def make_tree(config: MkosiConfig, path: Path, mode: int) -> None:
     else:
         result = 1
 
-    if result == 0:
-        path.chmod(mode)
-    else:
-        path.mkdir(mode)
+    if result != 0:
+        path.mkdir()
 
 
 def copy_tree(config: MkosiConfig, src: Path, dst: Path, *, preserve_owner: bool = True) -> None:
@@ -160,7 +158,8 @@ def install_tree(config: MkosiConfig, src: Path, dst: Path, target: Optional[Pat
     if target:
         t = dst / target.relative_to("/")
 
-    t.parent.mkdir(mode=0o755, parents=True, exist_ok=True)
+    with umask(~0o755):
+        t.parent.mkdir(parents=True, exist_ok=True)
 
     if src.is_dir():
         copy_tree(config, src, t, preserve_owner=False)
diff --git a/mkosi/util.py b/mkosi/util.py
index c52bf44ea..c1d8f66c5 100644
--- a/mkosi/util.py
+++ b/mkosi/util.py
@@ -229,3 +229,12 @@ def tar_binary() -> str:
 
 def one_zero(b: bool) -> str:
     return "1" if b else "0"
+
+
+@contextlib.contextmanager
+def umask(mask: int) -> Iterator[None]:
+    old = os.umask(mask)
+    try:
+        yield
+    finally:
+        os.umask(old)
diff --git a/tests/test_parse_load_args.py b/tests/test_parse_load_args.py
index 0a1decf30..118a18456 100644
--- a/tests/test_parse_load_args.py
+++ b/tests/test_parse_load_args.py
@@ -68,7 +68,7 @@ def test_os_distribution() -> None:
 def test_parse_config_files_filter() -> None:
     with cd_temp_dir():
         confd = Path("mkosi.conf.d")
-        confd.mkdir(0o755)
+        confd.mkdir()
 
         (confd / "10-file.conf").write_text("[Content]\nPackages=yes")
         (confd / "20-file.noconf").write_text("[Content]\nPackages=nope")