fetchfromtask: extracts propagator binary from a 'task' repo

devel/fetchfromtask.py

@@ -0,0 +1,137 @@
+#!/usr/bin/env python3
+
+# Retreive propagator rpm from 'task' repository and extract
+# /usr/sbin/propagator binary from it
+#
+# Dependencies:
+# pythonic:
+# - requests
+#   https://github.com/psf/requests, version 2.25.1 is known to work
+# others:
+# - coreutils
+#   sha1sum utility
+# - cpio
+# - rpm2cpio
+# - xz-utils
+#   xzcat utility
+
+import argparse
+import logging
+import os
+import re
+import requests
+import platform
+import shutil
+import subprocess
+
+NATIVE_ARCH = platform.uname().machine
+# Note: propagator is not built for armhf
+ALT_ARCHES = ['aarch64', 'i586', 'ppc64le', 'x86_64']
+
+
+def guess_url(taskid, pkgname='propagator', arch=NATIVE_ARCH):
+    pkg_regex = re.compile(f'^{pkgname}-(?!debuginfo).*\\.{arch}\\.rpm$')
+    baseurl = f"http://git.altlinux.org/tasks/{taskid}/build/repo/{arch}"
+
+    pkglist_url = f'{baseurl}/base/pkglist.task.xz'
+    logging.debug('guess_url: downloading pkglist from %s', pkglist_url)
+    req = requests.get(pkglist_url)
+    req.raise_for_status()
+    logging.debug('guess_url: uncompressing pkglist with xzcat')
+    xz = subprocess.Popen(['xzcat'],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            stdin=subprocess.PIPE)
+    pkglist, errs = xz.communicate(input=req.content)
+    rc = xz.wait()
+    if rc != 0:
+        logging.error('failed to unpack pkglist, error %d', rc)
+        raise RuntimeError(f"xzcat returned {rc}")
+
+    logging.debug('guess_url: extracting strings from pkglist')
+    proc = subprocess.Popen(['strings'],
+              stdout=subprocess.PIPE,
+              stderr=subprocess.PIPE,
+              stdin=subprocess.PIPE)
+    strings, errs = proc.communicate(input=pkglist)
+    rc = proc.wait()
+    if rc != 0:
+        logging.error('strings failed: %d', rc)
+        raise RuntimeError(f"strings failed: {rc}")
+    rpm_name = None
+    for line in strings.decode('utf-8').split('\n'):
+        logging.debug('guess_url: processing "%s"', line)
+        if pkg_regex.match(line) is not None:
+            logging.debug('OK, found "%s": "%s"', pkgname, line)
+            rpm_name = line
+            break
+    if rpm_name is None:
+        raise RuntimeError(f"No package {pkgname} in task {taskid} found")
+    return rpm_name, f"{baseurl}/RPMS.task/{rpm_name}"
+
+
+def download(url):
+    subprocess.check_call(['wget', '-N', url])
+
+
+def rm_rf(path):
+    try:
+        shutil.rmtree(path)
+    except FileNotFoundError:
+        pass
+
+def extract_rpm(rpm_name, arch=NATIVE_ARCH, pattern='*'):
+    rm_rf(arch)
+    os.makedirs(arch)
+    logging.debug('extract_rpm: running "rpm2cpio %s"', rpm_name)
+    rpm2cpio = subprocess.Popen(['rpm2cpio', rpm_name],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE)
+    logging.debug('extract_rpm: piping to "cpio -id"')
+    cpio = subprocess.Popen(['cpio', '-id', '--no-absolute-filenames', pattern],
+            stdin=rpm2cpio.stdout,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            cwd=arch)
+    rpm2cpio.stdout.close() # Allow rpm2cpio to receive a SIGPIPE if cpio exits
+    out, err = cpio.communicate()
+    rc1 = rpm2cpio.wait()
+    rc2 = cpio.wait()
+    if rc1 != 0:
+        raise RuntimeError(f'rpm2cpio failed: {rc1}')
+    if rc2 != 0:
+        logging.error('extract_rpm: cpio failed: %d, "%s"', rc2, err)
+        raise RuntimeError(f'cpio failed: {rc2}')
+
+
+def process(taskid, arch):
+    rpm_name, rpm_url = guess_url(taskid, pkgname='propagator', arch=arch)
+    download(rpm_url)
+    extract_rpm(rpm_name, arch=arch, pattern='./usr/sbin/propagator')
+
+
+def main():
+    parser = argparse.ArgumentParser(description='extract propagator binary from a "task" repo')
+    parser.add_argument('taskid', type=int)
+    parser.add_argument('-v', '--verbose', action='count', default=0,
+                        help='Be more verbose')
+    parser.add_argument('-a', '--arch',
+                        choices=ALT_ARCHES + ['all'],
+                        default=NATIVE_ARCH,
+                        help='architecture')
+    args = parser.parse_args()
+    loglevel = logging.INFO
+    if args.verbose >= 1:
+        loglevel = logging.DEBUG
+    logging.basicConfig(format='fetchfromtask:%(levelname)s:%(message)s',
+                        level=loglevel)
+    if args.arch == 'all':
+        for arch in ALT_ARCHES:
+            process(args.taskid, arch)
+    else:
+        process(args.taskid, args.arch)
+
+
+if __name__ == '__main__':
+
+    main()