Manage deb822 format and apt-key removal in trixie

Author: maisim

pyinfra/facts/apt.py

@@ -60,6 +60,75 @@ def parse_apt_repo(name):
     }
 
 
+def parse_deb822_stanza(lines: list[str]):
+    """Parse a deb822 style repository stanza.
+
+    deb822 sources are key/value pairs separated by blank lines, eg::
+
+        Types: deb
+        URIs: http://deb.debian.org/debian
+        Suites: bookworm
+        Components: main contrib
+        Architectures: amd64
+        Signed-By: /usr/share/keyrings/debian-archive-keyring.gpg
+
+    Returns a list of dicts matching the legacy ``parse_apt_repo`` output so the
+    rest of pyinfra can remain backwards compatible. A stanza may define
+    multiple types/URIs/suites which we expand into individual repo dicts.
+    """
+
+    if not lines:
+        return []
+
+    data: dict[str, str] = {}
+    for line in lines:
+        if not line or line.startswith("#"):
+            continue
+        # Field-Name: value
+        try:
+            key, value = line.split(":", 1)
+        except ValueError:  # malformed line
+            continue
+        data[key.strip()] = value.strip()
+
+    required = ("Types", "URIs", "Suites")
+    if not all(field in data for field in required):  # not a valid stanza
+        return []
+
+    types = data.get("Types", "").split()
+    uris = data.get("URIs", "").split()
+    suites = data.get("Suites", "").split()
+    components = data.get("Components", "").split()
+
+    # Map deb822 specific fields to legacy option names
+    options: dict[str, object] = {}
+    if architectures := data.get("Architectures"):
+        archs = architectures.split()
+        if archs:
+            options["arch"] = archs if len(archs) > 1 else archs[0]
+    if signed_by := data.get("Signed-By"):
+        signed = signed_by.split()
+        options["signed-by"] = signed if len(signed) > 1 else signed[0]
+    if trusted := data.get("Trusted"):
+        options["trusted"] = trusted.lower()
+
+    repos = []
+    # Produce combinations – in most real-world cases these will each be one.
+    for _type in types or ["deb"]:
+        for uri in uris:
+            for suite in suites:
+                repos.append(
+                    {
+                        "options": dict(options),  # copy per entry
+                        "type": _type,
+                        "url": uri,
+                        "distribution": suite,
+                        "components": components,
+                    }
+                )
+    return repos
+
+
 class AptSources(FactBase):
     """
     Returns a list of installed apt sources:
@@ -78,9 +147,11 @@ class AptSources(FactBase):
 
     @override
     def command(self) -> str:
+        # Include deb822 style .sources files in addition to legacy .list files.
         return make_cat_files_command(
             "/etc/apt/sources.list",
             "/etc/apt/sources.list.d/*.list",
+            "/etc/apt/sources.list.d/*.sources",
         )
 
     @override
@@ -93,11 +164,45 @@ def requires_command(self) -> str:
     def process(self, output):
         repos = []
 
-        for line in output:
+        deb822_buffer: list[str] = []
+        inside_deb822 = False
+
+        def flush_deb822():
+            nonlocal deb822_buffer, inside_deb822, repos
+            if deb822_buffer:
+                repos.extend(parse_deb822_stanza(deb822_buffer))
+                deb822_buffer = []
+            inside_deb822 = False
+
+        for raw_line in output:
+            line = raw_line.strip()
+
+            # Blank line -> possible end of deb822 stanza
+            if line == "":
+                flush_deb822()
+                continue
+
+            # Heuristic: deb822 stanzas use Key: Value capitalization, while
+            # legacy sources begin with 'deb' or 'deb-src'. Detect stanza start.
+            if re.match(r"^[A-Z][A-Za-z-]+:\s+", line):
+                # Starting or continuing a deb822 stanza
+                inside_deb822 = True
+                deb822_buffer.append(line)
+                continue
+
+            # If we were processing a stanza and hit a non deb822 line, flush it
+            if inside_deb822:
+                flush_deb822()
+
+            # Legacy single-line entry
             repo = parse_apt_repo(line)
             if repo:
                 repos.append(repo)
 
+        # Flush any remaining deb822 stanza at EOF
+        if inside_deb822:
+            flush_deb822()
+
         return repos
 
 
@@ -115,14 +220,30 @@ class AptKeys(GpgFactBase):
         }
     """
 
-    # This requires both apt-key *and* apt-key itself requires gpg
     @override
     def command(self) -> str:
