Fix #5883: bind session/view push channels to their HTTP session

impl/src/main/java/com/sun/faces/config/ConfigureListener.java

@@ -28,9 +28,13 @@
 import static com.sun.faces.config.WebConfiguration.BooleanWebContextInitParameter.ForceLoadFacesConfigFiles;
 import static com.sun.faces.config.WebConfiguration.BooleanWebContextInitParameter.VerifyFacesConfigObjects;
 import static com.sun.faces.config.WebConfiguration.WebContextInitParameter.JakartaFacesProjectStage;
+import static com.sun.faces.config.WebConfiguration.WebContextInitParameter.WebsocketEndpointIdleTimeout;
+import static com.sun.faces.config.WebConfiguration.WebContextInitParameter.WebsocketMaxSessionsPerChannel;
 import static com.sun.faces.context.SessionMap.createMutex;
 import static com.sun.faces.context.SessionMap.removeMutex;
 import static com.sun.faces.push.WebsocketEndpoint.URI_TEMPLATE;
+import static com.sun.faces.push.WebsocketEndpoint.USER_PROPERTY_IDLE_TIMEOUT;
+import static com.sun.faces.push.WebsocketEndpoint.USER_PROPERTY_MAX_SESSIONS_PER_CHANNEL;
 import static java.lang.Boolean.TRUE;
 import static java.text.MessageFormat.format;
 import static java.util.logging.Level.FINE;
@@ -99,6 +103,11 @@ public class ConfigureListener implements ServletRequestListener, HttpSessionLis
 
     private static final Logger LOGGER = FacesLogger.CONFIG.getLogger();
 
+    private static final String ERROR_INVALID_WEBSOCKET_ENDPOINT_IDLE_TIMEOUT =
+            "Context param ''{0}'' must represent a number of milliseconds of 0 or greater, but was: ''{1}''.";
+    private static final String ERROR_INVALID_WEBSOCKET_MAX_SESSIONS_PER_CHANNEL =
+            "Context param ''{0}'' must represent a number of 1 or greater, but was: ''{1}''.";
+
     private ScheduledThreadPoolExecutor webResourcePool;
 
     protected WebappLifecycleListener webAppListener;
@@ -230,7 +239,12 @@ public void contextInitialized(ServletContextEvent servletContextEvent) {
                         " The current websocket container implementation does not support programmatically registering a container-provided endpoint.");
                 }
 
-                serverContainer.addEndpoint(ServerEndpointConfig.Builder.create(WebsocketEndpoint.class, URI_TEMPLATE).build());
+                ServerEndpointConfig endpointConfig = ServerEndpointConfig.Builder.create(WebsocketEndpoint.class, URI_TEMPLATE)
+                        .configurator(new WebsocketEndpoint.Configurator())
+                        .build();
+                endpointConfig.getUserProperties().put(USER_PROPERTY_IDLE_TIMEOUT, getWebsocketEndpointIdleTimeout(webConfig));
+                endpointConfig.getUserProperties().put(USER_PROPERTY_MAX_SESSIONS_PER_CHANNEL, getWebsocketMaxSessionsPerChannel(webConfig));
+                serverContainer.addEndpoint(endpointConfig);
             }
 
             webConfig.doPostBringupActions();
@@ -264,6 +278,45 @@ public void contextInitialized(ServletContextEvent servletContextEvent) {
         }
     }
 
