0004403: Authentication interceptor out of memory

symmetric-core/src/main/java/org/jumpmind/symmetric/common/ParameterConstants.java

@@ -286,6 +286,7 @@ private ParameterConstants() {
     public final static String TRANSPORT_HTTP_COMPRESSION_STRATEGY = "compression.strategy";
     public final static String TRANSPORT_HTTP_USE_SESSION_AUTH = "http.use.session.auth";
     public final static String TRANSPORT_HTTP_SESSION_EXPIRE_SECONDS = "http.session.expire.seconds";
+    public final static String TRANSPORT_HTTP_SESSION_MAX_COUNT = "http.session.max.count";
     public final static String TRANSPORT_HTTP_USE_HEADER_SECURITY_TOKEN = "http.use.header.security.token";
     public final static String TRANSPORT_TYPE = "transport.type";
     public final static String TRANSPORT_MAX_BYTES_TO_SYNC = "transport.max.bytes.to.sync";

symmetric-core/src/main/resources/symmetric-default.properties

@@ -435,6 +435,15 @@ http.use.session.auth=true
 # Type: integer
 http.session.expire.seconds=14400
 
+# Maximum number of authenticated sessions to keep in memory before removing the oldest.
+# Normally, this won't be reached unless something is mis-configured, like a cluster that is
+# not using a sticky session load balancer.
+#
+# DatabaseOverridable: false
+# Tags: transport
+# Type: integer
+http.session.max.count=15000
+
 # When authenticating to a server node, send the security token in the request header instead
 # of using a URL parameter.  Using the request header avoids accidentally logging the
 # security token.  The transport uses the remote node's version to determine if it should use header or parameter.

symmetric-server/src/main/java/org/jumpmind/symmetric/web/AuthenticationInterceptor.java

@@ -21,15 +21,17 @@
 package org.jumpmind.symmetric.web;
 
 import java.io.IOException;
-import java.util.HashMap;
+import java.util.Iterator;
 import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 
 import javax.servlet.ServletException;
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
 
 import org.apache.commons.lang.StringUtils;
 import org.jumpmind.security.ISecurityService;
+import org.jumpmind.symmetric.common.ParameterConstants;
 import org.jumpmind.symmetric.service.INodeService;
 import org.jumpmind.symmetric.service.INodeService.AuthenticationStatus;
 import org.slf4j.Logger;
@@ -46,18 +48,23 @@ public class AuthenticationInterceptor implements IInterceptor {
 
     private ISecurityService securityService;
 
-    private Map<String, AuthenticationSession> sessions = new HashMap<String, AuthenticationSession>();
+    private Map<String, AuthenticationSession> sessions = new ConcurrentHashMap<String, AuthenticationSession>();
 
     private boolean useSessionAuth;
 
     private int sessionExpireMillis;
 
+    private int maxSessions;
+
+    private long maxSessionsLastTime;
+
     public AuthenticationInterceptor(INodeService nodeService, ISecurityService securityService, 
-            boolean useSessionAuth, int sessionExpireSeconds) {
+            boolean useSessionAuth, int sessionExpireSeconds, int maxSessions) {
         this.nodeService = nodeService;
         this.securityService = securityService;
         this.useSessionAuth = useSessionAuth;
         this.sessionExpireMillis = sessionExpireSeconds * 1000;
+        this.maxSessions = maxSessions;
     }
 
     public boolean before(HttpServletRequest req, HttpServletResponse resp) throws IOException, ServletException {
@@ -124,14 +131,43 @@ public boolean before(HttpServletRequest req, HttpServletResponse resp) throws I
 
     protected AuthenticationSession getSession(HttpServletRequest req, boolean create) {
         String sessionId = req.getHeader(WebConstants.HEADER_SESSION_ID);
-        AuthenticationSession session = sessions.get(sessionId);
+        AuthenticationSession session = null;
+        if (sessionId != null) {
+            session = sessions.get(sessionId);
+        }
         if (session == null && create) {
+            if (sessions.size() >= maxSessions) {
+                removeOldSessions();
+            }
             String id = securityService.nextSecureHexString(30);
             session = new AuthenticationSession(id);
             sessions.put(id, session);
         }
         return session;
     }
+
+    protected void removeOldSessions() {
+        long now = System.currentTimeMillis();
+        int removedSessions = 0;
+        AuthenticationSession oldestSession = null;
+        Iterator<AuthenticationSession> iter = sessions.values().iterator();
+        while (iter.hasNext()) {
+            AuthenticationSession session = iter.next();
+            if (now - session.getCreationTime() > sessionExpireMillis) {
+                iter.remove();
+                removedSessions++;
+            } else if (oldestSession == null || session.getCreationTime() < oldestSession.getCreationTime()) {
+                oldestSession = session;
+            }
+        }
+        if (removedSessions == 0 && oldestSession != null) {
+            sessions.remove(oldestSession.getId());
+        }
+        if (maxSessionsLastTime == 0 || now - maxSessionsLastTime > 60000) {
+            maxSessionsLastTime = now;
+            log.warn("Max node authentication sessions reached, removing old sessions. See parameter " + ParameterConstants.TRANSPORT_HTTP_SESSION_MAX_COUNT);
+        }
+    }
 
     public void after(HttpServletRequest req, HttpServletResponse res) throws IOException, ServletException {
     }

symmetric-server/src/main/java/org/jumpmind/symmetric/web/ServerSymmetricEngine.java

@@ -81,7 +81,8 @@ protected void init() {
 
         boolean useSessionAuth = parameterService.is(ParameterConstants.TRANSPORT_HTTP_USE_SESSION_AUTH);
         int sessionExpireSeconds = parameterService.getInt(ParameterConstants.TRANSPORT_HTTP_SESSION_EXPIRE_SECONDS);
-        AuthenticationInterceptor authInterceptor = new AuthenticationInterceptor(nodeService, securityService, useSessionAuth, sessionExpireSeconds);
+        int maxSessions = parameterService.getInt(ParameterConstants.TRANSPORT_HTTP_SESSION_MAX_COUNT, 1000);
+        AuthenticationInterceptor authInterceptor = new AuthenticationInterceptor(nodeService, securityService, useSessionAuth, sessionExpireSeconds, maxSessions);
         NodeConcurrencyInterceptor concurrencyInterceptor = new NodeConcurrencyInterceptor(
                 concurrentConnectionManager, configurationService, statisticManager);
         IInterceptor[] customInterceptors = buildCustomInterceptors();