This is an automated email from the ASF dual-hosted git repository. riemer pushed a commit to branch dev in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit 184bc66ee3c0209ec7d50ba624047a9ac6029e20 Author: Dominik Riemer <[email protected]> AuthorDate: Sun May 23 16:45:08 2021 +0200 [STREAMPIPES-374] Improve management of SecretStaticProperty values --- .../master/util/AdapterEncryptionService.java | 33 +----- .../manager/execution/http/PipelineExecutor.java | 43 +++----- .../execution/http/PipelineStorageService.java | 34 ++---- .../streampipes/manager/secret/ISecretHandler.java | 26 +++++ .../manager/secret/SecretDecrypter.java | 45 ++++++++ .../manager/secret/SecretEncrypter.java | 43 ++++++++ .../streampipes/manager/secret/SecretProvider.java | 30 ++++++ .../streampipes/manager/secret/SecretService.java | 52 +++++++++ .../streampipes/manager/secret/SecretVisitor.java | 116 +++++++++++++++++++++ 9 files changed, 338 insertions(+), 84 deletions(-) diff --git a/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/util/AdapterEncryptionService.java b/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/util/AdapterEncryptionService.java index b795bee..984cf58 100644 --- a/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/util/AdapterEncryptionService.java +++ b/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/util/AdapterEncryptionService.java @@ -17,13 +17,11 @@ */ package org.apache.streampipes.connect.container.master.util; +import org.apache.streampipes.manager.secret.SecretProvider; import org.apache.streampipes.model.connect.adapter.AdapterDescription; import org.apache.streampipes.model.connect.adapter.GenericAdapterDescription; -import org.apache.streampipes.model.staticproperty.SecretStaticProperty; import org.apache.streampipes.model.staticproperty.StaticProperty; -import org.apache.streampipes.user.management.encryption.CredentialsManager; -import java.security.GeneralSecurityException; import java.util.List; public class AdapterEncryptionService { @@ -58,35 +56,10 @@ public class AdapterEncryptionService { } private void encrypt(List<StaticProperty> staticProperties) { - staticProperties - .stream() - .filter(SecretStaticProperty.class::isInstance) - .forEach(secret -> { - if (!((SecretStaticProperty) secret).getEncrypted()) { - try { - String encrypted = CredentialsManager.encrypt(ad.getUserName(), - ((SecretStaticProperty) secret).getValue()); - ((SecretStaticProperty) secret).setValue(encrypted); - ((SecretStaticProperty) secret).setEncrypted(true); - } catch (GeneralSecurityException e) { - e.printStackTrace(); - } - } - }); + SecretProvider.getEncryptionService(ad.getUserName()).applyConfig(staticProperties); } private void decrypt(List<StaticProperty> staticProperties) { - staticProperties.stream() - .filter(SecretStaticProperty.class::isInstance) - .forEach(sp -> { - try { - String decrypted = CredentialsManager.decrypt(ad.getUserName(), - ((SecretStaticProperty) sp).getValue()); - ((SecretStaticProperty) sp).setValue(decrypted); - ((SecretStaticProperty) sp).setEncrypted(false); - } catch (GeneralSecurityException e) { - e.printStackTrace(); - } - }); + SecretProvider.getDecryptionService(ad.getUserName()).applyConfig(staticProperties); } } diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/PipelineExecutor.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/PipelineExecutor.java index c8d3e68..5775753 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/PipelineExecutor.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/PipelineExecutor.java @@ -19,6 +19,7 @@ package org.apache.streampipes.manager.execution.http; import org.apache.streampipes.manager.execution.status.PipelineStatusManager; +import org.apache.streampipes.manager.secret.SecretProvider; import org.apache.streampipes.manager.util.TemporaryGraphStorage; import org.apache.streampipes.model.SpDataSet; import org.apache.streampipes.model.base.InvocableStreamPipesEntity; @@ -30,13 +31,10 @@ import org.apache.streampipes.model.message.PipelineStatusMessageType; import org.apache.streampipes.model.pipeline.Pipeline; import org.apache.streampipes.model.pipeline.PipelineHealthStatus; import org.apache.streampipes.model.pipeline.PipelineOperationStatus; -import org.apache.streampipes.model.staticproperty.SecretStaticProperty; import org.apache.streampipes.storage.api.IPipelineStorage; import org.apache.streampipes.storage.management.StorageDispatcher; -import org.apache.streampipes.user.management.encryption.CredentialsManager; import org.lightcouch.DocumentConflictException; -import java.security.GeneralSecurityException; import java.util.*; import java.util.stream.Collectors; @@ -79,14 +77,16 @@ public class PipelineExecutor { graphs.addAll(sepas); graphs.addAll(secs); - List<InvocableStreamPipesEntity> decryptedGraphs = decryptSecrets(graphs); + decryptSecrets(graphs); - graphs.forEach(g -> g.setStreamRequirements(Collections.emptyList())); + //graphs.forEach(g -> g.setStreamRequirements(Collections.emptyList())); PipelineOperationStatus status = new GraphSubmitter(pipeline.getPipelineId(), - pipeline.getName(), decryptedGraphs, dataSets) + pipeline.getName(), graphs, dataSets) .invokeGraphs(); + encryptSecrets(graphs); + if (status.isSuccess()) { storeInvocationGraphs(pipeline.getPipelineId(), graphs, dataSets); @@ -110,31 +110,12 @@ public class PipelineExecutor { .forEach(tp -> tp.setGroupId(UUID.randomUUID().toString())); } - private List<InvocableStreamPipesEntity> decryptSecrets(List<InvocableStreamPipesEntity> graphs) { - List<InvocableStreamPipesEntity> decryptedGraphs = new ArrayList<>(); - graphs.stream().map(g -> { - if (g instanceof DataProcessorInvocation) { - return new DataProcessorInvocation((DataProcessorInvocation) g); - } else { - return new DataSinkInvocation((DataSinkInvocation) g); - } - }).forEach(g -> { - g.getStaticProperties() - .stream() - .filter(SecretStaticProperty.class::isInstance) - .forEach(sp -> { - try { - String decrypted = CredentialsManager.decrypt(pipeline.getCreatedByUser(), - ((SecretStaticProperty) sp).getValue()); - ((SecretStaticProperty) sp).setValue(decrypted); - ((SecretStaticProperty) sp).setEncrypted(false); - } catch (GeneralSecurityException e) { - e.printStackTrace(); - } - }); - decryptedGraphs.add(g); - }); - return decryptedGraphs; + private void decryptSecrets(List<InvocableStreamPipesEntity> graphs) { + SecretProvider.getDecryptionService(pipeline.getCreatedByUser()).apply(graphs); + } + + private void encryptSecrets(List<InvocableStreamPipesEntity> graphs) { + SecretProvider.getEncryptionService(pipeline.getCreatedByUser()).apply(graphs); } public PipelineOperationStatus stopPipeline() { diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/PipelineStorageService.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/PipelineStorageService.java index e0fc07e..d59a4b3 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/PipelineStorageService.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/PipelineStorageService.java @@ -21,15 +21,13 @@ package org.apache.streampipes.manager.execution.http; import org.apache.streampipes.manager.data.PipelineGraph; import org.apache.streampipes.manager.data.PipelineGraphBuilder; import org.apache.streampipes.manager.matching.InvocationGraphBuilder; +import org.apache.streampipes.manager.secret.SecretProvider; import org.apache.streampipes.model.base.InvocableStreamPipesEntity; -import org.apache.streampipes.model.pipeline.Pipeline; import org.apache.streampipes.model.graph.DataProcessorInvocation; import org.apache.streampipes.model.graph.DataSinkInvocation; -import org.apache.streampipes.model.staticproperty.SecretStaticProperty; +import org.apache.streampipes.model.pipeline.Pipeline; import org.apache.streampipes.storage.management.StorageDispatcher; -import org.apache.streampipes.user.management.encryption.CredentialsManager; -import java.security.GeneralSecurityException; import java.util.List; import java.util.stream.Collectors; @@ -42,7 +40,7 @@ public class PipelineStorageService { } public void updatePipeline() { - preparePipeline(); + encryptSecrets(pipeline); StorageDispatcher.INSTANCE.getNoSqlStore().getPipelineStorageAPI().updatePipeline(pipeline); } @@ -54,7 +52,8 @@ public class PipelineStorageService { private void preparePipeline() { PipelineGraph pipelineGraph = new PipelineGraphBuilder(pipeline).buildGraph(); InvocationGraphBuilder builder = new InvocationGraphBuilder(pipelineGraph, pipeline.getPipelineId()); - List<InvocableStreamPipesEntity> graphs = encryptSecrets(builder.buildGraphs()); + List<InvocableStreamPipesEntity> graphs = builder.buildGraphs(); + encryptSecrets(graphs); List<DataSinkInvocation> secs = filter(graphs, DataSinkInvocation.class); List<DataProcessorInvocation> sepas = filter(graphs, DataProcessorInvocation.class); @@ -63,23 +62,12 @@ public class PipelineStorageService { pipeline.setActions(secs); } - private List<InvocableStreamPipesEntity> encryptSecrets(List<InvocableStreamPipesEntity> graphs) { - graphs.forEach(g -> g.getStaticProperties() - .stream() - .filter(SecretStaticProperty.class::isInstance) - .forEach(secret -> { - if (!((SecretStaticProperty) secret).getEncrypted()) { - try { - String encrypted = CredentialsManager.encrypt(pipeline.getCreatedByUser(), - ((SecretStaticProperty) secret).getValue()); - ((SecretStaticProperty) secret).setValue(encrypted); - ((SecretStaticProperty) secret).setEncrypted(true); - } catch (GeneralSecurityException e) { - e.printStackTrace(); - } - } - })); - return graphs; + private void encryptSecrets(List<InvocableStreamPipesEntity> graphs) { + SecretProvider.getEncryptionService(pipeline.getCreatedByUser()).apply(graphs); + } + + private void encryptSecrets(Pipeline pipeline) { + SecretProvider.getEncryptionService(pipeline.getCreatedByUser()).apply(pipeline); } private <T> List<T> filter(List<InvocableStreamPipesEntity> graphs, Class<T> clazz) { diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/secret/ISecretHandler.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/secret/ISecretHandler.java new file mode 100644 index 0000000..2537ce3 --- /dev/null +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/secret/ISecretHandler.java @@ -0,0 +1,26 @@ +/* + * 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.secret; + +public interface ISecretHandler { + + String apply(String username, String extractedValue); + + boolean shouldApply(boolean encrypted); + +} diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/secret/SecretDecrypter.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/secret/SecretDecrypter.java new file mode 100644 index 0000000..9685632 --- /dev/null +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/secret/SecretDecrypter.java @@ -0,0 +1,45 @@ +/* + * 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.secret; + + +import org.apache.streampipes.user.management.encryption.CredentialsManager; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.security.GeneralSecurityException; + +public class SecretDecrypter implements ISecretHandler { + + private static final Logger LOG = LoggerFactory.getLogger(SecretDecrypter.class); + + @Override + public String apply(String username, String extractedValue) { + try { + return CredentialsManager.decrypt(username, extractedValue); + } catch (GeneralSecurityException e) { + LOG.error("Could not decrypt value, returning original value"); + return extractedValue; + } + } + + @Override + public boolean shouldApply(boolean encrypted) { + return encrypted; + } +} diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/secret/SecretEncrypter.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/secret/SecretEncrypter.java new file mode 100644 index 0000000..877858d --- /dev/null +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/secret/SecretEncrypter.java @@ -0,0 +1,43 @@ +/* + * 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.secret; + +import org.apache.streampipes.user.management.encryption.CredentialsManager; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.security.GeneralSecurityException; + +public class SecretEncrypter implements ISecretHandler { + + private static final Logger LOG = LoggerFactory.getLogger(SecretEncrypter.class); + @Override + public String apply(String username, String extractedValue) { + try { + return CredentialsManager.encrypt(username, extractedValue); + } catch (GeneralSecurityException e) { + LOG.error("Could not encrypt value, returning original value"); + return extractedValue; + } + } + + @Override + public boolean shouldApply(boolean encrypted) { + return !encrypted; + } +} diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/secret/SecretProvider.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/secret/SecretProvider.java new file mode 100644 index 0000000..9676677 --- /dev/null +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/secret/SecretProvider.java @@ -0,0 +1,30 @@ +/* + * 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.secret; + +public class SecretProvider { + + public static SecretService getEncryptionService(String username) { + return new SecretService(username, new SecretEncrypter()) { + }; + } + + public static SecretService getDecryptionService(String username) { + return new SecretService(username, new SecretDecrypter()); + } +} diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/secret/SecretService.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/secret/SecretService.java new file mode 100644 index 0000000..4f4223b --- /dev/null +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/secret/SecretService.java @@ -0,0 +1,52 @@ +/* + * 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.secret; + +import org.apache.streampipes.model.base.InvocableStreamPipesEntity; +import org.apache.streampipes.model.pipeline.Pipeline; +import org.apache.streampipes.model.staticproperty.StaticProperty; + +import java.util.List; + +public class SecretService { + + private SecretVisitor visitor; + + public SecretService(String username, + ISecretHandler secretHandler) { + this.visitor = new SecretVisitor(username, secretHandler); + } + + public void apply(Pipeline pipeline) { + pipeline.getSepas().forEach(this::apply); + pipeline.getActions().forEach(this::apply); + } + + public void apply(List<InvocableStreamPipesEntity> graphs) { + graphs.forEach(this::apply); + } + + public void apply(InvocableStreamPipesEntity graph) { + applyConfig(graph.getStaticProperties()); + } + + public void applyConfig(List<StaticProperty> staticProperties) { + staticProperties.forEach(sp -> sp.accept(visitor)); + } + +} diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/secret/SecretVisitor.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/secret/SecretVisitor.java new file mode 100644 index 0000000..de51d4e --- /dev/null +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/secret/SecretVisitor.java @@ -0,0 +1,116 @@ +/* + * 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.secret; + +import org.apache.streampipes.model.staticproperty.*; + +public class SecretVisitor implements StaticPropertyVisitor { + + private String username; + private ISecretHandler secretHandler; + + public SecretVisitor(String username, + ISecretHandler secretHandler) { + this.username = username; + this.secretHandler = secretHandler; + } + + @Override + public void visit(AnyStaticProperty property) { + + } + + @Override + public void visit(CodeInputStaticProperty codeInputStaticProperty) { + + } + + @Override + public void visit(CollectionStaticProperty collectionStaticProperty) { + collectionStaticProperty.getMembers().forEach(sp -> sp.accept(this)); + } + + @Override + public void visit(ColorPickerStaticProperty colorPickerStaticProperty) { + + } + + @Override + public void visit(DomainStaticProperty domainStaticProperty) { + + } + + @Override + public void visit(FileStaticProperty fileStaticProperty) { + + } + + @Override + public void visit(FreeTextStaticProperty freeTextStaticProperty) { + + } + + @Override + public void visit(MappingPropertyNary mappingPropertyNary) { + + } + + @Override + public void visit(MappingPropertyUnary mappingPropertyUnary) { + + } + + @Override + public void visit(MatchingStaticProperty matchingStaticProperty) { + + } + + @Override + public void visit(OneOfStaticProperty oneOfStaticProperty) { + + } + + @Override + public void visit(SecretStaticProperty secretStaticProperty) { + if (secretHandler.shouldApply(secretStaticProperty.getEncrypted())) { + String newValue = secretHandler.apply(username, secretStaticProperty.getValue()); + secretStaticProperty.setValue(newValue); + secretStaticProperty.setEncrypted(!secretStaticProperty.getEncrypted()); + } + } + + @Override + public void visit(StaticPropertyAlternative staticPropertyAlternative) { + staticPropertyAlternative.getStaticProperty().accept(this); + } + + @Override + public void visit(StaticPropertyAlternatives staticPropertyAlternatives) { + staticPropertyAlternatives.getAlternatives().forEach(sp -> sp.accept(this)); + } + + @Override + public void visit(StaticPropertyGroup staticPropertyGroup) { + staticPropertyGroup.getStaticProperties().forEach(sp -> sp.accept(this)); + } + + @Override + public void visit(RemoteOneOfStaticProperty remoteOneOfStaticProperty) { + + } +}
