"""Repo-side updates, read straight out of libalpm.
No parsing. pyalpm hands back versions, sizes, descriptions, install reasons and
dependency lists directly, which is what makes it possible to render a plan that
does not lie about what is about to happen.
The one thing pyalpm does not do for us is register the sync databases, so the
repo list is read from pacman.conf first.
"""
from __future__ import annotations
import contextlib
import fnmatch
import re
import shutil
import subprocess
import tempfile
import time
from dataclasses import dataclass, field
from pathlib import Path
import pyalpm
from .srcinfo import dep_name
DEFAULT_CONF = "/etc/pacman.conf"
REASON_EXPLICIT = 0
REASON_DEPEND = 1
@dataclass(frozen=True)
class Update:
"""One package moving from an installed version to a sync version."""
name: str
old: str
new: str
repo: str
download: int = 0
installed_old: int = 0
installed_new: int = 0
desc: str = ""
@property
def delta(self) -> int:
"""Change in on-disk size. Negative when the update shrinks the package."""
return self.installed_new - self.installed_old
@property
def is_downgrade(self) -> bool:
return pyalpm.vercmp(self.new, self.old) < 0
@dataclass
class Plan:
"""What `pacman -Su` would actually do."""
updates: list[Update] = field(default_factory=list)
#: newer in the db but held back by IgnorePkg
ignored: list[str] = field(default_factory=list)
# NOTE: there is deliberately no `removals` member. A package with no sync db
# entry is a *foreign* package (AUR), which `pacman -Su` leaves alone; and a
# package genuinely dropped from a repo cannot be detected at all, because
# pacman records only what is in a database now. Reporting either as a
# removal would be a lie. `foreign()` is the honest accessor.
def __len__(self) -> int:
return len(self.updates)
@property
def download_total(self) -> int:
return sum(u.download for u in self.updates)
@property
def installed_delta(self) -> int:
return sum(u.delta for u in self.updates)
@property
def repos(self) -> list[str]:
return sorted({u.repo for u in self.updates})
@dataclass(frozen=True)
class Config:
repos: tuple[str, ...]
ignore_pkg: frozenset[str]
ignore_group: frozenset[str]
def read_conf(path: str | Path = DEFAULT_CONF) -> Config:
"""Parse the bits of pacman.conf we need: repo sections and ignore lists.
`Include` is deliberately not followed; it points at mirrorlists, which have
no bearing on which databases to register.
"""
repos: list[str] = []
ignore_pkg: set[str] = set()
ignore_group: set[str] = set()
try:
text = Path(path).read_text(errors="replace")
except OSError:
return Config(("core", "extra"), frozenset(), frozenset())
section = None
for raw in text.splitlines():
line = raw.split("#", 1)[0].strip()
if not line:
continue
header = re.match(r"^\[(.+)]$", line)
if header:
name = header.group(1)
section = None if name == "options" else name
if section and section not in repos:
repos.append(section)
continue
if section is None:
key, _, value = line.partition("=")
key = key.strip()
for part in value.split():
if key == "IgnorePkg":
ignore_pkg.add(part)
elif key == "IgnoreGroup":
ignore_group.add(part)
return Config(tuple(repos), frozenset(ignore_pkg), frozenset(ignore_group))
@contextlib.contextmanager
def connect(root: str = "/", dbpath: str = "/var/lib/pacman", conf: str | Path = DEFAULT_CONF):
"""An alpm handle with every configured repo registered.
Without the explicit register_syncdb calls the handle silently reports zero
sync databases, and therefore zero updates, which looks exactly like a
system that is fully up to date.
"""
config = read_conf(conf)
handle = pyalpm.Handle(root, dbpath)
for name in config.repos:
try:
handle.register_syncdb(name, pyalpm.SIG_DATABASE_OPTIONAL)
except Exception: # noqa: BLE001, S112 - a missing optional repo is not fatal
continue
for path_ in ("/var/cache/pacman/pkg",):
with contextlib.suppress(Exception):
handle.add_cachedir(path_)
_register_ignores(handle, config)
try:
yield handle
finally:
del handle
def _register_ignores(handle, config: Config) -> None:
"""One owner for the ignore lists.
`connect()` registered them and `preview()` did not, and the plan is built
from the preview handle -- so `_is_ignored` always found an empty list and
every IgnorePkg package was offered as a pending upgrade while the apply,
which does use the preview handle, would have held it back.
"""
for name in config.ignore_pkg:
with contextlib.suppress(Exception):
handle.add_ignorepkg(name)
for name in config.ignore_group:
with contextlib.suppress(Exception):
handle.add_ignoregrp(name)
def _best_versions(handle) -> dict[str, object]:
"""Newest version of each package across all sync dbs."""
best: dict[str, object] = {}
for db in handle.get_syncdbs():
for pkg in db.pkgcache:
cur = best.get(pkg.name)
if cur is None or pyalpm.vercmp(pkg.version, cur.version) > 0:
best[pkg.name] = pkg
return best
def plan(handle) -> Plan:
"""What a system upgrade would do, computed against the *current* sync dbs.
This must be called after the dbs have been refreshed, otherwise it reports
the truth about stale data, which is worse than reporting nothing.
"""
best = _best_versions(handle)
local = handle.get_localdb()
result = Plan()
for lp in local.pkgcache:
sp = best.get(lp.name)
if sp is None:
continue # foreign (AUR); `pacman -Su` does not touch it
if pyalpm.vercmp(sp.version, lp.version) <= 0:
continue
if _is_ignored(handle, sp):
result.ignored.append(lp.name)
continue
result.updates.append(
Update(
name=lp.name,
old=lp.version,
new=sp.version,
repo=sp.db.name,
download=getattr(sp, "download_size", 0),
# `isize` is the installed size. `size` is NOT interchangeable:
# pyalpm reports 0 for it on a local package and the *download*
# size on a sync one, so using it made every delta look like a
# full-size increase.
installed_old=getattr(lp, "isize", 0),
installed_new=getattr(sp, "isize", 0),
desc=getattr(sp, "desc", "") or "",
)
)
result.updates.sort(key=lambda u: (u.repo, u.name))
result.ignored.sort()
return result
def _is_ignored(handle, pkg) -> bool:
"""`IgnorePkg` and `IgnoreGroup` are shell glob patterns, not literals.
pacman.conf documents both as accepting globs and hands the patterns to
libalpm, which fnmatches them. Comparing for equality meant `IgnorePkg =
linux*` matched nothing here and the plan listed packages pacman would
refuse to touch.
"""
if any(fnmatch.fnmatchcase(pkg.name, pat) for pat in (handle.ignorepkgs or [])):
return True
groups = getattr(pkg, "groups", []) or []
return any(
fnmatch.fnmatchcase(group, pat) for group in groups for pat in (handle.ignoregrps or [])
)
def foreign(handle) -> list[str]:
"""Installed packages with no sync db entry: the AUR set, in practice."""
known = set(_best_versions(handle))
return sorted(p.name for p in handle.get_localdb().pkgcache if p.name not in known)
def explicit(handle) -> list[str]:
return sorted(p.name for p in handle.get_localdb().pkgcache if p.reason == REASON_EXPLICIT)
def installed(handle) -> dict[str, str]:
"""name -> version for everything installed. Useful for cross-checks."""
return {p.name: p.version for p in handle.get_localdb().pkgcache}
def _satisfies(pkg, dep: str) -> bool:
"""Whether `pkg` satisfies a dependency string, per libalpm itself.
Delegated rather than reimplemented. Version constraints on `provides` are
load-bearing: ffmpeg4.4 provides `libavcodec.so=58-64` while callers require
`=63-64`, so a name-only comparison wrongly reports it as required.
"""
try:
return pyalpm.find_satisfier([pkg], dep) is not None
except Exception: # noqa: BLE001 - an unparsable dep string satisfies nothing
return False
def orphans(handle, *, include_optional: bool = False) -> list[str]:
"""Dependency-installed packages that nothing installed requires.
Matches pacman's two levels of strictness:
include_optional=False == `pacman -Qtdq` (an *optional* dependency of an
installed package counts as a reason to keep it)
include_optional=True == `pacman -Qttdq` (only hard dependencies count)
The distinction is not cosmetic: on a real system the two answers differ by
every package that anything merely *suggests*.
"""
packages = list(handle.get_localdb().pkgcache)
# Index every package under the names it could satisfy, so a dependency only
# has to be tested against the handful of packages that could match it.
by_name: dict[str, list] = {}
for pkg in packages:
by_name.setdefault(pkg.name, []).append(pkg)
for spec in getattr(pkg, "provides", []) or []:
by_name.setdefault(dep_name(spec), []).append(pkg)
specs: set[str] = set()
for pkg in packages:
specs.update(getattr(pkg, "depends", []) or [])
if not include_optional:
specs.update(getattr(pkg, "optdepends", []) or [])
required: set[str] = set()
for spec in specs:
for candidate in by_name.get(dep_name(spec), ()):
if _satisfies(candidate, spec):
required.add(candidate.name)
return sorted(
pkg.name
for pkg in packages
if pkg.reason == pyalpm.PKG_REASON_DEPEND and pkg.name not in required
)
# --------------------------------------------------------------------------
# refreshing the databases
# --------------------------------------------------------------------------
SYNC_TIMEOUT = 300
def sync_db_age(dbpath: str = "/var/lib/pacman") -> float | None:
"""Seconds since the newest sync database was written, or None."""
try:
stamps = [p.stat().st_mtime for p in (Path(dbpath) / "sync").glob("*.db")]
except OSError:
return None
return time.time() - max(stamps) if stamps else None
def _sync_argv(dbdir: str) -> list[str]:
"""`pacman -Sy` into an alternate dbpath, exactly as checkupdates does it.
`fakeroot` matters: pacman's sandbox refuses files it does not own, and the
databases it writes would otherwise be owned by the invoking user.
"""
args = ["pacman", "-Sy", "--noconfirm", "--dbpath", dbdir, "--logfile", "/dev/null"]
if shutil.which("fakeroot"):
return ["fakeroot", "--", args[0], "--disable-sandbox-filesystem", *args[1:]]
return args
@contextlib.contextmanager
def preview(
conf: str | Path = DEFAULT_CONF,
dbpath: str = "/var/lib/pacman",
*,
sync: bool = True,
timeout: int = SYNC_TIMEOUT,
):
"""A handle whose sync databases were refreshed into a throwaway directory.
This is what lets `updater plan` be accurate without asking for a password:
the real databases are never touched, and the local database is symlinked in
so installed versions are still visible.
Yields (handle, refreshed, note).
"""
config = read_conf(conf)
refreshed = False
note = ""
with tempfile.TemporaryDirectory(prefix="updater-sync-") as dbdir:
local = Path(dbpath) / "local"
if local.is_dir():
with contextlib.suppress(OSError):
(Path(dbdir) / "local").symlink_to(local, target_is_directory=True)
if sync:
try:
proc = subprocess.run(
_sync_argv(dbdir),
capture_output=True,
text=True,
errors="replace",
timeout=timeout,
check=False,
)
except (OSError, subprocess.SubprocessError) as exc:
note = f"could not refresh databases: {exc}"
else:
refreshed = proc.returncode == 0
if not refreshed:
detail = (proc.stderr or proc.stdout).strip().splitlines()
note = detail[0] if detail else f"pacman -Sy exited {proc.returncode}"
else:
note = "database refresh skipped"
if not refreshed:
# fall back to the real databases rather than reporting nothing
yield from _fallback(config, dbpath, refreshed, note)
return
handle = pyalpm.Handle("/", dbdir)
for name in config.repos:
with contextlib.suppress(Exception):
handle.register_syncdb(name, pyalpm.SIG_DATABASE_OPTIONAL)
_register_ignores(handle, config)
try:
yield handle, refreshed, note
finally:
del handle
def _fallback(config: Config, dbpath: str, refreshed: bool, note: str):
handle = pyalpm.Handle("/", dbpath)
for name in config.repos:
with contextlib.suppress(Exception):
handle.register_syncdb(name, pyalpm.SIG_DATABASE_OPTIONAL)
_register_ignores(handle, config)
try:
yield handle, refreshed, note
finally:
del handle