Strip off query part from url when operating on the path

This allows for using bmaptool, for example, with artifacts from gitlab
CI which have a ?job=... query parameter at the end of the URL. This
commit strips off the query part to operate on the filename extension or
compute the basename.

src/bmaptool/BmapCopy.py

@@ -705,7 +705,7 @@ def copy(self, sync=True, verify=True):
         # Read the image in '_batch_blocks' chunks and write them to the
         # destination file
         while True:
-            batch = self._batch_queue.get()
+            batch = self._batch_queue.get(timeout=60)
             if batch is None:
                 # No more data, the image is written
                 break

src/bmaptool/CLI.py

@@ -39,6 +39,7 @@
 import shutil
 import io
 import pathlib
+import urllib.parse
 from . import BmapCreate, BmapCopy, BmapHelpers, TransRead
 
 VERSION = "3.8.0"
@@ -156,7 +157,7 @@ def report_verification_results(context, sigs):
             )
 
 
-def verify_detached_bmap_signature(args, bmap_obj, bmap_path):
+def verify_detached_bmap_signature(args, bmap_obj, bmap_path, is_url):
     """
     This is a helper function for 'verify_bmap_signature()' which handles the
     detached signature case.
@@ -172,13 +173,20 @@ def verify_detached_bmap_signature(args, bmap_obj, bmap_path):
             error_out("cannot open bmap signature file '%s':\n%s", args.bmap_sig, err)
         sig_path = args.bmap_sig
     else:
+        def _add_ext(p, ext):
+            if not is_url:
+                return p + ext
+            # if the image is a url, add the extension to the 'path' part
+            # before the query string and url fragment
+            o = urllib.parse.urlparse(p)
+            return o._replace(path=o.path+ext).geturl()
         # Check if there is a stand-alone signature file
         try:
-            sig_path = bmap_path + ".asc"
+            sig_path = _add_ext(bmap_path, ".asc")
             sig_obj = TransRead.TransRead(sig_path)
         except TransRead.Error:
             try:
-                sig_path = bmap_path + ".sig"
+                sig_path = _add_ext(bmap_path, ".sig")
                 sig_obj = TransRead.TransRead(sig_path)
             except TransRead.Error:
                 # No signatures found
@@ -290,7 +298,7 @@ def verify_clearsign_bmap_signature(args, bmap_obj):
     return tmp_obj
 
 
-def verify_bmap_signature(args, bmap_obj, bmap_path):
+def verify_bmap_signature(args, bmap_obj, bmap_path, is_url):
     """
     Verify GPG signature of the bmap file if it is present. The signature may
     be in a separate file (detached) or it may be inside the bmap file itself
@@ -323,10 +331,10 @@ def verify_bmap_signature(args, bmap_obj, bmap_path):
     if buf == clearsign_marker:
         return verify_clearsign_bmap_signature(args, bmap_obj)
     else:
-        return verify_detached_bmap_signature(args, bmap_obj, bmap_path)
+        return verify_detached_bmap_signature(args, bmap_obj, bmap_path, is_url)
 
 
-def find_and_open_bmap(args):
+def find_and_open_bmap(args, is_url=False):
     """
     This is a helper function for 'open_files()' which discovers and opens the
     bmap file, then returns the corresponding file object and the bmap file
@@ -358,15 +366,27 @@ def find_and_open_bmap(args):
         # Automatically discover the bmap file
         image_path = args.image
         while True:
-            bmap_path = image_path + ".bmap"
+            if is_url:
+                # if the image is a url, add the extention to the 'path' part
+                # before the query string and url fragment
+                o = urllib.parse.urlparse(image_path)
+                bmap_path = o._replace(path=o.path + ".bmap").geturl()
+            else:
+                bmap_path = image_path + ".bmap"
             try:
                 bmap_obj = TransRead.TransRead(bmap_path)
                 log.info("discovered bmap file '%s'" % bmap_path)
                 break
             except TransRead.Error:
                 pass
 
-            image_path, ext = os.path.splitext(image_path)
+            if is_url:
+                # if the image is a url, split the extension from the 'path'
+                o = urllib.parse.urlparse(image_path)
+                p, ext = os.path.splitext(o.path)
+                image_path = o._replace(path=p).geturl()
+            else:
+                image_path, ext = os.path.splitext(image_path)
             if ext == "":
                 return (None, None)
 
@@ -408,7 +428,7 @@ def open_files(args):
 
     # Open the bmap file. Try to discover the bmap file automatically if it
     # was not specified.
-    (bmap_obj, bmap_path) = find_and_open_bmap(args)
+    (bmap_obj, bmap_path) = find_and_open_bmap(args, image_obj.is_url)
 
     if bmap_path == args.image:
         # Most probably the user specified the bmap file instead of the image
@@ -511,7 +531,7 @@ def copy_command(args):
             "the bmap signature file was specified, but bmap file was " "not found"
         )
 
-    f_obj = verify_bmap_signature(args, bmap_obj, bmap_path)
+    f_obj = verify_bmap_signature(args, bmap_obj, bmap_path, image_obj.is_url)
     if f_obj:
         bmap_obj.close()
         bmap_obj = f_obj
@@ -561,9 +581,15 @@ def copy_command(args):
                 writer.mapped_percent,
             )
         )
+        def _get_basename(p):
+            if image_obj.is_url:
+                # if this is a url, strip off potential query string and
+                # fragment from the end
+                p = urllib.parse.urlparse(p).path
+            return os.path.basename(p)
         log.info(
             "copying image '%s' to %s using bmap file '%s'"
-            % (os.path.basename(args.image), dest_str, os.path.basename(bmap_path))
+            % (_get_basename(args.image), dest_str, _get_basename(bmap_path))
         )
 
     if args.psplash_pipe:

src/bmaptool/TransRead.py

@@ -583,6 +583,10 @@ def _print_warning(timeout):
 
         parsed_url = urllib.parse.urlparse(url)
 
+        # figuring out the decompression program to use relies on the
+        # extension, so strip off any potential query parts
+        self.name = parsed_url.path
+
         if parsed_url.scheme == "ssh":
             # Unfortunately, urllib2 does not handle "ssh://" URLs
             self._open_url_ssh(parsed_url)