OF-2534: Add inline XEP-0198 stream resumption to SASL2 Bind2

xmppserver/src/main/java/org/jivesoftware/openfire/SessionManager.java

@@ -1653,6 +1653,19 @@ private class ClientSessionListener implements ConnectionCloseListener
         public CompletableFuture<Void> onConnectionClosing(Object handback)
         {
             final LocalClientSession session = (LocalClientSession) handback;
+            Log.trace("onConnectionClosing invoked for session with address {} and streamID {}: isDetached={}, resume={}, currentConnection={}.", session.getAddress(), session.getStreamID(), session.isDetached(), session.getStreamManager().getResume(), session.getConnection());
+
+            // A close notification can arrive for a connection that has since been superseded by a resumed session
+            // (XEP-0198, traditional or inline SASL2): the old connection is closed deliberately as part of the
+            // handoff, but this listener callback is asynchronous and may run after the session has already been
+            // reattached to a different, live connection. Treat that as a no-op rather than tearing down a session
+            // that is connected right now.
+            final Connection currentConnection = session.getConnection();
+            if (currentConnection != null && !currentConnection.isClosed()) {
+                Log.debug("Ignoring stale close notification for session with address {} and streamID {}: it already has a different, live connection.", session.getAddress(), session.getStreamID());
+                return CompletableFuture.completedFuture(null);
+            }
+
             if (session.isDetached()) {
                 Log.debug("Closing client session with address {} and streamID {} is detached already; this is a no-op.", session.getAddress(), session.getStreamID());
                 return CompletableFuture.completedFuture(null);

xmppserver/src/main/java/org/jivesoftware/openfire/net/SaslStreamFeatures.java

@@ -27,6 +27,7 @@
 import org.jivesoftware.openfire.session.ClientSession;
 import org.jivesoftware.openfire.session.LocalIncomingServerSession;
 import org.jivesoftware.openfire.session.LocalSession;
+import org.jivesoftware.openfire.streammanagement.StreamManager;
 import org.jivesoftware.util.JiveGlobals;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -173,8 +174,12 @@ static Element asSASLMechanismsElementForClientSessions(@Nonnull final Set<Strin
         if ( usingSASL2 )
         {
             Element inlineElement = result.addElement("inline");
+
+            if (StreamManager.isStreamManagementActive()) {
+                inlineElement.add(StreamManager.sasl2InlineFeatureElement());
+            }
             inlineElement.add(Bind2Request.featureElement());
-            // Element sm = inlineElement.addElement(...);
+
             if (FastTokenManager.ENABLE_FAST.getValue()) {
                 final Set<String> fastMechanisms = advertisableMechanismNames.stream()
                     .filter(MechanismName::isFast).collect(Collectors.toSet());

xmppserver/src/main/java/org/jivesoftware/openfire/net/StanzaHandler.java

@@ -92,6 +92,13 @@ public abstract class StanzaHandler {
      */
     protected boolean usingSASL2 = false;
 
+    /**
+     * Flag that indicates that SASL2 authentication succeeded by inline-resuming a pre-existing session (XEP-0198
+     * § 9.2), rather than by binding a (new or Bind2) resource. When set, {@link #sasl2Successful()} must not
+     * (re)send post-authentication stream features, per XEP-0198 § 9.2.
+     */
+    protected boolean sasl2SessionResumed = false;
+
     /**
      * SASL status based on the last SASL interaction
      */
@@ -243,20 +250,31 @@ else if ("auth".equals(tag)) {
             // User is trying to authenticate using SASL2.
             startedSASL = true;
             usingSASL2 = true;
-            saslStatus = SASLAuthentication.handle(session, doc, usingSASL2);
+            // An inline XEP-0198 resumption transfers the connection to the resumed session, which (through
+            // Connection#reinit) replaces this handler's 'session' field before handle() returns. Retain the session
+            // that is negotiating the authentication, as that is where the outcome of the negotiation is recorded.
+            final LocalSession authenticatingSession = session;
+            saslStatus = SASLAuthentication.handle(authenticatingSession, doc, usingSASL2);
             if (saslStatus == SASLAuthentication.Status.authenticated && usingSASL2) {
                 // No Bind2: send features synchronously now.
                 startedSASL = false; // Without a multi-step SASL mechanism, this can be reset here immediately, rather than in initiateSession (as SASL1 does).
                 sasl2Successful();
             } else if (saslStatus == SASLAuthentication.Status.authenticatedAwaitingFeatures) {
                 // Bind2: <success/> and features are delivered asynchronously by SASLAuthentication.
                 startedSASL = false;
+            } else if (saslStatus == SASLAuthentication.Status.authenticatedResumed) {
+                // Inline XEP-0198 resume: <success/> (with <resumed/>) was already delivered, over the resumed
+                // session, by SASLAuthentication. Adopt that session and suppress stream features (XEP-0198 § 9.2).
+                startedSASL = false;
+                adoptSasl2ResumedSession(authenticatingSession);
             }
             // If authenticatedAwaitingFeatures, <success/> and features are delivered asynchronously
             // by SASLAuthentication once Bind2 resource binding completes.
         } else if (startedSASL && ("response".equals(tag) || "abort".equals(tag))) {
             // User is responding to SASL challenge. Process response
-            saslStatus = SASLAuthentication.handle(session, doc, usingSASL2);
+            // See the 'authenticate' branch: an inline XEP-0198 resumption can replace this handler's session.
+            final LocalSession authenticatingSession = session;
+            saslStatus = SASLAuthentication.handle(authenticatingSession, doc, usingSASL2);
             if (saslStatus == SASLAuthentication.Status.failed) {
                 startedSASL = false;
                 usingSASL2 = false;
@@ -267,6 +285,11 @@ else if ("auth".equals(tag)) {
             } else if (saslStatus == SASLAuthentication.Status.authenticatedAwaitingFeatures) {
                 // Bind2: <success/> and features are delivered asynchronously by SASLAuthentication.
                 startedSASL = false;
+            } else if (saslStatus == SASLAuthentication.Status.authenticatedResumed) {
+                // Inline XEP-0198 resume: <success/> (with <resumed/>) was already delivered, over the resumed
+                // session, by SASLAuthentication. Adopt that session and suppress stream features (XEP-0198 § 9.2).
+                startedSASL = false;
+                adoptSasl2ResumedSession(authenticatingSession);
             }
             // If authenticatedAwaitingFeatures, <success/> and features are delivered asynchronously
             // by SASLAuthentication once Bind2 resource binding completes.
@@ -581,14 +604,57 @@ protected void saslSuccessful() {
 
     /**
      * Emits post-authentication stream features for SASL2 (XEP-0388), which does NOT restart the stream.
-     * On TCP the features element is sent inline in the existing stream. Transports with different framing
-     * (e.g. RFC 7395 WebSocket) override this.
+     *
+     * When the SASL2 authentication succeeded by inline-resuming a pre-existing session (XEP-0198 § 9.2), features
+     * are deliberately not (re)sent: the resumed stream is considered re-established immediately after the
+     * {@code <success/>} element, and XEP-0198 § 9.2 mandates that stream features MUST NOT be sent in this case.
      */
     protected void sasl2Successful() {
+        if (!sasl2SessionResumed) {
+            deliverSasl2Features();
+        }
+    }
+
+    /**
+     * Delivers post-authentication stream features for SASL2 (XEP-0388). On TCP the features element is sent
+     * inline in the existing stream. Transports with different framing (e.g. RFC 7395 WebSocket) override this.
+     */
+    protected void deliverSasl2Features() {
         final Element features = generateFeatures();
         connection.deliverRawText(features.asXML());
     }
 
+    /**
+     * Adopts the pre-existing session that a SASL2 authentication resumed inline (XEP-0198 § 9.2), replacing the
+     * temporary session that was negotiating the SASL2 authentication.
+     *
+     * The {@code <success/>} response (including the {@code <resumed/>} element) has already been delivered, over
+     * the resumed session, by {@link SASLAuthentication}, and XEP-0198 § 9.2 forbids sending stream features after
+     * it. This method therefore only switches this handler over to the resumed session; no features are sent. The
+     * {@link #sasl2SessionResumed} flag it sets guards {@link #sasl2Successful()} against a future caller that
+     * would.
+     *
+     * Note that transferring the connection re-initializes it for its new owner, which on some transports already
+     * replaces this handler's session. The switch is performed here regardless, so that this does not depend on the
+     * transport. For the same reason, the session that negotiated the authentication (which holds the outcome of
+     * that negotiation) must be provided by the caller, rather than read from {@link #session}.
+     *
+     * @param authenticatingSession the session that negotiated the SASL2 authentication (cannot be null).
+     */
+    protected void adoptSasl2ResumedSession(final LocalSession authenticatingSession) {
+        final Object data = authenticatingSession.removeSessionData(SASLAuthentication.SASL2_RESUMED_SESSION);
+        if (!(data instanceof LocalSession resumedSession)) {
+            // Unreachable in practice: SASLAuthentication only reports 'authenticatedResumed' after having stored the
+            // resumed session under this key. If it does happen, the client has already been told that its stream was
+            // resumed, over a connection that this handler can no longer serve. There is nothing to do but disconnect.
+            Log.error("Expected a resumed session to be available in session data under key '{}', but found: {}. Closing the connection.", SASLAuthentication.SASL2_RESUMED_SESSION, data);
+            connection.close(new StreamError(StreamError.Condition.internal_server_error, "Unable to complete inline stream resumption."));
+            return;
+        }
+        this.session = resumedSession;
+        sasl2SessionResumed = true;
+    }
+
     /**
      * Helper to generate stream:features, populated simply from the session.,
      *

xmppserver/src/main/java/org/jivesoftware/openfire/nio/NettyConnection.java

@@ -226,8 +226,15 @@ public void close(@Nullable final StreamError error) {
             ChannelFuture f;
 
             if (session != null) {
-                // If the stream was ended because of an error, it should not be possible to resume it (OF-2751).
-                if (error != null) {
+                // If the stream ended because of an error, it should not be possible to resume it (OF-2751).
+                // Exception: if the session is already detached by the time this runs, the error isn't reporting a
+                // genuine failure of this stream. It's StreamManager#detachIfNeeded() closing a connection that has
+                // already been handed off to a newly resumed session, choosing to still send a 'conflict' StreamError
+                // only because XEP-0198 §5 recommends that for a superseded former stream that is still open.
+                // detachIfNeeded() always detaches the session before closing its connection, so a detached session
+                // at this point reliably signals that hand-off, not a failure - and that hand-off must not disable
+                // resumption of the very session it's transferring the connection to.
+                if (error != null && !session.isDetached()) {
                     session.getStreamManager().formalClose();
                 }
 

xmppserver/src/main/java/org/jivesoftware/openfire/session/LocalClientSession.java

@@ -755,9 +755,10 @@ public void setOfflineFloodStopped(boolean offlineFloodStopped) {
         }
     }
 
-    public void reattach(LocalSession connectionProvider, long h)
+    @Override
+    protected void onReattached()
     {
-        super.reattach(connectionProvider, h);
+        super.onReattached();
 
         // XEP-0352: "After a previous stream was resumed using mechanisms like Stream Management (XEP-0198), the CSI
         // state is not restored. That is, stream resumption does not affect the current CSI state, which always

xmppserver/src/main/java/org/jivesoftware/openfire/session/LocalSession.java

@@ -187,6 +187,53 @@ Connection releaseConnection()
      * @param h the sequence number of the last handled stanza sent over the former stream
      */
     public void reattach(LocalSession connectionProvider, long h) {
+        reattachConnection(connectionProvider);
+        this.streamManager.onResume(new JID(null, this.serverName, null, true), h);
+        this.sessionManager.removeSession((LocalClientSession) connectionProvider);
+        onReattached();
+    }
+
+    /**
+     * Reattach the (existing) session to the connection provided by a new session, for an inline XEP-0198 resume
+     * request that is embedded in a SASL2 (XEP-0388) authentication exchange (see XEP-0198 § 9.2).
+     *
+     * This transfers the connection exactly as {@link #reattach(LocalSession, long)} does, but, unlike that method,
+     * does <em>not</em> complete the resumption: a SASL2 caller cannot have the stream manager write a
+     * {@code <resumed/>} directly to the connection, since that element needs to be embedded in the SASL2
+     * {@code <success/>} response that the caller is still constructing. Callers must invoke
+     * {@link #completeSasl2Resume(long)} after they have written that response.
+     *
+     * @param connectionProvider Session from which to obtain the connection from.
+     */
+    public void reattachForSasl2(LocalSession connectionProvider) {
+        reattachConnection(connectionProvider);
+        this.sessionManager.removeSession((LocalClientSession) connectionProvider);
+    }
+
+    /**
+     * Completes an inline SASL2 resumption, after the caller has delivered the SASL2 {@code <success/>} response
+     * carrying the {@code <resumed/>} element built by StreamManager#buildResumedElement().
+     *
+     * This performs the second half of what {@link StreamManager#onResume(JID, long)} does for the traditional flow:
+     * it processes the client's acknowledgement, retransmits anything still unacknowledged, and then invokes
+     * {@link #onReattached()}. It must not be invoked before the {@code <success/>} has been written: everything it
+     * delivers would otherwise precede the resumption confirmation on the wire.
+     *
+     * @param h the sequence number of the last handled stanza, as reported by the resuming client.
+     */
+    public void completeSasl2Resume(final long h) {
+        this.streamManager.redeliverUnackedStanzas(new JID(null, this.serverName, null, true), h);
+        onReattached();
+    }
+
+    /**
+     * Transfers the connection of connectionProvider to this session, closing any (stale) connection that this session
+     * might still have. Note that this does not invoke onReattached(): that is deferred until the resumption has been
+     * confirmed to the client (see reattach(LocalSession, long) and completeSasl2Resume(long)).
+     *
+     * @param connectionProvider Session from which to obtain the connection from.
+     */
+    private void reattachConnection(LocalSession connectionProvider) {
         lock.lock();
         try {
             Log.debug("Reattaching session with address {} and streamID {} using connection from session with address {} and streamID {}.", this.address, this.streamID, connectionProvider.getAddress(), connectionProvider.getStreamID());
@@ -196,13 +243,24 @@ public void reattach(LocalSession connectionProvider, long h) {
             }
             this.conn = connectionProvider.releaseConnection();
             this.conn.reinit(this);
-        }finally {
+        } finally {
             lock.unlock();
         }
         this.status = Session.Status.AUTHENTICATED;
         this.sessionManager.removeDetached(this);
-        this.streamManager.onResume(new JID(null, this.serverName, null, true), h);
-        this.sessionManager.removeSession((LocalClientSession)connectionProvider);
+        Log.debug("Reattach complete for session with address {} and streamID {}: status={}, resumable={}, detached={}.", this.address, this.streamID, this.status, this.streamManager.getResume(), this.isDetached());
+    }
+
+    /**
+     * Hook invoked after this session's connection has been transferred from another session, either through
+     * {@link #reattach(LocalSession, long)} or {@link #completeSasl2Resume(long)}.
+     *
+     * This is invoked only after the resumption has been confirmed and unacknowledged stanzas retransmitted.
+     *
+     * The default implementation does nothing; subclasses can override this to restore state that a resumed stream is
+     * expected to reset.
+     */
+    protected void onReattached() {
     }
 
     /**

xmppserver/src/main/java/org/jivesoftware/openfire/streammanagement/MalformedResumeRequestException.java

@@ -0,0 +1,28 @@
+/*
+ * Copyright (C) 2026 Ignite Realtime Foundation. All rights reserved.
+ *
+ * Licensed 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.
+ */
+package org.jivesoftware.openfire.streammanagement;
+
+/**
+ * Thrown by {@link ResumeRequest} when a XEP-0198 {@code <resume/>} element cannot be parsed, because it is
+ * missing a required attribute, or one of its attributes has an illegal value.
+ */
+public class MalformedResumeRequestException extends Exception
+{
+    public MalformedResumeRequestException(final String message)
+    {
+        super(message);
+    }
+}