Prevent stale, truncated, or changing-file copies (#10)

* Prevent stale, truncated, or changing-file copies

- copy_file() now loops until every byte is actually written (os.write
  may write fewer bytes than given) and compares fstat() on the
  source's own open descriptor before/after the read loop - this
  reflects the underlying inode directly, catching a source file
  truncated, grown, or overwritten in place while being read, and isn't
  fooled by the source path being renamed away mid-copy. Raises
  BackupCreationError (already a fatal, staging-discarding condition in
  initiate_backup) on a mismatch.
- copy_file() returns a sha256 hex digest of the copied content;
  copy_direntry()/update_direntry() propagate it; rsync() carries it
  through the previously-unused msg field for CREATE/REWRITE actions.
- initiate_backup() collects those digests into a checksums dict keyed
  by path relative to the snapshot root, recorded in the completion
  manifest (already JSON since the topology-validation work).
- Added verify_snapshot(): re-hashes every file with a recorded
  checksum and reports any mismatch, satisfying the release gate's
  "completed snapshot passes manifest verification" requirement.
- Switched every mtime comparison and os.utime() restore (rsync's
  file/directory sync, hardlink_dir's directory metadata, the
  hardlink-breaking copy-on-write step) from float seconds to
  nanoseconds, so a same-second modification can't be missed.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

* Address Copilot review: hash checksums for external rsync too, fix log/error text

- initiate_backup() now hashes CREATE/REWRITE files itself when msg is
  empty (fs.rsync_ext(), the external-rsync path, never yields a
  digest). Without this, verify_snapshot() silently became a no-op for
  every snapshot created with external_rsync=True.
- Fixed a pre-existing log message that said "different inodes" for
  the branch whose actual condition is src/dst sharing an inode.
- verify_snapshot()'s error message now also mentions a corrupt
  manifest, not just missing/pre-migration, matching what
  _read_manifest() actually distinguishes.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

---------

Co-authored-by: Maxim Shulga <maxim.shulga@alludo.com>
Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>

IMPROVEMENTS.md

@@ -74,11 +74,11 @@ Acceptance criteria:
 
 ### Prevent stale, truncated, or changing-file copies
 
-- [ ] Handle partial writes correctly or use a proven copy primitive.
+- [x] Handle partial writes correctly or use a proven copy primitive.
-- [ ] Use nanosecond timestamps and a documented integrity strategy.
+- [x] Use nanosecond timestamps and a documented integrity strategy.
-- [ ] Add checksum verification, either always or through an explicit mode.
+- [x] Add checksum verification, either always or through an explicit mode.
-- [ ] Stat files before and after copying and retry or fail if they changed.
+- [x] Stat files before and after copying and retry or fail if they changed.
-- [ ] Record integrity information in the completion manifest.
+- [x] Record integrity information in the completion manifest.
 
 Acceptance criteria:
 

curateipsum/backup.py

@@ -1,6 +1,7 @@
 """
 Module with backup functions.
 """
+import hashlib
 import json
 import logging
 import os
@@ -157,11 +158,13 @@ def _ensure_repo_id(backups_dir: str) -> str:
     return repo_id
 
 
-def _write_manifest(marker_path: str, backups_dir: str, snapshot_name: str):
+def _write_manifest(marker_path: str, backups_dir: str, snapshot_name: str,
+                    checksums: Optional[Dict[str, str]] = None):
     """Write and fsync the completion manifest for a finished snapshot."""
     manifest = json.dumps({
         "repo_id": _ensure_repo_id(backups_dir),
         "snapshot": snapshot_name,
+        "checksums": checksums or {},
     })
     _write_small_file_no_follow(marker_path, manifest)
 
@@ -192,6 +195,49 @@ def _read_manifest(marker_path: str) -> Union[dict, object, None]:
         return None
 
 
+def _sha256_file(path: str) -> str:
+    """sha256 hex digest of a file's current on-disk content."""
+    digest = hashlib.sha256()
+    with open(path, "rb") as f:
+        for chunk in iter(lambda: f.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def verify_snapshot(backup_entry: Union[os.DirEntry, fs.PseudoDirEntry]
+                    ) -> List[str]:
+    """
+    Verify a completed snapshot against the checksums recorded in its
+    manifest at backup time. Only files freshly copied or rewritten
+    during that run have a recorded checksum - files inherited unchanged
+    via hardlink from an earlier snapshot are not re-hashed here, since
+    they're covered by that earlier snapshot's own verification.
+
+    :return: relative paths whose current content no longer matches its
+        recorded checksum, or that are missing. Empty means verified OK.
+    :raises BackupFailedError: if the manifest itself is missing, legacy,
+        or corrupt - there's nothing to verify against.
+    """
+    manifest = _read_manifest(_get_backup_marker(backup_entry).path)
+    if manifest is None or manifest is _LEGACY_MANIFEST:
+        raise BackupFailedError(
+            "Cannot verify %s: missing, corrupt, or pre-migration manifest"
+            % backup_entry.name
+        )
+
+    mismatches = []
+    for relpath, expected_digest in manifest.get("checksums", {}).items():
+        full_path = os.path.join(backup_entry.path, relpath)
+        try:
+            actual_digest = _sha256_file(full_path)
+        except OSError:
+            mismatches.append(relpath)
+            continue
+        if actual_digest != expected_digest:
+            mismatches.append(relpath)
+    return mismatches
+
+
 def _verify_safe_to_delete(entry: os.DirEntry, backups_dir_abs: str,
                            repo_id: str) -> bool:
     """
@@ -752,6 +798,10 @@ def initiate_backup(sources,
     rsync_func = fs.rsync_ext if external_rsync else fs.rsync
 
     backup_changed = False
+    # sha256 digest of every freshly-copied-or-rewritten file this run,
+    # keyed by its path relative to the snapshot root. Recorded in the
+    # manifest so a completed snapshot's integrity can be verified later.
+    checksums: Dict[str, str] = {}
     try:
         for src in sources:
             src_abs = os.path.abspath(src)
@@ -766,11 +816,26 @@ def initiate_backup(sources,
                     raise BackupFailedError(
                         "Failed to copy %s: %s" % (entry_relpath, msg)
                     )
+                full_relpath = os.path.join(src_name, entry_relpath)
+                if action in (fs.Actions.CREATE, fs.Actions.REWRITE):
+                    if msg:
+                        # fs.rsync() already hashed the file while
+                        # copying it.
+                        checksums[full_relpath] = msg
+                    else:
+                        # fs.rsync_ext() (external rsync) never returns a
+                        # digest - hash the copied file directly so
+                        # verify_snapshot() isn't silently a no-op when
+                        # external_rsync=True.
+                        full_path = os.path.join(staging.path, full_relpath)
+                        if (os.path.isfile(full_path)
+                                and not os.path.islink(full_path)):
+                            checksums[full_relpath] = _sha256_file(full_path)
                 # TODO maybe should be run if first backup too?
                 if latest_backup is not None:
                     process_backed_entry(
                         backup_dir=staging.path,
-                        entry_relpath=os.path.join(src_name, entry_relpath),
+                        entry_relpath=full_relpath,
                         action=action,
                         msg=msg,
                     )
@@ -804,7 +869,7 @@ def initiate_backup(sources,
     # then atomically rename the staging dir into its final snapshot name
     marker_name = "%s_%s" % (BACKUP_MARKER, snapshot_name)
     _write_manifest(os.path.join(staging.path, marker_name),
-                    backups_dir, snapshot_name)
+                    backups_dir, snapshot_name, checksums=checksums)
     _fsync_dir(staging.path)
 
     final_path = os.path.join(backups_dir, snapshot_name)

curateipsum/fs.py

@@ -4,12 +4,13 @@ Module with filesystem-related functions.
 
 import enum
 import glob
+import hashlib
 import logging
 import os
 import subprocess
 import sys
 import uuid
-from typing import Iterable, Tuple, Union
+from typing import Iterable, Optional, Tuple, Union
 
 _lg = logging.getLogger(__name__)
 
@@ -221,16 +222,46 @@ WRITE_FLAGS = os.O_WRONLY | os.O_CREAT | os.O_TRUNC | O_BINARY
 BUFFER_SIZE = 128 * 1024
 
 
-def copy_file(src, dst):
+def _write_all(fd: int, data: bytes):
-    """ Copy file from src to dst. Faster than shutil.copy. """
+    """os.write() may write fewer bytes than given; loop until it's all out."""
+    view = memoryview(data)
+    while view:
+        written = os.write(fd, view)
+        view = view[written:]
+
+
+def copy_file(src, dst) -> str:
+    """
+    Copy file from src to dst. Faster than shutil.copy.
+
+    Compares fstat() on the source's own open file descriptor before and
+    after the read loop - this reflects the underlying inode directly, so
+    it catches the source being truncated, grown, or overwritten in place
+    while being read (and isn't fooled by the source path being renamed
+    or unlinked mid-copy, since the descriptor keeps referring to the
+    same inode regardless). Nanosecond mtime is used so a same-second
+    modification can't slip past undetected.
+
+    :return: sha256 hex digest of the bytes actually copied.
+    :raises BackupCreationError: if the source changed while being read.
+    """
     fin = None
     fout = None
     try:
         fin = os.open(src, READ_FLAGS)
-        fstat = os.fstat(fin)
+        stat_before = os.fstat(fin)
-        fout = os.open(dst, WRITE_FLAGS, fstat.st_mode)
+        fout = os.open(dst, WRITE_FLAGS, stat_before.st_mode)
-        for x in iter(lambda: os.read(fin, BUFFER_SIZE), b""):
+        digest = hashlib.sha256()
-            os.write(fout, x)
+        for chunk in iter(lambda: os.read(fin, BUFFER_SIZE), b""):
+            _write_all(fout, chunk)
+            digest.update(chunk)
+        stat_after = os.fstat(fin)
+        if (stat_after.st_size != stat_before.st_size
+                or stat_after.st_mtime_ns != stat_before.st_mtime_ns):
+            raise BackupCreationError(
+                "Source file changed while being copied: %s" % src
+            )
+        return digest.hexdigest()
     finally:
         if fout is not None:
             try:
@@ -272,9 +303,9 @@ def _break_shared_hardlink(path: str):
             _lg.debug("Cannot preserve ownership while breaking hardlink "
                       "(not root): %s", path)
         os.chmod(tmp_path, st.st_mode)
-        os.utime(tmp_path, (st.st_atime, st.st_mtime))
+        os.utime(tmp_path, ns=(st.st_atime_ns, st.st_mtime_ns))
         os.replace(tmp_path, path)
-    except OSError:
+    except (OSError, BackupCreationError):
         try:
             os.unlink(tmp_path)
         except OSError:
@@ -282,9 +313,15 @@ def _break_shared_hardlink(path: str):
         raise
 
 
-def copy_direntry(entry: Union[os.DirEntry, PseudoDirEntry], dst_path):
+def copy_direntry(entry: Union[os.DirEntry, PseudoDirEntry],
-    """ Non-recursive DirEntry (file/dir/symlink) copy. """
+                  dst_path) -> Optional[str]:
+    """
+    Non-recursive DirEntry (file/dir/symlink) copy.
+    :return: sha256 hex digest of the copied content for a regular file,
+        None for a directory or symlink (nothing to hash).
+    """
     src_stat = entry.stat(follow_symlinks=False)
+    digest = None
     # Classify with lstat semantics and check symlink-ness first: is_dir()
     # (and is_file()) default to follow_symlinks=True, so a symlink
     # pointing at a directory would otherwise be misclassified as a
@@ -298,8 +335,12 @@ def copy_direntry(entry: Union[os.DirEntry, PseudoDirEntry], dst_path):
         os.mkdir(dst_path)
 
     else:
-        copy_file(entry.path, dst_path)
+        digest = copy_file(entry.path, dst_path)
 
+    # Nanosecond timestamps: st_mtime (float seconds) loses sub-second
+    # precision, which could make a real content change on a fast-moving
+    # file look identical to an unrelated file with the same whole-second
+    # mtime once rounded.
     if entry.is_symlink():
         # change symlink attributes only if supported by OS
         if os.chown in os.supports_follow_symlinks:
@@ -311,7 +352,8 @@ def copy_direntry(entry: Union[os.DirEntry, PseudoDirEntry], dst_path):
         if os.chmod in os.supports_follow_symlinks:
             os.chmod(dst_path, src_stat.st_mode, follow_symlinks=False)
         if os.utime in os.supports_follow_symlinks:
-            os.utime(dst_path, (src_stat.st_atime, src_stat.st_mtime),
+            os.utime(dst_path,
+                     ns=(src_stat.st_atime_ns, src_stat.st_mtime_ns),
                      follow_symlinks=False)
     else:
         try:
@@ -319,17 +361,21 @@ def copy_direntry(entry: Union[os.DirEntry, PseudoDirEntry], dst_path):
         except PermissionError:
             _lg.debug("Cannot change ownership (not root): %s", dst_path)
         os.chmod(dst_path, src_stat.st_mode)
-        os.utime(dst_path, (src_stat.st_atime, src_stat.st_mtime))
+        os.utime(dst_path, ns=(src_stat.st_atime_ns, src_stat.st_mtime_ns))
+
+    return digest
 
 
-def update_direntry(src_entry: os.DirEntry, dst_entry: os.DirEntry):
+def update_direntry(src_entry: os.DirEntry,
+                    dst_entry: os.DirEntry) -> Optional[str]:
     """
     Make dst DirEntry (file/dir/symlink) same as src.
     If dst is directory, its content will be removed.
     Src dir content will not be copied into dst dir.
+    :return: see copy_direntry().
     """
     rm_direntry(dst_entry)
-    copy_direntry(src_entry, dst_entry.path)
+    return copy_direntry(src_entry, dst_entry.path)
 
 
 def rsync(src_dir,
@@ -390,9 +436,9 @@ def rsync(src_dir,
                           " (src is a file, dst is not a file): %s",
                           rel_path)
                 try:
-                    update_direntry(src_entry, dst_entry)
+                    digest = update_direntry(src_entry, dst_entry)
-                    yield rel_path, Actions.REWRITE, ""
+                    yield rel_path, Actions.REWRITE, digest or ""
-                except OSError as exc:
+                except (OSError, BackupCreationError) as exc:
                     yield rel_path, Actions.ERROR, str(exc)
                 continue
 
@@ -404,7 +450,7 @@ def rsync(src_dir,
                 try:
                     update_direntry(src_entry, dst_entry)
                     yield rel_path, Actions.REWRITE, ""
-                except OSError as exc:
+                except (OSError, BackupCreationError) as exc:
                     yield rel_path, Actions.ERROR, str(exc)
                 continue
 
@@ -416,17 +462,17 @@ def rsync(src_dir,
                 try:
                     update_direntry(src_entry, dst_entry)
                     yield rel_path, Actions.REWRITE, ""
-                except OSError as exc:
+                except (OSError, BackupCreationError) as exc:
                     yield rel_path, Actions.ERROR, str(exc)
                 continue
 
         # rewrite dst if it is hard link to src (bad for backups)
         if src_entry.inode() == dst_entry.inode():
-            _lg.debug("Rsync, rewriting (different inodes): %s", rel_path)
+            _lg.debug("Rsync, rewriting (shared inode with src): %s", rel_path)
             try:
-                update_direntry(src_entry, dst_entry)
+                digest = update_direntry(src_entry, dst_entry)
-                yield rel_path, Actions.REWRITE, ""
+                yield rel_path, Actions.REWRITE, digest or ""
-            except OSError as exc:
+            except (OSError, BackupCreationError) as exc:
                 yield rel_path, Actions.ERROR, str(exc)
             continue
 
@@ -436,15 +482,15 @@ def rsync(src_dir,
         # rewrite dst file/symlink which have different size or mtime than src
         if src_entry.is_file(follow_symlinks=False):
             same_size = src_stat.st_size == dst_stat.st_size
-            same_mtime = src_stat.st_mtime == dst_stat.st_mtime
+            same_mtime = src_stat.st_mtime_ns == dst_stat.st_mtime_ns
             if not (same_size and same_mtime):
                 reason = "size" if not same_size else "time"
                 _lg.debug("Rsync, rewriting (different %s): %s",
                           reason, rel_path)
                 try:
-                    update_direntry(src_entry, dst_entry)
+                    digest = update_direntry(src_entry, dst_entry)
-                    yield rel_path, Actions.REWRITE, ""
+                    yield rel_path, Actions.REWRITE, digest or ""
-                except OSError as exc:
+                except (OSError, BackupCreationError) as exc:
                     yield rel_path, Actions.ERROR, str(exc)
                 continue
 
@@ -456,7 +502,7 @@ def rsync(src_dir,
                 try:
                     update_direntry(src_entry, dst_entry)
                     yield rel_path, Actions.REWRITE, ""
-                except OSError as exc:
+                except (OSError, BackupCreationError) as exc:
                     yield rel_path, Actions.ERROR, str(exc)
                 continue
 
@@ -500,9 +546,9 @@ def rsync(src_dir,
         dst_path = os.path.join(dst_root_abs, rel_path)
         _lg.debug("Rsync, creating: %s", rel_path)
         try:
-            copy_direntry(src_entry, dst_path)
+            digest = copy_direntry(src_entry, dst_path)
-            yield rel_path, Actions.CREATE, ""
+            yield rel_path, Actions.CREATE, digest or ""
-        except OSError as exc:
+        except (OSError, BackupCreationError) as exc:
             yield rel_path, Actions.ERROR, str(exc)
 
     # restore dir mtimes in dst, updated by updating files
@@ -513,19 +559,19 @@ def rsync(src_dir,
         dst_path = os.path.join(dst_root_abs, rel_path)
         src_stat = src_entry.stat(follow_symlinks=False)
         dst_stat = os.lstat(dst_path)
-        if src_stat.st_mtime != dst_stat.st_mtime:
+        if src_stat.st_mtime_ns != dst_stat.st_mtime_ns:
             _lg.debug("Rsync, restoring directory mtime: %s", dst_path)
             os.utime(dst_path,
-                     (src_stat.st_atime, src_stat.st_mtime),
+                     ns=(src_stat.st_atime_ns, src_stat.st_mtime_ns),
                      follow_symlinks=False)
 
     # restore dst_root dir mtime
     src_root_stat = os.lstat(src_root_abs)
     dst_root_stat = os.lstat(dst_root_abs)
-    if src_root_stat.st_mtime != dst_root_stat.st_mtime:
+    if src_root_stat.st_mtime_ns != dst_root_stat.st_mtime_ns:
         _lg.debug("Rsync, restoring root directory mtime: %s", dst_root_abs)
         os.utime(dst_root_abs,
-                 (src_root_stat.st_atime, src_root_stat.st_mtime),
+                 ns=(src_root_stat.st_atime_ns, src_root_stat.st_mtime_ns),
                  follow_symlinks=False)
 
 
@@ -580,7 +626,8 @@ def _recursive_hardlink(src: str, dst: str) -> bool:
                 except PermissionError:
                     _lg.debug("Cannot change ownership (not root): %s", ent_dst_path)
                 os.chmod(ent_dst_path, ent_stat.st_mode)
-                os.utime(ent_dst_path, (ent_stat.st_atime, ent_stat.st_mtime))
+                os.utime(ent_dst_path,
+                        ns=(ent_stat.st_atime_ns, ent_stat.st_mtime_ns))
 
                 continue
             if ent.is_file(follow_symlinks=False) or ent.is_symlink():

tests/test_fs.py

@@ -164,6 +164,66 @@ class TestCopyFile:
         dst_stat = os.stat(dst_path)
         assert src_stat.st_mode == dst_stat.st_mode
 
+    def test_returns_sha256_of_content(self, tmp_path):
+        """copy_file returns the sha256 hex digest of the copied bytes"""
+        import hashlib
+        src_path = os.path.join(str(tmp_path), "source.txt")
+        dst_path = os.path.join(str(tmp_path), "dest.txt")
+        content = b"some content to hash"
+        with open(src_path, "wb") as f:
+            f.write(content)
+
+        digest = fs.copy_file(src_path, dst_path)
+
+        assert digest == hashlib.sha256(content).hexdigest()
+
+    def test_detects_source_truncated_during_copy(self, tmp_path):
+        """
+        A source file whose size changes between the read loop starting
+        and finishing (detected via fstat on the same descriptor) must
+        raise, not silently produce a truncated/inconsistent copy.
+        """
+        from unittest import mock
+
+        src_path = os.path.join(str(tmp_path), "source.txt")
+        dst_path = os.path.join(str(tmp_path), "dest.txt")
+        with open(src_path, "wb") as f:
+            f.write(b"x" * 1000)
+
+        real_read = os.read
+        call_count = {"n": 0}
+
+        def read_then_truncate(fd, n):
+            call_count["n"] += 1
+            data = real_read(fd, n)
+            if call_count["n"] == 1:
+                os.truncate(src_path, 10)
+            return data
+
+        with mock.patch("os.read", side_effect=read_then_truncate):
+            with pytest.raises(fs.BackupCreationError):
+                fs.copy_file(src_path, dst_path)
+
+    def test_partial_write_is_completed(self, tmp_path):
+        """os.write() may write fewer bytes than given; copy_file must
+        loop until everything requested has actually been written."""
+        from unittest import mock
+
+        src_path = os.path.join(str(tmp_path), "source.txt")
+        dst_path = os.path.join(str(tmp_path), "dest.txt")
+        content = b"y" * 5000
+        with open(src_path, "wb") as f:
+            f.write(content)
+
+        real_write = os.write
+        with mock.patch("os.write",
+                        side_effect=lambda fd, data: real_write(
+                            fd, bytes(data)[:1])):
+            fs.copy_file(src_path, dst_path)
+
+        with open(dst_path, "rb") as f:
+            assert f.read() == content
+
 
 class TestCopyDirEntry:
     """Test suite for copy_direntry function."""

tests/test_integration.py

@@ -973,3 +973,111 @@ class TestErrorRecovery:
         backups = [b for b in os.listdir(str(backups_dir))
                    if not b.startswith(".")]
         assert len(backups) == 1
+
+
+def test_completed_snapshot_passes_manifest_verification(integration_dirs):
+    """A completed snapshot's checksums must verify against its content."""
+    backups_dir, source_dir = integration_dirs
+    (source_dir / "file1.txt").write_text("content1")
+    (source_dir / "file2.txt").write_text("content2")
+
+    snapshot = bk.initiate_backup(
+        sources=[str(source_dir)],
+        backups_dir=str(backups_dir),
+        dry_run=False,
+    )
+
+    assert bk.verify_snapshot(snapshot) == []
+
+
+def test_verify_snapshot_detects_corruption(integration_dirs):
+    """Corrupting a file after backup must be caught by verify_snapshot."""
+    backups_dir, source_dir = integration_dirs
+    (source_dir / "file1.txt").write_text("content1")
+
+    snapshot = bk.initiate_backup(
+        sources=[str(source_dir)],
+        backups_dir=str(backups_dir),
+        dry_run=False,
+    )
+
+    source_name = os.path.basename(str(source_dir))
+    backed_up_file = os.path.join(
+        snapshot.path, source_name, "file1.txt"
+    )
+    with open(backed_up_file, "w") as f:
+        f.write("corrupted")
+
+    mismatches = bk.verify_snapshot(snapshot)
+    assert os.path.join(source_name, "file1.txt") in mismatches
+
+
+def test_source_changed_during_copy_fails_backup(integration_dirs,
+                                                   monkeypatch):
+    """
+    A source file that changes size while being copied must fail the
+    whole backup rather than produce an inconsistent snapshot.
+    """
+    backups_dir, source_dir = integration_dirs
+    src_file = source_dir / "changing.txt"
+    src_file.write_text("x" * 1000)
+
+    real_read = os.read
+    call_count = {"n": 0}
+
+    def read_then_truncate(fd, n):
+        call_count["n"] += 1
+        data = real_read(fd, n)
+        if call_count["n"] == 1:
+            os.truncate(str(src_file), 10)
+        return data
+
+    monkeypatch.setattr(os, "read", read_then_truncate)
+
+    with pytest.raises(bk.BackupFailedError):
+        bk.initiate_backup(
+            sources=[str(source_dir)],
+            backups_dir=str(backups_dir),
+            dry_run=False,
+        )
+
+    # no completed snapshot must exist after the failure
+    backups = [b for b in os.listdir(str(backups_dir))
+              if not b.startswith(".")]
+    assert backups == []
+
+
+@pytest.mark.skipif(not RSYNC_AVAILABLE, reason="rsync not available")
+def test_external_rsync_snapshot_still_gets_checksums(integration_dirs):
+    """
+    fs.rsync_ext() (external rsync) never yields a digest in msg, unlike
+    fs.rsync() - initiate_backup() must hash the copied files itself in
+    that case, or verify_snapshot() silently becomes a no-op whenever
+    --external-rsync is used.
+    """
+    backups_dir, source_dir = integration_dirs
+    (source_dir / "file1.txt").write_text("content1")
+
+    # Establish baseline with Python rsync first, same as
+    # test_external_rsync_creates_backup: a from-scratch external-rsync
+    # run emits a "created directory ..." header line this codebase's
+    # itemize-output parser doesn't handle, which is an unrelated,
+    # pre-existing limitation.
+    bk.initiate_backup(
+        sources=[str(source_dir)],
+        backups_dir=str(backups_dir),
+        dry_run=False,
+    )
+    time.sleep(1.1)
+    (source_dir / "file2.txt").write_text("content2")
+
+    snapshot = bk.initiate_backup(
+        sources=[str(source_dir)],
+        backups_dir=str(backups_dir),
+        dry_run=False,
+        external_rsync=True,
+    )
+
+    manifest = bk._read_manifest(bk._get_backup_marker(snapshot).path)
+    assert manifest["checksums"]
+    assert bk.verify_snapshot(snapshot) == []