test(snapshot): verify network data integrity after cross-kernel restore

Add check_network_data_integrity helper that generates random bytes on
the host, pushes them to the guest via SSH command-line (base64-encoded
to survive argv), has the guest decode and sha256 them, and asserts the
guest-side hash matches the host-side hash. This exercises the full
virtio-net RX path end-to-end beyond simple connectivity checks.

Signed-off-by: Jack Thomson <jackabt@amazon.com>

Author: JackThomson2

tests/framework/utils.py

@@ -2,7 +2,9 @@
 # SPDX-License-Identifier: Apache-2.0
 """Generic utility functions that are used in the framework."""
 
+import base64
 import errno
+import hashlib
 import json
 import logging
 import os
@@ -578,6 +580,19 @@ def check_entropy(ssh_connection):
     ssh_connection.check_output("dd if=/dev/hwrng of=/dev/null bs=4096 count=1")
 
 
+def check_network_data_integrity(ssh_connection, size_bytes=64 * 1024):
+    """Push random bytes to the guest over SSH and verify the guest-side sha256
+    matches the host-side hash. Exercises the virtio-net RX path end-to-end."""
+    payload = os.urandom(size_bytes)
+    host_hash = hashlib.sha256(payload).hexdigest()
+    b64 = base64.b64encode(payload).decode("ascii")
+    _, stdout, _ = ssh_connection.check_output(f"echo {b64} | base64 -d | sha256sum")
+    guest_hash = stdout.strip().split()[0]
+    assert (
+        guest_hash == host_hash
+    ), f"Guest hash {guest_hash} does not match host hash {host_hash}"
+
+
 @retry(wait=wait_fixed(0.5), stop=stop_after_attempt(5), reraise=True)
 def wait_process_running(process):
     """Wait for a process to run.

tests/integration_tests/functional/test_snapshot_restore_cross_kernel.py

@@ -13,6 +13,7 @@
 from framework.defs import FC_WORKSPACE_DIR
 from framework.utils import (
     check_entropy,
+    check_network_data_integrity,
     generate_mmds_get_request,
     generate_mmds_session_token,
     guest_run_fio_iteration,
@@ -129,6 +130,9 @@ def test_snap_restore_from_artifacts(
         logger.info("Testing net device %s...", iface["iface"].dev_name)
         vm.ssh_iface(idx).check_output("true")
 
+    logger.info("Testing network data integrity...")
+    check_network_data_integrity(vm.ssh)
+
     logger.info("Testing data store behavior...")
     _test_mmds(vm, vm.iface["eth3"]["iface"])