This is an automated email from the ASF dual-hosted git repository.
dominikriemer pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to refs/heads/dev by this push:
new a80b745239 fix: Improve handling of runtime-resolved configurations
(#4696)
a80b745239 is described below
commit a80b745239148a36cda69f125b86f3a5fe287a56
Author: Dominik Riemer <[email protected]>
AuthorDate: Wed Jul 8 17:15:30 2026 +0200
fix: Improve handling of runtime-resolved configurations (#4696)
---
.../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)
+ )));
+ }
+}
