Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
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
14 changes: 10 additions & 4 deletions vmupdate/agent/source/common/package_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -247,7 +247,7 @@ def _print_changes(self, changes):
if changes["installed"]:
for pkg in sorted(changes["installed"]):
result.out += self._print_to_string(
pkg, changes["installed"][pkg]
pkg, self._format_versions(changes["installed"][pkg])
)
else:
result.out += self._print_to_string("None")
Expand All @@ -257,9 +257,9 @@ def _print_changes(self, changes):
for pkg in sorted(changes["updated"]):
result.out += self._print_to_string(
pkg,
str(changes["updated"][pkg]["old"])[2:-2]
self._format_versions(changes["updated"][pkg]["old"])
+ " -> "
+ str(changes["updated"][pkg]["new"])[2:-2],
+ self._format_versions(changes["updated"][pkg]["new"]),
)
else:
result.out += self._print_to_string("None")
Expand All @@ -268,12 +268,18 @@ def _print_changes(self, changes):
if changes["removed"]:
for pkg in sorted(changes["removed"]):
result.out += self._print_to_string(
pkg, changes["removed"][pkg]
pkg, self._format_versions(changes["removed"][pkg])
)
else:
result.out += self._print_to_string("None")
return result

@staticmethod
def _format_versions(versions):
if isinstance(versions, (list, tuple)):
return ", ".join(str(version) for version in versions)
return str(versions)

@staticmethod
def _print_to_string(*args, **kwargs):
strio = io.StringIO()
Expand Down
4 changes: 3 additions & 1 deletion vmupdate/agent/source/common/process_result.py
Original file line number Diff line number Diff line change
Expand Up @@ -88,7 +88,9 @@ def sanitize_output(untrusted_bytes: bytes, single: bool = False) -> str:
[
c
for c in untrusted_str
if 0x20 <= ord(c) <= 0x7E or (c == "\n" and not single)
if 0x20 <= ord(c) <= 0x7E
or c == "\t"
or (c == "\n" and not single)
]
)

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.
260 changes: 260 additions & 0 deletions vmupdate/agent/source/guix/guix_cli.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,260 @@
# 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

SYSTEM_CONFIG = "/etc/config.scm"
SYSTEM_PROFILE = "/run/current-system/profile"
SERVICE_DIR = "/run/qubes-service"
# "guix pull" must run with root's real HOME so it persists root's
# current-guix profile; an ephemeral HOME would discard the pulled Qubes
# channel and leave the components pinned to the baked-in template.
GUIX_ENVIRONMENT = (
"HOME=/root",
)
# reconfigure must run via the just-pulled guix so it loads the newly
# pulled Qubes channel modules, not the image's baked-in guix.
PULLED_GUIX = "/var/guix/profiles/per-user/root/current-guix/bin/guix"
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_guix_environment(
self, command: List[str]
) -> List[str]:
env = list(self.GUIX_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_guix_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:
"""
Pull the channels declared in /etc/guix/channels.scm.

This advances both Guix itself and the Qubes channel that pins the
Qubes component versions, so the later reconfigure builds the updated
components instead of the ones baked into the installed template.
"""
cmd = [
self.package_manager,
"pull",
]
print("Pulling Guix channels for system reconfiguration.", 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()
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):
# guix prints tab-separated name, version, output, store path.
cols = [col.strip() for col in line.split("\t")]
if len(cols) != 4 or not all(cols):
cols = line.split(None, 3)
if len(cols) == 4 and all(col.strip() for col in cols):
return tuple(col.strip() for col in cols)
return None

def get_action(self, remove_obsolete) -> List[str]:
"""
Kept for the PackageManager interface; upgrade_internal reconfigures
the system from the pulled channels.
"""
return [
"system",
"reconfigure",
"--no-bootloader",
self.SYSTEM_CONFIG,
]

def _reconfigure_guix(self) -> str:
if os.access(self.PULLED_GUIX, os.X_OK):
return self.PULLED_GUIX
return self.package_manager

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._reconfigure_guix(), *self.get_action(remove_obsolete)]
print(
f"Reconfiguring Guix System from {self.SYSTEM_CONFIG}.",
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