From 580370d56789a3ebce8675355bfd95d2a6d58b1a Mon Sep 17 00:00:00 2001
From: Ivan Kruglov <mail@ikruglov.com>
Date: May 28 2024 23:57:46 +0000
Subject: [PATCH 1/2] draft of --verify-checksum which validates chechsum betwee 'sources' and already downloaded files


workflow:
1. $ ./rpmdev-spectool --get-files -C ~/tmp/files ~/tmp/retsnoop.spec
2. $ ./rpmdev-spectool --verify-checksum ~/tmp/sources -C ~/tmp/files ~/tmp/retsnoop.spec
Checksum: bpftool-7.2.0^20230926gite17d6cf.tar.gz OK
Checksum: libbpf-1.2.2^20230915git56069cd.tar.gz OK
Checksum: retsnoop-0.9.8.tar.gz OK

TODOs to discuss:
1. refactor main() to create clear helper functions, ex: one per mode:
   list_files
   get_files
   verify_checksum

2. verify_checksum auto download/discover 'sources' file
3. re-download source files instead of requiring to --get-files them
   first
4. output format

---

diff --git a/rpmdev-spectool b/rpmdev-spectool
index 2c06ef8..0bd0097 100755
--- a/rpmdev-spectool
+++ b/rpmdev-spectool
@@ -24,11 +24,13 @@ import argparse
 import os
 import tempfile
 import time
+import hashlib
 from collections import OrderedDict
 from typing import Any, Callable, Dict, List, Optional
 from typing import OrderedDict as OrderedDictT
 from typing import Tuple
 from urllib.parse import urlparse
+from pyrpkg import SourcesFile
 
 import progressbar
 import requests
