feat: validate transactions returned via reqresp + adjust peer scoring

yarn-project/p2p/src/client/p2p_client.ts

@@ -319,8 +319,6 @@ export class P2PClient implements P2P {
 
     this.log.debug(`Requested ${txHash.toString()} from peer | success = ${!!tx}`);
     if (tx) {
-      // TODO(https://github.com/AztecProtocol/aztec-packages/issues/8485): This check is not sufficient to validate the transaction. We need to validate the entire proof.
-      // TODO(https://github.com/AztecProtocol/aztec-packages/issues/8483): alter peer scoring system for a validator that returns an invalid transcation
       await this.txPool.addTxs([tx]);
     }
 

yarn-project/p2p/src/service/libp2p_service.ts

@@ -47,9 +47,11 @@ import { PeerErrorSeverity } from './peer_scoring.js';
 import { pingHandler, statusHandler } from './reqresp/handlers.js';
 import {
   DEFAULT_SUB_PROTOCOL_HANDLERS,
+  DEFAULT_SUB_PROTOCOL_VALIDATORS,
   PING_PROTOCOL,
   type ReqRespSubProtocol,
   type ReqRespSubProtocolHandlers,
+  ReqRespSubProtocolValidators,
   STATUS_PROTOCOL,
   type SubProtocolMap,
   TX_REQ_PROTOCOL,
@@ -162,7 +164,13 @@ export class LibP2PService implements P2PService {
       this.peerManager.heartbeat();
     }, this.config.peerCheckIntervalMS);
     this.discoveryRunningPromise.start();
-    await this.reqresp.start(this.requestResponseHandlers);
+
+    // Define the sub protocol validators - This is done within this start() method to gain a callback to the existing validateTx function
+    const reqrespSubProtocolValidators = {
+      ...DEFAULT_SUB_PROTOCOL_VALIDATORS,
+      [TX_REQ_PROTOCOL]: this.validateTx,
+    };
+    await this.reqresp.start(this.requestResponseHandlers, reqrespSubProtocolValidators);
   }
 
   /**
@@ -306,14 +314,7 @@ export class LibP2PService implements P2PService {
     protocol: SubProtocol,
     request: InstanceType<SubProtocolMap[SubProtocol]['request']>,
   ): Promise<InstanceType<SubProtocolMap[SubProtocol]['response']> | undefined> {
-    const pair = subProtocolMap[protocol];
-
-    const res = await this.reqresp.sendRequest(protocol, request.toBuffer());
-    if (!res) {
-      return undefined;
-    }
-
-    return pair.response.fromBuffer(res!);
+    return this.reqresp.sendRequest(protocol, request.toBuffer());
   }
 
   /**
@@ -425,7 +426,7 @@ export class LibP2PService implements P2PService {
     }
   }
 
-  private async validateTx(tx: Tx, peerId: PeerId): Promise<boolean> {
+  public async validateTx(tx: Tx, peerId: PeerId): Promise<boolean> {
     const blockNumber = (await this.l2BlockSource.getBlockNumber()) + 1;
     // basic data validation
     const dataValidator = new DataTxValidator();

yarn-project/p2p/src/service/reqresp/interface.ts

@@ -1,4 +1,5 @@
 import { Tx, TxHash } from '@aztec/circuit-types';
+import { PeerId } from '@libp2p/interface';
 
 /*
  * Request Response Sub Protocols
@@ -46,11 +47,25 @@ export interface ProtocolRateLimitQuota {
   globalLimit: RateLimitQuota;
 }
 
+const noopValidator = () => Promise.resolve(true);
+
 /**
  * A type mapping from supprotocol to it's handling funciton
  */
 export type ReqRespSubProtocolHandlers = Record<ReqRespSubProtocol, ReqRespSubProtocolHandler>;
 
