"""The commands a plan step will actually run, in order.
`paru -S --needed a b c` is not a command queue, it is a summary. What happens
underneath is a sequence: sources are fetched, checksums verified, the PKGBUILD's
own functions run, the result is packed and handed to pacman -- which downloads,
verifies, takes a snapshot, installs, and then runs the alpm hooks that rebuild
an initramfs or reload systemd.
None of that is guesswork. The PKGBUILD is parsed statically and its function
bodies are its commands. The hooks are files on disk with declared triggers. The
snapshots are a hook. This module assembles them in the order they run.
Nothing here executes anything.
"""
from __future__ import annotations
import re
from dataclasses import dataclass, field
from pathlib import Path
from . import hooks as hooks_mod
from .audit import PATTERNS, strip_source_prefix
from .srcinfo import parse_pkgbuild
#: how a stage is marked in the list
LEVEL_STEP = "step"
LEVEL_NOTE = "note"
LEVEL_HOOK = "hook"
LEVEL_WARN = "warn"
LEVEL_DANGER = "danger"
@dataclass
class Stage:
command: str # the literal thing that runs
what: str # one line saying what it is for
target: str = "" # the package or packages involved
level: str = LEVEL_STEP
phase: str = "" # the group this belongs to
finding: str = "" # the audit rule that flagged it, if any
@dataclass
class Queue:
"""A step's commands, grouped into phases."""
stages: list[Stage] = field(default_factory=list)
def add(
self,
phase: str,
command: str,
what: str,
*,
target: str = "",
level: str = LEVEL_STEP,
finding: str = "",
) -> None:
self.stages.append(
Stage(
command=command, what=what, target=target, level=level, phase=phase, finding=finding
)
)
@property
def warnings(self) -> list[Stage]:
return [s for s in self.stages if s.level in (LEVEL_WARN, LEVEL_DANGER)]
def phases(self) -> list[str]:
seen: list[str] = []
for stage in self.stages:
if stage.phase not in seen:
seen.append(stage.phase)
return seen
# --------------------------------------------------------------------------
# pacman
# --------------------------------------------------------------------------
def _hook_stage(hook: hooks_mod.Hook) -> Stage:
level = LEVEL_NOTE if hook.is_snapshot else LEVEL_HOOK
return Stage(
command=hook.exec_ or hook.path.name,
what=hook.label,
target=", ".join(sorted({t for tr in hook.triggers for t in tr.targets}))[:60],
level=level,
phase="snapshot" if hook.is_snapshot else "hooks",
)
def pacman_queue(
updates,
*,
repo_of=None,
refreshing: bool = True,
all_hooks: list[hooks_mod.Hook] | None = None,
paths_by_package: dict[str, list[str]] | None = None,
snapshotting: bool = True,
) -> Queue:
"""The transaction pacman will run for a set of repository upgrades."""
q = Queue()
if not updates:
return q
names = {u.name for u in updates}
hooks = all_hooks if all_hooks is not None else hooks_mod.load()
paths = paths_by_package or {}
before, after = hooks_mod.scheduled(hooks, names, paths)
snapshots = [h for h in before if h.is_snapshot]
other_before = [h for h in before if not h.is_snapshot]
if refreshing:
q.add("sync", "pacman -Sy", "refresh the package databases from the mirrors")
if snapshotting and snapshots:
for hook in snapshots:
q.add(
"snapshot",
hook.exec_ or hook.path.name,
"snapper takes a pre-transaction snapshot, so the upgrade can "
"be rolled back as a unit",
level=LEVEL_NOTE,
)
q.add(
"resolve",
f"resolve {len(updates)} package(s) against the new databases",
"pacman works out the full transaction, including anything pulled in "
"to satisfy dependencies",
)
for update in updates:
origin = f" from {repo_of(update.name)}" if repo_of else ""
q.add(
"download",
f"download {update.name}-{update.new}{origin}",
f"{update.name} {update.old} -> {update.new}",
target=update.name,
)
q.add(
"verify",
"verify the signatures of every downloaded package",
"each package is checked against the keyring before anything is "
"unpacked; pacman refuses the whole transaction if one fails",
)
q.add(
"verify",
"check package integrity",
"the file list and the compressed payload are validated",
)
q.add(
"verify",
"check for file conflicts",
"pacman confirms no two packages in the transaction own the same path",
)
for hook in other_before:
q.stages.append(_hook_stage(hook))
for update in updates:
q.add(
"install",
f"install {update.name} {update.old} -> {update.new}",
"unpacked over the running system; the old files are replaced",
target=update.name,
)
for hook in after:
q.stages.append(_hook_stage(hook))
return q
# --------------------------------------------------------------------------
# AUR
# --------------------------------------------------------------------------
#: shell noise that is not a command worth listing
_NOISE = re.compile(r"^\s*(?:#|$)")
#: `cd`, `local`, `export` and friends are bookkeeping, not actions
_BOOKKEEPING = re.compile(
r"^\s*(?:cd|local|export|declare|typeset|unset|readonly|shift|set|shopt)\b"
)
#: Bare shell structure is not something that runs, so it is dropped outright.
_STRUCTURE = re.compile(r"^\s*(?:fi|esac|done|then|else|elif|do|in|;;|;;&|;&|\{+|\}+|\(|\))\s*$")
#: A line that opens or branches control flow. Kept, because the condition is
#: what tells you whether the commands underneath will run at all, but marked as
#: structure so the eye can skip it.
_CONTROL = re.compile(r"^\s*(?:if|elif|while|until|for|case)\b|;;\s*$|^\s*[\w.|*$\"{}\[\]-]+\)\s")
def _commands_in(body: str) -> list[str]:
"""The lines of a PKGBUILD function that actually do something."""
out: list[str] = []
for raw in body.splitlines():
line = raw.strip()
if _NOISE.match(line) or _BOOKKEEPING.match(line):
continue
if _STRUCTURE.match(line):
continue
# strip a trailing comment that is not inside this line's arguments
hash_at = line.find(" #")
if hash_at > 0:
line = line[:hash_at].rstrip()
if line:
out.append(line)
return out
def _flag(line: str) -> tuple[str, str]:
"""(level, rule) if an audit rule flags this command.
Falls through to the next rule when there is no match. Getting that wrong
marked every line, which is worse than marking none: a warning that is
always present is a warning nobody reads.
"""
worst = ""
for pat in PATTERNS:
try:
match = pat.regex.search(line)
except Exception: # noqa: BLE001, S112 - a broken rule must not stop this
continue
if match is None:
continue
if pat.guard is not None and not pat.guard(line, match):
continue
if pat.severity.name == "BLOCK":
return LEVEL_DANGER, pat.title
worst = worst or pat.title
return (LEVEL_WARN, worst) if worst else ("", "")
def _source_name(entry: str) -> str:
"""The file a source entry produces, without the `filename::` prefix."""
cleaned = strip_source_prefix(entry).strip()
if "::" in entry:
return entry.split("::", 1)[0]
return cleaned.rsplit("/", 1)[-1] or cleaned
def aur_queue(
clone_dir: Path,
targets: list[str],
*,
meta: dict | None = None,
audits: dict | None = None,
all_hooks: list[hooks_mod.Hook] | None = None,
paths_by_package: dict[str, list[str]] | None = None,
) -> Queue:
"""The build paru will run, package by package, from the PKGBUILDs."""
q = Queue()
if not targets:
return q
hooks = all_hooks if all_hooks is not None else hooks_mod.load()
paths = paths_by_package or {}
for name in sorted(targets):
repo = clone_dir / name
pkgbuild = repo / "PKGBUILD"
entry = (meta or {}).get(name)
version = ""
if entry is not None:
version = f" {entry.version}"
q.add(
f"fetch:{name}",
f"git clone https://aur.archlinux.org/{name}.git",
f"the AUR repository, which is the PKGBUILD being reviewed{version}",
target=name,
)
if not pkgbuild.is_file():
q.add(
f"fetch:{name}",
"(no local clone)",
"paru has not downloaded this package, so its PKGBUILD is "
"not on disk and its commands cannot be listed",
target=name,
level=LEVEL_NOTE,
)
continue
text = pkgbuild.read_text(errors="replace")
pb = parse_pkgbuild(text)
for source in pb.get("source"):
q.add(
f"source:{name}",
f"fetch {_source_name(source)}",
f"declared in source=(): {strip_source_prefix(source)[:70]}",
target=name,
)
sums = pb.get("sha256sums") or pb.get("sha512sums") or pb.get("b2sums") or pb.get("md5sums")
if sums:
skips = [i for i, s in enumerate(sums) if s.strip().upper() == "SKIP"]
note = "PKGBUILD checksums are verified against the sources"
if skips:
note += f"; {len(skips)} entry(ies) are SKIP and are not verified"
q.add(
f"source:{name}",
"verify checksums",
note,
target=name,
level=LEVEL_WARN if skips else LEVEL_STEP,
)
report = (audits or {}).get(name)
# Keyed by the function a finding sits in. A file-wide rule reports
# `where="PKGBUILD"`, which covers every function in the file: the
# audit found something dangerous in this PKGBUILD, and the queue must
# not quietly soften it just because it does not name a function.
flagged: dict[str, bool] = {}
for finding in report.findings if report else []:
where = finding.where.split("[")[0].strip()
if where == "PKGBUILD" or not where:
flagged["*"] = True
else:
flagged[where.split("(")[0]] = True
for func in ("prepare", "build", "check", "package"):
body = pb.funcs.get(func, "")
if not body:
continue
q.add(f"{func}:{name}", f"{func}()", _phase_blurb(func, name), target=name)
for command in _commands_in(body):
level, rule = _flag(command)
if (
report is not None
and level == LEVEL_DANGER
and "*" not in flagged
and func not in flagged
):
# The audit has read this package's previous revision and
# declined to call it dangerous; the per-line scan has no
# such context. Deferring keeps the queue and the audit
# screen saying the same thing -- two views that disagree
# about whether something is safe is worse than either.
level = LEVEL_WARN
if level == "" and _CONTROL.match(command):
level = LEVEL_NOTE
q.add(
f"{func}:{name}",
command,
rule or "",
target=name,
level=level or LEVEL_STEP,
finding=rule,
)
q.add(
f"pack:{name}",
f"makepkg packs {name}",
"the built tree is compressed into a .pkg.tar.zst",
target=name,
)
q.add(
"install",
"install the built packages",
"each one is handed to pacman -U, and its hooks run as root",
)
names = set(targets)
before, after = hooks_mod.scheduled(hooks, names, paths)
for hook in before:
q.stages.append(_hook_stage(hook))
for hook in after:
q.stages.append(_hook_stage(hook))
return q
def _phase_blurb(func: str, name: str) -> str:
return {
"prepare": f"patches and readies the sources for {name}",
"build": f"compiles {name}; runs as you, in the cloned directory",
"check": f"runs the package's own test suite for {name}",
"package": f"installs the built files into the package tree for {name}",
}.get(func, func)
def flatpak_queue(updates) -> Queue:
"""Flatpak updates are opaque, and saying so is better than inventing them."""
q = Queue()
if not updates:
return q
q.add(
"fetch",
"flatpak update --noninteractive -y",
"flatpak resolves and applies every pending update in one pass",
)
for update in updates:
q.add(
"fetch",
f"update {update.application} {update.version}",
f"from {update.origin}",
target=update.application,
)
q.add(
"install",
"flatpak republishes each app from its remote",
"flatpak does not expose the per-application steps the way pacman "
"and makepkg do, so only the applications are listed here",
)
return q