@@ -88,6 +90,14 @@ def get_args() -> Dict[str, Any]:
     )
 
     ops.add_argument(
+        "--verify-checksum",
+        "-vc",
+        metavar="PATH_TO_SOURCES_FILE",
+        default="sources",
+        help="Verify checksums of downloaded sources",
+    )
+
+    ops.add_argument(
         "--version",
         "-V",
         action="version",
@@ -497,7 +507,7 @@ def main() -> int:
             print("RPM Failed to parse spec file.")
             return 1
 
-    if args["list_files"] and not args["get_files"]:
+    if args["list_files"] and not args["get_files"] and not args["verify_checksum"]:
         if args["source"]:
             numbers = split_numbers(args["source"])
 
@@ -524,14 +534,8 @@ def main() -> int:
         elif args["patches"] and not args["source"]:
             spec.list_patches()
 
-    if args["get_files"]:
-        force = args["force"]
-        dry = args["dry_run"]
-        headers = {}
-        for header in args["headers"]:
-            k, sep, v = header.partition(':')
-            headers[k.strip()] = v.strip()
-
+    # TODO(ikruglov): create helper functions
+    if args["get_files"] or args["verify_checksum"]:
         if args["directory"] and args["sourcedir"]:
             print("Conflicting requests for download directory.")
             return 1
@@ -543,6 +547,14 @@ def main() -> int:
         else:
             directory = os.getcwd()
 
+    if args["get_files"]:
+        force = args["force"]
+        dry = args["dry_run"]
+        headers = {}
+        for header in args["headers"]:
+            k, sep, v = header.partition(':')
+            headers[k.strip()] = v.strip()
+
         tasks: List[Tuple[Callable[..., bool], Tuple[Any, ...]]] = []
 
         if args["source"]:
@@ -584,6 +596,43 @@ def main() -> int:
         if failure:
             return 1
 
+    if args["verify_checksum"]:
+        sources_path = args["verify_checksum"]
+
+        try:
+            sources_file = SourcesFile(sources_path, 'bsd')
+        except ValueError:
+            print("RPM Failed to parse 'sources' file.")
+            return 1
+
+        failure = False
+        for entry in sources_file.entries:
+            computed_hash = None
+            message = "Unknown"
+            file = entry.file
+
+            # TODO(ikruglov): helper functions
+            try:
+                path = os.path.join(directory, file)
+                with open(path, 'rb', buffering=0) as f:
+                    computed_hash = hashlib.file_digest(f, entry.hashtype).hexdigest()
+            except FileNotFoundError:
+                message = "File Not Found"
+            except e:
+                message = "Unknown: {}".format(e)
+
+            if computed_hash is None:
+                print("Checksum: {} FAIL {}".format(file, message))
+                failure = True
+            elif computed_hash.lower() == entry.hash.lower():
+                print("Checksum: {} OK".format(file))
+            else:
+                print("Checksum: {} FAIL checksum mismatch".format(file))
+                failure = True
+
+        if failure:
+            return 1
+
     return 0
 
 

From 8de86c7ccf6d8688b350f1b0658409bdb0810165 Mon Sep 17 00:00:00 2001
From: Ivan Kruglov <mail@ikruglov.com>
Date: Aug 01 2024 01:59:58 +0000
Subject: [PATCH 2/2] embed SourcesFile


---

diff --git a/rpmdev-spectool b/rpmdev-spectool
index 0bd0097..dad1e05 100755
--- a/rpmdev-spectool
+++ b/rpmdev-spectool
@@ -22,6 +22,7 @@
 
 import argparse
 import os
+import re
 import tempfile
 import time
 import hashlib
@@ -30,7 +31,6 @@ from typing import Any, Callable, Dict, List, Optional
 from typing import OrderedDict as OrderedDictT
 from typing import Tuple
 from urllib.parse import urlparse
-from pyrpkg import SourcesFile
 
 import progressbar
 import requests
@@ -91,9 +91,8 @@ def get_args() -> Dict[str, Any]:
 
     ops.add_argument(
         "--verify-checksum",
-        "-vc",
+        "--vc",
         metavar="PATH_TO_SOURCES_FILE",
-        default="sources",
         help="Verify checksums of downloaded sources",
     )
 
@@ -452,6 +451,89 @@ class Spec:
 
         return failure
 
+# copy-paste with modification from https://pagure.io/rpkg/blob/master/f/pyrpkg/sources.py
+LINE_PATTERN = re.compile(
+    r'^(?P<hashtype>[^ ]+?) \((?P<file>[^ )]+?)\) = (?P<hash>[^ ]+?)$')
+
+class MalformedLineError(Exception):
+    """Raised when parsing a sources file with malformed lines"""
+    pass
+
+class SourcesFile:
+    def __init__(self, sourcesfile, entry_type):
+        self.sourcesfile = sourcesfile
+        self.entry_type = {'old': SourceFileEntry,
+                           'bsd': BSDSourceFileEntry}[entry_type]
+        self.entries = []
+        # if there is a directory with the same name, it causes a collision
+        # during reading the file and during its replacing as well
+        if os.path.exists(sourcesfile) and os.path.isdir(sourcesfile):
+            raise ValueError(
+                "'{0}' has to be a regular file, not a directory".format(sourcesfile))
+
+        with open(sourcesfile) as f:
+            for line in f:
+                entry = self.parse_line(line)
+
+                if entry and entry not in self.entries:
+                    self.entries.append(entry)
+
+    def parse_line(self, line):
+        stripped = line.strip()
+
+        if not stripped:
+            return
+
+        m = LINE_PATTERN.match(stripped)
+        if m is not None:
+            return self.entry_type(m.group('hashtype'), m.group('file'),
+                                   m.group('hash'))
+
+        # Try falling back on the old Fedora format
+        try:
+            hash, file = stripped.split('  ', 1)
+
+        except ValueError:
+            # Try old Centos Format
+            try:
+                hash, file_path = stripped.split(' ', 1)
+                if len(hash) == 128:
+                    hashtype = 'sha512'
+                elif len(hash) == 64:
+                    hashtype = 'sha256'
+                elif len(hash) == 40:
+                    hashtype = 'sha1'
+                elif len(hash) == 32:
+                    hashtype = 'md5'
+                else:
+                    raise MalformedLineError(
+                        'sources has invalid content: {0}'.format(stripped))
+                file = os.path.split(file_path)[1]
+                return self.entry_type(hashtype, file, hash)
+
+            except ValueError:
+                raise MalformedLineError(
+                    'sources has invalid content: {0}'.format(stripped))
+
+        return self.entry_type('md5', file, hash)
+
+class SourceFileEntry(object):
+    def __init__(self, hashtype, file, hash):
+        self.hashtype = hashtype.lower()
+        self.hash = hash
+        self.file = file
+
+    def __str__(self):
+        return '%s  %s\n' % (self.hash, self.file)
+
+    def __eq__(self, other):
+        return ((self.hashtype, self.hash, self.file) ==
+                (other.hashtype, other.hash, other.file))
+
+class BSDSourceFileEntry(SourceFileEntry):
+    def __str__(self):
+        return '%s (%s) = %s\n' % (self.hashtype.upper(), self.file,
+                                   self.hash)
 
 def main() -> int:
     args = get_args()
@@ -601,8 +683,8 @@ def main() -> int:
 
         try:
             sources_file = SourcesFile(sources_path, 'bsd')
-        except ValueError:
-            print("RPM Failed to parse 'sources' file.")
+        except (MalformedLineError, ValueError, IOError) as e:
+            print("Failed to parse 'sources' file: {}".format(e))
             return 1
 
         failure = False
@@ -611,15 +693,12 @@ def main() -> int:
             message = "Unknown"
             file = entry.file
 
-            # TODO(ikruglov): helper functions
             try:
                 path = os.path.join(directory, file)
-                with open(path, 'rb', buffering=0) as f:
-                    computed_hash = hashlib.file_digest(f, entry.hashtype).hexdigest()
-            except FileNotFoundError:
-                message = "File Not Found"
-            except e:
-                message = "Unknown: {}".format(e)
+                f = open(path, 'rb')
+                computed_hash = hashlib.file_digest(f, entry.hashtype).hexdigest()
+            except (ValueError, IOError) as e:
+                message = "Failed to compute hash of {}: {}".format(path, e)
 
             if computed_hash is None:
                 print("Checksum: {} FAIL {}".format(file, message))

