ryerraguntla commented on code in PR #3519: URL: https://github.com/apache/iggy/pull/3519#discussion_r3812622523
########## gateways/kafka/tests/version_firewall_tests.rs: ########## @@ -0,0 +1,742 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Version negotiation firewall - boundary tests for every scoped API key. + +#[path = "common/codec.rs"] +mod codec; +#[path = "common/fixtures.rs"] +mod fixtures; +#[path = "common/scope.rs"] +mod scope; +#[path = "common/server.rs"] +mod server; +#[path = "common/tcp.rs"] +mod tcp; +#[path = "common/wire.rs"] +mod wire; + +use std::time::Duration; + +use bytes::Bytes; +use tokio::io::AsyncWriteExt; +use tokio::net::TcpStream; + +use iggy_gateway_kafka::protocol::api::{ + API_KEY_API_VERSIONS, API_KEY_CREATE_TOPICS, API_KEY_FETCH, API_KEY_LIST_OFFSETS, + API_KEY_METADATA, API_KEY_PRODUCE, ERROR_INVALID_REQUEST, ERROR_NONE, + ERROR_UNSUPPORTED_VERSION, advertised_min_version, handle_request, is_supported_version, + supported_api_ranges, +}; + +use codec::Decoder; +use fixtures::{fixture_exists, load_fixture_body, load_fixture_body_or_skip}; +use scope::{SCOPED_API_KEYS, default_broker}; +use server::spawn_test_server; +use tcp::{ + ByteRead, build_list_offsets_v0_request_with_topic_t, build_metadata_legacy_request, + build_produce_flexible_body, build_produce_v2_body, build_produce_v3_body, build_request_frame, + parse_response_payload, read_byte_with_timeout, round_trip, scan_for_error_code, +}; +use wire::{ + OUT_OF_SCOPE_API_KEYS, build_api_versions_flexible_request, build_create_topics_empty_request, + build_fetch_empty_topics_request, build_list_offsets_request, + build_metadata_all_topics_flexible, build_metadata_all_topics_legacy, + build_metadata_flexible_request_v10, +}; + +#[test] +fn supported_ranges_table_has_six_entries() { + assert_eq!(supported_api_ranges().len(), 6); +} + +#[test] +fn is_supported_version_matches_scope_table() { + for &(api_key, _, min_ver, max_ver) in SCOPED_API_KEYS { + assert!( + !is_supported_version(api_key, min_ver - 1), + "key {api_key} must reject v{}", + min_ver - 1 + ); + assert!( + is_supported_version(api_key, min_ver), + "key {api_key} must accept min v{min_ver}" + ); + assert!( + is_supported_version(api_key, max_ver), + "key {api_key} must accept max v{max_ver}" + ); + assert!( + !is_supported_version(api_key, max_ver + 1), + "key {api_key} must reject v{}", + max_ver + 1 + ); + } +} + +#[test] +fn apiversions_advertises_exact_supported_ranges_v1() { + let body = handle_request(API_KEY_API_VERSIONS, 1, Bytes::new(), &default_broker()) + .expect_response("test request has acks != 0 and expects a response"); + let mut d = Decoder::new(body); + assert_eq!(d.read_i16().unwrap(), 0); + let count = usize::try_from(d.read_i32().unwrap()).expect("api count fits usize"); + assert_eq!(count, supported_api_ranges().len()); + + for expected in supported_api_ranges() { + let key = d.read_i16().unwrap(); + let min = d.read_i16().unwrap(); + let max = d.read_i16().unwrap(); + assert_eq!(key, expected.api_key); + assert_eq!( + min, + advertised_min_version(expected.api_key, expected.min_version) + ); + assert_eq!(max, expected.max_version); + } + assert_eq!(d.read_i32().unwrap(), 0); // throttle + assert_eq!(d.remaining(), 0); +} + +#[test] +fn apiversions_advertises_exact_supported_ranges_v3_flexible() { + let body = handle_request( + API_KEY_API_VERSIONS, + 3, + build_api_versions_flexible_request("iggy-test", "0.1.0"), + &default_broker(), + ) + .expect_response("test request has acks != 0 and expects a response"); + let mut d = Decoder::new(body); + assert_eq!(d.read_i16().unwrap(), 0); + let count = usize::try_from(d.read_varint().unwrap() - 1).expect("api count fits usize"); + assert_eq!(count, supported_api_ranges().len()); + + for expected in supported_api_ranges() { + let key = d.read_i16().unwrap(); + let min = d.read_i16().unwrap(); + let max = d.read_i16().unwrap(); + d.read_tagged_fields().unwrap(); + assert_eq!(key, expected.api_key); + assert_eq!( + min, + advertised_min_version(expected.api_key, expected.min_version) + ); + assert_eq!(max, expected.max_version); + } + assert_eq!(d.read_i32().unwrap(), 0); + d.read_tagged_fields().unwrap(); + assert_eq!(d.remaining(), 0); +} + +#[test] +fn apiversions_advertises_produce_min_zero_while_firewall_stays_three() { + let range = supported_api_ranges() + .iter() + .find(|r| r.api_key == API_KEY_PRODUCE) + .expect("produce range"); + assert_eq!(range.min_version, 3); + assert_eq!( + advertised_min_version(API_KEY_PRODUCE, range.min_version), + 0 + ); + assert!(!is_supported_version(API_KEY_PRODUCE, 0)); +} + +#[test] +fn apiversions_all_versions_return_success() { + for version in 0i16..=3 { + let request = if version >= 3 { + build_api_versions_flexible_request("iggy-test", "0.1.0") + } else { + Bytes::new() + }; + let body = handle_request(API_KEY_API_VERSIONS, version, request, &default_broker()) + .expect_response("test request has acks != 0 and expects a response"); + let mut d = Decoder::new(body); + assert_eq!(d.read_i16().unwrap(), 0, "ApiVersions v{version}"); + } +} + +#[test] +fn apiversions_out_of_range_returns_unsupported_in_body() { + let body = handle_request(API_KEY_API_VERSIONS, 99, Bytes::new(), &default_broker()) + .expect_response("test request has acks != 0 and expects a response"); + let mut d = Decoder::new(body); + assert_eq!(d.read_i16().unwrap(), ERROR_UNSUPPORTED_VERSION); +} + +fn metadata_request_one_topic() -> Bytes { + build_metadata_legacy_request(&["test-topic"]) +} + +#[test] +fn metadata_below_min_version_closes_connection() { + assert!( + handle_request( + API_KEY_METADATA, + -1, + metadata_request_one_topic(), + &default_broker(), + ) + .is_close(), + "Metadata below supported min must close rather than return a clamped body" + ); +} + +#[test] +fn metadata_above_max_version_closes_connection() { + // v10 request uses flexible encoding; a clamped v9 reply would not survive client parsing. + assert!( + handle_request( + API_KEY_METADATA, + 10, + build_metadata_flexible_request_v10(&["test-topic"]), + &default_broker(), + ) + .is_close(), + "Metadata above supported max must close rather than return a clamped body" + ); +} + +#[tokio::test] +async fn e2e_metadata_above_max_version_closes_tcp_connection() { + let (addr, _shutdown) = spawn_test_server().await; + let mut stream = TcpStream::connect(addr).await.expect("connect"); + + let body = build_metadata_flexible_request_v10(&["orders"]); + let frame = build_request_frame(API_KEY_METADATA, 10, 44, Some("n9-test"), &body); + stream.write_all(&frame).await.expect("write metadata v10"); + + assert_eq!( + read_byte_with_timeout(&mut stream, Duration::from_secs(2)).await, + ByteRead::Closed, + "unsupported Metadata version must close the connection" + ); +} + +#[test] +fn produce_below_min_version_with_nonzero_acks_closes_connection() { + // Produce v2 is below both the firewall min (3) and `kafka_protocol`'s schema floor (3-13) + // - no encodable response exists at this version, so a client expecting a reply (acks != 0) + // gets a close instead of the pre-migration downgraded error response. acks=0 still keeps + // the connection open - see `produce_advertises_min_zero_but_firewall_rejects_below_v3` and + // `api::handle_produce_request`'s hand-peeked acks path. + let body = handle_request( + API_KEY_PRODUCE, + 2, + build_produce_v2_body(1, 0), + &default_broker(), + ); + assert!( + body.is_close(), + "Produce v2 with acks != 0 has no encodable response shape and must close" + ); +} + +#[test] +fn fetch_below_min_version_closes_connection() { + // Fetch v3 is below both the firewall min (4) and `kafka_protocol`'s schema floor (4-18) - + // no encodable response exists at this version, so this closes instead of the + // pre-migration downgraded error response. + assert!( + handle_request(API_KEY_FETCH, 3, Bytes::new(), &default_broker()).is_close(), + "Fetch v3 has no encodable response shape and must close" + ); +} + +#[test] +fn fetch_unsupported_version_above_max_closes_connection() { + // Fetch v13+ response shape differs from the v12 encoder; a clamped body is unparsable. + assert!( + handle_request(API_KEY_FETCH, 13, Bytes::new(), &default_broker()).is_close(), + "Fetch above encoder max must close rather than return a clamped body" + ); +} + +#[test] +fn produce_unsupported_version_above_max_closes_connection() { + assert!( + handle_request(API_KEY_PRODUCE, 13, Bytes::new(), &default_broker()).is_close(), + "Produce above encoder max must close rather than return a clamped body" + ); +} + +#[test] +fn create_topics_unsupported_version_above_max_closes_connection() { + assert!( + handle_request(API_KEY_CREATE_TOPICS, 7, Bytes::new(), &default_broker()).is_close(), + "CreateTopics above encoder max must close rather than return a clamped body" + ); +} + +#[test] +fn list_offsets_unsupported_version_above_max_closes_connection() { + assert!( + handle_request(API_KEY_LIST_OFFSETS, 7, Bytes::new(), &default_broker()).is_close(), + "ListOffsets above encoder max must close rather than return a clamped body" + ); +} + +#[test] +fn list_offsets_v0_closes_connection() { + // `kafka_protocol` has no encoder for ListOffsets v0's legacy `old_style_offsets` shape (it + // predates the schema the crate generates from), so a v0 request - already below the + // firewall's min=1 - now closes instead of getting the pre-migration downgraded response. + assert!( + handle_request(API_KEY_LIST_OFFSETS, 0, Bytes::new(), &default_broker()).is_close(), + "ListOffsets v0 has no encodable response shape and must close" + ); +} + +#[test] +fn create_topics_below_min_version_closes_connection() { + // CreateTopics v1 is below both the firewall min (2) and `kafka_protocol`'s schema floor + // (2-7) - no encodable response exists at this version, so this closes instead of the + // pre-migration downgraded error response. + assert!( + handle_request(API_KEY_CREATE_TOPICS, 1, Bytes::new(), &default_broker()).is_close(), + "CreateTopics v1 has no encodable response shape and must close" + ); +} + +#[test] +fn unsupported_api_keys_close_connection() { + for key in [8, 9, 10, 11, 17, 20, 42, 999] { + let outcome = handle_request(key, 0, Bytes::new(), &default_broker()); + assert!( + outcome.is_close(), + "unknown api_key {key} must close (no parseable response schema)" + ); + } +} + +#[test] +fn supported_produce_versions_accept_valid_fixture() { + for version in 3i16..=9 { + let Some(body) = load_fixture_body_or_skip(0, "Produce", version) else { + continue; + }; + let resp = handle_request(API_KEY_PRODUCE, version, body, &default_broker()) + .expect_response("test request has acks != 0 and expects a response"); + assert!(!resp.is_empty(), "Produce v{version} response empty"); + } +} + +#[test] +fn supported_fetch_versions_accept_valid_fixture() { + for version in 4i16..=12 { + let Some(body) = load_fixture_body_or_skip(1, "Fetch", version) else { + continue; + }; + let resp = handle_request(API_KEY_FETCH, version, body, &default_broker()) + .expect_response("test request has acks != 0 and expects a response"); + assert!(!resp.is_empty(), "Fetch v{version} response empty"); + } +} + +#[test] +fn corrupt_produce_body_with_acks_stays_silent() { + // `kafka_protocol` decodes Produce in one shot, so a decode failure never exposes `acks` + // (unlike the pre-migration field-by-field decoder, which could still answer with + // INVALID_REQUEST once it knew acks was nonzero). Every Produce decode failure now stays + // silent regardless of whether acks was readable before the truncation. + let body = Bytes::from_static(&[ + 0xFF, 0xFF, // null transactional_id + 0x00, 0x01, // acks = 1 + 0x00, 0x00, 0x00, 0x00, // timeout_ms = 0 + 0xFF, 0xFF, 0xFF, // truncated topics count + ]); + assert!( + handle_request(API_KEY_PRODUCE, 3, body, &default_broker()).is_no_response(), + "malformed Produce body must stay silent regardless of acks" + ); +} + +#[test] +fn corrupt_produce_body_before_acks_is_silent() { + // Decode fails before acks is read: the client's response expectation is unknowable, and an + // error response could desync an acks=0 fire-and-forget client, so the server stays silent. + let body = Bytes::from_static(&[0xFF, 0xFF]); // null transactional_id, then EOF + let outcome = handle_request(API_KEY_PRODUCE, 3, body, &default_broker()); + assert!( + outcome.is_no_response(), + "produce decode failure before acks must be silent" + ); +} + +#[test] +fn corrupt_fetch_body_returns_invalid_request_error() { + let body = Bytes::from_static(&[0xFF, 0xFF, 0xFF]); + let resp = handle_request(API_KEY_FETCH, 4, body, &default_broker()) + .expect_response("test request has acks != 0 and expects a response"); + let mut d = Decoder::new(resp); + assert_eq!(d.read_i32().unwrap(), 0); + assert_eq!(d.read_i32().unwrap(), 1); + assert_eq!(d.read_nullable_string().unwrap(), Some(String::new())); + assert_eq!(d.read_i32().unwrap(), 1); + assert_eq!(d.read_i32().unwrap(), 0); + assert_eq!(d.read_i16().unwrap(), ERROR_INVALID_REQUEST); +} + +// ── ListOffsets v0 (no encodable representation in kafka_protocol) ───────── + +#[test] +fn list_offsets_v0_with_topic_closes_connection() { + let request_body = build_list_offsets_v0_request_with_topic_t(); + assert!( + handle_request(API_KEY_LIST_OFFSETS, 0, request_body, &default_broker()).is_close(), + "ListOffsets v0 has no encodable response shape and must close, even with a well-formed body" + ); +} + +// ── Comprehensive scoped-API coverage (correlation id, boundary versions) ── + +fn request_body_for_scoped_api(api_key: i16, name: &str, version: i16) -> Bytes { + match api_key { + API_KEY_METADATA => { + if version >= 9 { + build_metadata_all_topics_flexible(version) + } else { + build_metadata_all_topics_legacy(version) + } + } + API_KEY_API_VERSIONS => { + if version >= 3 { + build_api_versions_flexible_request("iggy-test", "0.1.0") + } else { + Bytes::new() + } + } + API_KEY_PRODUCE => { + if fixture_exists(api_key, name, version) { Review Comment: All 3 arms in request_body_for_scoped_api now call load_fixture_body_or_skip. cmd_generate (kafka-tool/src/main.rs:601-655) now bail!s when a requested --api-key yields zero files. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
