// SPDX-License-Identifier: GPL-3.0-or-later // Copyright 2026 YAMS Contributors // pi-lens-ignore: fatal error #include // pi-lens-ignore: fatal error #include #include #include #include "../../../src/daemon/p2p/p2p_fuzz.h" #include #include #include #include #include using namespace std::chrono_literals; using yams::daemon::p2p::InMemoryPeerTrustStore; using yams::daemon::p2p::P2pClientOptions; using yams::daemon::p2p::P2pConnection; using yams::daemon::p2p::P2pListener; using yams::daemon::p2p::PeerHandshakeConfig; using yams::daemon::p2p::PeerHandshakeResult; using yams::daemon::p2p::TlsIdentity; namespace { yams::Result validateHandshakeJson(const nlohmann::json& json) { const auto text = json.dump(); return yams::daemon::p2p::detail::validateHandshakeControlFrame( std::span(reinterpret_cast(text.data()), text.size())); } yams::Result validateHandshakeBytes(std::string_view text) { return yams::daemon::p2p::detail::validateHandshakeControlFrame( std::span(reinterpret_cast(text.data()), text.size())); } TlsIdentity identity(std::string nodeId, const yams::memory_sync::WriterKeyPair& keyPair) { auto result = TlsIdentity::fromPrivateKeyPem(std::move(nodeId), keyPair.privateKeyPem); REQUIRE(result.has_value()); return std::move(result.value()); } PeerHandshakeConfig config(std::string nodeId, std::string corpusId = "shared-corpus", std::uint64_t epoch = 1) { return PeerHandshakeConfig{.nodeId = std::move(nodeId), .corpusId = std::move(corpusId), .corpusEpoch = epoch, .localVersion = {}, .allowFirstContact = true, .timeout = 2s}; } void enableCurrentWriterWindow(PeerHandshakeConfig& config) { const auto counter = config.localVersion.get(config.nodeId); if (counter == 0) { return; } const auto commitment = config.localCommitments.at(config.nodeId); config.resolveLocalWindow = [counter, commitment](std::uint64_t peerCounter, std::size_t maxRecords, std::size_t maxWireBytes) { if (maxWireBytes == 0 || counter <= peerCounter || counter - peerCounter > maxRecords) { return yams::Result{yams::Error{ yams::ErrorCode::ResourceExhausted, "test writer window exceeds bound"}}; } return yams::Result{commitment}; }; } struct PairResult { yams::Result client; yams::Result server; std::string clientPin; std::string serverPin; }; PairResult runHandshake(PeerHandshakeConfig serverConfig, PeerHandshakeConfig clientConfig, InMemoryPeerTrustStore& serverTrust, InMemoryPeerTrustStore& clientTrust, std::string clientCertificateNodeId = {}) { auto serverKey = yams::memory_sync::generateWriterKeyPair(); auto clientKey = yams::memory_sync::generateWriterKeyPair(); REQUIRE(serverKey.has_value()); REQUIRE(clientKey.has_value()); auto serverIdentity = identity(serverConfig.nodeId, serverKey.value()); auto clientIdentity = identity(clientCertificateNodeId.empty() ? clientConfig.nodeId : clientCertificateNodeId, clientKey.value()); const std::string serverPin = serverIdentity.spkiPin(); const std::string clientPin = clientIdentity.spkiPin(); P2pListener listener(P2pListener::Options{.host = "127.0.0.1", .port = 0, .identity = std::move(serverIdentity), .allowedPeerPins = {}, .allowUnpinnedPeers = true, .handshakeTimeout = 2s, .maxConcurrentSessions = 1}); std::promise> serverDone; auto serverFuture = serverDone.get_future(); listener.setHandler([&](P2pConnection channel) { serverDone.set_value( yams::daemon::p2p::acceptPeerHandshake(channel, serverConfig, serverTrust)); }); REQUIRE(listener.start().has_value()); auto channel = yams::daemon::p2p::p2pConnect(P2pClientOptions{.host = "127.0.0.1", .port = listener.boundPort(), .identity = std::move(clientIdentity), .expectedPeerPin = {}, .allowUnpinnedPeer = true, .connectTimeout = 2s, .handshakeTimeout = 2s}); REQUIRE(channel.has_value()); auto clientResult = yams::daemon::p2p::initiatePeerHandshake(channel.value(), clientConfig, clientTrust); REQUIRE(serverFuture.wait_for(3s) == std::future_status::ready); auto serverResult = serverFuture.get(); listener.stop(); return PairResult{.client = std::move(clientResult), .server = std::move(serverResult), .clientPin = clientPin, .serverPin = serverPin}; } } // namespace TEST_CASE("p2p handshake fuzz seam enforces protocol-v4 control bounds", "[daemon][p2p][protocol][fuzz]") { const nlohmann::json hello{{"type", "hello"}, {"protocol", yams::daemon::p2p::kP2pProtocolVersion}, {"schema_version", yams::daemon::p2p::kP2pEnvelopeSchemaVersion}, {"node_id", "node-a"}, {"corpus_id", "corpus"}, {"corpus_epoch", 1}, {"max_writer_advance", 32}, {"max_writer_window_bytes", 4096}}; CHECK(validateHandshakeJson(hello).has_value()); auto legacy = hello; legacy["protocol"] = 3; CHECK_FALSE(validateHandshakeJson(legacy).has_value()); auto oversized = hello; oversized["node_id"] = std::string(yams::daemon::p2p::kMaxP2pIdentityBytes + 1, 'n'); CHECK_FALSE(validateHandshakeJson(oversized).has_value()); CHECK_FALSE(validateHandshakeJson(nlohmann::json{{"type", "unknown"}}).has_value()); } TEST_CASE("p2p handshake control frame rejects excessive JSON nesting depth", "[daemon][p2p][protocol][fuzz]") { // A structurally valid hello whose unknown field is pathologically nested must be rejected by // the depth bound rather than parsed (a pre-auth stack-exhaustion probe). std::string deep = "{\"type\":\"hello\",\"protocol\":4,\"schema_version\":4,\"node_id\":\"node-a\"," "\"corpus_id\":\"corpus\",\"corpus_epoch\":1,\"max_writer_advance\":32," "\"max_writer_window_bytes\":4096,\"nested\":"; deep.append(200, '['); deep.append(200, ']'); deep += '}'; CHECK_FALSE(validateHandshakeBytes(deep).has_value()); } TEST_CASE("p2p control frames admit only non-negative integer numbers", "[daemon][p2p][protocol][fuzz]") { // Found by fuzz_p2p_handshake: a hello whose integer field overflowed to a JSON float reached // get(), an undefined float-to-integer conversion before authentication. const nlohmann::json hello{{"type", "hello"}, {"protocol", yams::daemon::p2p::kP2pProtocolVersion}, {"schema_version", yams::daemon::p2p::kP2pEnvelopeSchemaVersion}, {"node_id", "node-a"}, {"corpus_id", "corpus"}, {"corpus_epoch", 1}, {"max_writer_advance", 32}, {"max_writer_window_bytes", 4096}}; REQUIRE(validateHandshakeJson(hello).has_value()); const auto rejectsRaw = [](std::string text) { return !validateHandshakeBytes(std::move(text)).has_value(); }; CHECK(rejectsRaw("{\"type\":\"hello\",\"protocol\":" "355555555555555555555555555555555555555555555,\"schema_version\":4," "\"node_id\":\"node-a\",\"corpus_id\":\"corpus\",\"corpus_epoch\":1," "\"max_writer_advance\":32,\"max_writer_window_bytes\":4096}")); auto fractional = hello; fractional["corpus_epoch"] = 1.0; CHECK_FALSE(validateHandshakeJson(fractional).has_value()); auto negative = hello; negative["max_writer_advance"] = -1; CHECK_FALSE(validateHandshakeJson(negative).has_value()); auto nestedFloat = hello; nestedFloat["extra"] = nlohmann::json::array({nlohmann::json::array({2.5})}); CHECK_FALSE(validateHandshakeJson(nestedFloat).has_value()); auto wideProtocol = hello; wideProtocol["protocol"] = std::uint64_t{yams::daemon::p2p::kP2pProtocolVersion} + (std::uint64_t{1} << 32); CHECK_FALSE(validateHandshakeJson(wideProtocol).has_value()); } TEST_CASE("P2P handshake freezes a bounded authenticated writer prefix", "[daemon][p2p][handshake][window][commitment]") { auto serverConfig = config("server-node"); serverConfig.localVersion.observe("server-node", 5); serverConfig.localCommitments["server-node"] = yams::memory_sync::WriterHistoryCommitment{.counter = 5, .digest = std::string(64, 'f')}; serverConfig.resolveLocalCommitment = [](std::uint64_t counter) { REQUIRE(counter == 2); return yams::Result{ yams::memory_sync::WriterHistoryCommitment{.counter = counter, .digest = std::string(64, 'a')}}; }; serverConfig.resolveLocalWindow = [](std::uint64_t peerCounter, std::size_t maxRecords, std::size_t maxWireBytes) { REQUIRE(peerCounter == 0); REQUIRE(maxRecords == 2); REQUIRE(maxWireBytes > 0); return yams::Result{ yams::memory_sync::WriterHistoryCommitment{.counter = 2, .digest = std::string(64, 'a')}}; }; serverConfig.maxWriterAdvance = 2; auto clientConfig = config("client-node"); clientConfig.maxWriterAdvance = 2; InMemoryPeerTrustStore serverTrust; InMemoryPeerTrustStore clientTrust; auto result = runHandshake(serverConfig, clientConfig, serverTrust, clientTrust); REQUIRE(result.client.has_value()); REQUIRE(result.server.has_value()); CHECK(result.client.value().peerVersion.get("server-node") == 2); REQUIRE(result.client.value().peerCommitments.contains("server-node")); CHECK(result.client.value().peerCommitments.at("server-node").counter == 2); CHECK(result.client.value().peerCommitments.at("server-node").digest == std::string(64, 'a')); CHECK(result.server.value().localVersion.get("server-node") == 2); } TEST_CASE("P2P peer trust store pins first contact and rejects key changes", "[daemon][p2p][handshake][trust]") { InMemoryPeerTrustStore trust; const std::string firstPin(64, 'a'); const std::string changedPin(64, 'b'); auto unknown = trust.verifyOrPin("peer", firstPin, false); REQUIRE_FALSE(unknown.has_value()); CHECK(unknown.error().code == yams::ErrorCode::Unauthorized); auto first = trust.verifyOrPin("peer", firstPin, true); REQUIRE(first.has_value()); CHECK(first.value().firstContactPinned); CHECK_FALSE(first.value().pinnedByOperator); CHECK(trust.pinnedPin("peer") == firstPin); auto known = trust.verifyOrPin("peer", firstPin, false); REQUIRE(known.has_value()); CHECK_FALSE(known.value().firstContactPinned); CHECK_FALSE(known.value().pinnedByOperator); REQUIRE(trust.enrollOperatorPeer("peer", firstPin).has_value()); auto enrolled = trust.verifyOrPin("peer", firstPin, false); REQUIRE(enrolled.has_value()); CHECK_FALSE(enrolled.value().firstContactPinned); CHECK(enrolled.value().pinnedByOperator); auto changed = trust.verifyOrPin("peer", changedPin, true); REQUIRE_FALSE(changed.has_value()); CHECK(changed.error().code == yams::ErrorCode::Unauthorized); CHECK(trust.pinnedPin("peer") == firstPin); } TEST_CASE("P2P hello binds TLS identities and exchanges causal watermarks", "[daemon][p2p][handshake][network]") { auto serverConfig = config("server-node"); auto clientConfig = config("client-node"); REQUIRE(serverConfig.localVersion.increment("server-node")); REQUIRE(serverConfig.localVersion.increment("server-node")); REQUIRE(serverConfig.localVersion.increment("server-node")); REQUIRE(clientConfig.localVersion.increment("client-node")); REQUIRE(clientConfig.localVersion.increment("client-node")); serverConfig.localCommitments["server-node"] = {.counter = 3, .digest = std::string(64, 'a')}; clientConfig.localCommitments["client-node"] = {.counter = 2, .digest = std::string(64, 'b')}; clientConfig.localQuarantinedWriters.insert("retired-client-writer"); enableCurrentWriterWindow(serverConfig); enableCurrentWriterWindow(clientConfig); InMemoryPeerTrustStore serverTrust; InMemoryPeerTrustStore clientTrust; auto pair = runHandshake(serverConfig, clientConfig, serverTrust, clientTrust); REQUIRE(pair.client.has_value()); REQUIRE(pair.server.has_value()); CHECK(pair.client.value().peerNodeId == "server-node"); CHECK(pair.server.value().peerNodeId == "client-node"); CHECK(pair.client.value().firstContactPinned); CHECK(pair.server.value().firstContactPinned); CHECK_FALSE(pair.client.value().pinnedByOperator); CHECK_FALSE(pair.server.value().pinnedByOperator); CHECK(pair.client.value().peerWatermark("server-node") == 3); CHECK(pair.server.value().peerWatermark("client-node") == 2); CHECK(pair.client.value().peerVersion.get("server-node") == 3); CHECK(pair.server.value().peerVersion.get("client-node") == 2); REQUIRE(pair.client.value().peerCommitments.contains("server-node")); CHECK(pair.client.value().peerCommitments.at("server-node").counter == 3); CHECK(pair.client.value().peerCommitments.at("server-node").digest == std::string(64, 'a')); CHECK(pair.server.value().peerQuarantinedWriters.contains("retired-client-writer")); CHECK(clientTrust.pinnedPin("server-node") == pair.serverPin); CHECK(serverTrust.pinnedPin("client-node") == pair.clientPin); } TEST_CASE("P2P history comparison binds only the authenticated peer writer", "[daemon][p2p][handshake][commitment]") { yams::memory_sync::ReplicationState local; REQUIRE(local.version.increment("peer-node")); local.commitments["peer-node"] = {.counter = 1, .digest = std::string(64, 'a')}; local.commitments["unrelated"] = {.counter = 7, .digest = std::string(64, 'b')}; PeerHandshakeResult peer; peer.peerNodeId = "peer-node"; REQUIRE(peer.peerVersion.increment("peer-node")); peer.peerCommitments["peer-node"] = {.counter = 1, .digest = std::string(64, 'a')}; peer.peerCommitments["unrelated"] = {.counter = 7, .digest = std::string(64, 'c')}; auto unrelatedMismatch = yams::daemon::p2p::requiresPeerWriterQuarantine(local, peer); REQUIRE(unrelatedMismatch.has_value()); CHECK_FALSE(unrelatedMismatch.value()); peer.peerCommitments["peer-node"].digest = std::string(64, 'd'); auto authenticatedMismatch = yams::daemon::p2p::requiresPeerWriterQuarantine(local, peer); REQUIRE(authenticatedMismatch.has_value()); CHECK(authenticatedMismatch.value()); peer.peerCommitments.erase("peer-node"); auto missing = yams::daemon::p2p::requiresPeerWriterQuarantine(local, peer); REQUIRE_FALSE(missing.has_value()); CHECK(missing.error().code == yams::ErrorCode::ValidationError); } TEST_CASE("P2P handshake bounds a live writer window by its frozen snapshot", "[daemon][p2p][handshake][window][concurrency]") { // The handshake snapshot froze the local writer at counter 2; concurrent local writes have // since advanced the live history (read by the resolver) to counter 4. const auto digestAt = [](std::uint64_t counter) { return std::string(64, static_cast('0' + counter)); }; auto serverConfig = config("server-node"); serverConfig.localVersion.observe("server-node", 2); serverConfig.localCommitments["server-node"] = {.counter = 2, .digest = digestAt(2)}; serverConfig.resolveLocalWindow = [digestAt](std::uint64_t peerCounter, std::size_t maxRecords, std::size_t) { constexpr std::uint64_t kLiveCounter = 4; const auto endpoint = std::min(kLiveCounter, peerCounter + maxRecords); return yams::Result{ yams::memory_sync::WriterHistoryCommitment{.counter = endpoint, .digest = digestAt(endpoint)}}; }; auto clientConfig = config("client-node"); InMemoryPeerTrustStore serverTrust; InMemoryPeerTrustStore clientTrust; auto result = runHandshake(serverConfig, clientConfig, serverTrust, clientTrust); REQUIRE(result.server.has_value()); REQUIRE(result.client.has_value()); CHECK(result.client.value().peerVersion.get("server-node") == 2); CHECK(result.client.value().peerCommitments.at("server-node").digest == digestAt(2)); } TEST_CASE("P2P history comparison accepts a local prefix advanced by a concurrent session", "[daemon][p2p][handshake][commitment][concurrency]") { // The handshake verified local prefix 2 and froze the peer frontier at 5; another session // with the same peer has since advanced the local prefix to 3. yams::memory_sync::ReplicationState local; local.version.observe("peer-node", 3); local.commitments["peer-node"] = {.counter = 3, .digest = std::string(64, 'a')}; PeerHandshakeResult peer; peer.peerNodeId = "peer-node"; peer.peerVersion.observe("peer-node", 5); peer.peerCommitments["peer-node"] = {.counter = 5, .digest = std::string(64, 'b')}; peer.peerPrefixCounter = 2; peer.peerPrefixVerified = true; peer.peerPrefixMatches = true; auto advanced = yams::daemon::p2p::requiresPeerWriterQuarantine(local, peer); REQUIRE(advanced.has_value()); CHECK_FALSE(advanced.value()); peer.peerPrefixVerified = false; auto unverified = yams::daemon::p2p::requiresPeerWriterQuarantine(local, peer); REQUIRE_FALSE(unverified.has_value()); peer.peerPrefixVerified = true; peer.peerPrefixCounter = 4; auto behindProof = yams::daemon::p2p::requiresPeerWriterQuarantine(local, peer); REQUIRE_FALSE(behindProof.has_value()); peer.peerPrefixCounter = 2; peer.peerPrefixMatches = false; auto forked = yams::daemon::p2p::requiresPeerWriterQuarantine(local, peer); REQUIRE(forked.has_value()); CHECK(forked.value()); } TEST_CASE("P2P handshake verifies a nonzero authenticated writer prefix before exchange", "[daemon][p2p][handshake][commitment][prefix]") { auto serverConfig = config("server-node"); auto clientConfig = config("client-node"); REQUIRE(serverConfig.localVersion.increment("server-node")); REQUIRE(serverConfig.localVersion.increment("server-node")); serverConfig.localCommitments["server-node"] = {.counter = 2, .digest = std::string(64, 'b')}; serverConfig.resolveLocalCommitment = [](std::uint64_t counter) -> yams::Result { if (counter != 1) { return yams::Error{yams::ErrorCode::InvalidState, "unexpected proof counter"}; } return yams::memory_sync::WriterHistoryCommitment{.counter = 1, .digest = std::string(64, 'a')}; }; REQUIRE(clientConfig.localVersion.increment("server-node")); clientConfig.localCommitments["server-node"] = {.counter = 1, .digest = std::string(64, 'a')}; enableCurrentWriterWindow(serverConfig); InMemoryPeerTrustStore serverTrust; InMemoryPeerTrustStore clientTrust; auto pair = runHandshake(serverConfig, clientConfig, serverTrust, clientTrust); REQUIRE(pair.client.has_value()); REQUIRE(pair.server.has_value()); CHECK(pair.client.value().peerPrefixVerified); CHECK(pair.client.value().peerPrefixCounter == 1); auto verified = yams::daemon::p2p::requiresPeerWriterQuarantine( yams::memory_sync::ReplicationState{.version = clientConfig.localVersion, .commitments = clientConfig.localCommitments}, pair.client.value()); REQUIRE(verified.has_value()); CHECK_FALSE(verified.value()); } TEST_CASE("P2P handshake verifies an acceptor prefix of the connector writer", "[daemon][p2p][handshake][commitment][prefix]") { auto serverConfig = config("server-node"); auto clientConfig = config("client-node"); REQUIRE(clientConfig.localVersion.increment("client-node")); REQUIRE(clientConfig.localVersion.increment("client-node")); clientConfig.localCommitments["client-node"] = {.counter = 2, .digest = std::string(64, 'd')}; clientConfig.resolveLocalCommitment = [](std::uint64_t counter) -> yams::Result { if (counter != 1) { return yams::Error{yams::ErrorCode::InvalidState, "unexpected proof counter"}; } return yams::memory_sync::WriterHistoryCommitment{.counter = 1, .digest = std::string(64, 'c')}; }; REQUIRE(serverConfig.localVersion.increment("client-node")); serverConfig.localCommitments["client-node"] = {.counter = 1, .digest = std::string(64, 'c')}; enableCurrentWriterWindow(clientConfig); InMemoryPeerTrustStore serverTrust; InMemoryPeerTrustStore clientTrust; auto pair = runHandshake(serverConfig, clientConfig, serverTrust, clientTrust); REQUIRE(pair.client.has_value()); REQUIRE(pair.server.has_value()); CHECK(pair.server.value().peerPrefixVerified); CHECK(pair.server.value().peerPrefixMatches); CHECK(pair.server.value().peerPrefixCounter == 1); } TEST_CASE("P2P handshake fails closed when writer prefix proof is unavailable", "[daemon][p2p][handshake][commitment][prefix][security]") { auto serverConfig = config("server-node"); auto clientConfig = config("client-node"); REQUIRE(serverConfig.localVersion.increment("server-node")); serverConfig.localCommitments["server-node"] = {.counter = 1, .digest = std::string(64, 'a')}; SECTION("historical prefix cannot be derived") { REQUIRE(serverConfig.localVersion.increment("server-node")); serverConfig.localCommitments["server-node"] = {.counter = 2, .digest = std::string(64, 'b')}; REQUIRE(clientConfig.localVersion.increment("server-node")); clientConfig.localCommitments["server-node"] = {.counter = 1, .digest = std::string(64, 'a')}; } SECTION("peer claims history ahead of the authenticated writer") { REQUIRE(clientConfig.localVersion.increment("server-node")); REQUIRE(clientConfig.localVersion.increment("server-node")); clientConfig.localCommitments["server-node"] = {.counter = 2, .digest = std::string(64, 'b')}; } enableCurrentWriterWindow(serverConfig); InMemoryPeerTrustStore serverTrust; InMemoryPeerTrustStore clientTrust; auto pair = runHandshake(serverConfig, clientConfig, serverTrust, clientTrust); REQUIRE_FALSE(pair.client.has_value()); REQUIRE_FALSE(pair.server.has_value()); CHECK_FALSE(serverTrust.pinnedPin("client-node").has_value()); CHECK_FALSE(clientTrust.pinnedPin("server-node").has_value()); } TEST_CASE("P2P handshake rejects unbounded local replication state", "[daemon][p2p][handshake][commitment][limits]") { auto serverConfig = config("server-node"); auto clientConfig = config("client-node"); for (std::size_t index = 0; index <= yams::daemon::p2p::kMaxP2pReplicationWriters; ++index) { const auto writer = "writer-" + std::to_string(index); REQUIRE(clientConfig.localVersion.increment(writer)); clientConfig.localCommitments[writer] = {.counter = 1, .digest = std::string(64, 'a')}; } InMemoryPeerTrustStore serverTrust; InMemoryPeerTrustStore clientTrust; auto pair = runHandshake(serverConfig, clientConfig, serverTrust, clientTrust); REQUIRE_FALSE(pair.client.has_value()); CHECK(pair.client.error().code == yams::ErrorCode::ResourceExhausted); REQUIRE_FALSE(pair.server.has_value()); CHECK_FALSE(serverTrust.pinnedPin("client-node").has_value()); } TEST_CASE("P2P hello rejects incompatible replication identities", "[daemon][p2p][handshake][network]") { auto serverConfig = config("server-node"); auto clientConfig = config("client-node"); std::string certificateNode; yams::ErrorCode expectedClientCode = yams::ErrorCode::Unauthorized; yams::ErrorCode expectedServerCode = yams::ErrorCode::ValidationError; bool clientClosedBeforeHello = false; SECTION("corpus mismatch") { clientConfig.corpusId = "other-corpus"; } SECTION("epoch mismatch") { clientConfig.corpusEpoch = 2; } SECTION("local claimed node differs from TLS certificate") { certificateNode = "certificate-node"; expectedClientCode = yams::ErrorCode::InvalidState; clientClosedBeforeHello = true; } InMemoryPeerTrustStore serverTrust; InMemoryPeerTrustStore clientTrust; auto pair = runHandshake(serverConfig, clientConfig, serverTrust, clientTrust, certificateNode); REQUIRE_FALSE(pair.client.has_value()); REQUIRE_FALSE(pair.server.has_value()); CHECK(pair.client.error().code == expectedClientCode); if (clientClosedBeforeHello) { const auto code = pair.server.error().code; CHECK((code == yams::ErrorCode::NetworkError || code == yams::ErrorCode::NotFound || code == yams::ErrorCode::OperationCancelled)); } else { CHECK(pair.server.error().code == expectedServerCode); } CHECK_FALSE(serverTrust.pinnedPin("client-node").has_value()); CHECK_FALSE(clientTrust.pinnedPin("server-node").has_value()); } TEST_CASE("P2P hello rejects unsupported wire protocol and schema versions", "[daemon][p2p][handshake][network]") { std::uint32_t protocol = yams::daemon::p2p::kP2pProtocolVersion; std::uint32_t schema = yams::daemon::p2p::kP2pEnvelopeSchemaVersion; SECTION("protocol mismatch") { ++protocol; } SECTION("schema mismatch") { --schema; } auto serverKey = yams::memory_sync::generateWriterKeyPair(); auto clientKey = yams::memory_sync::generateWriterKeyPair(); REQUIRE(serverKey.has_value()); REQUIRE(clientKey.has_value()); auto serverIdentity = identity("server-node", serverKey.value()); auto clientIdentity = identity("client-node", clientKey.value()); const std::string serverPin = serverIdentity.spkiPin(); const std::string clientPin = clientIdentity.spkiPin(); auto serverConfig = config("server-node"); InMemoryPeerTrustStore serverTrust; std::promise> serverDone; auto serverFuture = serverDone.get_future(); P2pListener listener(P2pListener::Options{.host = "127.0.0.1", .port = 0, .identity = std::move(serverIdentity), .allowedPeerPins = {clientPin}, .allowUnpinnedPeers = false, .handshakeTimeout = 2s, .maxConcurrentSessions = 1}); listener.setHandler([&](P2pConnection channel) { serverDone.set_value( yams::daemon::p2p::acceptPeerHandshake(channel, serverConfig, serverTrust)); }); REQUIRE(listener.start().has_value()); auto channel = yams::daemon::p2p::p2pConnect(P2pClientOptions{.host = "127.0.0.1", .port = listener.boundPort(), .identity = std::move(clientIdentity), .expectedPeerPin = serverPin, .allowUnpinnedPeer = false, .connectTimeout = 2s, .handshakeTimeout = 2s}); REQUIRE(channel.has_value()); const nlohmann::json hello{{"type", "hello"}, {"protocol", protocol}, {"schema_version", schema}, {"node_id", "client-node"}, {"corpus_id", "shared-corpus"}, {"corpus_epoch", 1}, {"max_writer_advance", 1024}, {"max_writer_window_bytes", 64 * 1024 * 1024}}; const std::string helloText = hello.dump(); REQUIRE( channel.value() .writeFrame(std::span( reinterpret_cast(helloText.data()), helloText.size()), 2s) .has_value()); auto acknowledgement = channel.value().readFrame(2s); REQUIRE(acknowledgement.has_value()); const std::string ackText(reinterpret_cast(acknowledgement.value().data()), acknowledgement.value().size()); const auto ack = nlohmann::json::parse(ackText); CHECK_FALSE(ack.at("accepted").get()); REQUIRE(serverFuture.wait_for(3s) == std::future_status::ready); auto serverResult = serverFuture.get(); REQUIRE_FALSE(serverResult.has_value()); CHECK(serverResult.error().code == yams::ErrorCode::NotSupported); CHECK_FALSE(serverTrust.pinnedPin("client-node").has_value()); listener.stop(); } TEST_CASE("P2P first contact is not pinned until state validation succeeds", "[daemon][p2p][handshake][network][trust]") { auto serverKey = yams::memory_sync::generateWriterKeyPair(); auto clientKey = yams::memory_sync::generateWriterKeyPair(); REQUIRE(serverKey.has_value()); REQUIRE(clientKey.has_value()); auto serverIdentity = identity("server-node", serverKey.value()); auto clientIdentity = identity("client-node", clientKey.value()); const std::string serverPin = serverIdentity.spkiPin(); const std::string clientPin = clientIdentity.spkiPin(); auto serverConfig = config("server-node"); InMemoryPeerTrustStore serverTrust; std::promise> serverDone; auto serverFuture = serverDone.get_future(); P2pListener listener(P2pListener::Options{.host = "127.0.0.1", .port = 0, .identity = std::move(serverIdentity), .allowedPeerPins = {clientPin}, .allowUnpinnedPeers = false, .handshakeTimeout = 2s, .maxConcurrentSessions = 1}); listener.setHandler([&](P2pConnection channel) { serverDone.set_value( yams::daemon::p2p::acceptPeerHandshake(channel, serverConfig, serverTrust)); }); REQUIRE(listener.start().has_value()); auto channel = yams::daemon::p2p::p2pConnect(P2pClientOptions{.host = "127.0.0.1", .port = listener.boundPort(), .identity = std::move(clientIdentity), .expectedPeerPin = serverPin, .allowUnpinnedPeer = false, .connectTimeout = 2s, .handshakeTimeout = 2s}); REQUIRE(channel.has_value()); const nlohmann::json hello{{"type", "hello"}, {"protocol", yams::daemon::p2p::kP2pProtocolVersion}, {"schema_version", yams::daemon::p2p::kP2pEnvelopeSchemaVersion}, {"node_id", "client-node"}, {"corpus_id", "shared-corpus"}, {"corpus_epoch", 1}, {"max_writer_advance", 1024}, {"max_writer_window_bytes", 64 * 1024 * 1024}}; const std::string helloText = hello.dump(); REQUIRE( channel.value() .writeFrame(std::span( reinterpret_cast(helloText.data()), helloText.size()), 2s) .has_value()); REQUIRE(channel.value().readFrame(2s).has_value()); // accepted hello_ack const nlohmann::json malformedState{{"type", "state"}, {"vv", {{"counters_", {{"client-node", 1}}}}}, {"seen", {{"client-node", 2}}}}; const std::string stateText = malformedState.dump(); REQUIRE( channel.value() .writeFrame(std::span( reinterpret_cast(stateText.data()), stateText.size()), 2s) .has_value()); REQUIRE(serverFuture.wait_for(3s) == std::future_status::ready); auto serverResult = serverFuture.get(); REQUIRE_FALSE(serverResult.has_value()); CHECK(serverResult.error().code == yams::ErrorCode::ValidationError); CHECK_FALSE(serverTrust.pinnedPin("client-node").has_value()); listener.stop(); } TEST_CASE("P2P hello rejects a previously pinned peer with a changed key", "[daemon][p2p][handshake][network][trust]") { auto serverConfig = config("server-node"); auto clientConfig = config("client-node"); serverConfig.allowFirstContact = false; InMemoryPeerTrustStore serverTrust; InMemoryPeerTrustStore clientTrust; REQUIRE(serverTrust.verifyOrPin("client-node", std::string(64, '0'), true).has_value()); auto pair = runHandshake(serverConfig, clientConfig, serverTrust, clientTrust); REQUIRE_FALSE(pair.client.has_value()); REQUIRE_FALSE(pair.server.has_value()); CHECK(pair.client.error().code == yams::ErrorCode::Unauthorized); CHECK(pair.server.error().code == yams::ErrorCode::Unauthorized); CHECK(serverTrust.pinnedPin("client-node") == std::string(64, '0')); }