yams/tests/unit/daemon/p2p_protocol_catch2_test.cpp

// SPDX-License-Identifier: GPL-3.0-or-later
// Copyright 2026 YAMS Contributors

// pi-lens-ignore: fatal error
#include <catch2/catch_test_macros.hpp>
// pi-lens-ignore: fatal error
#include <nlohmann/json.hpp>

#include <yams/daemon/p2p/p2p_protocol.h>
#include <yams/memory_sync/writer_auth.h>

#include "../../../src/daemon/p2p/p2p_fuzz.h"

#include <algorithm>
#include <chrono>
#include <future>
#include <string>
#include <utility>

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<void> validateHandshakeJson(const nlohmann::json& json) {
    const auto text = json.dump();
    return yams::daemon::p2p::detail::validateHandshakeControlFrame(
        std::span<const std::byte>(reinterpret_cast<const std::byte*>(text.data()), text.size()));
}

yams::Result<void> validateHandshakeBytes(std::string_view text) {
    return yams::daemon::p2p::detail::validateHandshakeControlFrame(
        std::span<const std::byte>(reinterpret_cast<const std::byte*>(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::memory_sync::WriterHistoryCommitment>{yams::Error{
                yams::ErrorCode::ResourceExhausted, "test writer window exceeds bound"}};
        }
        return yams::Result<yams::memory_sync::WriterHistoryCommitment>{commitment};
    };
}

struct PairResult {
    yams::Result<PeerHandshakeResult> client;
    yams::Result<PeerHandshakeResult> 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<yams::Result<PeerHandshakeResult>> 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<std::uint32_t>(), 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>{
            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>{
            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<char>('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<std::uint64_t>(kLiveCounter, peerCounter + maxRecords);
        return yams::Result<yams::memory_sync::WriterHistoryCommitment>{
            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<yams::memory_sync::WriterHistoryCommitment> {
        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<yams::memory_sync::WriterHistoryCommitment> {
        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<yams::Result<PeerHandshakeResult>> 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<const std::byte>(
                            reinterpret_cast<const std::byte*>(helloText.data()), helloText.size()),
                        2s)
            .has_value());
    auto acknowledgement = channel.value().readFrame(2s);
    REQUIRE(acknowledgement.has_value());
    const std::string ackText(reinterpret_cast<const char*>(acknowledgement.value().data()),
                              acknowledgement.value().size());
    const auto ack = nlohmann::json::parse(ackText);
    CHECK_FALSE(ack.at("accepted").get<bool>());

    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<yams::Result<PeerHandshakeResult>> 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<const std::byte>(
                            reinterpret_cast<const std::byte*>(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<const std::byte>(
                            reinterpret_cast<const std::byte*>(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'));
}