d012d95b07Keep `p2p status` text output machine-parseable by removing ANSI severity colorization, and stop connect/disconnect/forget from throwing on malformed daemon responses (safe .value defaults, uint64 counters). Bound handshake and delta timeouts to kP2pMaxOperationTimeout up front, and ceiling direct sync_interval_ms to the 5-minute reconnect bound so invalid config fails early instead of aborting daemon startup. Degrade memory-sync status to peerCount=0 on a transient peer-registry read instead of failing the whole status IPC. Add an explicit `p2p forget` operation (proto FORGET=9 through dispatcher, manager, and registry) so operators can rotate a peer's pinned key while preserving fail-closed enrollment. Signed-off-by: trvon <git@trevon.dev>
262 lines
11 KiB
C++
262 lines
11 KiB
C++
// SPDX-License-Identifier: GPL-3.0-or-later
|
|
// Copyright 2026 YAMS Contributors
|
|
|
|
// pi-lens-ignore: fatal error
|
|
#include <catch2/catch_test_macros.hpp>
|
|
|
|
#include <yams/compat/thread_stop_compat.h>
|
|
#include <yams/daemon/p2p/peer_registry.h>
|
|
|
|
#include <atomic>
|
|
#include <chrono>
|
|
#include <filesystem>
|
|
#include <latch>
|
|
#include <memory>
|
|
#include <string>
|
|
#include <vector>
|
|
|
|
using yams::daemon::p2p::PeerRegistry;
|
|
|
|
namespace {
|
|
|
|
struct TempDatabase {
|
|
explicit TempDatabase(std::string name) {
|
|
directory = std::filesystem::temp_directory_path() /
|
|
("yams-peer-registry-" + std::move(name) + "-" +
|
|
std::to_string(std::chrono::steady_clock::now().time_since_epoch().count()));
|
|
std::filesystem::create_directories(directory);
|
|
path = directory / "metadata.db";
|
|
}
|
|
~TempDatabase() {
|
|
std::error_code error;
|
|
std::filesystem::remove_all(directory, error);
|
|
}
|
|
std::filesystem::path directory;
|
|
std::filesystem::path path;
|
|
};
|
|
|
|
} // namespace
|
|
|
|
TEST_CASE("P2P peer registry migrates legacy trust schema", "[daemon][p2p][registry][sqlite]") {
|
|
TempDatabase database{"migration"};
|
|
{
|
|
yams::metadata::Database legacy;
|
|
REQUIRE(legacy.open(database.path.string(), yams::metadata::ConnectionMode::Create)
|
|
.has_value());
|
|
REQUIRE(legacy
|
|
.execute(R"sql(
|
|
CREATE TABLE p2p_peers (
|
|
node_id TEXT PRIMARY KEY NOT NULL,
|
|
spki_hash TEXT NOT NULL,
|
|
corpus_id TEXT NOT NULL DEFAULT '',
|
|
corpus_epoch INTEGER NOT NULL DEFAULT 0,
|
|
last_seen_vv TEXT NOT NULL DEFAULT '{"counters_":{}}',
|
|
last_connected_ms INTEGER NOT NULL DEFAULT 0,
|
|
pinned_by_operator INTEGER NOT NULL DEFAULT 0
|
|
)
|
|
)sql")
|
|
.has_value());
|
|
auto insert =
|
|
legacy.prepare("INSERT INTO p2p_peers(node_id, spki_hash, pinned_by_operator) "
|
|
"VALUES('legacy-peer', ?, 1)");
|
|
REQUIRE(insert.has_value());
|
|
auto statement = std::move(insert.value());
|
|
REQUIRE(statement.bind(1, std::string(64, 'a')).has_value());
|
|
REQUIRE(statement.execute().has_value());
|
|
}
|
|
|
|
auto opened = PeerRegistry::open(database.path, 8);
|
|
REQUIRE(opened.has_value());
|
|
auto peers = opened.value()->listPeers();
|
|
REQUIRE(peers.has_value());
|
|
REQUIRE(peers.value().size() == 1);
|
|
CHECK(peers.value().front().endpoint.empty());
|
|
CHECK_FALSE(peers.value().front().remembered);
|
|
CHECK_FALSE(peers.value().front().pinnedByOperator);
|
|
auto legacyTrust = opened.value()->verifyOrPin("legacy-peer", std::string(64, 'a'), false);
|
|
REQUIRE(legacyTrust.has_value());
|
|
CHECK_FALSE(legacyTrust.value().firstContactPinned);
|
|
CHECK_FALSE(legacyTrust.value().pinnedByOperator);
|
|
REQUIRE(opened.value()->enrollOperatorPeer("legacy-peer", std::string(64, 'a')).has_value());
|
|
auto reenrolled = opened.value()->verifyOrPin("legacy-peer", std::string(64, 'a'), false);
|
|
REQUIRE(reenrolled.has_value());
|
|
CHECK(reenrolled.value().pinnedByOperator);
|
|
yams::memory_sync::VersionVector version;
|
|
REQUIRE(opened.value()
|
|
->updatePeerState("legacy-peer", "corpus", 1, version, 7, "host:9721", true)
|
|
.has_value());
|
|
}
|
|
|
|
TEST_CASE("P2P peer registry persists operator enrollment",
|
|
"[daemon][p2p][registry][sqlite][security]") {
|
|
TempDatabase database{"operator-enrollment"};
|
|
auto opened = PeerRegistry::open(database.path, 1);
|
|
REQUIRE(opened.has_value());
|
|
auto& registry = *opened.value();
|
|
const std::string pin(64, 'a');
|
|
|
|
REQUIRE(registry.enrollOperatorPeer("peer-node", std::string(64, 'A')).has_value());
|
|
REQUIRE(registry.enrollOperatorPeer("peer-node", pin).has_value());
|
|
auto replacement = registry.enrollOperatorPeer("peer-node", std::string(64, 'b'));
|
|
REQUIRE_FALSE(replacement.has_value());
|
|
CHECK(replacement.error().code == yams::ErrorCode::Unauthorized);
|
|
auto full = registry.enrollOperatorPeer("other-node", std::string(64, 'c'));
|
|
REQUIRE_FALSE(full.has_value());
|
|
CHECK(full.error().code == yams::ErrorCode::ResourceExhausted);
|
|
|
|
auto peers = registry.listPeers();
|
|
REQUIRE(peers.has_value());
|
|
REQUIRE(peers.value().size() == 1);
|
|
CHECK(peers.value().front().nodeId == "peer-node");
|
|
CHECK(peers.value().front().spkiPin == pin);
|
|
CHECK(peers.value().front().pinnedByOperator);
|
|
CHECK_FALSE(peers.value().front().remembered);
|
|
auto trusted = registry.verifyOrPin("peer-node", pin, false);
|
|
REQUIRE(trusted.has_value());
|
|
CHECK(trusted.value().pinnedByOperator);
|
|
}
|
|
|
|
TEST_CASE("P2P peer registry supports explicit key rotation via removal",
|
|
"[daemon][p2p][registry][sqlite][security]") {
|
|
TempDatabase database{"key-rotation"};
|
|
auto opened = PeerRegistry::open(database.path, 1);
|
|
REQUIRE(opened.has_value());
|
|
auto& registry = *opened.value();
|
|
const std::string firstPin(64, 'a');
|
|
const std::string rotatedPin(64, 'b');
|
|
|
|
REQUIRE(registry.enrollOperatorPeer("peer-node", firstPin).has_value());
|
|
auto replace = registry.enrollOperatorPeer("peer-node", rotatedPin);
|
|
REQUIRE_FALSE(replace.has_value());
|
|
CHECK(replace.error().code == yams::ErrorCode::Unauthorized);
|
|
|
|
REQUIRE(registry.removePeer("peer-node").has_value());
|
|
REQUIRE(registry.enrollOperatorPeer("peer-node", rotatedPin).has_value());
|
|
|
|
auto peers = registry.listPeers();
|
|
REQUIRE(peers.has_value());
|
|
REQUIRE(peers.value().size() == 1);
|
|
CHECK(peers.value().front().spkiPin == rotatedPin);
|
|
CHECK(peers.value().front().pinnedByOperator);
|
|
auto trusted = registry.verifyOrPin("peer-node", rotatedPin, false);
|
|
REQUIRE(trusted.has_value());
|
|
CHECK(trusted.value().pinnedByOperator);
|
|
}
|
|
|
|
TEST_CASE("P2P peer registry persists trust and causal state across restart",
|
|
"[daemon][p2p][registry][sqlite]") {
|
|
TempDatabase database{"restart"};
|
|
const std::string pin(64, 'a');
|
|
yams::memory_sync::VersionVector version;
|
|
REQUIRE(version.increment("peer-node"));
|
|
REQUIRE(version.increment("peer-node"));
|
|
|
|
{
|
|
auto opened = PeerRegistry::open(database.path, 8);
|
|
REQUIRE(opened.has_value());
|
|
auto& registry = *opened.value();
|
|
auto unknown = registry.verifyOrPin("peer-node", pin, false);
|
|
REQUIRE_FALSE(unknown.has_value());
|
|
CHECK(unknown.error().code == yams::ErrorCode::Unauthorized);
|
|
auto first = registry.verifyOrPin("peer-node", std::string(64, 'A'), true);
|
|
REQUIRE(first.has_value());
|
|
CHECK(first.value().firstContactPinned);
|
|
CHECK_FALSE(first.value().pinnedByOperator);
|
|
REQUIRE(registry.enrollOperatorPeer("peer-node", pin).has_value());
|
|
auto promoted = registry.verifyOrPin("peer-node", pin, false);
|
|
REQUIRE(promoted.has_value());
|
|
CHECK(promoted.value().pinnedByOperator);
|
|
auto firstUpdate =
|
|
registry.updatePeerState("peer-node", "corpus", 7, version, 123456, {}, false);
|
|
const std::string firstUpdateError =
|
|
firstUpdate ? std::string{} : firstUpdate.error().message;
|
|
INFO(firstUpdateError);
|
|
REQUIRE(firstUpdate.has_value());
|
|
// A later non-operator update must not downgrade the pin source.
|
|
REQUIRE(registry
|
|
.updatePeerState("peer-node", "corpus", 7, version, 123999,
|
|
"ghost.example:9721", true)
|
|
.has_value());
|
|
}
|
|
|
|
auto reopened = PeerRegistry::open(database.path, 8);
|
|
REQUIRE(reopened.has_value());
|
|
auto& registry = *reopened.value();
|
|
CHECK(registry.pinnedPin("peer-node") == pin);
|
|
auto known = registry.verifyOrPin("peer-node", pin, false);
|
|
REQUIRE(known.has_value());
|
|
CHECK_FALSE(known.value().firstContactPinned);
|
|
CHECK(known.value().pinnedByOperator);
|
|
auto changed = registry.verifyOrPin("peer-node", std::string(64, 'b'), true);
|
|
REQUIRE_FALSE(changed.has_value());
|
|
CHECK(changed.error().code == yams::ErrorCode::Unauthorized);
|
|
|
|
auto peers = registry.listPeers();
|
|
REQUIRE(peers.has_value());
|
|
REQUIRE(peers.value().size() == 1);
|
|
const auto& peer = peers.value().front();
|
|
CHECK(peer.nodeId == "peer-node");
|
|
CHECK(peer.spkiPin == pin);
|
|
CHECK(peer.corpusId == "corpus");
|
|
CHECK(peer.corpusEpoch == 7);
|
|
CHECK(peer.lastSeenVersion.get("peer-node") == 2);
|
|
CHECK(peer.lastConnectedMs == 123999);
|
|
CHECK(peer.endpoint == "ghost.example:9721");
|
|
CHECK(peer.remembered);
|
|
CHECK(peer.pinnedByOperator);
|
|
}
|
|
|
|
TEST_CASE("P2P peer registry enforces capacity and supports removal",
|
|
"[daemon][p2p][registry][sqlite]") {
|
|
TempDatabase database{"capacity"};
|
|
auto opened = PeerRegistry::open(database.path, 2);
|
|
REQUIRE(opened.has_value());
|
|
auto& registry = *opened.value();
|
|
REQUIRE(registry.verifyOrPin("one", std::string(64, '1'), true).has_value());
|
|
REQUIRE(registry.verifyOrPin("two", std::string(64, '2'), true).has_value());
|
|
auto full = registry.verifyOrPin("three", std::string(64, '3'), true);
|
|
REQUIRE_FALSE(full.has_value());
|
|
CHECK(full.error().code == yams::ErrorCode::ResourceExhausted);
|
|
|
|
REQUIRE(registry.removePeer("one").has_value());
|
|
CHECK_FALSE(registry.pinnedPin("one").has_value());
|
|
REQUIRE(registry.verifyOrPin("three", std::string(64, '3'), true).has_value());
|
|
auto peers = registry.listPeers();
|
|
REQUIRE(peers.has_value());
|
|
CHECK(peers.value().size() == 2);
|
|
}
|
|
|
|
TEST_CASE("P2P peer registry serializes concurrent first contact",
|
|
"[daemon][p2p][registry][sqlite][concurrency]") {
|
|
TempDatabase database{"concurrency"};
|
|
auto opened = PeerRegistry::open(database.path, 8);
|
|
REQUIRE(opened.has_value());
|
|
auto& registry = *opened.value();
|
|
const std::string pin(64, 'c');
|
|
constexpr std::size_t kThreads = 8;
|
|
std::latch start(1);
|
|
std::atomic<std::size_t> successes{0};
|
|
std::atomic<std::size_t> firstContacts{0};
|
|
std::vector<yams::compat::jthread> workers;
|
|
workers.reserve(kThreads);
|
|
for (std::size_t index = 0; index < kThreads; ++index) {
|
|
workers.emplace_back([&] {
|
|
start.wait();
|
|
auto result = registry.verifyOrPin("shared-peer", pin, true);
|
|
if (result) {
|
|
successes.fetch_add(1, std::memory_order_relaxed);
|
|
if (result.value().firstContactPinned) {
|
|
firstContacts.fetch_add(1, std::memory_order_relaxed);
|
|
}
|
|
}
|
|
});
|
|
}
|
|
start.count_down();
|
|
workers.clear();
|
|
|
|
CHECK(successes.load(std::memory_order_relaxed) == kThreads);
|
|
CHECK(firstContacts.load(std::memory_order_relaxed) == 1);
|
|
auto peers = registry.listPeers();
|
|
REQUIRE(peers.has_value());
|
|
CHECK(peers.value().size() == 1);
|
|
}
|