Compare commits

..

1 Commits

Author SHA1 Message Date
12f98c759a add basestrap module 2023-12-28 02:58:55 +01:00
6 changed files with 623 additions and 193 deletions

View File

@@ -0,0 +1,66 @@
# SPDX-FileCopyrightText: no
# SPDX-License-Identifier: CC0-1.0
#
# The configuration for the package manager starts with the
# *backend* key, which picks one of the backends to use.
# In `main.py` there is a base class `PackageManager`.
# Implementations must subclass that and set a (class-level)
# property *backend* to the name of the backend (e.g. "dummy").
# That property is used to match against the *backend* key here.
#
# You will have to add such a class for your package manager.
# It is fairly simple Python code. The API is described in the
# abstract methods in class `PackageManager`. Mostly, the only
# trick is to figure out the correct commands to use, and in particular,
# whether additional switches are required or not. Some package managers
# have more installer-friendly defaults than others, e.g., DNF requires
# passing --disablerepo=* -C to allow removing packages without Internet
# connectivity, and it also returns an error exit code if the package did
# not exist to begin with.
---
#
# Which package manager to use, options are:
# - pacman - Pacman
#
# Not actually a package manager, but suitable for testing:
# - dummy - Dummy manager, only logs
#
backend: dummy
# pacman specific options
#
# *num_retries* should be a positive integer which specifies the
# number of times the call to pacman will be retried in the event of a
# failure. If it is missing, it will be set to 0.
#
# *disable_download_timeout* is a boolean that, when true, includes
# the flag --disable-download-timeout on calls to pacman. When missing,
# false is assumed.
#
# *needed_only* is a boolean that includes the pacman argument --needed
# when set to true. If missing, false is assumed.
# *handle_keyrings* is a boolean that includes initializing and populating keyrings
# when set to true. If missing, false is assumed.
pacman:
num_retries: 0
disable_download_timeout: false
needed_only: false
handle_keyrings: false
requirements:
- name: /etc
mode: "0o755"
- name: /var/cache/pacman/pkg
mode: "0o755"
- name: /var/lib/pacman
mode: "0o755"
keyrings:
- artix
# the artix base package allows selection of the init system tied to elogind
# this option is artix specific
# base_init: elogind
operations:
- install:
- base

View File

