Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion src/specify_cli/integrations/catalog.py
Original file line number Diff line number Diff line change
Expand Up @@ -841,5 +841,8 @@ def tools(self) -> List[Dict[str, Any]]:

def get_hash(self) -> str:
"""SHA-256 hash of the descriptor file."""
h = hashlib.sha256()
with open(self.path, "rb") as fh:
return f"sha256:{hashlib.sha256(fh.read()).hexdigest()}"
for chunk in iter(lambda: fh.read(65536), b""):
Comment thread
Quratulain-bilal marked this conversation as resolved.
h.update(chunk)
return f"sha256:{h.hexdigest()}"
68 changes: 68 additions & 0 deletions tests/integrations/test_integration_catalog.py
Original file line number Diff line number Diff line change
Expand Up @@ -646,6 +646,74 @@ def test_get_hash(self, tmp_path):
h = desc.get_hash()
assert h.startswith("sha256:")

def test_get_hash_incremental_chunking(self, tmp_path):
"""Regression: get_hash() must read in bounded chunks, not f.read().

Creates a file larger than 64 KiB to exercise multiple chunk
iterations, verifies the digest matches hashlib.sha256(content),
and asserts that raw read() is never called with no size limit.
"""
import hashlib
from unittest.mock import patch

# Build valid YAML content larger than 64 KiB. The padding field
# inflates the file without breaking YAML structure.
padding = "P" * (200 * 1024) # 200 KiB
data = {**VALID_DESCRIPTOR, "description": padding}
content = yaml.dump(data).encode("utf-8")
assert len(content) > 65536 # Ensure multi-chunk coverage.

p = tmp_path / "integration.yml"
p.write_bytes(content)
desc = IntegrationDescriptor(p)

# Spy on read() to reject unbounded calls. We cannot assign
# directly to fh.read on a BufferedReader (read-only attribute),
# so wrap the file in a proxy that intercepts read() while
# delegating everything else — including context-manager cleanup
# — to the real file object.
original_open = open
read_sizes: list[int] = []

class _ReadTrackingProxy:
"""Proxy that intercepts read() calls on a file object."""

def __init__(self, fh):
self._fh = fh

def read(self, n=-1):
if n == -1 or n is None:
raise RuntimeError(
"f.read() called without size limit — "
"use bounded chunked reads instead"
)
read_sizes.append(n)
return self._fh.read(n)

def __getattr__(self, name):
return getattr(self._fh, name)

def __enter__(self):
return self

def __exit__(self, *args):
return self._fh.__exit__(*args)

def tracking_open(path, *args, **kwargs):
fh = original_open(path, *args, **kwargs)
if (args and args[0] == "rb") or kwargs.get("mode") == "rb":
return _ReadTrackingProxy(fh)
return fh

with patch("builtins.open", side_effect=tracking_open):
result = desc.get_hash()

expected = f"sha256:{hashlib.sha256(content).hexdigest()}"
assert result == expected
# Verify bounded reads happened (at least 3 chunks of <= 65536 bytes).
assert len(read_sizes) >= 3
assert all(s <= 65536 for s in read_sizes)

def test_tools_accessor(self, tmp_path):
data = {**VALID_DESCRIPTOR, "requires": {
"speckit_version": ">=0.6.0",
Expand Down