22709357475

Committed 05 Mar 2026 08:35AM UTC coverage: 59.732% (-27.2%) from 86.974%

Build # 22709357475

Build Type

Pull #13880

github

Committed by

web-flow

Commit Message

Merge 28fcab93c into 710618057

Pull Request Pull Request #13880: Firehose: Replace TaggingService

Coverage Stats

12 of 12 new or added lines in 2 files covered. (100.0%)

20464 existing lines in 510 files now uncovered.

45290 of 75822 relevant lines covered (59.73%)

0.6 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

63.7

/localstack-core/localstack/utils/checksum.py

import hashlib
import logging
import os
import re
import tempfile
from abc import ABC, abstractmethod

from localstack.utils.files import load_file, rm_rf

# Setup logger
LOG = logging.getLogger(__name__)


class ChecksumException(Exception):
    """Base exception for checksum errors."""

    pass


class ChecksumFormat(ABC):
    """Abstract base class for checksum format parsers."""

    @abstractmethod
    def can_parse(self, content: str) -> bool:
        """
        Check if this parser can handle the given content.

        :param content: The content to check
        :return: True if parser can handle content, False otherwise
        """
        pass

    @abstractmethod
    def parse(self, content: str) -> dict[str, str]:
        """
        Parse the content and return filename to checksum mapping.

        :param content: The content to parse
        :return: Dictionary mapping filenames to checksums
        """
        pass


class StandardFormat(ChecksumFormat):
    """
    Handles standard checksum format.

    Supports formats like:

    * ``checksum  filename``
    * ``checksum *filename``
    """

    def can_parse(self, content: str) -> bool:
        lines = content.strip().split("\n")
        for line in lines[:5]:  # Check first 5 lines
            if re.match(r"^[a-fA-F0-9]{32,128}\s+\S+", line.strip()):
                return True
        return False

    def parse(self, content: str) -> dict[str, str]:
        checksums = {}
        for line in content.strip().split("\n"):
            line = line.strip()
            if not line or line.startswith("#"):
                continue

            # Match: checksum (whitespace) filename
            match = re.match(r"^([a-fA-F0-9]{32,128})\s+(\*?)(.+)$", line)
            if match:
                checksum, star, filename = match.groups()
                checksums[filename.strip()] = checksum.lower()

        return checksums


class BSDFormat(ChecksumFormat):
    """
    Handles BSD-style checksum format.

    Format: ``SHA512 (filename) = checksum``
    """

    def can_parse(self, content: str) -> bool:
        lines = content.strip().split("\n")
        for line in lines[:5]:
            if re.match(r"^(MD5|SHA1|SHA256|SHA512)\s*\(.+\)\s*=\s*[a-fA-F0-9]+", line):
                return True
        return False

    def parse(self, content: str) -> dict[str, str]:
        checksums = {}
        for line in content.strip().split("\n"):
            line = line.strip()
            if not line:
                continue

            # Match: ALGORITHM (filename) = checksum
            match = re.match(r"^(MD5|SHA1|SHA256|SHA512)\s*\((.+)\)\s*=\s*([a-fA-F0-9]+)$", line)
            if match:
                algo, filename, checksum = match.groups()
                checksums[filename.strip()] = checksum.lower()

        return checksums


class ApacheBSDFormat(ChecksumFormat):
    """
    Handles Apache's BSD-style format with split checksums.

    Format::

        filename: CHECKSUM_PART1
                 CHECKSUM_PART2
                 CHECKSUM_PART3
    """

    def can_parse(self, content: str) -> bool:
        lines = content.strip().split("\n")
        if lines and ":" in lines[0]:
            # Check if it looks like filename: hex_data
            parts = lines[0].split(":", 1)
            if len(parts) == 2 and re.search(r"[a-fA-F0-9\s]+", parts[1]):
                return True
        return False

    def parse(self, content: str) -> dict[str, str]:
        checksums = {}
        lines = content.strip().split("\n")

        current_file = None
        checksum_parts = []

        for line in lines:
            if ":" in line and not line.startswith(" "):
                # New file entry
                if current_file and checksum_parts:
                    # Save previous file's checksum
                    full_checksum = "".join(checksum_parts).replace(" ", "").lower()
                    if re.match(r"^[a-fA-F0-9]+$", full_checksum):
                        checksums[current_file] = full_checksum

                # Start new file
                parts = line.split(":", 1)
                current_file = parts[0].strip()
                checksum_part = parts[1].strip()
                checksum_parts = [checksum_part]
            elif line.strip() and current_file:
                # Continuation of checksum
                checksum_parts.append(line.strip())

        # Don't forget the last file
        if current_file and checksum_parts:
            full_checksum = "".join(checksum_parts).replace(" ", "").lower()
            if re.match(r"^[a-fA-F0-9]+$", full_checksum):
                checksums[current_file] = full_checksum

        return checksums


