Implemented filtering options for eth_subscribe with logs #27842

browser/brave_wallet/ethereum_provider_impl_unittest.cc

@@ -125,6 +125,15 @@ std::vector<uint8_t> DecodeHexHash(const std::string& hash_hex) {
   return hash;
 }
 
+absl::optional<base::Value> ToValue(const network::ResourceRequest& request) {
+  base::StringPiece request_string(request.request_body->elements()
+                                       ->at(0)
+                                       .As<network::DataElementBytes>()
+                                       .AsStringPiece());
+  return base::JSONReader::Read(request_string,
+                                base::JSONParserOptions::JSON_PARSE_RFC);
+}
+
 }  // namespace
 
 class TestEventsListener : public brave_wallet::mojom::EventsListener {
@@ -2171,6 +2180,70 @@ TEST_F(EthereumProviderImplUnitTest, EthSubscribeLogs) {
   EXPECT_FALSE(provider_->eth_logs_tracker_.IsRunning());
 }
 
+TEST_F(EthereumProviderImplUnitTest, EthSubscribeLogsFiltered) {
+  CreateWallet();
+  url_loader_factory_.SetInterceptor(
+      base::BindLambdaForTesting([&](const network::ResourceRequest& request) {
+        url_loader_factory_.ClearResponses();
+
+        std::string header_value;
+        EXPECT_TRUE(request.headers.GetHeader("X-Eth-Method", &header_value));
+
+        if (header_value == "eth_getLogs") {
+          const absl::optional<base::Value> req_body_payload =
+              base::JSONReader::Read(
+                  R"({"id":1,"jsonrpc":"2.0","method":"eth_getLogs","params":
+[{"address":["0x1111", "0x1112"],"fromBlock":"0x2211","toBlock":"0xab65",
+"topics":["0x2edc","0xb832","0x8dc8"]}]})",
+                  base::JSON_PARSE_CHROMIUM_EXTENSIONS |
+                      base::JSONParserOptions::JSON_PARSE_RFC);
+
+          const auto payload = ToValue(request);
+          EXPECT_EQ(*payload, req_body_payload.value());
+        }
+        url_loader_factory_.AddResponse(
+            request.url.spec(),
+            R"({"id":1,"jsonrpc":"2.0","result":[{"address":"0x91",
+                      "blockHash":"0xe8","blockNumber":"0x10","data":"0x0067",
+                      "logIndex":"0x0","removed":false,
+                      "topics":["0x4b","0x06e","0x085"],
+                      "transactionHash":"0x22f7","transactionIndex":"0x0"}]})");
+      }));
+
+  // Logs subscription with parameters
+  std::string request_payload_json =
+      R"({"id":1,"jsonrpc:": "2.0","method":"eth_subscribe",
+  "params": ["logs", {"address": ["0x1111", "0x1112"], "fromBlock": "0x2211",
+  "toBlock": "0xab65",  "topics":  ["0x2edc", "0xb832", "0x8dc8"]}]})";
+  absl::optional<base::Value> request_payload = base::JSONReader::Read(
+      request_payload_json, base::JSON_PARSE_CHROMIUM_EXTENSIONS |
+                                base::JSONParserOptions::JSON_PARSE_RFC);
+  std::string error_message;
+  auto response = CommonRequestOrSendAsync(request_payload.value());
+  EXPECT_EQ(response.first, false);
+  EXPECT_TRUE(response.second.is_string());
+  std::string subscription = *response.second.GetIfString();
+  browser_task_environment_.FastForwardBy(
+      base::Seconds(kLogTrackerDefaultTimeInSeconds));
+  EXPECT_TRUE(observer_->MessageEventFired());
+  base::Value rv = observer_->GetLastMessage();
+  ASSERT_TRUE(rv.is_dict());
+
+  std::string* address = rv.GetDict().FindString("address");
+  EXPECT_EQ(*address, "0x91");
+
+  // The first unsubscribe should not stop the block tracker
+  request_payload_json = base::StringPrintf(R"({"id":1,"jsonrpc:": "2.0",
+                              "method":"eth_unsubscribe",
+                              "params": ["%s"]})",
+                                            subscription.c_str());
+  request_payload = base::JSONReader::Read(
+      request_payload_json, base::JSON_PARSE_CHROMIUM_EXTENSIONS |
+                                base::JSONParserOptions::JSON_PARSE_RFC);
+  response = CommonRequestOrSendAsync(request_payload.value());
+  EXPECT_FALSE(provider_->eth_logs_tracker_.IsRunning());
+}
+
 TEST_F(EthereumProviderImplUnitTest, Web3ClientVersion) {
   std::string expected_version = base::StringPrintf(
       "BraveWallet/v%s", version_info::GetBraveChromiumVersionNumber().c_str());

components/brave_wallet/browser/asset_discovery_manager.cc

@@ -300,9 +300,13 @@ void AssetDiscoveryManager::OnGetEthTokenRegistry(
                                  weak_ptr_factory_.GetWeakPtr(),
                                  base::OwnedRef(std::move(tokens_to_search)),
                                  triggered_by_accounts_added, chain_id);
-  json_rpc_service_->EthGetLogs(chain_id, from_block, to_block,
-                                std::move(contract_addresses_to_search),
-                                std::move(topics), std::move(callback));
+  base::Value::Dict filtering;
+  filtering.Set("fromBlock", from_block);
+  filtering.Set("toBlock", to_block);
+  filtering.Set("address", std::move(contract_addresses_to_search));
+  filtering.Set("topics", std::move(topics));
+  json_rpc_service_->EthGetLogs(chain_id, std::move(filtering),
+                                std::move(callback));
 }
 
 void AssetDiscoveryManager::OnGetTokenTransferLogs(

components/brave_wallet/browser/eth_logs_tracker.cc

@@ -6,7 +6,7 @@
 #include "brave/components/brave_wallet/browser/eth_logs_tracker.h"
 
 #include <string>
-#include <vector>
+#include <utility>
 
 namespace brave_wallet {
 
@@ -31,6 +31,16 @@ bool EthLogsTracker::IsRunning() const {
   return timer_.IsRunning();
 }
 
+void EthLogsTracker::AddSubscriber(const std::string& subscription_id,
+                                   base::Value::Dict filter) {
+  eth_logs_subscription_info_.insert(std::pair<std::string, base::Value::Dict>(
+      subscription_id, std::move(filter)));
+}
+
+void EthLogsTracker::RemoveSubscriber(const std::string& subscription_id) {
+  eth_logs_subscription_info_.erase(subscription_id);
+}
+
 void EthLogsTracker::AddObserver(EthLogsTracker::Observer* observer) {
   observers_.AddObserver(observer);
 }
@@ -42,18 +52,22 @@ void EthLogsTracker::RemoveObserver(EthLogsTracker::Observer* observer) {
 void EthLogsTracker::GetLogs() {
   const auto chain_id = json_rpc_service_->GetChainId(mojom::CoinType::ETH);
 
-  json_rpc_service_->EthGetLogs(
-      chain_id, {}, {}, {}, {},
-      base::BindOnce(&EthLogsTracker::OnGetLogs, weak_factory_.GetWeakPtr()));
+  for (auto const& esi : std::as_const(eth_logs_subscription_info_)) {
+    json_rpc_service_->EthGetLogs(
+        chain_id, esi.second.Clone(),
+        base::BindOnce(&EthLogsTracker::OnGetLogs, weak_factory_.GetWeakPtr(),
+                       esi.first));
+  }
 }
 
-void EthLogsTracker::OnGetLogs([[maybe_unused]] const std::vector<Log>& logs,
+void EthLogsTracker::OnGetLogs(const std::string& subscription,
+                               [[maybe_unused]] const std::vector<Log>& logs,
                                base::Value rawlogs,
                                mojom::ProviderError error,
                                const std::string& error_message) {
   if (error == mojom::ProviderError::kSuccess && rawlogs.is_dict()) {
     for (auto& observer : observers_)
-      observer.OnLogsReceived(rawlogs.Clone());
+      observer.OnLogsReceived(subscription, rawlogs.Clone());
   } else {
     LOG(ERROR) << "OnGetLogs failed";
   }

components/brave_wallet/browser/eth_logs_tracker.h

@@ -6,6 +6,7 @@
 #ifndef BRAVE_COMPONENTS_BRAVE_WALLET_BROWSER_ETH_LOGS_TRACKER_H_
 #define BRAVE_COMPONENTS_BRAVE_WALLET_BROWSER_ETH_LOGS_TRACKER_H_
 
+#include <map>
 #include <string>
 #include <vector>
 
@@ -34,27 +35,35 @@ class EthLogsTracker {
 
   class Observer : public base::CheckedObserver {
    public:
-    virtual void OnLogsReceived(base::Value rawlogs) = 0;
+    virtual void OnLogsReceived(const std::string& subscription,
+                                base::Value rawlogs) = 0;
   };
 
   // If timer is already running, it will be replaced with new interval
   void Start(base::TimeDelta interval);
   void Stop();
   bool IsRunning() const;
 
+  void AddSubscriber(const std::string& subscription_id,
+                     base::Value::Dict filter);
+  void RemoveSubscriber(const std::string& subscription_id);
+
   void AddObserver(Observer* observer);
   void RemoveObserver(Observer* observer);
 
  private:
   void GetLogs();
-  void OnGetLogs(const std::vector<Log>& logs,
+  void OnGetLogs(const std::string& subscription,
+                 const std::vector<Log>& logs,
                  base::Value rawlogs,
                  mojom::ProviderError error,
                  const std::string& error_message);
 
   base::RepeatingTimer timer_;
   raw_ptr<JsonRpcService> json_rpc_service_ = nullptr;
 
+  std::map<std::string, base::Value::Dict> eth_logs_subscription_info_;
+
   base::ObserverList<Observer> observers_;
 
   base::WeakPtrFactory<EthLogsTracker> weak_factory_{this};

components/brave_wallet/browser/eth_requests.cc

@@ -336,27 +336,8 @@ std::string eth_getFilterLogs(const std::string& filter_id) {
   return GetJsonRpcString("eth_getFilterLogs", filter_id);
 }
 
-std::string eth_getLogs(const std::string& from_block_quantity_tag,
-                        const std::string& to_block_quantity_tag,
-                        base::Value::List addresses,
-                        base::Value::List topics,
-                        const std::string& block_hash) {
+std::string eth_getLogs(base::Value::Dict filter_options) {
   base::Value::List params;
-  base::Value::Dict filter_options;
-  // The `address` filter option accepts either a single address, or a list of
-  // addresses (See spec:
-  // https://ethereum.org/en/developers/docs/apis/json-rpc/#eth_getlogs). At
-  // time of writing Infura's documentation suggests they only support a single
-  // address, however we have verified they also support a list.
-  if (!addresses.empty()) {
-    filter_options.Set("address", std::move(addresses));
-  }
-  AddKeyIfNotEmpty(&filter_options, "fromBlock", from_block_quantity_tag);
-  AddKeyIfNotEmpty(&filter_options, "toBlock", to_block_quantity_tag);
-  if (!topics.empty()) {
-    filter_options.Set("topics", std::move(topics));
-  }
-  AddKeyIfNotEmpty(&filter_options, "blockhash", block_hash);
   params.Append(std::move(filter_options));
   base::Value::Dict dictionary =
       GetJsonRpcDictionary("eth_getLogs", std::move(params));

components/brave_wallet/browser/eth_requests.h

@@ -190,11 +190,7 @@ std::string eth_getFilterChanges(const std::string& filter_id);
 // Returns an array of all logs matching filter with given id.
 std::string eth_getFilterLogs(const std::string& filter_id);
 // Returns an array of all logs matching a given filter object.
-std::string eth_getLogs(const std::string& from_block_quantity_tag,
-                        const std::string& to_block_quantity_tag,
-                        base::Value::List addresses,
-                        base::Value::List topics,
-                        const std::string& block_hash);
+std::string eth_getLogs(base::Value::Dict filter_options);
 // Returns the hash of the current block, the seedHash, and the boundary
 // condition to be met (“target”).
 std::string eth_getWork();

components/brave_wallet/browser/eth_requests_unittest.cc

@@ -337,10 +337,17 @@ TEST(EthRequestUnitTest, eth_getLogs) {
 
   base::Value::List addresses;
   addresses.Append(base::Value("0x8888f1f195afa192cfee860698584c030f4c9db1"));
+
+  base::Value::Dict filtering;
+  filtering.Set("fromBlock", "0x1");
+  filtering.Set("toBlock", "0x2");
+  filtering.Set("address", std::move(addresses));
+  filtering.Set("topics", std::move(topics));
+  filtering.Set(
+      "blockhash",
+      "0xb903239f8543d04b5dc1ba6579132b143087c68db1b2168786408fcbce568238");
   ASSERT_EQ(
-      eth_getLogs(
-          "0x1", "0x2", std::move(addresses), std::move(topics),
-          "0xb903239f8543d04b5dc1ba6579132b143087c68db1b2168786408fcbce568238"),
+      eth_getLogs(std::move(filtering)),
       R"({"id":1,"jsonrpc":"2.0","method":"eth_getLogs","params":[{"address":["0x8888f1f195afa192cfee860698584c030f4c9db1"],"blockhash":"0xb903239f8543d04b5dc1ba6579132b143087c68db1b2168786408fcbce568238","fromBlock":"0x1","toBlock":"0x2","topics":["0x000000000000000000000000a94f5374fce5edbc8e2a8697c15331677e6ebf0b",["0x000000000000000000000000a94f5374fce5edbc8e2a8697c15331677e6ebf0b","0x0000000000000000000000000aff3454fce5edbc8cca8697c15331677e6ebccc"]]}]})");  // NOLINT
 }
 

components/brave_wallet/browser/ethereum_provider_impl.cc

@@ -578,9 +578,11 @@ void EthereumProviderImpl::RecoverAddress(const std::string& message,
                           false);
 }
 
-void EthereumProviderImpl::EthSubscribe(const std::string& event_type,
-                                        RequestCallback callback,
-                                        base::Value id) {
+void EthereumProviderImpl::EthSubscribe(
+    const std::string& event_type,
+    absl::optional<base::Value::Dict> filter,
+    RequestCallback callback,
+    base::Value id) {
   const auto generateHexBytes = [](std::vector<std::string>& subscriptions) {
     std::vector<uint8_t> bytes(16);
     crypto::RandBytes(&bytes.front(), bytes.size());
@@ -597,11 +599,15 @@ void EthereumProviderImpl::EthSubscribe(const std::string& event_type,
     }
     std::move(callback).Run(std::move(id), base::Value(std::get<1>(gen_res)),
                             false, "", false);
-  } else if (event_type == kEthSubscribeLogs) {
+  } else if (event_type == kEthSubscribeLogs && filter) {
     const auto gen_res = generateHexBytes(eth_log_subscriptions_);
+
     if (std::get<0>(gen_res)) {
       eth_logs_tracker_.Start(base::Seconds(kLogTrackerDefaultTimeInSeconds));
     }
+
+    eth_logs_tracker_.AddSubscriber(std::get<1>(gen_res), std::move(*filter));
+
     std::move(callback).Run(std::move(id), base::Value(std::get<1>(gen_res)),
                             false, "", false);
   } else {
@@ -641,9 +647,10 @@ bool EthereumProviderImpl::UnsubscribeBlockObserver(
 bool EthereumProviderImpl::UnsubscribeLogObserver(
     const std::string& subscription_id) {
   if (base::Erase(eth_log_subscriptions_, subscription_id)) {
-    if (eth_log_subscriptions_.empty())
+    eth_logs_tracker_.RemoveSubscriber(subscription_id);
+    if (eth_log_subscriptions_.empty()) {
       eth_logs_tracker_.Stop();
-
+    }
     return true;
   }
   return false;
@@ -1237,12 +1244,15 @@ void EthereumProviderImpl::CommonRequestOrSendAsync(base::ValueView input_value,
     Web3ClientVersion(std::move(callback), std::move(id));
   } else if (method == kEthSubscribe) {
     std::string event_type;
-    if (!ParseEthSubscribeParams(normalized_json_request, &event_type)) {
+    base::Value::Dict filter;
+    if (!ParseEthSubscribeParams(normalized_json_request, &event_type,
+                                 &filter)) {
       SendErrorOnRequest(error, error_message, std::move(callback),
                          std::move(id));
       return;
     }
-    EthSubscribe(event_type, std::move(callback), std::move(id));
+    EthSubscribe(event_type, std::move(filter), std::move(callback),
+                 std::move(id));
   } else if (method == kEthUnsubscribe) {
     std::string subscription_id;
     if (!ParseEthUnsubscribeParams(normalized_json_request, &subscription_id)) {
@@ -1705,7 +1715,8 @@ void EthereumProviderImpl::OnGetBlockByNumber(
 
 void EthereumProviderImpl::OnNewBlock(uint256_t block_num) {}
 
-void EthereumProviderImpl::OnLogsReceived(base::Value rawlogs) {
+void EthereumProviderImpl::OnLogsReceived(const std::string& subscription,
+                                          base::Value rawlogs) {
   if (!rawlogs.is_dict() || !events_listener_.is_bound()) {
     return;
   }
@@ -1718,8 +1729,7 @@ void EthereumProviderImpl::OnLogsReceived(base::Value rawlogs) {
   }
 
   for (auto& results_item : *results) {
-    for (const auto& subscription_id : eth_log_subscriptions_)
-      events_listener_->MessageEvent(subscription_id, results_item.Clone());
+    events_listener_->MessageEvent(subscription, results_item.Clone());
   }
 }
 

components/brave_wallet/browser/ethereum_provider_impl.h

@@ -89,6 +89,7 @@ class EthereumProviderImpl final
                       base::Value id);
 
   void EthSubscribe(const std::string& event_type,
+                    absl::optional<base::Value::Dict> filter,
                     RequestCallback callback,
                     base::Value id);
   void EthUnsubscribe(const std::string& subscription_id,
@@ -173,6 +174,8 @@ class EthereumProviderImpl final
                            RequestEthereumPermissionsWithAccounts);
   FRIEND_TEST_ALL_PREFIXES(EthereumProviderImplUnitTest, EthSubscribe);
   FRIEND_TEST_ALL_PREFIXES(EthereumProviderImplUnitTest, EthSubscribeLogs);
+  FRIEND_TEST_ALL_PREFIXES(EthereumProviderImplUnitTest,
+                           EthSubscribeLogsFiltered);
   friend class EthereumProviderImplUnitTest;
 
   // mojom::BraveWalletProvider:
@@ -378,7 +381,8 @@ class EthereumProviderImpl final
   bool UnsubscribeBlockObserver(const std::string& subscription_id);
 
   // EthLogsTracker::Observer:
-  void OnLogsReceived(base::Value rawlogs) override;
+  void OnLogsReceived(const std::string& subscription,
+                      base::Value rawlogs) override;
   bool UnsubscribeLogObserver(const std::string& subscription_id);
 
   raw_ptr<HostContentSettingsMap> host_content_settings_map_ = nullptr;