network_identity_subsystem.h

Go to the documentation of this file.

// Copyright (c) Microsoft Corporation. All rights reserved.
// Licensed under the Apache 2.0 License.
#pragma once
 
#include "ccf/network_identity_interface.h"
#include "ccf/service/tables/service.h"
#include "node/historical_queries.h"
#include "node/identity.h"
#include "node/rpc/node_interface.h"
#include "service/internal_tables_access.h"
 
#include <atomic>
 
namespace ccf
{
  static bool is_self_endorsement(const ccf::CoseEndorsement& endorsement)
  {
    return !endorsement.previous_version.has_value();
  }
 
  static void validate_fetched_endorsement(
    const ccf::CoseEndorsement& endorsement)
  {
    if (!is_self_endorsement(endorsement))
    {
      const auto [from, to] =
        ccf::crypto::extract_cose_endorsement_validity(endorsement.endorsement);
 
      const auto from_txid = ccf::TxID::from_str(from);
      if (!from_txid.has_value())
      {
        throw std::logic_error(fmt::format(
          "Cannot parse COSE endorsement header: {}",
          ccf::crypto::COSE_PHEADER_KEY_RANGE_BEGIN));
      }
 
      const auto to_txid = ccf::TxID::from_str(to);
      if (!to_txid.has_value())
      {
        throw std::logic_error(fmt::format(
          "Cannot parse COSE endorsement header: {}",
          ccf::crypto::COSE_PHEADER_KEY_RANGE_END));
      }
 
      if (!endorsement.endorsement_epoch_end.has_value())
      {
        throw std::logic_error(
          "COSE endorsement does not contain epoch end in the table entry");
      }
      if (
        endorsement.endorsement_epoch_begin != *from_txid ||
        *endorsement.endorsement_epoch_end != *to_txid)
      {
        throw std::logic_error(fmt ::format(
          "COSE endorsement fetched but range is invalid, epoch begin {}, "
          "epoch end {}, header epoch begin: {}, header epoch end: {}",
          endorsement.endorsement_epoch_begin.to_str(),
          endorsement.endorsement_epoch_end->to_str(),
          from,
          to));
      }
    }
  }
 
  static void validate_chain_integrity(
    const ccf::CoseEndorsement& newer, const ccf::CoseEndorsement& older)
  {
    if (!older.endorsement_epoch_end.has_value())
    {
      throw std::logic_error(fmt::format(
        "COSE endorsement chain integrity is violated, previous endorsement "
        "from {} does not have an epoch end",
        older.endorsement_epoch_begin.to_str()));
    }
 
    if (
      newer.endorsement_epoch_begin.view - aft::starting_view_change !=
        older.endorsement_epoch_end->view ||
      newer.endorsement_epoch_begin.seqno - 1 !=
        older.endorsement_epoch_end->seqno)
    {
      throw std::logic_error(fmt::format(
        "COSE endorsement chain integrity is violated, previous endorsement "
        "epoch end {} is not chained with newer endorsement epoch begin {}",
        older.endorsement_epoch_end->to_str(),
        newer.endorsement_epoch_begin.to_str()));
    }
  }
 
  class NetworkIdentitySubsystem : public NetworkIdentitySubsystemInterface
  {
  protected:
    AbstractNodeState& node_state;
    const std::unique_ptr<NetworkIdentity>& network_identity;
    std::shared_ptr<historical::StateCacheImpl> historical_cache;
    std::map<SeqNo, CoseEndorsement> endorsements;
    std::optional<TxID> current_service_from;
    SeqNo earliest_endorsed_seq{0};
    std::atomic<FetchStatus> fetch_status{FetchStatus::Retry};
    bool has_predecessors{false};
 
  public:
    NetworkIdentitySubsystem(
      AbstractNodeState& node_state_,
      const std::unique_ptr<NetworkIdentity>& network_identity_,
      std::shared_ptr<ccf::historical::StateCacheImpl> historical_cache_) :
      node_state(node_state_),
      network_identity(network_identity_),
      historical_cache(std::move(historical_cache_))
    {
      fetch_first();
    }
 
    [[nodiscard]] FetchStatus endorsements_fetching_status() const override
    {
      return fetch_status.load();
    }
 
