From 580370d56789a3ebce8675355bfd95d2a6d58b1a Mon Sep 17 00:00:00 2001 From: Ivan Kruglov 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 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[^ ]+?) \((?P[^ )]+?)\) = (?P[^ ]+?)$') + +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))