[improve][broker] Add strictAuthMethod to require explicit authentication method

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java

@@ -1796,6 +1796,14 @@ The max allowed delay for delayed delivery (in milliseconds). If the broker rece
         doc = "Enable authentication"
     )
     private boolean authenticationEnabled = false;
+
+    @FieldContext(
+        category = CATEGORY_AUTHENTICATION,
+        doc = "Strictly enforce authentication method. If specified, Pulsar will only attempt to authenticate with "
+                + "the provided method. If no method is provided, authentication fails."
+    )
+    private boolean strictAuthMethod = false;
+
     @FieldContext(
         category = CATEGORY_AUTHENTICATION,
         doc = "Authentication provider name list, which is a list of class names"

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationService.java

@@ -47,6 +47,7 @@
 public class AuthenticationService implements Closeable {
     private static final Logger LOG = LoggerFactory.getLogger(AuthenticationService.class);
     private final String anonymousUserRole;
+    private final boolean strictAuthMethod;
 
     private final Map<String, AuthenticationProvider> providers = new LinkedHashMap<>();
 
@@ -57,6 +58,7 @@ public AuthenticationService(ServiceConfiguration conf) throws PulsarServerExcep
     public AuthenticationService(ServiceConfiguration conf, OpenTelemetry openTelemetry)
             throws PulsarServerException {
         anonymousUserRole = conf.getAnonymousUserRole();
+        strictAuthMethod = conf.isStrictAuthMethod();
         if (conf.isAuthenticationEnabled()) {
             try {
                 Map<String, List<AuthenticationProvider>> providerMap = new LinkedHashMap<>();
@@ -138,6 +140,12 @@ public boolean authenticateHttpRequest(HttpServletRequest request, HttpServletRe
                 throw e;
             }
         } else {
+            if (strictAuthMethod) {
+                if (LOG.isDebugEnabled()) {
+                    LOG.debug("No authentication method provided while one was is required");
+                }
+                throw new AuthenticationException("Authentication method missing");
+            }
             for (AuthenticationProvider provider : providers.values()) {
                 try {
                     return provider.authenticateHttpRequest(request, response);

pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/AuthenticationServiceTest.java

@@ -221,6 +221,36 @@ public void testHttpRequestWithMultipleProviders() throws Exception {
                 (AuthenticationDataSource) null)).isEqualTo("role2");
     }
 
+    @Test(timeOut = 10000)
+    public void testStrictAuthMethodEnforcement() throws Exception {
+        ServiceConfiguration config = new ServiceConfiguration();
+        Set<String> providersClassNames = Sets.newHashSet(MockAuthenticationProvider.class.getName());
+        config.setAuthenticationProviders(providersClassNames);
+        config.setAuthenticationEnabled(true);
+        config.setStrictAuthMethod(true);
+        @Cleanup
+        AuthenticationService service = new AuthenticationService(config);
+
+        // Test: Request without auth method header should fail when strictAuthMethod is enabled
+        HttpServletRequest requestWithoutAuthMethod = mock(HttpServletRequest.class);
+        when(requestWithoutAuthMethod.getRemoteAddr()).thenReturn("192.168.1.1");
+        when(requestWithoutAuthMethod.getRemotePort()).thenReturn(8080);
+        // No X-Pulsar-Auth-Method-Name header set
+
+        assertThatThrownBy(() -> service.authenticateHttpRequest(requestWithoutAuthMethod, (HttpServletResponse) null))
+                .isInstanceOf(AuthenticationException.class)
+                .hasMessage("Authentication method missing");
+
+        // Test: Request with auth method header should still succeed
+        HttpServletRequest requestWithAuthMethod = mock(HttpServletRequest.class);
+        when(requestWithAuthMethod.getRemoteAddr()).thenReturn("192.168.1.1");
+        when(requestWithAuthMethod.getRemotePort()).thenReturn(8080);
+        when(requestWithAuthMethod.getHeader("X-Pulsar-Auth-Method-Name")).thenReturn("auth");
+
+        boolean result = service.authenticateHttpRequest(requestWithAuthMethod, (HttpServletResponse) null);
+        assertTrue(result, "Authentication should succeed when auth method is provided");
+    }
+
     public static class MockHttpAuthenticationProvider implements AuthenticationProvider {
         @Override
         public void close() throws IOException {