class ChecksumParser:
    """Main parser that tries different checksum formats."""

    def __init__(self):
        """Initialize parser with available format parsers."""
        self.formats = [
            StandardFormat(),
            BSDFormat(),
            ApacheBSDFormat(),
        ]

    def parse(self, content: str) -> dict[str, str]:
        """
        Try each format parser until one works.

        :param content: The content to parse
        :return: Dictionary mapping filenames to checksums
        """
        for format_parser in self.formats:
            if format_parser.can_parse(content):
                result = format_parser.parse(content)
                if result:
                    return result

        return {}


def parse_checksum_file_from_url(checksum_url: str) -> dict[str, str]:
    """
    Parse a SHA checksum file from a URL using multiple format parsers.

    :param checksum_url: URL of the checksum file
    :return: Dictionary mapping filenames to checksums
    """
    # import here to avoid circular dependency issues
    from localstack.utils.http import download

    checksum_name = os.path.basename(checksum_url)
    checksum_path = os.path.join(tempfile.gettempdir(), checksum_name)
    try:
        download(checksum_url, checksum_path)
        checksum_content = load_file(checksum_path)

        parser = ChecksumParser()
        checksums = parser.parse(checksum_content)

        return checksums
    finally:
        rm_rf(checksum_path)


def calculate_file_checksum(file_path: str, algorithm: str = "sha256") -> str:
    """
    Calculate checksum of a local file.

    :param file_path: Path to the file
    :param algorithm: Hash algorithm to use
    :return: Calculated checksum as hexadecimal string

    note: Supported algorithms: 'md5', 'sha1', 'sha256', 'sha512'
    """
    hash_func = getattr(hashlib, algorithm)()

    with open(file_path, "rb") as f:
        # Read file in chunks to handle large files efficiently
        for chunk in iter(lambda: f.read(8192), b""):
            hash_func.update(chunk)

    return hash_func.hexdigest()


def verify_local_file_with_checksum_url(file_path: str, checksum_url: str, filename=None) -> bool:
    """
    Verify a local file against checksums from an online checksum file.

    :param file_path: Path to the local file to verify
    :param checksum_url: URL of the checksum file
    :param filename: Filename to look for in checksum file (defaults to basename of file_path)
    :return: True if verification succeeds, False otherwise

    note: The algorithm is automatically detected based on checksum length:

       * 32 characters: MD5
       * 40 characters: SHA1
       * 64 characters: SHA256
       * 128 characters: SHA512
    """
    # Get checksums from URL
    LOG.debug("Fetching checksums from %s...", checksum_url)
    checksums = parse_checksum_file_from_url(checksum_url)

    if not checksums:
        raise ChecksumException(f"No checksums found in {checksum_url}")

    # Determine filename to look for
    if filename is None:
        filename = os.path.basename(file_path)

    # Find checksum for our file
    if filename not in checksums:
        # Try with different path variations
        possible_names = [
            filename,
            os.path.basename(filename),  # just filename without path
            filename.replace("\\", "/"),  # Unix-style paths
            filename.replace("/", "\\"),  # Windows-style paths
        ]

        found = False
        for name in possible_names:
            if name in checksums:
                filename = name
                found = True
                break

        if not found:
            raise ChecksumException(f"Checksum for {filename} not found in {checksum_url}")

    expected_checksum = checksums[filename]

    # Detect algorithm based on checksum length
    checksum_length = len(expected_checksum)
    if checksum_length == 32:
        algorithm = "md5"
    elif checksum_length == 40:
        algorithm = "sha1"
    elif checksum_length == 64:
        algorithm = "sha256"
    elif checksum_length == 128:
        algorithm = "sha512"
    else:
        raise ChecksumException(f"Unsupported checksum length: {checksum_length}")

    # Calculate checksum of local file
    LOG.debug("Calculating %s checksum of %s...", algorithm, file_path)
    calculated_checksum = calculate_file_checksum(file_path, algorithm)

    is_valid = calculated_checksum == expected_checksum.lower()

    if not is_valid:
        LOG.error(
            "Checksum mismatch for %s: calculated %s, expected %s",
            file_path,
            calculated_checksum,
            expected_checksum,
        )
        raise ChecksumException(
            f"Checksum mismatch for {file_path}: calculated {calculated_checksum}, expected {expected_checksum}"
        )
    LOG.debug("Checksum verification successful for %s", file_path)

    # Compare checksums
    return calculated_checksum == expected_checksum.lower()

