xmppftw/xmpp-rs
Watch
1
0
Fork
You've already forked xmpp-rs
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

stanzastream: handle malformed stanzas more gracefully

Browse source 
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.
This commit is contained in:
Jonas Schäfer 2026-06-06 13:55:27 +02:00
commit cc92d113fa
11 changed files with 401 additions and 31 deletions
Download patch file Download diff file Expand all files Collapse all files

4
tokio-xmpp/ChangeLog
View file

@ -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 <pep@bouah.net>

6
tokio-xmpp/src/client/login.rs
View file

@ -55,7 +55,11 @@ pub async fn auth<S: AsyncBufRead + AsyncWrite + Unpin>(
.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

6
tokio-xmpp/src/component/login.rs
View file

@ -43,7 +43,11 @@ pub async fn auth<S: AsyncBufRead + AsyncWrite + Unpin>(
.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(());
}

22
tokio-xmpp/src/component/stream.rs
View file

@ -18,16 +18,18 @@ impl<C: ServerConnector> Stream for Component<C> {
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
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,
}
}

6
tokio-xmpp/src/connect/starttls.rs
View file

@ -101,7 +101,11 @@ pub async fn starttls<S: TlsAsyncStream>(
.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;
}

28
tokio-xmpp/src/stanzastream/connected.rs
View file

@ -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

3
tokio-xmpp/src/stanzastream/negotiation.rs
View file

@ -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

102
tokio-xmpp/src/stanzastream/tests.rs
View file

@ -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<T>(v: Option<Result<T, ReadError>>) -> Result<T, io::Error> {
}
}
fn map_err(
opt: Option<Result<FallibleStreamElement, ReadError>>,
) -> Option<Result<XmppStreamElement, ReadError>> {
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 <presence/> 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 <presence/> 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:?}"),
}
}

4
tokio-xmpp/src/stanzastream/worker.rs
View file

@ -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<Box<dyn AsyncReadAndWrite + Send + 'static>, XmppStreamElement>;
crate::xmlstream::XmlStream<Box<dyn AsyncReadAndWrite + Send + 'static>, FallibleStreamElement>;
/// Underlying connection for a [`StanzaStream`][`super::StanzaStream`].
pub struct Connection {

6
tokio-xmpp/src/xmlstream/mod.rs
View file

@ -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<syntect::parsing::SyntaxSet> =
@ -502,4 +504,4 @@ impl<Io: AsyncWrite, T: FromXml> Future for Shutdown<'_, Io, T> {
}
/// Convenience alias for an XML stream using [`XmppStreamElement`].
pub type XmppStream<Io> = XmlStream<Io, XmppStreamElement>;
pub type XmppStream<Io> = XmlStream<Io, FallibleStreamElement>;

245
tokio-xmpp/src/xmlstream/xmpp.rs
View file

@ -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<String>,
to: Option<String>,
type_: Option<String>,
id: Option<String>,
},
}
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<CapturedMetadata>,
builder: FallibleBuilder<<XmppStreamElement as FromXml>::Builder, Error>,
}
impl FromEventsBuilder for FallibleStreamElementBuilder {
type Output = FallibleStreamElement;
fn feed(&mut self, ev: rxml::Event, ctx: &xso::Context) -> Result<Option<Self::Output>, 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<String>,
/// The unaltered `to` attribute, if present.
pub to: Option<String>,
/// The unaltered `type` attribute, if present.
pub type_: Option<String>,
/// The unaltered `id` attribute, if present.
pub id: Option<String>,
}
/// 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 `<iq/>`, `<presence/>` or `<message/>`
/// 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<XmppStreamElement, ReadError> {
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<Self::Builder, xso::error::FromEventsError> {
let metadata = Some(CapturedMetadata::new(&qname, &attrs));
let builder =
<Result<XmppStreamElement, Error> as FromXml>::from_events(qname, attrs, ctx)?;
Ok(FallibleStreamElementBuilder { metadata, builder })
}
}