diff --git a/src/controller/ha.c b/src/controller/ha.c index 10eca060..8d0b0710 100644 --- a/src/controller/ha.c +++ b/src/controller/ha.c @@ -1119,18 +1119,232 @@ static void on_vote_connected(endpoint_t *ep) } /* - * vote_reply_msg_handler: stub — implemented in task 2.3. - * Declared here so on_vote_connected can reference it. - * Signature matches msg_handler_t = int (*)(void*) from netio.h. + * xce_vote_reply_hook: test-only seam for vote_reply_msg_handler. + * + * Forward-declared here (before vote_reply_msg_handler) so the handler body + * can reference it. Defined only under XCE_TESTING; guarded below. + * + * When non-NULL, vote_reply_msg_handler() calls this INSTEAD OF the real + * handle_request_vote_reply() so tests can capture decoded fields without + * a live iomgr. Receives the decoded scalar fields — no capnp pointers. + * + * Pattern mirrors xce_vote_request_hook used in task 2.2. */ -static int vote_reply_msg_handler(void *ep) +#ifdef XCE_TESTING +int (*xce_vote_reply_hook)(ha_state_t *ha, ha_elect_ctx_t *ctx, int peer_idx, + uint64_t term, int vote_granted, + uint64_t peer_tag, uint64_t peer_lsn) = NULL; +#endif /* XCE_TESTING */ + +/* + * vote_reply_dispatch: internal helper — calls handle_request_vote_reply or + * the test hook (when XCE_TESTING and hook is non-NULL). + */ +static void vote_reply_dispatch(ha_state_t *ha, ha_elect_ctx_t *ctx, int idx, + uint64_t term, int vote_granted, + uint64_t peer_tag, uint64_t peer_lsn) { - (void)ep; - /* Task 2.3 will implement this; stub returns 0 to avoid crash if called */ +#ifdef XCE_TESTING + if (xce_vote_reply_hook != NULL) { + xce_vote_reply_hook(ha, ctx, idx, term, vote_granted, peer_tag, peer_lsn); + return; + } +#endif + handle_request_vote_reply(ha, ctx, idx, term, vote_granted, peer_tag, peer_lsn); +} + +/* + * vote_reply_msg_handler: iomgr message handler for incoming RequestVoteReply. + * + * Called by iomgr (via INVOKE_MSG_HANDLER) when a full packet arrives on an + * outgoing vote endpoint after on_vote_connected fired. + * + * Steps: + * 1. Cast ep to endpoint_t*; get packet from ep->fsm_in.pkt (owned by us). + * 2. Raw capnp decode: init capn_ctx, read outer Request wrapper. + * 3. read_RequestReply() into rep_s (capnp pointer — still valid). + * 4. read_RequestVoteReply() into vr_raw (scalars) BEFORE capn_free. + * 5. capn_free() — capnp segment released; vr_raw still valid (scalars only). + * 6. free_packet() + free(pkt) — packet memory released. + * 7. Term fencing: if vr_raw.term > ha->term, update ha->term atomically. + * 8. Call vote_reply_dispatch() with vr_raw scalar fields. + * + * On any decode error: treat peer as denied (grant=0, tag=0, lsn=0). + * + * ASAN safety: all capnp fields read before capn_free() — no heap-use-after-free. + * Requirements: 3.1 (iomgr packet), 3.2 (no capnp UAF), 3.3 (extract fields), + * 3.4 (term fencing). + */ +static int vote_reply_msg_handler(void *ep_raw) +{ + endpoint_t *ep = (endpoint_t *)ep_raw; + vote_ep_ctx_t *vctx = (vote_ep_ctx_t *)ep->data; + ha_state_t *ha = vctx->ha; + ha_elect_ctx_t *ctx = vctx->ctx; + int idx = vctx->peer_idx; + io_packet_t *pkt = ep->fsm_in.pkt; /* ownership transferred by INVOKE_MSG_HANDLER */ + int ok = 0; + + /* Decoded reply fields — populated on success path */ + uint64_t vr_term = 0; + int vr_granted = 0; + uint64_t vr_tag = 0; + uint64_t vr_lsn = 0; + + if (!pkt || !pkt->data || pkt->sz == 0) + goto done; + + { + struct capn capn_ctx; + struct Request outer_rq; + Request_ptr outer_rp; + struct RequestReply rep_s; + struct RequestVoteReply vr_raw; + int capn_ok; + + capn_ok = capn_init_mem(&capn_ctx, (const uint8_t *)pkt->data, + (size_t)pkt->sz, 0); + if (capn_ok != 0) + goto done; + + outer_rp.p = capn_getp(capn_root(&capn_ctx), 0, 1); + read_Request(&outer_rq, outer_rp); + + if (outer_rq.body_which != Request_body_replymsg || + !outer_rq.body.replymsg.p.data) { + capn_free(&capn_ctx); + goto done; + } + + read_RequestReply(&rep_s, outer_rq.body.replymsg); + + if (rep_s.data_which != RequestReply_data_requestvotereplymsg) { + capn_free(&capn_ctx); + goto done; + } + + /* CRITICAL: copy all scalars BEFORE capn_free to avoid UAF. + * struct RequestVoteReply fields: term, voteGranted, peerConfigTag, + * peerLastLogLsn — all scalars; safe to copy before freeing segment. */ + memset(&vr_raw, 0, sizeof(vr_raw)); + read_RequestVoteReply(&vr_raw, rep_s.data.requestvotereplymsg); + + capn_free(&capn_ctx); /* capnp segment released; vr_raw.* still valid */ + + vr_term = (uint64_t)vr_raw.term; + vr_granted = vr_raw.voteGranted ? 1 : 0; + vr_tag = (uint64_t)vr_raw.peerConfigTag; + vr_lsn = (uint64_t)vr_raw.peerLastLogLsn; + ok = 1; + } + +done: + /* Release packet — INVOKE_MSG_HANDLER will null ep->fsm_in.pkt after return */ + if (pkt) { free_packet(pkt); free(pkt); } + + if (ok) { + /* Term fencing: if peer's term exceeds ours, update atomically */ + uint64_t my_term = atomic_load_explicit(&ha->term, memory_order_acquire); + if (vr_term > my_term) + atomic_store_explicit(&ha->term, vr_term, memory_order_release); + + vote_reply_dispatch(ha, ctx, idx, vr_term, vr_granted, vr_tag, vr_lsn); + } else { + /* Decode error: treat peer as denied */ + vote_reply_dispatch(ha, ctx, idx, + atomic_load_explicit(&ha->term, memory_order_acquire), + 0, 0, 0); + } return 0; } #ifdef XCE_TESTING + +/* + * vote_reply_build_pkt_for_test: test helper that encodes a RequestVoteReply + * into a newly allocated io_packet_t that vote_reply_msg_handler can decode. + * + * Wire format mirrors the real peer reply: + * Request{opcode=reply, body.replymsg=RequestReply{ + * data_which=requestvotereplymsg, + * data.requestvotereplymsg=RequestVoteReply{term, voteGranted, ...} + * }} + * + * Returns a freshly allocated io_packet_t* on success, NULL on failure. + * Handler takes ownership: it calls free_packet()+free() when done. + * Tests set ep->fsm_in.pkt to the returned pointer before calling the handler. + */ +io_packet_t *vote_reply_build_pkt_for_test(uint64_t term, int vote_granted, + uint64_t peer_tag, uint64_t peer_lsn) +{ + struct capn cs; + struct capn_segment *seg; + struct RequestVoteReply vr_capn; + RequestVoteReply_ptr vr_ptr; + struct RequestReply rr_capn; + RequestReply_ptr rr_ptr; + struct Request rq_capn; + Request_ptr rq_ptr; + io_packet_t *pkt; + int rc; + + capn_init_malloc(&cs); + seg = capn_root(&cs).seg; + + /* RequestVoteReply — all scalar fields */ + memset(&vr_capn, 0, sizeof(vr_capn)); + vr_capn.term = term; + vr_capn.voteGranted = vote_granted ? 1 : 0; + vr_capn.peerConfigTag = peer_tag; + vr_capn.peerLastLogLsn = peer_lsn; + vr_ptr = new_RequestVoteReply(seg); + write_RequestVoteReply(&vr_capn, vr_ptr); + + /* RequestReply envelope */ + memset(&rr_capn, 0, sizeof(rr_capn)); + rr_capn.seq = 1; + rr_capn.opcode = OPCode_requestVote; + rr_capn.status = 0; + rr_capn.data_which = RequestReply_data_requestvotereplymsg; + rr_capn.data.requestvotereplymsg = vr_ptr; + rr_ptr = new_RequestReply(seg); + write_RequestReply(&rr_capn, rr_ptr); + + /* Outer Request{opcode=reply} */ + memset(&rq_capn, 0, sizeof(rq_capn)); + rq_capn.opcode = OPCode_reply; + rq_capn.seq = 1; + rq_capn.body_which = Request_body_replymsg; + rq_capn.body.replymsg = rr_ptr; + rq_ptr = new_Request(seg); + write_Request(&rq_capn, rq_ptr); + capn_setp(capn_root(&cs), 0, rq_ptr.p); + + pkt = calloc(1, sizeof(*pkt)); + if (!pkt) { capn_free(&cs); return NULL; } + + pkt->sz = capn_size(&cs); + rc = alloc_packet(pkt); + if (rc < 0) { capn_free(&cs); free(pkt); return NULL; } + + capn_write_mem(&cs, (uint8_t *)pkt->data, (size_t)pkt->sz, 0); + capn_free(&cs); + + return pkt; +} + +/* + * vote_reply_msg_handler_for_test: expose vote_reply_msg_handler under + * XCE_TESTING so tests can call it directly (it is otherwise static). + * Sets ep->fsm_in.pkt = pkt then calls vote_reply_msg_handler(ep). + * Caller owns ep->data (vote_ep_ctx_t); pkt ownership is transferred. + */ +int vote_reply_msg_handler_for_test(endpoint_t *ep, io_packet_t *pkt) +{ + ep->fsm_in.pkt = pkt; + return vote_reply_msg_handler(ep); +} + /* * vote_ep_ctx_build_for_test: test helper that allocates a vote_ep_ctx_t, * fills it from the supplied parameters, and wires up ep->data and diff --git a/src/controller/testsuite/test_ha_vote_send.c b/src/controller/testsuite/test_ha_vote_send.c index 2c2de1a5..d8610c9f 100644 --- a/src/controller/testsuite/test_ha_vote_send.c +++ b/src/controller/testsuite/test_ha_vote_send.c @@ -318,6 +318,245 @@ static void test_election_timeout_ctx_cancelled_flag(void) printf("test_election_timeout_ctx_cancelled_flag: PASS\n"); } +/* ── Task 2.3: vote_reply_msg_handler tests ───────────────────────────────── + * + * Test seam: xce_vote_reply_hook captures decoded fields without real iomgr. + * vote_reply_build_pkt_for_test encodes a real capnp RequestVoteReply packet. + * vote_reply_msg_handler_for_test sets ep->fsm_in.pkt and calls the handler. + * + * ASAN: handler calls free_packet()+free(pkt) before accessing vr_raw fields, + * so there is no heap-use-after-free under AddressSanitizer. + */ + +/* Forward declarations for task 2.3 test seams (defined in ha.c XCE_TESTING) */ +extern int (*xce_vote_reply_hook)(ha_state_t *ha, ha_elect_ctx_t *ctx, + int peer_idx, uint64_t term, int vote_granted, + uint64_t peer_tag, uint64_t peer_lsn); + +extern io_packet_t *vote_reply_build_pkt_for_test(uint64_t term, int vote_granted, + uint64_t peer_tag, + uint64_t peer_lsn); + +extern int vote_reply_msg_handler_for_test(endpoint_t *ep, io_packet_t *pkt); + +/* ── Mock vote_reply_hook ───────────────────────────────────────────────── */ + +static int g_reply_hook_calls = 0; +static uint64_t g_reply_term = 0; +static int g_reply_granted = -1; +static uint64_t g_reply_peer_tag = 0; +static uint64_t g_reply_peer_lsn = 0; +static int g_reply_peer_idx = -1; + +static int mock_vote_reply_hook(ha_state_t *ha, ha_elect_ctx_t *ctx, + int peer_idx, uint64_t term, int vote_granted, + uint64_t peer_tag, uint64_t peer_lsn) +{ + (void)ha; (void)ctx; + g_reply_hook_calls++; + g_reply_peer_idx = peer_idx; + g_reply_term = term; + g_reply_granted = vote_granted; + g_reply_peer_tag = peer_tag; + g_reply_peer_lsn = peer_lsn; + return 0; +} + +static void reset_reply_mock(void) +{ + g_reply_hook_calls = 0; + g_reply_term = 0; + g_reply_granted = -1; + g_reply_peer_tag = 0; + g_reply_peer_lsn = 0; + g_reply_peer_idx = -1; + xce_vote_reply_hook = mock_vote_reply_hook; +} + +/* ── Helper: build a minimal vote endpoint with reply context ────────────── */ + +static void make_vote_reply_ep(endpoint_t *ep, ha_state_t *ha, + ha_elect_ctx_t *ctx, int peer_idx) +{ + memset(ep, 0, sizeof(*ep)); + ep->fd = -1; + /* Build and attach a vote_ep_ctx_t via the existing test helper */ + vote_ep_ctx_build_for_test(ep, ha, ctx, peer_idx, + atomic_load_explicit(&ha->term, memory_order_acquire), + 100ULL, 50ULL, "ctrl-a"); +} + +/* ── Test: handler decodes correct term, granted, tag, lsn ─────────────── */ + +static void test_vote_reply_handler_decodes_correct_fields(void) +{ + reset_reply_mock(); + + ha_state_t ha; + ha_init(&ha, &g_vote_cfg, 0); + atomic_store_explicit(&ha.term, 5ULL, memory_order_release); + + ha_elect_ctx_t *ctx = alloc_ctx_for_vote(2); + pthread_mutex_init(&ctx->lock, NULL); + + endpoint_t ep; + make_vote_reply_ep(&ep, &ha, ctx, 0); + + /* Build a real capnp packet with known fields */ + io_packet_t *pkt = vote_reply_build_pkt_for_test( + 5ULL, /* term */ + 1, /* vote_granted = true */ + 200ULL, /* peer_tag */ + 99ULL /* peer_lsn */); + TEST_ASSERT_NOT_NULL_MESSAGE(pkt, "vote_reply_build_pkt_for_test must succeed"); + + /* Call handler — it takes ownership of pkt; frees it internally */ + int rc = vote_reply_msg_handler_for_test(&ep, pkt); + TEST_ASSERT_EQUAL_INT(0, rc); + + /* Verify hook was called exactly once */ + TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_reply_hook_calls, + "vote_reply_msg_handler must call hook exactly once"); + + /* Verify decoded fields */ + TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_reply_peer_idx, + "peer_idx must match (0)"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(5ULL, g_reply_term, + "decoded term must be 5"); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_reply_granted, + "decoded vote_granted must be 1"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(200ULL, g_reply_peer_tag, + "decoded peer_tag must be 200"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(99ULL, g_reply_peer_lsn, + "decoded peer_lsn must be 99"); + + /* ep->fsm_in.pkt is NOT freed here — handler already freed it */ + + free(ep.data); /* vote_ep_ctx_t allocated by build helper */ + free_ctx_for_vote(ctx); + ha_fini(&ha); + printf("test_vote_reply_handler_decodes_correct_fields: PASS\n"); +} + +/* ── Test: vote_granted=false correctly decoded ──────────────────────────── */ + +static void test_vote_reply_handler_denied_vote(void) +{ + reset_reply_mock(); + + ha_state_t ha; + ha_init(&ha, &g_vote_cfg, 0); + atomic_store_explicit(&ha.term, 3ULL, memory_order_release); + + ha_elect_ctx_t *ctx = alloc_ctx_for_vote(2); + pthread_mutex_init(&ctx->lock, NULL); + + endpoint_t ep; + make_vote_reply_ep(&ep, &ha, ctx, 1); + + io_packet_t *pkt = vote_reply_build_pkt_for_test( + 3ULL, /* term */ + 0, /* vote_granted = false */ + 50ULL, /* peer_tag */ + 20ULL /* peer_lsn */); + TEST_ASSERT_NOT_NULL_MESSAGE(pkt, "vote_reply_build_pkt_for_test must succeed"); + + int rc = vote_reply_msg_handler_for_test(&ep, pkt); + TEST_ASSERT_EQUAL_INT(0, rc); + + TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_reply_hook_calls, + "hook must be called once even for denied vote"); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, g_reply_granted, + "vote_granted must be 0 for denied vote"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(3ULL, g_reply_term, + "term must be decoded correctly even when denied"); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_reply_peer_idx, + "peer_idx must be 1"); + + free(ep.data); + free_ctx_for_vote(ctx); + ha_fini(&ha); + printf("test_vote_reply_handler_denied_vote: PASS\n"); +} + +/* ── Test: term fencing — higher peer term updates ha->term atomically ───── */ + +static void test_vote_reply_handler_term_fencing(void) +{ + reset_reply_mock(); + + ha_state_t ha; + ha_init(&ha, &g_vote_cfg, 0); + atomic_store_explicit(&ha.term, 3ULL, memory_order_release); + + ha_elect_ctx_t *ctx = alloc_ctx_for_vote(2); + pthread_mutex_init(&ctx->lock, NULL); + + endpoint_t ep; + make_vote_reply_ep(&ep, &ha, ctx, 0); + + /* Peer claims term 10 — higher than our term 3 */ + io_packet_t *pkt = vote_reply_build_pkt_for_test( + 10ULL, /* peer term > our term */ + 0, /* denied */ + 0ULL, + 0ULL); + TEST_ASSERT_NOT_NULL_MESSAGE(pkt, "vote_reply_build_pkt_for_test must succeed"); + + vote_reply_msg_handler_for_test(&ep, pkt); + + /* ha->term must now be updated to 10 */ + uint64_t updated_term = atomic_load_explicit(&ha.term, memory_order_acquire); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(10ULL, updated_term, + "ha->term must be atomically updated to peer term when peer term > ours"); + + free(ep.data); + free_ctx_for_vote(ctx); + ha_fini(&ha); + printf("test_vote_reply_handler_term_fencing: PASS\n"); +} + +/* ── Test: no UAF — pkt is freed before fields are used (ASAN observable) ── */ + +static void test_vote_reply_handler_no_uaf(void) +{ + /* This test proves ASAN-safety: vote_reply_msg_handler reads ALL capnp + * scalars into vr_term/vr_granted/vr_tag/vr_lsn BEFORE calling + * free_packet()+free(pkt). Under ASAN (enabled in xce.build via + * -fsanitize=address), accessing freed memory would fault immediately. + * If this test passes under ASAN, there is no heap-use-after-free. */ + reset_reply_mock(); + + ha_state_t ha; + ha_init(&ha, &g_vote_cfg, 0); + atomic_store_explicit(&ha.term, 2ULL, memory_order_release); + + ha_elect_ctx_t *ctx = alloc_ctx_for_vote(2); + pthread_mutex_init(&ctx->lock, NULL); + + endpoint_t ep; + make_vote_reply_ep(&ep, &ha, ctx, 0); + + io_packet_t *pkt = vote_reply_build_pkt_for_test( + 2ULL, 1, 777ULL, 888ULL); + TEST_ASSERT_NOT_NULL_MESSAGE(pkt, "pkt allocation must succeed"); + + /* Handler frees pkt internally; all field reads happen before the free */ + int rc = vote_reply_msg_handler_for_test(&ep, pkt); + TEST_ASSERT_EQUAL_INT(0, rc); + + /* Hook captured the fields — verify they are the values we encoded */ + TEST_ASSERT_EQUAL_UINT64_MESSAGE(2ULL, g_reply_term, "term matches"); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_reply_granted, "granted matches"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(777ULL, g_reply_peer_tag, "peer_tag matches"); + TEST_ASSERT_EQUAL_UINT64_MESSAGE(888ULL, g_reply_peer_lsn, "peer_lsn matches"); + + free(ep.data); + free_ctx_for_vote(ctx); + ha_fini(&ha); + printf("test_vote_reply_handler_no_uaf: PASS\n"); +} + /* ── Suite entry point ───────────────────────────────────────────────────── */ void test_ha_vote_send(void) @@ -326,4 +565,9 @@ void test_ha_vote_send(void) RUN_TEST(test_on_vote_connected_encodes_correct_fields); RUN_TEST(test_on_vote_connected_sets_msg_handler); RUN_TEST(test_election_timeout_ctx_cancelled_flag); + /* Task 2.3: vote_reply_msg_handler */ + RUN_TEST(test_vote_reply_handler_decodes_correct_fields); + RUN_TEST(test_vote_reply_handler_denied_vote); + RUN_TEST(test_vote_reply_handler_term_fencing); + RUN_TEST(test_vote_reply_handler_no_uaf); }