diff --git a/security/tar_safe.py b/security/tar_safe.py index 2aa3c3ad..3c6ef50a 100644 --- a/security/tar_safe.py +++ b/security/tar_safe.py @@ -197,7 +197,10 @@ def _extract_file( while bytes_copied < member.size: chunk = src.read(min(chunk_size, member.size - bytes_copied)) if not chunk: - break + raise UnsafeTarError( + f"Truncated file: {member.name} - expected {member.size} bytes, " + f"got {bytes_copied}" + ) bytes_copied += len(chunk) tmp.write(chunk) diff --git a/tests/test_security.py b/tests/test_security.py index cab638b9..b8600943 100644 --- a/tests/test_security.py +++ b/tests/test_security.py @@ -217,6 +217,27 @@ def test_nested_valid_directories(self, tmp_path): assert len(extracted) == 1 assert (extract_dir / 'dir1' / 'dir2' / 'file.txt').exists() + def test_truncated_file_detection(self, tmp_path): + """Test that truncated files (size metadata > actual data) are detected.""" + tar_file = tmp_path / "truncated.tar" + + with tarfile.open(tar_file, 'w') as tar: + info = tarfile.TarInfo(name='truncated.txt') + info.size = 1000 + info.type = tarfile.REGTYPE + content = b'x' * 1000 + fileobj = io.BytesIO(content) + tar.addfile(info, fileobj) + + data = tar_file.read_bytes() + tar_file.write_bytes(data[:512 + 500]) + + extract_dir = tmp_path / "extracted" + extractor = SafeTarExtractor() + + with pytest.raises(UnsafeTarError): + extractor.extract(tar_file, extract_dir) + def test_size_limits(self, tmp_path): """Test file size limits.""" tar_file = tmp_path / "large.tar"