Cow amm mitigation (#2402)

# Description
Changes required to test V1 of the CoW AMM. This PR can be reverted ~
next week when the CoW AMM has been patched.

See this
[thread](https://cowservices.slack.com/archives/C067ZLV8U13/p1707488519026799)
for more context.

# Limitations
Requires us to be aware of the contracts we'd like to apply the
mitigation for. Considered alternatives were to enforce 1 order per
EIP-1271 contract. Since we need to be aware about CoW AMM orders in the
autopilot anyway (to pick a reasonable one which is still TODO) we might
as well implement the less restrictive solution everywhere.

crates/autopilot/src/arguments.rs

@@ -217,6 +217,9 @@ pub struct Arguments {
     /// `order_events` database table.
     #[clap(long, env, default_value = "30d", value_parser = humantime::parse_duration)]
     pub order_events_cleanup_threshold: Duration,
+
+    #[clap(long, env)]
+    pub cow_amms: Vec<H160>,
 }
 
 impl std::fmt::Display for Arguments {
@@ -259,6 +262,7 @@ impl std::fmt::Display for Arguments {
             auction_update_interval,
             max_settlement_transaction_wait,
             s3,
+            cow_amms,
         } = self;
 
         write!(f, "{}", shared)?;
@@ -335,6 +339,7 @@ impl std::fmt::Display for Arguments {
             max_settlement_transaction_wait
         )?;
         writeln!(f, "s3: {:?}", s3)?;
+        writeln!(f, "cow_amms: {:?}", cow_amms)?;
         Ok(())
     }
 }

crates/autopilot/src/run.rs

@@ -554,6 +554,7 @@ pub async fn run(args: Arguments) {
             .try_into()
             .expect("limit order price factor can't be converted to BigDecimal"),
         domain::ProtocolFee::new(args.fee_policy.clone()),
+        args.cow_amms.into_iter().collect(),
     );
     solvable_orders_cache
         .update(block)

crates/autopilot/src/solvable_orders.rs

@@ -13,6 +13,7 @@ use {
     number::conversions::u256_to_big_decimal,
     primitive_types::{H160, H256, U256},
     prometheus::{IntCounter, IntCounterVec, IntGauge, IntGaugeVec},
+    rand::seq::SliceRandom,
     shared::{
         account_balances::{BalanceFetching, Query},
         bad_token::BadTokenDetecting,
@@ -77,6 +78,9 @@ pub struct SolvableOrdersCache {
     weth: H160,
     limit_order_price_factor: BigDecimal,
     protocol_fee: domain::ProtocolFee,
+    // TODO: remove ASAP since this we only use this for a mitigation that
+    // should be implemented on a smart contract level.
+    cow_amms: HashSet<H160>,
 }
 
 type Balances = HashMap<Query, U256>;
@@ -101,6 +105,7 @@ impl SolvableOrdersCache {
         weth: H160,
         limit_order_price_factor: BigDecimal,
         protocol_fee: domain::ProtocolFee,
+        cow_amms: HashSet<H160>,
     ) -> Arc<Self> {
         let self_ = Arc::new(Self {
             min_order_validity_period,
@@ -118,6 +123,7 @@ impl SolvableOrdersCache {
             weth,
             limit_order_price_factor,
             protocol_fee,
+            cow_amms,
         });
         tokio::task::spawn(
             update_task(Arc::downgrade(&self_), update_interval, current_block)
@@ -153,6 +159,10 @@ impl SolvableOrdersCache {
         let removed = counter.checkpoint("invalid_signature", &orders);
         invalid_order_uids.extend(removed);
 
+        let orders = filter_duplicate_cow_amm_orders(orders, &self.cow_amms);
+        let removed = counter.checkpoint("duplicate_cow_amm_order", &orders);
+        invalid_order_uids.extend(removed);
+
         let orders = filter_unsupported_tokens(orders, self.bad_token_detector.as_ref()).await?;
         let removed = counter.checkpoint("unsupported_token", &orders);
         invalid_order_uids.extend(removed);
@@ -519,6 +529,43 @@ async fn filter_unsupported_tokens(
     Ok(orders)
 }
 
+/// Enforces that for all CoW AMMs at most 1 order is in the auction.
+/// This is needed to protect against a known attack vector.
+fn filter_duplicate_cow_amm_orders(mut orders: Vec<Order>, cow_amms: &HashSet<H160>) -> Vec<Order> {
+    let mut amm_orders = HashMap::<H160, Vec<&Order>>::new();
+    for order in &orders {
+        let owner = order.metadata.owner;
+        if cow_amms.contains(&owner) {
+            amm_orders.entry(owner).or_default().push(order);
+        }
+    }
+    let canonical_amm_orders: HashSet<OrderUid> = amm_orders
+        .into_iter()
+        .map(|(owner, orders)| {
+            let uid = orders
+                .choose(&mut rand::thread_rng())
+                .expect("every group contains at least 1 order")
+                .metadata
+                .uid;
+            if orders.len() > 1 {
+                // TODO: find better heuristic to pick "canonical" order
+                tracing::warn!(
+                    num = orders.len(),
+                    amm = ?owner,
+                    order = ?uid,
+                    "multiple orders for the same CoW AMM; picked random order"
+                );
+            }
+            uid
+        })
+        .collect();
+
+    orders.retain(|o| {
+        !cow_amms.contains(&o.metadata.owner) || canonical_amm_orders.contains(&o.metadata.uid)
+    });
+    orders
+}
+
 /// Filter out limit orders which are far enough outside the estimated native
 /// token price.
 fn filter_mispriced_limit_orders(

crates/driver/src/domain/competition/mod.rs

@@ -103,6 +103,38 @@ impl Competition {
             }
         });
 
+        let solutions = solutions.filter(|solution| {
+            let mut amm_orders = HashSet::<eth::Address>::new();
+
+            for trade in solution.trades() {
+                let solution = solution.id();
+
+                let owner = match trade {
+                    solution::Trade::Fulfillment(f) => {
+                        let uid = f.order().uid;
+                        let Some(order) = auction.orders().iter().find(|o| o.uid == uid) else {
+                            tracing::warn!(?uid, ?solution, "order not in original auction");
+                            return false;
+                        };
+                        order.trader().0
+                    }
+                    solution::Trade::Jit(j) => j.order().signature.signer,
+                };
+
+                let is_cow_amm_order = self.eth.contracts().cow_amms().contains(&owner);
+                if is_cow_amm_order && !amm_orders.insert(owner) {
+                    tracing::warn!(
+                        amm = ?owner,
+                        ?solution,
+                        "solution contains more than 1 order for the same CoW AMM"
+                    );
+                    return false;
+                }
+            }
+
+            true
+        });
+
         // Encode solutions into settlements (streamed).
         let encoded = solutions
             .map(|solution| async move {

crates/driver/src/domain/competition/order/mod.rs

@@ -358,7 +358,7 @@ pub enum BuyTokenBalance {
 
 /// The address which placed the order.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
-pub struct Trader(eth::Address);
+pub struct Trader(pub eth::Address);
 
 impl From<Trader> for eth::Address {
     fn from(value: Trader) -> Self {

crates/driver/src/infra/blockchain/contracts.rs

@@ -1,6 +1,7 @@
 use {
     crate::{domain::eth, infra::blockchain::Ethereum},
     ethcontract::dyns::DynWeb3,
+    std::collections::HashSet,
     thiserror::Error,
 };
 
@@ -10,12 +11,14 @@ pub struct Contracts {
     vault_relayer: eth::ContractAddress,
     vault: contracts::BalancerV2Vault,
     weth: contracts::WETH9,
+    cow_amms: HashSet<eth::Address>,
 }
 
-#[derive(Debug, Default, Clone, Copy)]
+#[derive(Debug, Default, Clone)]
 pub struct Addresses {
     pub settlement: Option<eth::ContractAddress>,
     pub weth: Option<eth::ContractAddress>,
+    pub cow_amms: Option<HashSet<eth::Address>>,
 }
 
 impl Contracts {
@@ -53,6 +56,7 @@ impl Contracts {
             vault_relayer,
             vault,
             weth,
+            cow_amms: addresses.cow_amms.unwrap_or_default(),
         })
     }
 
@@ -75,6 +79,10 @@ impl Contracts {
     pub fn weth_address(&self) -> eth::WethAddress {
         self.weth.address().into()
     }
+
+    pub fn cow_amms(&self) -> &HashSet<eth::Address> {
+        &self.cow_amms
+    }
 }
 
 /// Returns the address of a contract for the specified network, or `None` if

crates/driver/src/infra/config/file/load.rs

@@ -314,6 +314,10 @@ pub async fn load(network: &blockchain::Network, path: &Path) -> infra::Config {
         contracts: blockchain::contracts::Addresses {
             settlement: config.contracts.gp_v2_settlement.map(Into::into),
             weth: config.contracts.weth.map(Into::into),
+            cow_amms: config
+                .contracts
+                .cow_amms
+                .map(|contracts| contracts.into_iter().map(eth::Address).collect()),
         },
         disable_access_list_simulation: config.disable_access_list_simulation,
         disable_gas_simulation: config.disable_gas_simulation.map(Into::into),

crates/driver/src/infra/config/file/mod.rs

@@ -237,6 +237,8 @@ struct ContractsConfig {
 
     /// Override the default address of the WETH contract.
     weth: Option<eth::H160>,
+
+    cow_amms: Option<Vec<eth::H160>>,
 }
 
 #[derive(Debug, Deserialize)]

crates/driver/src/run.rs

@@ -143,7 +143,7 @@ async fn ethereum(config: &infra::Config, ethrpc: blockchain::Rpc) -> Ethereum {
             .await
             .expect("initialize gas price estimator"),
     );
-    Ethereum::new(ethrpc, config.contracts, gas).await
+    Ethereum::new(ethrpc, config.contracts.clone(), gas).await
 }
 
 fn solvers(config: &config::Config, eth: &Ethereum) -> Vec<Solver> {

crates/driver/src/tests/setup/solver.rs

@@ -220,6 +220,7 @@ impl Solver {
             Addresses {
                 settlement: Some(config.blockchain.settlement.address().into()),
                 weth: Some(config.blockchain.weth.address().into()),
+                cow_amms: Some(Default::default()),
             },
             gas,
         )