-
Notifications
You must be signed in to change notification settings - Fork 276
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Support datastreams as an AuditLog Sink (#4257)
Signed-off-by: tmanninger <[email protected]>
- Loading branch information
1 parent
a62e99a
commit 9c5e32a
Showing
7 changed files
with
498 additions
and
36 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
83 changes: 83 additions & 0 deletions
83
src/main/java/org/opensearch/security/auditlog/sink/AbstractInternalOpenSearchSink.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,83 @@ | ||
| /* | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| * | ||
| * The OpenSearch Contributors require contributions made to | ||
| * this file be licensed under the Apache-2.0 license or a | ||
| * compatible open source license. | ||
| * | ||
| * Modifications Copyright OpenSearch Contributors. See | ||
| * GitHub history for details. | ||
| */ | ||
|
|
||
| package org.opensearch.security.auditlog.sink; | ||
|
|
||
| import java.io.IOException; | ||
|
|
||
| import org.opensearch.action.DocWriteRequest; | ||
| import org.opensearch.action.index.IndexRequestBuilder; | ||
| import org.opensearch.action.support.WriteRequest.RefreshPolicy; | ||
| import org.opensearch.client.Client; | ||
| import org.opensearch.common.settings.Settings; | ||
| import org.opensearch.common.unit.TimeValue; | ||
| import org.opensearch.common.util.concurrent.ThreadContext.StoredContext; | ||
| import org.opensearch.security.auditlog.impl.AuditMessage; | ||
| import org.opensearch.security.support.ConfigConstants; | ||
| import org.opensearch.security.support.HeaderHelper; | ||
| import org.opensearch.threadpool.ThreadPool; | ||
|
|
||
| public abstract class AbstractInternalOpenSearchSink extends AuditLogSink { | ||
|
|
||
| protected final Client clientProvider; | ||
| private final ThreadPool threadPool; | ||
| private final DocWriteRequest.OpType storeOpType; | ||
|
|
||
| public AbstractInternalOpenSearchSink( | ||
| final String name, | ||
| final Settings settings, | ||
| final String settingsPrefix, | ||
| final Client clientProvider, | ||
| ThreadPool threadPool, | ||
| AuditLogSink fallbackSink, | ||
| DocWriteRequest.OpType storeOpType | ||
| ) { | ||
| super(name, settings, settingsPrefix, fallbackSink); | ||
| this.clientProvider = clientProvider; | ||
| this.threadPool = threadPool; | ||
| this.storeOpType = storeOpType; | ||
| } | ||
|
|
||
| @Override | ||
| public void close() throws IOException { | ||
|
|
||
| } | ||
|
|
||
| public boolean doStore(final AuditMessage msg, String indexName) { | ||
|
|
||
| if (Boolean.parseBoolean( | ||
| HeaderHelper.getSafeFromHeader(threadPool.getThreadContext(), ConfigConstants.OPENDISTRO_SECURITY_CONF_REQUEST_HEADER) | ||
| )) { | ||
| if (log.isTraceEnabled()) { | ||
| log.trace("audit log of audit log will not be executed"); | ||
| } | ||
| return true; | ||
| } | ||
|
|
||
| try (StoredContext ctx = threadPool.getThreadContext().stashContext()) { | ||
| try { | ||
| final IndexRequestBuilder irb = clientProvider.prepareIndex(indexName) | ||
| .setRefreshPolicy(RefreshPolicy.IMMEDIATE) | ||
| .setSource(msg.getAsMap()); | ||
| threadPool.getThreadContext().putHeader(ConfigConstants.OPENDISTRO_SECURITY_CONF_REQUEST_HEADER, "true"); | ||
| irb.setTimeout(TimeValue.timeValueMinutes(1)); | ||
| if (this.storeOpType != null) { | ||
| irb.setOpType(this.storeOpType); | ||
| } | ||
| irb.execute().actionGet(); | ||
| return true; | ||
| } catch (final Exception e) { | ||
| log.error("Unable to index audit log {} due to", msg, e); | ||
| return false; | ||
| } | ||
| } | ||
| } | ||
| } |
149 changes: 149 additions & 0 deletions
149
src/main/java/org/opensearch/security/auditlog/sink/InternalOpenSearchDataStreamSink.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,149 @@ | ||
| /* | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| * | ||
| * The OpenSearch Contributors require contributions made to | ||
| * this file be licensed under the Apache-2.0 license or a | ||
| * compatible open source license. | ||
| * | ||
| * Modifications Copyright OpenSearch Contributors. See | ||
| * GitHub history for details. | ||
| */ | ||
|
|
||
| package org.opensearch.security.auditlog.sink; | ||
|
|
||
| // CS-SUPPRESS-SINGLE: RegexpSingleline https://github.com/opensearch-project/OpenSearch/issues/3663 | ||
| import java.io.IOException; | ||
| import java.nio.file.Path; | ||
| import java.util.List; | ||
|
|
||
| import org.opensearch.ResourceAlreadyExistsException; | ||
| import org.opensearch.action.DocWriteRequest; | ||
| import org.opensearch.action.admin.indices.datastream.CreateDataStreamAction; | ||
| import org.opensearch.action.admin.indices.template.put.PutComposableIndexTemplateAction; | ||
| import org.opensearch.action.support.master.AcknowledgedResponse; | ||
| import org.opensearch.client.Client; | ||
| import org.opensearch.cluster.metadata.ComposableIndexTemplate; | ||
| import org.opensearch.cluster.metadata.DataStream; | ||
| import org.opensearch.cluster.metadata.Template; | ||
| import org.opensearch.common.settings.Settings; | ||
| import org.opensearch.security.auditlog.impl.AuditMessage; | ||
| import org.opensearch.security.support.ConfigConstants; | ||
| import org.opensearch.threadpool.ThreadPool; | ||
| import org.opensearch.transport.RemoteTransportException; | ||
|
|
||
| public final class InternalOpenSearchDataStreamSink extends AbstractInternalOpenSearchSink { | ||
|
|
||
| String dataStreamName; | ||
| private boolean dataStreamInitialized = false; | ||
|
|
||
| public InternalOpenSearchDataStreamSink( | ||
| final String name, | ||
| final Settings settings, | ||
| final String settingsPrefix, | ||
| final Path configPath, | ||
| final Client clientProvider, | ||
| ThreadPool threadPool, | ||
| AuditLogSink fallbackSink | ||
| ) { | ||
| super(name, settings, settingsPrefix, clientProvider, threadPool, fallbackSink, DocWriteRequest.OpType.CREATE); | ||
| Settings sinkSettings = getSinkSettings(settingsPrefix); | ||
|
|
||
| this.dataStreamName = sinkSettings.get(ConfigConstants.SECURITY_AUDIT_OPENSEARCH_DATASTREAM_NAME, "opensearch-security-auditlog"); | ||
|
|
||
| // Node is no ready yet... this.initDataStream() must be called later (in method doStore()) | ||
| } | ||
|
|
||
| private boolean initDataStream() { | ||
|
|
||
| if (this.dataStreamInitialized) { | ||
| return true; | ||
| } | ||
|
|
||
| Settings sinkSettings = getSinkSettings(settingsPrefix); | ||
|
|
||
| final boolean templateManage = sinkSettings.getAsBoolean( | ||
| ConfigConstants.SECURITY_AUDIT_OPENSEARCH_DATASTREAM_TEMPLATE_MANAGE, | ||
| true | ||
| ); | ||
|
|
||
| // Create datastream template | ||
| if (templateManage) { | ||
|
|
||
| final String templateName = sinkSettings.get( | ||
| ConfigConstants.SECURITY_AUDIT_OPENSEARCH_DATASTREAM_TEMPLATE_NAME, | ||
| "opensearch-security-auditlog" | ||
| ); | ||
| final Integer numberOfReplicas = sinkSettings.getAsInt( | ||
| ConfigConstants.SECURITY_AUDIT_OPENSEARCH_DATASTREAM_TEMPLATE_NUMBER_OF_REPLICAS, | ||
| 0 | ||
| ); | ||
| final Integer numberOfShards = sinkSettings.getAsInt( | ||
| ConfigConstants.SECURITY_AUDIT_OPENSEARCH_DATASTREAM_TEMPLATE_NUMBER_OF_SHARDS, | ||
| 1 | ||
| ); | ||
|
|
||
| ComposableIndexTemplate template = new ComposableIndexTemplate( | ||
| List.of(dataStreamName), | ||
| new Template( | ||
| Settings.builder().put("number_of_shards", numberOfShards).put("number_of_replicas", numberOfReplicas).build(), | ||
| null, | ||
| null | ||
| ), | ||
| null, | ||
| null, | ||
| null, | ||
| null, | ||
| new ComposableIndexTemplate.DataStreamTemplate(new DataStream.TimestampField("@timestamp")) | ||
| ); | ||
|
|
||
| try { | ||
| PutComposableIndexTemplateAction.Request request = new PutComposableIndexTemplateAction.Request(templateName); | ||
| request.indexTemplate(template); | ||
| AcknowledgedResponse response = clientProvider.execute(PutComposableIndexTemplateAction.INSTANCE, request).get(); | ||
| if (!response.isAcknowledged()) { | ||
| log.error("Failed to create index template {}", templateName); | ||
| return false; | ||
| } | ||
| } catch (final Exception e) { | ||
| log.error("Cannot create index template {} due to", templateName, e); | ||
| return false; | ||
| } | ||
| } | ||
|
|
||
| CreateDataStreamAction.Request createDataStreamRequest = new CreateDataStreamAction.Request(dataStreamName); | ||
| try { | ||
| AcknowledgedResponse response = clientProvider.admin().indices().createDataStream(createDataStreamRequest).get(); | ||
| if (!response.isAcknowledged()) { | ||
| log.error("Failed to create datastream {}", dataStreamName); | ||
| } | ||
| this.dataStreamInitialized = true; | ||
| } catch (final Exception e) { | ||
| if (e.getCause() instanceof ResourceAlreadyExistsException | ||
| || (e.getCause() instanceof RemoteTransportException | ||
| && e.getCause().getCause() instanceof ResourceAlreadyExistsException)) { | ||
| log.trace("Datastream {} already exists", dataStreamName); | ||
| this.dataStreamInitialized = true; | ||
| } else { | ||
| log.error("Cannot create datastream {} due to", dataStreamName, e); | ||
| return false; | ||
| } | ||
| } | ||
|
|
||
| return this.dataStreamInitialized; | ||
| } | ||
|
|
||
| @Override | ||
| public void close() throws IOException { | ||
|
|
||
| } | ||
|
|
||
| public boolean doStore(final AuditMessage msg) { | ||
|
|
||
| if (!this.initDataStream()) { | ||
| log.error("Datastream initializaten failed. Cannot write to auditlog"); | ||
| return false; | ||
| } | ||
|
|
||
| return super.doStore(msg, this.dataStreamName); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.