From 9f3792e61de5eb880d7a720c7b0fac16eeffc996 Mon Sep 17 00:00:00 2001 From: feynmanlin <315157973@qq.com> Date: Tue, 4 Jul 2023 20:00:53 +0800 Subject: [PATCH 1/6] [fix][broker] Fix potential OOM due to client use sasl authentication --- .../AuthenticationProviderSasl.java | 19 ++++-- .../authentication/SaslAuthenticateTest.java | 62 +++++++++++++++++-- .../broker/ServiceConfigurationUtils.java | 14 +++++ 3 files changed, 86 insertions(+), 9 deletions(-) diff --git a/pulsar-broker-auth-sasl/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderSasl.java b/pulsar-broker-auth-sasl/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderSasl.java index 93f7d3f420699..ef888c81aee91 100644 --- a/pulsar-broker-auth-sasl/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderSasl.java +++ b/pulsar-broker-auth-sasl/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderSasl.java @@ -33,6 +33,8 @@ import static org.apache.pulsar.common.sasl.SaslConstants.SASL_STATE_NEGOTIATE; import static org.apache.pulsar.common.sasl.SaslConstants.SASL_STATE_SERVER; import static org.apache.pulsar.common.sasl.SaslConstants.SASL_STATE_SERVER_CHECK_TOKEN; +import com.github.benmanes.caffeine.cache.Cache; +import com.github.benmanes.caffeine.cache.Caffeine; import java.io.IOException; import java.net.SocketAddress; import java.net.URI; @@ -41,7 +43,7 @@ import java.util.Base64; import java.util.HashMap; import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; import java.util.regex.Pattern; import java.util.regex.PatternSyntaxException; import javax.naming.AuthenticationException; @@ -52,6 +54,7 @@ import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; import org.apache.pulsar.broker.ServiceConfiguration; +import org.apache.pulsar.broker.ServiceConfigurationUtils; import org.apache.pulsar.common.api.AuthData; import org.apache.pulsar.common.sasl.JAASCredentialsContainer; import org.apache.pulsar.common.sasl.SaslConstants; @@ -72,6 +75,10 @@ public class AuthenticationProviderSasl implements AuthenticationProvider { private JAASCredentialsContainer jaasCredentialsContainer; private String loginContextName; + private Cache authStates; + + private static final String AUTHENTICATION_SASL_PREFIX = "authentication_sasl_prefix"; + private static final long AUTHENTICATION_SASL_DEFAULT_EXPIRE_MS = 60_000; @Override public void initialize(ServiceConfiguration config) throws IOException { @@ -110,6 +117,10 @@ public void initialize(ServiceConfiguration config) throws IOException { throw new IllegalArgumentException(msg); } this.signer = new SaslRoleTokenSigner(secret); + this.authStates = Caffeine.newBuilder() + .expireAfterWrite( + ServiceConfigurationUtils.getLongPropertyOrDefault(config, AUTHENTICATION_SASL_PREFIX, + AUTHENTICATION_SASL_DEFAULT_EXPIRE_MS), TimeUnit.MILLISECONDS).build(); } @Override @@ -198,8 +209,6 @@ private byte[] readSecretFromUrl(String secretConfUrl) throws IOException { } } - private ConcurrentHashMap authStates = new ConcurrentHashMap<>(); - // return authState if it is in cache. private AuthenticationState getAuthState(HttpServletRequest request) { String id = request.getHeader(SASL_STATE_SERVER); @@ -208,7 +217,7 @@ private AuthenticationState getAuthState(HttpServletRequest request) { } try { - return authStates.get(Long.parseLong(id)); + return authStates.getIfPresent(Long.parseLong(id)); } catch (NumberFormatException e) { log.error("[{}] Wrong Id String in Token {}. e:", request.getRequestURI(), id, e); @@ -295,7 +304,7 @@ public boolean authenticateHttpRequest(HttpServletRequest request, HttpServletRe response.setStatus(HttpServletResponse.SC_OK); // auth completed, no need to keep authState - authStates.remove(state.getStateId()); + authStates.invalidate(state.getStateId()); return false; } else { // auth not complete diff --git a/pulsar-broker-auth-sasl/src/test/java/org/apache/pulsar/broker/authentication/SaslAuthenticateTest.java b/pulsar-broker-auth-sasl/src/test/java/org/apache/pulsar/broker/authentication/SaslAuthenticateTest.java index 8a0d0392d1333..f37d5f59e9619 100644 --- a/pulsar-broker-auth-sasl/src/test/java/org/apache/pulsar/broker/authentication/SaslAuthenticateTest.java +++ b/pulsar-broker-auth-sasl/src/test/java/org/apache/pulsar/broker/authentication/SaslAuthenticateTest.java @@ -18,26 +18,31 @@ */ package org.apache.pulsar.broker.authentication; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mock; +import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertFalse; import static org.testng.Assert.assertNull; import static org.testng.Assert.assertTrue; import static org.testng.Assert.fail; - +import com.github.benmanes.caffeine.cache.Cache; +import com.google.common.collect.ImmutableSet; import java.io.File; import java.io.FileWriter; +import java.lang.reflect.Field; import java.net.URI; import java.nio.file.Files; import java.nio.file.Paths; +import java.util.Base64; import java.util.HashMap; import java.util.HashSet; import java.util.Map; import java.util.Properties; import java.util.Set; import java.util.concurrent.TimeUnit; - import javax.security.auth.login.Configuration; - -import com.google.common.collect.ImmutableSet; +import javax.servlet.http.HttpServletRequest; +import javax.servlet.http.HttpServletResponse; import lombok.extern.slf4j.Slf4j; import org.apache.commons.io.FileUtils; import org.apache.pulsar.client.admin.PulsarAdmin; @@ -59,6 +64,7 @@ import org.testng.annotations.BeforeClass; import org.testng.annotations.BeforeMethod; import org.testng.annotations.Test; +import org.testng.collections.CollectionUtils; @Slf4j public class SaslAuthenticateTest extends ProducerConsumerBase { @@ -295,4 +301,52 @@ public void testSaslServerAndClientAuth() throws Exception { log.info("-- {} -- end", methodName); } + @Test + public void testSaslServerAndClientAuth2() throws Exception { + AuthenticationProviderSasl saslServer = (AuthenticationProviderSasl) pulsar.getBrokerService() + .getAuthenticationService().getAuthenticationProvider(SaslConstants.AUTH_METHOD_NAME); + + HttpServletRequest servletRequest = mock(HttpServletRequest.class); + doReturn("Init").when(servletRequest).getHeader("State"); + // 10 clients only do one-stage verification, resulting in 10 auth info remaining in memory + for (int i = 0; i < 10; i++) { + AuthenticationDataProvider dataProvider = authSasl.getAuthData("localhost"); + AuthData initData1 = dataProvider.authenticate(AuthData.INIT_AUTH_DATA); + doReturn(Base64.getEncoder().encodeToString(initData1.getBytes())).when( + servletRequest).getHeader("SASL-Token"); + doReturn(String.valueOf(i)).when(servletRequest).getHeader("SASL-Server-ID"); + saslServer.authenticateHttpRequest(servletRequest, mock(HttpServletResponse.class)); + } + Field field = AuthenticationProviderSasl.class.getDeclaredField("authStates"); + field.setAccessible(true); + Cache cache = (Cache) field.get(saslServer); + assertEquals(cache.estimatedSize(), 10); + // The cache expiration time is set to 1ms. Residual auth info should be cleaned up + Properties properties = new Properties(); + properties.setProperty("authentication_sasl_prefix", "1"); + conf.setProperties(properties); + saslServer.initialize(conf); + // Add more auth info into memory + for (int i = 0; i < 10; i++) { + AuthenticationDataProvider dataProvider = authSasl.getAuthData("localhost"); + AuthData initData1 = dataProvider.authenticate(AuthData.INIT_AUTH_DATA); + doReturn(Base64.getEncoder().encodeToString(initData1.getBytes())).when( + servletRequest).getHeader("SASL-Token"); + doReturn(String.valueOf(10 + i)).when(servletRequest).getHeader("SASL-Server-ID"); + saslServer.authenticateHttpRequest(servletRequest, mock(HttpServletResponse.class)); + } + long start = System.currentTimeMillis(); + while (true) { + if (System.currentTimeMillis() - start > 10_00) { + fail(); + } + cache = (Cache) field.get(saslServer); + // Residual auth info should be cleaned up + if (CollectionUtils.hasElements(cache.asMap())) { + break; + } + Thread.yield(); + } + } + } diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfigurationUtils.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfigurationUtils.java index b5a0af97f8f10..c27eac89f4506 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfigurationUtils.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfigurationUtils.java @@ -137,4 +137,18 @@ public static String webServiceUrl(String host, int port) { public static String webServiceUrlTls(String host, int port) { return String.format("https://%s:%d", host, port); } + + public static long getLongPropertyOrDefault(ServiceConfiguration config, String key, long defaultValue) { + Object value = config.getProperties().get(key); + if (value instanceof Integer) { + return (Long) value; + } else if (value instanceof String) { + try { + return Long.parseLong((String) value); + } catch (NumberFormatException e) { + return defaultValue; + } + } + return defaultValue; + } } From 44032c027c14afd4a05581b9ad5e08fe0814dbb4 Mon Sep 17 00:00:00 2001 From: feynmanlin <315157973@qq.com> Date: Tue, 4 Jul 2023 20:17:13 +0800 Subject: [PATCH 2/6] [fix][broker] Fix potential OOM due to client use sasl authentication --- .../org/apache/pulsar/broker/ServiceConfigurationUtils.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfigurationUtils.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfigurationUtils.java index c27eac89f4506..9cd9954ecb022 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfigurationUtils.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfigurationUtils.java @@ -140,7 +140,7 @@ public static String webServiceUrlTls(String host, int port) { public static long getLongPropertyOrDefault(ServiceConfiguration config, String key, long defaultValue) { Object value = config.getProperties().get(key); - if (value instanceof Integer) { + if (value instanceof Long) { return (Long) value; } else if (value instanceof String) { try { From 665ad965abc1dae2b54104fdf681adaaa533f744 Mon Sep 17 00:00:00 2001 From: feynmanlin <315157973@qq.com> Date: Tue, 4 Jul 2023 20:19:05 +0800 Subject: [PATCH 3/6] fix test --- .../pulsar/broker/authentication/SaslAuthenticateTest.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pulsar-broker-auth-sasl/src/test/java/org/apache/pulsar/broker/authentication/SaslAuthenticateTest.java b/pulsar-broker-auth-sasl/src/test/java/org/apache/pulsar/broker/authentication/SaslAuthenticateTest.java index f37d5f59e9619..c6043f3d2accd 100644 --- a/pulsar-broker-auth-sasl/src/test/java/org/apache/pulsar/broker/authentication/SaslAuthenticateTest.java +++ b/pulsar-broker-auth-sasl/src/test/java/org/apache/pulsar/broker/authentication/SaslAuthenticateTest.java @@ -302,7 +302,7 @@ public void testSaslServerAndClientAuth() throws Exception { } @Test - public void testSaslServerAndClientAuth2() throws Exception { + public void testSaslOnlyAuthFirstStage() throws Exception { AuthenticationProviderSasl saslServer = (AuthenticationProviderSasl) pulsar.getBrokerService() .getAuthenticationService().getAuthenticationProvider(SaslConstants.AUTH_METHOD_NAME); @@ -320,7 +320,7 @@ public void testSaslServerAndClientAuth2() throws Exception { Field field = AuthenticationProviderSasl.class.getDeclaredField("authStates"); field.setAccessible(true); Cache cache = (Cache) field.get(saslServer); - assertEquals(cache.estimatedSize(), 10); + assertEquals(cache.asMap().size(), 10); // The cache expiration time is set to 1ms. Residual auth info should be cleaned up Properties properties = new Properties(); properties.setProperty("authentication_sasl_prefix", "1"); From e2686caff0130bfd1681691e4915c3eb90e140cf Mon Sep 17 00:00:00 2001 From: feynmanlin <315157973@qq.com> Date: Tue, 4 Jul 2023 20:48:06 +0800 Subject: [PATCH 4/6] Add unit test and address comment --- .../broker/ServiceConfigurationUtils.java | 3 ++ .../broker/ServiceConfigurationUtilsTest.java | 44 +++++++++++++++++++ 2 files changed, 47 insertions(+) create mode 100644 pulsar-broker-common/src/test/java/org/apache/pulsar/broker/ServiceConfigurationUtilsTest.java diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfigurationUtils.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfigurationUtils.java index 9cd9954ecb022..3c1f1e8d178f2 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfigurationUtils.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfigurationUtils.java @@ -140,6 +140,9 @@ public static String webServiceUrlTls(String host, int port) { public static long getLongPropertyOrDefault(ServiceConfiguration config, String key, long defaultValue) { Object value = config.getProperties().get(key); + if (value instanceof Integer) { + return ((Integer) value).longValue(); + } if (value instanceof Long) { return (Long) value; } else if (value instanceof String) { diff --git a/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/ServiceConfigurationUtilsTest.java b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/ServiceConfigurationUtilsTest.java new file mode 100644 index 0000000000000..1a9d1351a2f8d --- /dev/null +++ b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/ServiceConfigurationUtilsTest.java @@ -0,0 +1,44 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.apache.pulsar.broker; + +import static org.testng.Assert.assertEquals; +import java.util.Properties; +import org.testng.annotations.Test; + +public class ServiceConfigurationUtilsTest { + @Test + public void testGetLongPropertyOrDefault() { + ServiceConfiguration configuration = new ServiceConfiguration(); + Properties properties = new Properties(); + properties.put("test", 1); + configuration.setProperties(properties); + long value = ServiceConfigurationUtils.getLongPropertyOrDefault(configuration, "test", 999L); + assertEquals(value, 1L); + properties.put("test", 2L); + value = ServiceConfigurationUtils.getLongPropertyOrDefault(configuration, "test", 999L); + assertEquals(value, 2L); + properties.put("test", "3"); + value = ServiceConfigurationUtils.getLongPropertyOrDefault(configuration, "test", 999L); + assertEquals(value, 3L); + properties.put("test", ""); + value = ServiceConfigurationUtils.getLongPropertyOrDefault(configuration, "test", 999L); + assertEquals(value, 999L); + } +} From 60db04de3a287219d7b526aff8b6544f7029cfc0 Mon Sep 17 00:00:00 2001 From: feynmanlin <315157973@qq.com> Date: Wed, 19 Jul 2023 10:09:11 +0800 Subject: [PATCH 5/6] Add configuration item --- .../AuthenticationProviderSasl.java | 8 +--- .../pulsar/broker/ServiceConfiguration.java | 6 +++ .../broker/ServiceConfigurationUtils.java | 17 ------- .../broker/ServiceConfigurationUtilsTest.java | 44 ------------------- 4 files changed, 7 insertions(+), 68 deletions(-) delete mode 100644 pulsar-broker-common/src/test/java/org/apache/pulsar/broker/ServiceConfigurationUtilsTest.java diff --git a/pulsar-broker-auth-sasl/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderSasl.java b/pulsar-broker-auth-sasl/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderSasl.java index ef888c81aee91..bbec814c5b71f 100644 --- a/pulsar-broker-auth-sasl/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderSasl.java +++ b/pulsar-broker-auth-sasl/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderSasl.java @@ -54,7 +54,6 @@ import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; import org.apache.pulsar.broker.ServiceConfiguration; -import org.apache.pulsar.broker.ServiceConfigurationUtils; import org.apache.pulsar.common.api.AuthData; import org.apache.pulsar.common.sasl.JAASCredentialsContainer; import org.apache.pulsar.common.sasl.SaslConstants; @@ -77,9 +76,6 @@ public class AuthenticationProviderSasl implements AuthenticationProvider { private String loginContextName; private Cache authStates; - private static final String AUTHENTICATION_SASL_PREFIX = "authentication_sasl_prefix"; - private static final long AUTHENTICATION_SASL_DEFAULT_EXPIRE_MS = 60_000; - @Override public void initialize(ServiceConfiguration config) throws IOException { this.configuration = new HashMap<>(); @@ -118,9 +114,7 @@ public void initialize(ServiceConfiguration config) throws IOException { } this.signer = new SaslRoleTokenSigner(secret); this.authStates = Caffeine.newBuilder() - .expireAfterWrite( - ServiceConfigurationUtils.getLongPropertyOrDefault(config, AUTHENTICATION_SASL_PREFIX, - AUTHENTICATION_SASL_DEFAULT_EXPIRE_MS), TimeUnit.MILLISECONDS).build(); + .expireAfterWrite(config.getInflightSaslContextExpiryMs(), TimeUnit.MILLISECONDS).build(); } @Override diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java index b41a562fbd788..d20ed655872be 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java @@ -1654,6 +1654,12 @@ The delayed message index time step(in seconds) in per bucket snapshot segment, ) private String kinitCommand = "/usr/bin/kinit"; + @FieldContext( + category = CATEGORY_SASL_AUTH, + doc = "how often the broker expires the inflight SASL context." + ) + private long inflightSaslContextExpiryMs = 30_000L; + /**** --- BookKeeper Client. --- ****/ @FieldContext( category = CATEGORY_STORAGE_BK, diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfigurationUtils.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfigurationUtils.java index 3c1f1e8d178f2..b5a0af97f8f10 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfigurationUtils.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfigurationUtils.java @@ -137,21 +137,4 @@ public static String webServiceUrl(String host, int port) { public static String webServiceUrlTls(String host, int port) { return String.format("https://%s:%d", host, port); } - - public static long getLongPropertyOrDefault(ServiceConfiguration config, String key, long defaultValue) { - Object value = config.getProperties().get(key); - if (value instanceof Integer) { - return ((Integer) value).longValue(); - } - if (value instanceof Long) { - return (Long) value; - } else if (value instanceof String) { - try { - return Long.parseLong((String) value); - } catch (NumberFormatException e) { - return defaultValue; - } - } - return defaultValue; - } } diff --git a/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/ServiceConfigurationUtilsTest.java b/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/ServiceConfigurationUtilsTest.java deleted file mode 100644 index 1a9d1351a2f8d..0000000000000 --- a/pulsar-broker-common/src/test/java/org/apache/pulsar/broker/ServiceConfigurationUtilsTest.java +++ /dev/null @@ -1,44 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you 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.apache.pulsar.broker; - -import static org.testng.Assert.assertEquals; -import java.util.Properties; -import org.testng.annotations.Test; - -public class ServiceConfigurationUtilsTest { - @Test - public void testGetLongPropertyOrDefault() { - ServiceConfiguration configuration = new ServiceConfiguration(); - Properties properties = new Properties(); - properties.put("test", 1); - configuration.setProperties(properties); - long value = ServiceConfigurationUtils.getLongPropertyOrDefault(configuration, "test", 999L); - assertEquals(value, 1L); - properties.put("test", 2L); - value = ServiceConfigurationUtils.getLongPropertyOrDefault(configuration, "test", 999L); - assertEquals(value, 2L); - properties.put("test", "3"); - value = ServiceConfigurationUtils.getLongPropertyOrDefault(configuration, "test", 999L); - assertEquals(value, 3L); - properties.put("test", ""); - value = ServiceConfigurationUtils.getLongPropertyOrDefault(configuration, "test", 999L); - assertEquals(value, 999L); - } -} From 9da80b8633e3a37646d575a76c47f79db4e4fee3 Mon Sep 17 00:00:00 2001 From: feynmanlin <315157973@qq.com> Date: Thu, 20 Jul 2023 10:17:16 +0800 Subject: [PATCH 6/6] Add max context limit --- .../AuthenticationProviderSasl.java | 1 + .../authentication/SaslAuthenticateTest.java | 29 +++++++++++++++++-- .../pulsar/broker/ServiceConfiguration.java | 6 ++++ 3 files changed, 33 insertions(+), 3 deletions(-) diff --git a/pulsar-broker-auth-sasl/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderSasl.java b/pulsar-broker-auth-sasl/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderSasl.java index bbec814c5b71f..ed2d05ab7227f 100644 --- a/pulsar-broker-auth-sasl/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderSasl.java +++ b/pulsar-broker-auth-sasl/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationProviderSasl.java @@ -114,6 +114,7 @@ public void initialize(ServiceConfiguration config) throws IOException { } this.signer = new SaslRoleTokenSigner(secret); this.authStates = Caffeine.newBuilder() + .maximumSize(config.getMaxInflightSaslContext()) .expireAfterWrite(config.getInflightSaslContextExpiryMs(), TimeUnit.MILLISECONDS).build(); } diff --git a/pulsar-broker-auth-sasl/src/test/java/org/apache/pulsar/broker/authentication/SaslAuthenticateTest.java b/pulsar-broker-auth-sasl/src/test/java/org/apache/pulsar/broker/authentication/SaslAuthenticateTest.java index c6043f3d2accd..5cace2221dea8 100644 --- a/pulsar-broker-auth-sasl/src/test/java/org/apache/pulsar/broker/authentication/SaslAuthenticateTest.java +++ b/pulsar-broker-auth-sasl/src/test/java/org/apache/pulsar/broker/authentication/SaslAuthenticateTest.java @@ -322,9 +322,7 @@ public void testSaslOnlyAuthFirstStage() throws Exception { Cache cache = (Cache) field.get(saslServer); assertEquals(cache.asMap().size(), 10); // The cache expiration time is set to 1ms. Residual auth info should be cleaned up - Properties properties = new Properties(); - properties.setProperty("authentication_sasl_prefix", "1"); - conf.setProperties(properties); + conf.setInflightSaslContextExpiryMs(1); saslServer.initialize(conf); // Add more auth info into memory for (int i = 0; i < 10; i++) { @@ -349,4 +347,29 @@ public void testSaslOnlyAuthFirstStage() throws Exception { } } + @Test + public void testMaxInflightContext() throws Exception { + AuthenticationProviderSasl saslServer = (AuthenticationProviderSasl) pulsar.getBrokerService() + .getAuthenticationService().getAuthenticationProvider(SaslConstants.AUTH_METHOD_NAME); + HttpServletRequest servletRequest = mock(HttpServletRequest.class); + doReturn("Init").when(servletRequest).getHeader("State"); + conf.setInflightSaslContextExpiryMs(Integer.MAX_VALUE); + conf.setMaxInflightSaslContext(1); + saslServer.initialize(conf); + // add 10 inflight sasl context + for (int i = 0; i < 10; i++) { + AuthenticationDataProvider dataProvider = authSasl.getAuthData("localhost"); + AuthData initData1 = dataProvider.authenticate(AuthData.INIT_AUTH_DATA); + doReturn(Base64.getEncoder().encodeToString(initData1.getBytes())).when( + servletRequest).getHeader("SASL-Token"); + doReturn(String.valueOf(i)).when(servletRequest).getHeader("SASL-Server-ID"); + saslServer.authenticateHttpRequest(servletRequest, mock(HttpServletResponse.class)); + } + Field field = AuthenticationProviderSasl.class.getDeclaredField("authStates"); + field.setAccessible(true); + Cache cache = (Cache) field.get(saslServer); + //only 1 context was left in the memory + assertEquals(cache.asMap().size(), 1); + } + } diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java index d20ed655872be..a83c3b77bbf37 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java @@ -1660,6 +1660,12 @@ The delayed message index time step(in seconds) in per bucket snapshot segment, ) private long inflightSaslContextExpiryMs = 30_000L; + @FieldContext( + category = CATEGORY_SASL_AUTH, + doc = "Maximum number of inflight sasl context." + ) + private long maxInflightSaslContext = 50_000L; + /**** --- BookKeeper Client. --- ****/ @FieldContext( category = CATEGORY_STORAGE_BK,