Added bubblewrap / apparmor fixes and diagnostic tools for troubleshooting it on related distros.
This commit is contained in:
822
ptlinux/container_host.py
Normal file
822
ptlinux/container_host.py
Normal file
@@ -0,0 +1,822 @@
|
||||
"""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
|
||||
Reference in New Issue
Block a user