We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Support libarchive cpio
Hi,

please support libarchive cpio(bsdcpio), which does not have `--reproducible`

https://lists.gnu.org/archive/html/bug-cpio/2014-11/msg00000.html

https://github.com/systemd/mkosi/blob/2c45d0effb1871750a2e9f897510d2745cb6d6b9/mkosi/__init__.py#L3489

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `mkosi/archive.py`
Content:
```
1 # SPDX-License-Identifier: LGPL-2.1+
2 
3 import os
4 from collections.abc import Iterable
5 from pathlib import Path
6 from typing import Optional
7 
8 from mkosi.log import log_step
9 from mkosi.run import bwrap, finalize_passwd_mounts
10 from mkosi.util import tar_binary
11 
12 
13 def tar_exclude_apivfs_tmp() -> list[str]:
14     return [
15         "--exclude", "./dev/*",
16         "--exclude", "./proc/*",
17         "--exclude", "./sys/*",
18         "--exclude", "./tmp/*",
19         "--exclude", "./run/*",
20         "--exclude", "./var/tmp/*",
21     ]
22 
23 
24 def make_tar(src: Path, dst: Path) -> None:
25     log_step(f"Creating tar archive {dst}…")
26     bwrap(
27         [
28             tar_binary(),
29             "--create",
30             "--file", dst,
31             "--directory", src,
32             "--acls",
33             "--selinux",
34             "--xattrs",
35             "--sparse",
36             "--force-local",
37             *tar_exclude_apivfs_tmp(),
38             ".",
39         ],
40         # Make sure tar uses user/group information from the root directory instead of the host.
41         options=finalize_passwd_mounts(src) if (src / "etc/passwd").exists() else [],
42     )
43 
44 
45 def extract_tar(src: Path, dst: Path, log: bool = True) -> None:
46     if log:
47         log_step(f"Extracting tar archive {src}…")
48     bwrap(
49         [
50             tar_binary(),
51             "--extract",
52             "--file", src,
53             "--directory", dst,
54             "--keep-directory-symlink",
55             "--no-overwrite-dir",
56             "--same-permissions",
57             "--same-owner" if (dst / "etc/passwd").exists() else "--numeric-owner",
58             "--same-order",
59             "--acls",
60             "--selinux",
61             "--xattrs",
62             "--force-local",
63             *tar_exclude_apivfs_tmp(),
64         ],
65         # Make sure tar uses user/group information from the root directory instead of the host.
66         options=finalize_passwd_mounts(dst) if (dst / "etc/passwd").exists() else [],
67     )
68 
69 
70 def make_cpio(src: Path, dst: Path, files: Optional[Iterable[Path]] = None) -> None:
71     if not files:
72         files = src.rglob("*")
73 
74     log_step(f"Creating cpio archive {dst}…")
75     bwrap(
76         [
77             "cpio",
78             "--create",
79             "--reproducible",
80             "--null",
81             "--format=newc",
82             "--quiet",
83             "--directory", src,
84             "-O", dst,
85         ],
86         input="\0".join(os.fspath(f.relative_to(src)) for f in files),
87         # Make sure tar uses user/group information from the root directory instead of the host.
88         options=finalize_passwd_mounts(dst),
89     )
90
```