    const std::unique_ptr<NetworkIdentity>& get() override
    {
      return network_identity;
    }
 
    [[nodiscard]] std::optional<CoseEndorsementsChain>
    get_cose_endorsements_chain(ccf::SeqNo seqno) const override
    {
      // All other cases must be handled after recovery has been completed and
      // identities have been successfully fetched.
      if (fetch_status.load() != FetchStatus::Done)
      {
        return std::nullopt;
      }
 
      if (!current_service_from.has_value())
      {
        LOG_FAIL_FMT(
          "Unset current_service_from when fetching endorsements chain");
        return std::nullopt;
      }
 
      if (!has_predecessors || seqno >= current_service_from->seqno)
      {
        return CoseEndorsementsChain{};
      }
 
      auto it = endorsements.upper_bound(seqno);
      if (it == endorsements.begin())
      {
        LOG_INFO_FMT(
          "No endorsements found for seqno {}, earliest endorsed is {}",
          seqno,
          earliest_endorsed_seq);
        return {};
      }
 
      CoseEndorsementsChain result;
      for (--it; it != endorsements.end(); ++it)
      {
        result.push_back(it->second.endorsement);
      }
      std::reverse(result.begin(), result.end());
      return result;
    }
 
  private:
    void retry_first_fetch()
    {
      using namespace std::chrono_literals;
      static constexpr auto retry_after = 1s;
      ccf::tasks::add_delayed_task(
        ccf::tasks::make_basic_task([this]() { this->fetch_first(); }),
        retry_after);
    }
 
    void fail_fetching(const std::string& err = "")
    {
      if (!err.empty())
      {
        LOG_FAIL_FMT("Failed fetching network identity: {}", err);
      }
      fetch_status.store(FetchStatus::Failed);
    }
 
    void complete_fetching()
    {
      if (!current_service_from.has_value())
      {
        fail_fetching("Unset current_service_from when completing fetching");
        return;
      }
 
      if (!endorsements.empty())
      {
        try
        {
          auto next = endorsements.begin();
          auto prev = next++;
          while (next != endorsements.end())
          {
            validate_chain_integrity(next->second, prev->second);
            ++prev;
            ++next;
          }
 
          const auto& last = prev->second;
          if (!last.endorsement_epoch_end.has_value())
          {
            fail_fetching(fmt::format(
              "The last fetched endorsement at {} has no epoch end",
              last.endorsement_epoch_begin.seqno));
            return;
          }
 
          if (
            current_service_from->view - aft::starting_view_change !=
              last.endorsement_epoch_end->view ||
            current_service_from->seqno - 1 !=
              last.endorsement_epoch_end->seqno)
          {
            fail_fetching(fmt::format(
              "COSE endorsement chain integrity is violated, the current "
              "service start at {} is not chained with previous endorsement "
              "ending at {}",
              current_service_from->to_str(),
              last.endorsement_epoch_end->to_str()));
            return;
          }
        }
        catch (const std::exception& e)
        {
          fail_fetching(e.what());
          return;
        }
      }
 
      fetch_status.store(FetchStatus::Done);
    }
 
