clan/data-mesher
8
2
Fork 2
Code Issues 11 Pull Requests 2 Actions Packages Projects Releases 1 Wiki Activity
Files
main
data-mesher/nix/nixosModules/data-mesher/tests/basic.nix
T
pinpox 3d6f88aeb0 nix/tests: cover per-file TTL in the basic NixOS VM test 
Adds three TTL-specific test cases to the basic data-mesher VM test,
plus the two new settings options needed to drive them:

  - sweep_interval: how often the background sweeper runs.
  - clock_skew_tolerance: gossip ingress slack for already-expired
    remote sigs.

The settings were already available via the freeform module type, but
declaring them explicitly makes them discoverable and lets the test
read them as ordinary options.

Test nodes are configured with sweep_interval=2s so the sweeper runs
fast enough to make TTL behaviour observable in seconds rather than
minutes. This is harmless for the existing non-TTL tests because their
files are published with valid_for=0 and the sweeper never touches
them.

Test 11 — Local TTL expiry:
  Upload with --expires-in 5s, confirm the file appears in the GET
  /files listing while live, then verify both the read-path filter
  (listing) and the sweeper (on-disk file) drop it after expiry.

Test 12 — Cluster-wide independent expiry:
  Upload with --expires-in 12s, wait for gossip propagation to a2 and
  a3, then verify each node independently expires the file. No
  tombstone gossip is required — expiry is decided locally from
  signed metadata. The log on every node shows
  "swept expired files removed=1" within the sweep interval.

Test 13 — Heartbeat refresh:
  Upload with --expires-in 6s, sleep 3s, re-upload with the same TTL
  (the heartbeat pattern dm-wireguard-star peers will use). Verify
  the file is still alive 5s after the second upload — past when the
  first TTL would have fired (3 + 5 > 6) but inside the refreshed
  window. Then verify it expires and gets swept after the refreshed
  TTL elapses.

Adds pkgs.curl to the test nodes so the test script can hit GET
/files via curl to validate the read-path filter independently of
the sweeper.
2026-04-10 15:26:31 +01:00

436 lines
17 KiB
Nix
Raw Permalink Blame History

