diff --git a/components/camel-google/camel-google-secret-manager/pom.xml b/components/camel-google/camel-google-secret-manager/pom.xml index 7dac5778f195c..b7ed680aa4367 100644 --- a/components/camel-google/camel-google-secret-manager/pom.xml +++ b/components/camel-google/camel-google-secret-manager/pom.xml @@ -83,5 +83,10 @@ camel-test-junit5 test + + org.assertj + assertj-core + test + diff --git a/components/camel-google/camel-google-secret-manager/src/main/java/org/apache/camel/component/google/secret/manager/GoogleSecretManagerPropertiesFunction.java b/components/camel-google/camel-google-secret-manager/src/main/java/org/apache/camel/component/google/secret/manager/GoogleSecretManagerPropertiesFunction.java index 82dc1e3c29060..dc2539badc3bc 100644 --- a/components/camel-google/camel-google-secret-manager/src/main/java/org/apache/camel/component/google/secret/manager/GoogleSecretManagerPropertiesFunction.java +++ b/components/camel-google/camel-google-secret-manager/src/main/java/org/apache/camel/component/google/secret/manager/GoogleSecretManagerPropertiesFunction.java @@ -55,8 +55,8 @@ * Otherwise it is possible to specify the credentials as properties: * * *

* @@ -79,6 +79,8 @@ public class GoogleSecretManagerPropertiesFunction extends ServiceSupport implem private static final String CAMEL_VAULT_GCP_PROJECT_ID = "CAMEL_VAULT_GCP_PROJECT_ID"; private static final String CAMEL_VAULT_GCP_USE_DEFAULT_INSTANCE = "CAMEL_VAULT_GCP_USE_DEFAULT_INSTANCE"; + private static final ObjectMapper MAPPER = new ObjectMapper(); + boolean useDefaultInstance; private CamelContext camelContext; @@ -207,8 +209,7 @@ private String getSecretFromSource( returnValue = response.getPayload().getData().toStringUtf8(); } if (ObjectHelper.isNotEmpty(subkey) && ObjectHelper.isNotEmpty(returnValue)) { - ObjectMapper mapper = new ObjectMapper(); - JsonNode actualObj = mapper.readTree(returnValue); + JsonNode actualObj = MAPPER.readTree(returnValue); JsonNode field = actualObj.get(subkey); if (ObjectHelper.isNotEmpty(field)) { returnValue = field.textValue(); diff --git a/components/camel-google/camel-google-secret-manager/src/main/java/org/apache/camel/component/google/secret/manager/vault/PubsubReloadTriggerTask.java b/components/camel-google/camel-google-secret-manager/src/main/java/org/apache/camel/component/google/secret/manager/vault/PubsubReloadTriggerTask.java index a0ca3bcb01ce7..6edf740236fbd 100644 --- a/components/camel-google/camel-google-secret-manager/src/main/java/org/apache/camel/component/google/secret/manager/vault/PubsubReloadTriggerTask.java +++ b/components/camel-google/camel-google-secret-manager/src/main/java/org/apache/camel/component/google/secret/manager/vault/PubsubReloadTriggerTask.java @@ -24,6 +24,7 @@ import java.util.Map; import java.util.Set; +import com.google.api.core.ApiService; import com.google.api.gax.core.FixedCredentialsProvider; import com.google.auth.Credentials; import com.google.auth.oauth2.GoogleCredentials; @@ -89,7 +90,7 @@ public boolean isReloadEnabled() { } /** - * Whether Camel should be reloaded on AWS secret updated + * Whether Camel should be reloaded on GCP secret updated */ public void setReloadEnabled(boolean reloadEnabled) { this.reloadEnabled = reloadEnabled; @@ -120,6 +121,9 @@ public Instant getLastReloadTime() { protected void doStart() throws Exception { super.doStart(); + // specific secrets + secrets = camelContext.getVaultConfiguration().gcp().getSecrets(); + // auto-detect secrets in-use PropertiesComponent pc = camelContext.getPropertiesComponent(); PropertiesFunction pf = pc.getPropertiesFunction("gcp"); @@ -127,8 +131,7 @@ protected void doStart() throws Exception { propertiesFunction = (GoogleSecretManagerPropertiesFunction) pf; LOG.debug("Auto-detecting secrets from properties-function: {}", pf.getName()); } - // specific secrets - secrets = camelContext.getVaultConfiguration().aws().getSecrets(); + if (ObjectHelper.isEmpty(secrets) && propertiesFunction == null) { throw new IllegalArgumentException("Secrets must be configured on GCP vault configuration"); } @@ -187,7 +190,14 @@ protected void doShutdown() throws Exception { public void run() { lastCheckTime = Instant.now(); - subscriber.startAsync().awaitRunning(); + if (subscriber == null) { + return; + } + // the subscriber is push based, so it only has to be started once. A Google ApiService can only be + // started while it is still NEW, starting it again on every period would fail with an IllegalStateException + if (subscriber.state() == ApiService.State.NEW) { + subscriber.startAsync().awaitRunning(); + } } protected boolean matchSecret(String name) { @@ -220,14 +230,13 @@ public class FilteringEventMessageReceiver implements MessageReceiver { private static final String SECRET_UPDATE = "SECRET_UPDATE"; private static final String SECRET_VERSION_ADD = "SECRET_VERSION_ADD"; - private boolean triggerReloading; - @Override public void receiveMessage(PubsubMessage message, AckReplyConsumer consumer) { + boolean triggerReloading = false; String secretId = message.getAttributesMap().get("secretId"); String eventType = message.getAttributesMap().get("eventType"); - if (eventType.equalsIgnoreCase(SECRET_UPDATE) || eventType.equalsIgnoreCase(SECRET_VERSION_ADD)) { - if (matchSecret(secretId)) { + if (SECRET_UPDATE.equalsIgnoreCase(eventType) || SECRET_VERSION_ADD.equalsIgnoreCase(eventType)) { + if (secretId != null && matchSecret(secretId)) { int secretNameBeginInd = secretId.lastIndexOf("/") + 1; updates.put(secretId.substring(secretNameBeginInd), Instant.ofEpochSecond(message.getPublishTime().getSeconds(), message.getPublishTime().getNanos())); diff --git a/components/camel-google/camel-google-secret-manager/src/test/java/org/apache/camel/component/google/secret/manager/vault/PubsubReloadTriggerTaskTest.java b/components/camel-google/camel-google-secret-manager/src/test/java/org/apache/camel/component/google/secret/manager/vault/PubsubReloadTriggerTaskTest.java new file mode 100644 index 0000000000000..56c244fed5567 --- /dev/null +++ b/components/camel-google/camel-google-secret-manager/src/test/java/org/apache/camel/component/google/secret/manager/vault/PubsubReloadTriggerTaskTest.java @@ -0,0 +1,156 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.google.secret.manager.vault; + +import java.lang.reflect.Field; +import java.util.concurrent.atomic.AtomicInteger; + +import com.google.cloud.pubsub.v1.AckReplyConsumer; +import com.google.pubsub.v1.PubsubMessage; +import org.apache.camel.RuntimeCamelException; +import org.apache.camel.impl.DefaultCamelContext; +import org.apache.camel.support.DefaultContextReloadStrategy; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class PubsubReloadTriggerTaskTest { + + private static final class CountingReloadStrategy extends DefaultContextReloadStrategy { + private final AtomicInteger reloads = new AtomicInteger(); + + @Override + public void onReload(Object source) { + reloads.incrementAndGet(); + } + } + + private static final class RecordingAck implements AckReplyConsumer { + private boolean acked; + + @Override + public void ack() { + acked = true; + } + + @Override + public void nack() { + } + } + + private static void setField(Object target, String name, Object value) throws Exception { + Field f = PubsubReloadTriggerTask.class.getDeclaredField(name); + f.setAccessible(true); + f.set(target, value); + } + + private static Object getField(Object target, String name) throws Exception { + Field f = PubsubReloadTriggerTask.class.getDeclaredField(name); + f.setAccessible(true); + return f.get(target); + } + + private static PubsubMessage message(String eventType, String secretId) { + PubsubMessage.Builder builder = PubsubMessage.newBuilder(); + if (eventType != null) { + builder.putAttributes("eventType", eventType); + } + if (secretId != null) { + builder.putAttributes("secretId", secretId); + } + return builder.build(); + } + + @Test + void secretsAreReadFromTheGcpVaultConfiguration() throws Exception { + DefaultCamelContext context = new DefaultCamelContext(); + context.getVaultConfiguration().gcp().setSecrets("gcp-secret"); + context.getVaultConfiguration().aws().setSecrets("aws-secret"); + + PubsubReloadTriggerTask task = new PubsubReloadTriggerTask(); + task.setCamelContext(context); + + // no GCP credentials are configured, so starting the task fails once it reaches the gcp properties + // function. By then the secrets to watch must already have been read from the GCP vault configuration + assertThatThrownBy(task::start).isInstanceOf(RuntimeCamelException.class); + assertThat(getField(task, "secrets")).isEqualTo("gcp-secret"); + } + + @Test + void reloadIsTriggeredOnlyForMatchingSecrets() throws Exception { + DefaultCamelContext context = new DefaultCamelContext(); + CountingReloadStrategy reload = new CountingReloadStrategy(); + context.addService(reload); + context.start(); + + PubsubReloadTriggerTask task = new PubsubReloadTriggerTask(); + task.setCamelContext(context); + setField(task, "secrets", "tracked"); + + PubsubReloadTriggerTask.FilteringEventMessageReceiver receiver = task.new FilteringEventMessageReceiver(); + + RecordingAck matching = new RecordingAck(); + receiver.receiveMessage(message("SECRET_UPDATE", "projects/p/secrets/tracked"), matching); + assertThat(reload.reloads.get()).isEqualTo(1); + assertThat(matching.acked).isTrue(); + assertThat(task.getUpdates()).containsKey("tracked"); + + // a message for a secret that is not watched must not trigger another reload + RecordingAck other = new RecordingAck(); + receiver.receiveMessage(message("SECRET_UPDATE", "projects/p/secrets/other"), other); + assertThat(reload.reloads.get()).isEqualTo(1); + assertThat(other.acked).isTrue(); + + // neither must an event type that is not a secret update + RecordingAck unrelated = new RecordingAck(); + receiver.receiveMessage(message("SECRET_DELETE", "projects/p/secrets/tracked"), unrelated); + assertThat(reload.reloads.get()).isEqualTo(1); + assertThat(unrelated.acked).isTrue(); + + context.stop(); + } + + @Test + void messagesWithoutAttributesAreAcknowledged() throws Exception { + DefaultCamelContext context = new DefaultCamelContext(); + PubsubReloadTriggerTask task = new PubsubReloadTriggerTask(); + task.setCamelContext(context); + setField(task, "secrets", "tracked"); + + PubsubReloadTriggerTask.FilteringEventMessageReceiver receiver = task.new FilteringEventMessageReceiver(); + + // a message published on the subscription by something other than the secret manager event feed + RecordingAck noAttributes = new RecordingAck(); + assertThatCode(() -> receiver.receiveMessage(message(null, null), noAttributes)).doesNotThrowAnyException(); + assertThat(noAttributes.acked).isTrue(); + + RecordingAck noSecretId = new RecordingAck(); + assertThatCode(() -> receiver.receiveMessage(message("SECRET_UPDATE", null), noSecretId)).doesNotThrowAnyException(); + assertThat(noSecretId.acked).isTrue(); + } + + @Test + void runWithoutSubscriberDoesNotThrow() { + PubsubReloadTriggerTask task = new PubsubReloadTriggerTask(); + task.setCamelContext(new DefaultCamelContext()); + + assertThatCode(task::run).doesNotThrowAnyException(); + assertThat(task.getLastCheckTime()).isNotNull(); + } +}