Downloads latest qcow2 images of Debian and Rocky. Use in automated deployment environments, expand to other OSes as desired.

qcow2-image-downloader.py

# qcow2-image-downloader.py - downloads qcow2 images from Debian and Rocky repositories.
#
# meant to be run periodically by a systemd unit/timer.
#
# use for automated deployment environments where having the latest OS patches at first boot is a necessity.
#
# license: BSD-3-Clause (https://opensource.org/license/bsd-3-clause)
# 
# copyright (C) 2025 Dayton Hasty (dayt0n)

import hashlib
import os
import re
import sys
from typing import Dict, Literal, Optional
import requests

if len(sys.argv) < 2:
    print(f"Incorrect usage\nusage: {sys.argv[0]} [path/to/download/directory]")
    sys.exit(-1)

outdir = sys.argv[1]


class OS:
    name: str = None
    version: int = None
    url_format: str = None
    hash_type: str = None
    session: requests.Session = None
    checksums: Dict = None
    latest_format: str = None
    version_name: Optional[str] = None

    def __init__(
        self,
        name,
        version,
        version_name=None,
        url_format="https://dl.rockylinux.org/pub/{name}/{version}/images/x86_64",
        checksum_file="CHECKSUM",
        checksum_format=r"(?P<hash>[0-9a-f]+)\s+(?P<file>.*)",
        hash_type: Literal["SHA256", "SHA512", "MD5"] = "SHA256",
        latest_format="Rocky-{version_number}-GenericCloud.latest.x86_64.qcow2",
    ):
        self.session = requests.session()

        self.name = name
        self.version = version
        self.version_name = version_name
        self.url_format = url_format
        self.hash_type = hash_type
        self.checksums = self.getsums(checksum_file, re.compile(checksum_format))
        self.latest_format = latest_format

    @property
    def baseurl(self):
        return self.url_format.format(
            name=self.name, version=self.version, version_name=self.version_name
        )

    def getsums(self, checksum_file, checksum_regex) -> Dict:
        r = self.session.get(f"{self.baseurl}/{checksum_file}")
        r.raise_for_status()
        sums = {}
        for line in r.text.splitlines():
            res = checksum_regex.search(line)
            if res is None:
                continue
            d = res.groupdict()
            h = d.get("hash")
            f = d.get("file")
            if not h or not f:
                continue
            sums.update({f: h})
        return sums

    def get_hash(self, filename):
        return self.checksums.get(filename)

    def hash_file(self, path):
        with open(path, "rb", buffering=0) as fp:
            return hashlib.file_digest(fp, self.hash_type.lower()).hexdigest()

    def pull(self):
        fn = self.latest_format.format(version_number=self.version)
        outfile = os.path.join(outdir, f"{self.name.lower()}{self.version}.qcow2")
        known_hash = self.checksums.get(fn)
        if not known_hash:
            raise Exception("Unable to find checksum for file")
        if os.path.exists(outfile) and self.hash_file(outfile) == known_hash:
            print(
                f"{self.name.capitalize()} {self.version} already downloaded, skipping..."
            )
            return
        print(f"Downloading {self.name.capitalize()} {self.version}...")
        with self.session.get(f"{self.baseurl}/{fn}", stream=True) as r:
            r.raise_for_status()
            with open(outfile, "wb") as fp:
                for chunk in r.iter_content(chunk_size=8192):
                    fp.write(chunk)
        fh = self.hash_file(outfile)
        if known_hash != fh:
            raise Exception("File hash does not match known value")

        return outfile


class RockyOS(OS):
    def __init__(
        self,
        version,
    ):
        checksum_format = r"SHA256 \((?P<file>.+)\) = (?P<hash>[0-9a-f]+)"
        super().__init__(
            "rocky",
            version,
            checksum_format=checksum_format,
        )


class DebianOS(OS):
    def __init__(
        self,
        version_name,
        version,
    ):
        super().__init__(
            "debian",
            version,
            version_name=version_name,
            url_format="https://cloud.debian.org/images/cloud/{version_name}/latest",
            checksum_file="SHA512SUMS",
            hash_type="SHA512",
            latest_format="debian-{version_number}-generic-amd64.qcow2",
        )


# define images to download here
to_pull = [
    DebianOS("bullseye", 11),
    DebianOS("bookworm", 12),
    DebianOS("trixie", 13),
    RockyOS(9),
]


def main():
    os.makedirs(outdir, mode=0o777, exist_ok=True)
    image: OS
    for image in to_pull:
        try:
            image.pull()
        except Exception as e:
            print(f"Failed to pull image: {e}")


if __name__ == "__main__":
    main()