@@ -0,0 +1,550 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# === This file is part of Calamares - <https://calamares.io> ===
#
# SPDX-FileCopyrightText: 2014 Pier Luigi Fiorini <pierluigi.fiorini@gmail.com>
# SPDX-FileCopyrightText: 2015-2017 Teo Mrnjavac <teo@kde.org>
# SPDX-FileCopyrightText: 2016-2017 Kyle Robbertze <kyle@aims.ac.za>
# SPDX-FileCopyrightText: 2017 Alf Gaida <agaida@siduction.org>
# SPDX-FileCopyrightText: 2018 Adriaan de Groot <groot@kde.org>
# SPDX-FileCopyrightText: 2018 Philip Müller <philm@manjaro.org>
# SPDX-FileCopyrightText: 2023 Artoo <artoo@artixlinux.org>
# SPDX-License-Identifier: GPL-3.0-or-later
#
# Calamares is Free Software: see the License-Identifier above.
#
import abc
from string import Template
import os, shutil, subprocess, sys
import libcalamares
from libcalamares.utils import host_env_process_output, target_env_process_output
from libcalamares.utils import gettext_path, gettext_languages
from os.path import join
import gettext
_translation = gettext.translation("calamares-python",
localedir=gettext_path(),
languages=gettext_languages(),
fallback=True)
_ = _translation.gettext
_n = _translation.ngettext
total_packages = 0 # For the entire job
completed_packages = 0 # Done so far for this job
group_packages = 0 # One group of packages from an -install or -remove entry
# A PM object may set this to a string (take care of translations!)
# to override the string produced by pretty_status_message()
custom_status_message = None
INSTALL = object()
REMOVE = object()
mode_packages = None # Changes to INSTALL or REMOVE
def _change_mode(mode):
global mode_packages
mode_packages = mode
libcalamares.job.setprogress(completed_packages * 1.0 / total_packages)
def pretty_name():
return _("Install packages.")
def pretty_status_message():
if custom_status_message is not None:
return custom_status_message
if not group_packages:
if (total_packages > 0):
# Outside the context of an operation
s = _("Processing packages (%(count)d / %(total)d)")
else:
s = _("Install packages.")
elif mode_packages is INSTALL:
s = _n("Installing one package.",
"Installing %(num)d packages.", group_packages)
elif mode_packages is REMOVE:
s = _n("Removing one package.",
"Removing %(num)d packages.", group_packages)
else:
# No mode, generic description
s = _("Install packages.")
return s % {"num": group_packages,
"count": completed_packages,
"total": total_packages}
class PackageManager(metaclass=abc.ABCMeta):
"""
Package manager base class. A subclass implements package management
for a specific backend, and must have a class property `backend`
with the string identifier for that backend.
Subclasses are collected below to populate the list of possible
backends.
"""
backend = None
@abc.abstractmethod
def install(self, pkgs, from_local=False):
"""
Install a list of packages (named) into the system.
Although this handles lists, in practice it is called
with one package at a time.
@param pkgs: list[str]
list of package names
@param from_local: bool
if True, then these are local packages (on disk) and the
pkgs names are paths.
"""
pass
@abc.abstractmethod
def remove(self, pkgs):
"""
Removes packages.
@param pkgs: list[str]
list of package names
"""
pass
def run(self, script):
if script != "":
host_env_process_output(script.split(" "))
def install_package(self, packagedata, from_local=False):
"""
Install a package from a single entry in the install list.
This can be either a single package name, or an object
with pre- and post-scripts. If @p packagedata is a dict,
it is assumed to follow the documented structure.
@param packagedata: str|dict
@param from_local: bool
see install.from_local
"""
if isinstance(packagedata, str):
self.install([packagedata], from_local=from_local)
else:
self.run(packagedata["pre-script"])
self.install([packagedata["package"]], from_local=from_local)
self.run(packagedata["post-script"])
def remove_package(self, packagedata):
"""
Remove a package from a single entry in the remove list.
This can be either a single package name, or an object
with pre- and post-scripts. If @p packagedata is a dict,
it is assumed to follow the documented structure.
@param packagedata: str|dict
"""
if isinstance(packagedata, str):
self.remove([packagedata])
else:
self.run(packagedata["pre-script"])
self.remove([packagedata["package"]])
self.run(packagedata["post-script"])
def operation_install(self, package_list, from_local=False):
"""
Installs the list of packages named in @p package_list .
These can be strings -- plain package names -- or
structures (with a pre- and post-install step).
This operation is called for "critical" packages,
which are expected to succeed, or fail, all together.
However, if there are packages with pre- or post-scripts,
then packages are installed one-by-one instead.
NOTE: package managers may reimplement this method
NOTE: exceptions are expected to leave this method, to indicate
failure of the installation.
"""
if all([isinstance(x, str) for x in package_list]):
self.install(package_list, from_local=from_local)
else:
for package in package_list:
self.install_package(package, from_local=from_local)
def operation_try_install(self, package_list):
"""
Installs the list of packages named in @p package_list .
These can be strings -- plain package names -- or
structures (with a pre- and post-install step).
This operation is called for "non-critical" packages,
which can succeed or fail without affecting the overall installation.
Packages are installed one-by-one to support package managers
that do not have a "install as much as you can" mode.
NOTE: package managers may reimplement this method
NOTE: no package-installation exceptions should be raised
"""
# we make a separate package manager call for each package so a
# single failing package won't stop all of them
for package in package_list:
try:
self.install_package(package)
except subprocess.CalledProcessError:
libcalamares.utils.warning("Could not install package %s" % package)
def operation_remove(self, package_list):
"""
Removes the list of packages named in @p package_list .
These can be strings -- plain package names -- or
structures (with a pre- and post-install step).
This operation is called for "critical" packages, which are
expected to succeed or fail all together.
However, if there are packages with pre- or post-scripts,
then packages are removed one-by-one instead.
NOTE: package managers may reimplement this method
NOTE: exceptions should be raised to indicate failure
"""
if all([isinstance(x, str) for x in package_list]):
self.remove(package_list)
else:
for package in package_list:
self.remove_package(package)
def operation_try_remove(self, package_list):
"""
Same relation as try_install has to install, except it removes
packages instead. Packages are removed one-by-one.
NOTE: package managers may reimplement this method
NOTE: no package-installation exceptions should be raised
"""
for package in package_list:
try:
self.remove_package(package)
except subprocess.CalledProcessError:
libcalamares.utils.warning("Could not remove package %s" % package)
### PACKAGE MANAGER IMPLEMENTATIONS
#
# Keep these alphabetical (presumably both by class name and backend name),
# even the Dummy implementation.
#
class PMPacman(PackageManager):
backend = "pacman"
def __init__(self):
import re
progress_match = re.compile("^\\((\\d+)/(\\d+)\\)")
def line_cb(line):
if line.startswith(":: "):
self.in_package_changes = "package" in line or "hooks" in line
else:
if self.in_package_changes and line.endswith("...\n"):
# Update the message, untranslated; do not change the
# progress percentage, since there may be more "installing..."
# lines in the output for the group, than packages listed
# explicitly. We don't know how to calculate proper progress.
global custom_status_message
custom_status_message = "pacman: " + line.strip()
libcalamares.job.setprogress(self.progress_fraction)
libcalamares.utils.debug(line)
self.in_package_changes = False
self.line_cb = line_cb
pacman = libcalamares.job.configuration.get("pacman", None)
if pacman is None:
pacman = dict()
if type(pacman) is not dict:
libcalamares.utils.warning("Job configuration *pacman* will be ignored.")
pacman = dict()
self.pacman_num_retries = pacman.get("num_retries", 0)
self.pacman_disable_timeout = pacman.get("disable_download_timeout", False)
self.pacman_needed_only = pacman.get("needed_only", False)
self.pacman_key = pacman.get("handle_keyrings", False)
self.pacman_requirements = pacman.get("requirements", [])
self.pacman_keyrings = pacman.get("keyrings", [])
def reset_progress(self):
self.in_package_changes = False
# These are globals
self.progress_fraction = (completed_packages * 1.0 / total_packages)
def run_pacman(self, command, callback=False):
"""
Call pacman in a loop until it is successful or the number of retries is exceeded
:param command: The pacman command to run
:param callback: An optional boolean that indicates if this pacman run should use the callback
:return:
"""
pacman_count = 0
while pacman_count <= self.pacman_num_retries:
pacman_count += 1
try:
if False: # callback:
host_env_process_output(command, self.line_cb)
else:
host_env_process_output(command)
return
except subprocess.CalledProcessError:
if pacman_count <= self.pacman_num_retries:
pass
else:
raise
def install(self, pkgs, from_local=False):
install_root = libcalamares.globalstorage.value("rootMountPoint")
cal_umask = os.umask(0)
for target in self.pacman_requirements:
dest = install_root + target["name"]
if not os.path.exists(dest):
libcalamares.utils.debug("Create: {!s}".format(dest))
mod = int(target["mode"],8)
libcalamares.utils.debug("Mode: {!s}".format(oct(mod)))
os.makedirs(dest, mode=mod)
path = join(install_root, "run")
os.chmod(path, 0o755)
os.umask(cal_umask)
f = "etc/resolv.conf"
if os.path.exists(join("/",f)):
shutil.copy2(join("/",f), join(install_root, f))
command = ["pacman"]
cachedir = join(install_root, "var/cache/pacman/pkg")
dbdir = join(install_root, "var/lib/pacman")
pacman_args = ["--root", install_root, "--dbpath", dbdir, "--cachedir", cachedir]
command.extend(pacman_args)
# Don't ask for user intervention, take the default action
command.append("--noconfirm")
# Don't report download progress for each file
command.append("--noprogressbar")
if self.pacman_needed_only is True:
command.append("--needed")
if self.pacman_disable_timeout is True:
command.append("--disable-download-timeout")
if from_local:
command.append("-U")
else:
command.append("-Sy")
command += pkgs
libcalamares.utils.debug("Command: {!s}".format(command))
self.reset_progress()
self.run_pacman(command, True)
if self.pacman_key:
self.init_keyring()
self.populate_keyring()
def remove(self, pkgs):
self.reset_progress()
self.run_pacman(["pacman", "-Rs", "--noconfirm"] + pkgs, True)
def init_keyring(self):
target_env_process_output(["pacman-key", "--init"])
def populate_keyring(self):
target_env_process_output(["pacman-key", "--populate"] + self.pacman_keyrings)
# Collect all the subclasses of PackageManager defined above,
# and index them based on the backend property of each class.
backend_managers = [
(c.backend, c)
for c in globals().values()
if type(c) is abc.ABCMeta and issubclass(c, PackageManager) and c.backend]
def subst_locale(plist):
"""
Returns a locale-aware list of packages, based on @p plist.
Package names that contain LOCALE are localized with the
BCP47 name of the chosen system locale; if the system
locale is 'en' (e.g. English, US) then these localized
packages are dropped from the list.
@param plist: list[str|dict]
Candidate packages to install.
@return: list[str|dict]
"""
locale = libcalamares.globalstorage.value("locale")
if not locale:
# It is possible to skip the locale-setting entirely.
# Then pretend it is "en", so that {LOCALE}-decorated
# package names are removed from the list.
locale = "en"
ret = []
for packagedata in plist:
if isinstance(packagedata, str):
packagename = packagedata
else:
packagename = packagedata["package"]
# Update packagename: substitute LOCALE, and drop packages
# if locale is en and LOCALE is in the package name.
if locale != "en":
packagename = Template(packagename).safe_substitute(LOCALE=locale)
elif 'LOCALE' in packagename:
packagename = None
if packagename is not None:
# Put it back in packagedata
if isinstance(packagedata, str):
packagedata = packagename
else:
packagedata["package"] = packagename
ret.append(packagedata)
return ret
def run_operations(pkgman, entry):
"""
Call package manager with suitable parameters for the given
package actions.
:param pkgman: PackageManager
This is the manager that does the actual work.
:param entry: dict
Keys are the actions -- e.g. "install" -- to take, and the values
are the (list of) packages to apply the action to. The actions are
not iterated in a specific order, so it is recommended to use only
one action per dictionary. The list of packages may be package
names (strings) or package information dictionaries with pre-
and post-scripts.
"""
global group_packages, completed_packages, mode_packages
for key in entry.keys():
package_list = subst_locale(entry[key])
group_packages = len(package_list)
if key == "install":
_change_mode(INSTALL)
pkgman.operation_install(package_list)
elif key == "try_install":
_change_mode(INSTALL)
pkgman.operation_try_install(package_list)
elif key == "remove":
_change_mode(REMOVE)
pkgman.operation_remove(package_list)
elif key == "try_remove":
_change_mode(REMOVE)
pkgman.operation_try_remove(package_list)
elif key == "localInstall":
_change_mode(INSTALL)
pkgman.operation_install(package_list, from_local=True)
elif key == "source":
libcalamares.utils.debug("Package-list from {!s}".format(entry[key]))
else:
libcalamares.utils.warning("Unknown package-operation key {!s}".format(key))
completed_packages += len(package_list)
libcalamares.job.setprogress(completed_packages * 1.0 / total_packages)
libcalamares.utils.debug("Pretty name: {!s}, setting progress..".format(pretty_name()))
group_packages = 0
_change_mode(None)
def run():
"""
Calls routine with detected package manager to install locale packages
or remove drivers not needed on the installed system.
:return:
"""
global mode_packages, total_packages, completed_packages, group_packages
backend = libcalamares.job.configuration.get("backend")
for identifier, impl in backend_managers:
if identifier == backend:
pkgman = impl()
break
else:
return "Bad backend", "backend=\"{}\"".format(backend)
if not libcalamares.globalstorage.value("hasInternet"):
libcalamares.utils.warning( "Package installation has been skipped: no internet" )
return None
operations = libcalamares.job.configuration.get("operations", [])
# if libcalamares.globalstorage.contains("packageOperations"):
# operations += libcalamares.globalstorage.value("packageOperations")
if libcalamares.globalstorage.contains("packagechooser_baseinit"):
base_init = libcalamares.globalstorage.value("packagechooser_baseinit")
libcalamares.utils.debug("Package added: {!s}".format(base_init))
operations[0]["install"].append(base_init)
if libcalamares.job.configuration.get("base_init"):
base_init = libcalamares.job.configuration.get("base_init", None)
if libcalamares.globalstorage.contains("netinstallAdd"):
data = libcalamares.globalstorage.value("netinstallAdd")
init_provider = data[0]["name"]
libcalamares.utils.debug("Init provider: {!s}".format(init_provider))
init_pkg = base_init + '-' + init_provider
libcalamares.utils.debug("Package added: {!s}".format(init_pkg))
operations[0]["install"].append(init_pkg)
if init_provider is not None:
libcalamares.globalstorage.insert("initProvider", init_provider)
libcalamares.globalstorage.insert("packageOperationsBasestrap", operations)
mode_packages = None
total_packages = 0
completed_packages = 0
for op in operations:
for packagelist in op.values():
total_packages += len(subst_locale(packagelist))
if not total_packages:
# Avoids potential divide-by-zero in progress reporting
return None
for entry in operations:
group_packages = 0
libcalamares.utils.debug(pretty_name())
try:
run_operations(pkgman, entry)
except subprocess.CalledProcessError as e:
libcalamares.utils.warning(str(e))
libcalamares.utils.debug("stdout:" + str(e.stdout))
libcalamares.utils.debug("stderr:" + str(e.stderr))
return (_("Package Manager error"),
_("The package manager could not make changes to the installed system. The command <pre>{!s}</pre> returned error code {!s}.")
.format(e.cmd, e.returncode))
mode_packages = None
libcalamares.job.setprogress(1.0)
return None

