This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a commit to branch mtls_kafka_proxy
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/mtls_kafka_proxy by this push:
new b277988ac address review
b277988ac is described below
commit b277988acdf599836f2df4595fc82433b2849ef4
Author: Grzegorz Koszyk <[email protected]>
AuthorDate: Thu Oct 15 09:59:56 2026 +0200
address review
---
gateways/kafka/README.md | 5 +-
gateways/kafka/docs/AUTHENTICATION.md | 6 +-
gateways/kafka/src/bridge/iggy_bridge/fetch.rs | 40 ++++++++
gateways/kafka/src/bridge/iggy_bridge/mod.rs | 17 ++-
gateways/kafka/src/tls/identity.rs | 137 ++++++++++++++++++++++---
5 files changed, 182 insertions(+), 23 deletions(-)
diff --git a/gateways/kafka/README.md b/gateways/kafka/README.md
index e55370350..bfcac9934 100644
--- a/gateways/kafka/README.md
+++ b/gateways/kafka/README.md
@@ -312,8 +312,9 @@ pattern matches the whole subject, `DEFAULT` keeps the
subject as is. Iggy accep
3 to 50 characters, but a raw subject is usually longer than that, and is
rarely the name the
operator created, so a deployment that maps identities sets at least one rule.
Patterns are a
Java-compatible subset: `\d \s \w` match ASCII only and `.` stops at Java's
line terminators, as on
-a broker, while `(?i)`, the `u` and `U` flags, `\p{..}`, POSIX classes and
`\b` are refused at
-startup ([docs/AUTHENTICATION.md](docs/AUTHENTICATION.md) says why). A
replacement that
+a broker, while `(?i)`, the `u`, `U`, `m`, `x` and `R` flags, `\p{..}`, POSIX
classes, `\b`, `\v`,
+possessive quantifiers and the `--` and `~~` class operations are refused at
startup
+([docs/AUTHENTICATION.md](docs/AUTHENTICATION.md) says why). A replacement that
names a group the pattern does not have is refused at startup. That is
stricter than Kafka, which
writes such a `$N` literally: `RULE:^CN=(.*)$/$2/` maps to `$2` on a broker.
The reverse case,
`/$1\$2/`, makes Kafka throw while the gateway maps `alice` to `alice$2`,
which is harmless because
diff --git a/gateways/kafka/docs/AUTHENTICATION.md
b/gateways/kafka/docs/AUTHENTICATION.md
index a217cb83c..31ee2d072 100644
--- a/gateways/kafka/docs/AUTHENTICATION.md
+++ b/gateways/kafka/docs/AUTHENTICATION.md
@@ -259,8 +259,10 @@ first frame:
ASCII only, and `.` outside `(?s)` stops at `\r`, U+0085, U+2028 and U+2029
as well as `\n`, both as
in Java. Rust would read them over Unicode, so a CA-signed `CN=\u212Aafka`
(Kelvin sign) would match
`\w+` and lowercase to `kafka`. Case-insensitive matching (`(?i)`), the `u`
and `U` flags, `\p{..}`
- and `\P{..}`, POSIX classes such as `[[:alpha:]]`, and word boundaries
(`\b`, `\B`) are refused at
- startup with the construct named: each has no faithful translation.
Non-ASCII names still map, so
+ and `\P{..}`, POSIX classes such as `[[:alpha:]]`, word boundaries (`\b`,
`\B`), the `m`, `x` and
+ `R` flags, `\v`, a quantifier directly after another (`.*+`, possessive in
Java) and the `--` and
+ `~~` class operations are refused at startup with the construct named: each
has no faithful
+ translation. Non-ASCII names still map, so
`CN=Zoë` under `RULE:^CN=([^,]+).*$/$1/` maps to `Zoë` as on a broker.
3. **Look the name up** through the bridge's own session. The user must exist
and be active, or the
connection is closed after the handshake with the name in the log at
`warn`: a certificate the CA signed
diff --git a/gateways/kafka/src/bridge/iggy_bridge/fetch.rs
b/gateways/kafka/src/bridge/iggy_bridge/fetch.rs
index 9ce900bbb..e404d8fc4 100644
--- a/gateways/kafka/src/bridge/iggy_bridge/fetch.rs
+++ b/gateways/kafka/src/bridge/iggy_bridge/fetch.rs
@@ -129,6 +129,23 @@ impl LazyClient {
result
}
+ /// [`Self::until`] on a task of its own, so the deadline still drops a
stuck client when the
+ /// caller stops waiting first.
+ pub(super) async fn until_detached<T, Call>(
+ &self,
+ deadline: Instant,
+ call: Call,
+ ) -> Result<T, BridgeError>
+ where
+ T: Send + 'static,
+ Call: Future<Output = Result<T, BridgeError>> + Send + 'static,
+ {
+ let client = self.clone();
+ tokio::spawn(async move { client.until(deadline, call).await })
+ .await
+ .unwrap_or(Err(BridgeError::Timeout))
+ }
+
/// Shuts the client down, if it connected.
pub(super) async fn close(&self) -> Result<(), BridgeError> {
let Some(client) = self.0.lock().await.take() else {
@@ -375,6 +392,29 @@ mod tests {
Arc::new(client)
}
+ #[tokio::test]
+ async fn
given_a_detached_call_whose_caller_gave_up_when_its_deadline_passes_should_drop_the_client()
+ {
+ let client = LazyClient::default();
+ *client.0.lock().await = Some(unconnected());
+ let deadline = Instant::now() + Duration::from_millis(50);
+
+ let caller = tokio::time::timeout(
+ Duration::from_millis(10),
+ client.until_detached(deadline, std::future::pending::<Result<u8,
BridgeError>>()),
+ )
+ .await;
+ assert!(
+ caller.is_err(),
+ "the caller stops waiting before the deadline"
+ );
+ tokio::time::sleep(Duration::from_millis(150)).await;
+ assert!(
+ client.0.lock().await.is_none(),
+ "the stuck client is dropped at its own deadline, not left for the
next lookup"
+ );
+ }
+
#[tokio::test]
async fn
given_a_call_past_its_deadline_when_given_up_on_should_drop_the_client() {
let client = LazyClient::default();
diff --git a/gateways/kafka/src/bridge/iggy_bridge/mod.rs
b/gateways/kafka/src/bridge/iggy_bridge/mod.rs
index 049a63700..18ee64c97 100644
--- a/gateways/kafka/src/bridge/iggy_bridge/mod.rs
+++ b/gateways/kafka/src/bridge/iggy_bridge/mod.rs
@@ -280,15 +280,22 @@ impl std::fmt::Debug for IggyBridge {
///
/// Not the lockstep client: a stuck Produce send would hold every new mTLS
connection past its
/// handshake budget, and a burst of handshakes would queue ahead of Produce.
+///
+/// Detached because the caller's `pre_auth_timeout` can be shorter than
`REQUEST_TIMEOUT`:
+/// cancelling the lookup there would also cancel the deadline that drops a
stuck client, and
+/// every later lookup would queue behind it.
#[async_trait]
impl PrincipalResolver for IggyBridge {
async fn resolve(&self, username: &str) -> Result<AuthenticatedPrincipal,
AuthError> {
- let client = &self.lookup_client;
- let lookup = async {
- let connected = client.connected(&self.config).await?;
- Ok(resolve_principal(connected.as_ref(), username).await)
+ let client = self.lookup_client.clone();
+ let config = self.config.clone();
+ let username = username.to_owned();
+ let lookup = async move {
+ let connected = client.connected(&config).await?;
+ Ok(resolve_principal(connected.as_ref(), &username).await)
};
- match client.until(Instant::now() + REQUEST_TIMEOUT, lookup).await {
+ let deadline = Instant::now() + REQUEST_TIMEOUT;
+ match self.lookup_client.until_detached(deadline, lookup).await {
Ok(resolved) => resolved,
Err(error) => {
warn!(%error, "could not reach Iggy to look up the
certificate's user");
diff --git a/gateways/kafka/src/tls/identity.rs
b/gateways/kafka/src/tls/identity.rs
index 877e5061a..e9ede9016 100644
--- a/gateways/kafka/src/tls/identity.rs
+++ b/gateways/kafka/src/tls/identity.rs
@@ -29,7 +29,8 @@ use std::str::FromStr;
use regex::{Captures, Regex};
use regex_syntax::ast::{
- self, Assertion, AssertionKind, Ast, ClassPerl, ClassPerlKind,
ClassSetItem, Flag, Flags, Span,
+ self, Assertion, AssertionKind, Ast, ClassPerl, ClassPerlKind,
ClassSetBinaryOp,
+ ClassSetBinaryOpKind, ClassSetItem, Flag, Flags, LiteralKind, Span,
SpecialLiteralKind,
Visitor,
};
use thiserror::Error;
@@ -99,6 +100,21 @@ pub enum JavaIncompatibility {
PosixClass,
#[error("word boundaries are not supported")]
WordBoundary,
+ #[error(
+ "the 'm', 'x' and 'R' flags differ between Rust and Java in which line
ends and which \
+ whitespace and comments they cover"
+ )]
+ LineOrWhitespaceFlag,
+ #[error("'\\v' is a vertical tab in Rust and any vertical whitespace in
Java")]
+ VerticalWhitespace,
+ #[error(
+ "Java reads a quantifier after a quantifier as possessive, Rust as a
nested repetition"
+ )]
+ StackedQuantifier,
+ #[error(
+ "Java reads '--' and '~~' inside a class as literal characters, Rust
as set operations"
+ )]
+ ClassSetOperation,
}
/// A replacement that `java.util.regex.Matcher` would throw on when the rule
is applied.
@@ -455,6 +471,11 @@ impl JavaRewriter {
ast::FlagsItemKind::Flag(Flag::Unicode | Flag::SwapGreed) => {
return Err((item.span, JavaIncompatibility::UnicodeFlag));
}
+ ast::FlagsItemKind::Flag(
+ flag @ (Flag::MultiLine | Flag::IgnoreWhitespace |
Flag::CRLF),
+ ) if flags.flag_state(flag) == Some(true) => {
+ return Err((item.span,
JavaIncompatibility::LineOrWhitespaceFlag));
+ }
_ => {}
}
}
@@ -511,6 +532,14 @@ impl Visitor for JavaRewriter {
return Err((class.span, JavaIncompatibility::UnicodeClass));
}
Ast::Assertion(assertion) => refuse_word_boundary(assertion)?,
+ Ast::Literal(literal)
+ if literal.kind ==
LiteralKind::Special(SpecialLiteralKind::VerticalTab) =>
+ {
+ return Err((literal.span,
JavaIncompatibility::VerticalWhitespace));
+ }
+ Ast::Repetition(repetition) if matches!(*repetition.ast,
Ast::Repetition(_)) => {
+ return Err((repetition.op.span,
JavaIncompatibility::StackedQuantifier));
+ }
_ => {}
}
Ok(())
@@ -532,10 +561,24 @@ impl Visitor for JavaRewriter {
ClassSetItem::Ascii(class) => {
return Err((class.span, JavaIncompatibility::PosixClass));
}
+ ClassSetItem::Literal(literal)
+ if literal.kind ==
LiteralKind::Special(SpecialLiteralKind::VerticalTab) =>
+ {
+ return Err((literal.span,
JavaIncompatibility::VerticalWhitespace));
+ }
_ => {}
}
Ok(())
}
+
+ fn visit_class_set_binary_op_pre(&mut self, op: &ClassSetBinaryOp) ->
Result<(), Self::Err> {
+ match op.kind {
+ ClassSetBinaryOpKind::Difference |
ClassSetBinaryOpKind::SymmetricDifference => {
+ Err((op.span, JavaIncompatibility::ClassSetOperation))
+ }
+ ClassSetBinaryOpKind::Intersection => Ok(()),
+ }
+ }
}
const fn refuse_word_boundary(assertion: &Assertion) -> Result<(), (Span,
JavaIncompatibility)> {
@@ -1034,9 +1077,25 @@ mod tests {
}
}
+ /// Asserts each rule list is refused, naming `construct` for `reason`.
+ fn assert_refused(cases: &[(&str, &str, JavaIncompatibility)]) {
+ for &(raw, construct, expected) in cases {
+ match raw.parse::<PrincipalMappingRules>() {
+ Err(MappingRuleError::JavaIncompatible {
+ construct: found,
+ reason,
+ ..
+ }) => {
+ assert_eq!((found.as_str(), reason), (construct,
expected), "{raw}");
+ }
+ other => panic!("'{raw}' must be refused, got {other:?}"),
+ }
+ }
+ }
+
#[test]
fn given_a_construct_without_java_semantics_should_refuse_the_rule() {
- for (raw, construct, expected) in [
+ assert_refused(&[
(
"RULE:(?i)^CN=(\u{17F}vc)$/$1/",
"i",
@@ -1082,21 +1141,71 @@ mod tests {
r"\B",
JavaIncompatibility::WordBoundary,
),
- ] {
- match raw.parse::<PrincipalMappingRules>() {
- Err(MappingRuleError::JavaIncompatible {
- construct: found,
- reason,
- ..
- }) => {
- assert_eq!((found.as_str(), reason), (construct,
expected), "{raw}");
- }
- other => panic!("'{raw}' must be refused, got {other:?}"),
- }
- }
+ ]);
let case_sensitive: PrincipalMappingRules = "RULE:(?-i)^CN=(.*)$/$1/"
.parse()
.expect("(?-i) is Java's default");
assert_eq!(case_sensitive.map("CN=a").as_deref(), Some("a"));
}
+
+ #[test]
+ fn given_syntax_rust_and_java_read_differently_should_refuse_the_rule() {
+ assert_refused(&[
+ (
+ "RULE:(?m)^CN=(.*)$/$1/",
+ "m",
+ JavaIncompatibility::LineOrWhitespaceFlag,
+ ),
+ (
+ "RULE:(?x)^CN=(.*)$/$1/",
+ "x",
+ JavaIncompatibility::LineOrWhitespaceFlag,
+ ),
+ (
+ "RULE:^CN=(?R:.*)$/CN/",
+ "R",
+ JavaIncompatibility::LineOrWhitespaceFlag,
+ ),
+ (
+ r"RULE:^CN=([^\v,]+)$/$1/",
+ r"\v",
+ JavaIncompatibility::VerticalWhitespace,
+ ),
+ (
+ r"RULE:^CN=(a\v)$/$1/",
+ r"\v",
+ JavaIncompatibility::VerticalWhitespace,
+ ),
+ (
+ "RULE:^CN=(.*+)$/$1/",
+ "+",
+ JavaIncompatibility::StackedQuantifier,
+ ),
+ (
+ "RULE:^CN=(a{2}+)$/$1/",
+ "+",
+ JavaIncompatibility::StackedQuantifier,
+ ),
+ (
+ "RULE:^CN=([a-z--b]+)$/$1/",
+ "a-z--b",
+ JavaIncompatibility::ClassSetOperation,
+ ),
+ (
+ "RULE:^CN=([a-z~~b]+)$/$1/",
+ "a-z~~b",
+ JavaIncompatibility::ClassSetOperation,
+ ),
+ ]);
+ // Turning a flag off keeps Java's default, so it stays admissible.
+ let explicit_defaults: PrincipalMappingRules =
"RULE:(?-mx)^CN=([^,]+)$/$1/"
+ .parse()
+ .expect("(?-mx) is Java's default");
+ assert_eq!(explicit_defaults.map("CN=a").as_deref(), Some("a"));
+ // A lazy quantifier and a class intersection mean the same in both.
+ let same_in_both: PrincipalMappingRules =
"RULE:^CN=([a-z&&[^q]]+?)$/$1/"
+ .parse()
+ .expect("lazy quantifiers and && match Java");
+ assert_eq!(same_in_both.map("CN=abc").as_deref(), Some("abc"));
+ }
}