+    private static long getWebsocketEndpointIdleTimeout(WebConfiguration webConfig) {
+        String value = webConfig.getOptionValue(WebsocketEndpointIdleTimeout);
+        long idleTimeout = toLong(value, -1); // A non-numeric value maps to -1 because 0 is a valid value meaning no timeout.
+
+        if (idleTimeout < 0) {
+            throw new IllegalArgumentException(format(ERROR_INVALID_WEBSOCKET_ENDPOINT_IDLE_TIMEOUT, WebsocketEndpointIdleTimeout.getQualifiedName(), value));
+        }
+
+        return idleTimeout;
+    }
+
+    private static int getWebsocketMaxSessionsPerChannel(WebConfiguration webConfig) {
+        String value = webConfig.getOptionValue(WebsocketMaxSessionsPerChannel);
+
+        if (value == null || value.isEmpty()) {
+            return Integer.MAX_VALUE;
+        }
+
+        long maxSessionsPerChannel = toLong(value, 0);
+
+        if (maxSessionsPerChannel < 1 || maxSessionsPerChannel > Integer.MAX_VALUE) {
+            throw new IllegalArgumentException(format(ERROR_INVALID_WEBSOCKET_MAX_SESSIONS_PER_CHANNEL, WebsocketMaxSessionsPerChannel.getQualifiedName(), value));
+        }
+
+        return (int) maxSessionsPerChannel;
+    }
+
+    private static long toLong(String value, long fallback) {
+        if (value == null) {
+            return fallback;
+        }
+
+        try {
+            return Long.parseLong(value.trim());
+        } catch (NumberFormatException e) {
+            return fallback;
+        }
+    }
+
     @Override
     public void contextDestroyed(ServletContextEvent sce) {
         ServletContext context = sce.getServletContext();

impl/src/main/java/com/sun/faces/config/WebConfiguration.java

@@ -785,6 +785,8 @@ public enum WebContextInitParameter {
         WebAppContractsDirectory(ResourceHandler.WEBAPP_CONTRACTS_DIRECTORY_PARAM_NAME, "/contracts"),
         ExceptionTypesToIgnoreInLogging("com.sun.faces.exceptionTypesToIgnoreInLogging", ""),
         CspPolicy(ResourceHandlerImpl.CSP_POLICY_PARAM_NAME, ResourceHandlerImpl.DEFAULT_CSP_POLICY),
+        WebsocketEndpointIdleTimeout("com.sun.faces.websocketEndpointIdleTimeout", "0"), // in milliseconds; 0 means no timeout
+        WebsocketMaxSessionsPerChannel("com.sun.faces.websocketMaxSessionsPerChannel", ""), // empty means unbounded
         ;
 
         private String defaultValue;

impl/src/main/java/com/sun/faces/push/WebsocketChannelManager.java

@@ -31,13 +31,15 @@
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.CopyOnWriteArraySet;
 
 import jakarta.annotation.PreDestroy;
 import jakarta.enterprise.context.SessionScoped;
 import jakarta.faces.context.FacesContext;
 import jakarta.faces.push.Push;
 import jakarta.faces.view.ViewScoped;
 import jakarta.inject.Inject;
+import jakarta.servlet.http.HttpSession;
 
 /**
  * <p class="changed_added_2_3">
@@ -68,6 +70,9 @@ public class WebsocketChannelManager implements Serializable {
     static final int ESTIMATED_TOTAL_CHANNELS = ESTIMATED_CHANNELS_PER_APPLICATION + ESTIMATED_CHANNELS_PER_SESSION + ESTIMATED_CHANNELS_PER_VIEW;
     static final Map<String, String> EMPTY_SCOPE = emptyMap();
 
+    /** The HTTP session attribute name under which the session and view scoped channel identifiers owned by that HTTP session are held. */
+    private static final String SESSION_SCOPE_CHANNEL_IDS = "com.sun.faces.push.SESSION_SCOPE_CHANNEL_IDS";
+
     private enum Scope {
         APPLICATION, SESSION, VIEW;
 
@@ -153,9 +158,30 @@ private String register(FacesContext context, Serializable user, String channel,
         }
 
         socketSessions.register(channelId);
+
+        if (targetScope != APPLICATION_SCOPE) {
+            registerChannelIdInSession(context, channelId); // Session and view scoped channels may only be connected to by the owning HTTP session.
+        }
+
         return url + "?" + channelId;
     }
 
+    private static void registerChannelIdInSession(FacesContext context, String channelId) {
+        HttpSession httpSession = (HttpSession) context.getExternalContext().getSession(true);
+
+        synchronized (httpSession) {
+            @SuppressWarnings("unchecked")
+            Set<String> channelIds = (Set<String>) httpSession.getAttribute(SESSION_SCOPE_CHANNEL_IDS);
+
+            if (channelIds == null) {
+                channelIds = new CopyOnWriteArraySet<>();
+                httpSession.setAttribute(SESSION_SCOPE_CHANNEL_IDS, channelIds);
+            }
+
+            channelIds.add(channelId);
+        }
+    }
+
     /**
      * When current session scope is about to be destroyed, deregister all session scope channels and explicitly close any
      * open web sockets associated with it to avoid stale websockets. If any, also deregister session users.
@@ -233,6 +259,29 @@ static String getChannelId(String channel, Map<String, String> sessionScope, Map
         return channelId;
     }
 
+    /**
+     * For internal usage only. This makes it possible for {@link WebsocketEndpoint.Configurator} to determine whether
+     * the given channel identifier represents an application scoped channel, which is by design not bound to any HTTP
+     * session.
+     */
+    static boolean isApplicationScopedChannelId(String channelId) {
+        return channelId != null && APPLICATION_SCOPE.containsValue(channelId);
+    }
+
+    /**
+     * For internal usage only. This makes it possible for {@link WebsocketEndpoint.Configurator} to determine whether
+     * the given session or view scoped channel identifier was registered by the given HTTP session.
+     */
+    static boolean isChannelIdRegisteredInSession(HttpSession httpSession, String channelId) {
+        if (httpSession == null) {
+            return false;
+        }
+
+        @SuppressWarnings("unchecked")
+        Set<String> channelIds = (Set<String>) httpSession.getAttribute(SESSION_SCOPE_CHANNEL_IDS);
+        return channelIds != null && channelIds.contains(channelId);
+    }
+
     // Serialization --------------------------------------------------------------------------------------------------
 
     private void writeObject(ObjectOutputStream output) throws IOException {

impl/src/main/java/com/sun/faces/push/WebsocketEndpoint.java

@@ -25,10 +25,14 @@
 import java.util.logging.Logger;
 
 import jakarta.faces.push.Push;
+import jakarta.servlet.http.HttpSession;
 import jakarta.websocket.CloseReason;
 import jakarta.websocket.Endpoint;
 import jakarta.websocket.EndpointConfig;
+import jakarta.websocket.HandshakeResponse;
 import jakarta.websocket.Session;
+import jakarta.websocket.server.HandshakeRequest;
+import jakarta.websocket.server.ServerEndpointConfig;
 
 /**
  * <p class="changed_added_2_3">
@@ -48,6 +52,12 @@ public class WebsocketEndpoint extends Endpoint {
     /** The context-relative URI template where the web socket endpoint should listen on. */
     public static final String URI_TEMPLATE = URI_PREFIX + "/{" + PARAM_CHANNEL + "}";
 
+    /** The endpoint config user property name holding the maximum idle timeout in milliseconds as a {@link Long}. */
+    public static final String USER_PROPERTY_IDLE_TIMEOUT = "com.sun.faces.push.IDLE_TIMEOUT";
+
+    /** The endpoint config user property name holding the maximum number of concurrent sessions per channel as an {@link Integer}. */
+    public static final String USER_PROPERTY_MAX_SESSIONS_PER_CHANNEL = "com.sun.faces.push.MAX_SESSIONS_PER_CHANNEL";
+
     private static final Logger logger = Logger.getLogger(WebsocketEndpoint.class.getName());
     private static final CloseReason REASON_UNKNOWN_CHANNEL = new CloseReason(VIOLATED_POLICY, "Unknown channel");
     private static final String ERROR_EXCEPTION = "WebsocketEndpoint: An exception occurred during processing web socket request.";
@@ -56,15 +66,16 @@ public class WebsocketEndpoint extends Endpoint {
 
     /**
      * Add given web socket session to the <code>WebocketSessionManager</code>. If web socket session is not accepted (i.e. the
-     * channel identifier is unknown), then immediately close with reason <code>VIOLATED_POLICY</code> (close code 1008).
+     * channel identifier is unknown or the channel has already reached its maximum number of concurrent sessions), then
+     * immediately close with reason <code>VIOLATED_POLICY</code> (close code 1008).
      *
      * @param session The opened web socket session.
      * @param config The endpoint configuration.
      */
     @Override
     public void onOpen(Session session, EndpointConfig config) {
-        if (WebsocketSessionManager.getInstance().add(session)) { // @Inject in Endpoint doesn't work in Tomcat+Weld/OWB.
-            session.setMaxIdleTimeout(0);
+        if (WebsocketSessionManager.getInstance().add(session, getMaxSessionsPerChannel(config))) { // @Inject in Endpoint doesn't work in Tomcat+Weld/OWB.
+            session.setMaxIdleTimeout(getIdleTimeout(config));
         } else {
             try {
                 session.close(REASON_UNKNOWN_CHANNEL);
@@ -107,4 +118,41 @@ public void onClose(Session session, CloseReason reason) {
         }
     }
 
+    // Helpers --------------------------------------------------------------------------------------------------------
+
+    private static long getIdleTimeout(EndpointConfig config) {
+        Object idleTimeout = config.getUserProperties().get(USER_PROPERTY_IDLE_TIMEOUT);
+        return idleTimeout instanceof Long ? (Long) idleTimeout : 0;
+    }
+
+    private static int getMaxSessionsPerChannel(EndpointConfig config) {
+        Object maxSessionsPerChannel = config.getUserProperties().get(USER_PROPERTY_MAX_SESSIONS_PER_CHANNEL);
+        return maxSessionsPerChannel instanceof Integer ? (Integer) maxSessionsPerChannel : Integer.MAX_VALUE;
+    }
+
+    // Nested classes -------------------------------------------------------------------------------------------------
+
+    /**
+     * This handshake configurator enforces that a session or view scoped channel can only be connected to by the HTTP
+     * session which registered it. Application scoped channels are by design not bound to any HTTP session.
+     *
+     * @author Bauke Scholtz
+     * @see WebsocketChannelManager
+     */
+    public static class Configurator extends ServerEndpointConfig.Configurator {
+
+        private static final String ERROR_UNAUTHORIZED_CHANNEL = "f:websocket channel is not registered in the current HTTP session.";
+
+        @Override
+        public void modifyHandshake(ServerEndpointConfig config, HandshakeRequest request, HandshakeResponse response) {
+            String channelId = request.getQueryString();
+
+            if (!WebsocketChannelManager.isApplicationScopedChannelId(channelId)
+                    && !WebsocketChannelManager.isChannelIdRegisteredInSession((HttpSession) request.getHttpSession(), channelId)) {
+                throw new IllegalStateException(ERROR_UNAUTHORIZED_CHANNEL);
+            }
+        }
+
+    }
+
 }

impl/src/main/java/com/sun/faces/push/WebsocketSessionManager.java

@@ -114,17 +114,19 @@ protected void register(Iterable<String> channelIds) {
 
     /**
      * On open, add given web socket session to the mapping associated with its channel identifier and returns
-     * <code>true</code> if it's accepted (i.e. the channel identifier is known) and the same session hasn't been added
-     * before, otherwise <code>false</code>.
+     * <code>true</code> if it's accepted (i.e. the channel identifier is known, the channel has not yet reached the
+     * given maximum number of concurrent sessions, and the same session hasn't been added before), otherwise
+     * <code>false</code>.
      *
      * @param session The opened web socket session.
+     * @param maxSessionsPerChannel The maximum number of concurrent web socket sessions allowed per channel.
      * @return <code>true</code> if given web socket session is accepted and is new, otherwise <code>false</code>.
      */
-    protected boolean add(Session session) {
+    protected boolean add(Session session, int maxSessionsPerChannel) {
         String channelId = getChannelId(session);
         Collection<Session> sessions = socketSessions.get(channelId);
 
-        if (sessions != null && sessions.add(session)) {
+        if (sessions != null && !isChannelFull(sessions, maxSessionsPerChannel) && sessions.add(session)) {
             Serializable user = socketUsers.getUser(getChannel(session), channelId);
 
             if (user != null) {
@@ -281,6 +283,10 @@ static WebsocketSessionManager getInstance() {
 
     // Helpers --------------------------------------------------------------------------------------------------------
 
+    private static boolean isChannelFull(Collection<Session> sessions, int maxSessionsPerChannel) {
+        return maxSessionsPerChannel != Integer.MAX_VALUE && sessions.size() >= maxSessionsPerChannel; // Size check is skipped when unbounded because it is O(n) on the underlying queue.
+    }
+
     private static String getChannel(Session session) {
         return session.getPathParameters().get(PARAM_CHANNEL);
     }