    void fetch_first()
    {
      if (!node_state.is_part_of_network())
      {
        LOG_INFO_FMT(
          "Retry fetching network identity as node is not part of the network "
          "yet");
        retry_first_fetch();
        return;
      }
 
      auto store = node_state.get_store();
      auto tx = store->create_read_only_tx();
 
      if (!current_service_from.has_value())
      {
        auto* service_info_handle =
          tx.template ro<ccf::Service>(ccf::Tables::SERVICE);
        auto service_info = service_info_handle->get();
        if (
          !service_info ||
          !service_info->current_service_create_txid.has_value())
        {
          LOG_INFO_FMT(
            "Retrying fetching network identity as current service create txid "
            "is not yet available");
          retry_first_fetch();
          return;
        }
 
        current_service_from = service_info->current_service_create_txid;
      }
 
      auto* previous_identity_endorsement =
        tx.ro<ccf::PreviousServiceIdentityEndorsement>(
          ccf::Tables::PREVIOUS_SERVICE_IDENTITY_ENDORSEMENT);
 
      auto endorsement = previous_identity_endorsement->get();
      if (!endorsement.has_value())
      {
        LOG_INFO_FMT(
          "Retrying fetching network identity as there is no previous service "
          "identity endorsement yet");
        retry_first_fetch();
        return;
      }
 
      if (is_self_endorsement(endorsement.value()))
      {
        if (
          current_service_from->seqno !=
          endorsement->endorsement_epoch_begin.seqno)
        {
          fail_fetching(fmt::format(
            "The first fetched endorsement is a self-endorsement with seqno {} "
            "which is different from current_service_create_txid {}",
            endorsement->endorsement_epoch_begin.seqno,
            current_service_from->seqno));
          return;
        }
 
        LOG_INFO_FMT(
          "The very first service endorsement is self-signed at {}, no "
          "endorsement chain will be preloaded",
          current_service_from->seqno);
 
        has_predecessors = false;
        complete_fetching();
        return;
      }
 
      has_predecessors = true;
      earliest_endorsed_seq = current_service_from->seqno;
      process_endorsement(endorsement.value());
    }
 
    void process_endorsement(const ccf::CoseEndorsement& endorsement)
    {
      const auto from = endorsement.endorsement_epoch_begin.seqno;
      if (is_self_endorsement(endorsement))
      {
        if (endorsements.find(from) == endorsements.end())
        {
          fail_fetching(fmt::format(
            "Fetched self-endorsement with seqno {} which has not been seen",
            from));
          return;
        }
        LOG_INFO_FMT("Got self-endorsement at {}, stopping fetching", from);
        complete_fetching();
        return;
      }
 
      if (from >= earliest_endorsed_seq)
      {
        fail_fetching(fmt::format(
          "Fetched service endorsement with seqno {} which is greater than "
          "the earliest known in the chain {}",
          from,
          earliest_endorsed_seq));
        return;
      }
 
      if (!endorsement.endorsement_epoch_end.has_value())
      {
        fail_fetching(
          fmt::format("Fetched endorsement at {} has no epoch end", from));
        return;
      }
 
      earliest_endorsed_seq = from;
      if (endorsements.find(from) != endorsements.end())
      {
        fail_fetching(fmt::format(
          "Fetched service endorsement with seqno {} which already exists",
          from));
        return;
      }
 
      LOG_INFO_FMT(
        "Fetched service endorsement from {} to {}",
        from,
        endorsement.endorsement_epoch_end->seqno);
      endorsements.insert({from, endorsement});
 
      if (endorsement.previous_version.has_value())
      {
        fetch_next_at(endorsement.previous_version.value());
        return;
      }
 
      complete_fetching();
    }
 
    void fetch_next_at(ccf::SeqNo seq)
    {
      auto state = historical_cache->get_state_at(
        ccf::historical::CompoundHandle{
          ccf::historical::RequestNamespace::System, seq},
        seq);
      if (!state)
      {
        retry_fetch_next(seq);
        return;
      }
 
      if (!state->store)
      {
        fail_fetching(fmt::format(
          "Fetched historical state with seqno {} with missing store", seq));
        return;
      }
      auto htx = state->store->create_read_only_tx();
      const auto endorsement =
        htx
          .template ro<ccf::PreviousServiceIdentityEndorsement>(
            ccf::Tables::PREVIOUS_SERVICE_IDENTITY_ENDORSEMENT)
          ->get();
 
      if (!endorsement.has_value())
      {
        fail_fetching(
          fmt::format("Fetched COSE endorsement for {} is invalid", seq));
        return;
      }
 
      try
      {
        validate_fetched_endorsement(endorsement.value());
      }
      catch (const std::exception& e)
      {
        fail_fetching(e.what());
        return;
      }
 
      process_endorsement(endorsement.value());
    }
 
    void retry_fetch_next(ccf::SeqNo seq)
    {
      using namespace std::chrono_literals;
      static constexpr auto retry_after = 100ms;
      ccf::tasks::add_delayed_task(
        ccf::tasks::make_basic_task(
          [this, seq]() { this->fetch_next_at(seq); }),
        retry_after);
    }
  };
}