Stop using string interpolation on AA SQL

- Don't use SQL string interpolation for Account Abstraction query
- It doesn't look like a SQL injection there could be possible, but just in case

Author: Uxio0

safe_transaction_service/history/models.py

@@ -1,4 +1,5 @@
 import datetime
+import json
 from collections.abc import Iterator, Sequence
 from decimal import Decimal
 from enum import Enum
@@ -407,10 +408,12 @@ def account_abstraction_txs(self) -> RawQuerySet:
         """
         :return: Transactions containing ERC4337 `UserOperation` event
         """
-        query = '{"topics": ["' + to_0x_hex_str(USER_OPERATION_EVENT_TOPIC) + '"]}'
+        # Use json.dumps to safely construct the JSON query string
+        query_json = json.dumps({"topics": [to_0x_hex_str(USER_OPERATION_EVENT_TOPIC)]})
 
         return self.raw(
-            f"SELECT * FROM history_ethereumtx WHERE '{query}'::jsonb <@ ANY (logs)"
+            "SELECT * FROM history_ethereumtx WHERE %s::jsonb <@ ANY (logs)",
+            [query_json],
         )
 
 

safe_transaction_service/history/tests/test_models.py

@@ -350,20 +350,26 @@ def test_create_from_tx_dict(self):
                     ethereum_tx.transaction_index, tx_receipt["transactionIndex"]
                 )
 
-    def test_account_abstraction_tx_hashes(self):
+    def test_account_abstraction_txs(self):
         self.assertEqual(len(EthereumTx.objects.account_abstraction_txs()), 0)
 
         # Insert random transaction
         EthereumTxFactory()
         self.assertEqual(len(EthereumTx.objects.account_abstraction_txs()), 0)
 
+        # Insert a non 4337 transaction
+        _ethereum_tx = EthereumTxFactory(
+            logs=[clean_receipt_log(log) for log in type_0_tx["receipt"]["logs"]]
+        )
+        self.assertEqual(len(EthereumTx.objects.account_abstraction_txs()), 0)
+
         # Insert a 4337 transaction
-        ethereum_tx = EthereumTxFactory(
+        aa_ethereum_tx = EthereumTxFactory(
             logs=[clean_receipt_log(log) for log in aa_tx_receipt_mock["logs"]]
         )
         ethereum_txs = EthereumTx.objects.account_abstraction_txs()
         self.assertEqual(len(ethereum_txs), 1)
-        self.assertEqual(ethereum_txs[0], ethereum_tx)
+        self.assertEqual(ethereum_txs[0], aa_ethereum_tx)
 
     def test_get_deployed_proxies_from_logs(self):
         ethereum_tx = EthereumTxFactory(