Path: `mkosi/util.py`
Content:
```
1 # SPDX-License-Identifier: LGPL-2.1+
2 
3 import ast
4 import contextlib
5 import copy
6 import enum
7 import errno
8 import fcntl
9 import functools
10 import importlib
11 import itertools
12 import logging
13 import os
14 import pwd
15 import re
16 import resource
17 import shutil
18 import stat
19 import sys
20 import tempfile
21 from collections.abc import Iterable, Iterator, Mapping, Sequence
22 from pathlib import Path
23 from typing import Any, Callable, Optional, TypeVar
24 
25 T = TypeVar("T")
26 V = TypeVar("V")
27 
28 
29 def dictify(f: Callable[..., Iterator[tuple[T, V]]]) -> Callable[..., dict[T, V]]:
30     def wrapper(*args: Any, **kwargs: Any) -> dict[T, V]:
31         return dict(f(*args, **kwargs))
32 
33     return functools.update_wrapper(wrapper, f)
34 
35 
36 @dictify
37 def read_os_release() -> Iterator[tuple[str, str]]:
38     try:
39         filename = "/etc/os-release"
40         f = open(filename)
41     except FileNotFoundError:
42         filename = "/usr/lib/os-release"
43         f = open(filename)
44 
45     with f:
46         for line_number, line in enumerate(f, start=1):
47             line = line.rstrip()
48             if not line or line.startswith("#"):
49                 continue
50             if (m := re.match(r"([A-Z][A-Z_0-9]+)=(.*)", line)):
51                 name, val = m.groups()
52                 if val and val[0] in "\"'":
53                     val = ast.literal_eval(val)
54                 yield name, val
55             else:
56                 print(f"{filename}:{line_number}: bad line {line!r}", file=sys.stderr)
57 
58 
59 def format_rlimit(rlimit: int) -> str:
60     limits = resource.getrlimit(rlimit)
61     soft = "infinity" if limits[0] == resource.RLIM_INFINITY else str(limits[0])
62     hard = "infinity" if limits[1] == resource.RLIM_INFINITY else str(limits[1])
63     return f"{soft}:{hard}"
64 
65 
66 def sort_packages(packages: Iterable[str]) -> list[str]:
67     """Sorts packages: normal first, paths second, conditional third"""
68 
69     m = {"(": 2, "/": 1}
70     sort = lambda name: (m.get(name[0], 0), name)
71     return sorted(packages, key=sort)
72 
73 
74 def flatten(lists: Iterable[Iterable[T]]) -> list[T]:
75     """Flatten a sequence of sequences into a single list."""
76     return list(itertools.chain.from_iterable(lists))
77 
78 
79 class InvokingUser:
80     @staticmethod
81     def _uid_from_env() -> Optional[int]:
82         uid = os.getenv("SUDO_UID") or os.getenv("PKEXEC_UID")
83         return int(uid) if uid is not None else None
84 
85     @classmethod
86     def uid(cls) -> int:
87         return cls._uid_from_env() or os.getuid()
88 
89     @classmethod
90     def uid_gid(cls) -> tuple[int, int]:
91         if (uid := cls._uid_from_env()) is not None:
92             gid = int(os.getenv("SUDO_GID", pwd.getpwuid(uid).pw_gid))
93             return uid, gid
94         return os.getuid(), os.getgid()
95 
96     @classmethod
97     def name(cls) -> str:
98         return pwd.getpwuid(cls.uid()).pw_name
99 
100     @classmethod
101     def home(cls) -> Path:
102         return Path(f"~{cls.name()}").expanduser()
103 
104     @classmethod
105     def is_running_user(cls) -> bool:
106         return cls.uid() == os.getuid()
107 
108 
109 @contextlib.contextmanager
110 def chdir(directory: Path) -> Iterator[None]:
111     old = Path.cwd()
112 
113     if old == directory:
114         yield
115         return
116 
117     try:
118         os.chdir(directory)
119         yield
120     finally:
121         os.chdir(old)
122 
123 
124 def qemu_check_kvm_support(log: bool) -> bool:
125     # some CI runners may present a non-working KVM device
126     try:
127         os.close(os.open("/dev/kvm", os.O_RDWR|os.O_CLOEXEC))
128     except OSError as e:
129         if e.errno == errno.ENOENT:
130             if log:
131                 logging.warning("/dev/kvm not found. Not using KVM acceleration.")
132             return False
133         elif e.errno in (errno.EPERM, errno.EACCES):
134             if log:
135                 logging.warning("Permission denied to access /dev/kvm. Not using KVM acceleration")
136             return False
137 
138         raise e
139 
140     return True
141 
142 
143 def qemu_check_vsock_support(log: bool) -> bool:
144     try:
145         os.close(os.open("/dev/vhost-vsock", os.O_RDWR|os.O_CLOEXEC))
146     except OSError as e:
147         if e.errno == errno.ENOENT:
148             if log:
149                 logging.warning("/dev/vhost-vsock not found. Not adding a vsock device to the virtual machine.")
150             return False
151         elif e.errno in (errno.EPERM, errno.EACCES):
152             if log:
153                 logging.warning("Permission denied to access /dev/vhost-vsock. Not adding a vsock device to the virtual machine.")
154             return False
155 
156         raise e
157 
158     return True
159 
160 
161 def format_bytes(num_bytes: int) -> str:
162     if num_bytes >= 1024**3:
163         return f"{num_bytes/1024**3 :0.1f}G"
164     if num_bytes >= 1024**2:
165         return f"{num_bytes/1024**2 :0.1f}M"
166     if num_bytes >= 1024:
167         return f"{num_bytes/1024 :0.1f}K"
168 
169     return f"{num_bytes}B"
170 
171 
172 def make_executable(path: Path) -> None:
173     st = path.stat()
174     os.chmod(path, st.st_mode | stat.S_IEXEC)
175 
176 
177 def try_import(module: str) -> None:
178     try:
179         importlib.import_module(module)
180     except ModuleNotFoundError:
181         pass
182 
183 
184 @contextlib.contextmanager
185 def flock(path: Path) -> Iterator[int]:
186     fd = os.open(path, os.O_CLOEXEC|os.O_RDONLY)
187     try:
188         fcntl.fcntl(fd, fcntl.FD_CLOEXEC)
189         fcntl.flock(fd, fcntl.LOCK_EX)
190         yield fd
191     finally:
192         os.close(fd)
193 
194 
195 @contextlib.contextmanager
196 def scopedenv(env: Mapping[str, Any]) -> Iterator[None]:
197     old = copy.copy(os.environ)
198     os.environ |= env
199 
200     # python caches the default temporary directory so when we might modify TMPDIR we have to make sure it
201     # gets recalculated (see https://docs.python.org/3/library/tempfile.html#tempfile.tempdir).
202     tempfile.tempdir = None
203 
204     try:
205         yield
206     finally:
207         os.environ = old
208         tempfile.tempdir = None
209 
210 
211 class StrEnum(enum.Enum):
212     def __str__(self) -> str:
213         assert isinstance(self.value, str)
214         return self.value
215 
216     # Used by enum.auto() to get the next value.
217     @staticmethod
218     def _generate_next_value_(name: str, start: int, count: int, last_values: Sequence[str]) -> str:
219         return name.replace("_", "-")
220 
221     @classmethod
222     def values(cls) -> list[str]:
223         return list(map(str, cls))
224 
225 
226 def tar_binary() -> str:
227     # Some distros (Mandriva) install BSD tar as "tar", hence prefer
228     # "gtar" if it exists, which should be GNU tar wherever it exists.
229     # We are interested in exposing same behaviour everywhere hence
230     # it's preferable to use the same implementation of tar
231     # everywhere. In particular given the limited/different SELinux
232     # support in BSD tar and the different command line syntax
233     # compared to GNU tar.
234     return "gtar" if shutil.which("gtar") else "tar"
235 
236 
237 def one_zero(b: bool) -> str:
238     return "1" if b else "0"
239 
240 
241 @contextlib.contextmanager
242 def umask(mask: int) -> Iterator[None]:
243     old = os.umask(mask)
244     try:
245         yield
246     finally:
247         os.umask(old)
248
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/mkosi/archive.py b/mkosi/archive.py
--- a/mkosi/archive.py
+++ b/mkosi/archive.py
@@ -1,13 +1,28 @@
 # SPDX-License-Identifier: LGPL-2.1+
 
 import os
+import shutil
 from collections.abc import Iterable
 from pathlib import Path
 from typing import Optional
 
 from mkosi.log import log_step
 from mkosi.run import bwrap, finalize_passwd_mounts
-from mkosi.util import tar_binary
+
+
+def tar_binary() -> str:
+    # Some distros (Mandriva) install BSD tar as "tar", hence prefer
+    # "gtar" if it exists, which should be GNU tar wherever it exists.
+    # We are interested in exposing same behaviour everywhere hence
+    # it's preferable to use the same implementation of tar
+    # everywhere. In particular given the limited/different SELinux
+    # support in BSD tar and the different command line syntax
+    # compared to GNU tar.
+    return "gtar" if shutil.which("gtar") else "tar"
+
+
+def cpio_binary() -> str:
+    return "gcpio" if shutil.which("gcpio") else "cpio"
 
 
 def tar_exclude_apivfs_tmp() -> list[str]:
@@ -74,7 +89,7 @@
     log_step(f"Creating cpio archive {dst}…")
     bwrap(
         [
-            "cpio",
+            cpio_binary(),
             "--create",
             "--reproducible",
             "--null",
diff --git a/mkosi/util.py b/mkosi/util.py
--- a/mkosi/util.py
+++ b/mkosi/util.py
@@ -14,7 +14,6 @@
 import pwd
 import re
 import resource
-import shutil
 import stat
 import sys
 import tempfile
@@ -223,17 +222,6 @@
         return list(map(str, cls))
 
 
-def tar_binary() -> str:
-    # Some distros (Mandriva) install BSD tar as "tar", hence prefer
-    # "gtar" if it exists, which should be GNU tar wherever it exists.
-    # We are interested in exposing same behaviour everywhere hence
-    # it's preferable to use the same implementation of tar
-    # everywhere. In particular given the limited/different SELinux
-    # support in BSD tar and the different command line syntax
-    # compared to GNU tar.
-    return "gtar" if shutil.which("gtar") else "tar"
-
-
 def one_zero(b: bool) -> str:
     return "1" if b else "0"

{"golden_diff": "diff --git a/mkosi/archive.py b/mkosi/archive.py\n--- a/mkosi/archive.py\n+++ b/mkosi/archive.py\n@@ -1,13 +1,28 @@\n # SPDX-License-Identifier: LGPL-2.1+\n \n import os\n+import shutil\n from collections.abc import Iterable\n from pathlib import Path\n from typing import Optional\n \n from mkosi.log import log_step\n from mkosi.run import bwrap, finalize_passwd_mounts\n-from mkosi.util import tar_binary\n+\n+\n+def tar_binary() -> str:\n+    # Some distros (Mandriva) install BSD tar as \"tar\", hence prefer\n+    # \"gtar\" if it exists, which should be GNU tar wherever it exists.\n+    # We are interested in exposing same behaviour everywhere hence\n+    # it's preferable to use the same implementation of tar\n+    # everywhere. In particular given the limited/different SELinux\n+    # support in BSD tar and the different command line syntax\n+    # compared to GNU tar.\n+    return \"gtar\" if shutil.which(\"gtar\") else \"tar\"\n+\n+\n+def cpio_binary() -> str:\n+    return \"gcpio\" if shutil.which(\"gcpio\") else \"cpio\"\n \n \n def tar_exclude_apivfs_tmp() -> list[str]:\n@@ -74,7 +89,7 @@\n     log_step(f\"Creating cpio archive {dst}\u2026\")\n     bwrap(\n         [\n-            \"cpio\",\n+            cpio_binary(),\n             \"--create\",\n             \"--reproducible\",\n             \"--null\",\ndiff --git a/mkosi/util.py b/mkosi/util.py\n--- a/mkosi/util.py\n+++ b/mkosi/util.py\n@@ -14,7 +14,6 @@\n import pwd\n import re\n import resource\n-import shutil\n import stat\n import sys\n import tempfile\n@@ -223,17 +222,6 @@\n         return list(map(str, cls))\n \n \n-def tar_binary() -> str:\n-    # Some distros (Mandriva) install BSD tar as \"tar\", hence prefer\n-    # \"gtar\" if it exists, which should be GNU tar wherever it exists.\n-    # We are interested in exposing same behaviour everywhere hence\n-    # it's preferable to use the same implementation of tar\n-    # everywhere. In particular given the limited/different SELinux\n-    # support in BSD tar and the different command line syntax\n-    # compared to GNU tar.\n-    return \"gtar\" if shutil.which(\"gtar\") else \"tar\"\n-\n-\n def one_zero(b: bool) -> str:\n     return \"1\" if b else \"0\"\n", "issue": "Support libarchive cpio\nHi,\r\n\r\nplease support libarchive cpio(bsdcpio), which does not have `--reproducible`\r\n\r\nhttps://lists.gnu.org/archive/html/bug-cpio/2014-11/msg00000.html\r\n\r\nhttps://github.com/systemd/mkosi/blob/2c45d0effb1871750a2e9f897510d2745cb6d6b9/mkosi/__init__.py#L3489\n", "before_files": [{"content": "# SPDX-License-Identifier: LGPL-2.1+\n\nimport os\nfrom collections.abc import Iterable\nfrom pathlib import Path\nfrom typing import Optional\n\nfrom mkosi.log import log_step\nfrom mkosi.run import bwrap, finalize_passwd_mounts\nfrom mkosi.util import tar_binary\n\n\ndef tar_exclude_apivfs_tmp() -> list[str]:\n    return [\n        \"--exclude\", \"./dev/*\",\n        \"--exclude\", \"./proc/*\",\n        \"--exclude\", \"./sys/*\",\n        \"--exclude\", \"./tmp/*\",\n        \"--exclude\", \"./run/*\",\n        \"--exclude\", \"./var/tmp/*\",\n    ]\n\n\ndef make_tar(src: Path, dst: Path) -> None:\n    log_step(f\"Creating tar archive {dst}\u2026\")\n    bwrap(\n        [\n            tar_binary(),\n            \"--create\",\n            \"--file\", dst,\n            \"--directory\", src,\n            \"--acls\",\n            \"--selinux\",\n            \"--xattrs\",\n            \"--sparse\",\n            \"--force-local\",\n            *tar_exclude_apivfs_tmp(),\n            \".\",\n        ],\n        # Make sure tar uses user/group information from the root directory instead of the host.\n        options=finalize_passwd_mounts(src) if (src / \"etc/passwd\").exists() else [],\n    )\n\n\ndef extract_tar(src: Path, dst: Path, log: bool = True) -> None:\n    if log:\n        log_step(f\"Extracting tar archive {src}\u2026\")\n    bwrap(\n        [\n            tar_binary(),\n            \"--extract\",\n            \"--file\", src,\n            \"--directory\", dst,\n            \"--keep-directory-symlink\",\n            \"--no-overwrite-dir\",\n            \"--same-permissions\",\n            \"--same-owner\" if (dst / \"etc/passwd\").exists() else \"--numeric-owner\",\n            \"--same-order\",\n            \"--acls\",\n            \"--selinux\",\n            \"--xattrs\",\n            \"--force-local\",\n            *tar_exclude_apivfs_tmp(),\n        ],\n        # Make sure tar uses user/group information from the root directory instead of the host.\n        options=finalize_passwd_mounts(dst) if (dst / \"etc/passwd\").exists() else [],\n    )\n\n\ndef make_cpio(src: Path, dst: Path, files: Optional[Iterable[Path]] = None) -> None:\n    if not files:\n        files = src.rglob(\"*\")\n\n    log_step(f\"Creating cpio archive {dst}\u2026\")\n    bwrap(\n        [\n            \"cpio\",\n            \"--create\",\n            \"--reproducible\",\n            \"--null\",\n            \"--format=newc\",\n            \"--quiet\",\n            \"--directory\", src,\n            \"-O\", dst,\n        ],\n        input=\"\\0\".join(os.fspath(f.relative_to(src)) for f in files),\n        # Make sure tar uses user/group information from the root directory instead of the host.\n        options=finalize_passwd_mounts(dst),\n    )\n", "path": "mkosi/archive.py"}, {"content": "# SPDX-License-Identifier: LGPL-2.1+\n\nimport ast\nimport contextlib\nimport copy\nimport enum\nimport errno\nimport fcntl\nimport functools\nimport importlib\nimport itertools\nimport logging\nimport os\nimport pwd\nimport re\nimport resource\nimport shutil\nimport stat\nimport sys\nimport tempfile\nfrom collections.abc import Iterable, Iterator, Mapping, Sequence\nfrom pathlib import Path\nfrom typing import Any, Callable, Optional, TypeVar\n\nT = TypeVar(\"T\")\nV = TypeVar(\"V\")\n\n\ndef dictify(f: Callable[..., Iterator[tuple[T, V]]]) -> Callable[..., dict[T, V]]:\n    def wrapper(*args: Any, **kwargs: Any) -> dict[T, V]:\n        return dict(f(*args, **kwargs))\n\n    return functools.update_wrapper(wrapper, f)\n\n\n@dictify\ndef read_os_release() -> Iterator[tuple[str, str]]:\n    try:\n        filename = \"/etc/os-release\"\n        f = open(filename)\n    except FileNotFoundError:\n        filename = \"/usr/lib/os-release\"\n        f = open(filename)\n\n    with f:\n        for line_number, line in enumerate(f, start=1):\n            line = line.rstrip()\n            if not line or line.startswith(\"#\"):\n                continue\n            if (m := re.match(r\"([A-Z][A-Z_0-9]+)=(.*)\", line)):\n                name, val = m.groups()\n                if val and val[0] in \"\\\"'\":\n                    val = ast.literal_eval(val)\n                yield name, val\n            else:\n                print(f\"{filename}:{line_number}: bad line {line!r}\", file=sys.stderr)\n\n\ndef format_rlimit(rlimit: int) -> str:\n    limits = resource.getrlimit(rlimit)\n    soft = \"infinity\" if limits[0] == resource.RLIM_INFINITY else str(limits[0])\n    hard = \"infinity\" if limits[1] == resource.RLIM_INFINITY else str(limits[1])\n    return f\"{soft}:{hard}\"\n\n\ndef sort_packages(packages: Iterable[str]) -> list[str]:\n    \"\"\"Sorts packages: normal first, paths second, conditional third\"\"\"\n\n    m = {\"(\": 2, \"/\": 1}\n    sort = lambda name: (m.get(name[0], 0), name)\n    return sorted(packages, key=sort)\n\n\ndef flatten(lists: Iterable[Iterable[T]]) -> list[T]:\n    \"\"\"Flatten a sequence of sequences into a single list.\"\"\"\n    return list(itertools.chain.from_iterable(lists))\n\n\nclass InvokingUser:\n    @staticmethod\n    def _uid_from_env() -> Optional[int]:\n        uid = os.getenv(\"SUDO_UID\") or os.getenv(\"PKEXEC_UID\")\n        return int(uid) if uid is not None else None\n\n    @classmethod\n    def uid(cls) -> int:\n        return cls._uid_from_env() or os.getuid()\n\n    @classmethod\n    def uid_gid(cls) -> tuple[int, int]:\n        if (uid := cls._uid_from_env()) is not None:\n            gid = int(os.getenv(\"SUDO_GID\", pwd.getpwuid(uid).pw_gid))\n            return uid, gid\n        return os.getuid(), os.getgid()\n\n    @classmethod\n    def name(cls) -> str:\n        return pwd.getpwuid(cls.uid()).pw_name\n\n    @classmethod\n    def home(cls) -> Path:\n        return Path(f\"~{cls.name()}\").expanduser()\n\n    @classmethod\n    def is_running_user(cls) -> bool:\n        return cls.uid() == os.getuid()\n\n\n@contextlib.contextmanager\ndef chdir(directory: Path) -> Iterator[None]:\n    old = Path.cwd()\n\n    if old == directory:\n        yield\n        return\n\n    try:\n        os.chdir(directory)\n        yield\n    finally:\n        os.chdir(old)\n\n\ndef qemu_check_kvm_support(log: bool) -> bool:\n    # some CI runners may present a non-working KVM device\n    try:\n        os.close(os.open(\"/dev/kvm\", os.O_RDWR|os.O_CLOEXEC))\n    except OSError as e:\n        if e.errno == errno.ENOENT:\n            if log:\n                logging.warning(\"/dev/kvm not found. Not using KVM acceleration.\")\n            return False\n        elif e.errno in (errno.EPERM, errno.EACCES):\n            if log:\n                logging.warning(\"Permission denied to access /dev/kvm. Not using KVM acceleration\")\n            return False\n\n        raise e\n\n    return True\n\n\ndef qemu_check_vsock_support(log: bool) -> bool:\n    try:\n        os.close(os.open(\"/dev/vhost-vsock\", os.O_RDWR|os.O_CLOEXEC))\n    except OSError as e:\n        if e.errno == errno.ENOENT:\n            if log:\n                logging.warning(\"/dev/vhost-vsock not found. Not adding a vsock device to the virtual machine.\")\n            return False\n        elif e.errno in (errno.EPERM, errno.EACCES):\n            if log:\n                logging.warning(\"Permission denied to access /dev/vhost-vsock. Not adding a vsock device to the virtual machine.\")\n            return False\n\n        raise e\n\n    return True\n\n\ndef format_bytes(num_bytes: int) -> str:\n    if num_bytes >= 1024**3:\n        return f\"{num_bytes/1024**3 :0.1f}G\"\n    if num_bytes >= 1024**2:\n        return f\"{num_bytes/1024**2 :0.1f}M\"\n    if num_bytes >= 1024:\n        return f\"{num_bytes/1024 :0.1f}K\"\n\n    return f\"{num_bytes}B\"\n\n\ndef make_executable(path: Path) -> None:\n    st = path.stat()\n    os.chmod(path, st.st_mode | stat.S_IEXEC)\n\n\ndef try_import(module: str) -> None:\n    try:\n        importlib.import_module(module)\n    except ModuleNotFoundError:\n        pass\n\n\n@contextlib.contextmanager\ndef flock(path: Path) -> Iterator[int]:\n    fd = os.open(path, os.O_CLOEXEC|os.O_RDONLY)\n    try:\n        fcntl.fcntl(fd, fcntl.FD_CLOEXEC)\n        fcntl.flock(fd, fcntl.LOCK_EX)\n        yield fd\n    finally:\n        os.close(fd)\n\n\n@contextlib.contextmanager\ndef scopedenv(env: Mapping[str, Any]) -> Iterator[None]:\n    old = copy.copy(os.environ)\n    os.environ |= env\n\n    # python caches the default temporary directory so when we might modify TMPDIR we have to make sure it\n    # gets recalculated (see https://docs.python.org/3/library/tempfile.html#tempfile.tempdir).\n    tempfile.tempdir = None\n\n    try:\n        yield\n    finally:\n        os.environ = old\n        tempfile.tempdir = None\n\n\nclass StrEnum(enum.Enum):\n    def __str__(self) -> str:\n        assert isinstance(self.value, str)\n        return self.value\n\n    # Used by enum.auto() to get the next value.\n    @staticmethod\n    def _generate_next_value_(name: str, start: int, count: int, last_values: Sequence[str]) -> str:\n        return name.replace(\"_\", \"-\")\n\n    @classmethod\n    def values(cls) -> list[str]:\n        return list(map(str, cls))\n\n\ndef tar_binary() -> str:\n    # Some distros (Mandriva) install BSD tar as \"tar\", hence prefer\n    # \"gtar\" if it exists, which should be GNU tar wherever it exists.\n    # We are interested in exposing same behaviour everywhere hence\n    # it's preferable to use the same implementation of tar\n    # everywhere. In particular given the limited/different SELinux\n    # support in BSD tar and the different command line syntax\n    # compared to GNU tar.\n    return \"gtar\" if shutil.which(\"gtar\") else \"tar\"\n\n\ndef one_zero(b: bool) -> str:\n    return \"1\" if b else \"0\"\n\n\n@contextlib.contextmanager\ndef umask(mask: int) -> Iterator[None]:\n    old = os.umask(mask)\n    try:\n        yield\n    finally:\n        os.umask(old)\n", "path": "mkosi/util.py"}], "after_files": [{"content": "# SPDX-License-Identifier: LGPL-2.1+\n\nimport os\nimport shutil\nfrom collections.abc import Iterable\nfrom pathlib import Path\nfrom typing import Optional\n\nfrom mkosi.log import log_step\nfrom mkosi.run import bwrap, finalize_passwd_mounts\n\n\ndef tar_binary() -> str:\n    # Some distros (Mandriva) install BSD tar as \"tar\", hence prefer\n    # \"gtar\" if it exists, which should be GNU tar wherever it exists.\n    # We are interested in exposing same behaviour everywhere hence\n    # it's preferable to use the same implementation of tar\n    # everywhere. In particular given the limited/different SELinux\n    # support in BSD tar and the different command line syntax\n    # compared to GNU tar.\n    return \"gtar\" if shutil.which(\"gtar\") else \"tar\"\n\n\ndef cpio_binary() -> str:\n    return \"gcpio\" if shutil.which(\"gcpio\") else \"cpio\"\n\n\ndef tar_exclude_apivfs_tmp() -> list[str]:\n    return [\n        \"--exclude\", \"./dev/*\",\n        \"--exclude\", \"./proc/*\",\n        \"--exclude\", \"./sys/*\",\n        \"--exclude\", \"./tmp/*\",\n        \"--exclude\", \"./run/*\",\n        \"--exclude\", \"./var/tmp/*\",\n    ]\n\n\ndef make_tar(src: Path, dst: Path) -> None:\n    log_step(f\"Creating tar archive {dst}\u2026\")\n    bwrap(\n        [\n            tar_binary(),\n            \"--create\",\n            \"--file\", dst,\n            \"--directory\", src,\n            \"--acls\",\n            \"--selinux\",\n            \"--xattrs\",\n            \"--sparse\",\n            \"--force-local\",\n            *tar_exclude_apivfs_tmp(),\n            \".\",\n        ],\n        # Make sure tar uses user/group information from the root directory instead of the host.\n        options=finalize_passwd_mounts(src) if (src / \"etc/passwd\").exists() else [],\n    )\n\n\ndef extract_tar(src: Path, dst: Path, log: bool = True) -> None:\n    if log:\n        log_step(f\"Extracting tar archive {src}\u2026\")\n    bwrap(\n        [\n            tar_binary(),\n            \"--extract\",\n            \"--file\", src,\n            \"--directory\", dst,\n            \"--keep-directory-symlink\",\n            \"--no-overwrite-dir\",\n            \"--same-permissions\",\n            \"--same-owner\" if (dst / \"etc/passwd\").exists() else \"--numeric-owner\",\n            \"--same-order\",\n            \"--acls\",\n            \"--selinux\",\n            \"--xattrs\",\n            \"--force-local\",\n            *tar_exclude_apivfs_tmp(),\n        ],\n        # Make sure tar uses user/group information from the root directory instead of the host.\n        options=finalize_passwd_mounts(dst) if (dst / \"etc/passwd\").exists() else [],\n    )\n\n\ndef make_cpio(src: Path, dst: Path, files: Optional[Iterable[Path]] = None) -> None:\n    if not files:\n        files = src.rglob(\"*\")\n\n    log_step(f\"Creating cpio archive {dst}\u2026\")\n    bwrap(\n        [\n            cpio_binary(),\n            \"--create\",\n            \"--reproducible\",\n            \"--null\",\n            \"--format=newc\",\n            \"--quiet\",\n            \"--directory\", src,\n            \"-O\", dst,\n        ],\n        input=\"\\0\".join(os.fspath(f.relative_to(src)) for f in files),\n        # Make sure tar uses user/group information from the root directory instead of the host.\n        options=finalize_passwd_mounts(dst),\n    )\n", "path": "mkosi/archive.py"}, {"content": "# SPDX-License-Identifier: LGPL-2.1+\n\nimport ast\nimport contextlib\nimport copy\nimport enum\nimport errno\nimport fcntl\nimport functools\nimport importlib\nimport itertools\nimport logging\nimport os\nimport pwd\nimport re\nimport resource\nimport stat\nimport sys\nimport tempfile\nfrom collections.abc import Iterable, Iterator, Mapping, Sequence\nfrom pathlib import Path\nfrom typing import Any, Callable, Optional, TypeVar\n\nT = TypeVar(\"T\")\nV = TypeVar(\"V\")\n\n\ndef dictify(f: Callable[..., Iterator[tuple[T, V]]]) -> Callable[..., dict[T, V]]:\n    def wrapper(*args: Any, **kwargs: Any) -> dict[T, V]:\n        return dict(f(*args, **kwargs))\n\n    return functools.update_wrapper(wrapper, f)\n\n\n@dictify\ndef read_os_release() -> Iterator[tuple[str, str]]:\n    try:\n        filename = \"/etc/os-release\"\n        f = open(filename)\n    except FileNotFoundError:\n        filename = \"/usr/lib/os-release\"\n        f = open(filename)\n\n    with f:\n        for line_number, line in enumerate(f, start=1):\n            line = line.rstrip()\n            if not line or line.startswith(\"#\"):\n                continue\n            if (m := re.match(r\"([A-Z][A-Z_0-9]+)=(.*)\", line)):\n                name, val = m.groups()\n                if val and val[0] in \"\\\"'\":\n                    val = ast.literal_eval(val)\n                yield name, val\n            else:\n                print(f\"{filename}:{line_number}: bad line {line!r}\", file=sys.stderr)\n\n\ndef format_rlimit(rlimit: int) -> str:\n    limits = resource.getrlimit(rlimit)\n    soft = \"infinity\" if limits[0] == resource.RLIM_INFINITY else str(limits[0])\n    hard = \"infinity\" if limits[1] == resource.RLIM_INFINITY else str(limits[1])\n    return f\"{soft}:{hard}\"\n\n\ndef sort_packages(packages: Iterable[str]) -> list[str]:\n    \"\"\"Sorts packages: normal first, paths second, conditional third\"\"\"\n\n    m = {\"(\": 2, \"/\": 1}\n    sort = lambda name: (m.get(name[0], 0), name)\n    return sorted(packages, key=sort)\n\n\ndef flatten(lists: Iterable[Iterable[T]]) -> list[T]:\n    \"\"\"Flatten a sequence of sequences into a single list.\"\"\"\n    return list(itertools.chain.from_iterable(lists))\n\n\nclass InvokingUser:\n    @staticmethod\n    def _uid_from_env() -> Optional[int]:\n        uid = os.getenv(\"SUDO_UID\") or os.getenv(\"PKEXEC_UID\")\n        return int(uid) if uid is not None else None\n\n    @classmethod\n    def uid(cls) -> int:\n        return cls._uid_from_env() or os.getuid()\n\n    @classmethod\n    def uid_gid(cls) -> tuple[int, int]:\n        if (uid := cls._uid_from_env()) is not None:\n            gid = int(os.getenv(\"SUDO_GID\", pwd.getpwuid(uid).pw_gid))\n            return uid, gid\n        return os.getuid(), os.getgid()\n\n    @classmethod\n    def name(cls) -> str:\n        return pwd.getpwuid(cls.uid()).pw_name\n\n    @classmethod\n    def home(cls) -> Path:\n        return Path(f\"~{cls.name()}\").expanduser()\n\n    @classmethod\n    def is_running_user(cls) -> bool:\n        return cls.uid() == os.getuid()\n\n\n@contextlib.contextmanager\ndef chdir(directory: Path) -> Iterator[None]:\n    old = Path.cwd()\n\n    if old == directory:\n        yield\n        return\n\n    try:\n        os.chdir(directory)\n        yield\n    finally:\n        os.chdir(old)\n\n\ndef qemu_check_kvm_support(log: bool) -> bool:\n    # some CI runners may present a non-working KVM device\n    try:\n        os.close(os.open(\"/dev/kvm\", os.O_RDWR|os.O_CLOEXEC))\n    except OSError as e:\n        if e.errno == errno.ENOENT:\n            if log:\n                logging.warning(\"/dev/kvm not found. Not using KVM acceleration.\")\n            return False\n        elif e.errno in (errno.EPERM, errno.EACCES):\n            if log:\n                logging.warning(\"Permission denied to access /dev/kvm. Not using KVM acceleration\")\n            return False\n\n        raise e\n\n    return True\n\n\ndef qemu_check_vsock_support(log: bool) -> bool:\n    try:\n        os.close(os.open(\"/dev/vhost-vsock\", os.O_RDWR|os.O_CLOEXEC))\n    except OSError as e:\n        if e.errno == errno.ENOENT:\n            if log:\n                logging.warning(\"/dev/vhost-vsock not found. Not adding a vsock device to the virtual machine.\")\n            return False\n        elif e.errno in (errno.EPERM, errno.EACCES):\n            if log:\n                logging.warning(\"Permission denied to access /dev/vhost-vsock. Not adding a vsock device to the virtual machine.\")\n            return False\n\n        raise e\n\n    return True\n\n\ndef format_bytes(num_bytes: int) -> str:\n    if num_bytes >= 1024**3:\n        return f\"{num_bytes/1024**3 :0.1f}G\"\n    if num_bytes >= 1024**2:\n        return f\"{num_bytes/1024**2 :0.1f}M\"\n    if num_bytes >= 1024:\n        return f\"{num_bytes/1024 :0.1f}K\"\n\n    return f\"{num_bytes}B\"\n\n\ndef make_executable(path: Path) -> None:\n    st = path.stat()\n    os.chmod(path, st.st_mode | stat.S_IEXEC)\n\n\ndef try_import(module: str) -> None:\n    try:\n        importlib.import_module(module)\n    except ModuleNotFoundError:\n        pass\n\n\n@contextlib.contextmanager\ndef flock(path: Path) -> Iterator[int]:\n    fd = os.open(path, os.O_CLOEXEC|os.O_RDONLY)\n    try:\n        fcntl.fcntl(fd, fcntl.FD_CLOEXEC)\n        fcntl.flock(fd, fcntl.LOCK_EX)\n        yield fd\n    finally:\n        os.close(fd)\n\n\n@contextlib.contextmanager\ndef scopedenv(env: Mapping[str, Any]) -> Iterator[None]:\n    old = copy.copy(os.environ)\n    os.environ |= env\n\n    # python caches the default temporary directory so when we might modify TMPDIR we have to make sure it\n    # gets recalculated (see https://docs.python.org/3/library/tempfile.html#tempfile.tempdir).\n    tempfile.tempdir = None\n\n    try:\n        yield\n    finally:\n        os.environ = old\n        tempfile.tempdir = None\n\n\nclass StrEnum(enum.Enum):\n    def __str__(self) -> str:\n        assert isinstance(self.value, str)\n        return self.value\n\n    # Used by enum.auto() to get the next value.\n    @staticmethod\n    def _generate_next_value_(name: str, start: int, count: int, last_values: Sequence[str]) -> str:\n        return name.replace(\"_\", \"-\")\n\n    @classmethod\n    def values(cls) -> list[str]:\n        return list(map(str, cls))\n\n\ndef one_zero(b: bool) -> str:\n    return \"1\" if b else \"0\"\n\n\n@contextlib.contextmanager\ndef umask(mask: int) -> Iterator[None]:\n    old = os.umask(mask)\n    try:\n        yield\n    finally:\n        os.umask(old)\n", "path": "mkosi/util.py"}]}