You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/10/01 19:50:18 UTC
(pulsar) 19/32: [fix][fn] Apply the worker connectors list checks to the deprecated functions connectors endpoint (#26745)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch lh-branch-4.0-backports-2026-10-01
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 1407cb090773c4e44a74ce7bcf76e468ad56ba74
Author: Lari Hotari <lh...@users.noreply.github.com>
AuthorDate: Wed Sep 30 09:53:02 2026 +0300
[fix][fn] Apply the worker connectors list checks to the deprecated functions connectors endpoint (#26745)
(cherry picked from commit c9f0e421288365e0e31ebf53850acf34dd3c145f)
---
.../pulsar/broker/admin/impl/FunctionsBase.java | 2 +-
.../apache/pulsar/broker/admin/v2/Functions.java | 2 +-
.../broker/admin/FunctionsConnectorsListTest.java | 156 +++++++++++++++++++++
.../org/apache/pulsar/client/admin/Functions.java | 9 ++
.../worker/rest/api/v2/FunctionsApiV2Resource.java | 2 +-
.../worker/rest/api/v3/FunctionsApiV3Resource.java | 2 +-
.../rest/FunctionsConnectorsListResourceTest.java | 153 ++++++++++++++++++++
7 files changed, 322 insertions(+), 4 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/FunctionsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/FunctionsBase.java
index 42971ae231c0..034bce8c9539 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/FunctionsBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/FunctionsBase.java
@@ -738,7 +738,7 @@ public class FunctionsBase extends AdminResource {
* Deprecated in favor of moving endpoint to {@link org.apache.pulsar.broker.admin.v2.Worker}
*/
public List<ConnectorDefinition> getConnectorsList() throws IOException {
- return functions().getListOfConnectors();
+ return validateAndGetWorkerService().getWorkers().getListOfConnectors(authParams());
}
@POST
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Functions.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Functions.java
index 69f99a42d068..3b77b52216c0 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Functions.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Functions.java
@@ -324,6 +324,6 @@ public class Functions extends AdminResource {
})
@Path("/connectors")
public List<ConnectorDefinition> getConnectorsList() throws IOException {
- return functions().getListOfConnectors();
+ return validateAndGetWorkerService().getWorkers().getListOfConnectors(authParams());
}
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/FunctionsConnectorsListTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/FunctionsConnectorsListTest.java
new file mode 100644
index 000000000000..a9cd98085c9f
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/FunctionsConnectorsListTest.java
@@ -0,0 +1,156 @@
+/*
+ * 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.admin;
+
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.expectThrows;
+import java.util.List;
+import java.util.Set;
+import java.util.function.Supplier;
+import javax.servlet.ServletContext;
+import javax.servlet.http.HttpServletRequest;
+import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.admin.v3.Functions;
+import org.apache.pulsar.broker.authorization.AuthorizationService;
+import org.apache.pulsar.broker.resources.PulsarResources;
+import org.apache.pulsar.broker.web.AuthenticationFilter;
+import org.apache.pulsar.broker.web.WebService;
+import org.apache.pulsar.common.configuration.PulsarConfigurationLoader;
+import org.apache.pulsar.common.io.ConnectorDefinition;
+import org.apache.pulsar.common.util.RestException;
+import org.apache.pulsar.functions.worker.ConnectorsManager;
+import org.apache.pulsar.functions.worker.PulsarWorkerService;
+import org.apache.pulsar.functions.worker.WorkerConfig;
+import org.apache.pulsar.functions.worker.rest.api.FunctionsImpl;
+import org.apache.pulsar.functions.worker.rest.api.FunctionsImplV2;
+import org.apache.pulsar.functions.worker.rest.api.WorkerImpl;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+/**
+ * Verifies that the broker functions connectors list endpoints apply the same checks as the worker connectors
+ * endpoint.
+ */
+@SuppressWarnings("deprecation")
+@Test(groups = "broker-admin")
+public class FunctionsConnectorsListTest {
+
+ private static final String SUPER_USER = "superuser";
+ private static final String PROXY_ROLE = "proxy";
+ private static final String ORIGINAL_PRINCIPAL_HEADER = "X-Original-Principal";
+
+ private WorkerConfig workerConfig;
+ private PulsarService pulsar;
+ private List<ConnectorDefinition> connectorDefinitions;
+
+ @BeforeMethod(alwaysRun = true)
+ public void setup() throws Exception {
+ workerConfig = new WorkerConfig();
+ workerConfig.setPulsarFunctionsCluster("test");
+ workerConfig.setAuthorizationEnabled(true);
+ workerConfig.setSuperUserRoles(Set.of(SUPER_USER, PROXY_ROLE));
+ workerConfig.setProxyRoles(Set.of(PROXY_ROLE));
+ AuthorizationService authorizationService = new AuthorizationService(
+ PulsarConfigurationLoader.convertFrom(workerConfig), mock(PulsarResources.class));
+
+ ConnectorDefinition source = new ConnectorDefinition();
+ source.setName("test-source");
+ source.setSourceClass("org.example.TestSource");
+ ConnectorDefinition sink = new ConnectorDefinition();
+ sink.setName("test-sink");
+ sink.setSinkClass("org.example.TestSink");
+ connectorDefinitions = List.of(source, sink);
+ ConnectorsManager connectorsManager = mock(ConnectorsManager.class);
+ when(connectorsManager.getConnectorDefinitions()).thenReturn(connectorDefinitions);
+
+ PulsarWorkerService workerService = mock(PulsarWorkerService.class);
+ when(workerService.isInitialized()).thenReturn(true);
+ when(workerService.getWorkerConfig()).thenReturn(workerConfig);
+ when(workerService.getAuthorizationService()).thenReturn(authorizationService);
+ when(workerService.getConnectorsManager()).thenReturn(connectorsManager);
+ Supplier<PulsarWorkerService> supplier = () -> workerService;
+ FunctionsImpl functions = new FunctionsImpl(supplier);
+ doReturn(functions).when(workerService).getFunctions();
+ doReturn(new FunctionsImplV2(functions)).when(workerService).getFunctionsV2();
+ doReturn(new WorkerImpl(supplier)).when(workerService).getWorkers();
+
+ pulsar = mock(PulsarService.class);
+ when(pulsar.getWorkerService()).thenReturn(workerService);
+ }
+
+ @DataProvider(name = "resources")
+ public Object[][] resources() {
+ return new Object[][] {{"v2"}, {"v3"}};
+ }
+
+ private static void init(AdminResource resource, ServletContext servletContext, HttpServletRequest request)
+ throws IllegalAccessException {
+ resource.setServletContext(servletContext);
+ FieldUtils.writeField(resource, "httpRequest", request, true);
+ }
+
+ private List<ConnectorDefinition> getConnectorsList(String version, String role, String originalPrincipal)
+ throws Exception {
+ ServletContext servletContext = mock(ServletContext.class);
+ when(servletContext.getAttribute(WebService.ATTRIBUTE_PULSAR_NAME)).thenReturn(pulsar);
+ HttpServletRequest request = mock(HttpServletRequest.class);
+ when(request.getAttribute(AuthenticationFilter.AuthenticatedRoleAttributeName)).thenReturn(role);
+ when(request.getHeader(ORIGINAL_PRINCIPAL_HEADER)).thenReturn(originalPrincipal);
+ if ("v2".equals(version)) {
+ org.apache.pulsar.broker.admin.v2.Functions resource = new org.apache.pulsar.broker.admin.v2.Functions();
+ init(resource, servletContext, request);
+ return resource.getConnectorsList();
+ } else {
+ Functions resource = new Functions();
+ init(resource, servletContext, request);
+ return resource.getConnectorsList();
+ }
+ }
+
+ private void assertStatus401(String version, String role, String originalPrincipal) {
+ RestException e = expectThrows(RestException.class,
+ () -> getConnectorsList(version, role, originalPrincipal));
+ assertEquals(e.getResponse().getStatus(), 401);
+ }
+
+ @Test(dataProvider = "resources")
+ public void testConnectorsListRequiresSuperUser(String version) throws Exception {
+ assertStatus401(version, null, null);
+ assertStatus401(version, "user", null);
+ assertEquals(getConnectorsList(version, SUPER_USER, null), connectorDefinitions);
+ }
+
+ @Test(dataProvider = "resources")
+ public void testConnectorsListChecksOriginalPrincipal(String version) throws Exception {
+ assertStatus401(version, PROXY_ROLE, "user");
+ assertStatus401(version, PROXY_ROLE, null);
+ assertEquals(getConnectorsList(version, PROXY_ROLE, SUPER_USER), connectorDefinitions);
+ }
+
+ @Test(dataProvider = "resources")
+ public void testConnectorsListWithoutAuthorization(String version) throws Exception {
+ workerConfig.setAuthorizationEnabled(false);
+ assertEquals(getConnectorsList(version, "user", null), connectorDefinitions);
+ }
+}
diff --git a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Functions.java b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Functions.java
index 3932ccf6c918..7dfd817f931a 100644
--- a/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Functions.java
+++ b/pulsar-client-admin-api/src/main/java/org/apache/pulsar/client/admin/Functions.java
@@ -797,6 +797,9 @@ public interface Functions {
* Deprecated in favor of getting sources and sinks for their own APIs.
* <p/>
* Fetches a list of supported Pulsar IO connectors currently running in cluster mode
+ * <p/>
+ * When authorization is enabled, this requires super-user access. Use {@link Sources#getBuiltInSources()} and
+ * {@link Sinks#getBuiltInSinks()} instead.
*
* @throws PulsarAdminException
* Unexpected error
@@ -809,6 +812,9 @@ public interface Functions {
* Deprecated in favor of getting sources and sinks for their own APIs.
* <p/>
* Fetches a list of supported Pulsar IO sources currently running in cluster mode
+ * <p/>
+ * When authorization is enabled, this requires super-user access. Use {@link Sources#getBuiltInSources()} and
+ * {@link Sinks#getBuiltInSinks()} instead.
*
* @throws PulsarAdminException
* Unexpected error
@@ -821,6 +827,9 @@ public interface Functions {
* Deprecated in favor of getting sources and sinks for their own APIs.
* <p/>
* Fetches a list of supported Pulsar IO sinks currently running in cluster mode
+ * <p/>
+ * When authorization is enabled, this requires super-user access. Use {@link Sources#getBuiltInSources()} and
+ * {@link Sinks#getBuiltInSinks()} instead.
*
* @throws PulsarAdminException
* Unexpected error
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/FunctionsApiV2Resource.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/FunctionsApiV2Resource.java
index 0b125756b30a..69cfcb6344ae 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/FunctionsApiV2Resource.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v2/FunctionsApiV2Resource.java
@@ -324,6 +324,6 @@ public class FunctionsApiV2Resource extends FunctionApiResource {
*/
@Deprecated
public List<ConnectorDefinition> getConnectorsList() throws IOException {
- return functions().getListOfConnectors();
+ return get().getWorkers().getListOfConnectors(authParams());
}
}
diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v3/FunctionsApiV3Resource.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v3/FunctionsApiV3Resource.java
index 7bdc86d5fae3..12debf7eb931 100644
--- a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v3/FunctionsApiV3Resource.java
+++ b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/rest/api/v3/FunctionsApiV3Resource.java
@@ -354,7 +354,7 @@ public class FunctionsApiV3Resource extends FunctionApiResource {
*/
@Deprecated
public List<ConnectorDefinition> getConnectorsList() throws IOException {
- return functions().getListOfConnectors();
+ return get().getWorkers().getListOfConnectors(authParams());
}
@POST
diff --git a/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/FunctionsConnectorsListResourceTest.java b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/FunctionsConnectorsListResourceTest.java
new file mode 100644
index 000000000000..8bdcbdefaae7
--- /dev/null
+++ b/pulsar-functions/worker/src/test/java/org/apache/pulsar/functions/worker/rest/FunctionsConnectorsListResourceTest.java
@@ -0,0 +1,153 @@
+/*
+ * 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.functions.worker.rest;
+
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.expectThrows;
+import java.util.List;
+import java.util.Set;
+import java.util.function.Supplier;
+import javax.servlet.ServletContext;
+import javax.servlet.http.HttpServletRequest;
+import org.apache.pulsar.broker.authorization.AuthorizationService;
+import org.apache.pulsar.broker.resources.PulsarResources;
+import org.apache.pulsar.broker.web.AuthenticationFilter;
+import org.apache.pulsar.common.configuration.PulsarConfigurationLoader;
+import org.apache.pulsar.common.io.ConnectorDefinition;
+import org.apache.pulsar.common.util.RestException;
+import org.apache.pulsar.functions.worker.ConnectorsManager;
+import org.apache.pulsar.functions.worker.PulsarWorkerService;
+import org.apache.pulsar.functions.worker.WorkerConfig;
+import org.apache.pulsar.functions.worker.rest.api.FunctionsImpl;
+import org.apache.pulsar.functions.worker.rest.api.FunctionsImplV2;
+import org.apache.pulsar.functions.worker.rest.api.SinksImpl;
+import org.apache.pulsar.functions.worker.rest.api.SourcesImpl;
+import org.apache.pulsar.functions.worker.rest.api.WorkerImpl;
+import org.apache.pulsar.functions.worker.rest.api.v2.FunctionsApiV2Resource;
+import org.apache.pulsar.functions.worker.rest.api.v3.FunctionsApiV3Resource;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+/**
+ * Verifies that the functions connectors list endpoints apply the same checks as the worker connectors endpoint.
+ */
+@SuppressWarnings("deprecation")
+public class FunctionsConnectorsListResourceTest {
+
+ private static final String SUPER_USER = "superuser";
+ private static final String PROXY_ROLE = "proxy";
+
+ private WorkerConfig workerConfig;
+ private PulsarWorkerService workerService;
+ private List<ConnectorDefinition> connectorDefinitions;
+
+ @BeforeMethod
+ public void setup() throws Exception {
+ workerConfig = new WorkerConfig();
+ workerConfig.setPulsarFunctionsCluster("test");
+ workerConfig.setAuthorizationEnabled(true);
+ workerConfig.setSuperUserRoles(Set.of(SUPER_USER, PROXY_ROLE));
+ workerConfig.setProxyRoles(Set.of(PROXY_ROLE));
+ AuthorizationService authorizationService = new AuthorizationService(
+ PulsarConfigurationLoader.convertFrom(workerConfig), mock(PulsarResources.class));
+
+ ConnectorDefinition source = new ConnectorDefinition();
+ source.setName("test-source");
+ source.setSourceClass("org.example.TestSource");
+ ConnectorDefinition sink = new ConnectorDefinition();
+ sink.setName("test-sink");
+ sink.setSinkClass("org.example.TestSink");
+ connectorDefinitions = List.of(source, sink);
+ ConnectorsManager connectorsManager = mock(ConnectorsManager.class);
+ when(connectorsManager.getConnectorDefinitions()).thenReturn(connectorDefinitions);
+
+ workerService = mock(PulsarWorkerService.class);
+ when(workerService.isInitialized()).thenReturn(true);
+ when(workerService.getWorkerConfig()).thenReturn(workerConfig);
+ when(workerService.getAuthorizationService()).thenReturn(authorizationService);
+ when(workerService.getConnectorsManager()).thenReturn(connectorsManager);
+ Supplier<PulsarWorkerService> supplier = () -> workerService;
+ FunctionsImpl functions = new FunctionsImpl(supplier);
+ doReturn(functions).when(workerService).getFunctions();
+ doReturn(new FunctionsImplV2(functions)).when(workerService).getFunctionsV2();
+ doReturn(new WorkerImpl(supplier)).when(workerService).getWorkers();
+ }
+
+ @DataProvider(name = "resources")
+ public Object[][] resources() {
+ return new Object[][] {{"v2"}, {"v3"}};
+ }
+
+ private List<ConnectorDefinition> getConnectorsList(String version, String role, String originalPrincipal)
+ throws Exception {
+ ServletContext servletContext = mock(ServletContext.class);
+ when(servletContext.getAttribute(FunctionApiResource.ATTRIBUTE_FUNCTION_WORKER)).thenReturn(workerService);
+ HttpServletRequest request = mock(HttpServletRequest.class);
+ when(request.getAttribute(AuthenticationFilter.AuthenticatedRoleAttributeName)).thenReturn(role);
+ when(request.getHeader(FunctionApiResource.ORIGINAL_PRINCIPAL_HEADER)).thenReturn(originalPrincipal);
+ if ("v2".equals(version)) {
+ FunctionsApiV2Resource resource = new FunctionsApiV2Resource();
+ resource.servletContext = servletContext;
+ resource.httpRequest = request;
+ return resource.getConnectorsList();
+ } else {
+ FunctionsApiV3Resource resource = new FunctionsApiV3Resource();
+ resource.servletContext = servletContext;
+ resource.httpRequest = request;
+ return resource.getConnectorsList();
+ }
+ }
+
+ private void assertStatus401(String version, String role, String originalPrincipal) {
+ RestException e = expectThrows(RestException.class,
+ () -> getConnectorsList(version, role, originalPrincipal));
+ assertEquals(e.getResponse().getStatus(), 401);
+ }
+
+ @Test(dataProvider = "resources")
+ public void testConnectorsListRequiresSuperUser(String version) throws Exception {
+ assertStatus401(version, null, null);
+ assertStatus401(version, "user", null);
+ assertEquals(getConnectorsList(version, SUPER_USER, null), connectorDefinitions);
+ }
+
+ @Test(dataProvider = "resources")
+ public void testConnectorsListChecksOriginalPrincipal(String version) throws Exception {
+ assertStatus401(version, PROXY_ROLE, "user");
+ assertStatus401(version, PROXY_ROLE, null);
+ assertEquals(getConnectorsList(version, PROXY_ROLE, SUPER_USER), connectorDefinitions);
+ }
+
+ @Test(dataProvider = "resources")
+ public void testConnectorsListWithoutAuthorization(String version) throws Exception {
+ workerConfig.setAuthorizationEnabled(false);
+ assertEquals(getConnectorsList(version, "user", null), connectorDefinitions);
+ }
+
+ @Test
+ public void testBuiltinSourcesAndSinksListsUnchanged() {
+ Supplier<PulsarWorkerService> supplier = () -> workerService;
+ assertEquals(new SourcesImpl(supplier).getSourceList(), List.of(connectorDefinitions.get(0)));
+ assertEquals(new SinksImpl(supplier).getSinkList(), List.of(connectorDefinitions.get(1)));
+ }
+}