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 <[email protected]> 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) + ))); + } +}
