From cc92d113fa6882b8313ab004e49d50e80421b133 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jonas=20Sch=C3=A4fer?= Date: Sat, 6 Jun 2026 13:55:27 +0200 Subject: [PATCH] stanzastream: handle malformed stanzas more gracefully This introduces the FallibleStreamElement type which, instead of failing parsing altogether, captures certain types of parse errors and allows the user (in this case, the stanzastream) to react to these errors appropriately. This allows us to drop invalid stanzas (e.g. with malformed values in strictly-checked fields) instead of failing the entire stream. Fixes #172. --- tokio-xmpp/ChangeLog | 4 + tokio-xmpp/src/client/login.rs | 6 +- tokio-xmpp/src/component/login.rs | 6 +- tokio-xmpp/src/component/stream.rs | 22 +- tokio-xmpp/src/connect/starttls.rs | 6 +- tokio-xmpp/src/stanzastream/connected.rs | 28 ++- tokio-xmpp/src/stanzastream/negotiation.rs | 3 + tokio-xmpp/src/stanzastream/tests.rs | 102 ++++++++- tokio-xmpp/src/stanzastream/worker.rs | 4 +- tokio-xmpp/src/xmlstream/mod.rs | 6 +- tokio-xmpp/src/xmlstream/xmpp.rs | 245 ++++++++++++++++++++- 11 files changed, 401 insertions(+), 31 deletions(-) diff --git a/tokio-xmpp/ChangeLog b/tokio-xmpp/ChangeLog index 740695e7..9ecf95fa 100644 --- a/tokio-xmpp/ChangeLog +++ b/tokio-xmpp/ChangeLog @@ -26,6 +26,10 @@ Version NEXT: - Drive the XMPP client stream in the background (!631) - Add option to split XMPP client into read and write half (!631) - Remove T: AsXml bound from XmlStream (!675) + - Invalid stanzas are now handled dropped (after increasing the SM + counters) instead of failing the stream. Previously, that might have + caused continuous reconnect loops esp. with stream management (#172, + !675). Version 5.0.0: 2025-10-28 pep diff --git a/tokio-xmpp/src/client/login.rs b/tokio-xmpp/src/client/login.rs index a13f0e49..7d69a2eb 100644 --- a/tokio-xmpp/src/client/login.rs +++ b/tokio-xmpp/src/client/login.rs @@ -55,7 +55,11 @@ pub async fn auth( .await?; loop { - match stream.next().await { + match stream + .next() + .await + .map(|v| v.map(|v| v.into_read_error()).flatten()) + { Some(Ok(XmppStreamElement::Sasl(sasl))) => match sasl { Nonza::Challenge(challenge) => { let response = mechanism diff --git a/tokio-xmpp/src/component/login.rs b/tokio-xmpp/src/component/login.rs index 08b93334..d09638c6 100644 --- a/tokio-xmpp/src/component/login.rs +++ b/tokio-xmpp/src/component/login.rs @@ -43,7 +43,11 @@ pub async fn auth( .await?; loop { - match stream.next().await { + match stream + .next() + .await + .map(|v| v.map(|v| v.into_read_error()).flatten()) + { Some(Ok(XmppStreamElement::ComponentHandshake(_))) => { return Ok(()); } diff --git a/tokio-xmpp/src/component/stream.rs b/tokio-xmpp/src/component/stream.rs index 07c6b766..13ec22d2 100644 --- a/tokio-xmpp/src/component/stream.rs +++ b/tokio-xmpp/src/component/stream.rs @@ -18,16 +18,18 @@ impl Stream for Component { fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll> { loop { match Pin::new(&mut self.stream).poll_next(cx) { - Poll::Ready(Some(Ok(XmppStreamElement::Stanza(stanza)))) => { - return Poll::Ready(Some(stanza)) - } - Poll::Ready(Some(Ok(_))) => - // unexpected - { - return Poll::Ready(None) - } - Poll::Ready(Some(Err(_))) => return Poll::Ready(None), - Poll::Ready(None) => return Poll::Ready(None), + Poll::Ready(v) => match v.map(|v| v.map(|v| v.into_read_error()).flatten()) { + Some(Ok(XmppStreamElement::Stanza(stanza))) => { + return Poll::Ready(Some(stanza)) + } + Some(Ok(_)) => + // unexpected + { + return Poll::Ready(None) + } + Some(Err(_)) => return Poll::Ready(None), + None => return Poll::Ready(None), + }, Poll::Pending => return Poll::Pending, } } diff --git a/tokio-xmpp/src/connect/starttls.rs b/tokio-xmpp/src/connect/starttls.rs index 9d30fad4..be0c3d71 100644 --- a/tokio-xmpp/src/connect/starttls.rs +++ b/tokio-xmpp/src/connect/starttls.rs @@ -101,7 +101,11 @@ pub async fn starttls( .await?; loop { - match stream.next().await { + match stream + .next() + .await + .map(|v| v.map(|v| v.into_read_error()).flatten()) + { Some(Ok(XmppStreamElement::Starttls(starttls::Nonza::Proceed(_)))) => { break; } diff --git a/tokio-xmpp/src/stanzastream/connected.rs b/tokio-xmpp/src/stanzastream/connected.rs index 6155b500..f9648f47 100644 --- a/tokio-xmpp/src/stanzastream/connected.rs +++ b/tokio-xmpp/src/stanzastream/connected.rs @@ -19,7 +19,7 @@ use xmpp_parsers::{ stream_features::StreamFeatures, }; -use crate::xmlstream::{ReadError, XmppStreamElement}; +use crate::xmlstream::{FallibleStreamElement, ReadError, StreamElementError, XmppStreamElement}; use crate::Stanza; use super::negotiation::{NegotiationResult, NegotiationState}; @@ -542,14 +542,14 @@ impl ConnectedState { }); match item { // Easy case, we got some data. - Ok(XmppStreamElement::Stanza(data)) => { + Ok(FallibleStreamElement::Ok(XmppStreamElement::Stanza(data))) => { if let Some(sm_state) = sm_state.as_mut() { sm_state.received(); } Poll::Ready(Some(ConnectedEvent::Worker(WorkerEvent::Stanza(data)))) } - Ok(XmppStreamElement::SM(sm::Nonza::Ack(ack))) => { + Ok(FallibleStreamElement::Ok(XmppStreamElement::SM(sm::Nonza::Ack(ack)))) => { if let Some(sm_state) = sm_state { match sm_state.remote_acked(ack.h) { Ok(()) => Poll::Ready(None), @@ -567,7 +567,7 @@ impl ConnectedState { } } - Ok(XmppStreamElement::SM(sm::Nonza::Req(_))) => { + Ok(FallibleStreamElement::Ok(XmppStreamElement::SM(sm::Nonza::Req(_)))) => { if let Some(sm_state) = sm_state { match sm_state.pending_acks.checked_add(1) { None => panic!("Too many pending ACKs, something is wrong."), @@ -588,7 +588,7 @@ impl ConnectedState { Poll::Ready(None) } - Ok(other) => { + Ok(FallibleStreamElement::Ok(other)) => { log::warn!( "Received unsupported stream element: {other:?}. Emitting stream error.", ); @@ -602,6 +602,16 @@ impl ConnectedState { Poll::Ready(None) } + Ok(FallibleStreamElement::Err( + e @ StreamElementError::InvalidStanza { .. }, + )) => { + log::warn!("Received invalid stanza: {e}; discarding silently."); + if let Some(sm_state) = sm_state.as_mut() { + sm_state.received(); + } + Poll::Ready(None) + } + // Another easy case: Soft timeouts are passed through // to the caller for handling. Err(ReadError::SoftTimeout) => { @@ -610,9 +620,11 @@ impl ConnectedState { // Parse errors are also just passed through (and will // likely cause us to send a stream error). - Err(ReadError::ParseError(e)) => { - Poll::Ready(Some(ConnectedEvent::Worker(WorkerEvent::ParseError(e)))) - } + Err(ReadError::ParseError(e)) + | Ok(FallibleStreamElement::Err(StreamElementError::InvalidNonza { + error: e, + .. + })) => Poll::Ready(Some(ConnectedEvent::Worker(WorkerEvent::ParseError(e)))), // I/O errors cause the stream to be considerde // broken; we drop it and send a Disconnect event with diff --git a/tokio-xmpp/src/stanzastream/negotiation.rs b/tokio-xmpp/src/stanzastream/negotiation.rs index f80086c3..52f703d9 100644 --- a/tokio-xmpp/src/stanzastream/negotiation.rs +++ b/tokio-xmpp/src/stanzastream/negotiation.rs @@ -197,6 +197,7 @@ impl NegotiationState { "eof before stream footer", ))) }); + let item = item.map(|v| v.into_read_error()).flatten(); match item { Ok(XmppStreamElement::Stanza(data)) => match data { @@ -387,6 +388,8 @@ impl NegotiationState { "eof before stream footer", ))) }); + let item = item.map(|v| v.into_read_error()).flatten(); + match item { // Pre-SM data. Note that we mustn't count this while we // are still in negotiating state: we transit to diff --git a/tokio-xmpp/src/stanzastream/tests.rs b/tokio-xmpp/src/stanzastream/tests.rs index 74758b7f..c9a3771b 100644 --- a/tokio-xmpp/src/stanzastream/tests.rs +++ b/tokio-xmpp/src/stanzastream/tests.rs @@ -7,13 +7,15 @@ use xmpp_parsers::{ bind::{BindFeature, BindQuery, BindResponse}, iq::Iq, message::Message, + presence::{Presence, Show, Type as PresenceType}, sm, }; use crate::jid::{BareJid, ResourcePart}; +use crate::minidom::Element; use crate::xmlstream::{ - accept_stream, initiate_stream, ReadError, RecvFeaturesError, StreamHeader, Timeouts, - XmppStreamElement, + accept_stream, initiate_stream, FallibleStreamElement, ReadError, RecvFeaturesError, + StreamHeader, Timeouts, XmppStreamElement, }; use super::*; @@ -49,6 +51,12 @@ fn map_io(v: Option>) -> Result { } } +fn map_err( + opt: Option>, +) -> Option> { + opt.map(|res| res.map(|v| v.into_read_error()).flatten()) +} + async fn custom_stream_pair( identity: BareJid, mut features: StreamFeatures, @@ -117,7 +125,7 @@ async fn custom_stream_pair( 16, ); - match map_io(server.next().await)? { + match map_io(map_err(server.next().await))? { XmppStreamElement::Stanza(stanza) => match stanza { Stanza::Iq(Iq::Set { from, @@ -168,7 +176,7 @@ async fn fresh_sm_pair() -> io::Result<(StanzaStream, XmppStream)> { features.stream_management = Some(sm::StreamManagement { optional: false }); let (mut client, mut server) = custom_stream_pair("client@domain.example".parse().unwrap(), features).await?; - match map_io(server.next().await)? { + match map_io(map_err(server.next().await))? { XmppStreamElement::SM(sm) => match sm { sm::Nonza::Enable(sm::Enable { max, resume }) => { server @@ -199,8 +207,92 @@ async fn fresh_sm_pair() -> io::Result<(StanzaStream, XmppStream)> { async fn sm_negotiation() { let (client, mut server) = fresh_sm_pair().await.expect("stream setup failed"); client.send(Box::new(Message::new(None).into())).await; - match map_io(server.next().await) { + match map_io(map_err(server.next().await)) { Ok(XmppStreamElement::Stanza(Stanza::Message(Message { .. }))) => (), other => panic!("unexpected recv result on server side: {other:?}"), } } + +#[cfg(not(feature = "component"))] +#[tokio::test] +async fn drop_stanza_with_unparsable_payload_nonsm() { + let (mut client, mut server) = plain_pair().await.expect("stream setup failed"); + + // Here, we construct a deliberately broken stanza. + let mut presence = Presence::new(PresenceType::None); + presence.show = Some(Show::Away); + let mut presence: Element = presence.into(); + for child in presence.children_mut() { + if child.is("show", "jabber:client") { + child.append_text("-foo"); + } + } + + server + .send(&presence) + .await + .expect("unexpected send result on server side"); + server + .send(&Presence::new(PresenceType::Unavailable)) + .await + .expect("unexpected send result on server side"); + + match client.next().await { + Some(Event::Stanza(Stanza::Presence(Presence { type_, .. }))) => { + assert_eq!(type_, PresenceType::Unavailable); + } + other => panic!("unexpected recv result on client side: {other:?}"), + } +} + +#[cfg(not(feature = "component"))] +#[tokio::test] +async fn drop_stanza_with_unparsable_payload_sm() { + let (mut client, mut server) = fresh_sm_pair().await.expect("stream setup failed"); + + // Here, we construct a deliberately broken stanza. + let mut presence = Presence::new(PresenceType::None); + presence.show = Some(Show::Away); + let mut presence: Element = presence.into(); + for child in presence.children_mut() { + if child.is("show", "jabber:client") { + child.append_text("-foo"); + } + } + + server + .send(&presence) + .await + .expect("unexpected send result on server side"); + server + .send(&sm::R) + .await + .expect("unexpected send result on server side"); + match map_io(map_err(server.next().await)) { + Ok(XmppStreamElement::SM(sm::Nonza::Ack(sm::A { h }))) => { + assert_eq!(h, 1); + } + other => panic!("unexpected recv result on the server side: {other:?}"), + } + server + .send(&Presence::new(PresenceType::Unavailable)) + .await + .expect("unexpected send result on server side"); + server + .send(&sm::R) + .await + .expect("unexpected send result on server side"); + match map_io(map_err(server.next().await)) { + Ok(XmppStreamElement::SM(sm::Nonza::Ack(sm::A { h }))) => { + assert_eq!(h, 2); + } + other => panic!("unexpected recv result on the server side: {other:?}"), + } + + match client.next().await { + Some(Event::Stanza(Stanza::Presence(Presence { type_, .. }))) => { + assert_eq!(type_, PresenceType::Unavailable); + } + other => panic!("unexpected recv result on client side: {other:?}"), + } +} diff --git a/tokio-xmpp/src/stanzastream/worker.rs b/tokio-xmpp/src/stanzastream/worker.rs index 3a1eec7e..a214bf46 100644 --- a/tokio-xmpp/src/stanzastream/worker.rs +++ b/tokio-xmpp/src/stanzastream/worker.rs @@ -26,7 +26,7 @@ use xmpp_parsers::{ }; use crate::connect::AsyncReadAndWrite; -use crate::xmlstream::{ReadError, XmppStreamElement}; +use crate::xmlstream::{FallibleStreamElement, ReadError}; use crate::Stanza; use super::connected::{ConnectedEvent, ConnectedState}; @@ -38,7 +38,7 @@ use super::{Event, StreamEvent}; /// Convenience alias for [`XmlStreams`][`crate::xmlstream::XmlStream`] which /// may be used with [`StanzaStream`][`super::StanzaStream`]. pub type XmppStream = - crate::xmlstream::XmlStream, XmppStreamElement>; + crate::xmlstream::XmlStream, FallibleStreamElement>; /// Underlying connection for a [`StanzaStream`][`super::StanzaStream`]. pub struct Connection { diff --git a/tokio-xmpp/src/xmlstream/mod.rs b/tokio-xmpp/src/xmlstream/mod.rs index 828fe6e2..3f9b60df 100644 --- a/tokio-xmpp/src/xmlstream/mod.rs +++ b/tokio-xmpp/src/xmlstream/mod.rs @@ -85,7 +85,9 @@ use self::common::{RawError, RawXmlStream, ReadXsoError, ReadXsoState}; pub use self::common::{StreamHeader, Timeouts}; pub use self::initiator::{InitiatingStream, PendingFeaturesRecv, RecvFeaturesError}; pub use self::responder::{AcceptedStream, PendingFeaturesSend}; -pub use self::xmpp::XmppStreamElement; +pub use self::xmpp::{ + FallibleStreamElement, RawStanzaHeader, StreamElementError, XmppStreamElement, +}; #[cfg(feature = "syntax-highlighting")] static PS: LazyLock = @@ -502,4 +504,4 @@ impl Future for Shutdown<'_, Io, T> { } /// Convenience alias for an XML stream using [`XmppStreamElement`]. -pub type XmppStream = XmlStream; +pub type XmppStream = XmlStream; diff --git a/tokio-xmpp/src/xmlstream/xmpp.rs b/tokio-xmpp/src/xmlstream/xmpp.rs index 600b181d..18364169 100644 --- a/tokio-xmpp/src/xmlstream/xmpp.rs +++ b/tokio-xmpp/src/xmlstream/xmpp.rs @@ -4,12 +4,19 @@ // License, v. 2.0. If a copy of the MPL was not distributed with this // file, You can obtain one at http://mozilla.org/MPL/2.0/. -use xso::{AsXml, FromXml}; +use std::fmt; +use std::io; + +use rxml::NcNameStr; + +use xso::{error::Error, fromxml::FallibleBuilder, AsXml, FromEventsBuilder, FromXml}; use xmpp_parsers::{component, sasl, sm, starttls, stream_error::ReceivedStreamError}; use crate::Stanza; +use super::ReadError; + /// Any valid XMPP stream-level element. #[derive(FromXml, AsXml, Debug)] #[xml()] @@ -38,3 +45,239 @@ pub enum XmppStreamElement { #[xml(transparent)] SM(sm::Nonza), } + +#[derive(Debug)] +pub enum PartialStanza { + Presence, + Iq, + Message, +} + +impl fmt::Display for PartialStanza { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + match self { + Self::Presence => f.write_str("presence"), + Self::Message => f.write_str("message"), + Self::Iq => f.write_str("iq"), + } + } +} + +impl PartialStanza { + pub fn to_ncname(&self) -> &'static NcNameStr { + match self { + Self::Presence => rxml::xml_ncname!("presence"), + Self::Message => rxml::xml_ncname!("message"), + Self::Iq => rxml::xml_ncname!("iq"), + } + } +} + +enum CapturedMetadata { + Nonza { + qname: rxml::QName, + }, + Stanza { + ns: rxml::Namespace<'static>, + name: PartialStanza, + from: Option, + to: Option, + type_: Option, + id: Option, + }, +} + +impl CapturedMetadata { + pub fn new(qname: &rxml::QName, attrs: &rxml::AttrMap) -> Self { + let kind = match qname.1.as_str() { + "presence" => Some(PartialStanza::Presence), + "iq" => Some(PartialStanza::Iq), + "message" => Some(PartialStanza::Message), + _ => None, + }; + if let Some(kind) = kind { + CapturedMetadata::Stanza { + ns: qname.0.clone(), + name: kind, + from: attrs.get(&rxml::Namespace::NONE, "from").cloned(), + to: attrs.get(&rxml::Namespace::NONE, "to").cloned(), + id: attrs.get(&rxml::Namespace::NONE, "id").cloned(), + type_: attrs.get(&rxml::Namespace::NONE, "type").cloned(), + } + } else { + CapturedMetadata::Nonza { + qname: qname.clone(), + } + } + } +} + +pub struct FallibleStreamElementBuilder { + metadata: Option, + builder: FallibleBuilder<::Builder, Error>, +} + +impl FromEventsBuilder for FallibleStreamElementBuilder { + type Output = FallibleStreamElement; + + fn feed(&mut self, ev: rxml::Event, ctx: &xso::Context) -> Result, Error> { + match self.builder.feed(ev, ctx) { + Ok(Some(output)) => Ok(Some(match output { + Ok(v) => FallibleStreamElement::Ok(v), + + // The FallibleBuilder should never ever emit an rxml::Error, + // because it cannot *receive* rxml::Error via `feed`. + Err(Error::XmlError(e)) => unreachable!("feed somehow saw an rxml error: {e}"), + + // This error condition can not be emitted from feed. + Err(Error::TypeMismatch) => unreachable!("feed somehow saw a TypeMismatch"), + + Err(error) => FallibleStreamElement::Err( + match self.metadata.take().expect("feed called after completion") { + CapturedMetadata::Nonza { qname } => { + StreamElementError::InvalidNonza { qname, error } + } + CapturedMetadata::Stanza { + ns, + name, + from, + to, + type_, + id, + } => StreamElementError::InvalidStanza { + ns, + name, + header: RawStanzaHeader { + from, + to, + type_, + id, + }, + error, + }, + }, + ), + })), + Ok(None) => Ok(None), + Err(e) => Err(e), + } + } +} + +/// Container for unparsed stanza attributes. +#[derive(Debug)] +pub struct RawStanzaHeader { + /// The unaltered `from` attribute, if present. + pub from: Option, + + /// The unaltered `to` attribute, if present. + pub to: Option, + + /// The unaltered `type` attribute, if present. + pub type_: Option, + + /// The unaltered `id` attribute, if present. + pub id: Option, +} + +/// Error condition arising from failing to convert a stream-level element +/// into a [`xmpp_parsers`] struct. +#[derive(Debug)] +pub enum StreamElementError { + /// Failed to convert an expected ``, `` or `` + /// element into a struct. + InvalidStanza { + /// Namespace of the element. + ns: rxml::Namespace<'static>, + + /// Name of the element. + name: PartialStanza, + + /// Header attributes of the invalid stanza. + header: RawStanzaHeader, + + /// The error which caused the stanza to fail to parse. + /// + /// Note that this is never `xso::error::Error::TypeMismatch`, because + /// type mismatches do not even start to parse with `FromXml`. + error: xso::error::Error, + }, + + /// Invalid top-level stream element. + /// + /// This is reported if the element header matched the + /// [`XmppStreamElement`] type, but the payload failed to parse and it was + /// not a stanza. + InvalidNonza { + /// Qualified name of the element which failed to parse. + qname: rxml::QName, + + /// The error which caused the stanza to fail to parse. + /// + /// Note that this is never `xso::error::Error::TypeMismatch`, because + /// type mismatches do not even start to parse with `FromXml`. + error: xso::error::Error, + }, +} + +impl fmt::Display for StreamElementError { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + match self { + Self::InvalidNonza { qname, error, .. } => write!( + f, + "invalid nonza received: <{{{}}}{}/> ({error})", + qname.0, qname.1 + ), + Self::InvalidStanza { + ns, name, error, .. + } => write!( + f, + "invalid stanza received: <{{{}}}{}/> ({error})", + ns, name + ), + } + } +} + +impl core::error::Error for StreamElementError {} + +/// Wrapper type to catch parse errors of [`XmppStreamElement`] items. +#[derive(Debug)] +pub enum FallibleStreamElement { + /// Parsing succeeded. + Ok(XmppStreamElement), + + /// Parsing failed. + Err(StreamElementError), +} + +impl FallibleStreamElement { + /// Convert the contained error condition (if any) to a [`ReadError`]. + /// + /// This can be used in places where you do not care to handle the various + /// error conditions separately. + pub fn into_read_error(self) -> Result { + match self { + Self::Ok(v) => Ok(v), + Self::Err(e) => Err(ReadError::HardError(io::Error::new( + io::ErrorKind::InvalidData, + e, + ))), + } + } +} + +impl FromXml for FallibleStreamElement { + type Builder = FallibleStreamElementBuilder; + + fn from_events( + qname: rxml::QName, + attrs: rxml::AttrMap, + ctx: &xso::Context, + ) -> Result { + let metadata = Some(CapturedMetadata::new(&qname, &attrs)); + let builder = + as FromXml>::from_events(qname, attrs, ctx)?; + Ok(FallibleStreamElementBuilder { metadata, builder }) + } +}