Holds test methods that are valid regardless of whether the audit target + * is a plain concrete index or a write alias. Each subclass provides its own + * {@code @ClassRule} {@link LocalCluster} and overrides + * {@link #cluster()} and {@link #auditTarget()} to supply the correct + * cluster and query pattern for that configuration variant.
+ * + *Using a shared base class avoids duplicating {@code testPersistsAuditEventsToTarget} + * and {@code testAuditDocumentContainsMandatoryFields} across the two concrete + * integration-test classes that differ only in cluster setup.
+ * + * @see InternalOpenSearchSinkIntegrationTestConcreteIndex concrete date-based index variant + * @see InternalOpenSearchSinkIntegrationTestAuditAlias write-alias variant + */ +abstract class AbstractInternalOpenSearchSinkIntegrationTest { + + /** + * Returns the cluster configured for this test variant. + * Implementations must expose the {@code @ClassRule}-annotated cluster field. + */ + abstract LocalCluster cluster(); + + /** + * Returns the index name or alias used to query audit documents. + * + *Examples:
+ *Tested Code Path: From the second event onward, + * {@code createIndexIfAbsent()} detects the target already exists + * (via {@code metadata.hasIndex()} for concrete indices or + * {@code metadata.hasAlias()} for aliases) and returns {@code true} + * without attempting recreation. Successful persistence of both events + * confirms the early-return branch works correctly.
+ */ + @Test + public void testPersistsAuditEventsToTarget() { + try (Client client = cluster().getInternalNodeClient()) { + long before = countAuditDocs(client); + + generateAuditEvent("_cluster/health"); + generateAuditEvent("_cluster/stats"); + generateAuditEvent("_nodes"); + + await().atMost(10, SECONDS).pollInterval(200, MILLISECONDS).untilAsserted(() -> { + refreshAuditTarget(client); + assertThat("At least 3 new audit events must be persisted", countAuditDocs(client) - before, greaterThanOrEqualTo(3L)); + }); + } + } + + /** + * Validates that audit documents contain all mandatory fields defined by the + * audit log specification. + * + *Core Fields (all events):
+ *REST-Specific Fields:
+ *Absent Fields (transport audit disabled):
+ *Shared tests ({@code testPersistsAuditEventsToTarget} and + * {@code testAuditDocumentContainsMandatoryFields}) are inherited from + * {@link AbstractInternalOpenSearchSinkIntegrationTest}.
+ * + * @see InternalOpenSearchSinkIntegrationTestConcreteIndex for the concrete date-based index variant + * @see InternalOpenSearchSinkTest for unit tests covering exception and race-condition branches */ -public class InternalOpenSearchSinkIntegrationTestAuditAlias { +public class InternalOpenSearchSinkIntegrationTestAuditAlias extends AbstractInternalOpenSearchSinkIntegrationTest { private static final String AUDIT_ALIAS = "security-audit-write-alias"; private static final String BACKING_INDEX = "security-audit-backend-000001"; @@ -52,27 +57,21 @@ public class InternalOpenSearchSinkIntegrationTestAuditAlias { static final TestAlias auditAlias = new TestAlias(AUDIT_ALIAS).on(backingIndex).writeIndex(backingIndex); @ClassRule - public static final LocalCluster cluster = new LocalCluster.Builder().clusterManager(ClusterManager.SINGLENODE) + public static final LocalCluster CLUSTER = new LocalCluster.Builder().clusterManager(ClusterManager.SINGLENODE) .nodeSettings(Map.of("plugins.security.audit.config.index", AUDIT_ALIAS)) .internalAudit(new AuditConfiguration(true).filters(new AuditFilters().enabledRest(true).enabledTransport(false))) .indices(backingIndex) .aliases(auditAlias) .build(); - /** Counts all audit documents reachable through the write alias. */ - private long countAuditDocs(TestRestClient client) { - HttpResponse response = client.postJson(AUDIT_ALIAS + "/_search", """ - {"query": {"match_all": {}}, "size": 0} - """); - response.assertStatusCode(200); - return response.getLongFromJsonBody("/hits/total/value"); + @Override + LocalCluster cluster() { + return CLUSTER; } - /** Issues an authenticated REST GET that triggers an {@code AUTHENTICATED} audit event. */ - private void generateAuditEvent(String path) { - try (TestRestClient restClient = cluster.getRestClient(cluster.getAdminCertificate())) { - restClient.get(path); - } + @Override + String auditTarget() { + return AUDIT_ALIAS; } /** @@ -81,13 +80,23 @@ private void generateAuditEvent(String path) { * *Generates one event, then checks that the alias still resolves to the * backing index and no spurious concrete index was created.
+ * + *Tested Code Path: {@code metadata.hasAlias(indexName)} returns + * {@code true}, so the method logs a debug message and returns {@code true} + * immediately without attempting index creation.
*/ @Test public void testRecognizesAuditTargetAsWriteAlias() { - try (TestRestClient client = cluster.getRestClient(cluster.getAdminCertificate())) { + try (TestRestClient client = CLUSTER.getRestClient(CLUSTER.getAdminCertificate())) { generateAuditEvent("_cluster/health"); - await().until(() -> countAuditDocs(client) > 0); + await().until(() -> { + HttpResponse countResponse = client.postJson(AUDIT_ALIAS + "/_search", """ + {"query": {"match_all": {}}, "size": 0} + """); + countResponse.assertStatusCode(200); + return countResponse.getLongFromJsonBody("/hits/total/value") > 0; + }); HttpResponse aliasResponse = client.get("_alias/" + AUDIT_ALIAS); aliasResponse.assertStatusCode(200); @@ -106,55 +115,4 @@ public void testRecognizesAuditTargetAsWriteAlias() { assertThat("Backing index must exist physically", indexExistsResponse.getStatusCode(), is(200)); } } - - /** - * The alias branch is invoked on every {@code doStore} call. - * Generates three distinct events and asserts all are persisted, confirming - * that repeated writes through the alias succeed. - */ - @Test - public void testWritesEventsToAliasSuccessfully() { - try (TestRestClient client = cluster.getRestClient(cluster.getAdminCertificate())) { - long before = countAuditDocs(client); - - generateAuditEvent("_cluster/health"); - generateAuditEvent("_cluster/stats"); - generateAuditEvent("_nodes/info"); - - await().untilAsserted( - () -> assertThat("At least 3 events must be written through alias", countAuditDocs(client) - before, greaterThan(2L)) - ); - } - } - - /** - * Documents written via the alias must contain the same mandatory audit fields - * as those written to a concrete index (category, timestamp, REST method/path, - * layer and origin). Transport-specific fields must be absent since transport - * audit is disabled. - */ - @Test - public void testAuditDocumentsViaAliasContainMandatoryFields() { - try (TestRestClient client = cluster.getRestClient(cluster.getAdminCertificate())) { - long before = countAuditDocs(client); - generateAuditEvent("_cluster/health"); - - await().until(() -> countAuditDocs(client) > before); - - HttpResponse response = client.postJson(AUDIT_ALIAS + "/_search", """ - {"query": {"match_all": {}}, "size": 1, "sort": [{"@timestamp": "desc"}]} - """); - response.assertStatusCode(200); - - JsonNode source = response.bodyAsJsonNode().get("hits").get("hits").get(0).get("_source"); - - assertThat(source.has("audit_category"), is(true)); - assertThat(source.has("@timestamp"), is(true)); - assertThat(source.has("audit_rest_request_method"), is(true)); - assertThat(source.has("audit_rest_request_path"), is(true)); - assertThat(source.get("audit_request_layer").asText(), is("REST")); - assertThat(source.get("audit_request_origin").asText(), is("REST")); - assertThat(source.has("audit_transport_request_type"), is(false)); - } - } } diff --git a/src/integrationTest/java/org/opensearch/security/auditlog/sink/InternalOpenSearchSinkIntegrationTestConcreteIndex.java b/src/integrationTest/java/org/opensearch/security/auditlog/sink/InternalOpenSearchSinkIntegrationTestConcreteIndex.java new file mode 100644 index 0000000000..de6d73df96 --- /dev/null +++ b/src/integrationTest/java/org/opensearch/security/auditlog/sink/InternalOpenSearchSinkIntegrationTestConcreteIndex.java @@ -0,0 +1,115 @@ +/* + * 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.util.Arrays; + +import org.junit.ClassRule; +import org.junit.Test; + +import org.opensearch.action.admin.indices.get.GetIndexRequest; +import org.opensearch.action.admin.indices.get.GetIndexResponse; +import org.opensearch.test.framework.AuditConfiguration; +import org.opensearch.test.framework.AuditFilters; +import org.opensearch.test.framework.cluster.ClusterManager; +import org.opensearch.test.framework.cluster.LocalCluster; +import org.opensearch.transport.client.Client; + +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static java.util.concurrent.TimeUnit.SECONDS; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.greaterThan; +import static org.hamcrest.Matchers.is; +import static org.awaitility.Awaitility.await; + +/** + * Integration tests for {@link InternalOpenSearchSink} with a plain concrete index + * (no alias configured). + * + *Exercises the regular index creation path: the default pattern + * {@code 'security-auditlog-'YYYY.MM.dd} produces daily indices + * (e.g., {@code security-auditlog-2025.01.11}). No pre-existing index or alias is + * present when the cluster starts, so the sink must create the index on first write.
+ * + *Shared tests ({@code testPersistsAuditEventsToTarget} and + * {@code testAuditDocumentContainsMandatoryFields}) are inherited from + * {@link AbstractInternalOpenSearchSinkIntegrationTest}.
+ * + * @see InternalOpenSearchSinkIntegrationTestAuditAlias for the write-alias variant + * @see InternalOpenSearchSinkTest for unit tests covering exception and race-condition branches + */ +public class InternalOpenSearchSinkIntegrationTestConcreteIndex extends AbstractInternalOpenSearchSinkIntegrationTest { + + private static final String AUDIT_INDEX_PREFIX = "security-auditlog-"; + + @ClassRule + public static final LocalCluster CLUSTER = new LocalCluster.Builder().clusterManager(ClusterManager.SINGLENODE) + .anonymousAuth(true) + .internalAudit(new AuditConfiguration(true).filters(new AuditFilters().enabledRest(true).enabledTransport(false))) + .build(); + + @Override + LocalCluster cluster() { + return CLUSTER; + } + + @Override + String auditTarget() { + return AUDIT_INDEX_PREFIX + "*"; + } + + /** + * Verifies that the audit sink automatically creates a date-based index + * on the first audit event. + * + *Tested Code Path: Both {@code metadata.hasAlias()} and + * {@code metadata.hasIndex()} return {@code false} for a brand-new index name, + * so {@code CreateIndexRequest} is executed. The resulting index name must + * match the pattern {@code security-auditlog-YYYY.MM.dd} + * (e.g., {@code security-auditlog-2025.01.11}).
+ */ + @Test + public void testCreatesAuditIndexAutomatically() { + try (Client client = CLUSTER.getInternalNodeClient()) { + long before = countAuditDocs(client); + + generateAuditEvent("_cluster/health"); + + await().atMost(10, SECONDS).pollInterval(200, MILLISECONDS).untilAsserted(() -> { + refreshAuditTarget(client); + assertThat("At least one new audit event must be generated", countAuditDocs(client), greaterThan(before)); + }); + + await().atMost(10, SECONDS).pollInterval(200, MILLISECONDS).untilAsserted(() -> { + GetIndexResponse response = client.admin() + .indices() + .getIndex(new GetIndexRequest().indices(AUDIT_INDEX_PREFIX + "*")) + .actionGet(); + + assertThat("At least one audit index must exist", response.indices().length, greaterThan(0)); + assertThat( + "All audit indices must follow date-based pattern security-auditlog-YYYY.MM.dd", + Arrays.stream(response.indices()).allMatch(name -> name.matches(AUDIT_INDEX_PREFIX + "\\d{4}\\.\\d{2}\\.\\d{2}$")), + is(true) + ); + }); + } + } +} diff --git a/src/test/java/org/opensearch/security/auditlog/sink/InternalOpenSearchSinkTest.java b/src/test/java/org/opensearch/security/auditlog/sink/InternalOpenSearchSinkTest.java new file mode 100644 index 0000000000..38f2ccaa65 --- /dev/null +++ b/src/test/java/org/opensearch/security/auditlog/sink/InternalOpenSearchSinkTest.java @@ -0,0 +1,202 @@ +/* + * 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 org.apache.lucene.tests.util.LuceneTestCase; +import org.junit.Before; +import org.junit.Test; + +import org.opensearch.ResourceAlreadyExistsException; +import org.opensearch.action.admin.indices.create.CreateIndexRequest; +import org.opensearch.action.admin.indices.create.CreateIndexResponse; +import org.opensearch.cluster.ClusterState; +import org.opensearch.cluster.metadata.Metadata; +import org.opensearch.cluster.service.ClusterService; +import org.opensearch.common.action.ActionFuture; +import org.opensearch.common.settings.Settings; +import org.opensearch.threadpool.ThreadPool; +import org.opensearch.transport.client.AdminClient; +import org.opensearch.transport.client.Client; +import org.opensearch.transport.client.IndicesAdminClient; + +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.is; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * Unit tests for {@link InternalOpenSearchSink#createIndexIfAbsent(String)}. + * + *Covers three branches that cannot be reliably triggered in an integration + * test against a single-node embedded cluster:
+ *The remaining branches ({@code hasAlias()}, {@code hasIndex()}, and + * successful creation) are covered by the integration tests:
+ *Scenario: In a multi-node cluster, two nodes can call + * {@code createIndexIfAbsent()} concurrently. The first node creates the index + * successfully; the second finds {@code hasIndex()} returning {@code false} + * (cluster state not yet propagated) and attempts creation, receiving + * {@link ResourceAlreadyExistsException}. The sink must treat this as a + * no-op success rather than a failure.
+ * + *Tested Code Path:
+ *{@code
+ * } catch (ResourceAlreadyExistsException e) {
+ * log.debug("Audit log index '{}' was created by another node", indexName);
+ * return true;
+ * }
+ */
+ @Test
+ public void createIndexIfAbsent_returnsTrue_onConcurrentCreationByAnotherNode() {
+ when(indicesAdminClient.create(any(CreateIndexRequest.class)))
+ .thenThrow(new ResourceAlreadyExistsException(TEST_INDEX));
+
+ boolean result = sink.createIndexIfAbsent(TEST_INDEX);
+
+ assertThat("Must return true when index was concurrently created by another node",
+ result, is(true));
+ verify(indicesAdminClient).create(any(CreateIndexRequest.class));
+ }
+
+ /**
+ * Verifies that {@code createIndexIfAbsent()} returns {@code false} when an
+ * unexpected exception is thrown, without propagating it to the caller.
+ *
+ * Scenario: An unforeseen cluster error (e.g., node disconnection, + * authorization failure, or I/O error) prevents index creation. The sink must + * log the error and return {@code false} so that the calling {@code doStore()} + * can handle the failure gracefully, rather than crashing the audit pipeline.
+ * + *Tested Code Path:
+ *{@code
+ * } catch (Exception e) {
+ * log.error("Error creating audit log index '{}'", indexName, e);
+ * return false;
+ * }
+ */
+ @Test
+ public void createIndexIfAbsent_returnsFalse_onUnexpectedException() {
+ when(indicesAdminClient.create(any(CreateIndexRequest.class)))
+ .thenThrow(new RuntimeException("simulated cluster failure"));
+
+ boolean result = sink.createIndexIfAbsent(TEST_INDEX);
+
+ assertThat("Must return false without propagating the exception to the caller",
+ result, is(false));
+ verify(indicesAdminClient).create(any(CreateIndexRequest.class));
+ }
+
+ /**
+ * Verifies that {@code createIndexIfAbsent()} returns {@code false} when the
+ * cluster manager acknowledges the request but the response is not acknowledged.
+ *
+ * Scenario: The {@code CreateIndexRequest} completes without throwing, + * but {@link CreateIndexResponse#isAcknowledged()} returns {@code false}. This + * can happen when the cluster manager times out waiting for all shard copies to + * confirm the mapping update. The index likely exists on the cluster manager node + * but may not yet be visible to other nodes. The sink logs an error and returns + * {@code false}, causing the current audit event to be dropped silently. The + * next event will find the index present and succeed normally.
+ * + *Tested Code Path:
+ *{@code
+ * final boolean acknowledged = clientProvider.admin().indices()
+ * .create(createIndexRequest).actionGet().isAcknowledged();
+ * if (acknowledged) { ... } else {
+ * log.error("Failed to create audit log index '{}'. Index creation was not acknowledged.", indexName);
+ * }
+ * return acknowledged; // returns false
+ * }
+ *
+ * Implementation note: {@code CreateIndexResponse} is instantiated directly + * (not mocked) because {@link org.opensearch.action.support.clustermanager.AcknowledgedResponse#isAcknowledged()} + * is declared {@code final} and cannot be stubbed by Mockito. {@code doReturn().when()} + * is used for {@code actionGet()} to avoid calling the generic method on the mock during + * stub setup (type erasure would confuse the standard {@code when().thenReturn()} form).
+ */ + @Test + public void createIndexIfAbsent_returnsFalse_whenCreationNotAcknowledged() { + // CreateIndexResponse is instantiated directly because isAcknowledged() is final + // in AcknowledgedResponse and cannot be stubbed by Mockito. + // doReturn().when() is used instead of when().thenReturn() to avoid calling the + // generic actionGet() on the mock during stub setup (prevents type-erasure issues). + CreateIndexResponse notAcknowledged = new CreateIndexResponse(false, false, TEST_INDEX); + doReturn(notAcknowledged).when(createIndexFuture).actionGet(); + when(indicesAdminClient.create(any(CreateIndexRequest.class))).thenReturn(createIndexFuture); + + boolean result = sink.createIndexIfAbsent(TEST_INDEX); + + assertThat("Must return false when index creation is not acknowledged by the cluster", result, is(false)); + verify(indicesAdminClient).create(any(CreateIndexRequest.class)); + } +}