823 lines
24 KiB
Python
823 lines
24 KiB
Python
"""Diagnose and narrowly repair the Bubblewrap host sandbox used by UMU."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import asdict, dataclass, replace
|
|
from datetime import datetime
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from typing import Callable, Mapping, Sequence
|
|
|
|
from . import __version__
|
|
from .paths import AppPaths
|
|
|
|
|
|
UID_MAP_PERMISSION_DENIED = "bwrap: setting up uid map: permission denied"
|
|
_BWRAP_ATTACHMENT_PATTERN = re.compile(
|
|
r"(?m)^[ \t]*(?:profile[ \t]+[^\s{]+[ \t]+)?/usr/bin/bwrap(?=[ \t{])"
|
|
)
|
|
_MAX_APPARMOR_PROFILE_BYTES = 1024 * 1024
|
|
|
|
# AppArmor upstream's restrictive Bubblewrap profile, pinned to a commit and
|
|
# its AppArmor 4.0 ABI so it works on Ubuntu 24.04. It grants Bubblewrap the
|
|
# setup capabilities it needs, then stacks children under unpriv_bwrap, where
|
|
# capabilities are explicitly denied. This is intentionally not an unconfined
|
|
# profile and does not disable Ubuntu's global user-namespace restriction.
|
|
BWRAP_APPARMOR_PROFILE = """# Installed by the PristonTale EU Linux launcher.
|
|
# Upstream source:
|
|
# https://gitlab.com/apparmor/apparmor/-/blob/8e431ebcd915216a03ebc8d01e72b1741bb2f855/profiles/apparmor/profiles/extras/bwrap-userns-restrict
|
|
#
|
|
# This profile allows almost everything and only exists to allow bwrap
|
|
# to work on a system with user namespace restrictions being enforced.
|
|
# bwrap is allowed access to user namespaces and capabilities within
|
|
# the user namespace, but its children do not have capabilities,
|
|
# blocking bwrap from being able to be used to arbitrarily by-pass the
|
|
# user namespace restrictions.
|
|
|
|
# Note: the bwrap child is stacked against the bwrap profile due to
|
|
# bwraps use of no-new-privs.
|
|
|
|
abi <abi/4.0>,
|
|
|
|
include <tunables/global>
|
|
|
|
profile bwrap /usr/bin/bwrap flags=(attach_disconnected,mediate_deleted) {
|
|
allow capability,
|
|
# not allow all, to allow for pix stack on systems that don't support
|
|
# rule priority.
|
|
#
|
|
# sadly we have to allow 'm' every where to allow children to work under
|
|
# profile stacking atm.
|
|
allow file rwlkm /{**,},
|
|
allow network,
|
|
allow unix,
|
|
allow ptrace,
|
|
allow signal,
|
|
allow mqueue,
|
|
allow io_uring,
|
|
allow userns,
|
|
allow mount,
|
|
allow umount,
|
|
allow pivot_root,
|
|
allow dbus,
|
|
|
|
# stacked like this due to no-new-privs restriction
|
|
# this will stack a target profile against bwrap and unpriv_bwrap
|
|
# Ideally
|
|
# - there would be a transition at userns creation first. This would allow
|
|
# for the bwrap profile to be tighter, and looser within the user
|
|
# ns. bwrap will still have to fairly loose until a transition at
|
|
# namespacing in general (not just user ns) is available.
|
|
# - there would be an independent second target as fallback
|
|
# This would allow for select target profiles to be used, and not
|
|
# necessarily stack the unpriv_bwrap in cases where this is desired
|
|
#
|
|
# the ix works here because stack will apply to ix fallback
|
|
# Ideally we would sanitize the environment across a privilege boundry
|
|
# (leaving bwarp into application) but flatpak etc use environment glibc
|
|
# sanitized environment variables as part of the sandbox setup.
|
|
allow pix /** -> &bwrap//&unpriv_bwrap,
|
|
|
|
# the local include should not be used without understanding the userns
|
|
# restriction.
|
|
# Site-specific additions and overrides. See local/README for details.
|
|
include if exists <local/bwrap-userns-restrict>
|
|
}
|
|
|
|
# The unpriv_bwrap profile is used to strip capabilities within the userns
|
|
profile unpriv_bwrap flags=(attach_disconnected,mediate_deleted) {
|
|
# not allow all, to allow for pix stack
|
|
allow file rwlkm /{**,},
|
|
allow network,
|
|
allow unix,
|
|
allow ptrace,
|
|
allow signal,
|
|
allow mqueue,
|
|
allow io_uring,
|
|
allow userns,
|
|
allow mount,
|
|
allow umount,
|
|
allow pivot_root,
|
|
allow dbus,
|
|
|
|
# bwrap profile does stacking against itself this will keep the target
|
|
# profile from having elevated privileges in the container.
|
|
# If done recursively the stack will remove any duplicate
|
|
allow pix /** -> &unpriv_bwrap,
|
|
|
|
audit deny capability,
|
|
|
|
# the local include should not be used without understanding the userns
|
|
# restriction.
|
|
# Site-specific additions and overrides. See local/README for details.
|
|
include if exists <local/unpriv_bwrap>
|
|
}
|
|
"""
|
|
BWRAP_APPARMOR_PROFILE_SHA256 = hashlib.sha256(
|
|
BWRAP_APPARMOR_PROFILE.encode("utf-8")
|
|
).hexdigest()
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ContainerHostFiles:
|
|
"""Fixed host paths, injectable so repair logic can be tested safely."""
|
|
|
|
bwrap: Path = Path("/usr/bin/bwrap")
|
|
probe_program: Path = Path("/usr/bin/true")
|
|
os_release: Path = Path("/etc/os-release")
|
|
apparmor_parser: Path = Path("/usr/sbin/apparmor_parser")
|
|
apparmor_directory: Path = Path("/etc/apparmor.d")
|
|
install: Path = Path("/usr/bin/install")
|
|
link: Path = Path("/usr/bin/ln")
|
|
remove: Path = Path("/usr/bin/rm")
|
|
obsolete_umu_profile_target: Path = Path(
|
|
"/usr/share/apparmor/extra-profiles/bwrap-userns-restrict"
|
|
)
|
|
apparmor_restrict_userns: Path = Path(
|
|
"/proc/sys/kernel/apparmor_restrict_unprivileged_userns"
|
|
)
|
|
unprivileged_userns_clone: Path = Path(
|
|
"/proc/sys/kernel/unprivileged_userns_clone"
|
|
)
|
|
max_user_namespaces: Path = Path("/proc/sys/user/max_user_namespaces")
|
|
|
|
@property
|
|
def distribution_profile(self) -> Path:
|
|
return self.apparmor_directory / "bwrap-userns-restrict"
|
|
|
|
@property
|
|
def managed_profile(self) -> Path:
|
|
return self.apparmor_directory / "bwrap-userns-restrict-ptlinux"
|
|
|
|
@property
|
|
def umu_profile_link(self) -> Path:
|
|
return self.apparmor_directory / "bwrap-userns-restrict-umu"
|
|
|
|
|
|
DEFAULT_CONTAINER_HOST_FILES = ContainerHostFiles()
|
|
CommandExecutor = Callable[[Sequence[str], float], subprocess.CompletedProcess[str]]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ContainerHostReport:
|
|
ready: bool
|
|
issue: str | None
|
|
distribution: str
|
|
bwrap_path: str | None
|
|
probe_returncode: int | None
|
|
probe_stderr: str
|
|
apparmor_restrict_unprivileged_userns: str | None
|
|
unprivileged_userns_clone: str | None
|
|
max_user_namespaces: str | None
|
|
active_profile_source: str | None
|
|
conflicting_bwrap_profile: str | None
|
|
broken_umu_profile_link: bool
|
|
repairable: bool
|
|
recommendation: str | None
|
|
repair_attempted: bool = False
|
|
repaired: bool = False
|
|
repair_actions: tuple[str, ...] = ()
|
|
repair_error: str | None = None
|
|
|
|
def to_dict(self) -> dict[str, object]:
|
|
return asdict(self)
|
|
|
|
|
|
def inspect_container_host(
|
|
*,
|
|
environment: Mapping[str, str] | None = None,
|
|
files: ContainerHostFiles = DEFAULT_CONTAINER_HOST_FILES,
|
|
command_executor: CommandExecutor | None = None,
|
|
timeout_seconds: float = 5.0,
|
|
) -> ContainerHostReport:
|
|
"""Run a harmless user-namespace probe and classify the exact failure."""
|
|
|
|
_ = environment
|
|
distribution_data = _read_os_release(files.os_release)
|
|
distribution = distribution_data.get("ID", "unknown")
|
|
bwrap_available = files.bwrap.is_file() and os.access(files.bwrap, os.X_OK)
|
|
restrict_userns = _read_setting(files.apparmor_restrict_userns)
|
|
userns_clone = _read_setting(files.unprivileged_userns_clone)
|
|
max_user_namespaces = _read_setting(files.max_user_namespaces)
|
|
profile = _active_profile_source(files)
|
|
broken_link = _known_broken_umu_profile_link(files)
|
|
returncode: int | None = None
|
|
stderr = ""
|
|
issue: str | None = None
|
|
if not bwrap_available:
|
|
issue = "bubblewrap-missing"
|
|
else:
|
|
command = (
|
|
str(files.bwrap),
|
|
"--ro-bind",
|
|
"/",
|
|
"/",
|
|
"--unshare-user",
|
|
"--",
|
|
str(files.probe_program),
|
|
)
|
|
try:
|
|
result = _run(command, timeout_seconds, command_executor)
|
|
returncode = result.returncode
|
|
stderr = _bounded_output(result.stderr)
|
|
except subprocess.TimeoutExpired as exc:
|
|
issue = "bubblewrap-probe-timeout"
|
|
stderr = _bounded_output(_timeout_stderr(exc))
|
|
except OSError as exc:
|
|
issue = "bubblewrap-probe-error"
|
|
stderr = str(exc)
|
|
if issue is None and returncode != 0:
|
|
folded = stderr.casefold()
|
|
if UID_MAP_PERMISSION_DENIED in folded:
|
|
issue = "uid-map-permission-denied"
|
|
elif userns_clone == "0":
|
|
issue = "unprivileged-userns-disabled"
|
|
elif max_user_namespaces == "0":
|
|
issue = "user-namespace-limit-zero"
|
|
else:
|
|
issue = "bubblewrap-probe-failed"
|
|
|
|
conflicting_profile = (
|
|
_find_conflicting_bwrap_profile(files)
|
|
if issue == "uid-map-permission-denied" or broken_link
|
|
else None
|
|
)
|
|
ubuntu_family = _is_ubuntu_family(distribution_data)
|
|
managed_collision = _managed_profile_collision(files)
|
|
repairable = (
|
|
ubuntu_family
|
|
and restrict_userns == "1"
|
|
and files.apparmor_parser.is_file()
|
|
and os.access(files.apparmor_parser, os.X_OK)
|
|
and bwrap_available
|
|
and not managed_collision
|
|
and conflicting_profile is None
|
|
and (issue == "uid-map-permission-denied" or broken_link)
|
|
)
|
|
recommendation = _recommendation(
|
|
issue,
|
|
repairable=repairable,
|
|
broken_link=broken_link,
|
|
conflicting_profile=conflicting_profile,
|
|
stderr=stderr,
|
|
)
|
|
return ContainerHostReport(
|
|
ready=issue is None,
|
|
issue=issue,
|
|
distribution=distribution,
|
|
bwrap_path=str(files.bwrap) if bwrap_available else None,
|
|
probe_returncode=returncode,
|
|
probe_stderr=stderr,
|
|
apparmor_restrict_unprivileged_userns=restrict_userns,
|
|
unprivileged_userns_clone=userns_clone,
|
|
max_user_namespaces=max_user_namespaces,
|
|
active_profile_source=str(profile) if profile else None,
|
|
conflicting_bwrap_profile=(
|
|
str(conflicting_profile) if conflicting_profile else None
|
|
),
|
|
broken_umu_profile_link=broken_link,
|
|
repairable=repairable,
|
|
recommendation=recommendation,
|
|
)
|
|
|
|
|
|
def repair_container_host(
|
|
*,
|
|
environment: Mapping[str, str] | None = None,
|
|
files: ContainerHostFiles = DEFAULT_CONTAINER_HOST_FILES,
|
|
command_executor: CommandExecutor | None = None,
|
|
privileged_executor: CommandExecutor | None = None,
|
|
) -> ContainerHostReport:
|
|
"""Repair only known Ubuntu AppArmor states, then repeat the probe."""
|
|
|
|
before = inspect_container_host(
|
|
environment=environment,
|
|
files=files,
|
|
command_executor=command_executor,
|
|
)
|
|
if before.ready and not before.broken_umu_profile_link:
|
|
return before
|
|
if not before.repairable:
|
|
return replace(
|
|
before,
|
|
repair_attempted=True,
|
|
repair_error=(
|
|
before.recommendation or "No safe automatic repair is available."
|
|
),
|
|
)
|
|
actions: list[str] = []
|
|
created_managed_profile = False
|
|
profile_loaded = False
|
|
selected_profile = _active_profile_source(files)
|
|
try:
|
|
if selected_profile is None:
|
|
selected_profile = files.managed_profile
|
|
_validate_and_install_managed_profile(
|
|
files,
|
|
command_executor=command_executor,
|
|
privileged_executor=privileged_executor,
|
|
)
|
|
created_managed_profile = True
|
|
actions.append(f"installed {selected_profile}")
|
|
else:
|
|
_validate_profile(
|
|
selected_profile,
|
|
files,
|
|
command_executor=command_executor,
|
|
)
|
|
|
|
loaded = _run_privileged(
|
|
(str(files.apparmor_parser), "-r", str(selected_profile)),
|
|
120.0,
|
|
privileged_executor,
|
|
)
|
|
_require_success(loaded, f"could not load AppArmor profile {selected_profile}")
|
|
profile_loaded = True
|
|
actions.append(f"loaded {selected_profile}")
|
|
|
|
if _known_broken_umu_profile_link(files):
|
|
linked = _run_privileged(
|
|
(
|
|
str(files.link),
|
|
"-sfn",
|
|
"--",
|
|
str(selected_profile),
|
|
str(files.umu_profile_link),
|
|
),
|
|
120.0,
|
|
privileged_executor,
|
|
)
|
|
_require_success(
|
|
linked,
|
|
"could not repair UMU's broken AppArmor profile link",
|
|
)
|
|
actions.append(f"repaired {files.umu_profile_link}")
|
|
except (OSError, RuntimeError, subprocess.TimeoutExpired) as exc:
|
|
rollback_error = None
|
|
if created_managed_profile and not profile_loaded:
|
|
rollback_error = _rollback_managed_profile(files, privileged_executor)
|
|
if rollback_error is None:
|
|
actions.append(f"rolled back {files.managed_profile}")
|
|
after_failure = inspect_container_host(
|
|
environment=environment,
|
|
files=files,
|
|
command_executor=command_executor,
|
|
)
|
|
error = str(exc)
|
|
if rollback_error:
|
|
error = f"{error}; rollback also failed: {rollback_error}"
|
|
return replace(
|
|
after_failure,
|
|
repair_attempted=True,
|
|
repair_actions=tuple(actions),
|
|
repair_error=error,
|
|
)
|
|
|
|
after = inspect_container_host(
|
|
environment=environment,
|
|
files=files,
|
|
command_executor=command_executor,
|
|
)
|
|
error = None
|
|
if not after.ready:
|
|
error = (
|
|
"The AppArmor profile loaded, but the Bubblewrap user-namespace probe "
|
|
"still fails. "
|
|
+ (after.recommendation or "Inspect the host security policy.")
|
|
)
|
|
return replace(
|
|
after,
|
|
repair_attempted=True,
|
|
repaired=after.ready,
|
|
repair_actions=tuple(actions),
|
|
repair_error=error,
|
|
)
|
|
|
|
|
|
def repair_container_host_after_update(paths: AppPaths) -> ContainerHostReport:
|
|
"""Run the repair once after activation, including upgrades by older code."""
|
|
|
|
status = inspect_container_host()
|
|
if status.ready and not status.broken_umu_profile_link:
|
|
return status
|
|
if not _launcher_was_updated_to_current_version(paths):
|
|
return status
|
|
marker = paths.container_host_repair_state
|
|
if _repair_was_attempted(marker, status):
|
|
return status
|
|
print(
|
|
"Launcher update: checking Steam Runtime sandbox permissions; "
|
|
"authentication may be requested for a narrow AppArmor repair.",
|
|
file=sys.stderr,
|
|
)
|
|
report = repair_container_host()
|
|
try:
|
|
_write_repair_marker(marker, report)
|
|
except OSError as exc:
|
|
print(
|
|
f"Warning: could not record the container-host repair attempt: {exc}",
|
|
file=sys.stderr,
|
|
)
|
|
return report
|
|
|
|
|
|
def container_host_error(report: ContainerHostReport) -> str:
|
|
if report.ready and report.broken_umu_profile_link:
|
|
return report.repair_error or report.recommendation or (
|
|
"UMU's AppArmor profile link remains broken."
|
|
)
|
|
message = (
|
|
"The Steam Runtime container cannot create its Bubblewrap user namespace."
|
|
)
|
|
if report.repair_error:
|
|
return f"{message} Host repair failed: {report.repair_error}"
|
|
if report.recommendation:
|
|
return f"{message} {report.recommendation}"
|
|
if report.probe_stderr:
|
|
return f"{message} Bubblewrap reported: {report.probe_stderr}"
|
|
return message
|
|
|
|
|
|
def _validate_and_install_managed_profile(
|
|
files: ContainerHostFiles,
|
|
*,
|
|
command_executor: CommandExecutor | None,
|
|
privileged_executor: CommandExecutor | None,
|
|
) -> None:
|
|
if files.managed_profile.exists() or files.managed_profile.is_symlink():
|
|
if _is_managed_profile(files.managed_profile):
|
|
_validate_profile(
|
|
files.managed_profile,
|
|
files,
|
|
command_executor=command_executor,
|
|
)
|
|
return
|
|
raise RuntimeError(
|
|
f"refusing to overwrite an unmanaged AppArmor file: {files.managed_profile}"
|
|
)
|
|
with tempfile.TemporaryDirectory(prefix="ptlinux-bwrap-profile-") as temporary:
|
|
candidate = Path(temporary) / "bwrap-userns-restrict-ptlinux"
|
|
candidate.write_text(BWRAP_APPARMOR_PROFILE, encoding="utf-8")
|
|
candidate.chmod(0o600)
|
|
_validate_profile(candidate, files, command_executor=command_executor)
|
|
installed = _run_privileged(
|
|
(
|
|
str(files.install),
|
|
"-m",
|
|
"0644",
|
|
"--",
|
|
str(candidate),
|
|
str(files.managed_profile),
|
|
),
|
|
120.0,
|
|
privileged_executor,
|
|
)
|
|
_require_success(installed, "could not install the managed AppArmor profile")
|
|
|
|
|
|
def _validate_profile(
|
|
profile: Path,
|
|
files: ContainerHostFiles,
|
|
*,
|
|
command_executor: CommandExecutor | None,
|
|
) -> None:
|
|
validated = _run(
|
|
(str(files.apparmor_parser), "-Q", "-K", str(profile)),
|
|
30.0,
|
|
command_executor,
|
|
)
|
|
_require_success(validated, f"AppArmor rejected profile {profile}")
|
|
|
|
|
|
def _rollback_managed_profile(
|
|
files: ContainerHostFiles,
|
|
privileged_executor: CommandExecutor | None,
|
|
) -> str | None:
|
|
try:
|
|
removed = _run_privileged(
|
|
(str(files.remove), "--", str(files.managed_profile)),
|
|
120.0,
|
|
privileged_executor,
|
|
)
|
|
_require_success(removed, "could not remove the newly installed profile")
|
|
except (OSError, RuntimeError, subprocess.TimeoutExpired) as exc:
|
|
return str(exc)
|
|
return None
|
|
|
|
|
|
def _run(
|
|
command: Sequence[str],
|
|
timeout_seconds: float,
|
|
executor: CommandExecutor | None,
|
|
) -> subprocess.CompletedProcess[str]:
|
|
if executor is not None:
|
|
return executor(tuple(command), timeout_seconds)
|
|
return subprocess.run(
|
|
list(command),
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=timeout_seconds,
|
|
check=False,
|
|
)
|
|
|
|
|
|
def _run_privileged(
|
|
command: Sequence[str],
|
|
timeout_seconds: float,
|
|
executor: CommandExecutor | None,
|
|
) -> subprocess.CompletedProcess[str]:
|
|
if executor is not None:
|
|
return executor(tuple(command), timeout_seconds)
|
|
prefix = _privilege_prefix()
|
|
return subprocess.run(
|
|
[*prefix, *command],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=timeout_seconds,
|
|
check=False,
|
|
)
|
|
|
|
|
|
def _privilege_prefix() -> tuple[str, ...]:
|
|
if os.geteuid() == 0:
|
|
return ()
|
|
if sys.stdin.isatty():
|
|
for candidate in (Path("/usr/bin/sudo"), Path("/usr/bin/doas")):
|
|
if candidate.is_file() and os.access(candidate, os.X_OK):
|
|
return (str(candidate),)
|
|
pkexec = Path("/usr/bin/pkexec")
|
|
if pkexec.is_file() and os.access(pkexec, os.X_OK):
|
|
return (str(pkexec),)
|
|
sudo = Path("/usr/bin/sudo")
|
|
if sudo.is_file() and os.access(sudo, os.X_OK):
|
|
return str(sudo), "-n"
|
|
raise RuntimeError(
|
|
"root access is required for the AppArmor repair, but no usable "
|
|
"sudo, doas, or pkexec command was found"
|
|
)
|
|
|
|
|
|
def _require_success(result: subprocess.CompletedProcess[str], message: str) -> None:
|
|
if result.returncode == 0:
|
|
return
|
|
detail = _bounded_output(result.stderr) or _bounded_output(result.stdout)
|
|
if detail:
|
|
raise RuntimeError(f"{message}: {detail}")
|
|
raise RuntimeError(f"{message} (exit status {result.returncode})")
|
|
|
|
|
|
def _active_profile_source(files: ContainerHostFiles) -> Path | None:
|
|
if files.distribution_profile.is_file():
|
|
return files.distribution_profile
|
|
if _trusted_umu_profile_link(files):
|
|
return files.umu_profile_link
|
|
if _is_managed_profile(files.managed_profile):
|
|
return files.managed_profile
|
|
return None
|
|
|
|
|
|
def _is_managed_profile(path: Path) -> bool:
|
|
if not path.is_file() or path.is_symlink():
|
|
return False
|
|
try:
|
|
content = path.read_bytes()
|
|
except OSError:
|
|
return False
|
|
return hashlib.sha256(content).hexdigest() == BWRAP_APPARMOR_PROFILE_SHA256
|
|
|
|
|
|
def _managed_profile_collision(files: ContainerHostFiles) -> bool:
|
|
path = files.managed_profile
|
|
return (path.exists() or path.is_symlink()) and not _is_managed_profile(path)
|
|
|
|
|
|
def _find_conflicting_bwrap_profile(files: ContainerHostFiles) -> Path | None:
|
|
known_names = {
|
|
files.distribution_profile.name,
|
|
files.managed_profile.name,
|
|
}
|
|
if _trusted_umu_profile_link(files):
|
|
known_names.add(files.umu_profile_link.name)
|
|
try:
|
|
candidates = sorted(
|
|
files.apparmor_directory.iterdir(),
|
|
key=lambda path: path.name,
|
|
)
|
|
except OSError:
|
|
return None
|
|
for candidate in candidates:
|
|
if candidate.name in known_names or not candidate.is_file():
|
|
continue
|
|
try:
|
|
if candidate.stat().st_size > _MAX_APPARMOR_PROFILE_BYTES:
|
|
continue
|
|
content = candidate.read_text(encoding="utf-8")
|
|
except (OSError, UnicodeError):
|
|
continue
|
|
policy = "\n".join(line.split("#", 1)[0] for line in content.splitlines())
|
|
if _BWRAP_ATTACHMENT_PATTERN.search(policy):
|
|
return candidate
|
|
return None
|
|
|
|
|
|
def _trusted_umu_profile_link(files: ContainerHostFiles) -> bool:
|
|
link = files.umu_profile_link
|
|
if not link.is_symlink() or not link.is_file():
|
|
return False
|
|
try:
|
|
target_text = os.readlink(link)
|
|
target = Path(target_text)
|
|
if not target.is_absolute():
|
|
target = link.parent / target
|
|
target = target.resolve()
|
|
allowed = {
|
|
files.distribution_profile.resolve(),
|
|
files.managed_profile.resolve(),
|
|
files.obsolete_umu_profile_target.resolve(),
|
|
}
|
|
except OSError:
|
|
return False
|
|
return target in allowed
|
|
|
|
|
|
def _known_broken_umu_profile_link(files: ContainerHostFiles) -> bool:
|
|
link = files.umu_profile_link
|
|
if not link.is_symlink():
|
|
return False
|
|
try:
|
|
target_text = os.readlink(link)
|
|
except OSError:
|
|
return False
|
|
if target_text != str(files.obsolete_umu_profile_target):
|
|
return False
|
|
return not files.obsolete_umu_profile_target.exists()
|
|
|
|
|
|
def _read_os_release(path: Path) -> dict[str, str]:
|
|
try:
|
|
lines = path.read_text(encoding="utf-8").splitlines()
|
|
except OSError:
|
|
return {}
|
|
result: dict[str, str] = {}
|
|
for line in lines:
|
|
if "=" not in line or line.startswith("#"):
|
|
continue
|
|
key, value = line.split("=", 1)
|
|
result[key] = value.strip().strip('"').strip("'")
|
|
return result
|
|
|
|
|
|
def _is_ubuntu_family(os_release: Mapping[str, str]) -> bool:
|
|
distribution = os_release.get("ID", "").casefold()
|
|
like = set(os_release.get("ID_LIKE", "").casefold().split())
|
|
return distribution == "ubuntu" or "ubuntu" in like
|
|
|
|
|
|
def _read_setting(path: Path) -> str | None:
|
|
try:
|
|
return path.read_text(encoding="ascii").strip()[:64]
|
|
except (OSError, UnicodeError):
|
|
return None
|
|
|
|
|
|
def _recommendation(
|
|
issue: str | None,
|
|
*,
|
|
repairable: bool,
|
|
broken_link: bool,
|
|
conflicting_profile: Path | None,
|
|
stderr: str,
|
|
) -> str | None:
|
|
if issue is None:
|
|
if broken_link:
|
|
return (
|
|
"UMU's AppArmor profile link is broken; run "
|
|
"`pristontale-eu container-host --repair` before rebooting."
|
|
)
|
|
return None
|
|
if issue == "bubblewrap-missing":
|
|
return (
|
|
"Install the distribution's bubblewrap package and run the check again."
|
|
)
|
|
if issue == "uid-map-permission-denied":
|
|
if repairable:
|
|
return (
|
|
"Ubuntu AppArmor denied Bubblewrap's UID map. Run "
|
|
"`pristontale-eu container-host --repair`; authentication is "
|
|
"required."
|
|
)
|
|
if conflicting_profile is not None:
|
|
return (
|
|
"Bubblewrap's UID map was denied, but an unrelated AppArmor profile "
|
|
f"already attaches to /usr/bin/bwrap: {conflicting_profile}. The "
|
|
"launcher will not install a competing profile automatically."
|
|
)
|
|
return (
|
|
"Bubblewrap's UID map was denied, but this host does not match the "
|
|
"narrow Ubuntu AppArmor repair. Inspect the loaded bwrap policy."
|
|
)
|
|
if issue == "unprivileged-userns-disabled":
|
|
return (
|
|
"kernel.unprivileged_userns_clone is disabled. The launcher will not "
|
|
"weaken this global security setting automatically."
|
|
)
|
|
if issue == "user-namespace-limit-zero":
|
|
return (
|
|
"user.max_user_namespaces is zero. The launcher will not change this "
|
|
"global limit automatically."
|
|
)
|
|
if issue == "bubblewrap-probe-timeout":
|
|
return (
|
|
"The Bubblewrap sandbox probe timed out; inspect the host sandbox policy."
|
|
)
|
|
if stderr:
|
|
return f"Bubblewrap's sandbox probe failed: {stderr}"
|
|
return "Bubblewrap's sandbox probe failed without a diagnostic message."
|
|
|
|
|
|
def _timeout_stderr(error: subprocess.TimeoutExpired) -> str:
|
|
value = error.stderr
|
|
if isinstance(value, bytes):
|
|
return value.decode("utf-8", errors="replace")
|
|
return value or ""
|
|
|
|
|
|
def _bounded_output(value: str | None, limit: int = 2048) -> str:
|
|
if not value:
|
|
return ""
|
|
cleaned = value.strip()
|
|
if len(cleaned) <= limit:
|
|
return cleaned
|
|
return cleaned[:limit] + "…"
|
|
|
|
|
|
def _launcher_was_updated_to_current_version(paths: AppPaths) -> bool:
|
|
try:
|
|
data = paths.launcher_update_state.read_bytes()
|
|
if len(data) > 64 * 1024:
|
|
return False
|
|
payload = json.loads(data)
|
|
except (OSError, UnicodeError, json.JSONDecodeError):
|
|
return False
|
|
return isinstance(payload, dict) and payload.get("to_version") == __version__
|
|
|
|
|
|
def _repair_was_attempted(path: Path, status: ContainerHostReport) -> bool:
|
|
try:
|
|
data = path.read_bytes()
|
|
if len(data) > 64 * 1024:
|
|
return False
|
|
payload = json.loads(data)
|
|
except (OSError, UnicodeError, json.JSONDecodeError):
|
|
return False
|
|
return (
|
|
isinstance(payload, dict)
|
|
and payload.get("schema") == 1
|
|
and payload.get("launcher_version") == __version__
|
|
and payload.get("profile_sha256") == BWRAP_APPARMOR_PROFILE_SHA256
|
|
and payload.get("issue") == status.issue
|
|
and payload.get("broken_umu_profile_link") == status.broken_umu_profile_link
|
|
)
|
|
|
|
|
|
def _write_repair_marker(path: Path, report: ContainerHostReport) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
payload = {
|
|
"schema": 1,
|
|
"attempted_at": datetime.now().astimezone().isoformat(),
|
|
"launcher_version": __version__,
|
|
"profile_sha256": BWRAP_APPARMOR_PROFILE_SHA256,
|
|
"issue": report.issue,
|
|
"broken_umu_profile_link": report.broken_umu_profile_link,
|
|
"ready": report.ready,
|
|
"repair_error": report.repair_error,
|
|
}
|
|
with tempfile.NamedTemporaryFile(
|
|
mode="w",
|
|
encoding="utf-8",
|
|
dir=path.parent,
|
|
prefix=f".{path.name}.",
|
|
delete=False,
|
|
) as handle:
|
|
temporary = Path(handle.name)
|
|
try:
|
|
json.dump(payload, handle, indent=2, sort_keys=True)
|
|
handle.write("\n")
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
except BaseException:
|
|
temporary.unlink(missing_ok=True)
|
|
raise
|
|
try:
|
|
temporary.chmod(0o600)
|
|
temporary.replace(path)
|
|
except BaseException:
|
|
temporary.unlink(missing_ok=True)
|
|
raise
|