krishvishal commented on code in PR #4535:
URL: https://github.com/apache/iggy/pull/4535#discussion_r4279607547
##########
gateways/kafka/src/bridge/iggy_bridge/mod.rs:
##########
@@ -243,13 +250,50 @@ async fn connect_client(config: &IggyBridgeConfig) ->
Result<IggyClient, BridgeE
let credentials =
Credentials::UsernamePassword(config.username.clone(),
config.password.clone());
- let client = IggyClientBuilder::new()
+ let mut builder = IggyClientBuilder::new()
.with_tcp()
.with_server_address(config.address.clone())
.with_auto_sign_in(AutoLogin::Enabled(credentials))
- .with_reconnection_max_retries(Some(RECONNECTION_RETRIES))
- .build()
- .map_err(BridgeError::Iggy)?;
+ .with_reconnection_max_retries(Some(RECONNECTION_RETRIES));
+ if config.tls.enabled {
+ builder = builder.with_tls_enabled(true);
+ if !config.tls.domain.is_empty() {
+ builder = builder.with_tls_domain(config.tls.domain.clone());
+ }
+ if let Some(ref ca_file) = config.tls.ca_file {
+ builder = builder.with_tls_ca_file(ca_file.clone());
+ }
+ }
+ let client = builder.build().map_err(BridgeError::Iggy)?;
with_request_timeout(client.connect()).await?;
Ok(client)
}
+
+impl std::fmt::Debug for IggyBridge {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("IggyBridge").finish_non_exhaustive()
+ }
+}
+
+/// A session of the bridge's own answers for the users client certificates
name. It runs as the
+/// bridge's Iggy user, so that account needs `read_users` for any certificate
to be admitted.
+///
+/// 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.
+#[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)
+ };
+ match client.until(Instant::now() + REQUEST_TIMEOUT, lookup).await {
Review Comment:
If `pre_auth_timeout` is under 15s, `within_auth_budget` cancels `resolve`
before `until` hits its deadline, and the stuck `lookup_client` never gets
evicted. New mTLS connections then fail for up to 30s.
Could you spawn the lookup like `in_slot` and `ProbeBoard::probe` do, or
evict from a drop guard?
##########
gateways/kafka/src/tls/identity.rs:
##########
@@ -0,0 +1,1102 @@
+// 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.
+
+//! The principal a client certificate names.
+//!
+//! Kafka hands `ssl.principal.mapping.rules` the subject as Java's
`X500Principal.getName()`
+//! renders it: RFC 2253, most specific RDN first, no spaces after the commas.
Operators write
+//! their rules against that rendering, so this one follows
`sun.security.x509.AVA`'s
+//! `toRFC2253String` (JDK 11 to 21) rather than x509-parser's own `Display`,
which keeps the
+//! certificate's order and pads with spaces. Malformed UTF-8 and BMP values
are the one case not
+//! checked against Java: both sides substitute U+FFFD, possibly not one for
one.
+
+use std::fmt::Write as _;
+use std::str::FromStr;
+
+use regex::{Captures, Regex};
+use regex_syntax::ast::{
+ self, Assertion, AssertionKind, Ast, ClassPerl, ClassPerlKind,
ClassSetItem, Flag, Flags, Span,
+ Visitor,
+};
+use thiserror::Error;
+use x509_parser::asn1_rs::{Any, Tag, ToDer};
+use x509_parser::error::X509Error;
+use x509_parser::nom;
+use x509_parser::parse_x509_certificate;
+use x509_parser::x509::{AttributeTypeAndValue, RelativeDistinguishedName,
X509Name};
+
+/// Why a client certificate yielded no principal.
+#[derive(Debug, Error)]
+pub enum IdentityError {
+ #[error("the client certificate could not be parsed: {0}")]
+ Certificate(#[from] nom::Err<X509Error>),
+ #[error("no principal mapping rule matches the certificate subject
'{subject}'")]
+ NoMatchingRule { subject: String },
+}
+
+/// Why `ssl.principal.mapping.rules` text did not parse.
+#[derive(Debug, Error)]
+pub enum MappingRuleError {
+ #[error("the rule list is empty; the default is DEFAULT")]
+ Empty,
+ #[error("the rule list ends with a comma")]
+ TrailingComma,
+ #[error("'{0}' is neither DEFAULT nor RULE:pattern/replacement/[LU]")]
+ UnknownRule(String),
+ #[error("expected ',' or the end of the list before '{0}'")]
+ MissingSeparator(String),
+ #[error("rule '{0}' has no '/' after its pattern")]
+ UnterminatedPattern(String),
+ #[error("rule '{0}' has no '/' after its replacement")]
+ UnterminatedReplacement(String),
+ #[error("rule pattern '{pattern}' is not a valid regex: {source}")]
+ InvalidPattern {
+ pattern: String,
+ #[source]
+ source: regex::Error,
+ },
+ #[error("rule pattern '{pattern}' uses '{construct}': {reason}")]
+ JavaIncompatible {
+ pattern: String,
+ construct: String,
+ reason: JavaIncompatibility,
+ },
+ #[error("replacement '{replacement}' of rule pattern '{pattern}' is
invalid: {source}")]
+ InvalidReplacement {
+ pattern: String,
+ replacement: String,
+ #[source]
+ source: ReplacementError,
+ },
+}
+
+/// A pattern construct whose Rust `regex` meaning admits subjects Java's
`Pattern` rejects.
+#[derive(Debug, Error, PartialEq, Eq, Clone, Copy)]
+pub enum JavaIncompatibility {
+ #[error(
+ "the case-insensitive flag folds Unicode in Rust (the Kelvin sign
matches 'k'), ASCII in Java"
+ )]
+ CaseInsensitive,
+ #[error("the 'u' and 'U' flags mean different things in Rust and Java")]
+ UnicodeFlag,
+ #[error("Unicode property classes are not supported")]
+ UnicodeClass,
+ #[error("Java reads '[:name:]' as a set of literal characters, not a POSIX
class")]
+ PosixClass,
+ #[error("word boundaries are not supported")]
+ WordBoundary,
+}
+
+/// A replacement that `java.util.regex.Matcher` would throw on when the rule
is applied.
+#[derive(Debug, Error, PartialEq, Eq)]
+pub enum ReplacementError {
+ #[error("a trailing '\\' escapes nothing")]
+ DanglingEscape,
+ #[error("'$' is followed by neither a group number nor '{{name}}'")]
+ IllegalGroupReference,
+ #[error("the pattern has no group {0}")]
+ NoGroup(usize),
+ #[error("'${{' has no closing '}}'")]
+ UnterminatedGroupName,
+ #[error("the pattern has no group named '{0}'")]
+ NoNamedGroup(String),
+}
+
+/// The subject of a DER certificate, as Kafka's principal builder sees it.
+///
+/// # Errors
+///
+/// Returns [`IdentityError::Certificate`] when `certificate` is not a DER
X.509 certificate.
+pub fn subject_distinguished_name(certificate: &[u8]) -> Result<String,
IdentityError> {
+ let (_, parsed) = parse_x509_certificate(certificate)?;
+ Ok(render_rfc2253(parsed.subject()))
+}
+
+/// RFC 2253 string form: RDNs in reverse certificate order, `,` between RDNs,
`+` inside a
+/// multi-valued one, and the attribute keywords Java emits.
+fn render_rfc2253(name: &X509Name<'_>) -> String {
+ let rendered: Vec<String> = name.iter_rdn().map(render_rdn).collect();
+ let mut out = String::new();
+ for (index, rdn) in rendered.iter().rev().enumerate() {
+ if index > 0 {
+ out.push(',');
+ }
+ out.push_str(rdn);
+ }
+ out
+}
+
+fn render_rdn(rdn: &RelativeDistinguishedName<'_>) -> String {
+ let mut out = String::new();
+ for (index, attribute) in rdn.iter().enumerate() {
+ if index > 0 {
+ out.push('+');
+ }
+ render_attribute(&mut out, attribute);
+ }
+ out
+}
+
+/// `KEYWORD=value` for the attribute types Java gives a keyword, `OID=#hex`
for the rest, and
+/// `#hex` of the DER for a value whose string type Java does not decode.
+fn render_attribute(out: &mut String, attribute: &AttributeTypeAndValue<'_>) {
+ let value = attribute.attr_value();
+ if let Some(keyword) = rfc2253_keyword(&attribute.attr_type().to_string())
{
+ out.push_str(keyword);
+ out.push('=');
+ match string_value(value) {
+ Some(text) => escape_into(out, &text),
+ None => push_hex(out, value),
+ }
+ } else {
+ let _ = write!(out, "{}=", attribute.attr_type());
+ push_hex(out, value);
+ }
+}
+
+/// The keyword Java's RFC 2253 rendering uses for an attribute type, or
`None` for an OID it
+/// spells out. Java's set is RFC 2253's seven plus `DC` and `UID` from RFC
4514; everything
+/// else, `emailAddress` included, is rendered as a dotted OID with a hex
value.
+fn rfc2253_keyword(oid: &str) -> Option<&'static str> {
+ Some(match oid {
+ "2.5.4.3" => "CN",
+ "2.5.4.6" => "C",
+ "2.5.4.7" => "L",
+ "2.5.4.8" => "ST",
+ "2.5.4.9" => "STREET",
+ "2.5.4.10" => "O",
+ "2.5.4.11" => "OU",
+ "0.9.2342.19200300.100.1.25" => "DC",
+ "0.9.2342.19200300.100.1.1" => "UID",
+ _ => return None,
+ })
+}
+
+/// The text of a directory string, decoded with the charset `AVA.getCharset`
picks for its tag,
+/// or `None` for the tags `AVA.isDerString` rejects, which Java renders as hex
+/// (`NumericString` and `UniversalString` among them).
+fn string_value(value: &Any<'_>) -> Option<String> {
+ let data = value.data;
+ match value.tag() {
+ // Latin-1, not ASCII: Java keeps a non-ASCII byte in these types as
its own code point.
+ Tag::PrintableString | Tag::Ia5String | Tag::GeneralString |
Tag::T61String => {
+ Some(data.iter().map(|&byte| char::from(byte)).collect())
+ }
+ Tag::Utf8String => Some(String::from_utf8_lossy(data).into_owned()),
+ Tag::BmpString => {
+ let units: Vec<u16> = data
+ .chunks(2)
+ .map(|pair| match pair {
+ [high, low] => u16::from_be_bytes([*high, *low]),
+ _ => 0xFFFD,
+ })
+ .collect();
+ Some(String::from_utf16_lossy(&units))
+ }
+ _ => None,
+ }
+}
+
+fn push_hex(out: &mut String, value: &Any<'_>) {
+ out.push('#');
+ let der = value
+ .to_der_vec()
+ .unwrap_or_else(|_| value.as_bytes().to_vec());
+ for byte in der {
+ let _ = write!(out, "{byte:02x}");
+ }
+}
+
+/// Backslash-escapes what Java's RFC 2253 rendering escapes: `,=+<>#;"\`
anywhere, NUL as `\00`,
+/// and every space or carriage return in the leading and trailing runs.
Everything else, other
+/// control characters included, is written as is.
+fn escape_into(out: &mut String, text: &str) {
+ let mut escaped: Vec<char> = Vec::with_capacity(text.len());
+ for character in text.chars() {
+ if character == '\0' {
+ escaped.extend(['\\', '0', '0']);
+ continue;
+ }
+ if matches!(
+ character,
+ ',' | '=' | '+' | '<' | '>' | '#' | ';' | '"' | '\\'
+ ) {
+ escaped.push('\\');
+ }
+ escaped.push(character);
+ }
+ let is_edge_space = |character: &char| matches!(character, ' ' | '\r');
+ let lead = escaped
+ .iter()
+ .position(|character| !is_edge_space(character))
+ .unwrap_or(escaped.len());
+ let trail = escaped
+ .iter()
+ .rposition(|character| !is_edge_space(character));
+ for (index, character) in escaped.into_iter().enumerate() {
+ let in_trailing_run = trail.is_none_or(|trail| index > trail);
+ if index < lead || in_trailing_run {
+ out.push('\\');
+ }
+ out.push(character);
+ }
+}
+
+/// Kafka's `ssl.principal.mapping.rules`: `RULE:pattern/replacement/[LU]` and
`DEFAULT`, comma
+/// separated, tried in order.
+///
+/// A rule applies when its pattern matches the whole subject (Java's
`matches()`). Its result is
+/// then Java's `subject.replaceAll(pattern, replacement)`: every match of the
unanchored pattern
+/// in the subject is replaced, an empty match at the end included, so
`RULE:(.*)/$1@x/` maps
+/// `alice` to `alice@x@x` exactly as Kafka does.
+///
+/// Patterns are a Java-compatible subset: `\d \s \w` are rewritten to ASCII
classes and `.` to
+/// exclude Java's line terminators, and constructs whose Rust meaning is
wider than Java's are
+/// refused as [`MappingRuleError::JavaIncompatible`].
+#[derive(Debug, Clone)]
+pub struct PrincipalMappingRules {
+ rules: Vec<Rule>,
+}
+
+#[derive(Debug, Clone)]
+enum Rule {
+ /// The subject string itself.
+ Default,
+ Rewrite {
+ whole: Regex,
+ finder: Regex,
+ replacement: Vec<Piece>,
+ case: Case,
+ },
+}
+
+/// One part of a parsed replacement.
+#[derive(Debug, Clone, PartialEq, Eq)]
+enum Piece {
+ Literal(String),
+ Group(usize),
+ Named(String),
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum Case {
+ Keep,
+ Lower,
+ Upper,
+}
+
+impl Default for PrincipalMappingRules {
+ fn default() -> Self {
+ Self {
+ rules: vec![Rule::Default],
+ }
+ }
+}
+
+impl PrincipalMappingRules {
+ /// Maps a rendered subject to a principal name, or `None` when no rule
matches.
+ #[must_use]
+ pub fn map(&self, subject: &str) -> Option<String> {
+ self.rules.iter().find_map(|rule| match rule {
+ Rule::Default => Some(subject.to_string()),
+ Rule::Rewrite {
+ whole,
+ finder,
+ replacement,
+ case,
+ } => {
+ if !whole.is_match(subject) {
+ return None;
+ }
+ let rewritten = replace_all(finder, subject, replacement);
+ Some(match case {
+ Case::Keep => rewritten,
+ Case::Lower => rewritten.to_lowercase(),
+ Case::Upper => rewritten.to_uppercase(),
+ })
+ }
+ })
+ }
+}
+
+impl FromStr for PrincipalMappingRules {
+ type Err = MappingRuleError;
+
+ fn from_str(raw: &str) -> Result<Self, Self::Err> {
+ let mut rules = Vec::new();
+ let mut rest = raw.trim_start();
+ if rest.is_empty() {
+ return Err(MappingRuleError::Empty);
+ }
+ while !rest.is_empty() {
+ if let Some(after) = rest.strip_prefix("DEFAULT") {
+ rules.push(Rule::Default);
+ rest = after;
+ } else if let Some(after) = rest.strip_prefix("RULE:") {
+ let (rule, after) = parse_rewrite(rest, after)?;
+ rules.push(rule);
+ rest = after;
+ } else {
+ return Err(MappingRuleError::UnknownRule(rest.to_string()));
+ }
+ rest = rest.trim_start();
+ match rest.strip_prefix(',') {
+ Some(after) => {
+ rest = after.trim_start();
+ if rest.is_empty() {
+ return Err(MappingRuleError::TrailingComma);
+ }
+ }
+ None if rest.is_empty() => {}
+ None => return
Err(MappingRuleError::MissingSeparator(rest.to_string())),
+ }
+ }
+ Ok(Self { rules })
+ }
+}
+
+/// Parses the `pattern/replacement/[LU]` after `RULE:`, returning the rule
and what follows it.
+fn parse_rewrite<'a>(rule: &str, body: &'a str) -> Result<(Rule, &'a str),
MappingRuleError> {
+ let (pattern, after) = take_until_slash(body)
+ .ok_or_else(||
MappingRuleError::UnterminatedPattern(rule.to_string()))?;
+ let (replacement, after) = take_until_slash(after)
+ .ok_or_else(||
MappingRuleError::UnterminatedReplacement(rule.to_string()))?;
+ let (case, after) = match after.chars().next() {
+ Some('L') => (Case::Lower, &after[1..]),
+ Some('U') => (Case::Upper, &after[1..]),
+ _ => (Case::Keep, after),
+ };
+ let invalid_pattern = |source| MappingRuleError::InvalidPattern {
+ pattern: pattern.clone(),
+ source,
+ };
+ let java = java_semantics(&pattern)?;
+ let finder = Regex::new(&java).map_err(invalid_pattern)?;
+ let whole = Regex::new(&format!("^(?:{java})$")).map_err(invalid_pattern)?;
+ let pieces = parse_replacement(&replacement, &finder).map_err(|source| {
+ MappingRuleError::InvalidReplacement {
+ pattern: pattern.clone(),
+ replacement: replacement.clone(),
+ source,
+ }
+ })?;
+ Ok((
+ Rule::Rewrite {
+ whole,
+ finder,
+ replacement: pieces,
+ case,
+ },
+ after,
+ ))
+}
+
+/// Rewrites `pattern` so Rust matches what `java.util.regex.Pattern` matches
without flags:
+/// `\d \s \w` and their negations become ASCII classes, and `.` outside
`(?s)` also stops at
+/// `\r`, U+0085, U+2028 and U+2029. Constructs with no faithful rewrite are
refused with their
+/// span.
+fn java_semantics(pattern: &str) -> Result<String, MappingRuleError> {
+ let parsed = ast::parse::Parser::new().parse(pattern).map_err(|error| {
+ MappingRuleError::InvalidPattern {
+ pattern: pattern.to_string(),
+ source: regex::Error::Syntax(error.to_string()),
+ }
+ })?;
+ let rewriter = JavaRewriter {
+ dot_all: vec![false],
+ edits: Vec::new(),
+ };
+ let edits = ast::visit(&parsed, rewriter).map_err(|(span, reason)| {
+ MappingRuleError::JavaIncompatible {
+ pattern: pattern.to_string(),
+ construct: pattern[span.start.offset..span.end.offset].to_string(),
+ reason,
+ }
+ })?;
+ let mut rewritten = pattern.to_string();
+ for (span, replacement) in edits.iter().rev() {
+ rewritten.replace_range(span.start.offset..span.end.offset,
replacement);
+ }
+ Ok(rewritten)
+}
+
+const JAVA_DOT: &str = r"[^\n\r\x{85}\x{2028}\x{2029}]";
+
+/// Collects `(span, replacement)` edits in pattern order.
+struct JavaRewriter {
+ /// Whether `.` matches line terminators, one entry per open group.
+ dot_all: Vec<bool>,
+ edits: Vec<(Span, String)>,
+}
+
+impl JavaRewriter {
+ fn apply_flags(&mut self, flags: &Flags) -> Result<(), (Span,
JavaIncompatibility)> {
Review Comment:
`JavaRewriter` still lets through `(?m)`, `(?x)`, `\v`, possessive
quantifiers like `.*+`, and `--`/`~~` inside classes. Rust reads all of them
differently from Java, so a rule can match DNs that it wouldn't match in Kafka.
I'd refuse them as `JavaIncompatible` like `(?i)` and `(?U)` already are,
the flags here in `apply_flags` and the rest in the visitor.
--
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]