-        return "! command -v gpg || apt-key list --with-colons"
-
-    @override
-    def requires_command(self) -> str:
-        return "apt-key"
+        # Prefer not to use deprecated apt-key even if present. Iterate over keyrings
+        # directly. This maintains backwards compatibility of output with the
+        # previous implementation which fell back to this method.
+        return (
+            "for f in "
+            "  /etc/apt/trusted.gpg "
+            "  /etc/apt/trusted.gpg.d/*.gpg /etc/apt/trusted.gpg.d/*.asc "
+            "  /etc/apt/keyrings/*.gpg /etc/apt/keyrings/*.asc "
+            "  /usr/share/keyrings/*.gpg /usr/share/keyrings/*.asc "
+            "; do "
+            "  [ -e \"$f\" ] || continue; "
+            "  case \"$f\" in "
+            "    *.asc) "
+            "      gpg --batch --show-keys --with-colons --keyid-format LONG \"$f\" "
+            "      ;; "
+            "    *) "
+            "      gpg --batch --no-default-keyring --keyring \"$f\" "
+            "          --list-keys --with-colons --keyid-format LONG "
+            "      ;; "
+            "  esac; "
+            "done"
+        )
 
 
 class AptSimulationDict(TypedDict):

pyinfra/operations/apt.py

@@ -4,6 +4,7 @@
 
 from __future__ import annotations
 
+import re
 from datetime import timedelta
 from urllib.parse import urlparse
 
@@ -45,57 +46,139 @@ def _simulate_then_perform(command: str):
         yield noninteractive_apt(command)
 
 
-@operation()
-def key(src: str | None = None, keyserver: str | None = None, keyid: str | list[str] | None = None):
+def _sanitize_apt_keyring_name(name: str) -> str:
     """
-    Add apt gpg keys with ``apt-key``.
-
-    + src: filename or URL
-    + keyserver: URL of keyserver to fetch key from
-    + keyid: key ID or list of key IDs when using keyserver
+    Produce a filesystem-friendly name from an URL host/basename or a local filename.
+    """
+    name = name.strip().lower()
+    name = re.sub(r"[^\w.-]+", "_", name)
+    name = re.sub(r"_+", "_", name).strip("_.")
+    return name or "apt-keyring"
 
-    keyserver/id:
-        These must be provided together.
 
-    .. warning::
-        ``apt-key`` is deprecated in Debian, it is recommended NOT to use this
-        operation and instead follow the instructions here:
+def _derive_dest_from_src_and_keyids(src: str | None, keyids: list[str] | None, dest: str | None) -> str:
+    """
+    Compute a stable destination path in /etc/apt/keyrings/.
+    Priority:
+      1) explicit dest if provided
+      2) from src (URL host + basename, or local basename)
+      3) from keyids (joined)
+      4) fallback "apt-keyring.gpg"
+    """
+    if dest:
+        # Ensure it ends with .gpg and is absolute under /etc/apt/keyrings
+        if not dest.endswith(".gpg"):
+            dest += ".gpg"
+        if not dest.startswith("/"):
+            dest = f"/etc/apt/keyrings/{dest}"
+        return dest
+
+    base = None
+    if src:
+        parsed = urlparse(src)
+        if parsed.scheme and parsed.netloc:
+            host = _sanitize_apt_keyring_name(parsed.netloc.replace(":", "_"))
+            bn = _sanitize_apt_keyring_name((parsed.path.rsplit("/", 1)[-1] or "key").replace(".asc", "").replace(".gpg", ""))
+            base = f"{host}-{bn}"
+        else:
+            bn = _sanitize_apt_keyring_name(src.rsplit("/", 1)[-1].replace(".asc", "").replace(".gpg", ""))
+            base = bn or "key"
+    elif keyids:
+        base = "keyserver-" + _sanitize_apt_keyring_name("-".join(keyids))
+    else:
+        base = "apt-keyring"
 
-        https://wiki.debian.org/DebianRepository/UseThirdParty
+    return f"/etc/apt/keyrings/{base}.gpg"
 
-    **Examples:**
 
-    .. code:: python
-
-        # Note: If using URL, wget is assumed to be installed.
+@operation()
+def key(
+    src: str | None = None,
+    keyserver: str | None = None,
+    keyid: str | list[str] | None = None,
+    dest: str | None = None,
+):
+    """
+    Add apt GPG keys *without* apt-key:
+      - Keys are stored under /etc/apt/keyrings/<name>.gpg (binary, dearmored if needed).
+      - You must reference the resulting file in your apt source via `signed-by=...`.
+
+    Args:
+        src: filename or URL to a key (ASCII .asc or binary .gpg)
+        keyserver: keyserver URL for fetching keys by ID
+        keyid: key ID or list of key IDs (required with keyserver)
+        dest: optional keyring filename/path ('.gpg' will be enforced, defaults under /etc/apt/keyrings)
+
+    Behavior:
+        - Idempotent via AptKeys: if the key IDs are already present in any apt keyring, nothing is changed.
+        - If src is ASCII (.asc), it will be dearmored; if binary (.gpg), it's copied as-is.
+        - Keyserver flow uses a temporary GNUPGHOME, then exports and dearmors to the destination keyring.
+
+    Examples:
         apt.key(
-            name="Add the Docker apt gpg key",
-            src="https://download.docker.com/linux/ubuntu/gpg",
+            name="Add Docker apt GPG key",
+            src="https://download.docker.com/linux/debian/gpg",
+            dest="docker.gpg",
         )
 
         apt.key(
             name="Install VirtualBox key",
             src="https://www.virtualbox.org/download/oracle_vbox_2016.asc",
+            dest="oracle-virtualbox.gpg",
+        )
+
+        apt.key(
+            name="Fetch keys from keyserver",
+            keyserver="hkps://keyserver.ubuntu.com",
+            keyid=["0xD88E42B4", "0x7EA0A9C3"],
+            dest="vendor-archive.gpg",
         )
     """
 
