fix: limit number of bad pieces to accept from a webseed (#6875)

* perf: initialise blame bitfield by piece count

* refactor: set blame for all peers

* refactor: make `tr_swarm::add_strike()` work for all peers

* refactor: move `tr_peer::do_purge` to `tr_peerMsgs`

* fix: limit number of bad pieces to accept from a webseed

libtransmission/peer-common.h

@@ -215,6 +215,8 @@ struct tr_peer
     {
     }
 
+    virtual void ban() = 0;
+
     tr_session* const session;
 
     tr_swarm* const swarm;
@@ -229,9 +231,6 @@ struct tr_peer
     // whether or not this peer sent us any given block
     tr_bitfield blame;
 
-    // whether or not we should free this peer soon.
-    bool do_purge = false;
-
     // how many bad pieces this piece has contributed to
     uint8_t strikes = 0;
 

libtransmission/peer-mgr.cc

@@ -560,6 +560,26 @@ public:
             // not currently supported
             break;
 
+        case tr_peer_event::Type::Error:
+            if (event.err == ERANGE || event.err == EMSGSIZE || event.err == ENOTCONN)
+            {
+                // some protocol error from the peer
+                msgs->disconnect_soon();
+                tr_logAddDebugSwarm(
+                    s,
+                    fmt::format(
+                        "setting {} is_disconnecting_ flag because we got [({}) {}]",
+                        msgs->display_name(),
+                        event.err,
+                        tr_strerror(event.err)));
+            }
+            else
+            {
+                tr_logAddDebugSwarm(s, fmt::format("unhandled error: ({}) {}", event.err, tr_strerror(event.err)));
+            }
+
+            break;
+
         default:
             peer_callback_common(msgs, event, s);
             break;
@@ -619,7 +639,7 @@ private:
         stats.active_webseed_count = 0;
     }
 
-    void add_strike(tr_peerMsgs* peer) const
+    void add_strike(tr_peer* peer) const
     {
         tr_logAddTraceSwarm(
             this,
@@ -627,8 +647,7 @@ private:
 
         if (++peer->strikes >= MaxBadPiecesPerPeer)
         {
-            peer->peer_info->ban();
+            peer->ban();
-            peer->do_purge = true;
             tr_logAddTraceSwarm(this, fmt::format("banning peer {}", peer->display_name()));
         }
     }
@@ -716,9 +735,7 @@ private:
 
     void on_got_bad_piece(tr_piece_index_t piece)
     {
-        auto const byte_count = tor->piece_size(piece);
+        auto const maybe_add_strike = [this, piece](tr_peer* const peer)
-
-        for (auto* const peer : peers)
         {
             if (peer->blame.test(piece))
             {
@@ -731,6 +748,18 @@ private:
                         peer->strikes + 1));
                 add_strike(peer);
             }
+        };
+
+        auto const byte_count = tor->piece_size(piece);
+
+        for (auto* const peer : peers)
+        {
+            maybe_add_strike(peer);
+        }
+
+        for (auto& webseed : webseeds)
+        {
+            maybe_add_strike(webseed.get());
         }
 
         tr_announcerAddBytes(tor, TR_ANN_CORRUPT, byte_count);
@@ -791,32 +820,13 @@ private:
                 auto const loc = tor->piece_loc(event.pieceIndex, event.offset);
                 s->cancel_all_requests_for_block(loc.block, peer);
                 peer->blocks_sent_to_client.add(tr_time(), 1);
+                peer->blame.set(loc.piece);
                 tor->on_block_received(loc.block);
                 s->got_block.emit(tor, event.pieceIndex, loc.block);
             }
 
             break;
 
-        case tr_peer_event::Type::Error:
-            if (event.err == ERANGE || event.err == EMSGSIZE || event.err == ENOTCONN)
-            {
-                /* some protocol error from the peer */
-                peer->do_purge = true;
-                tr_logAddDebugSwarm(
-                    s,
-                    fmt::format(
-                        "setting {} do_purge flag because we got [({}) {}]",
-                        peer->display_name(),
-                        event.err,
-                        tr_strerror(event.err)));
-            }
-            else
-            {
-                tr_logAddDebugSwarm(s, fmt::format("unhandled error: ({}) {}", event.err, tr_strerror(event.err)));
-            }
-
-            break;
-
         default:
             TR_ASSERT_MSG(false, "This should be unreachable code");
             break;
@@ -890,13 +900,13 @@ EXIT:
                 std::end(peers),
                 [&info_that](tr_peerMsgs const* const peer) { return peer->peer_info == info_that; });
             TR_ASSERT(it != std::end(peers));
-            (*it)->do_purge = true;
+            (*it)->disconnect_soon();
 
             return false;
         }
 
         info_that->merge(*info_this);
