Reintroduce repair ping forwarding

Forwards ping frames arriving at the 'repair ingress' port from
the shred tile to the repair tile.

This was previously done at the net tile level, but is moved to
the shred/repair tiles to keep the network stack clean.

Author: riptl

src/app/shared/commands/get_identity.c

@@ -30,7 +30,7 @@ get_identity_cmd_fn( args_t *   args   FD_PARAM_UNUSED,
   fd_topo_join_workspace( &config->topo, &config->topo.workspaces[ shred_wksp_id ], FD_SHMEM_JOIN_MODE_READ_ONLY );
 
   /* Cast to shred context structure */
-  fd_shred_ctx_t const * shred_ctx = fd_topo_obj_laddr( &config->topo, shred_tile->tile_obj_id );
+  fd_shred_ctx_hdr_t const * shred_ctx = fd_topo_obj_laddr( &config->topo, shred_tile->tile_obj_id );
   if( FD_UNLIKELY( !shred_ctx ) ) {
     fd_topo_leave_workspaces( &config->topo );
     FD_LOG_ERR(( "Failed to access shred tile object" ));

src/disco/shred/fd_shred_tile.c

@@ -1,4 +1,4 @@
-#include "../tiles.h"
+#include "fd_shred_tile.h"
 
 #include "generated/fd_shred_tile_seccomp.h"
 #include "../../util/pod/fd_pod_format.h"
@@ -172,6 +172,7 @@ typedef struct {
   ushort net_id;
 
   int skip_frag;
+  int ping_frag;
 
   ulong                    adtl_dests_leader_cnt;
   fd_shred_dest_weighted_t adtl_dests_leader    [ FD_TOPO_ADTL_DESTS_MAX ];
@@ -349,6 +350,64 @@ before_frag( fd_shred_ctx_t * ctx,
   return 0;
 }
 
+/* *** Ping forwarding ***
+   Solana peers use a simple 'ping-pong' to do a primitive form of
+   address and endpoint validation.  (Mainly defeats reflection-like
+   flood attacks, by allowing receivers to decline flows)
+
+   Unfortunately, ping-pong flows that belong to the repair tile reuse
+   the 'repair intake' port, which ends up at the shred tile.  So, the
+   shred tile must forward pings back to repair. */
+
+static int
+forward_ping_prepare(
+    fd_shred_ctx_t * ctx,
+    uchar const *    buf,
+    ulong            sz
+) {
+  /* Don't do anything if the repair tile doesn't exist in this topology */
+  if( FD_UNLIKELY( !ctx->repair_out_mem ) ) return 0;
+
+  /* Extract IPv4 and UDP info, so downstream can generate a response */
+  if( FD_UNLIKELY( sz<sizeof(fd_eth_hdr_t)+sizeof(fd_ip4_hdr_t) ) ) return 0;
+  fd_ip4_hdr_t ip4     = FD_LOAD( fd_ip4_hdr_t, buf+sizeof(fd_eth_hdr_t) );
+  uint         ip4_len = FD_IP4_GET_LEN( ip4 );
+  if( FD_UNLIKELY( sz<sizeof(fd_eth_hdr_t)+ip4_len+sizeof(fd_udp_hdr_t)+FD_REPAIR_PING_SZ ) ) return 0;
+  fd_udp_hdr_t udp     = FD_LOAD( fd_udp_hdr_t, buf+sizeof(fd_eth_hdr_t)+FD_IP4_GET_LEN( ip4 ) );
+
+  /* Write a ping frame to the shred->repair link.
+     Pings are smaller than the shred_repair MTU, so we always have
+     space to send a frame. */
+  FD_STATIC_ASSERT( FD_SHRED_REPAIR_MTU>=sizeof(fd_repair_ping_fwd_t), mtu );
+  fd_repair_ping_fwd_t * dst = fd_chunk_to_laddr( ctx->repair_out_mem, ctx->repair_out_chunk );
+  dst->src_ip4  = ip4.saddr;
+  dst->src_port = fd_ushort_bswap( udp.net_sport );
+  fd_memcpy( dst->ping, buf, FD_REPAIR_PING_SZ );
+  return 1;
+}
+
+static void
+forward_ping_commit(
+    fd_shred_ctx_t *    ctx,
+    fd_stem_context_t * stem
+) {
+  /* Don't do anything if the repair tile doesn't exist in this topology */
+  if( FD_UNLIKELY( !ctx->repair_out_mem ) ) return;
+
+  /* Commit a previous shred->repair ping forward */
+  ulong out_idx = ctx->repair_out_idx;
+  ulong sig     = ULONG_MAX; /* repair ping */
+  ulong chunk   = ctx->repair_out_chunk;
+  ulong sz      = sizeof(fd_repair_ping_fwd_t);
+  ulong ctl     = 0UL; /* unused */
+  ulong tsorig  = 0UL; /* TODO forward tsorig from upstream packet */
+  ulong tspub   = fd_frag_meta_ts_comp( fd_tickcount() );
+  fd_stem_publish( stem, out_idx, sig, chunk, sz, ctl, tsorig, tspub );
+
+  /* Wind up for next iteration */
+  ctx->repair_out_chunk = fd_dcache_compact_next( chunk, sz, ctx->repair_out_chunk0, ctx->repair_out_wmark );
+}
+
 static void
 during_frag( fd_shred_ctx_t * ctx,
              ulong            in_idx,
@@ -635,6 +694,16 @@ during_frag( fd_shred_ctx_t * ctx,
     uchar const * dcache_entry = fd_net_rx_translate_frag( &ctx->in[ in_idx ].net_rx, chunk, ctl, sz );
     ulong hdr_sz = fd_disco_netmux_sig_hdr_sz( sig );
     FD_TEST( hdr_sz <= sz ); /* Should be ensured by the net tile */
+
+    /* Ping traffic generated by the repair tile can end up with the
+       shred tile ~ forward it. */
+    if( FD_UNLIKELY( (sz-hdr_sz)==FD_REPAIR_PING_SZ ) ) {
+      ctx->ping_frag = !!forward_ping_prepare( ctx, dcache_entry, sz );
+      return;
+    } else {
+      ctx->ping_frag = 0;
+    }
+
     fd_shred_t const * shred = fd_shred_parse( dcache_entry+hdr_sz, sz-hdr_sz );
     if( FD_UNLIKELY( !shred ) ) {
       ctx->skip_frag = 1;
@@ -808,6 +877,11 @@ after_frag( fd_shred_ctx_t *    ctx,
   ulong fanout = 200UL; /* Default Agave's DATA_PLANE_FANOUT = 200UL */
 
   if( FD_LIKELY( ctx->in_kind[ in_idx ]==IN_KIND_NET ) ) {
+    if( FD_UNLIKELY( ctx->ping_frag ) ) {
+      forward_ping_commit( ctx, stem );
+      return;
+    }
+
     uchar * shred_buffer    = ctx->shred_buffer;
     ulong   shred_buffer_sz = ctx->shred_buffer_sz;
 

src/disco/shred/fd_shred_tile.h

@@ -4,12 +4,30 @@
 #include "../tiles.h"
 #include "../../flamenco/types/fd_types_custom.h"
 
+/* FD_REPAIR_PING_SZ is the UDP payload size of a 'ping'-related packet.
+   These incoming packets are forwarded to the repair tile. */
+#define FD_REPAIR_PING_SZ (132UL)
+
+struct fd_repair_ping_fwd {
+  uint   src_ip4;
+  uint   dst_ip4;
+  ushort src_port;
+
+  /* FIXME: Just have a wire-format struct here for the ping frame.
+     This is currently not possible due to use of fd_types, which is not
+     guaranteed to have the same in-memory format as the wire format. */
+  uchar ping[ FD_REPAIR_PING_SZ ];
+};
+
+typedef struct fd_repair_ping_fwd fd_repair_ping_fwd_t;
+
 /* Forward declarations */
 typedef struct fd_fec_resolver fd_fec_resolver_t;
 typedef struct fd_keyswitch_private fd_keyswitch_t;
 typedef struct fd_keyguard_client fd_keyguard_client_t;
 
-/* Shred tile context structure */
+/* Part of the shred tile context struct
+   FIXME remove this and just use fd_shred_ctx_t everywhere */
 typedef struct {
   fd_shredder_t      * shredder;
   fd_fec_resolver_t  * resolver;
@@ -27,6 +45,6 @@ typedef struct {
   fd_keyswitch_t *     keyswitch;
   fd_keyguard_client_t keyguard_client[1];
   /* ... rest of the structure members ... */
-} fd_shred_ctx_t;
+} fd_shred_ctx_hdr_t;
 
 #endif /* HEADER_fd_src_disco_shred_fd_shred_tile_h */

src/discof/repair/fd_repair_tile.c

@@ -2,6 +2,7 @@
 #define _GNU_SOURCE
 
 #include "../../disco/topo/fd_topo.h"
+#include "../../disco/shred/fd_shred_tile.h"
 #include "generated/fd_repair_tile_seccomp.h"
 
 #include "../../flamenco/repair/fd_repair.h"
@@ -253,14 +254,12 @@ handle_new_cluster_contact_info( fd_repair_tile_ctx_t * ctx,
   }
 }
 
-ulong
-fd_repair_handle_ping( fd_repair_tile_ctx_t *  repair_tile_ctx,
-                       fd_repair_t *                 glob,
-                       fd_gossip_ping_t const *      ping,
-                       fd_gossip_peer_addr_t const * peer_addr FD_PARAM_UNUSED,
-                       uint                          self_ip4_addr FD_PARAM_UNUSED,
-                       uchar *                       msg_buf,
-                       ulong                         msg_buf_sz ) {
+static ulong
+fd_repair_handle_ping( fd_repair_tile_ctx_t *   repair_tile_ctx,
+                       fd_repair_t *            glob,
+                       fd_gossip_ping_t const * ping,
+                       uchar *                  msg_buf,
+                       ulong                    msg_buf_sz ) {
   fd_repair_protocol_t protocol;
   fd_repair_protocol_new_disc(&protocol, fd_repair_protocol_enum_pong);
   fd_gossip_ping_t * pong = &protocol.inner.pong;
@@ -286,8 +285,22 @@ fd_repair_handle_ping( fd_repair_tile_ctx_t *  repair_tile_ctx,
   return buflen;
 }
 
+static void
+fd_repair_handle_ping1( fd_repair_tile_ctx_t *   repair_tile_ctx,
+                        fd_repair_t *            glob,
+                        fd_stem_context_t *      stem,
+                        fd_gossip_ping_t const * ping,
+                        uint const               src_ip,
+                        uint const               dst_port,
+                        ushort const             src_port ) {
+  uchar buf[1024];
+  ulong buflen = fd_repair_handle_ping( repair_tile_ctx, glob, ping, buf, sizeof(buf) );
+  ulong tsorig = fd_frag_meta_ts_comp( fd_tickcount() );
+  send_packet( repair_tile_ctx, stem, 1, src_ip, src_port, dst_port, buf, buflen, tsorig );
+}
+
 /* Pass a raw client response packet into the protocol. addr is the address of the sender */
-static int
+static void
 fd_repair_recv_clnt_packet( fd_repair_tile_ctx_t *        repair_tile_ctx,
                             fd_stem_context_t *           stem,
                             fd_repair_t *                 glob,
@@ -297,35 +310,25 @@ fd_repair_recv_clnt_packet( fd_repair_tile_ctx_t *        repair_tile_ctx,
                             uint                          dst_ip4_addr ) {
   glob->metrics.recv_clnt_pkt++;
 
-  FD_SCRATCH_SCOPE_BEGIN {
-    while( 1 ) {
-      ulong decoded_sz;
-      fd_repair_response_t * gmsg = fd_bincode_decode1_scratch(
-          repair_response, msg, msglen, NULL, &decoded_sz );
-      if( FD_UNLIKELY( !gmsg ) ) {
-        /* Solana falls back to assuming we got a shred in this case
-           https://github.com/solana-labs/solana/blob/master/core/src/repair/serve_repair.rs#L1198 */
-        break;
-      }
-      if( FD_UNLIKELY( decoded_sz != msglen ) ) {
-        break;
-      }
-
-      switch( gmsg->discriminant ) {
-      case fd_repair_response_enum_ping:
-        {
-          uchar buf[1024];
-          ulong buflen = fd_repair_handle_ping( repair_tile_ctx, glob, &gmsg->inner.ping, src_addr, dst_ip4_addr, buf, sizeof(buf) );
-          ulong tsorig = fd_frag_meta_ts_comp( fd_tickcount() );
-          send_packet( repair_tile_ctx, stem, 1, src_addr->addr, src_addr->port, dst_ip4_addr, buf, buflen, tsorig );
-          break;
-        }
-      }
-
-      return 0;
+  if( FD_UNLIKELY( msglen<sizeof(uint) ) ) {
+    glob->metrics.recv_pkt_corrupted_msg++;
+    return;
+  }
+  uint msg_type = FD_LOAD( uint, msg );
+  msg    += sizeof(uint);
+  msglen -= sizeof(uint);
+
+  switch( msg_type ) {
+  case 0: /* ping */
+    if( FD_UNLIKELY( msglen!=132 ) ) {
+      glob->metrics.recv_pkt_corrupted_msg++;
+      return;
     }
-  } FD_SCRATCH_SCOPE_END;
-  return 0;
+    fd_repair_handle_ping1( repair_tile_ctx, glob, stem, fd_type_pun_const( msg ), src_addr->addr, dst_ip4_addr, src_addr->port );
+    break;
+  default:
+    break;
+  }
 }
 
 static ulong
@@ -427,7 +430,10 @@ before_frag( fd_repair_tile_ctx_t * ctx,
              ulong                  sig ) {
   uint in_kind = ctx->in_kind[ in_idx ];
   if( FD_LIKELY  ( in_kind==IN_KIND_NET   ) ) return fd_disco_netmux_sig_proto( sig )!=DST_PROTO_REPAIR;
-  if( FD_UNLIKELY( in_kind==IN_KIND_SHRED ) ) return fd_int_if( fd_forest_root_slot( ctx->forest )==ULONG_MAX, -1, 0 ); /* not ready to read frag */
+  if( FD_UNLIKELY( in_kind==IN_KIND_SHRED ) ) {
+    if( FD_UNLIKELY( sig==ULONG_MAX ) ) return 0; /* repair ping */
+    return fd_int_if( fd_forest_root_slot( ctx->forest )==ULONG_MAX, -1, 0 ); /* not ready to read frag */
+  }
   return 0;
 }
 
@@ -646,6 +652,13 @@ after_frag( fd_repair_tile_ctx_t * ctx,
     return;
   }
 
+  if( FD_UNLIKELY( in_kind==IN_KIND_NET && sig==ULONG_MAX ) ) {
+    fd_repair_ping_fwd_t const * fwd  = fd_type_pun_const( ctx->buffer );
+    fd_gossip_ping_t const *     ping = fd_type_pun_const( fwd->ping );
+    fd_repair_handle_ping1( ctx, ctx->repair, stem, ping, fwd->src_ip4, fwd->dst_ip4, fwd->src_port );
+    return;
+  }
+
   fd_eth_hdr_t const * eth  = (fd_eth_hdr_t const *)ctx->buffer;
   fd_ip4_hdr_t const * ip4  = (fd_ip4_hdr_t const *)( (ulong)eth + sizeof(fd_eth_hdr_t) );
   fd_udp_hdr_t const * udp  = (fd_udp_hdr_t const *)( (ulong)ip4 + FD_IP4_GET_LEN( *ip4 ) );

src/flamenco/types/fd_fuzz_types.h

@@ -3675,23 +3675,6 @@ void *fd_repair_protocol_generate( void *mem, void **alloc_mem, fd_rng_t * rng )
   return mem;
 }
 
-void fd_repair_response_inner_generate( fd_repair_response_inner_t * self, void **alloc_mem, uint discriminant, fd_rng_t * rng ) {
-  switch (discriminant) {
-  case 0: {
-    fd_gossip_ping_generate( &self->ping, alloc_mem, rng );
-    break;
-  }
-  }
-}
-void *fd_repair_response_generate( void *mem, void **alloc_mem, fd_rng_t * rng ) {
-  fd_repair_response_t *self = (fd_repair_response_t *) mem;
-  *alloc_mem = (uchar *) *alloc_mem + sizeof(fd_repair_response_t);
-  fd_repair_response_new(mem);
-  self->discriminant = fd_rng_uint( rng ) % 1;
-  fd_repair_response_inner_generate( &self->inner, alloc_mem, self->discriminant, rng );
-  return mem;
-}
-
 void fd_instr_error_enum_inner_generate( fd_instr_error_enum_inner_t * self, void **alloc_mem, uint discriminant, fd_rng_t * rng ) {
   switch (discriminant) {
   case 25: {