Support SSZ request body for POST /beacon/blinded_blocks endpoints (v1 & v2) (#4504)

## Issue Addressed

#4262 

## Proposed Changes

add SSZ support in request body for POST /beacon/blinded_blocks endpoints (v1 & v2)

## Additional Info

beacon_node/http_api/tests/broadcast_validation_tests.rs

@@ -851,6 +851,49 @@ pub async fn blinded_gossip_full_pass() {
         .block_is_known_to_fork_choice(&block.canonical_root()));
 }
 
+// This test checks that a block that is valid from both a gossip and consensus perspective is accepted when using `broadcast_validation=gossip`.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+pub async fn blinded_gossip_full_pass_ssz() {
+    /* this test targets gossip-level validation */
+    let validation_level: Option<BroadcastValidation> = Some(BroadcastValidation::Gossip);
+
+    // Validator count needs to be at least 32 or proposer boost gets set to 0 when computing
+    // `validator_count // 32`.
+    let validator_count = 64;
+    let num_initial: u64 = 31;
+    let tester = InteractiveTester::<E>::new(None, validator_count).await;
+
+    // Create some chain depth.
+    tester.harness.advance_slot();
+    tester
+        .harness
+        .extend_chain(
+            num_initial as usize,
+            BlockStrategy::OnCanonicalHead,
+            AttestationStrategy::AllValidators,
+        )
+        .await;
+    tester.harness.advance_slot();
+
+    let slot_a = Slot::new(num_initial);
+    let slot_b = slot_a + 1;
+
+    let state_a = tester.harness.get_current_state();
+    let (block, _): (SignedBlindedBeaconBlock<E>, _) =
+        tester.harness.make_blinded_block(state_a, slot_b).await;
+
+    let response: Result<(), eth2::Error> = tester
+        .client
+        .post_beacon_blinded_blocks_v2_ssz(&block, validation_level)
+        .await;
+
+    assert!(response.is_ok());
+    assert!(tester
+        .harness
+        .chain
+        .block_is_known_to_fork_choice(&block.canonical_root()));
+}
+
 /// This test checks that a block that is **invalid** from a gossip perspective gets rejected when using `broadcast_validation=consensus`.
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 pub async fn blinded_consensus_invalid() {

beacon_node/http_api/tests/tests.rs

@@ -2578,6 +2578,66 @@ impl ApiTester {
         }
     }
 
+    pub async fn test_blinded_block_production_ssz<Payload: AbstractExecPayload<E>>(&self) {
+        let fork = self.chain.canonical_head.cached_head().head_fork();
+        let genesis_validators_root = self.chain.genesis_validators_root;
+
+        for _ in 0..E::slots_per_epoch() * 3 {
+            let slot = self.chain.slot().unwrap();
+            let epoch = self.chain.epoch().unwrap();
+
+            let proposer_pubkey_bytes = self
+                .client
+                .get_validator_duties_proposer(epoch)
+                .await
+                .unwrap()
+                .data
+                .into_iter()
+                .find(|duty| duty.slot == slot)
+                .map(|duty| duty.pubkey)
+                .unwrap();
+            let proposer_pubkey = (&proposer_pubkey_bytes).try_into().unwrap();
+
+            let sk = self
+                .validator_keypairs()
+                .iter()
+                .find(|kp| kp.pk == proposer_pubkey)
+                .map(|kp| kp.sk.clone())
+                .unwrap();
+
+            let randao_reveal = {
+                let domain = self.chain.spec.get_domain(
+                    epoch,
+                    Domain::Randao,
+                    &fork,
+                    genesis_validators_root,
+                );
+                let message = epoch.signing_root(domain);
+                sk.sign(message).into()
+            };
+
+            let block = self
+                .client
+                .get_validator_blinded_blocks::<E, Payload>(slot, &randao_reveal, None)
+                .await
+                .unwrap()
+                .data;
+
+            let signed_block = block.sign(&sk, &fork, genesis_validators_root, &self.chain.spec);
+
+            self.client
+                .post_beacon_blinded_blocks_ssz(&signed_block)
+                .await
+                .unwrap();
+
+            // This converts the generic `Payload` to a concrete type for comparison.
+            let head_block = SignedBeaconBlock::from(signed_block.clone());
+            assert_eq!(head_block, signed_block);
+
+            self.chain.slot_clock.set_slot(slot.as_u64() + 1);
+        }
+    }
+
     pub async fn test_blinded_block_production_no_verify_randao<Payload: AbstractExecPayload<E>>(
         self,
     ) -> Self {
@@ -4704,6 +4764,14 @@ async fn blinded_block_production_full_payload_premerge() {
         .await;
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn blinded_block_production_ssz_full_payload_premerge() {
+    ApiTester::new()
+        .await
+        .test_blinded_block_production_ssz::<FullPayload<_>>()
+        .await;
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn blinded_block_production_with_skip_slots_full_payload_premerge() {
     ApiTester::new()
@@ -4713,6 +4781,15 @@ async fn blinded_block_production_with_skip_slots_full_payload_premerge() {
         .await;
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn blinded_block_production_ssz_with_skip_slots_full_payload_premerge() {
+    ApiTester::new()
+        .await
+        .skip_slots(E::slots_per_epoch() * 2)
+        .test_blinded_block_production_ssz::<FullPayload<_>>()
+        .await;
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn blinded_block_production_no_verify_randao_full_payload_premerge() {
     ApiTester::new()