Introduce basic UDP packet filtering scheme (#23957)

* Introduce basic UDP packet filtering scheme

- Embedded class devices on Wi-Fi networks can see large number of
  mDNS packets from normal network. If these arrive while device
  is processing CASE/PASE or any other long running system activity,
  the amount of queuing can overrun the packet buffer pools or
  event queues, which can cause very bad outcomes.
- Filtering of packets can always be done by a product deep in its
  network stack. However, allowing some level of basic filtering
  for UDP packets (that cover most of the traffic that could be
  problematic and need queuing) for at least lwIP devices allows
  sharing code for filters that may be smarter.

Follow-up PR will show an example hook-up to ESP32

Issue #23258
Issue #23180

This PR:
- Adds generic EndpointQueueFilter.h, which is portable/testable
  on its own.
- Adds hook-ups to UDPEndPointImplLwIP to run an EndpointQueueFilter
  on incoming packets

Testing done:
- Full unit test suite for the filtering framework and basic
  filter.

* Address review comments

src/inet/BUILD.gn

@@ -122,6 +122,8 @@ static_library("inet") {
 
   if (chip_inet_config_enable_udp_endpoint) {
     sources += [
+      "BasicPacketFilters.h",
+      "EndpointQueueFilter.h",
       "UDPEndPoint.cpp",
       "UDPEndPoint.h",
       "UDPEndPointImpl${chip_system_config_inet}.cpp",

src/inet/BasicPacketFilters.h

@@ -0,0 +1,185 @@
+/*
+ *    Copyright (c) 2022 Project CHIP Authors
+ *    All rights reserved.
+ *
+ *    Licensed under the Apache License, Version 2.0 (the "License");
+ *    you may not use this file except in compliance with the License.
+ *    You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *    Unless required by applicable law or agreed to in writing, software
+ *    distributed under the License is distributed on an "AS IS" BASIS,
+ *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *    See the License for the specific language governing permissions and
+ *    limitations under the License.
+ */
+
+#pragma once
+
+#include <atomic>
+#include <inet/EndpointQueueFilter.h>
+#include <inet/IPPacketInfo.h>
+#include <lib/support/CodeUtils.h>
+#include <system/SystemPacketBuffer.h>
+
+namespace chip {
+namespace Inet {
+
+/**
+ * @brief Basic filter that counts how many pending (not yet dequeued) packets
+ *        are accumulated that match a predicate function, and drops those that
+ *        would cause crossing of the threshold.
+ */
+class DropIfTooManyQueuedPacketsFilter : public chip::Inet::EndpointQueueFilter
+{
+public:
+    typedef bool (*PacketMatchPredicateFunc)(void * context, const void * endpoint, const chip::Inet::IPPacketInfo & pktInfo,
+                                             const chip::System::PacketBufferHandle & pktPayload);
+
+    /**
+     * @brief Initialize the packet filter with a starting limit
+     *
+     * @param maxAllowedQueuedPackets - max number of pending-in-queue not yet processed predicate-matching packets
+     */
+    DropIfTooManyQueuedPacketsFilter(size_t maxAllowedQueuedPackets) : mMaxAllowedQueuedPackets(maxAllowedQueuedPackets) {}
+
+    /**
+     * @brief Set the predicate to use for filtering
+     *
+     * @warning DO NOT modify at runtime while the filter is being called. If you do so, the queue accounting could
+     *          get out of sync, and cause the filtering to fail to properly work.
+     *
+     * @param predicateFunc - Predicate function to apply. If nullptr, no filtering will take place
+     * @param context - Pointer to predicate-specific context that will be provided to predicate at every call. May be nullptr.
+     */
+    void SetPredicate(PacketMatchPredicateFunc predicateFunc, void * context)
+    {
+        mPredicate = predicateFunc;
+        mContext   = context;
+    }
+
+    /**
+     * @brief Set the ceiling for max allowed packets queued up that matched the predicate.
+     *
+     * @note Changing this at runtime while packets are coming only affects future dropping, and
+     *       does not remove packets from the queue if the limit is lowered below the currently-in-queue
+     *       count.
+     *
+     * @param maxAllowedQueuedPackets - number of packets currently pending allowed.
+     */
+    void SetMaxQueuedPacketsLimit(int maxAllowedQueuedPackets) { mMaxAllowedQueuedPackets.store(maxAllowedQueuedPackets); }
+
+    /**
+     * @return the total number of packets dropped so far by the filter
+     */
+    size_t GetNumDroppedPackets() const { return mNumDroppedPackets.load(); }
+
+    /**
+     * @brief Reset the counter of dropped packets.
+     */
+    void ClearNumDroppedPackets() { mNumDroppedPackets.store(0); }
+
+    /**
+     * @brief Method called when a packet is dropped due to high watermark getting reached, based on predicate.
+     *
+     * Subclasses may use this to implement additional behavior or diagnostics.
+     *
+     * This is called once for every dropped packet. If there is no filter predicate, this is not called.
+     *
+     * @param endpoint - pointer to endpoint instance (platform-dependent, which is why it's void)
+     * @param pktInfo - info about source/dest of packet
+     * @param pktPayload - payload content of packet
+     */
+    virtual void OnDropped(const void * endpoint, const chip::Inet::IPPacketInfo & pktInfo,
+                           const chip::System::PacketBufferHandle & pktPayload)
+    {}
+
+    /**
+     * @brief Method called whenever queue of accumulated packets is now empty, based on predicate.
+     *
+     * Subclasses may use this to implement additional behavior or diagnostics.
+     *
+     * This is possibly called repeatedly in a row, if the queue actually never gets above one.
+     *
+     * This is only called for packets that had matched the filtering rule, where they had
+     * been explicitly allowed in the past. If there is no filter predicate, this is not called.
+     *
+     * @param endpoint - pointer to endpoint instance (platform-dependent, which is why it's void)
+     * @param pktInfo - info about source/dest of packet
+     * @param pktPayload - payload content of packet
+     */
+    virtual void OnLastMatchDequeued(const void * endpoint, const chip::Inet::IPPacketInfo & pktInfo,
+                                     const chip::System::PacketBufferHandle & pktPayload)
+    {}
+
+    /**
+     * @brief Implementation of filtering before queueing that applies the predicate.
+     *
+     * See base class for arguments
+     */
+    FilterOutcome FilterBeforeEnqueue(const void * endpoint, const chip::Inet::IPPacketInfo & pktInfo,
+                                      const chip::System::PacketBufferHandle & pktPayload) override
+    {
+        // WARNING: This is likely called in a different context than `FilterAfterDequeue`. We use an atomic for the counter.
+
+        // Non-matching is never accounted, always allowed. Lack of predicate is equivalent to non-matching.
+        if ((mPredicate == nullptr) || !mPredicate(mContext, endpoint, pktInfo, pktPayload))
+        {
+            return FilterOutcome::kAllowPacket;
+        }
+
+        if (mNumQueuedPackets.load() >= mMaxAllowedQueuedPackets)
+        {
+            ++mNumDroppedPackets;
+            OnDropped(endpoint, pktInfo, pktPayload);
+            return FilterOutcome::kDropPacket;
+        }
+
+        ++mNumQueuedPackets;
+
+        return FilterOutcome::kAllowPacket;
+    }
+
+    /**
+     * @brief Implementation of filtering after dequeueing that applies the predicate.
+     *
+     * See base class for arguments
+     */
+    FilterOutcome FilterAfterDequeue(const void * endpoint, const chip::Inet::IPPacketInfo & pktInfo,
+                                     const chip::System::PacketBufferHandle & pktPayload) override
+    {
+        // WARNING: This is likely called in a different context than `FilterBeforeEnqueue`. We use an atomic for the counter.
+        // NOTE: This is always called from Matter platform event loop
+
+        // Non-matching is never accounted, always allowed. Lack of predicate is equivalent to non-matching.
+        if ((mPredicate == nullptr) || !mPredicate(mContext, endpoint, pktInfo, pktPayload))
+        {
+            return FilterOutcome::kAllowPacket;
+        }
+
+        --mNumQueuedPackets;
+        int numQueuedPackets = mNumQueuedPackets.load();
+        if (numQueuedPackets == 0)
+        {
+            OnLastMatchDequeued(endpoint, pktInfo, pktPayload);
+        }
+
+        // If we ever go negative, we have mismatch ingress/egress filter via predicate and
+        // device may eventually starve.
+        VerifyOrDie(numQueuedPackets >= 0);
+
+        // We always allow the packet and just do accounting, since all dropping is prior to queue entry.
+        return FilterOutcome::kAllowPacket;
+    }
+
+protected:
+    PacketMatchPredicateFunc mPredicate = nullptr;
+    void * mContext                     = nullptr;
+    std::atomic_int mNumQueuedPackets{ 0 };
+    std::atomic_int mMaxAllowedQueuedPackets{ 0 };
+    std::atomic_size_t mNumDroppedPackets{ 0u };
+};
+
+} // namespace Inet
+} // namespace chip

src/inet/EndpointQueueFilter.h

@@ -0,0 +1,88 @@
+/*
+ *    Copyright (c) 2022 Project CHIP Authors
+ *    All rights reserved.
+ *
+ *    Licensed under the Apache License, Version 2.0 (the "License");
+ *    you may not use this file except in compliance with the License.
+ *    You may obtain a copy of the License at
+ *
+ *        http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *    Unless required by applicable law or agreed to in writing, software
+ *    distributed under the License is distributed on an "AS IS" BASIS,
+ *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *    See the License for the specific language governing permissions and
+ *    limitations under the License.
+ */
+
+#pragma once
+
+#include <inet/IPPacketInfo.h>
+#include <system/SystemPacketBuffer.h>
+
+namespace chip {
+namespace Inet {
+
+/**
+ * @brief Filter for UDP Packets going into and out of UDPEndPoint queue.
+ *
+ * NOTE: This is only used by some low-level implementations of UDPEndPoint
+ */
+class EndpointQueueFilter
+{
+public:
+    enum FilterOutcome
+    {
+        kAllowPacket = 0,
+        kDropPacket  = 1,
+    };
+
+    virtual ~EndpointQueueFilter() {}
+
+    /**
+     * @brief Run filter prior to inserting in queue.
+     *
+     * If filter returns `kAllowPacket`, packet will be enqueued, and `FilterAfterDequeue` will
+     * be called when it gets dequeued. If filter returns `kDropPacket`, packet will be dropped
+     * rather than enqueued and `FilterAfterDequeue` method will not be called.
+     *
+     * WARNING: This is likely called from non-Matter-eventloop context, from network layer code.
+     *          Be extremely careful about accessing any system data which may belong to Matter
+     *          stack from this method.
+     *
+     * @param endpoint - pointer to endpoint instance (platform-dependent, which is why it's void)
+     * @param pktInfo - info about source/dest of packet
+     * @param pktPayload - payload content of packet
+     *
+     * @return kAllowPacket to allow packet to enqueue or kDropPacket to drop the packet
+     */
+    virtual FilterOutcome FilterBeforeEnqueue(const void * endpoint, const IPPacketInfo & pktInfo,
+                                              const chip::System::PacketBufferHandle & pktPayload)
+    {
+        return FilterOutcome::kAllowPacket;
+    }
+
+    /**
+     * @brief Run filter after dequeuing, prior to processing.
+     *
+     * If filter returns `kAllowPacket`, packet will be processed after dequeuing. If filter returns
+     * `kDropPacket`, packet will be dropped and not processed, even though it was dequeued.
+     *
+     * WARNING: This is called from Matter thread context. Be extremely careful about accessing any
+     *          data which may belong to different threads from this method.
+     *
+     * @param endpoint - pointer to endpoint instance (platform-dependent, which is why it's void)
+     * @param pktInfo - info about source/dest of packet
+     * @param pktPayload - payload content of packet
+     *
+     * @return kAllowPacket to allow packet to be processed or kDropPacket to drop the packet
+     */
+    virtual FilterOutcome FilterAfterDequeue(const void * endpoint, const IPPacketInfo & pktInfo,
+                                             const chip::System::PacketBufferHandle & pktPayload)
+    {
+        return FilterOutcome::kAllowPacket;
+    }
+};
+
+} // namespace Inet
+} // namespace chip

src/inet/UDPEndPointImplLwIP.cpp

@@ -78,6 +78,8 @@ struct Deleter<struct pbuf>
 namespace chip {
 namespace Inet {
 
+EndpointQueueFilter * UDPEndPointImplLwIP::sQueueFilter = nullptr;
+
 CHIP_ERROR UDPEndPointImplLwIP::BindImpl(IPAddressType addressType, const IPAddress & address, uint16_t port,
                                          InterfaceId interfaceId)
 {
@@ -287,7 +289,16 @@ void UDPEndPointImplLwIP::Free()
 
 void UDPEndPointImplLwIP::HandleDataReceived(System::PacketBufferHandle && msg, IPPacketInfo * pktInfo)
 {
-    if ((mState == State::kListening) && (OnMessageReceived != nullptr))
+    // Process packet filter if needed. May cause packet to get dropped before processing.
+    bool dropPacket = false;
+    if ((pktInfo != nullptr) && (sQueueFilter != nullptr))
+    {
+        auto outcome = sQueueFilter->FilterAfterDequeue(this, *pktInfo, msg);
+        dropPacket   = (outcome == EndpointQueueFilter::FilterOutcome::kDropPacket);
+    }
+
+    // Process actual packet if allowed
+    if ((mState == State::kListening) && (OnMessageReceived != nullptr) && !dropPacket)
     {
         if (pktInfo != nullptr)
         {
@@ -424,14 +435,28 @@ void UDPEndPointImplLwIP::LwIPReceiveUDPMessage(void * arg, struct udp_pcb * pcb
     pktInfo->SrcPort     = port;
     pktInfo->DestPort    = pcb->local_port;
 
+    auto filterOutcome = EndpointQueueFilter::FilterOutcome::kAllowPacket;
+    if (sQueueFilter != nullptr)
+    {
+        filterOutcome = sQueueFilter->FilterBeforeEnqueue(ep, *(pktInfo.get()), buf);
+    }
+
+    if (filterOutcome != EndpointQueueFilter::FilterOutcome::kAllowPacket)
+    {
+        // Logging, if any, should be at the choice of the filter impl at time of filtering.
+        return;
+    }
+
     // Increase mDelayReleaseCount to delay release of this UDP EndPoint while the HandleDataReceived call is
     // pending on it.
     ep->mDelayReleaseCount++;
 
     CHIP_ERROR err = ep->GetSystemLayer().ScheduleLambda(
         [ep, p = System::LwIPPacketBufferView::UnsafeGetLwIPpbuf(buf), pktInfo = pktInfo.get()] {
             ep->mDelayReleaseCount--;
-            ep->HandleDataReceived(System::PacketBufferHandle::Adopt(p), pktInfo);
+
+            auto handle = System::PacketBufferHandle::Adopt(p);
+            ep->HandleDataReceived(std::move(handle), pktInfo);
         });
 
     if (err == CHIP_NO_ERROR)
@@ -443,6 +468,14 @@ void UDPEndPointImplLwIP::LwIPReceiveUDPMessage(void * arg, struct udp_pcb * pcb
     }
     else
     {
+        // On failure to enqueue the processing, we have to tell the filter that
+        // the packet is basically dequeued, if it tries to keep track of the lifecycle.
+        if (sQueueFilter != nullptr)
+        {
+            (void) sQueueFilter->FilterAfterDequeue(ep, *(pktInfo.get()), buf);
+            ChipLogError(Inet, "Dequeue ERROR err = %" CHIP_ERROR_FORMAT, err.Format());
+        }
+
         ep->mDelayReleaseCount--;
     }
 }

src/inet/UDPEndPointImplLwIP.h

@@ -24,6 +24,7 @@
 #pragma once
 
 #include <inet/EndPointStateLwIP.h>
+#include <inet/EndpointQueueFilter.h>
 #include <inet/UDPEndPoint.h>
 
 namespace chip {
@@ -40,6 +41,20 @@ class UDPEndPointImplLwIP : public UDPEndPoint, public EndPointStateLwIP
     uint16_t GetBoundPort() const override;
     void Free() override;
 
+    /**
+     * @brief Set the queue filter for all UDP endpoints
+     *
+     * Responsibility is on the caller to avoid changing the filter while packets are being
+     * processed. Setting the queue filter to `nullptr` disables the filtering.
+     *
+     * NOTE: There is only one EndpointQueueFilter instance settable. However it's possible
+     *       to create an instance of EndpointQueueFilter that combines several other
+     *       EndpointQueueFilter by composition to achieve the effect of multiple filters.
+     *
+     * @param queueFilter - queue filter instance to set, owned by caller
+     */
+    static void SetQueueFilter(EndpointQueueFilter * queueFilter) { sQueueFilter = queueFilter; }
+
 private:
     // UDPEndPoint overrides.
 #if INET_CONFIG_ENABLE_IPV4
@@ -62,6 +77,8 @@ class UDPEndPointImplLwIP : public UDPEndPoint, public EndPointStateLwIP
 
     udp_pcb * mUDP; // LwIP User datagram protocol (UDP) control block.
     std::atomic_int mDelayReleaseCount{ 0 };
+
+    static EndpointQueueFilter * sQueueFilter;
 };
 
 using UDPEndPointImpl = UDPEndPointImplLwIP;