View File

@@ -0,0 +1,7 @@
# SPDX-FileCopyrightText: no
# SPDX-License-Identifier: CC0-1.0
---
type: "job"
name: "basestrap"
interface: "python"
script: "main.py"

View File

@@ -1,144 +0,0 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
# === This file is part of Calamares - <https://github.com/calamares> ===
#
# Copyright 2018-2019, Adriaan de Groot <groot@kde.org>
# Copyright 2019, Artoo <artoo@artixlinux.org>
#
# Calamares 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 3 of the License, or
# (at your option) any later version.
#
# Calamares 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 Calamares. If not, see <http://www.gnu.org/licenses/>.
import libcalamares
from libcalamares.utils import target_env_call, warning
from os.path import exists, join
import gettext
_ = gettext.translation("calamares-python",
localedir=libcalamares.utils.gettext_path(),
languages=libcalamares.utils.gettext_languages(),
fallback=True).gettext
def pretty_name():
return _("Configure Runit services")
class RunitController:
"""
This is the runit service controller.
All of its state comes from global storage and the job
configuration at initialization time.
"""
def __init__(self):
self.root = libcalamares.globalstorage.value('rootMountPoint')
# Translate the entries in the config to the actions passed to sv-helper
self.services = dict()
self.services["enable"] = libcalamares.job.configuration.get('services', [])
self.services["disable"] = libcalamares.job.configuration.get('disable', [])
self.svDir = libcalamares.job.configuration['svDir']
self.runsvDir = libcalamares.job.configuration['runsvDir']
def make_failure_description(self, state, name, runlevel):
"""
Returns a generic "could not <foo>" failure message, specialized
for the action @p state and the specific service @p name in @p runlevel.
"""
if state == "enable":
description = _("Cannot enable service {name!s} to run-level {level!s}.")
elif state == "disable":
description = _("Cannot disable service {name!s} from run-level {level!s}.")
else:
description = _("Unknown service-action <code>{arg!s}</code> for service {name!s} in run-level {level!s}.")
return description.format(arg=state, name=name, level=runlevel)
def update(self, state):
"""
Call sv-helper for each service listed
in services for the given @p state.
"""
for svc in self.services.get(state, []):
if isinstance(svc, str):
name = svc
runlevel = "default"
mandatory = False
else:
name = svc["name"]
runlevel = svc.get("runlevel", "default")
mandatory = svc.get("mandatory", False)
service_path = self.root + self.svDir + "/" + name
runlevel_path = self.root + self.runsvDir + "/" + runlevel
src = self.svDir + "/" + name
dest = self.runsvDir + "/" + runlevel + "/"
if state == 'enable':
cmd = ["ln", "-sv", src, dest]
elif state == 'disable':
cmd = ["rm", "-rv", dest]
if exists(service_path):
if exists(runlevel_path):
ec = target_env_call(cmd)
if ec != 0:
warning("Cannot {} service {} to {}".format(state, name, runlevel))
warning("{} returned error code {!s}".format(cmd, ec))
if mandatory:
title = _("Cannot modify service")
diagnostic = _("<code>cmd {arg!s}</code> call in chroot returned error code {num!s}.").format(arg=state, num=ec)
return (title,
self.make_failure_description(state, name, runlevel) + " " + diagnostic
)
else:
warning("Target runlevel {} does not exist for {}.".format(runlevel, name))
if mandatory:
title = _("Target runlevel does not exist")
diagnostic = _("The path for runlevel {level!s} is <code>{path!s}</code>, which does not exist.").format(level=runlevel, path=runlevel_path)
return (title,
self.make_failure_description(state, name, runlevel) + " " + diagnostic
)
else:
warning("Target service {} does not exist in {}.".format(name, self.svDir))
if mandatory:
title = _("Target service does not exist")
diagnostic = _("The path for service {name!s} is <code>{path!s}</code>, which does not exist.").format(name=name, path=service_path)
return (title,
self.make_failure_description(state, name, runlevel) + " " + diagnostic
)
def run(self):
"""Run the controller
"""
for state in ("enable", "disable"):
r = self.update(state)
if r is not None:
return r
def run():
"""
Setup services
"""
return RunitController().run()

