You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ac...@apache.org on 2022/09/22 09:30:29 UTC
[camel] 01/04: camel-azure-key-vault - Add dev console for secrets
This is an automated email from the ASF dual-hosted git repository.
acosentino pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
commit 167d1895ae0fdefa282b79b7e69f85c26704172f
Author: Andrea Cosentino <an...@gmail.com>
AuthorDate: Thu Sep 22 10:13:54 2022 +0200
camel-azure-key-vault - Add dev console for secrets
---
.../camel-azure/camel-azure-key-vault/pom.xml | 4 +
.../key/vault/EventhubsReloadTriggerTask.java | 376 +++++++++++----------
2 files changed, 201 insertions(+), 179 deletions(-)
diff --git a/components/camel-azure/camel-azure-key-vault/pom.xml b/components/camel-azure/camel-azure-key-vault/pom.xml
index 5f02311ad27..cb5d0aa85ad 100644
--- a/components/camel-azure/camel-azure-key-vault/pom.xml
+++ b/components/camel-azure/camel-azure-key-vault/pom.xml
@@ -40,6 +40,10 @@
<groupId>org.apache.camel</groupId>
<artifactId>camel-support</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-console</artifactId>
+ </dependency>
<!-- azure sdk -->
<dependency>
diff --git a/components/camel-azure/camel-azure-key-vault/src/main/java/org/apache/camel/component/azure/key/vault/EventhubsReloadTriggerTask.java b/components/camel-azure/camel-azure-key-vault/src/main/java/org/apache/camel/component/azure/key/vault/EventhubsReloadTriggerTask.java
index 0c8949ba805..fee5f1170e7 100644
--- a/components/camel-azure/camel-azure-key-vault/src/main/java/org/apache/camel/component/azure/key/vault/EventhubsReloadTriggerTask.java
+++ b/components/camel-azure/camel-azure-key-vault/src/main/java/org/apache/camel/component/azure/key/vault/EventhubsReloadTriggerTask.java
@@ -52,187 +52,205 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * Period task which checks if Azure Key Vaults secrets has been updated and can trigger Camel to be reloaded.
+ * Period task which checks if Azure Key Vaults secrets has been updated and can
+ * trigger Camel to be reloaded.
*/
@PeriodicTask("azure-secret-refresh")
public class EventhubsReloadTriggerTask extends ServiceSupport implements CamelContextAware, Runnable {
- private static final Logger LOG = LoggerFactory.getLogger(EventhubsReloadTriggerTask.class);
-
- private static final String BLOB_SERVICE_URI_SEGMENT = ".blob.core.windows.net";
- private static final String SECRET_VERSION_ADD = "Microsoft.KeyVault.SecretNewVersionCreated";
-
- private CamelContext camelContext;
- private boolean reloadEnabled = true;
- private String secrets;
- private EventProcessorClient eventProcessorClient;
- private KeyVaultPropertiesFunction propertiesFunction;
- private final Map<String, Instant> updates = new HashMap<>();
-
- public EventhubsReloadTriggerTask() {
- }
-
- @Override
- public CamelContext getCamelContext() {
- return camelContext;
- }
-
- @Override
- public void setCamelContext(CamelContext camelContext) {
- this.camelContext = camelContext;
- }
-
- public boolean isReloadEnabled() {
- return reloadEnabled;
- }
-
- /**
- * Whether Camel should be reloaded on Azure secret updated
- */
- public void setReloadEnabled(boolean reloadEnabled) {
- this.reloadEnabled = reloadEnabled;
- }
-
- /**
- * A map of the updated secrets with the latest updated time.
- */
- public Map<String, Instant> getUpdates() {
- return Collections.unmodifiableMap(updates);
- }
-
- @Override
- protected void doStart() throws Exception {
- super.doStart();
-
- // auto-detect secrets in-use
- PropertiesComponent pc = camelContext.getPropertiesComponent();
- PropertiesFunction pf = pc.getPropertiesFunction("azure");
- if (pf instanceof KeyVaultPropertiesFunction) {
- propertiesFunction = (KeyVaultPropertiesFunction) pf;
- LOG.debug("Auto-detecting secrets from properties-function: {}", pf.getName());
- }
- // specific secrets
- secrets = camelContext.getVaultConfiguration().azure().getSecrets();
- if (ObjectHelper.isEmpty(secrets) && propertiesFunction == null) {
- throw new IllegalArgumentException("Secrets must be configured on Azure vault configuration");
- }
-
- String eventhubConnectionString = null;
- String blobAccessKey = null;
- String blobAccountName = null;
- String blobContainerName = null;
- AzureVaultConfiguration azureVaultConfiguration = getCamelContext().getVaultConfiguration().azure();
- if (ObjectHelper.isNotEmpty(azureVaultConfiguration)) {
- eventhubConnectionString = azureVaultConfiguration.getEventhubConnectionString();
- blobAccessKey = azureVaultConfiguration.getBlobAccessKey();
- blobAccountName = azureVaultConfiguration.getBlobAccountName();
- blobContainerName = azureVaultConfiguration.getBlobContainerName();
- }
- if (ObjectHelper.isNotEmpty(eventhubConnectionString) && ObjectHelper.isNotEmpty(blobAccessKey)
- && ObjectHelper.isNotEmpty(blobAccountName) && ObjectHelper.isNotEmpty(blobContainerName)) {
- BlobContainerAsyncClient c = new BlobContainerClientBuilder()
- .endpoint(String.format(Locale.ROOT, "https://%s" + BLOB_SERVICE_URI_SEGMENT, blobAccountName))
- .containerName(blobContainerName)
- .credential(new StorageSharedKeyCredential(blobAccountName, blobAccessKey))
- .buildAsyncClient();
-
- EventProcessorClientBuilder eventProcessorClientBuilder = new EventProcessorClientBuilder()
- .checkpointStore(new BlobCheckpointStore(c))
- .consumerGroup("$Default")
- .connectionString(eventhubConnectionString)
- .processEvent(this::onEventListener)
- .processError(this::onErrorListener)
- .transportType(AmqpTransportType.AMQP);
-
- eventProcessorClient = eventProcessorClientBuilder.buildEventProcessorClient();
- eventProcessorClient.start();
- } else {
- throw new RuntimeCamelException(
- "Using the Azure Key Vault Secret refresh task requires setting eventhubs connection String, Blob Account Name, Blob Access Key and Blob Container Name as application properties ");
- }
- }
-
- @Override
- protected void doShutdown() throws Exception {
- super.doShutdown();
-
- if (eventProcessorClient != null) {
- try {
- eventProcessorClient.stop();
- } catch (Exception e) {
- // ignore
- }
- eventProcessorClient = null;
- }
-
- updates.clear();
- }
-
- @Override
- public void run() {
- }
-
- protected boolean matchSecret(String name) {
- Set<String> set = new HashSet<>();
- if (secrets != null) {
- Collections.addAll(set, secrets.split(","));
- }
- if (propertiesFunction != null) {
- set.addAll(propertiesFunction.getSecrets());
- }
-
- for (String part : set) {
- boolean result = name.contains(part) || PatternHelper.matchPattern(name, part);
- LOG.trace("Matching secret id: {}={} -> {}", name, part, result);
- if (result) {
- return true;
- }
- }
-
- return false;
- }
-
- @Override
- public String toString() {
- return "Azure Secrets Refresh Task";
- }
-
- protected void onEventListener(final EventContext eventContext) {
- boolean triggerReloading = false;
-
- ObjectMapper mapper = new ObjectMapper();
- JsonNode actualObj = null;
- try {
- actualObj = mapper.readTree(eventContext.getEventData().getBodyAsString());
- } catch (JsonMappingException e) {
- e.printStackTrace();
- } catch (JsonProcessingException e) {
- e.printStackTrace();
- }
- for (int i = 0; i < actualObj.size(); i++) {
- String secret = actualObj.get(i).get("subject").textValue();
- String eventType = actualObj.get(i).get("eventType").textValue();
- if (ObjectHelper.isNotEmpty(secret) && ObjectHelper.isNotEmpty(eventType)) {
- if (eventType.equalsIgnoreCase(SECRET_VERSION_ADD)) {
- if (matchSecret(secret)) {
- if (isReloadEnabled()) {
- LOG.info("Update for Azure secret: {} detected, triggering CamelContext reload", secret);
- triggerReloading = true;
- }
- }
- }
- }
- }
- if (triggerReloading) {
- ContextReloadStrategy reload = camelContext.hasService(ContextReloadStrategy.class);
- if (reload != null) {
- // trigger reload
- reload.onReload(this);
- }
- }
- }
-
- public void onErrorListener(final ErrorContext errorContext) {
- //NOOP
- }
+ private static final Logger LOG = LoggerFactory.getLogger(EventhubsReloadTriggerTask.class);
+
+ private static final String BLOB_SERVICE_URI_SEGMENT = ".blob.core.windows.net";
+ private static final String SECRET_VERSION_ADD = "Microsoft.KeyVault.SecretNewVersionCreated";
+
+ private CamelContext camelContext;
+ private boolean reloadEnabled = true;
+ private String secrets;
+ private EventProcessorClient eventProcessorClient;
+ private KeyVaultPropertiesFunction propertiesFunction;
+ private volatile Instant lastCheckTime;
+ private volatile Instant lastReloadTime;
+ private final Map<String, Instant> updates = new HashMap<>();
+
+ public EventhubsReloadTriggerTask() {
+ }
+
+ @Override
+ public CamelContext getCamelContext() {
+ return camelContext;
+ }
+
+ @Override
+ public void setCamelContext(CamelContext camelContext) {
+ this.camelContext = camelContext;
+ }
+
+ public boolean isReloadEnabled() {
+ return reloadEnabled;
+ }
+
+ /**
+ * Whether Camel should be reloaded on Azure secret updated
+ */
+ public void setReloadEnabled(boolean reloadEnabled) {
+ this.reloadEnabled = reloadEnabled;
+ }
+
+ /**
+ * A map of the updated secrets with the latest updated time.
+ */
+ public Map<String, Instant> getUpdates() {
+ return Collections.unmodifiableMap(updates);
+ }
+
+ /**
+ * Last time this task checked GCP for updated secrets.
+ */
+ public Instant getLastCheckTime() {
+ return lastCheckTime;
+ }
+
+ /**
+ * Last time GCP secrets update triggered reload.
+ */
+ public Instant getLastReloadTime() {
+ return lastReloadTime;
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+
+ // auto-detect secrets in-use
+ PropertiesComponent pc = camelContext.getPropertiesComponent();
+ PropertiesFunction pf = pc.getPropertiesFunction("azure");
+ if (pf instanceof KeyVaultPropertiesFunction) {
+ propertiesFunction = (KeyVaultPropertiesFunction) pf;
+ LOG.debug("Auto-detecting secrets from properties-function: {}", pf.getName());
+ }
+ // specific secrets
+ secrets = camelContext.getVaultConfiguration().azure().getSecrets();
+ if (ObjectHelper.isEmpty(secrets) && propertiesFunction == null) {
+ throw new IllegalArgumentException("Secrets must be configured on Azure vault configuration");
+ }
+
+ String eventhubConnectionString = null;
+ String blobAccessKey = null;
+ String blobAccountName = null;
+ String blobContainerName = null;
+ AzureVaultConfiguration azureVaultConfiguration = getCamelContext().getVaultConfiguration().azure();
+ if (ObjectHelper.isNotEmpty(azureVaultConfiguration)) {
+ eventhubConnectionString = azureVaultConfiguration.getEventhubConnectionString();
+ blobAccessKey = azureVaultConfiguration.getBlobAccessKey();
+ blobAccountName = azureVaultConfiguration.getBlobAccountName();
+ blobContainerName = azureVaultConfiguration.getBlobContainerName();
+ }
+ if (ObjectHelper.isNotEmpty(eventhubConnectionString) && ObjectHelper.isNotEmpty(blobAccessKey)
+ && ObjectHelper.isNotEmpty(blobAccountName) && ObjectHelper.isNotEmpty(blobContainerName)) {
+ BlobContainerAsyncClient c = new BlobContainerClientBuilder()
+ .endpoint(String.format(Locale.ROOT, "https://%s" + BLOB_SERVICE_URI_SEGMENT, blobAccountName))
+ .containerName(blobContainerName)
+ .credential(new StorageSharedKeyCredential(blobAccountName, blobAccessKey)).buildAsyncClient();
+
+ EventProcessorClientBuilder eventProcessorClientBuilder = new EventProcessorClientBuilder()
+ .checkpointStore(new BlobCheckpointStore(c)).consumerGroup("$Default")
+ .connectionString(eventhubConnectionString).processEvent(this::onEventListener)
+ .processError(this::onErrorListener).transportType(AmqpTransportType.AMQP);
+
+ eventProcessorClient = eventProcessorClientBuilder.buildEventProcessorClient();
+ eventProcessorClient.start();
+ } else {
+ throw new RuntimeCamelException(
+ "Using the Azure Key Vault Secret refresh task requires setting eventhubs connection String, Blob Account Name, Blob Access Key and Blob Container Name as application properties ");
+ }
+ }
+
+ @Override
+ protected void doShutdown() throws Exception {
+ super.doShutdown();
+
+ if (eventProcessorClient != null) {
+ try {
+ eventProcessorClient.stop();
+ } catch (Exception e) {
+ // ignore
+ }
+ eventProcessorClient = null;
+ }
+
+ updates.clear();
+ }
+
+ @Override
+ public void run() {
+ lastCheckTime = Instant.now();
+ }
+
+ protected boolean matchSecret(String name) {
+ Set<String> set = new HashSet<>();
+ if (secrets != null) {
+ Collections.addAll(set, secrets.split(","));
+ }
+ if (propertiesFunction != null) {
+ set.addAll(propertiesFunction.getSecrets());
+ }
+
+ for (String part : set) {
+ boolean result = name.contains(part) || PatternHelper.matchPattern(name, part);
+ LOG.trace("Matching secret id: {}={} -> {}", name, part, result);
+ if (result) {
+ return true;
+ }
+ }
+
+ return false;
+ }
+
+ @Override
+ public String toString() {
+ return "Azure Secrets Refresh Task";
+ }
+
+ protected void onEventListener(final EventContext eventContext) {
+ boolean triggerReloading = false;
+
+ ObjectMapper mapper = new ObjectMapper();
+ JsonNode actualObj = null;
+ try {
+ actualObj = mapper.readTree(eventContext.getEventData().getBodyAsString());
+ } catch (JsonMappingException e) {
+ e.printStackTrace();
+ } catch (JsonProcessingException e) {
+ e.printStackTrace();
+ }
+ for (int i = 0; i < actualObj.size(); i++) {
+ String secret = actualObj.get(i).get("subject").textValue();
+ String eventType = actualObj.get(i).get("eventType").textValue();
+ if (ObjectHelper.isNotEmpty(secret) && ObjectHelper.isNotEmpty(eventType)) {
+ if (eventType.equalsIgnoreCase(SECRET_VERSION_ADD)) {
+ if (matchSecret(secret)) {
+ if (ObjectHelper.isNotEmpty(eventContext.getEventData().getEnqueuedTime())) {
+ updates.put(secret, eventContext.getEventData().getEnqueuedTime());
+ }
+ if (isReloadEnabled()) {
+ LOG.info("Update for Azure secret: {} detected, triggering CamelContext reload", secret);
+ triggerReloading = true;
+ }
+ }
+ }
+ }
+ }
+ if (triggerReloading) {
+ ContextReloadStrategy reload = camelContext.hasService(ContextReloadStrategy.class);
+ if (reload != null) {
+ // trigger reload
+ lastReloadTime = Instant.now();
+ reload.onReload(this);
+ }
+ }
+ }
+
+ public void onErrorListener(final ErrorContext errorContext) {
+ // NOOP
+ }
}