diff --git a/impl/src/main/java/com/sun/faces/config/ConfigureListener.java b/impl/src/main/java/com/sun/faces/config/ConfigureListener.java index 15e11b130c..be6edc08c3 100644 --- a/impl/src/main/java/com/sun/faces/config/ConfigureListener.java +++ b/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(); diff --git a/impl/src/main/java/com/sun/faces/config/WebConfiguration.java b/impl/src/main/java/com/sun/faces/config/WebConfiguration.java index d99d8124e9..dad1edc3da 100644 --- a/impl/src/main/java/com/sun/faces/config/WebConfiguration.java +++ b/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; diff --git a/impl/src/main/java/com/sun/faces/push/WebsocketChannelManager.java b/impl/src/main/java/com/sun/faces/push/WebsocketChannelManager.java index 31824730d8..8328c7ec4b 100644 --- a/impl/src/main/java/com/sun/faces/push/WebsocketChannelManager.java +++ b/impl/src/main/java/com/sun/faces/push/WebsocketChannelManager.java @@ -31,6 +31,7 @@ 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; @@ -38,6 +39,7 @@ import jakarta.faces.push.Push; import jakarta.faces.view.ViewScoped; import jakarta.inject.Inject; +import jakarta.servlet.http.HttpSession; /** *

@@ -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 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 channelIds = (Set) 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 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 channelIds = (Set) httpSession.getAttribute(SESSION_SCOPE_CHANNEL_IDS); + return channelIds != null && channelIds.contains(channelId); + } + // Serialization -------------------------------------------------------------------------------------------------- private void writeObject(ObjectOutputStream output) throws IOException { diff --git a/impl/src/main/java/com/sun/faces/push/WebsocketEndpoint.java b/impl/src/main/java/com/sun/faces/push/WebsocketEndpoint.java index 9ce69b05af..21eab508ea 100644 --- a/impl/src/main/java/com/sun/faces/push/WebsocketEndpoint.java +++ b/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; /** *

@@ -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 WebocketSessionManager. If web socket session is not accepted (i.e. the - * channel identifier is unknown), then immediately close with reason VIOLATED_POLICY (close code 1008). + * channel identifier is unknown or the channel has already reached its maximum number of concurrent sessions), then + * immediately close with reason VIOLATED_POLICY (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); + } + } + + } + } diff --git a/impl/src/main/java/com/sun/faces/push/WebsocketSessionManager.java b/impl/src/main/java/com/sun/faces/push/WebsocketSessionManager.java index b56d195a85..932e2da71b 100644 --- a/impl/src/main/java/com/sun/faces/push/WebsocketSessionManager.java +++ b/impl/src/main/java/com/sun/faces/push/WebsocketSessionManager.java @@ -114,17 +114,19 @@ protected void register(Iterable channelIds) { /** * On open, add given web socket session to the mapping associated with its channel identifier and returns - * true if it's accepted (i.e. the channel identifier is known) and the same session hasn't been added - * before, otherwise false. + * true 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 + * false. * * @param session The opened web socket session. + * @param maxSessionsPerChannel The maximum number of concurrent web socket sessions allowed per channel. * @return true if given web socket session is accepted and is new, otherwise false. */ - protected boolean add(Session session) { + protected boolean add(Session session, int maxSessionsPerChannel) { String channelId = getChannelId(session); Collection 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 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); }