View File

@@ -1,5 +0,0 @@
---
type: "job"
name: "services-runit"
interface: "python"
script: "main.py"

View File

@@ -1,44 +0,0 @@
# runit services module to modify service runlevels via symlinks in the chroot
#
# Services can be added (to any runlevel, or multiple runlevels) or deleted.
# Handle disable with care and only use it if absolutely necessary.
#
# if a service is listed in the conf but is not present/detected on the target system,
# or a runlevel does not exist, it will be ignored and skipped; a warning is logged.
#
---
# svDir: holds the runit service directory location
svDir: /etc/runit/sv
# runsvDir: holds the runlevels directory location
runsvDir: /etc/runit/runsvdir
# services: a list of entries to **enable**
# disable: a list of entries to **disable**
#
# Each entry has three fields:
# - name: the service name
# - (optional) runlevel: can hold any runlevel present on the target
# system; if no runlevel is provided, "default" is assumed.
# - (optional) mandatory: if set to true, a failure to modify
# the service will result in installation failure, rather than just
# a warning. The default is false.
#
# an entry may also be a single string, which is interpreted
# as the name field (runlevel "default" is assumed then, and not-mandatory).
#
# # Example services and disable settings:
# # - add foo1 to default, but it must succeed
# # - add foo2 to nonetwork
# # - remove foo3 from default
# # - remove foo4 from default
# services:
# - name: foo1
# mandatory: true
# - name: foo2
# runlevel: nonetwork
# disable:
# - name: foo3
# - foo4
services: []
disable: []