+    # Gather currently installed keys (across trusted.gpg.d/, keyrings/, etc.)
     existing_keys = host.get_fact(AptKeys)
 
+    # --- src branch: install a key from URL or local file ---
     if src:
-        key_data = host.get_fact(GpgKey, src=src)
-        if key_data:
-            keyid = list(key_data.keys())
+        key_data = host.get_fact(GpgKey, src=src)  # Parses the key(s) from src to extract key IDs
+        keyids_from_src = list(key_data.keys()) if key_data else []
+
+        # If we don't know the IDs (eg. unreachable URL), we cannot determine idempotency -> try to install.
+        # Otherwise, skip if all key IDs are already present.
+        if (not keyids_from_src) or (not all(kid in existing_keys for kid in keyids_from_src)):
+            dest_path = _derive_dest_from_src_and_keyids(src, keyids_from_src or None, dest)
 
-        if not keyid or not all(kid in existing_keys for kid in keyid):
-            # If URL, wget the key to stdout and pipe into apt-key, because the "adv"
-            # apt-key passes to gpg which doesn't always support https!
             if urlparse(src).scheme:
-                yield "(wget -O - {0} || curl -sSLf {0}) | apt-key add -".format(src)
+                # Remote source: download to a temp file, then install/dearmor accordingly
+                yield (
+                    "sh -c 'set -e;"
+                    " install -d -m 0755 /etc/apt/keyrings;"
+                    " tmp=$(mktemp);"
+                    f" (wget -qO \"$tmp\" {src} || curl -sSLf -o \"$tmp\" {src});"
+                    " if grep -q \"BEGIN PGP PUBLIC KEY BLOCK\" \"$tmp\"; then"
+                    f"   gpg --batch --dearmor -o \"{dest_path}\" \"$tmp\";"
+                    " else"
+                    f"   install -m 0644 \"$tmp\" \"{dest_path}\";"
+                    " fi;"
+                    " rm -f \"$tmp\";"
+                    f" chmod 0644 \"{dest_path}\"'"
+                )
             else:
-                yield "apt-key add {0}".format(src)
+                # Local file already present on the target
+                yield (
+                    "sh -c 'set -e;"
+                    " install -d -m 0755 /etc/apt/keyrings;"
+                    f" if grep -q \"BEGIN PGP PUBLIC KEY BLOCK\" \"{src}\"; then"
+                    f"   gpg --batch --dearmor -o \"{dest_path}\" \"{src}\";"
+                    " else"
+                    f"   install -m 0644 \"{src}\" \"{dest_path}\";"
+                    " fi;"
+                    f" chmod 0644 \"{dest_path}\"'"
+                )
         else:
-            host.noop("All keys from {0} are already available in the apt keychain".format(src))
+            host.noop(f"All keys from {src} are already available in the apt keychain")
 
+    # --- keyserver branch: fetch one or multiple keys by ID ---
     if keyserver:
         if not keyid:
             raise OperationError("`keyid` must be provided with `keyserver`")
@@ -105,16 +188,22 @@ def key(src: str | None = None, keyserver: str | None = None, keyid: str | list[
 
         needed_keys = sorted(set(keyid) - set(existing_keys.keys()))
         if needed_keys:
-            yield "apt-key adv --keyserver {0} --recv-keys {1}".format(
-                keyserver,
-                " ".join(needed_keys),
+            dest_path = _derive_dest_from_src_and_keyids(None, needed_keys, dest)
+            joined = " ".join(needed_keys)
+            # Use a temporary GNUPGHOME so we don't pollute the system/user keyring,
+            # then export and dearmor to the APT keyring destination.
+            yield (
+                "sh -c 'set -e;"
+                " install -d -m 0755 /etc/apt/keyrings;"
+                " tmp=$(mktemp -d);"
+                " export GNUPGHOME=\"$tmp\";"
+                f" gpg --batch --keyserver \"{keyserver}\" --recv-keys {joined};"
+                f" gpg --batch --export {joined} | gpg --batch --dearmor -o \"{dest_path}\";"
+                " rm -rf \"$tmp\";"
+                f" chmod 0644 \"{dest_path}\"'"
             )
         else:
-            host.noop(
-                "Keys {0} are already available in the apt keychain".format(
-                    ", ".join(keyid),
-                ),
-            )
+            host.noop(f"Keys {', '.join(keyid)} are already available in the apt keychain")
 
 
 @operation()