Add checksum_url option

Signed-off-by: Matthias Harzer <matthias.harzer@inovex.de>

Author: matthiasharzer

etc/schema.yaml

@@ -45,6 +45,7 @@ meta:
 versions:
   build_date: any(day(), timestamp())
   checksum: regex(r'\w+:([a-f0-9]{32}|[a-f0-9]{40}|[a-f0-9]{64}|[a-f0-9]{128})$', name='valid checksum', required=False)
+  checksum_url: regex(r'^(http|ftp)s?:\/\/\w+.*', name='valid URL', required=False)
   checksums_url: regex(r'^(http|ftp)s?:\/\/\w+.*', name='valid URL', required=False)
   hidden: bool(required=False)
   image_description: str(required=False)

openstack_image_manager/main.py

@@ -174,7 +174,15 @@ def read_image_files(self, return_all_images=False) -> list:
                     logger.error(exc)
         return all_images
 
-    def get_checksum(self, url: str, checksums_url: str) -> str:
+    def is_checksum(self, string: str) -> bool:
+        return (
+            len(string) == 128
+            or len(string) == 64
+            or len(string) == 40
+            or len(string) == 32
+        ) and "." not in string
+
+    def get_checksum_from_checksums_url(self, url: str, checksums_url: str) -> str:
         """
         Get the checksum of an upstream image by parsing its corresponding checksums file
 
@@ -191,15 +199,27 @@ def get_checksum(self, url: str, checksums_url: str) -> str:
             if filename in line:
                 split = line.split(" ")
                 for elem in split:
-                    if (
-                        len(elem) == 128
-                        or len(elem) == 64
-                        or len(elem) == 40
-                        or len(elem) == 32
-                    ) and "." not in elem:
+                    if self.is_checksum(elem):
                         return elem
         return ""
 
+    def get_checksum_from_checksum_url(self, checksum_url: str) -> str:
+        """
+        Get the checksum from a checksum_url
+
+        Params:
+            checksum_url: the URL of the checksum file
+
+        Returns:
+            the checksum, if it is available or else an empty string
+        """
+        checksum_file_content = requests.get(checksum_url).text.strip()
+
+        if self.is_checksum(checksum_file_content):
+            return checksum_file_content
+
+        return ""
+
     def create_connection(self) -> None:
         if "OS_AUTH_URL" in os.environ:
             self.conn = openstack.connect()
@@ -323,13 +343,21 @@ def process_images(self, images) -> set:
                         versions[version["version"]]["hidden"] = version["hidden"]
 
                     if version["version"] == "latest":  #
+                        if "checksums_url" in version and "checksum_url" in version:
+                            raise ValueError(
+                                'You may only specify either "checksums_url" or "checksum_url", not both'
+                            )
                         if "checksums_url" in version:
                             versions[version["version"]]["checksums_url"] = version[
                                 "checksums_url"
                             ]
+                        elif "checksum_url" in version:
+                            versions[version["version"]]["checksum_url"] = version[
+                                "checksum_url"
+                            ]
                         else:
                             raise ValueError(
-                                'Key "checksums_url" is required when using version "latest"'
+                                'Key "checksums_url" or "checksum_url" is required when using version "latest"'
                             )
 
                     if "meta" in version:
@@ -609,13 +637,21 @@ def process_image(
                 existence = image["name"] in cloud_images
 
             if version == "latest":
-                checksums_url = versions[version]["checksums_url"]
-                upstream_checksum = self.get_checksum(
-                    versions[version]["url"], checksums_url
-                )
+                checksums_url = versions[version].get("checksums_url")
+                checksum_url = versions[version].get("checksum_url")
+
+                if checksums_url:
+                    upstream_checksum = self.get_checksum_from_checksums_url(
+                        versions[version]["url"], checksums_url
+                    )
+                else:
+                    upstream_checksum = self.get_checksum_from_checksum_url(
+                        checksum_url
+                    )
+
                 if not upstream_checksum:
                     logger.error(
-                        f"Could not find checksum for image '{image['name']}', check the checksums_url"
+                        f"Could not find checksum for image '{image['name']}', check the checksums_url or checksum_url"
                     )
                     return existing_images, imported_image, previous_image