You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampipes.apache.org by ri...@apache.org on 2026/07/07 20:14:05 UTC
(streampipes) branch improve-runtime-configuration-handling created (now 367cc335f7)
This is an automated email from the ASF dual-hosted git repository.
dominikriemer pushed a change to branch improve-runtime-configuration-handling
in repository https://gitbox.apache.org/repos/asf/streampipes.git
at 367cc335f7 fix: Improve handling of runtime-resolved configurations
This branch includes the following new commits:
new 367cc335f7 fix: Improve handling of runtime-resolved configurations
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(streampipes) 01/01: fix: Improve handling of runtime-resolved configurations
Posted by ri...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dominikriemer pushed a commit to branch improve-runtime-configuration-handling
in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit 367cc335f74f37fb4ff5d1fd6f6087b353e74a7a
Author: Dominik Riemer <do...@gmail.com>
AuthorDate: Tue Jul 7 22:13:47 2026 +0200
fix: Improve handling of runtime-resolved configurations
---
.../execution/http/InvokeExtensionRequest.java | 37 +++++++++--
.../execution/task/StorePipelineStatusTask.java | 3 +
.../CustomTransformOutputSchemaGenerator.java | 10 ++-
.../pipeline/PipelineElementUserCleaner.java | 36 +++++++++++
.../manager/pipeline/PipelineManager.java | 4 --
.../pipeline/update/PipelineUpdateCoordinator.java | 2 +
.../remote/ContainerProvidedOptionsHandler.java | 3 +
.../manager/storage/PipelineStorageService.java | 2 +
.../execution/http/InvokeExtensionRequestTest.java | 63 +++++++++++++++++++
.../CustomTransformOutputSchemaGeneratorTest.java | 73 ++++++++++++++++++++++
.../pipeline/PipelineElementUserCleanerTest.java | 48 ++++++++++++++
.../resource/management/secret/SecretVisitor.java | 2 +-
.../management/secret/SecretServiceTest.java | 64 +++++++++++++++++++
.../core/filter/TokenAuthenticationFilter.java | 9 ++-
.../core/filter/TokenAuthenticationFilterTest.java | 53 ++++++++++++++++
15 files changed, 396 insertions(+), 13 deletions(-)
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/InvokeExtensionRequest.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/InvokeExtensionRequest.java
index d77a44326c..a67c878153 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/InvokeExtensionRequest.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/InvokeExtensionRequest.java
@@ -24,8 +24,10 @@ import org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestTarg
import org.apache.streampipes.manager.api.extensions.ExtensionServiceRequests;
import org.apache.streampipes.manager.execution.endpoint.ExtensionsServiceEndpointUtils;
import org.apache.streampipes.manager.util.AuthTokenProvider;
-import org.apache.streampipes.model.api.EndpointSelectable;
import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
+import org.apache.streampipes.model.client.user.Permission;
+import org.apache.streampipes.model.graph.DataProcessorInvocation;
+import org.apache.streampipes.model.graph.DataSinkInvocation;
import org.apache.streampipes.resource.management.SpResourceManager;
import org.apache.streampipes.serializers.json.JacksonSerializer;
@@ -58,7 +60,7 @@ public class InvokeExtensionRequest extends PipelineElementExtensionRequest {
var authToken = new AuthTokenProvider(resourceManager).getAuthToken(pipelineId);
return requestManager().request(
ExtensionServiceRequests
- .pipelineElementInvocation(requestTarget, toJson(pipelineElement), authToken)
+ .pipelineElementInvocation(requestTarget, toJson(pipelineElement, pipelineId), authToken)
);
}
@@ -70,7 +72,34 @@ public class InvokeExtensionRequest extends PipelineElementExtensionRequest {
endpointUrl, pipelineElementName, exceptionMessage);
}
- private String toJson(EndpointSelectable pipelineElement) throws JsonProcessingException {
- return JacksonSerializer.getObjectMapper().writeValueAsString(pipelineElement);
+ String toJson(InvocableStreamPipesEntity pipelineElement,
+ String pipelineId) throws JsonProcessingException {
+ var invocation = makeInvocationPayload(pipelineElement);
+ if (pipelineId != null) {
+ invocation.setCorrespondingUser(getPipelineOwnerSid(pipelineId));
+ }
+ return JacksonSerializer.getObjectMapper().writeValueAsString(invocation);
+ }
+
+ private InvocableStreamPipesEntity makeInvocationPayload(InvocableStreamPipesEntity pipelineElement) {
+ InvocableStreamPipesEntity invocation;
+ if (pipelineElement instanceof DataProcessorInvocation processorInvocation) {
+ invocation = new DataProcessorInvocation(processorInvocation);
+ } else if (pipelineElement instanceof DataSinkInvocation sinkInvocation) {
+ invocation = new DataSinkInvocation(sinkInvocation);
+ } else {
+ throw new IllegalArgumentException("Unsupported pipeline element type: "
+ + pipelineElement.getClass().getCanonicalName());
+ }
+ invocation.setSelectedServiceId(pipelineElement.getSelectedServiceId());
+ return invocation;
+ }
+
+ private String getPipelineOwnerSid(String pipelineId) {
+ return resourceManager.managePermissions().findForObjectId(pipelineId)
+ .stream()
+ .findFirst()
+ .map(Permission::getOwnerSid)
+ .orElseThrow(() -> new IllegalArgumentException("Could not find owner for pipeline " + pipelineId));
}
}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/task/StorePipelineStatusTask.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/task/StorePipelineStatusTask.java
index 0c4916f05a..941a2e37e1 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/task/StorePipelineStatusTask.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/task/StorePipelineStatusTask.java
@@ -20,6 +20,7 @@ package org.apache.streampipes.manager.execution.task;
import org.apache.streampipes.commons.prometheus.pipelines.PipelinesStats;
import org.apache.streampipes.manager.execution.PipelineExecutionInfo;
+import org.apache.streampipes.manager.pipeline.PipelineElementUserCleaner;
import org.apache.streampipes.model.pipeline.Pipeline;
import org.apache.streampipes.model.pipeline.PipelineHealthStatus;
import org.apache.streampipes.resource.management.SpResourceManager;
@@ -69,6 +70,7 @@ public class StorePipelineStatusTask implements PipelineExecutionTask {
private void setPipelineStarted(Pipeline pipeline) {
pipeline.setRunning(true);
pipeline.setStartedAt(new Date().getTime());
+ PipelineElementUserCleaner.clearCorrespondingUsers(pipeline);
pipelinesStats.updatePipelineRunningState(pipeline.getElementId(),pipeline.getName()
, true);
pipelinesStats.updatePipelineHealthState(pipeline.getElementId(),pipeline.getName(), pipeline.getHealthStatus().toString());
@@ -81,6 +83,7 @@ public class StorePipelineStatusTask implements PipelineExecutionTask {
private void setPipelineStopped(Pipeline pipeline) {
pipeline.setRunning(false);
+ PipelineElementUserCleaner.clearCorrespondingUsers(pipeline);
pipelinesStats.updatePipelineRunningState(pipeline.getElementId(),pipeline.getName()
, false);
pipelinesStats.updatePipelineHealthState(
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/output/CustomTransformOutputSchemaGenerator.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/output/CustomTransformOutputSchemaGenerator.java
index 0386095f22..70e1e35d76 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/output/CustomTransformOutputSchemaGenerator.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/output/CustomTransformOutputSchemaGenerator.java
@@ -27,6 +27,8 @@ import org.apache.streampipes.model.graph.DataProcessorInvocation;
import org.apache.streampipes.model.output.CustomTransformOutputStrategy;
import org.apache.streampipes.model.output.OutputStrategy;
import org.apache.streampipes.model.schema.EventSchema;
+import org.apache.streampipes.resource.management.secret.SecretDecrypter;
+import org.apache.streampipes.resource.management.secret.SecretService;
import org.apache.streampipes.sdk.helpers.Tuple2;
import org.apache.streampipes.serializers.json.JacksonSerializer;
import org.apache.streampipes.svcdiscovery.api.model.SpServiceUrlProvider;
@@ -72,7 +74,7 @@ public class CustomTransformOutputSchemaGenerator extends OutputSchemaGenerator<
private EventSchema makeRequest() {
try {
- String httpRequestBody = JacksonSerializer.getObjectMapper().writeValueAsString(dataProcessorInvocation);
+ String httpRequestBody = makeRequestBody();
var service = new ExtensionsServiceEndpointGenerator().selectService(
dataProcessorInvocation.getAppId(),
SpServiceUrlProvider.DATA_PROCESSOR,
@@ -91,6 +93,12 @@ public class CustomTransformOutputSchemaGenerator extends OutputSchemaGenerator<
}
}
+ String makeRequestBody() throws IOException {
+ var requestInvocation = new DataProcessorInvocation(dataProcessorInvocation);
+ new SecretService(new SecretDecrypter()).apply(requestInvocation);
+ return JacksonSerializer.getObjectMapper().writeValueAsString(requestInvocation);
+ }
+
private EventSchema handleResponse(ExtensionServiceOperationResult response) throws JsonSyntaxException, IOException {
if (!response.isSuccess()) {
throw new IOException("Could not compute output schema, status code: " + response.statusCode());
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/PipelineElementUserCleaner.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/PipelineElementUserCleaner.java
new file mode 100644
index 0000000000..9faaec7244
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/PipelineElementUserCleaner.java
@@ -0,0 +1,36 @@
+/*
+ * 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.streampipes.manager.pipeline;
+
+import org.apache.streampipes.model.pipeline.Pipeline;
+
+import java.util.Optional;
+
+public final class PipelineElementUserCleaner {
+
+ private PipelineElementUserCleaner() {
+ }
+
+ public static void clearCorrespondingUsers(Pipeline pipeline) {
+ Optional.ofNullable(pipeline.getSepas())
+ .ifPresent(processors -> processors.forEach(processor -> processor.setCorrespondingUser(null)));
+ Optional.ofNullable(pipeline.getActions())
+ .ifPresent(actions -> actions.forEach(action -> action.setCorrespondingUser(null)));
+ }
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/PipelineManager.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/PipelineManager.java
index 58104134d0..c1615688f6 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/PipelineManager.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/PipelineManager.java
@@ -190,9 +190,5 @@ public class PipelineManager {
pipeline.setHealthStatus(PipelineHealthStatus.OK);
pipeline.setCreatedByUser(username);
pipeline.setCreatedAt(new Date().getTime());
- pipeline.getSepas()
- .forEach(processor -> processor.setCorrespondingUser(username));
- pipeline.getActions()
- .forEach(action -> action.setCorrespondingUser(username));
}
}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/PipelineUpdateCoordinator.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/PipelineUpdateCoordinator.java
index dd4b162d60..e89d9476eb 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/PipelineUpdateCoordinator.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/update/PipelineUpdateCoordinator.java
@@ -23,6 +23,7 @@ import org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestMana
import org.apache.streampipes.manager.execution.PipelineExecutor;
import org.apache.streampipes.manager.matching.PipelineVerificationHandlerV2;
import org.apache.streampipes.manager.matching.v2.pipeline.MeasurementChangeValidationStep;
+import org.apache.streampipes.manager.pipeline.PipelineElementUserCleaner;
import org.apache.streampipes.manager.pipeline.PipelineManager;
import org.apache.streampipes.model.SpDataStream;
import org.apache.streampipes.model.base.NamedStreamPipesEntity;
@@ -138,6 +139,7 @@ public class PipelineUpdateCoordinator {
modifiedPipeline.setValid(false);
}
+ PipelineElementUserCleaner.clearCorrespondingUsers(modifiedPipeline);
pipelineStorage.updateElement(modifiedPipeline);
if (shouldRestartPipeline && canAutoMigrate) {
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/remote/ContainerProvidedOptionsHandler.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/remote/ContainerProvidedOptionsHandler.java
index 1b54cfb76a..361b4d72d1 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/remote/ContainerProvidedOptionsHandler.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/remote/ContainerProvidedOptionsHandler.java
@@ -30,6 +30,8 @@ import org.apache.streampipes.manager.util.AuthTokenProvider;
import org.apache.streampipes.model.runtime.RuntimeOptionsRequest;
import org.apache.streampipes.model.runtime.RuntimeOptionsResponse;
import org.apache.streampipes.resource.management.SpResourceManager;
+import org.apache.streampipes.resource.management.secret.SecretDecrypter;
+import org.apache.streampipes.resource.management.secret.SecretService;
import org.apache.streampipes.serializers.json.JacksonSerializer;
import org.apache.streampipes.svcdiscovery.api.model.SpServiceUrlProvider;
@@ -54,6 +56,7 @@ public class ContainerProvidedOptionsHandler {
throws SpConfigurationException, SpRuntimeException {
try {
+ new SecretService(new SecretDecrypter()).applyConfig(request.getStaticProperties());
var payload = JacksonSerializer.getObjectMapper().writeValueAsString(request);
var requestTarget = getEndpointRequestTarget(request.getAppId());
var authToken = new AuthTokenProvider(resourceManager).getAuthTokenForCurrentUser();
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/storage/PipelineStorageService.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/storage/PipelineStorageService.java
index 8858a81c2a..e1cc8444ea 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/storage/PipelineStorageService.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/storage/PipelineStorageService.java
@@ -20,6 +20,7 @@ package org.apache.streampipes.manager.storage;
import org.apache.streampipes.manager.data.PipelineGraph;
import org.apache.streampipes.manager.data.PipelineGraphBuilder;
+import org.apache.streampipes.manager.pipeline.PipelineElementUserCleaner;
import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
import org.apache.streampipes.model.graph.DataProcessorInvocation;
import org.apache.streampipes.model.graph.DataSinkInvocation;
@@ -64,6 +65,7 @@ public class PipelineStorageService {
List<DataProcessorInvocation> sepas = filter(graphs, DataProcessorInvocation.class);
pipeline.setSepas(sepas);
pipeline.setActions(secs);
+ PipelineElementUserCleaner.clearCorrespondingUsers(pipeline);
}
private void encryptSecrets(List<InvocableStreamPipesEntity> graphs) {
diff --git a/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/execution/http/InvokeExtensionRequestTest.java b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/execution/http/InvokeExtensionRequestTest.java
new file mode 100644
index 0000000000..59c4ca52ae
--- /dev/null
+++ b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/execution/http/InvokeExtensionRequestTest.java
@@ -0,0 +1,63 @@
+/*
+ * 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.streampipes.manager.execution.http;
+
+import org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestManager;
+import org.apache.streampipes.model.client.user.Permission;
+import org.apache.streampipes.model.graph.DataSinkInvocation;
+import org.apache.streampipes.resource.management.PermissionResourceManager;
+import org.apache.streampipes.resource.management.SpResourceManager;
+import org.apache.streampipes.serializers.json.JacksonSerializer;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class InvokeExtensionRequestTest {
+
+ @Test
+ void toJsonUsesPipelineOwnerAsCorrespondingUserWithoutMutatingOriginal() throws Exception {
+ var pipelineId = "pipeline-1";
+ var ownerSid = "owner-sid";
+ var storedUser = "stale-user";
+ var resourceManager = mock(SpResourceManager.class);
+ var permissionResourceManager = mock(PermissionResourceManager.class);
+ var permission = new Permission();
+ permission.setOwnerSid(ownerSid);
+ when(resourceManager.managePermissions()).thenReturn(permissionResourceManager);
+ when(permissionResourceManager.findForObjectId(pipelineId)).thenReturn(List.of(permission));
+
+ var sinkInvocation = new DataSinkInvocation();
+ sinkInvocation.setCorrespondingUser(storedUser);
+ sinkInvocation.setSelectedServiceId("service-1");
+
+ var request = new InvokeExtensionRequest(mock(ExtensionServiceRequestManager.class), resourceManager);
+
+ var json = request.toJson(sinkInvocation, pipelineId);
+
+ var payload = JacksonSerializer.getObjectMapper().readValue(json, DataSinkInvocation.class);
+ assertEquals(ownerSid, payload.getCorrespondingUser());
+ assertEquals("service-1", payload.getSelectedServiceId());
+ assertEquals(storedUser, sinkInvocation.getCorrespondingUser());
+ }
+}
diff --git a/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/matching/output/CustomTransformOutputSchemaGeneratorTest.java b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/matching/output/CustomTransformOutputSchemaGeneratorTest.java
new file mode 100644
index 0000000000..4e33b8d7c8
--- /dev/null
+++ b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/matching/output/CustomTransformOutputSchemaGeneratorTest.java
@@ -0,0 +1,73 @@
+/*
+ * 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.streampipes.manager.matching.output;
+
+import org.apache.streampipes.manager.api.extensions.ExtensionServiceRequestManager;
+import org.apache.streampipes.model.graph.DataProcessorInvocation;
+import org.apache.streampipes.model.output.CustomTransformOutputStrategy;
+import org.apache.streampipes.model.staticproperty.SecretStaticProperty;
+import org.apache.streampipes.serializers.json.JacksonSerializer;
+import org.apache.streampipes.user.management.encryption.SecretEncryptionManager;
+
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+
+class CustomTransformOutputSchemaGeneratorTest {
+
+ @Test
+ void makeRequestBodyDecryptsSecretWithoutMutatingInvocation() throws IOException {
+ var plainValue = "my-secret-value";
+ var encryptedValue = SecretEncryptionManager.encrypt(plainValue);
+ var secretProperty = makeSecretProperty(encryptedValue, true);
+ var invocation = new DataProcessorInvocation();
+ invocation.setStaticProperties(List.of(secretProperty));
+
+ var generator = new CustomTransformOutputSchemaGenerator(
+ new CustomTransformOutputStrategy(),
+ invocation,
+ mock(ExtensionServiceRequestManager.class)
+ );
+
+ var requestBody = generator.makeRequestBody();
+
+ var requestInvocation = JacksonSerializer.getObjectMapper()
+ .readValue(requestBody, DataProcessorInvocation.class);
+ var requestSecret = (SecretStaticProperty) requestInvocation.getStaticProperties().get(0);
+ assertEquals(plainValue, requestSecret.getValue());
+ assertFalse(requestSecret.getEncrypted());
+
+ assertEquals(encryptedValue, secretProperty.getValue());
+ assertTrue(secretProperty.getEncrypted());
+ }
+
+ private SecretStaticProperty makeSecretProperty(String value,
+ boolean encrypted) {
+ var secretProperty = new SecretStaticProperty("secret", "Secret", "");
+ secretProperty.setValue(value);
+ secretProperty.setEncrypted(encrypted);
+ return secretProperty;
+ }
+}
diff --git a/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/pipeline/PipelineElementUserCleanerTest.java b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/pipeline/PipelineElementUserCleanerTest.java
new file mode 100644
index 0000000000..5e57c6874d
--- /dev/null
+++ b/streampipes-pipeline-management/src/test/java/org/apache/streampipes/manager/pipeline/PipelineElementUserCleanerTest.java
@@ -0,0 +1,48 @@
+/*
+ * 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.streampipes.manager.pipeline;
+
+import org.apache.streampipes.model.graph.DataProcessorInvocation;
+import org.apache.streampipes.model.graph.DataSinkInvocation;
+import org.apache.streampipes.model.pipeline.Pipeline;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+class PipelineElementUserCleanerTest {
+
+ @Test
+ void clearCorrespondingUsersRemovesUsersFromProcessorsAndSinks() {
+ var processor = new DataProcessorInvocation();
+ processor.setCorrespondingUser("user-1");
+ var sink = new DataSinkInvocation();
+ sink.setCorrespondingUser("user-1");
+ var pipeline = new Pipeline();
+ pipeline.setSepas(List.of(processor));
+ pipeline.setActions(List.of(sink));
+
+ PipelineElementUserCleaner.clearCorrespondingUsers(pipeline);
+
+ assertNull(processor.getCorrespondingUser());
+ assertNull(sink.getCorrespondingUser());
+ }
+}
diff --git a/streampipes-resource-management/src/main/java/org/apache/streampipes/resource/management/secret/SecretVisitor.java b/streampipes-resource-management/src/main/java/org/apache/streampipes/resource/management/secret/SecretVisitor.java
index 04cd621b93..92ff003fc7 100644
--- a/streampipes-resource-management/src/main/java/org/apache/streampipes/resource/management/secret/SecretVisitor.java
+++ b/streampipes-resource-management/src/main/java/org/apache/streampipes/resource/management/secret/SecretVisitor.java
@@ -134,6 +134,6 @@ public class SecretVisitor implements StaticPropertyVisitor {
@Override
public void visit(RuntimeResolvableGroupStaticProperty groupStaticProperty) {
- // Do nothing
+ visit((StaticPropertyGroup) groupStaticProperty);
}
}
diff --git a/streampipes-resource-management/src/test/java/org/apache/streampipes/resource/management/secret/SecretServiceTest.java b/streampipes-resource-management/src/test/java/org/apache/streampipes/resource/management/secret/SecretServiceTest.java
new file mode 100644
index 0000000000..f444a944b6
--- /dev/null
+++ b/streampipes-resource-management/src/test/java/org/apache/streampipes/resource/management/secret/SecretServiceTest.java
@@ -0,0 +1,64 @@
+/*
+ * 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.streampipes.resource.management.secret;
+
+import org.apache.streampipes.model.staticproperty.RuntimeResolvableGroupStaticProperty;
+import org.apache.streampipes.model.staticproperty.SecretStaticProperty;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+
+class SecretServiceTest {
+
+ @Test
+ void applyConfigTraversesRuntimeResolvableGroups() {
+ var secretProperty = makeSecretProperty();
+ var groupProperty = new RuntimeResolvableGroupStaticProperty("group", "Group", "", List.of());
+ groupProperty.setStaticProperties(List.of(secretProperty));
+
+ new SecretService(new TestSecretHandler()).applyConfig(List.of(groupProperty));
+
+ assertEquals("decrypted-value", secretProperty.getValue());
+ assertFalse(secretProperty.getEncrypted());
+ }
+
+ private SecretStaticProperty makeSecretProperty() {
+ var secretProperty = new SecretStaticProperty("secret", "Secret", "");
+ secretProperty.setValue("encrypted-value");
+ secretProperty.setEncrypted(true);
+ return secretProperty;
+ }
+
+ private static class TestSecretHandler implements ISecretHandler {
+
+ @Override
+ public String apply(String extractedValue) {
+ return "decrypted-value";
+ }
+
+ @Override
+ public boolean shouldApply(boolean encrypted) {
+ return encrypted;
+ }
+ }
+}
diff --git a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/filter/TokenAuthenticationFilter.java b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/filter/TokenAuthenticationFilter.java
index b486832594..4cec18cb10 100644
--- a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/filter/TokenAuthenticationFilter.java
+++ b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/filter/TokenAuthenticationFilter.java
@@ -38,6 +38,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.HttpHeaders;
import org.springframework.security.authentication.UsernamePasswordAuthenticationToken;
+import org.springframework.security.core.GrantedAuthority;
import org.springframework.security.core.context.SecurityContext;
import org.springframework.security.core.context.SecurityContextHolder;
import org.springframework.security.web.authentication.WebAuthenticationDetailsSource;
@@ -54,6 +55,7 @@ import java.io.IOException;
import java.security.NoSuchAlgorithmException;
import java.security.spec.InvalidKeySpecException;
import java.util.Base64;
+import java.util.Collection;
import java.util.List;
import java.util.Objects;
@@ -149,7 +151,7 @@ public class TokenAuthenticationFilter extends OncePerRequestFilter {
Principal user = userStorage.getUser(username);
PrincipalUserDetails<?> userDetails = makeDetails(user);
var onBehalfOfHeader = request.getHeader(HttpConstants.X_ON_BEHALF_OF);
- if (isAdminUser(userDetails) && onBehalfOfHeader != null) {
+ if (canActOnBehalfOf(userDetails.getAuthorities()) && onBehalfOfHeader != null) {
var onBehalfOf = userStorage.getUserById(onBehalfOfHeader);
if (onBehalfOf != null) {
userDetails = makeDetails(onBehalfOf);
@@ -177,10 +179,11 @@ public class TokenAuthenticationFilter extends OncePerRequestFilter {
);
}
- private boolean isAdminUser(PrincipalUserDetails<?> userDetails) {
- return userDetails.getAuthorities().stream()
+ static boolean canActOnBehalfOf(Collection<? extends GrantedAuthority> authorities) {
+ return authorities.stream()
.anyMatch(a ->
Objects.equals(a.getAuthority(), DefaultRole.Constants.ROLE_ADMIN_VALUE)
+ || Objects.equals(a.getAuthority(), DefaultRole.Constants.ROLE_SERVICE_ADMIN_VALUE)
);
}
diff --git a/streampipes-service-core/src/test/java/org/apache/streampipes/service/core/filter/TokenAuthenticationFilterTest.java b/streampipes-service-core/src/test/java/org/apache/streampipes/service/core/filter/TokenAuthenticationFilterTest.java
new file mode 100644
index 0000000000..f46d88d72a
--- /dev/null
+++ b/streampipes-service-core/src/test/java/org/apache/streampipes/service/core/filter/TokenAuthenticationFilterTest.java
@@ -0,0 +1,53 @@
+/*
+ * 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.streampipes.service.core.filter;
+
+import org.apache.streampipes.model.client.user.DefaultRole;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.security.core.authority.SimpleGrantedAuthority;
+
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class TokenAuthenticationFilterTest {
+
+ @Test
+ void canActOnBehalfOfAllowsAdminRole() {
+ assertTrue(TokenAuthenticationFilter.canActOnBehalfOf(List.of(
+ new SimpleGrantedAuthority(DefaultRole.Constants.ROLE_ADMIN_VALUE)
+ )));
+ }
+
+ @Test
+ void canActOnBehalfOfAllowsServiceAdminRole() {
+ assertTrue(TokenAuthenticationFilter.canActOnBehalfOf(List.of(
+ new SimpleGrantedAuthority(DefaultRole.Constants.ROLE_SERVICE_ADMIN_VALUE)
+ )));
+ }
+
+ @Test
+ void canActOnBehalfOfRejectsOtherRoles() {
+ assertFalse(TokenAuthenticationFilter.canActOnBehalfOf(List.of(
+ new SimpleGrantedAuthority(DefaultRole.Constants.ROLE_PIPELINE_ADMIN_VALUE)
+ )));
+ }
+}