+type ResponseValidator<T> = (msg: T, peerId: PeerId) => Promise<boolean>;
+
+export type ReqRespSubProtocolValidators = {
+  [S in ReqRespSubProtocol]: ResponseValidator<any>;
+}
+
+export const DEFAULT_SUB_PROTOCOL_VALIDATORS: ReqRespSubProtocolValidators = {
+  [PING_PROTOCOL]: noopValidator,
+  [STATUS_PROTOCOL]: noopValidator,
+  [TX_REQ_PROTOCOL]: noopValidator,
+};
+
 /**
  * Sub protocol map determines the request and response types for each
  * Req Resp protocol
@@ -91,6 +106,7 @@ interface RequestResponsePair<Req, Res> {
   };
 }
 
+
 /**
  * RequestableBuffer is a wrapper around a buffer that allows it to be
  * used in generic request response protocols

yarn-project/p2p/src/service/reqresp/reqresp.ts

@@ -12,6 +12,10 @@ import { type PeerManager } from '../peer_manager.js';
 import { type P2PReqRespConfig } from './config.js';
 import {
   DEFAULT_SUB_PROTOCOL_HANDLERS,
+  DEFAULT_SUB_PROTOCOL_VALIDATORS,
+  ReqRespSubProtocolValidators,
+  SubProtocolMap,
+  subProtocolMap,
   type ReqRespSubProtocol,
   type ReqRespSubProtocolHandlers,
 } from './interface.js';
@@ -36,7 +40,10 @@ export class ReqResp {
   private overallRequestTimeoutMs: number;
   private individualRequestTimeoutMs: number;
 
+  // Warning, if the `start` function is not called as the parent class constructor, then the default sub protocol handlers will be used ( not good )
   private subProtocolHandlers: ReqRespSubProtocolHandlers = DEFAULT_SUB_PROTOCOL_HANDLERS;
+  private subProtocolValidators: ReqRespSubProtocolValidators = DEFAULT_SUB_PROTOCOL_VALIDATORS;
+
   private rateLimiter: RequestResponseRateLimiter;
 
   constructor(config: P2PReqRespConfig, protected readonly libp2p: Libp2p, peerManager: PeerManager) {
@@ -51,8 +58,10 @@ export class ReqResp {
   /**
    * Start the reqresp service
    */
-  async start(subProtocolHandlers: ReqRespSubProtocolHandlers) {
+  async start(subProtocolHandlers: ReqRespSubProtocolHandlers, subProtocolValidators: ReqRespSubProtocolValidators) {
     this.subProtocolHandlers = subProtocolHandlers;
+    this.subProtocolValidators = subProtocolValidators;
+
     // Register all protocol handlers
     for (const subProtocol of Object.keys(this.subProtocolHandlers)) {
       await this.libp2p.handle(subProtocol, this.streamHandler.bind(this, subProtocol as ReqRespSubProtocol));
@@ -80,7 +89,7 @@ export class ReqResp {
    * @param payload - The payload to send
    * @returns - The response from the peer, otherwise undefined
    */
-  async sendRequest(subProtocol: ReqRespSubProtocol, payload: Buffer): Promise<Buffer | undefined> {
+  async sendRequest<SubProtocol extends ReqRespSubProtocol>(subProtocol: SubProtocol, payload: Buffer): Promise<InstanceType<SubProtocolMap[SubProtocol]['response']> | undefined> {
     const requestFunction = async () => {
       // Get active peers
       const peers = this.libp2p.getPeers();
@@ -92,14 +101,22 @@ export class ReqResp {
         // If we get a response, return it, otherwise we iterate onto the next peer
         // We do not consider it a success if we have an empty buffer
         if (response && response.length > 0) {
-          return response;
+
+          // TODO(md): indepth describe what this code does
+          const validator = this.subProtocolValidators[subProtocol];
+          const object = subProtocolMap[subProtocol].response.fromBuffer(response);
+
+          const isValid = await validator(object, peer);
+          if (isValid) {
+            return object;
+          }
         }
       }
       return undefined;
     };
 
     try {
-      return await executeTimeoutWithCustomError<Buffer | undefined>(
+      return await executeTimeoutWithCustomError<InstanceType<SubProtocolMap[SubProtocol]['response']> | undefined>(
         requestFunction,
         this.overallRequestTimeoutMs,
         () => new CollectiveReqRespTimeoutError(),