{
lib,
pkgs,
dmModule,
testers,
}:
let
urlKey = {
a1 = lib.trim (builtins.readFile ./keys/network_a/a1.pub.url_encoded);
a2 = lib.trim (builtins.readFile ./keys/network_a/a2.pub.url_encoded);
a3 = lib.trim (builtins.readFile ./keys/network_a/a3.pub.url_encoded);
};
in
testers.nixosTest {
name = "data-mesher-basic";
nodes =
let
peerID = {
a1 = builtins.readFile ./keys/network_a/a1.peerID;
a2 = builtins.readFile ./keys/network_a/a2.peerID;
};
mkNode =
{ name, bootstrapPeers }:
{
imports = [ dmModule ];
config = {
# curl is used by the TTL tests below to query GET /files
environment.systemPackages = [ pkgs.curl ];
system.activationScripts.identity_key = {
text = ''
# Copy network public key
cp ${./keys/network_a/network.pub} /run/network_a.pub
chown data-mesher:data-mesher /run/network_a.pub
chmod 600 /run/network_a.pub
# Copy peer identity key
cp ${./keys/network_a/${name}.key} /run/identity.key
chown data-mesher:data-mesher /run/identity.key
chmod 600 /run/identity.key
# Copy peer identity certificate signed by the network
cp ${./keys/network_a/${name}.cert} /run/identity.cert
chown data-mesher:data-mesher /run/identity.cert
chmod 600 /run/identity.cert
'';
};
services.data-mesher = {
enable = true;
openFirewall = true;
pluginDirectories = [ "test_plugin" ];
settings = {
log_level = "debug";
# speed up TTL sweeping for the TTL tests at the end of this
# test script. Harmless for the non-TTL tests because their
# files are published with valid_for=0.
sweep_interval = "2s";
cluster = {
port = 7946;
bootstrap_peers = bootstrapPeers;
# speed up for testing
push_pull_interval = "5s";
identity_key = "/run/identity.key";
identity_cert = "/run/identity.cert";
};
network = {
id = "/run/network_a.pub";
files = {
"shared_file" = [
"${./keys/signers/alice.pub}"
"${./keys/signers/bob.pub}"
];
"alice_only" = [
"${./keys/signers/alice.pub}"
];
};
namespaces = [ "ns" ];
};
};
};
# A minimal plugin service that verifies BindReadOnlyPaths access to a data-mesher plugin directory.
systemd.services.test-plugin = {
after = [ "data-mesher.service" ];
requires = [ "data-mesher.service" ];
wantedBy = [ "multi-user.target" ];
serviceConfig = {
Type = "oneshot";
RemainAfterExit = true;
BindReadOnlyPaths = [
"/var/lib/data-mesher/files/home/test_plugin"
];
ExecStart = "${pkgs.bash}/bin/bash -c 'test -d /var/lib/data-mesher/files/home/test_plugin && echo ok > /tmp/plugin-check'";
};
};
# override just for testing
# when booting the interface we bind to is not ready, causing an error
systemd.services.data-mesher.serviceConfig.RestartSec = lib.mkForce "1s";
};
};
bootstrapPeers = [
"/dns/a1/tcp/7946/p2p/${peerID.a1}"
"/dns/a2/tcp/7946/p2p/${peerID.a2}"
];
in
{
a1 = mkNode {
name = "a1";
inherit bootstrapPeers;
};
a2 = mkNode {
name = "a2";
inherit bootstrapPeers;
};
# a3 only knows about a2 so this will test DHT propagation of peers
a3 = mkNode {
name = "a3";
bootstrapPeers = [ "/dns/a2/tcp/7946/p2p/${peerID.a2}" ];
};
};
testScript = ''
# The default network uses the name "home" for file storage
NETWORK_NAME = "home"
# some helper functions
def file_path(filename):
"""Return the on-disk path for a file, accounting for per-network storage."""
return f"/var/lib/data-mesher/files/{NETWORK_NAME}/{filename}"
def upload_file(node, filename, content, key_path="${./keys/signers/alice.key}"):
"""Create a file and upload it via the cli"""
node.succeed(f"echo -n '{content}' > /tmp/{filename}")
node.succeed(f"data-mesher file update /tmp/{filename} --url http://[::1]:7331 --key {key_path} --network-id ${./keys/network_a/network.pub}")
def get_file_content(node, filename):
"""Get file content from the state directory"""
return node.succeed(f"cat {file_path(filename)}").strip()
def wait_for_file(node, filename, expected_content, timeout=30):
"""Wait until a file exists with the expected content"""
fp = file_path(filename)
node.wait_until_succeeds(
f"test -f {fp} && "
f"grep -q '{expected_content}' {fp}",
timeout
)
def file_exists(node, filename):
"""Check if a file exists in the state directory"""
fp = file_path(filename)
return node.succeed(f"test -f {fp} && echo yes || echo no").strip() == "yes"
def delete_file(node, filename, key_path="${./keys/signers/alice.key}"):
"""Delete a file via the cli (creates a tombstone)"""
node.succeed(f"data-mesher file delete {filename} --url http://[::1]:7331 --key {key_path} --network-id ${./keys/network_a/network.pub}")
def wait_for_file_deleted(node, filename, timeout=30):
"""Wait until a file no longer exists"""
fp = file_path(filename)
node.wait_until_succeeds(
f"test ! -f {fp}",
timeout
)
# ========================================
# Test 0: Plugin directory creation
# ========================================
a1.wait_for_unit("data-mesher.service")
# Verify the plugin directory was created by data-mesher
a1.succeed("test -d /var/lib/data-mesher/files/home/test_plugin")
# Verify the test-plugin service started and could access the directory via BindReadOnlyPaths
a1.wait_for_unit("test-plugin.service")
result = a1.succeed("cat /tmp/plugin-check").strip()
assert result == "ok", f"Expected 'ok' but got '{result}'"
print("Test 0 PASSED: Plugin directory creation and BindReadOnlyPaths access")
# ========================================
# Test 1: Basic upload and local retrieval
# ========================================
# Upload a file to a1
upload_file(a1, "shared_file", "version1_from_a1")
# Verify file exists locally
wait_for_file(a1, "shared_file", "version1_from_a1")
print("Test 1 PASSED: Basic upload and local retrieval")
# ========================================
# Test 2: File propagation to cluster
# ========================================
a2.wait_for_unit("data-mesher.service")
a3.wait_for_unit("data-mesher.service")
# Wait for the file to propagate to a2 and a3
wait_for_file(a2, "shared_file", "version1_from_a1")
wait_for_file(a3, "shared_file", "version1_from_a1")
print("Test 2 PASSED: File propagation to cluster")
# ========================================
# Test 3: Conflict resolution (newer wins)
# ========================================
# Upload a newer version from a2 (a2 is authorized for shared_file)
upload_file(a2, "shared_file", "version2_from_a2")
# All nodes should eventually have the newer version
wait_for_file(a1, "shared_file", "version2_from_a2")
wait_for_file(a2, "shared_file", "version2_from_a2")
wait_for_file(a3, "shared_file", "version2_from_a2")
print("Test 3 PASSED: Conflict resolution (newer timestamp wins)")
# ========================================
# Test 4: Node restart recovery
# ========================================
a2.shutdown()
# Upload a new file while a2 is down
upload_file(a1, "alice_only", "created_while_a2_down")
# Verify a1 and a3 have it
wait_for_file(a1, "alice_only", "created_while_a2_down")
wait_for_file(a3, "alice_only", "created_while_a2_down")
# Restart a2
a2.start()
a2.wait_for_unit("data-mesher.service")
# a2 should receive the file after rejoining
wait_for_file(a2, "alice_only", "created_while_a2_down")
print("Test 4 PASSED: Node restart recovery")
# ========================================
# Test 5: Unauthorized signer rejection
# ========================================
# a3 is NOT authorized to sign alice_only, so this should fail
a3.succeed("echo -n 'unauthorized' > /tmp/alice_only")
a3.fail("data-mesher file update /tmp/alice_only --url http://[::1]:7331 --key ${./keys/signers/bob.key} --network-id ${./keys/network_a/network.pub}")
print("Test 5 PASSED: Unauthorized signer rejection")
# ========================================
# Test 6: Tombstone propagation
# ========================================
# Verify shared_file exists on all nodes
for node in [a1, a2, a3]:
assert file_exists(node, "shared_file"), "shared_file should exist before deletion"
# a1 deletes the shared file (creates tombstone)
delete_file(a1, "shared_file")
# Verify file is deleted on a1
wait_for_file_deleted(a1, "shared_file")
# Verify tombstone propagates to a2 and a3
wait_for_file_deleted(a2, "shared_file")
wait_for_file_deleted(a3, "shared_file")
print("Test 6 PASSED: Tombstone propagation")
# ========================================
# Test 7: Network key is implicit signer
# ========================================
# The network key should be able to update any configured file, even alice_only
# which only lists alice as an explicit signer
upload_file(a1, "alice_only", "network_key_update", key_path="${./keys/network_a/network.key}")
wait_for_file(a1, "alice_only", "network_key_update")
upload_file(a1, "shared_file", "network_key_shared", key_path="${./keys/network_a/network.key}")
wait_for_file(a1, "shared_file", "network_key_shared")
print("Test 7 PASSED: Network key is implicit signer for all files")
# ========================================
# Test 8: Namespace upload and propagation
# ========================================
# Upload a namespace file from a1 using its identity key and certificate
# The namespace path is ns/{a1_url_encoded_public_key}
ns_filename = "ns/${urlKey.a1}"
a1.succeed("echo -n 'namespace_data_from_a1' > /tmp/ns_file")
a1.succeed(f"data-mesher file update /tmp/ns_file --name '{ns_filename}' --url http://[::1]:7331 --key /run/identity.key --network-id ${./keys/network_a/network.pub} --cert /run/identity.cert")
# Verify the file exists locally on a1
wait_for_file(a1, ns_filename, "namespace_data_from_a1")
# Verify the file propagates to a2 and a3
wait_for_file(a2, ns_filename, "namespace_data_from_a1")
wait_for_file(a3, ns_filename, "namespace_data_from_a1")
print("Test 8 PASSED: Namespace upload and propagation")
# ========================================
# Test 9: Namespace rejection without cert
# ========================================
# Uploading a namespace file without a certificate should fail
ns_filename_alice = "ns/${urlKey.a2}"
a2.succeed("echo -n 'no_cert' > /tmp/ns_nocert")
a2.fail(f"data-mesher file update /tmp/ns_nocert --name '{ns_filename_alice}' --url http://[::1]:7331 --key /run/identity.key --network-id ${./keys/network_a/network.pub}")
print("Test 9 PASSED: Namespace rejection without certificate")
# ========================================
# Test 10: Namespace rejection for unconfigured namespace
# ========================================
# Uploading to a namespace not in the configured list should fail,
# even with a valid certificate
bad_ns_filename = "badns/${urlKey.a1}"
a1.succeed("echo -n 'bad_namespace' > /tmp/ns_badns")
a1.fail(f"data-mesher file update /tmp/ns_badns --name '{bad_ns_filename}' --url http://[::1]:7331 --key /run/identity.key --network-id ${./keys/network_a/network.pub} --cert /run/identity.cert")
print("Test 10 PASSED: Namespace rejection for unconfigured namespace")
# ========================================
# TTL tests — exercise per-file ValidFor expiry
# ========================================
#
# Nodes are configured with sweep_interval=2s above, so the background
# sweeper runs frequently enough to make these tests fast. The clock
# skew tolerance is the default 5m, so gossip ingress accepts files
# whose effective expiry is up to 5 minutes in the past — easily
# generous enough for our short TTLs and ~5s push/pull interval.
def upload_file_with_ttl(node, filename, content, ttl, key_path="${./keys/signers/alice.key}"):
"""Create a file and upload it via the cli with a --expires-in TTL."""
node.succeed(f"echo -n '{content}' > /tmp/{filename}")
node.succeed(
f"data-mesher file update /tmp/{filename} "
f"--url http://[::1]:7331 "
f"--key {key_path} "
f"--network-id ${./keys/network_a/network.pub} "
f"--expires-in {ttl}"
)
def list_files(node):
"""Fetch the GET /files NDJSON listing as a single string."""
return node.succeed("curl -sS http://[::1]:7331/files")
# ========================================
# Test 11: Local TTL expiry (read filter + sweeper)
# ========================================
# Upload with a 5s TTL. The file should be readable immediately, then
# disappear from the listing once expired (read-path filter), and then
# be removed from disk by the sweeper shortly after.
upload_file_with_ttl(a1, "shared_file", "ttl_local_test", "5s")
wait_for_file(a1, "shared_file", "ttl_local_test")
# Confirm it shows up in the HTTP listing while still live.
listing = list_files(a1)
assert "shared_file" in listing, "shared_file should be in the GET /files listing while live"
# Wait long enough for the sweeper to run and remove the on-disk file.
# Effective expiry = SignedAt + 5s; sweeper runs every 2s; allow margin.
wait_for_file_deleted(a1, "shared_file", timeout=15)
# And confirm the listing endpoint also no longer reports it.
a1.wait_until_succeeds(
"curl -sS http://[::1]:7331/files | grep -v shared_file > /dev/null || true; "
"! curl -sS http://[::1]:7331/files | grep -q shared_file",
timeout=10,
)
print("Test 11 PASSED: Local TTL expiry removes file from disk and listing")
# ========================================
# Test 12: Cluster-wide independent expiry
# ========================================
# Upload from a1 with a TTL long enough for gossip to propagate before
# expiry kicks in (push_pull_interval=5s). Then verify each node
# independently expires the file — no tombstone gossip required.
upload_file_with_ttl(a1, "shared_file", "ttl_cluster_test", "12s")
# Wait for propagation to a2 and a3 while the file is still alive.
wait_for_file(a1, "shared_file", "ttl_cluster_test")
wait_for_file(a2, "shared_file", "ttl_cluster_test")
wait_for_file(a3, "shared_file", "ttl_cluster_test")
# Wait until every node has independently dropped the file.
# 12s expiry + 2s sweeper interval + slack
for node in [a1, a2, a3]:
wait_for_file_deleted(node, "shared_file", timeout=25)
print("Test 12 PASSED: Cluster-wide TTL expiry without tombstones")
# ========================================
# Test 13: Heartbeat refresh resets the TTL window
# ========================================
# Upload with a 6s TTL, then re-upload (heartbeat) before it expires.
# The second upload has a fresh SignedAt, so the file should remain
# alive past when the first TTL would have fired.
upload_file_with_ttl(a1, "shared_file", "ttl_heartbeat_v1", "6s")
wait_for_file(a1, "shared_file", "ttl_heartbeat_v1")
# Sleep for half the TTL, then refresh.
a1.succeed("sleep 3")
upload_file_with_ttl(a1, "shared_file", "ttl_heartbeat_v2", "6s")
wait_for_file(a1, "shared_file", "ttl_heartbeat_v2")
# Sleep for 5s — past the original TTL of 6s (3 + 5 = 8 > 6), but still
# within the refreshed TTL (5 < 6).
a1.succeed("sleep 5")
assert file_exists(a1, "shared_file"), \
"heartbeat-refreshed file should still exist past its original expiry"
# Now wait for the refreshed TTL to fire and the sweeper to clean up.
wait_for_file_deleted(a1, "shared_file", timeout=15)
print("Test 13 PASSED: Heartbeat refresh extends TTL window")
print("All tests PASSED!")
'';
}
Reference in New Issue View Git Blame Copy Permalink