diff --git a/cpworker/tests/integration/control_socket_test.go b/cpworker/tests/integration/control_socket_test.go index 5722130d..7007aced 100644 --- a/cpworker/tests/integration/control_socket_test.go +++ b/cpworker/tests/integration/control_socket_test.go @@ -4,6 +4,7 @@ import ( "bufio" "encoding/json" "fmt" + "io" "net" "os" "path/filepath" @@ -87,36 +88,69 @@ func (c *controlConn) send(gap time.Duration, parts ...string) error { return nil } -// expectOK reads one reply line and checks that its status is OK. -func (c *controlConn) expectOK() error { +// readReply reads one reply line and decodes it. +func (c *controlConn) readReply() (map[string]any, error) { c.conn.SetReadDeadline(time.Now().Add(controlIOTimeout)) line, err := c.r.ReadBytes('\n') if err != nil { - return fmt.Errorf("read reply: %w", err) + return nil, fmt.Errorf("read reply: %w", err) } var resp map[string]any if err := json.Unmarshal(line, &resp); err != nil { - return fmt.Errorf("decode reply %q: %w", line, err) + return nil, fmt.Errorf("decode reply %q: %w", line, err) + } + return resp, nil +} + +// expectOK reads one reply line and checks that its status is OK. +func (c *controlConn) expectOK() error { + resp, err := c.readReply() + if err != nil { + return err } if resp["status"] != "OK" { - return fmt.Errorf("reply status is not OK: %s", line) + return fmt.Errorf("reply status is not OK: %v", resp) } return nil } -// handshakeAndPing runs the v1 handshake and one ping on a new connection. -func handshakeAndPing(sock string) error { +// expectClosed checks that the manager closes the connection without replying. +func (c *controlConn) expectClosed() error { + c.conn.SetReadDeadline(time.Now().Add(controlIOTimeout)) + line, err := c.r.ReadBytes('\n') + if err == io.EOF && len(line) == 0 { + return nil + } + if err != nil { + return fmt.Errorf("want the connection closed, got: %w", err) + } + return fmt.Errorf("want the connection closed, got reply %q", line) +} + +// dialHandshaken opens a connection and runs the v1 handshake on it. +func dialHandshaken(sock string) (*controlConn, error) { c, err := dialControl(sock) if err != nil { - return fmt.Errorf("dial: %w", err) + return nil, fmt.Errorf("dial: %w", err) } - defer c.conn.Close() if err := c.send(0, `{"version":"v1"}`+"\n"); err != nil { - return fmt.Errorf("send handshake: %w", err) + c.conn.Close() + return nil, fmt.Errorf("send handshake: %w", err) } if err := c.expectOK(); err != nil { - return fmt.Errorf("handshake: %w", err) + c.conn.Close() + return nil, fmt.Errorf("handshake: %w", err) + } + return c, nil +} + +// handshakeAndPing runs the v1 handshake and one ping on a new connection. +func handshakeAndPing(sock string) error { + c, err := dialHandshaken(sock) + if err != nil { + return err } + defer c.conn.Close() if err := c.send(0, `{"command":"ping"}`+"\n"); err != nil { return fmt.Errorf("send ping: %w", err) } @@ -134,6 +168,8 @@ func handshakeAndPing(sock string) error { // - request_split_across_writes: a request that arrives in several writes is answered (#261) // - silent_client_does_not_block_others: a client that connects and never sends the // handshake does not stop the manager from serving other clients +// - rejected_requests: a bad handshake, a request that is not JSON or has no string +// "command" closes that connection; an unknown command is answered with ERROR func TestControlSocket(t *testing.T) { if os.Getenv("CPWORKER_BIN") == "" { t.Skip("CPWORKER_BIN not set") @@ -208,4 +244,63 @@ func TestControlSocket(t *testing.T) { t.Errorf("client behind a silent connection: %v", err) } }) + + t.Run("rejected_requests", func(t *testing.T) { + sock := startControlWorker(t, "rejected_requests") + + for _, hs := range []string{`not json`, `{"version":"v2"}`, `{"version":1}`, `{}`} { + c, err := dialControl(sock) + if err != nil { + t.Fatalf("dial: %v", err) + } + if err := c.send(0, hs+"\n"); err != nil { + t.Fatalf("send handshake %s: %v", hs, err) + } + if err := c.expectClosed(); err != nil { + t.Errorf("handshake %s: %v", hs, err) + } + c.conn.Close() + } + + // An unknown command is answered, and the connection stays usable. + c, err := dialHandshaken(sock) + if err != nil { + t.Fatal(err) + } + if err := c.send(0, `{"command":"no_such_command"}`+"\n"); err != nil { + t.Fatalf("send unknown command: %v", err) + } + resp, err := c.readReply() + if err != nil { + t.Fatalf("unknown command: %v", err) + } + if resp["status"] != "ERROR" || resp["message"] != "unknown command" { + t.Errorf("unknown command: got %v, want status ERROR, message \"unknown command\"", resp) + } + if err := c.send(0, `{"command":"ping"}`+"\n"); err != nil { + t.Fatalf("send ping: %v", err) + } + if err := c.expectOK(); err != nil { + t.Errorf("ping after unknown command: %v", err) + } + c.conn.Close() + + for _, req := range []string{`not json`, `{"command":42}`, `{}`} { + c, err := dialHandshaken(sock) + if err != nil { + t.Fatal(err) + } + if err := c.send(0, req+"\n"); err != nil { + t.Fatalf("send request %s: %v", req, err) + } + if err := c.expectClosed(); err != nil { + t.Errorf("request %s: %v", req, err) + } + c.conn.Close() + } + + if err := handshakeAndPing(sock); err != nil { + t.Errorf("client after the rejected requests: %v", err) + } + }) } diff --git a/cpworker/tests/unit/output_zmq.c b/cpworker/tests/unit/output_zmq.c index a4db13ae..476f9c46 100644 --- a/cpworker/tests/unit/output_zmq.c +++ b/cpworker/tests/unit/output_zmq.c @@ -545,6 +545,75 @@ void test_rate_limit_charges_and_counts_sliced_frame_length(void) TEST_ASSERT_EQUAL_UINT64(1000, stats.ratelimit_drop_bytes.bytes); } +/* ---- docs/ZMQ-WIRE-FORMAT.md ยง8: batch flush rules ---- */ + +// A batch is flushed only when the next record would take it past ZMQ_MAX_BATCH_BUF_SIZE. A +// 2451-byte frame makes a 2473-byte record (length, packet header, frame, MPLS label), and 424 +// of them after the batch header fill the buffer exactly. +void test_batch_flushed_when_next_record_exceeds_buffer(void) +{ + char errbuf[ERROR_BUFFER_SIZE] = {0}; + zmq_output_t *output = new_output(0, (char *)VALID_UUID, errbuf); + TEST_ASSERT_NOT_NULL_MESSAGE(output, errbuf); + + static uint8_t frame[2451]; + memset(frame, 0x5a, sizeof(frame)); + frame[12] = 0x08; + frame[13] = 0x00; + + for (int i = 0; i < 424; i++) + TEST_ASSERT_EQUAL_INT(0, send_frame(output, frame, sizeof(frame))); + TEST_ASSERT_EQUAL_UINT32(ZMQ_MAX_BATCH_BUF_SIZE, output->pkts_buf.batch_bufpos); + TEST_ASSERT_EQUAL_UINT16(424, output->pkts_buf.batch_hdr.pkts_num); + + TEST_ASSERT_EQUAL_INT(0, send_frame(output, frame, sizeof(frame))); + TEST_ASSERT_EQUAL_UINT16(1, output->pkts_buf.batch_hdr.pkts_num); + TEST_ASSERT_EQUAL_INT(ZMQ_MAX_BATCH_BUF_SIZE, recv_batch()); + TEST_ASSERT_EQUAL_UINT16(424, batch_pkts_num()); + + zmq_output_destroy(&output->base); +} + +// A batch is flushed when a packet arrives more than ZMQ_PKTS_FLUSH_MAX_DUR_SEC after its first +// packet; that packet starts the next batch. +void test_batch_flushed_when_packet_is_over_a_second_later(void) +{ + char errbuf[ERROR_BUFFER_SIZE] = {0}; + zmq_output_t *output = new_output(0, (char *)VALID_UUID, errbuf); + TEST_ASSERT_NOT_NULL_MESSAGE(output, errbuf); + + send_frame_at(output, 100); + send_frame_at(output, 101); + TEST_ASSERT_EQUAL_UINT16(2, output->pkts_buf.batch_hdr.pkts_num); + + send_frame_at(output, 102); + TEST_ASSERT_EQUAL_UINT16(1, output->pkts_buf.batch_hdr.pkts_num); + TEST_ASSERT_GREATER_THAN_INT(0, recv_batch()); + TEST_ASSERT_EQUAL_UINT16(2, batch_pkts_num()); + + zmq_output_destroy(&output->base); +} + +// With no traffic, the periodic zmq_heartbeat() call flushes a batch once its first packet is +// more than ZMQ_PKTS_FLUSH_MAX_DUR_SEC old, even when heartbeats are disabled. +void test_idle_batch_flushed_after_a_second(void) +{ + char errbuf[ERROR_BUFFER_SIZE] = {0}; + zmq_output_t *output = new_output(0, (char *)VALID_UUID, errbuf); + TEST_ASSERT_NOT_NULL_MESSAGE(output, errbuf); + + send_frame_at(output, 100); + zmq_heartbeat(&output->base, 101); + TEST_ASSERT_EQUAL_UINT16(1, output->pkts_buf.batch_hdr.pkts_num); + + zmq_heartbeat(&output->base, 102); + TEST_ASSERT_EQUAL_UINT16(0, output->pkts_buf.batch_hdr.pkts_num); + TEST_ASSERT_GREATER_THAN_INT(0, recv_batch()); + TEST_ASSERT_EQUAL_UINT16(1, batch_pkts_num()); + + zmq_output_destroy(&output->base); +} + int main(void) { UNITY_BEGIN(); @@ -573,5 +642,9 @@ int main(void) RUN_TEST(test_counters_use_sliced_frame_length_without_headers); RUN_TEST(test_rate_limit_charges_and_counts_sliced_frame_length); + RUN_TEST(test_batch_flushed_when_next_record_exceeds_buffer); + RUN_TEST(test_batch_flushed_when_packet_is_over_a_second_later); + RUN_TEST(test_idle_batch_flushed_after_a_second); + return UNITY_END(); } diff --git a/cpworker/tests/unit/packet_split.c b/cpworker/tests/unit/packet_split.c index 95c8beb3..18863926 100644 --- a/cpworker/tests/unit/packet_split.c +++ b/cpworker/tests/unit/packet_split.c @@ -542,6 +542,76 @@ void test_udp_checksum_ipv6_roundtrip(void) TEST_ASSERT_EQUAL_HEX16(0, calculate_udp_checksum(NULL, ip6, udp, udp_len)); } +/* ====================================================================== + * Tests: calculate_tcp_checksum / calculate_udp_checksum, known-answer vectors + * + * The segment is built by the helpers above with payload byte i = i * 13 + 1. The expected + * values come from an independent RFC 1071 implementation in Python over the RFC 793 / RFC 768 + * pseudo-header (RFC 8200 section 8.1 for IPv6), not from calculate_*_checksum. The two lengths + * per case reach the scalar loops and the SSE2 loop, and both end on an odd byte. + * ====================================================================== */ +static uint16_t known_answer_checksum(bool ipv6, uint8_t protocol, uint16_t payload_len) +{ + uint8_t pkt[2048]; + uint8_t payload[64]; + uint16_t i; + for (i = 0; i < payload_len; i++) + payload[i] = (uint8_t)(i * 13 + 1); + + uint16_t ip_offset = sizeof(struct ether_header); + if (ipv6) + { + struct in6_addr saddr, daddr; + inet_pton(AF_INET6, "2001:db8:1:2:3:4:5:6", &saddr); + inet_pton(AF_INET6, "2001:db8:a:b:c:d:e:f", &daddr); + struct ipv6_hdr *ip6 = (struct ipv6_hdr *)(pkt + ip_offset); + uint8_t *l4 = pkt + ip_offset + sizeof(struct ipv6_hdr); + if (protocol == IPPROTO_TCP) + { + build_ipv6_tcp_packet(pkt, payload, payload_len, &saddr, &daddr, 49152, 443, 0x01020304); + return ntohs(calculate_tcp_checksum(NULL, ip6, (struct tcphdr *)l4, 20 + payload_len)); + } + build_ipv6_udp_packet(pkt, payload, payload_len, &saddr, &daddr, 49152, 53); + return ntohs(calculate_udp_checksum(NULL, ip6, (struct udphdr *)l4, sizeof(struct udphdr) + payload_len)); + } + + uint32_t saddr = htonl(0xC0000201); /* 192.0.2.1 */ + uint32_t daddr = htonl(0xC6336402); /* 198.51.100.2 */ + struct ipv4_hdr *ip = (struct ipv4_hdr *)(pkt + ip_offset); + uint8_t *l4 = pkt + ip_offset + 20; + if (protocol == IPPROTO_TCP) + { + build_ipv4_tcp_packet(pkt, payload, payload_len, saddr, daddr, 49152, 443, 0x01020304); + return ntohs(calculate_tcp_checksum(ip, NULL, (struct tcphdr *)l4, 20 + payload_len)); + } + build_ipv4_udp_packet(pkt, payload, payload_len, saddr, daddr, 49152, 53); + return ntohs(calculate_udp_checksum(ip, NULL, (struct udphdr *)l4, sizeof(struct udphdr) + payload_len)); +} + +void test_tcp_checksum_ipv4_known_answer(void) +{ + TEST_ASSERT_EQUAL_HEX16(0x5D5C, known_answer_checksum(false, IPPROTO_TCP, 7)); + TEST_ASSERT_EQUAL_HEX16(0xA2C0, known_answer_checksum(false, IPPROTO_TCP, 59)); +} + +void test_tcp_checksum_ipv6_known_answer(void) +{ + TEST_ASSERT_EQUAL_HEX16(0xEDC1, known_answer_checksum(true, IPPROTO_TCP, 7)); + TEST_ASSERT_EQUAL_HEX16(0x3326, known_answer_checksum(true, IPPROTO_TCP, 59)); +} + +void test_udp_checksum_ipv4_known_answer(void) +{ + TEST_ASSERT_EQUAL_HEX16(0xB320, known_answer_checksum(false, IPPROTO_UDP, 19)); + TEST_ASSERT_EQUAL_HEX16(0xF81A, known_answer_checksum(false, IPPROTO_UDP, 59)); +} + +void test_udp_checksum_ipv6_known_answer(void) +{ + TEST_ASSERT_EQUAL_HEX16(0x4386, known_answer_checksum(true, IPPROTO_UDP, 19)); + TEST_ASSERT_EQUAL_HEX16(0x8880, known_answer_checksum(true, IPPROTO_UDP, 59)); +} + /* ====================================================================== * Tests: parse_packet * ====================================================================== */ @@ -1159,6 +1229,60 @@ void test_parse_packet_ipv6_unsplittable_ext_rejected(void) } } +/* ====================================================================== + * Tests: parse_packet, frames that end where the L4 header ends + * + * A pure ACK or an empty UDP datagram has no payload, so the captured frame ends at the last + * header byte. It parses with an empty payload; one byte less does not parse. + * ====================================================================== */ +static void assert_header_only_frame(const uint8_t *pkt, int header_end) +{ + packet_parse_result_t result; + TEST_ASSERT_TRUE(parse_packet(pkt, header_end, &result)); + TEST_ASSERT_EQUAL_UINT16(header_end, result.payload_offset); + TEST_ASSERT_EQUAL_UINT16(0, result.payload_len); + + TEST_ASSERT_FALSE(parse_packet(pkt, header_end - 1, &result)); +} + +void test_parse_packet_ipv4_header_only_frames(void) +{ + uint8_t pkt[2048]; + + TEST_ASSERT_EQUAL_INT(14 + 20 + 20, build_ipv4_tcp_packet(pkt, NULL, 0, 1, 2, 80, 80, 0)); + assert_header_only_frame(pkt, 14 + 20 + 20); + + TEST_ASSERT_EQUAL_INT(14 + 20 + 8, build_ipv4_udp_packet(pkt, NULL, 0, 1, 2, 53, 53)); + assert_header_only_frame(pkt, 14 + 20 + 8); + + TEST_ASSERT_EQUAL_INT(14 + 4 + 20 + 20, build_vlan_ipv4_tcp_packet(pkt, NULL, 0, 100)); + assert_header_only_frame(pkt, 14 + 4 + 20 + 20); + + /* TCP with 12 bytes of options: data offset 8 */ + uint8_t options[12] = {0}; + build_ipv4_tcp_packet(pkt, options, sizeof(options), 1, 2, 80, 80, 0); + ((struct tcphdr *)(pkt + 14 + 20))->offx2 = (8 << 4); + assert_header_only_frame(pkt, 14 + 20 + 32); +} + +void test_parse_packet_ipv6_header_only_frames(void) +{ + uint8_t pkt[2048]; + struct in6_addr saddr, daddr; + inet_pton(AF_INET6, "2001:db8::1", &saddr); + inet_pton(AF_INET6, "2001:db8::2", &daddr); + + TEST_ASSERT_EQUAL_INT(14 + 40 + 20, build_ipv6_tcp_packet(pkt, NULL, 0, &saddr, &daddr, 80, 80, 0)); + assert_header_only_frame(pkt, 14 + 40 + 20); + + TEST_ASSERT_EQUAL_INT(14 + 40 + 8, build_ipv6_udp_packet(pkt, NULL, 0, &saddr, &daddr, 53, 53)); + assert_header_only_frame(pkt, 14 + 40 + 8); + + const uint8_t exts[] = {IPPROTO_DSTOPTS}; + TEST_ASSERT_EQUAL_INT(14 + 40 + 8 + 20, build_ipv6_ext_tcp_packet(pkt, exts, 1, 0)); + assert_header_only_frame(pkt, 14 + 40 + 8 + 20); +} + /* ====================================================================== * Tests: build_fragment checksums, known-answer vectors * @@ -1262,6 +1386,12 @@ int main(void) RUN_TEST(test_udp_checksum_ipv4_empty_payload); RUN_TEST(test_udp_checksum_ipv6_roundtrip); + /* TCP / UDP checksum: known-answer vectors */ + RUN_TEST(test_tcp_checksum_ipv4_known_answer); + RUN_TEST(test_tcp_checksum_ipv6_known_answer); + RUN_TEST(test_udp_checksum_ipv4_known_answer); + RUN_TEST(test_udp_checksum_ipv6_known_answer); + /* parse_packet */ RUN_TEST(test_parse_packet_ipv4_tcp); RUN_TEST(test_parse_packet_ipv4_udp); @@ -1292,6 +1422,8 @@ int main(void) RUN_TEST(test_parse_packet_ipv4_df_is_not_a_fragment); RUN_TEST(test_parse_packet_ipv6_skips_hopopts_routing_dstopts); RUN_TEST(test_parse_packet_ipv6_unsplittable_ext_rejected); + RUN_TEST(test_parse_packet_ipv4_header_only_frames); + RUN_TEST(test_parse_packet_ipv6_header_only_frames); /* build_fragment: checksum known-answer vectors */ RUN_TEST(test_build_fragment_ipv4_udp_zero_checksum_sent_as_ffff); diff --git a/cpworker/tests/unit/unix_control.c b/cpworker/tests/unit/unix_control.c index 5fe53780..0242b130 100644 --- a/cpworker/tests/unit/unix_control.c +++ b/cpworker/tests/unit/unix_control.c @@ -216,11 +216,13 @@ static int64_t json_time_sec(cJSON *resp, const char *key) void test_stats_summary_has_wall_time_and_monotonic_time(void) { - struct timespec mono_before, mono_after; + // Bracket the snapshot with the clocks it reads: time() may use the coarse realtime clock, + // which lags CLOCK_REALTIME by up to a tick and so can still show the previous second. + struct timespec mono_before, mono_after, wall_before, wall_after; clock_gettime(CLOCK_MONOTONIC, &mono_before); - time_t wall_before = time(NULL); + clock_gettime(CLOCK_REALTIME, &wall_before); task_manager_update_stats(); - time_t wall_after = time(NULL); + clock_gettime(CLOCK_REALTIME, &wall_after); clock_gettime(CLOCK_MONOTONIC, &mono_after); cJSON *resp = cJSON_CreateObject(); @@ -228,8 +230,8 @@ void test_stats_summary_has_wall_time_and_monotonic_time(void) TEST_ASSERT_EQUAL(0, task_manager_collect_stats_summary_command(NULL, resp, NULL)); int64_t wall = json_time_sec(resp, "wall_time"); - TEST_ASSERT_GREATER_OR_EQUAL_INT64((int64_t)wall_before, wall); - TEST_ASSERT_LESS_OR_EQUAL_INT64((int64_t)wall_after, wall); + TEST_ASSERT_GREATER_OR_EQUAL_INT64((int64_t)wall_before.tv_sec, wall); + TEST_ASSERT_LESS_OR_EQUAL_INT64((int64_t)wall_after.tv_sec, wall); int64_t mono = json_time_sec(resp, "time"); TEST_ASSERT_GREATER_OR_EQUAL_INT64((int64_t)mono_before.tv_sec, mono);