1	import hashlib	1✔
2	import logging	1✔
3	import os	1✔
4	import re	1✔
5	import tempfile	1✔
6	from abc import ABC, abstractmethod	1✔
7
8	from localstack.utils.files import load_file, rm_rf	1✔
9
10	# Setup logger
11	LOG = logging.getLogger(__name__)	1✔
12
13
14	class ChecksumException(Exception):	1✔
15	"""Base exception for checksum errors."""
16
17	pass	1✔
18
19
20	class ChecksumFormat(ABC):	1✔
21	"""Abstract base class for checksum format parsers."""
22
23	@abstractmethod	1✔
24	def can_parse(self, content: str) -> bool:	1✔
25	"""
26	Check if this parser can handle the given content.
27
28	:param content: The content to check
29	:return: True if parser can handle content, False otherwise
30	"""
31	pass	×
32
33	@abstractmethod	1✔
34	def parse(self, content: str) -> dict[str, str]:	1✔
35	"""
36	Parse the content and return filename to checksum mapping.
37
38	:param content: The content to parse
39	:return: Dictionary mapping filenames to checksums
40	"""
41	pass	×
42
43
44	class StandardFormat(ChecksumFormat):	1✔
45	"""
46	Handles standard checksum format.
47
48	Supports formats like:
49
50	* ``checksum filename``
51	* ``checksum *filename``
52	"""
53
54	def can_parse(self, content: str) -> bool:	1✔
55	lines = content.strip().split("\n")	1✔
56	for line in lines[:5]: # Check first 5 lines	1✔
57	if re.match(r"^[a-fA-F0-9]{32,128}\s+\S+", line.strip()):	1✔
58	return True	1✔
59	return False	1✔
60
61	def parse(self, content: str) -> dict[str, str]:	1✔
62	checksums = {}	1✔
63	for line in content.strip().split("\n"):	1✔
64	line = line.strip()	1✔
65	if not line or line.startswith("#"):	1✔
66	continue	1✔
67
68	# Match: checksum (whitespace) filename
69	match = re.match(r"^([a-fA-F0-9]{32,128})\s+(\*?)(.+)$", line)	1✔
70	if match:	1✔
71	checksum, star, filename = match.groups()	1✔
72	checksums[filename.strip()] = checksum.lower()	1✔
73
74	return checksums	1✔
75
76
77	class BSDFormat(ChecksumFormat):	1✔
78	"""
79	Handles BSD-style checksum format.
80
81	Format: ``SHA512 (filename) = checksum``
82	"""
83
84	def can_parse(self, content: str) -> bool:	1✔
85	lines = content.strip().split("\n")	1✔
86	for line in lines[:5]:	1✔
87	if re.match(r"^(MD5\|SHA1\|SHA256\|SHA512)\s\(.+\)\s=\s*[a-fA-F0-9]+", line):	1✔
88	return True	1✔
89	return False	1✔
90
91	def parse(self, content: str) -> dict[str, str]:	1✔
92	checksums = {}	1✔
93	for line in content.strip().split("\n"):	1✔
94	line = line.strip()	1✔
95	if not line:	1✔
96	continue	×
97
98	# Match: ALGORITHM (filename) = checksum
99	match = re.match(r"^(MD5\|SHA1\|SHA256\|SHA512)\s\((.+)\)\s=\s*([a-fA-F0-9]+)$", line)	1✔
100	if match:	1✔
101	algo, filename, checksum = match.groups()	1✔
102	checksums[filename.strip()] = checksum.lower()	1✔
103
104	return checksums	1✔
105
106
107	class ApacheBSDFormat(ChecksumFormat):	1✔
108	"""
109	Handles Apache's BSD-style format with split checksums.
110
111	Format::
112
113	filename: CHECKSUM_PART1
114	CHECKSUM_PART2
115	CHECKSUM_PART3
116	"""
117
118	def can_parse(self, content: str) -> bool:	1✔
119	lines = content.strip().split("\n")	1✔
120	if lines and ":" in lines[0]:	1✔
121	# Check if it looks like filename: hex_data
122	parts = lines[0].split(":", 1)	1✔
123	if len(parts) == 2 and re.search(r"[a-fA-F0-9\s]+", parts[1]):	1✔
124	return True	1✔
125	return False	1✔
126
127	def parse(self, content: str) -> dict[str, str]:	1✔
128	checksums = {}	1✔
129	lines = content.strip().split("\n")	1✔
130
131	current_file = None	1✔
132	checksum_parts = []	1✔
133
134	for line in lines:	1✔
135	if ":" in line and not line.startswith(" "):	1✔
136	# New file entry
137	if current_file and checksum_parts:	1✔
138	# Save previous file's checksum
139	full_checksum = "".join(checksum_parts).replace(" ", "").lower()	1✔
140	if re.match(r"^[a-fA-F0-9]+$", full_checksum):	1✔
141	checksums[current_file] = full_checksum	1✔
142
143	# Start new file
144	parts = line.split(":", 1)	1✔
145	current_file = parts[0].strip()	1✔
146	checksum_part = parts[1].strip()	1✔
147	checksum_parts = [checksum_part]	1✔
148	elif line.strip() and current_file:	1✔
149	# Continuation of checksum
150	checksum_parts.append(line.strip())	1✔
151
152	# Don't forget the last file
153	if current_file and checksum_parts:	1✔
154	full_checksum = "".join(checksum_parts).replace(" ", "").lower()	1✔
155	if re.match(r"^[a-fA-F0-9]+$", full_checksum):	1✔
156	checksums[current_file] = full_checksum	1✔
157
158	return checksums	1✔
159
160
161	class ChecksumParser:	1✔
162	"""Main parser that tries different checksum formats."""
163
164	def __init__(self):	1✔
165	"""Initialize parser with available format parsers."""
166	self.formats = [	1✔
167	StandardFormat(),
168	BSDFormat(),
169	ApacheBSDFormat(),
170	]
171
172	def parse(self, content: str) -> dict[str, str]:	1✔
173	"""
174	Try each format parser until one works.
175
176	:param content: The content to parse
177	:return: Dictionary mapping filenames to checksums
178	"""
179	for format_parser in self.formats:	1✔
180	if format_parser.can_parse(content):	1✔
181	result = format_parser.parse(content)	1✔
182	if result:	1✔
183	return result	1✔
184
185	return {}	1✔
186
187
188	def parse_checksum_file_from_url(checksum_url: str) -> dict[str, str]:	1✔
189	"""
190	Parse a SHA checksum file from a URL using multiple format parsers.
191
192	:param checksum_url: URL of the checksum file
193	:return: Dictionary mapping filenames to checksums
194	"""
195	# import here to avoid circular dependency issues
UNCOV 196	from localstack.utils.http import download	×
197
UNCOV 198	checksum_name = os.path.basename(checksum_url)	×
UNCOV 199	checksum_path = os.path.join(tempfile.gettempdir(), checksum_name)	×
UNCOV 200	try:	×
UNCOV 201	download(checksum_url, checksum_path)	×
UNCOV 202	checksum_content = load_file(checksum_path)	×
203
UNCOV 204	parser = ChecksumParser()	×
UNCOV 205	checksums = parser.parse(checksum_content)	×
206
UNCOV 207	return checksums	×
208	finally:
UNCOV 209	rm_rf(checksum_path)	×
210
211
212	def calculate_file_checksum(file_path: str, algorithm: str = "sha256") -> str:	1✔
213	"""
214	Calculate checksum of a local file.
215
216	:param file_path: Path to the file
217	:param algorithm: Hash algorithm to use
218	:return: Calculated checksum as hexadecimal string
219
220	note: Supported algorithms: 'md5', 'sha1', 'sha256', 'sha512'
221	"""
UNCOV 222	hash_func = getattr(hashlib, algorithm)()	×
223
UNCOV 224	with open(file_path, "rb") as f:	×
225	# Read file in chunks to handle large files efficiently
UNCOV 226	for chunk in iter(lambda: f.read(8192), b""):	×
UNCOV 227	hash_func.update(chunk)	×
228
UNCOV 229	return hash_func.hexdigest()	×
230
231
232	def verify_local_file_with_checksum_url(file_path: str, checksum_url: str, filename=None) -> bool:	1✔
233	"""
234	Verify a local file against checksums from an online checksum file.
235
236	:param file_path: Path to the local file to verify
237	:param checksum_url: URL of the checksum file
238	:param filename: Filename to look for in checksum file (defaults to basename of file_path)
239	:return: True if verification succeeds, False otherwise
240
241	note: The algorithm is automatically detected based on checksum length:
242
243	* 32 characters: MD5
244	* 40 characters: SHA1
245	* 64 characters: SHA256
246	* 128 characters: SHA512
247	"""
248	# Get checksums from URL
UNCOV 249	LOG.debug("Fetching checksums from %s...", checksum_url)	×
UNCOV 250	checksums = parse_checksum_file_from_url(checksum_url)	×
251
UNCOV 252	if not checksums:	×
253	raise ChecksumException(f"No checksums found in {checksum_url}")	×
254
255	# Determine filename to look for
UNCOV 256	if filename is None:	×
UNCOV 257	filename = os.path.basename(file_path)	×
258
259	# Find checksum for our file
UNCOV 260	if filename not in checksums:	×
261	# Try with different path variations
262	possible_names = [	×
263	filename,
264	os.path.basename(filename), # just filename without path
265	filename.replace("\\", "/"), # Unix-style paths
266	filename.replace("/", "\\"), # Windows-style paths
267	]
268
269	found = False	×
270	for name in possible_names:	×
271	if name in checksums:	×
272	filename = name	×
273	found = True	×
274	break	×
275
276	if not found:	×
277	raise ChecksumException(f"Checksum for {filename} not found in {checksum_url}")	×
278
UNCOV 279	expected_checksum = checksums[filename]	×
280
281	# Detect algorithm based on checksum length
UNCOV 282	checksum_length = len(expected_checksum)	×
UNCOV 283	if checksum_length == 32:	×
284	algorithm = "md5"	×
UNCOV 285	elif checksum_length == 40:	×
286	algorithm = "sha1"	×
UNCOV 287	elif checksum_length == 64:	×
UNCOV 288	algorithm = "sha256"	×
289	elif checksum_length == 128:	×
290	algorithm = "sha512"	×
291	else:
292	raise ChecksumException(f"Unsupported checksum length: {checksum_length}")	×
293
294	# Calculate checksum of local file
UNCOV 295	LOG.debug("Calculating %s checksum of %s...", algorithm, file_path)	×
UNCOV 296	calculated_checksum = calculate_file_checksum(file_path, algorithm)	×
297
UNCOV 298	is_valid = calculated_checksum == expected_checksum.lower()	×
299
UNCOV 300	if not is_valid:	×
301	LOG.error(	×
302	"Checksum mismatch for %s: calculated %s, expected %s",
303	file_path,
304	calculated_checksum,
305	expected_checksum,
306	)
307	raise ChecksumException(	×
308	f"Checksum mismatch for {file_path}: calculated {calculated_checksum}, expected {expected_checksum}"
309	)
UNCOV 310	LOG.debug("Checksum verification successful for %s", file_path)	×
311
312	# Compare checksums
UNCOV 313	return calculated_checksum == expected_checksum.lower()	×

localstack / localstack / 22709357475

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous