This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new ec6b651b2 fix(common): reject malformed user headers during
deserialization (#4670)
ec6b651b2 is described below
commit ec6b651b2679f37c58ad0d5c03b3d9ec23fbe4fc
Author: Prabhath <[email protected]>
AuthorDate: Wed Nov 4 02:13:40 2026 +0530
fix(common): reject malformed user headers during deserialization (#4670)
## Which issue does this PR address?
Closes #4604
## Rationale
The HTTP produce endpoint accepted typed user headers that violate the
wire protocol's TLV bounds (0-byte values, widths mismatched with
fixed-size types like `int32`, or values > 255 bytes). When encoded by
`user_headers_to_wire`, these headers corrupted the message TLV stream,
causing consumers polling the message to fail header decoding
(`Err(InvalidHeaderValue)`) or silently drop all headers on the message
(`Ok(None)`).
## What changed?
- **Marker-specific error mapping:** Added a sealed `HeaderFieldMarker`
trait on `KeyMarker` and `ValueMarker` returning
`IggyError::InvalidHeaderKey` and `IggyError::InvalidHeaderValue`
respectively.
- **Constructor validation:** Parameterized `HeaderField<T>::from_raw`
by `T: HeaderFieldMarker`, returning the marker-specific error on empty
fields, lengths > 255 bytes, or fixed-kind width mismatches.
- **Serde deserialization validation:** Implemented custom `Deserialize`
for `HeaderField<T>` that deserializes raw base64 data and routes it
through `Self::from_raw()`, rejecting malformed keys/values at the
HTTP/JSON boundary (HTTP 400).
- **Defense-in-depth:** Added a `debug_assert!` in
`user_headers_to_wire` verifying that the encoded buffer is structurally
valid TLV.
- **Tests:** Enabled and passed the boundary rejection tests in
`send_messages.rs`, and added unit tests for `HeaderField`,
`HeaderEntry`, `HeaderKey`, `HeaderValue`, and `IggyMessage`
deserialization.
---
core/common/src/http/messages/send_messages.rs | 202 ++++++++++--
core/common/src/types/message/iggy_message.rs | 53 ++++
core/common/src/types/message/mod.rs | 5 +-
core/common/src/types/message/user_headers.rs | 391 ++++++++++++++++++++++--
core/common/src/wire_conversions.rs | 18 +-
core/integration/tests/server/http_vsr.rs | 83 +++++
core/sdk/src/prelude.rs | 34 +--
core/server/src/http/extractor.rs | 40 ++-
core/server/src/http/handlers.rs | 12 +-
foreign/cpp/src/type_conversion.rs | 18 +-
foreign/python/src/user_headers.rs | 7 +-
foreign/python/tests/test_message_operations.py | 4 +-
12 files changed, 784 insertions(+), 83 deletions(-)
diff --git a/core/common/src/http/messages/send_messages.rs
b/core/common/src/http/messages/send_messages.rs
index 5851f3de6..7b1856c35 100644
--- a/core/common/src/http/messages/send_messages.rs
+++ b/core/common/src/http/messages/send_messages.rs
@@ -21,8 +21,8 @@ use crate::PartitionContext;
use crate::PartitioningKind;
use crate::Validatable;
use crate::error::IggyError;
-use crate::types::message::HeaderEntry;
use crate::types::message::partitioning::Partitioning;
+use crate::types::message::{HeaderEntry, RawHeaderEntry, UserHeaders};
use crate::{
IggyMessage, IggyMessagesBatch, SendMessagesConfirmationResponse,
SendMessagesResponse,
};
@@ -31,7 +31,6 @@ use bytes::Bytes;
use serde::de::{self, MapAccess, Visitor};
use serde::ser::SerializeStruct;
use serde::{Deserialize, Deserializer, Serialize, Serializer};
-use std::collections::BTreeMap;
use std::fmt::Formatter;
/// `SendMessages` command is used to send messages to a topic in a stream.
@@ -139,6 +138,33 @@ impl Serialize for SendMessages {
impl<'de> Deserialize<'de> for SendMessages {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+ where
+ D: Deserializer<'de>,
+ {
+ Self::deserialize_recording(deserializer, &mut None)
+ }
+}
+
+impl SendMessages {
+ /// The error of the typed user header that refused the JSON `body`, `None`
+ /// when the body parses or failed on something else. A refused header is
+ /// otherwise only a serde error, so the server reads this to answer the
+ /// header's own `InvalidHeaderKey` or `InvalidHeaderValue`.
+ pub fn refused_user_header(body: &[u8]) -> Option<IggyError> {
+ let mut refused = None;
+ let parsed = Self::deserialize_recording(
+ &mut serde_json::Deserializer::from_slice(body),
+ &mut refused,
+ );
+ parsed.err().and(refused)
+ }
+
+ /// Deserializes the HTTP body, keeping in `refused` the error of a typed
+ /// user header that the header codec refuses.
+ fn deserialize_recording<'de, D>(
+ deserializer: D,
+ refused: &mut Option<IggyError>,
+ ) -> Result<Self, D::Error>
where
D: Deserializer<'de>,
{
@@ -181,9 +207,11 @@ impl<'de> Deserialize<'de> for SendMessages {
}
}
- struct SendMessagesVisitor;
+ struct SendMessagesVisitor<'a> {
+ refused: &'a mut Option<IggyError>,
+ }
- impl<'de> Visitor<'de> for SendMessagesVisitor {
+ impl<'de> Visitor<'de> for SendMessagesVisitor<'_> {
type Value = SendMessages;
fn expecting(&self, formatter: &mut Formatter) -> std::fmt::Result
{
@@ -246,18 +274,7 @@ impl<'de> Deserialize<'de> for SendMessages {
})?;
(None, Some(Bytes::from(raw)))
} else {
- let entries: Vec<HeaderEntry> =
serde_json::from_value(
- headers.clone(),
- )
- .map_err(|e| {
- de::Error::custom(format!(
- "Invalid headers format:
{e}"
- ))
- })?;
- let mut map = BTreeMap::new();
- for entry in entries {
- map.insert(entry.key,
entry.value);
- }
+ let map =
typed_user_headers(headers, self.refused)?;
(Some(map), None)
}
} else {
@@ -317,7 +334,7 @@ impl<'de> Deserialize<'de> for SendMessages {
}
}
- deserializer.deserialize_struct("SendMessages", FIELDS,
SendMessagesVisitor)
+ deserializer.deserialize_struct("SendMessages", FIELDS,
SendMessagesVisitor { refused })
}
}
@@ -375,6 +392,27 @@ impl From<SendMessagesConfirmations> for
SendMessagesResponse {
}
}
+/// The typed `user_headers` of one HTTP message, keeping in `refused` the
+/// error of a field the header codec refuses.
+fn typed_user_headers<E: de::Error>(
+ headers: &serde_json::Value,
+ refused: &mut Option<IggyError>,
+) -> Result<UserHeaders, E> {
+ let entries: Vec<RawHeaderEntry> = serde_json::from_value(headers.clone())
+ .map_err(|e| E::custom(format!("Invalid headers format: {e}")))?;
+ entries
+ .into_iter()
+ .map(|entry| {
+ let entry = HeaderEntry::try_from(entry).map_err(|error| {
+ let message = format!("Invalid headers format: {error}");
+ *refused = Some(error);
+ E::custom(message)
+ })?;
+ Ok((entry.key, entry.value))
+ })
+ .collect()
+}
+
fn parse_message_id(value: Option<&serde_json::Value>) -> Result<u128, String>
{
let value = match value {
Some(v) => v,
@@ -577,4 +615,134 @@ mod tests {
let result = parse_message_id(Some(&value));
assert!(result.is_err());
}
+
+ fn send(headers: serde_json::Value) -> Result<SendMessages,
serde_json::Error> {
+ serde_json::from_value(serde_json::json!({
+ "partitioning": { "kind": "balanced", "value": "" },
+ "messages": [{ "id": 1, "payload": "aGVsbG8=", "user_headers":
headers }]
+ }))
+ }
+
+ #[test]
+ fn typed_header_with_valid_headers_is_accepted_at_the_http_boundary() {
+ let parsed = send(serde_json::json!([{
+ "key": { "kind": "string", "value": "a2V5" },
+ "value": { "kind": "string", "value": "dmFs" }
+ }]));
+ assert!(parsed.is_ok());
+ }
+
+ #[test]
+ fn typed_header_with_wrong_size_value_is_rejected_at_the_http_boundary() {
+ let parsed = send(serde_json::json!([{
+ "key": { "kind": "string", "value": "a2V5" },
+ "value": { "kind": "int32", "value": "AQ==" }
+ }]));
+ assert!(parsed.is_err());
+ }
+
+ #[test]
+ fn typed_header_with_empty_value_is_rejected_at_the_http_boundary() {
+ let parsed = send(serde_json::json!([
+ { "key": { "kind": "string", "value": "b2s=" }, "value": { "kind":
"string", "value": "dmFs" } },
+ { "key": { "kind": "string", "value": "a2V5" }, "value": { "kind":
"raw", "value": "" } }
+ ]));
+ assert!(parsed.is_err());
+ }
+
+ #[test]
+ fn typed_header_with_wrong_size_key_is_rejected_at_the_http_boundary() {
+ let parsed = send(serde_json::json!([{
+ "key": { "kind": "int32", "value": "AQ==" },
+ "value": { "kind": "string", "value": "dmFs" }
+ }]));
+ assert!(parsed.is_err());
+ }
+
+ #[test]
+ fn typed_header_with_empty_key_is_rejected_at_the_http_boundary() {
+ let parsed = send(serde_json::json!([{
+ "key": { "kind": "string", "value": "" },
+ "value": { "kind": "string", "value": "dmFs" }
+ }]));
+ assert!(parsed.is_err());
+ }
+
+ #[test]
+ fn typed_header_with_oversized_fields_is_rejected_at_the_http_boundary() {
+ let oversized = BASE64.encode([0u8; 256]);
+ let parsed_key = send(serde_json::json!([{
+ "key": { "kind": "string", "value": oversized },
+ "value": { "kind": "string", "value": "dmFs" }
+ }]));
+ assert!(parsed_key.is_err());
+
+ let parsed_val = send(serde_json::json!([{
+ "key": { "kind": "string", "value": "a2V5" },
+ "value": { "kind": "string", "value": oversized }
+ }]));
+ assert!(parsed_val.is_err());
+ }
+
+ fn body(payload: &str, headers: serde_json::Value) -> Vec<u8> {
+ serde_json::to_vec(&serde_json::json!({
+ "partitioning": { "kind": "balanced", "value": "" },
+ "messages": [
+ { "id": 1, "payload": "aGVsbG8=" },
+ { "id": 2, "payload": payload, "user_headers": headers }
+ ]
+ }))
+ .unwrap()
+ }
+
+ #[test]
+ fn refused_user_header_names_the_error_of_the_refused_field() {
+ let wrong_size_key = body(
+ "aGVsbG8=",
+ serde_json::json!([{
+ "key": { "kind": "int32", "value": "AQ==" },
+ "value": { "kind": "string", "value": "dmFs" }
+ }]),
+ );
+ assert_eq!(
+ SendMessages::refused_user_header(&wrong_size_key),
+ Some(IggyError::InvalidHeaderKey)
+ );
+
+ let empty_value = body(
+ "aGVsbG8=",
+ serde_json::json!([
+ { "key": { "kind": "string", "value": "b2s=" }, "value": {
"kind": "string", "value": "dmFs" } },
+ { "key": { "kind": "string", "value": "a2V5" }, "value": {
"kind": "raw", "value": "" } }
+ ]),
+ );
+ assert_eq!(
+ SendMessages::refused_user_header(&empty_value),
+ Some(IggyError::InvalidHeaderValue)
+ );
+ }
+
+ #[test]
+ fn refused_user_header_is_none_when_no_header_refused_the_body() {
+ let valid_header = serde_json::json!([{
+ "key": { "kind": "string", "value": "a2V5" },
+ "value": { "kind": "int32", "value": "AQAAAA==" }
+ }]);
+ assert_eq!(
+ SendMessages::refused_user_header(&body("aGVsbG8=",
valid_header.clone())),
+ None
+ );
+ assert_eq!(
+ SendMessages::refused_user_header(&body("not base64!",
valid_header)),
+ None
+ );
+ let missing_kind = serde_json::json!([{
+ "key": { "value": "a2V5" },
+ "value": { "kind": "string", "value": "dmFs" }
+ }]);
+ assert_eq!(
+ SendMessages::refused_user_header(&body("aGVsbG8=", missing_kind)),
+ None
+ );
+ }
}
diff --git a/core/common/src/types/message/iggy_message.rs
b/core/common/src/types/message/iggy_message.rs
index 7fe3a3a7c..6b3da9b50 100644
--- a/core/common/src/types/message/iggy_message.rs
+++ b/core/common/src/types/message/iggy_message.rs
@@ -884,4 +884,57 @@ mod tests {
assert_eq!(&buf[..8], b"existing");
assert_eq!(&buf[8..], &message.to_bytes()[..]);
}
+
+ #[test]
+ fn deserializing_message_with_malformed_typed_headers_fails() {
+ let header =
serde_json::to_value(IggyMessageHeader::default()).unwrap();
+
+ // Valid headers deserialize successfully
+ let json_valid = serde_json::json!({
+ "header": header,
+ "payload": "aGVsbG8=",
+ "user_headers": [{
+ "key": { "kind": "string", "value": "a2V5" },
+ "value": { "kind": "string", "value": "dmFs" }
+ }]
+ });
+ let result: Result<IggyMessage, _> =
serde_json::from_value(json_valid);
+ assert!(result.is_ok());
+
+ // Wrong size value (int32 with 1 byte)
+ let json_wrong_val = serde_json::json!({
+ "header": header,
+ "payload": "aGVsbG8=",
+ "user_headers": [{
+ "key": { "kind": "string", "value": "a2V5" },
+ "value": { "kind": "int32", "value": "AQ==" }
+ }]
+ });
+ let result: Result<IggyMessage, _> =
serde_json::from_value(json_wrong_val);
+ assert!(result.is_err());
+
+ // Empty value
+ let json_empty_val = serde_json::json!({
+ "header": header,
+ "payload": "aGVsbG8=",
+ "user_headers": [{
+ "key": { "kind": "string", "value": "a2V5" },
+ "value": { "kind": "raw", "value": "" }
+ }]
+ });
+ let result: Result<IggyMessage, _> =
serde_json::from_value(json_empty_val);
+ assert!(result.is_err());
+
+ // Wrong size key
+ let json_wrong_key = serde_json::json!({
+ "header": header,
+ "payload": "aGVsbG8=",
+ "user_headers": [{
+ "key": { "kind": "int32", "value": "AQ==" },
+ "value": { "kind": "string", "value": "dmFs" }
+ }]
+ });
+ let result: Result<IggyMessage, _> =
serde_json::from_value(json_wrong_key);
+ assert!(result.is_err());
+ }
}
diff --git a/core/common/src/types/message/mod.rs
b/core/common/src/types/message/mod.rs
index 5a4de21ab..ea1f90b34 100644
--- a/core/common/src/types/message/mod.rs
+++ b/core/common/src/types/message/mod.rs
@@ -70,7 +70,8 @@ pub use poll_options::{AutoCommit, CommitAfter, CommitEvery,
CommitPolicy, PollO
pub use polled_messages::{DecodeError, PolledMessages, UndecodedRecord,
record_at};
pub use polling_kind::PollingKind;
pub use polling_strategy::PollingStrategy;
+pub(crate) use user_headers::RawHeaderEntry;
pub use user_headers::{
- HeaderEntry, HeaderField, HeaderKey, HeaderKind, HeaderValue, KeyMarker,
UserHeaders,
- ValueMarker, deserialize_headers, serialize_headers,
+ HeaderEntry, HeaderField, HeaderFieldMarker, HeaderKey, HeaderKind,
HeaderValue, KeyMarker,
+ UserHeaders, ValueMarker, deserialize_headers, serialize_headers,
};
diff --git a/core/common/src/types/message/user_headers.rs
b/core/common/src/types/message/user_headers.rs
index 07223ce80..a4ce80fb7 100644
--- a/core/common/src/types/message/user_headers.rs
+++ b/core/common/src/types/message/user_headers.rs
@@ -91,6 +91,31 @@ pub struct KeyMarker;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct ValueMarker;
+mod private {
+ pub trait Sealed {}
+ impl Sealed for super::KeyMarker {}
+ impl Sealed for super::ValueMarker {}
+}
+
+/// Sealed trait marking valid header field markers ([`KeyMarker`] and
[`ValueMarker`]).
+pub trait HeaderFieldMarker:
+ std::fmt::Debug + Clone + Copy + PartialEq + Eq + Hash + private::Sealed
+{
+ fn invalid_error() -> IggyError;
+}
+
+impl HeaderFieldMarker for KeyMarker {
+ fn invalid_error() -> IggyError {
+ IggyError::InvalidHeaderKey
+ }
+}
+
+impl HeaderFieldMarker for ValueMarker {
+ fn invalid_error() -> IggyError {
+ IggyError::InvalidHeaderValue
+ }
+}
+
/// A typed header field that can be used as either a key or value in message
headers.
///
/// `HeaderField` is a generic struct parameterized by a marker type
(`KeyMarker` or `ValueMarker`)
@@ -142,7 +167,7 @@ pub struct ValueMarker;
/// # Ok::<(), IggyError>(())
/// ```
#[serde_as]
-#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
+#[derive(Debug, Serialize, Clone, PartialEq, Eq)]
pub struct HeaderField<T> {
kind: HeaderKind,
#[serde_as(as = "Base64")]
@@ -151,6 +176,24 @@ pub struct HeaderField<T> {
_marker: PhantomData<T>,
}
+#[serde_as]
+#[derive(Deserialize)]
+struct RawHeaderField {
+ kind: HeaderKind,
+ #[serde_as(as = "Base64")]
+ value: Bytes,
+}
+
+impl<'de, T: HeaderFieldMarker> Deserialize<'de> for HeaderField<T> {
+ fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+ where
+ D: serde::Deserializer<'de>,
+ {
+ let raw = RawHeaderField::deserialize(deserializer)?;
+ Self::from_raw_bytes(raw.kind,
raw.value).map_err(serde::de::Error::custom)
+ }
+}
+
impl<T> HeaderField<T> {
/// Returns the kind of this header field.
pub fn kind(&self) -> HeaderKind {
@@ -353,7 +396,7 @@ impl<T> Hash for HeaderField<T> {
}
}
-impl<T> FromStr for HeaderField<T> {
+impl<T: HeaderFieldMarker> FromStr for HeaderField<T> {
type Err = IggyError;
fn from_str(s: &str) -> Result<Self, Self::Err> {
Self::try_from(s)
@@ -616,7 +659,9 @@ impl<T> HeaderField<T> {
_marker: PhantomData,
}
}
+}
+impl<T: HeaderFieldMarker> HeaderField<T> {
/// Wrap raw wire bytes as a field of `kind`.
///
/// The option catalog hands out each key's default as a kind plus bytes,
so
@@ -624,19 +669,43 @@ impl<T> HeaderField<T> {
///
/// # Errors
///
- /// `IggyError::InvalidHeaderValue` when a fixed-size kind's payload is the
- /// wrong length, or the payload exceeds the field length limit.
+ /// `IggyError::InvalidHeaderKey` or `IggyError::InvalidHeaderValue` when a
+ /// fixed-size kind's payload is the wrong length, or the payload exceeds
+ /// the field length limit.
pub fn from_raw(kind: HeaderKind, value: &[u8]) -> Result<Self, IggyError>
{
+ Self::validate_raw(kind, value.len())?;
+ Ok(Self::new_unchecked(kind, value))
+ }
+
+ /// Wrap owned wire `Bytes` as a field of `kind`.
+ ///
+ /// Validates length constraints and moves `value` without copying.
+ ///
+ /// # Errors
+ ///
+ /// `IggyError::InvalidHeaderKey` or `IggyError::InvalidHeaderValue` when a
+ /// fixed-size kind's payload is the wrong length, or the payload exceeds
+ /// the field length limit.
+ pub fn from_raw_bytes(kind: HeaderKind, value: Bytes) -> Result<Self,
IggyError> {
+ Self::validate_raw(kind, value.len())?;
+ Ok(Self {
+ kind,
+ value,
+ _marker: PhantomData,
+ })
+ }
+
+ fn validate_raw(kind: HeaderKind, len: usize) -> Result<(), IggyError> {
// 255 is the field length the TLV codec's own `TryFrom` impls enforce.
- if value.is_empty() || value.len() > 255 {
- return Err(IggyError::InvalidHeaderValue);
+ if len == 0 || len > 255 {
+ return Err(T::invalid_error());
}
if let Some(expected) = kind.expected_size()
- && value.len() != expected
+ && len != expected
{
- return Err(IggyError::InvalidHeaderValue);
+ return Err(T::invalid_error());
}
- Ok(Self::new_unchecked(kind, value))
+ Ok(())
}
}
@@ -652,34 +721,34 @@ impl<T: Eq> PartialOrd for HeaderField<T> {
}
}
-impl<T> TryFrom<&str> for HeaderField<T> {
+impl<T: HeaderFieldMarker> TryFrom<&str> for HeaderField<T> {
type Error = IggyError;
fn try_from(value: &str) -> Result<Self, Self::Error> {
if value.is_empty() || value.len() > 255 {
- return Err(IggyError::InvalidHeaderValue);
+ return Err(T::invalid_error());
}
Ok(Self::new_unchecked(HeaderKind::String, value.as_bytes()))
}
}
-impl<T> TryFrom<String> for HeaderField<T> {
+impl<T: HeaderFieldMarker> TryFrom<String> for HeaderField<T> {
type Error = IggyError;
fn try_from(value: String) -> Result<Self, Self::Error> {
Self::try_from(value.as_str())
}
}
-impl<T> TryFrom<&[u8]> for HeaderField<T> {
+impl<T: HeaderFieldMarker> TryFrom<&[u8]> for HeaderField<T> {
type Error = IggyError;
fn try_from(value: &[u8]) -> Result<Self, Self::Error> {
if value.is_empty() || value.len() > 255 {
- return Err(IggyError::InvalidHeaderValue);
+ return Err(T::invalid_error());
}
Ok(Self::new_unchecked(HeaderKind::Raw, value))
}
}
-impl<T> TryFrom<Vec<u8>> for HeaderField<T> {
+impl<T: HeaderFieldMarker> TryFrom<Vec<u8>> for HeaderField<T> {
type Error = IggyError;
fn try_from(value: Vec<u8>) -> Result<Self, Self::Error> {
Self::try_from(value.as_slice())
@@ -980,6 +1049,26 @@ pub struct HeaderEntry {
pub value: HeaderValue,
}
+/// A [`HeaderEntry`] as JSON carries it, before either field is checked, so a
+/// caller that needs the refusal as an [`IggyError`] rather than a serde error
+/// can convert it with `HeaderEntry::try_from`.
+#[derive(Deserialize)]
+pub(crate) struct RawHeaderEntry {
+ key: RawHeaderField,
+ value: RawHeaderField,
+}
+
+impl TryFrom<RawHeaderEntry> for HeaderEntry {
+ type Error = IggyError;
+
+ fn try_from(entry: RawHeaderEntry) -> Result<Self, Self::Error> {
+ Ok(Self {
+ key: HeaderKey::from_raw_bytes(entry.key.kind, entry.key.value)?,
+ value: HeaderValue::from_raw_bytes(entry.value.kind,
entry.value.value)?,
+ })
+ }
+}
+
pub fn serialize_headers<S>(headers: &Option<UserHeaders>, serializer: S) ->
Result<S::Ok, S::Error>
where
S: serde::Serializer,
@@ -1022,6 +1111,7 @@ where
mod tests {
use super::*;
use crate::wire_conversions::{user_headers_from_wire,
user_headers_to_wire};
+ use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64};
use bytes::{BufMut, BytesMut};
use iggy_binary_protocol::WireUserHeaders;
@@ -1041,7 +1131,35 @@ mod tests {
let header_key = HeaderKey::try_from(value);
assert!(header_key.is_err());
let error = header_key.unwrap_err();
- assert_eq!(error.as_code(), IggyError::InvalidHeaderValue.as_code());
+ assert_eq!(error.as_code(), IggyError::InvalidHeaderKey.as_code());
+
+ let from_str_res = HeaderKey::from_str(value);
+ assert!(from_str_res.is_err());
+ assert_eq!(
+ from_str_res.unwrap_err().as_code(),
+ IggyError::InvalidHeaderKey.as_code()
+ );
+
+ let from_string_res = HeaderKey::try_from(value.to_string());
+ assert!(from_string_res.is_err());
+ assert_eq!(
+ from_string_res.unwrap_err().as_code(),
+ IggyError::InvalidHeaderKey.as_code()
+ );
+
+ let from_slice_res = HeaderKey::try_from([].as_slice());
+ assert!(from_slice_res.is_err());
+ assert_eq!(
+ from_slice_res.unwrap_err().as_code(),
+ IggyError::InvalidHeaderKey.as_code()
+ );
+
+ let from_vec_res = HeaderKey::try_from(Vec::<u8>::new());
+ assert!(from_vec_res.is_err());
+ assert_eq!(
+ from_vec_res.unwrap_err().as_code(),
+ IggyError::InvalidHeaderKey.as_code()
+ );
}
#[test]
@@ -1050,7 +1168,35 @@ mod tests {
let header_key = HeaderKey::try_from(value.as_str());
assert!(header_key.is_err());
let error = header_key.unwrap_err();
- assert_eq!(error.as_code(), IggyError::InvalidHeaderValue.as_code());
+ assert_eq!(error.as_code(), IggyError::InvalidHeaderKey.as_code());
+
+ let from_str_res = HeaderKey::from_str(value.as_str());
+ assert!(from_str_res.is_err());
+ assert_eq!(
+ from_str_res.unwrap_err().as_code(),
+ IggyError::InvalidHeaderKey.as_code()
+ );
+
+ let from_string_res = HeaderKey::try_from(value.clone());
+ assert!(from_string_res.is_err());
+ assert_eq!(
+ from_string_res.unwrap_err().as_code(),
+ IggyError::InvalidHeaderKey.as_code()
+ );
+
+ let from_slice_res = HeaderKey::try_from(vec![0u8; 256].as_slice());
+ assert!(from_slice_res.is_err());
+ assert_eq!(
+ from_slice_res.unwrap_err().as_code(),
+ IggyError::InvalidHeaderKey.as_code()
+ );
+
+ let from_vec_res = HeaderKey::try_from(vec![0u8; 256]);
+ assert!(from_vec_res.is_err());
+ assert_eq!(
+ from_vec_res.unwrap_err().as_code(),
+ IggyError::InvalidHeaderKey.as_code()
+ );
}
#[test]
@@ -1659,4 +1805,215 @@ mod tests {
};
assert_eq!(field.to_string_value(), "<malformed bool>");
}
+
+ #[test]
+ fn header_key_from_raw_rejects_empty_and_oversized_and_wrong_size() {
+ assert_eq!(
+ HeaderKey::from_raw(HeaderKind::String, b"")
+ .unwrap_err()
+ .as_code(),
+ IggyError::InvalidHeaderKey.as_code()
+ );
+ assert_eq!(
+ HeaderKey::from_raw(HeaderKind::String, &[0u8; 256])
+ .unwrap_err()
+ .as_code(),
+ IggyError::InvalidHeaderKey.as_code()
+ );
+ assert_eq!(
+ HeaderKey::from_raw(HeaderKind::Int32, &[1, 2])
+ .unwrap_err()
+ .as_code(),
+ IggyError::InvalidHeaderKey.as_code()
+ );
+ }
+
+ #[test]
+ fn header_value_from_raw_rejects_empty_and_oversized_and_wrong_size() {
+ assert_eq!(
+ HeaderValue::from_raw(HeaderKind::Raw, b"")
+ .unwrap_err()
+ .as_code(),
+ IggyError::InvalidHeaderValue.as_code()
+ );
+ assert_eq!(
+ HeaderValue::from_raw(HeaderKind::Raw, &[0u8; 256])
+ .unwrap_err()
+ .as_code(),
+ IggyError::InvalidHeaderValue.as_code()
+ );
+ assert_eq!(
+ HeaderValue::from_raw(HeaderKind::Int32, &[1, 2])
+ .unwrap_err()
+ .as_code(),
+ IggyError::InvalidHeaderValue.as_code()
+ );
+ }
+
+ #[test]
+ fn header_key_from_raw_bytes_accepts_valid_and_rejects_invalid() {
+ let valid = HeaderKey::from_raw_bytes(HeaderKind::String,
Bytes::from_static(b"my-key"));
+ assert!(valid.is_ok());
+ let key = valid.unwrap();
+ assert_eq!(key.as_str().unwrap(), "my-key");
+ assert_eq!(key.value(), Bytes::from_static(b"my-key"));
+
+ assert_eq!(
+ HeaderKey::from_raw_bytes(HeaderKind::String, Bytes::new())
+ .unwrap_err()
+ .as_code(),
+ IggyError::InvalidHeaderKey.as_code()
+ );
+ assert_eq!(
+ HeaderKey::from_raw_bytes(HeaderKind::String,
Bytes::copy_from_slice(&[0u8; 256]))
+ .unwrap_err()
+ .as_code(),
+ IggyError::InvalidHeaderKey.as_code()
+ );
+ assert_eq!(
+ HeaderKey::from_raw_bytes(HeaderKind::Int32,
Bytes::from_static(&[1, 2]))
+ .unwrap_err()
+ .as_code(),
+ IggyError::InvalidHeaderKey.as_code()
+ );
+ }
+
+ #[test]
+ fn header_value_from_raw_bytes_accepts_valid_and_rejects_invalid() {
+ let valid = HeaderValue::from_raw_bytes(
+ HeaderKind::Int32,
+ Bytes::copy_from_slice(&42i32.to_le_bytes()),
+ );
+ assert!(valid.is_ok());
+ let value = valid.unwrap();
+ assert_eq!(value.as_int32().unwrap(), 42);
+
+ assert_eq!(
+ HeaderValue::from_raw_bytes(HeaderKind::Raw, Bytes::new())
+ .unwrap_err()
+ .as_code(),
+ IggyError::InvalidHeaderValue.as_code()
+ );
+ assert_eq!(
+ HeaderValue::from_raw_bytes(HeaderKind::Raw,
Bytes::copy_from_slice(&[0u8; 256]))
+ .unwrap_err()
+ .as_code(),
+ IggyError::InvalidHeaderValue.as_code()
+ );
+ assert_eq!(
+ HeaderValue::from_raw_bytes(HeaderKind::Int32,
Bytes::from_static(&[1, 2]))
+ .unwrap_err()
+ .as_code(),
+ IggyError::InvalidHeaderValue.as_code()
+ );
+ }
+
+ #[test]
+ fn deserialize_header_field_validates_value() {
+ // Valid string header
+ let valid_json = serde_json::json!({
+ "kind": "string",
+ "value": "aGVsbG8=" // "hello"
+ });
+ let key: Result<HeaderKey, _> =
serde_json::from_value(valid_json.clone());
+ assert!(key.is_ok());
+ let val: Result<HeaderValue, _> = serde_json::from_value(valid_json);
+ assert!(val.is_ok());
+
+ // Empty value rejected
+ let empty_json = serde_json::json!({
+ "kind": "raw",
+ "value": ""
+ });
+ let key_err: Result<HeaderKey, _> =
serde_json::from_value(empty_json.clone());
+ assert!(key_err.is_err());
+ assert!(
+ key_err
+ .unwrap_err()
+ .to_string()
+ .contains("Invalid header key")
+ );
+
+ let val_err: Result<HeaderValue, _> =
serde_json::from_value(empty_json);
+ assert!(val_err.is_err());
+ assert!(
+ val_err
+ .unwrap_err()
+ .to_string()
+ .contains("Invalid header value")
+ );
+
+ // Wrong size for fixed-size kind rejected (int32 needs 4 bytes, got 1)
+ let wrong_size_json = serde_json::json!({
+ "kind": "int32",
+ "value": "AQ==" // 1 byte [1]
+ });
+ let key_err: Result<HeaderKey, _> =
serde_json::from_value(wrong_size_json.clone());
+ assert!(key_err.is_err());
+ assert!(
+ key_err
+ .unwrap_err()
+ .to_string()
+ .contains("Invalid header key")
+ );
+
+ let val_err: Result<HeaderValue, _> =
serde_json::from_value(wrong_size_json);
+ assert!(val_err.is_err());
+ assert!(
+ val_err
+ .unwrap_err()
+ .to_string()
+ .contains("Invalid header value")
+ );
+
+ // Oversized (> 255 bytes) rejected
+ let oversized_b64 = BASE64.encode([0u8; 256]);
+ let oversized_json = serde_json::json!({
+ "kind": "string",
+ "value": oversized_b64
+ });
+ let key_err: Result<HeaderKey, _> =
serde_json::from_value(oversized_json.clone());
+ assert!(key_err.is_err());
+ let val_err: Result<HeaderValue, _> =
serde_json::from_value(oversized_json);
+ assert!(val_err.is_err());
+ }
+
+ #[test]
+ fn deserialize_header_entry_validates_key_and_value() {
+ // Valid entry
+ let valid_entry = serde_json::json!({
+ "key": { "kind": "string", "value": "a2V5" },
+ "value": { "kind": "int32", "value": "AQAAAA==" }
+ });
+ let entry: Result<HeaderEntry, _> =
serde_json::from_value(valid_entry);
+ assert!(entry.is_ok());
+
+ // Malformed key
+ let malformed_key = serde_json::json!({
+ "key": { "kind": "int32", "value": "AQ==" },
+ "value": { "kind": "string", "value": "dmFs" }
+ });
+ let entry: Result<HeaderEntry, _> =
serde_json::from_value(malformed_key);
+ assert!(entry.is_err());
+ assert!(
+ entry
+ .unwrap_err()
+ .to_string()
+ .contains("Invalid header key")
+ );
+
+ // Malformed value
+ let malformed_value = serde_json::json!({
+ "key": { "kind": "string", "value": "a2V5" },
+ "value": { "kind": "int32", "value": "AQ==" }
+ });
+ let entry: Result<HeaderEntry, _> =
serde_json::from_value(malformed_value);
+ assert!(entry.is_err());
+ assert!(
+ entry
+ .unwrap_err()
+ .to_string()
+ .contains("Invalid header value")
+ );
+ }
}
diff --git a/core/common/src/wire_conversions.rs
b/core/common/src/wire_conversions.rs
index 965fc4713..e87723ba7 100644
--- a/core/common/src/wire_conversions.rs
+++ b/core/common/src/wire_conversions.rs
@@ -789,6 +789,10 @@ pub fn user_headers_to_wire(
}
// Buffer was just encoded from valid HeaderKey/HeaderValue entries,
// so structural TLV validity is guaranteed by construction.
+ debug_assert!(
+ iggy_binary_protocol::validate_user_headers(&buf).is_ok(),
+ "encoded user headers must be structurally valid TLV"
+ );
WireUserHeaders::from_validated(buf.freeze())
}
@@ -850,20 +854,10 @@ fn headers_from_validated_slice(
return Err(error);
}
};
- if let Some(expected) = key_kind.expected_size()
- && entry.key.len() != expected
- {
- return Err(IggyError::InvalidHeaderKey);
- }
- if let Some(expected) = value_kind.expected_size()
- && entry.value.len() != expected
- {
- return Err(IggyError::InvalidHeaderValue);
- }
headers.insert(
- HeaderKey::new_unchecked(key_kind, entry.key),
- HeaderValue::new_unchecked(value_kind, entry.value),
+ HeaderKey::from_raw(key_kind, entry.key)?,
+ HeaderValue::from_raw(value_kind, entry.value)?,
);
}
Ok(headers)
diff --git a/core/integration/tests/server/http_vsr.rs
b/core/integration/tests/server/http_vsr.rs
index 27f30cd48..297ddaaa1 100644
--- a/core/integration/tests/server/http_vsr.rs
+++ b/core/integration/tests/server/http_vsr.rs
@@ -854,6 +854,89 @@ async fn
given_oversized_body_when_producing_should_reject_413(harness: &TestHar
);
}
+#[iggy_harness]
+async fn
given_refused_typed_header_when_producing_should_reject_400_with_header_error(
+ harness: &TestHarness,
+) {
+ let http = HttpClient::login_root(harness).await;
+ http.create_stream_and_topic("http-headers", "typed", 1)
+ .await;
+ let path = "/streams/http-headers/topics/typed/messages";
+ let produce = async |key: serde_json::Value, value: serde_json::Value| {
+ let body = json!({
+ "partitioning": { "kind": "balanced", "value": "" },
+ "messages": [{
+ "id": 1,
+ "payload": "aGVsbG8=",
+ "user_headers": [{ "key": key, "value": value }]
+ }]
+ });
+ http.client
+ .post(http.url(path))
+ .bearer_auth(&http.token)
+ .json(&body)
+ .send()
+ .await
+ .expect("produce request")
+ };
+ let string = |value: &str| json!({ "kind": "string", "value": value });
+ let one_byte_int32 = json!({ "kind": "int32", "value": "AQ==" });
+
+ let cases = [
+ (
+ one_byte_int32.clone(),
+ string("dmFs"),
+ IggyError::InvalidHeaderKey,
+ ),
+ (string(""), string("dmFs"), IggyError::InvalidHeaderKey),
+ (
+ string("a2V5"),
+ one_byte_int32,
+ IggyError::InvalidHeaderValue,
+ ),
+ (
+ string("a2V5"),
+ json!({ "kind": "raw", "value": "" }),
+ IggyError::InvalidHeaderValue,
+ ),
+ ];
+ for (key, value, refusal) in cases {
+ let case = format!("key={key} value={value}");
+ let response = produce(key, value).await;
+ assert_eq!(response.status(), StatusCode::BAD_REQUEST, "{case}");
+ let error: serde_json::Value = response.json().await.expect("decode
the error body");
+ assert_eq!(error["id"], refusal.as_code(), "{case}");
+ assert_eq!(error["code"], refusal.as_string(), "{case}");
+ }
+
+ let response = produce(
+ string("a2V5"),
+ json!({ "kind": "int32", "value": "AQAAAA==" }),
+ )
+ .await;
+ assert_eq!(
+ response.status(),
+ StatusCode::CREATED,
+ "a well-formed typed header must still commit"
+ );
+ let polled = http
+ .poll("http-headers", "typed", PARTITION_ID, 0, 10)
+ .await;
+ assert_eq!(
+ polled.messages.len(),
+ 1,
+ "only the well-formed produce must land"
+ );
+ let user_headers = polled.messages[0]
+ .user_headers_map()
+ .expect("headers decode")
+ .expect("the message carries user headers");
+ assert_eq!(
+ user_headers.get(&HeaderKey::try_from("key").expect("header key")),
+ Some(&HeaderValue::from(1i32)),
+ );
+}
+
#[iggy_harness]
async fn given_ack_none_when_producing_should_return_202_and_commit(harness:
&TestHarness) {
const BURST_SIZE: u32 = 8;
diff --git a/core/sdk/src/prelude.rs b/core/sdk/src/prelude.rs
index e252507ff..345322b9f 100644
--- a/core/sdk/src/prelude.rs
+++ b/core/sdk/src/prelude.rs
@@ -55,23 +55,23 @@ pub use iggy_common::{
ClusterNodeRole, ClusterNodeStatus, CommitAfter, CommitEvery, CommitPolicy,
CompressionAlgorithm, Consumer, ConsumerGroup, ConsumerGroupAssignment,
ConsumerGroupDetails,
ConsumerGroupMember, ConsumerKind, Credentials, Durability, EncryptorKind,
GlobalPermissions,
- HeaderField, HeaderKey, HeaderKind, HeaderValue, HttpClientConfig,
HttpClientConfigBuilder,
- HttpMethod, IdKind, Identifier, IdentityInfo, IggyByteSize, IggyDuration,
IggyError,
- IggyExpiry, IggyIndexView, IggyMessage, IggyMessageHeader,
IggyMessageHeaderView,
- IggyMessageView, IggyMessageViewIterator, IggyTimestamp, MaxTopicSize,
NonZeroDurationError,
- NonZeroIggyDuration, OptionKey, OptionSpec, OptionValue, OptionsScope,
Partition,
- PartitionContext, PartitionResizePolicy, Partitioner, Partitioning,
PendingOperationsConfig,
- Permissions, PersonalAccessTokenExpiry, PollMessages, PollOptions,
PolledMessages, PollingKind,
- PollingStrategy, QuicClientConfig, QuicClientConfigBuilder,
QuicClientReconnectionConfig,
- ResourceOptions, SendMessages, SendMessagesConfirmationResponse,
SendMessagesResponse,
- Sizeable, SnapshotCompression, Stats, Stream, StreamCreateOptions,
StreamDetails,
- StreamEncryptionKey, StreamEncryptionSource, StreamPermissions,
StreamUpdateOptions,
- SystemSnapshotType, TcpClientConfig, TcpClientConfigBuilder,
TcpClientReconnectionConfig,
- Topic, TopicCreateOptions, TopicDetails, TopicPermissions,
TopicResolvedOptions,
- TopicUpdateOptions, TransportEndpoints, TransportProtocol,
UndecodedRecord, UserCreateOptions,
- UserId, UserInfo, UserInfoDetails, UserStatus, UserUpdateOptions,
Validatable,
- WebSocketClientConfig, WebSocketClientConfigBuilder,
WebSocketClientReconnectionConfig,
- WebSocketConfig, defaults, locking, topic_option_keys,
+ HeaderField, HeaderFieldMarker, HeaderKey, HeaderKind, HeaderValue,
HttpClientConfig,
+ HttpClientConfigBuilder, HttpMethod, IdKind, Identifier, IdentityInfo,
IggyByteSize,
+ IggyDuration, IggyError, IggyExpiry, IggyIndexView, IggyMessage,
IggyMessageHeader,
+ IggyMessageHeaderView, IggyMessageView, IggyMessageViewIterator,
IggyTimestamp, MaxTopicSize,
+ NonZeroDurationError, NonZeroIggyDuration, OptionKey, OptionSpec,
OptionValue, OptionsScope,
+ Partition, PartitionContext, PartitionResizePolicy, Partitioner,
Partitioning,
+ PendingOperationsConfig, Permissions, PersonalAccessTokenExpiry,
PollMessages, PollOptions,
+ PolledMessages, PollingKind, PollingStrategy, QuicClientConfig,
QuicClientConfigBuilder,
+ QuicClientReconnectionConfig, ResourceOptions, SendMessages,
SendMessagesConfirmationResponse,
+ SendMessagesResponse, Sizeable, SnapshotCompression, Stats, Stream,
StreamCreateOptions,
+ StreamDetails, StreamEncryptionKey, StreamEncryptionSource,
StreamPermissions,
+ StreamUpdateOptions, SystemSnapshotType, TcpClientConfig,
TcpClientConfigBuilder,
+ TcpClientReconnectionConfig, Topic, TopicCreateOptions, TopicDetails,
TopicPermissions,
+ TopicResolvedOptions, TopicUpdateOptions, TransportEndpoints,
TransportProtocol,
+ UndecodedRecord, UserCreateOptions, UserId, UserInfo, UserInfoDetails,
UserStatus,
+ UserUpdateOptions, Validatable, WebSocketClientConfig,
WebSocketClientConfigBuilder,
+ WebSocketClientReconnectionConfig, WebSocketConfig, defaults, locking,
topic_option_keys,
};
pub use iggy_common::{
Client, ClusterClient, ConsumerGroupClient, ConsumerOffsetClient,
MessageClient,
diff --git a/core/server/src/http/extractor.rs
b/core/server/src/http/extractor.rs
index 130a63b15..78376d3a2 100644
--- a/core/server/src/http/extractor.rs
+++ b/core/server/src/http/extractor.rs
@@ -25,6 +25,7 @@ use std::rc::Rc;
use axum::Json;
use axum::body::{Body, Bytes};
+use axum::extract::rejection::JsonRejection;
use axum::extract::{ConnectInfo, FromRequest, FromRequestParts, Request};
use axum::http::HeaderMap;
use axum::http::header::AUTHORIZATION;
@@ -32,7 +33,7 @@ use axum::http::request::Parts;
use axum::response::{IntoResponse, Response};
use iggy_common::{
IggyError, IggyTimestamp, OptionInputValue, OptionsUpdate,
PersonalAccessToken, RESET_FIELD,
- erase_buffer,
+ SendMessages, erase_buffer,
};
use send_wrapper::SendWrapper;
use serde::Deserializer as _;
@@ -40,7 +41,7 @@ use serde::de::{DeserializeOwned, Error as _, IgnoredAny,
MapAccess, Visitor};
use tracing::debug;
use super::HttpState;
-use super::error::{AuthError, WriteError};
+use super::error::{AuthError, PartitionWriteError, WriteError};
use super::jwt::TokenSubject;
use super::session::HttpSession;
use super::state::HttpInner;
@@ -92,6 +93,12 @@ pub struct JsonMembers<T> {
/// at fault where one is.
pub(in crate::http) struct JsonUpdate(pub OptionsUpdate);
+/// A produce body. A typed user header the header codec refuses answers 400
+/// with that header's `InvalidHeaderKey` or `InvalidHeaderValue`, the body
+/// `SendMessages::validate` gives on the same route. Every other rejection
+/// stays axum's own.
+pub(in crate::http) struct JsonSendMessages(pub SendMessages);
+
/// Walks an update body with the member types [`OptionsUpdate`] parses and
/// records the first member that does not parse: a repeated name, a `reset`
/// that is not a list of distinct names, or another member that is not a
@@ -178,6 +185,35 @@ where
}
}
+impl<S> FromRequest<S> for JsonSendMessages
+where
+ S: Send + Sync,
+{
+ type Rejection = Response;
+
+ async fn from_request(request: Request, state: &S) -> Result<Self,
Self::Rejection> {
+ let (parts, body) = request.into_parts();
+ let bytes = Bytes::from_request(Request::from_parts(parts.clone(),
body), state)
+ .await
+ .map_err(IntoResponse::into_response)?;
+ match Json::<SendMessages>::from_request(
+ Request::from_parts(parts, Body::from(bytes.clone())),
+ state,
+ )
+ .await
+ {
+ Ok(Json(command)) => Ok(Self(command)),
+ Err(JsonRejection::JsonDataError(rejection)) => {
+ Err(SendMessages::refused_user_header(&bytes).map_or_else(
+ || rejection.into_response(),
+ |error|
PartitionWriteError::Rejected(error).into_response(),
+ ))
+ }
+ Err(rejection) => Err(rejection.into_response()),
+ }
+ }
+}
+
impl<'de> Visitor<'de> for RefusedMemberVisitor<'_> {
type Value = ();
diff --git a/core/server/src/http/handlers.rs b/core/server/src/http/handlers.rs
index 1e379d0dc..510aab447 100644
--- a/core/server/src/http/handlers.rs
+++ b/core/server/src/http/handlers.rs
@@ -105,8 +105,8 @@ use iggy_common::{
ConsumerOffsetInfo, DEFAULT_MAX_TOPIC_SIZE, Identifier, IdentityInfo,
IggyError, IggyTimestamp,
MetadataChanges, MetadataChangesQuery, OptionSpec, OptionStorage,
OptionsScope, Partitioning,
PersonalAccessTokenInfo, PollMessages, PolledMessages, PollingStrategy,
RawPersonalAccessToken,
- SendMessages, SendMessagesConfirmations, Stats, Stream, StreamDetails,
TokenInfo, Topic,
- TopicDetails, UserInfo, UserInfoDetails, Validatable, descriptors_for,
erase_buffer,
+ SendMessagesConfirmations, Stats, Stream, StreamDetails, TokenInfo, Topic,
TopicDetails,
+ UserInfo, UserInfoDetails, Validatable, descriptors_for, erase_buffer,
};
use metadata::impls::metadata::StreamsFrontend;
use metadata::permissioner::Permissioner;
@@ -129,7 +129,9 @@ use crate::http::error::{
Consistency, ConsistencyQuery, CustomError, PartitionWriteError,
ProduceAck, ProduceQuery,
ReadError, WriteError,
};
-use crate::http::extractor::{Authenticated, Credential, Identity, JsonMembers,
JsonUpdate};
+use crate::http::extractor::{
+ Authenticated, Credential, Identity, JsonMembers, JsonSendMessages,
JsonUpdate,
+};
use crate::http::forward::{
FirstAttempt, PartitionEffect, PartitionHop, PartitionRequest,
require_local_primary,
serve_partition,
@@ -1677,7 +1679,7 @@ pub(in crate::http) async fn send_messages(
Path((stream_id, topic_id)): Path<(String, String)>,
Query(query): Query<ProduceQuery>,
Extension(request): Extension<Arc<PartitionRequest>>,
- Json(mut command): Json<SendMessages>,
+ JsonSendMessages(mut command): JsonSendMessages,
) -> Result<Response, PartitionWriteError> {
let stream_id =
Identifier::from_str_value(&stream_id).map_err(PartitionWriteError::Rejected)?;
@@ -2361,7 +2363,7 @@ mod tests {
use iggy_binary_protocol::responses::messages::{
SendMessagesConfirmationResponse, SendMessagesResponse,
};
- use iggy_common::{OptionInputMap, OptionInputValue, OptionKey};
+ use iggy_common::{OptionInputMap, OptionInputValue, OptionKey,
SendMessages};
use serde_json::Value;
use crate::http::reply::classify_partition_reply;
diff --git a/foreign/cpp/src/type_conversion.rs
b/foreign/cpp/src/type_conversion.rs
index 6ac247690..0ad50a458 100644
--- a/foreign/cpp/src/type_conversion.rs
+++ b/foreign/cpp/src/type_conversion.rs
@@ -56,12 +56,13 @@ use iggy_common::{
ConsumerGroup as RustConsumerGroup, ConsumerGroupInfo as
RustConsumerGroupInfo,
ConsumerGroupMember as RustConsumerGroupMember, ConsumerOffsetInfo as
RustConsumerOffsetInfo,
GlobalPermissions as RustGlobalPermissions, HeaderEntry as RustHeaderEntry,
- HeaderField as RustHeaderField, HeaderKind as RustHeaderKind, IdentityInfo
as RustIdentityInfo,
- IggyError, IggyTimestamp, MetadataChange, MetadataChangeCursor,
MetadataChangeGroup,
- MetadataChangeRecord, MetadataChangeStart, MetadataChanges,
MetadataChangesQuery,
- MetadataConsumerGroupCreated, MetadataConsumerGroupDeleted,
MetadataConsumerGroupItem,
- MetadataCurrentItem, MetadataCurrentPageToken, MetadataMemberItem,
MetadataMemberJoined,
- MetadataMemberLeft, MetadataOptions, MetadataPartitionHandedOver,
MetadataPartitionsChanged,
+ HeaderField as RustHeaderField, HeaderFieldMarker, HeaderKind as
RustHeaderKind,
+ IdentityInfo as RustIdentityInfo, IggyError, IggyTimestamp, MetadataChange,
+ MetadataChangeCursor, MetadataChangeGroup, MetadataChangeRecord,
MetadataChangeStart,
+ MetadataChanges, MetadataChangesQuery, MetadataConsumerGroupCreated,
+ MetadataConsumerGroupDeleted, MetadataConsumerGroupItem,
MetadataCurrentItem,
+ MetadataCurrentPageToken, MetadataMemberItem, MetadataMemberJoined,
MetadataMemberLeft,
+ MetadataOptions, MetadataPartitionHandedOver, MetadataPartitionsChanged,
MetadataPartitionsItem, MetadataSession, MetadataSessionClosed,
MetadataStreamChange,
MetadataStreamDeleted, MetadataTokenChange, MetadataTokenDeleted,
MetadataTopicChange,
MetadataTopicDeleted, MetadataTopicItem, MetadataUserChange,
MetadataUserIdentity,
@@ -1025,7 +1026,10 @@ impl TryFrom<ffi::HeaderEntry> for RustHeaderEntry {
}
}
-fn decode_field<T>(kind: u8, value: Vec<u8>) -> Result<RustHeaderField<T>,
String> {
+fn decode_field<T: HeaderFieldMarker>(
+ kind: u8,
+ value: Vec<u8>,
+) -> Result<RustHeaderField<T>, String> {
let kind = RustHeaderKind::from_code(kind)
.map_err(|error| format!("Could not convert header field: {error}"))?;
diff --git a/foreign/python/src/user_headers.rs
b/foreign/python/src/user_headers.rs
index e3fc21082..bb2d9d20b 100644
--- a/foreign/python/src/user_headers.rs
+++ b/foreign/python/src/user_headers.rs
@@ -20,7 +20,8 @@ use std::collections::hash_map::DefaultHasher;
use std::hash::{Hash, Hasher};
use iggy::prelude::{
- HeaderField, HeaderKey as RustHeaderKey, HeaderKind, HeaderValue as
RustHeaderValue,
+ HeaderField, HeaderFieldMarker, HeaderKey as RustHeaderKey, HeaderKind,
+ HeaderValue as RustHeaderValue,
};
use pyo3::PyClass;
use pyo3::exceptions::PyValueError;
@@ -698,7 +699,9 @@ fn py_header_to_plain<'py, T: ToPlain + PyClass>(
/// values). Returns `Ok(None)` when the object is not a supported plain type
so
/// the caller can raise a message naming keys vs values; genuine range/length
/// errors are returned as `Err`.
-fn plain_to_rust_header_field<T>(obj: &Bound<'_, PyAny>) ->
PyResult<Option<HeaderField<T>>> {
+fn plain_to_rust_header_field<T: HeaderFieldMarker>(
+ obj: &Bound<'_, PyAny>,
+) -> PyResult<Option<HeaderField<T>>> {
if obj.is_instance_of::<PyBool>() {
return Ok(Some(obj.extract::<bool>()?.into()));
}
diff --git a/foreign/python/tests/test_message_operations.py
b/foreign/python/tests/test_message_operations.py
index c15c34968..e24207491 100644
--- a/foreign/python/tests/test_message_operations.py
+++ b/foreign/python/tests/test_message_operations.py
@@ -873,8 +873,8 @@ class TestMessageOperations:
@pytest.mark.parametrize(
("headers", "error"),
[
- ({"": "value"}, "Invalid header value"),
- ({"x" * 256: "value"}, "Invalid header value"),
+ ({"": "value"}, "Invalid header key"),
+ ({"x" * 256: "value"}, "Invalid header key"),
({"key": ""}, "Invalid header value"),
({"key": b""}, "Invalid header value"),
({"key": "x" * 256}, "Invalid header value"),