-        msgs->do_purge = true;
+        msgs->disconnect_soon();
         stats.known_peer_from_count[info_this->from_first()] -= connectable_pool.erase(info_this->listen_socket_address());
 
         return true;
@@ -1152,7 +1162,7 @@ private:
 tr_peer::tr_peer(tr_torrent const& tor)
     : session{ tor.session }
     , swarm{ tor.swarm }
-    , blame{ tor.block_count() }
+    , blame{ tor.piece_count() }
 {
 }
 
@@ -2104,9 +2114,9 @@ auto constexpr MaxUploadIdleSecs = time_t{ 60 * 5 };
 [[nodiscard]] bool shouldPeerBeClosed(tr_swarm const* s, tr_peerMsgs const* peer, size_t peer_count, time_t const now)
 {
     /* if it's marked for purging, close it */
-    if (peer->do_purge)
+    if (peer->is_disconnecting())
     {
-        tr_logAddTraceSwarm(s, fmt::format("purging peer {} because its do_purge flag is set", peer->display_name()));
+        tr_logAddTraceSwarm(s, fmt::format("purging peer {} because its is_disconnecting_ flag is set", peer->display_name()));
         return true;
     }
 
@@ -2157,9 +2167,9 @@ constexpr struct
 {
     [[nodiscard]] static int compare(tr_peerMsgs const* a, tr_peerMsgs const* b) // <=>
     {
-        if (a->do_purge != b->do_purge)
+        if (a->is_disconnecting() != b->is_disconnecting())
         {
-            return a->do_purge ? 1 : -1;
+            return a->is_disconnecting() ? 1 : -1;
         }
 
         return -a->peer_info->compare_by_piece_data_time(*b->peer_info);

libtransmission/peer-msgs.cc

@@ -397,6 +397,12 @@ public:
 
     // ---
 
+    void ban() override
+    {
+        peer_info->ban();
+        disconnect_soon();
+    }
+
     void on_torrent_got_metainfo() noexcept override
     {
         update_active();
@@ -1682,7 +1688,6 @@ int tr_peerMsgsImpl::client_got_block(std::unique_ptr<Cache::BlockData> block_da
         return err;
     }
 
-    blame.set(loc.piece);
     publish(tr_peer_event::GotBlock(tor_.block_info(), block));
 
     return 0;

libtransmission/peer-msgs.h

@@ -95,6 +95,16 @@ public:
         return is_active_[direction];
     }
 
+    [[nodiscard]] constexpr auto is_disconnecting() const noexcept
+    {
+        return is_disconnecting_;
+    }
+
+    constexpr void disconnect_soon() noexcept
+    {
+        is_disconnecting_ = true;
+    }
+
     [[nodiscard]] virtual tr_socket_address socket_address() const = 0;
 
     virtual void set_choke(bool peer_is_choked) = 0;
@@ -172,6 +182,9 @@ private:
 
     // whether or not the peer has indicated it will download from us
     bool peer_is_interested_ = false;
+
+    // whether or not we should free this peer soon.
+    bool is_disconnecting_ = false;
 };
 
 /* @} */

libtransmission/webseed.cc

@@ -201,9 +201,7 @@ public:
 
     ~tr_webseed_impl() override
     {
-        // flag all the pending tasks as dead
+        stop();
-        std::for_each(std::begin(tasks), std::end(tasks), [](auto* task) { task->dead = true; });
-        tasks.clear();
     }
 
     [[nodiscard]] Speed get_piece_speed(uint64_t now, tr_direction dir) const override
@@ -248,6 +246,21 @@ public:
         return have_;
     }
 
+    void stop()
+    {
+        idle_timer_->stop();
+
+        // flag all the pending tasks as dead
+        std::for_each(std::begin(tasks), std::end(tasks), [](auto* task) { task->dead = true; });
+        tasks.clear();
+    }
+
+    void ban() override
+    {
+        is_banned_ = true;
+        stop();
+    }
+
     void got_piece_data(uint32_t n_bytes)
     {
         bandwidth_.notify_bandwidth_consumed(TR_DOWN, n_bytes, true, tr_time_msec());
@@ -265,7 +278,7 @@ public:
 
     void request_blocks(tr_block_span_t const* block_spans, size_t n_spans) override
     {
-        if (!tor.is_running() || tor.is_done())
+        if (is_banned_ || !tor.is_running() || tor.is_done())
         {
             return;
         }
@@ -282,6 +295,11 @@ public:
 
     void on_idle()
     {
+        if (is_banned_)
+        {
+            return;
+        }
+
         auto const [max_spans, max_blocks] = max_available_reqs();
         if (max_spans == 0 || max_blocks == 0)
         {
@@ -344,6 +362,8 @@ private:
 
     tr_peer_callback_webseed const callback_;
     void* const callback_data_;
+
+    bool is_banned_ = false;
 };
 
 // ---