Skip to content
Draft
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 5 additions & 1 deletion vmupdate/agent/entrypoint.py
Original file line number Diff line number Diff line change
Expand Up @@ -90,12 +90,16 @@ def get_package_manager(
elif os_data["os_family"] == "ArchLinux":
from source.pacman.pacman_cli import PACMANCLI as PackageManager

print("Progress reporting not supported.", flush=True)
elif os_data["os_family"] == "Guix":
from source.guix.guix_cli import GUIXCLI as PackageManager

print("Progress reporting not supported.", flush=True)
elif os_data["os_family"] == "Qubes":
PackageManager = import_dom0_package_manager(os_data, log, no_progress)
else:
raise NotImplementedError(
"Only Debian, RedHat and ArchLinux based OS is supported."
"Only Debian, RedHat, ArchLinux, Qubes and Guix based OS is supported."
)

pkg_mng = PackageManager(log_handler, log_level, agent_type)
Expand Down
20 changes: 20 additions & 0 deletions vmupdate/agent/source/guix/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,20 @@
# coding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org
#
# Copyright (C) 2026 The Qubes OS Project
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301,
# USA.
318 changes: 318 additions & 0 deletions vmupdate/agent/source/guix/guix_cli.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,318 @@
# coding=utf-8
#
# The Qubes OS Project, https://www.qubes-os.org
#
# Copyright (C) 2026 The Qubes OS Project
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301,
# USA.

import os
import shlex
import shutil
import subprocess
import sys
from typing import Dict, List, Optional

from source.common.package_manager import AgentType, PackageManager
from source.common.process_result import ProcessResult
from source.common.exit_codes import EXIT


class GUIXCLI(PackageManager):
PROGRESS_REPORTING = False

TIME_MACHINE_BRANCH = "master"
SYSTEM_CONFIG = "/etc/config.scm"
SYSTEM_PROFILE = "/run/current-system/profile"
SERVICE_DIR = "/run/qubes-service"
TIME_MACHINE_ENVIRONMENT = (
"HOME=/tmp",
"XDG_CONFIG_HOME=/tmp/qubes-vm-update-guix-config",
"XDG_CACHE_HOME=/tmp/qubes-vm-update-guix-cache",
)
MANIFEST_SEPARATOR = "|"
STATE_PATHS = {
"guix-system": "/run/current-system",
}
GUIX_CANDIDATES = (
"/run/qubes/bin/guix",
"/root/.config/guix/current/bin/guix",
"/var/guix/profiles/per-user/root/current-guix/bin/guix",
"/run/current-system/profile/bin/guix",
)

def __init__(
self, log_handler, log_level, agent_type: AgentType
):
super().__init__(log_handler, log_level, agent_type)
self.package_manager = self._find_guix(self.GUIX_CANDIDATES)

def _find_guix(self, candidates) -> str:
for path in candidates:
if os.access(path, os.X_OK):
return path

path = shutil.which("guix")
if path is not None:
return path

raise RuntimeError("Package manager not found!")

def _uses_qubes_update_proxy(self) -> bool:
# updates-proxy-setup marks update clients. A VM with
# qubes-updates-proxy provides the proxy and must not route its own
# Guix traffic back through the local forwarder.
return (
os.path.exists(os.path.join(self.SERVICE_DIR,
"updates-proxy-setup"))
and not os.path.exists(os.path.join(self.SERVICE_DIR,
"qubes-updates-proxy"))
)

def _with_time_machine_environment(
self, command: List[str]
) -> List[str]:
env = list(self.TIME_MACHINE_ENVIRONMENT)

if self._uses_qubes_update_proxy():
proxy = "http://127.0.0.1:8082/"
no_proxy = "127.0.0.1,localhost"
env.extend([
f"http_proxy={proxy}",
f"https_proxy={proxy}",
f"HTTP_PROXY={proxy}",
f"HTTPS_PROXY={proxy}",
f"all_proxy={proxy}",
f"ALL_PROXY={proxy}",
f"no_proxy={no_proxy}",
f"NO_PROXY={no_proxy}",
])

return ["env", *env, *command]

def _run_guix(self, command: List[str]) -> ProcessResult:
result = self.run_cmd(self._with_time_machine_environment(command))
if result and not (result.out.strip() or result.err.strip()):
result.err = (
f"Guix command failed with exit code {result.code}: "
f"{shlex.join(command)}"
)
result.posted = False
return result

def refresh(self, hard_fail: bool) -> ProcessResult:
"""
Refresh Guix channel metadata for system reconfiguration.

Use guix time-machine so Qubes vmupdate does not mutate root's Guix
checkout as package-manager state. The update target is the Guix
System generation produced by the later reconfigure step.
"""
cmd = [
self.package_manager,
"time-machine",
f"--branch={self.TIME_MACHINE_BRANCH}",
"--",
"describe",
]
print(
f"Refreshing Guix channel metadata from "
f"{self.TIME_MACHINE_BRANCH}.",
flush=True,
)
return self._run_guix(cmd)

def get_packages(self) -> Dict[str, List[str]]:
"""
Report Guix System profile entries as update state.

The shared updater summary compares package/version dictionaries.
Guix profiles expose manifest entries as name, version, output, and
store path, so report each system profile output plus the current
system generation symlink.
"""
packages: Dict[str, List[str]] = {}
for name, path in self.STATE_PATHS.items():
if os.path.exists(path):
packages[name] = [os.path.realpath(path)]

result = self._list_installed_packages()
if result:
self.log.warning(
"Unable to list Guix system profile packages: %s",
result.err or result.out,
)
return packages

for line in result.out.splitlines():
if not line.strip():
continue
entry = self._parse_manifest_entry(line)
if entry is None:
self.log.warning(
"Ignoring unexpected Guix package entry: %s", line
)
continue
name, version, output, store_path = entry
package = f"{name}:{output}"
packages.setdefault(package, []).append(
f"{version} {store_path}"
)
return packages

def _list_installed_packages(self) -> ProcessResult:
command = [
self.package_manager,
"package",
f"--profile={self.SYSTEM_PROFILE}",
"--list-installed",
]
self.log.debug("run command: %s", " ".join(command))
with subprocess.Popen(
command,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
) as proc:
out, err = proc.communicate()
out = out.replace(b"\t", self.MANIFEST_SEPARATOR.encode())
result = ProcessResult.from_untrusted_out_err(out, err)
result.code = proc.returncode
self.log.debug("command exit code: %i", result.code)
return result

@staticmethod
def _parse_manifest_entry(line):
if GUIXCLI.MANIFEST_SEPARATOR in line:
cols = [
col.strip()
for col in line.split(GUIXCLI.MANIFEST_SEPARATOR, 3)
]
if len(cols) == 4 and all(cols):
return tuple(cols)

store_marker = "/gnu/store/"
store_start = line.find(store_marker)
if store_start != -1:
store_path = line[store_start:].strip()
fields = line[:store_start].strip().split()
if len(fields) >= 3:
name, version, output = fields[:3]
return name, version, output, store_path
entry = GUIXCLI._parse_sanitized_manifest_entry(
fields, store_path
)
if entry is not None:
return entry

cols = line.split(None, 3)
if len(cols) == 4:
return tuple(cols)

return None

@staticmethod
def _parse_sanitized_manifest_entry(fields, store_path):
"""
Recover fields after ProcessResult stripped tabs from Guix output.

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Looks like a lot of effort to un-break that output. Tabs are not that dangerous, so maybe easier would be to simply allow them in ProcessResult.sanitize_output function?


Guix separates name, version, output, and store path with tabs.
ProcessResult removes tabs from untrusted output before callers parse
it. When a column is wider than Guix's padding, adjacent fields can be
glued together; the store item basename keeps the name-version boundary.
"""
store_item = os.path.basename(store_path)
try:
_store_hash, store_name_version = store_item.split("-", 1)
except ValueError:
return None

if len(fields) == 2:
first, second = fields
if store_name_version.startswith(f"{first}-"):
version = store_name_version[len(first) + 1:]
if second.startswith(version):
output = second[len(version):]
if output:
return first, version, output, store_path

for index in range(1, len(first)):
name = first[:index]
version = first[index:]
if f"{name}-{version}" == store_name_version:
return name, version, second, store_path

return None

def get_action(self, remove_obsolete) -> List[str]:
"""
Kept for the PackageManager interface; upgrade_internal runs the
reconfiguration through guix time-machine.
"""
return [
"time-machine",
f"--branch={self.TIME_MACHINE_BRANCH}",
"--",
"system",
"reconfigure",
"--no-bootloader",
self.SYSTEM_CONFIG,
]

def upgrade_internal(self, remove_obsolete: bool) -> ProcessResult:
if not os.path.exists(self.SYSTEM_CONFIG):
return ProcessResult(
EXIT.ERR_VM_UPDATE,
err=f"missing Guix system configuration: {self.SYSTEM_CONFIG}")

cmd = [self.package_manager, *self.get_action(remove_obsolete)]
print(
f"Reconfiguring Guix System from {self.SYSTEM_CONFIG} "
f"using {self.TIME_MACHINE_BRANCH}.",
flush=True,
)
result = self._run_guix(cmd)
if not result:
print("Reconfigured Guix System.", flush=True)
else:
print(
"Guix System reconfiguration failed.",
file=sys.stderr,
flush=True,
)
return result

def install_requirements(
self,
requirements: Optional[Dict[str, str]],
curr_pkg: Dict[str, List[str]]
) -> ProcessResult:
"""
Qubes vmupdate plugins do not currently declare Guix requirements.
Avoid installing ad hoc root profile packages as hidden update policy.
"""
if requirements:
packages = ", ".join(sorted(requirements))
return ProcessResult(
EXIT.ERR_VM_PRE,
err=f"Guix vmupdate requirements are unsupported: {packages}")
return ProcessResult()

def clean(self) -> int:
"""
Keep Guix generations for rollback; do not collect garbage implicitly.
"""
return EXIT.OK
5 changes: 4 additions & 1 deletion vmupdate/agent/source/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ def get_os_data(logger: Optional = None) -> Dict[str, Any]:
name: "Linux" or a string identifying the operating system,
codename (optional): an operating system release code name,
release (optional): version string,
os_family: "Unknown", "RedHat", "Debian", "ArchLinux".
os_family: "Unknown", "RedHat", "Debian", "ArchLinux", "Guix".
"""
data = {}

Expand Down Expand Up @@ -69,6 +69,9 @@ def get_os_data(logger: Optional = None) -> Dict[str, Any]:
if "arch" in family:
data["os_family"] = "ArchLinux"

if "guix" in family:
data["os_family"] = "Guix"

return data


Expand Down
Loading