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 }) + } +}