Merge remote-tracking branch 'igor/fix/noise-bytes-written' into proto-abs

include/libp2p/security/noise/noise_connection.hpp

@@ -6,6 +6,8 @@
 #ifndef LIBP2P_INCLUDE_LIBP2P_SECURITY_NOISE_NOISE_CONNECTION_HPP
 #define LIBP2P_INCLUDE_LIBP2P_SECURITY_NOISE_NOISE_CONNECTION_HPP
 
+#include <list>
+
 #include <libp2p/connection/secure_connection.hpp>
 
 #include <libp2p/common/metrics/instance_count.hpp>
@@ -22,6 +24,14 @@ namespace libp2p::connection {
   class NoiseConnection : public SecureConnection,
                           public std::enable_shared_from_this<NoiseConnection> {
    public:
+    using BufferList = std::list<common::ByteArray>;
+
+    struct OperationContext {
+      size_t bytes_served;                /// written or read bytes count
+      const size_t total_bytes;           /// total size to process
+      BufferList::iterator write_buffer;  /// temporary data storage
+    };
+
     ~NoiseConnection() override = default;
 
     NoiseConnection(
@@ -65,6 +75,17 @@ namespace libp2p::connection {
     outcome::result<crypto::PublicKey> remotePublicKey() const override;
 
    private:
+    void read(gsl::span<uint8_t> out, size_t bytes, OperationContext ctx,
+              ReadCallbackFunc cb);
+
+    void readSome(gsl::span<uint8_t> out, size_t bytes, OperationContext ctx,
+                  ReadCallbackFunc cb);
+
+    void write(gsl::span<const uint8_t> in, size_t bytes, OperationContext ctx,
+               WriteCallbackFunc cb);
+
+    void eraseWriteBuffer(BufferList::iterator &iterator);
+
     std::shared_ptr<RawConnection> raw_connection_;
     crypto::PublicKey local_;
     crypto::PublicKey remote_;
@@ -73,10 +94,7 @@ namespace libp2p::connection {
     std::shared_ptr<security::noise::CipherState> decoder_cs_;
     std::shared_ptr<common::ByteArray> frame_buffer_;
     std::shared_ptr<security::noise::InsecureReadWriter> framer_;
-    size_t already_read_;
-    size_t already_wrote_;
-    size_t plaintext_len_to_write_;
-    common::ByteArray writing_;
+    BufferList write_buffers_;
     log::Logger log_ = log::createLogger("NoiseConnection");
 
    public:

src/common/hexutil.cpp

@@ -3,6 +3,8 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
+#include <ios>
+
 #include <libp2p/common/hexutil.hpp>
 
 #include <boost/algorithm/hex.hpp>
@@ -24,13 +26,13 @@ namespace libp2p::common {
 
   std::string int_to_hex(uint64_t n, size_t fixed_width) noexcept {
     std::stringstream result;
-    result.width(fixed_width);
+    result.width(static_cast<std::streamsize>(fixed_width));
     result.fill('0');
     result << std::hex << std::uppercase << n;
     auto str = result.str();
     if (str.length() % 2 != 0) {
       str.push_back('\0');
-      for (int64_t i = str.length() - 2; i >= 0; --i) {
+      for (int64_t i = static_cast<int64_t>(str.length()) - 2; i >= 0; --i) {
         str[i + 1] = str[i];
       }
       str[0] = '0';

src/crypto/common_functions.cpp

@@ -25,8 +25,8 @@ namespace libp2p::crypto {
     }
 
     // turn private key bytes into big number
-    BIGNUM *private_bignum{
-        BN_bin2bn(private_key.data(), private_key.size(), nullptr)};
+    BIGNUM *private_bignum{BN_bin2bn(
+        private_key.data(), static_cast<int>(private_key.size()), nullptr)};
     if (nullptr == private_bignum) {
       return FAILED;
     }
@@ -92,10 +92,11 @@ namespace libp2p::crypto {
       int, gsl::span<const uint8_t>, decltype(EVP_PKEY_new_raw_public_key) *);
 
   outcome::result<std::vector<uint8_t>> GenerateEcSignature(
-      gsl::span<const uint8_t> digest,
-      const std::shared_ptr<EC_KEY> &key) {
+      gsl::span<const uint8_t> digest, const std::shared_ptr<EC_KEY> &key) {
     std::shared_ptr<ECDSA_SIG> signature{
-        ECDSA_do_sign(digest.data(), digest.size(), key.get()), ECDSA_SIG_free};
+        ECDSA_do_sign(digest.data(), static_cast<int>(digest.size()),
+                      key.get()),
+        ECDSA_SIG_free};
     if (signature == nullptr) {
       return CryptoProviderError::SIGNATURE_GENERATION_FAILED;
     }
@@ -110,12 +111,12 @@ namespace libp2p::crypto {
     return std::move(signature_bytes);
   }
 
-  outcome::result<bool> VerifyEcSignature(
-      gsl::span<const uint8_t> digest,
+  outcome::result<bool> VerifyEcSignature(gsl::span<const uint8_t> digest,
                                           gsl::span<const uint8_t> signature,
                                           const std::shared_ptr<EC_KEY> &key) {
-    int result = ECDSA_verify(0, digest.data(), digest.size(), signature.data(),
-                              signature.size(), key.get());
+    int result = ECDSA_verify(0, digest.data(), static_cast<int>(digest.size()),
+                              signature.data(),
+                              static_cast<int>(signature.size()), key.get());
     if (result < 0) {
       return CryptoProviderError::SIGNATURE_VERIFICATION_FAILED;
     }

src/crypto/key_marshaller/key_marshaller_impl.cpp

@@ -84,7 +84,7 @@ namespace libp2p::crypto::marshaller {
       const ProtobufKey &proto_key) const {
     protobuf::PublicKey protobuf_key;
     if (!protobuf_key.ParseFromArray(proto_key.key.data(),
-                                     proto_key.key.size())) {
+                                     static_cast<int>(proto_key.key.size()))) {
       return CryptoProviderError::FAILED_UNMARSHAL_DATA;
     }
 
@@ -101,7 +101,7 @@ namespace libp2p::crypto::marshaller {
       const ProtobufKey &proto_key) const {
     protobuf::PublicKey protobuf_key;
     if (!protobuf_key.ParseFromArray(proto_key.key.data(),
-                                     proto_key.key.size())) {
+                                     static_cast<int>(proto_key.key.size()))) {
       return CryptoProviderError::FAILED_UNMARSHAL_DATA;
     }
 

src/security/noise/handshake_message_marshaller_impl.cpp

@@ -59,7 +59,8 @@ namespace libp2p::security::noise {
       const HandshakeMessage &msg) const {
     OUTCOME_TRY(proto_msg, handyToProto(msg));
     common::ByteArray out_msg(proto_msg.ByteSizeLong());
-    if (not proto_msg.SerializeToArray(out_msg.data(), out_msg.size())) {
+    if (not proto_msg.SerializeToArray(out_msg.data(),
+                                       static_cast<int>(out_msg.size()))) {
       return Error::MESSAGE_SERIALIZING_ERROR;
     }
     return out_msg;
@@ -69,7 +70,8 @@ namespace libp2p::security::noise {
   HandshakeMessageMarshallerImpl::unmarshal(
       gsl::span<const uint8_t> msg_bytes) const {
     protobuf::NoiseHandshakePayload proto_msg;
-    if (not proto_msg.ParseFromArray(msg_bytes.data(), msg_bytes.size())) {
+    if (not proto_msg.ParseFromArray(msg_bytes.data(),
+                                     static_cast<int>(msg_bytes.size()))) {
       return Error::MESSAGE_DESERIALIZING_ERROR;
     }
     return protoToHandy(proto_msg);

src/security/noise/noise_connection.cpp

@@ -15,6 +15,7 @@
 #define OUTCOME_CB_I(var, res)                \
   auto && (var) = (res);                      \
   if ((var).has_error()) {                    \
+    self->eraseWriteBuffer(ctx.write_buffer); \
     return cb((var).error());                 \
   }
 
@@ -40,10 +41,7 @@ namespace libp2p::connection {
         frame_buffer_{
             std::make_shared<common::ByteArray>(security::noise::kMaxMsgLen)},
         framer_{std::make_shared<security::noise::InsecureReadWriter>(
-            raw_connection_, frame_buffer_)},
-        already_read_{0},
-        already_wrote_{0},
-        plaintext_len_to_write_{0} {
+            raw_connection_, frame_buffer_)} {
     BOOST_ASSERT(raw_connection_);
     BOOST_ASSERT(key_marshaller_);
     BOOST_ASSERT(encoder_cs_);
@@ -63,62 +61,89 @@ namespace libp2p::connection {
 
   void NoiseConnection::read(gsl::span<uint8_t> out, size_t bytes,
                              libp2p::basic::Reader::ReadCallbackFunc cb) {
+    OperationContext context{.bytes_served = 0,
+                             .total_bytes = bytes,
+                             .write_buffer = write_buffers_.end()};
+    read(out, bytes, context, std::move(cb));
+  }
+
+  void NoiseConnection::read(gsl::span<uint8_t> out, size_t bytes,
+                             OperationContext ctx, ReadCallbackFunc cb) {
     size_t out_size{out.empty() ? 0u : static_cast<size_t>(out.size())};
     BOOST_ASSERT(out_size >= bytes);
-    if (bytes == 0) {
-      auto n{already_read_};
-      already_read_ = 0;
-      return cb(n);
+    if (0 == bytes) {
+      BOOST_ASSERT(ctx.bytes_served == ctx.total_bytes);
+      return cb(ctx.bytes_served);
     }
     readSome(out, bytes,
-             [self{shared_from_this()}, out, bytes,
-              cb{std::move(cb)}](auto _n) mutable {
+             [self{shared_from_this()}, out, bytes, cb{std::move(cb)},
+              ctx](auto _n) mutable {
                OUTCOME_CB(n, _n);
-               self->already_read_ += n;
-               self->read(out.subspan(n), bytes - n, std::move(cb));
+               ctx.bytes_served += n;
+               self->read(out.subspan(n), bytes - n, ctx, std::move(cb));
              });
   }
 
   void NoiseConnection::readSome(gsl::span<uint8_t> out, size_t bytes,
                                  libp2p::basic::Reader::ReadCallbackFunc cb) {
+    OperationContext context{.bytes_served = 0,
+                             .total_bytes = bytes,
+                             .write_buffer = write_buffers_.end()};
+    readSome(out, bytes, context, std::move(cb));
+  }
+
+  void NoiseConnection::readSome(gsl::span<uint8_t> out, size_t bytes,
+                                 OperationContext ctx, ReadCallbackFunc cb) {
     if (not frame_buffer_->empty()) {
       auto n{std::min(bytes, frame_buffer_->size())};
       auto begin{frame_buffer_->begin()};
-      auto end{begin + n};
+      auto end{begin + static_cast<int64_t>(n)};
       std::copy(begin, end, out.begin());
       frame_buffer_->erase(begin, end);
       return cb(n);
     }
-    framer_->read([self{shared_from_this()}, out, bytes,
-                   cb{std::move(cb)}](auto _data) mutable {
+    framer_->read([self{shared_from_this()}, out, bytes, cb{std::move(cb)},
+                   ctx](auto _data) mutable {
       OUTCOME_CB(data, _data);
       OUTCOME_CB(decrypted, self->decoder_cs_->decrypt({}, *data, {}));
       self->frame_buffer_->assign(decrypted.begin(), decrypted.end());
-      self->readSome(out, bytes, std::move(cb));
+      self->readSome(out, bytes, ctx, std::move(cb));
     });
   }
 
   void NoiseConnection::write(gsl::span<const uint8_t> in, size_t bytes,
                               libp2p::basic::Writer::WriteCallbackFunc cb) {
-    if (0 == plaintext_len_to_write_) {
-      plaintext_len_to_write_ = bytes;
-    }
-    if (bytes == 0) {
-      BOOST_ASSERT(already_wrote_ >= plaintext_len_to_write_);
-      auto n{plaintext_len_to_write_};
-      already_wrote_ = 0;
-      plaintext_len_to_write_ = 0;
-      return cb(n);
+    OperationContext context{.bytes_served = 0,
+                             .total_bytes = bytes,
+                             .write_buffer = write_buffers_.end()};
+    write(in, bytes, context, std::move(cb));
+  }
+
+  void NoiseConnection::write(gsl::span<const uint8_t> in, size_t bytes,
+                              NoiseConnection::OperationContext ctx,
+                              basic::Writer::WriteCallbackFunc cb) {
+    auto *self{this};  // for OUTCOME_CB
+    if (0 == bytes) {
+      BOOST_ASSERT(ctx.bytes_served >= ctx.total_bytes);
+      eraseWriteBuffer(ctx.write_buffer);
+      return cb(ctx.total_bytes);
     }
     auto n{std::min(bytes, security::noise::kMaxPlainText)};
     OUTCOME_CB(encrypted, encoder_cs_->encrypt({}, in.subspan(0, n), {}));
-    writing_ = std::move(encrypted);
-    framer_->write(writing_,
-                   [self{shared_from_this()}, in{in.subspan(n)},
-                    bytes{bytes - n}, cb{std::move(cb)}](auto _n) mutable {
+    if (write_buffers_.end() == ctx.write_buffer) {
+      constexpr auto dummy_size = 1;
+      constexpr auto dummy_value = 0x0;
+      ctx.write_buffer =
+          write_buffers_.emplace(write_buffers_.end(), dummy_size, dummy_value);
+    }
+    ctx.write_buffer->swap(encrypted);
+    framer_->write(
+        *ctx.write_buffer,
+        [self{shared_from_this()}, in{in.subspan(static_cast<int64_t>(n))},
+         bytes{bytes - n}, cb{std::move(cb)}, ctx](auto _n) mutable {
           OUTCOME_CB(n, _n);
-                     self->already_wrote_ += n;
-                     self->write(in, bytes, std::move(cb));
+          ctx.bytes_served += n;
+          self->write(in, bytes, ctx, std::move(cb));
         });
   }
 
@@ -165,4 +190,12 @@ namespace libp2p::connection {
       const {
     return remote_;
   }
+
+  void NoiseConnection::eraseWriteBuffer(BufferList::iterator &iterator) {
+    if (write_buffers_.end() == iterator) {
+      return;
+    }
+    write_buffers_.erase(iterator);
+    iterator = write_buffers_.end();
+  }
 }  // namespace libp2p::connection