diff --git a/build.gradle b/build.gradle index 3d0c160c5..1d612a47a 100644 --- a/build.gradle +++ b/build.gradle @@ -141,6 +141,14 @@ repositories { maven { url "https://aws.oss.sonatype.org/content/repositories/snapshots" } } +compileJava { + options.compilerArgs.addAll(["-processor", 'lombok.launch.AnnotationProcessorHider$AnnotationProcessor']) +} +compileTestJava { + options.compilerArgs.addAll(["-processor", 'lombok.launch.AnnotationProcessorHider$AnnotationProcessor']) +} + + sourceSets.main.java.srcDirs = ['src/main/generated','src/main/java'] configurations { zipArchive diff --git a/src/main/java/org/opensearch/securityanalytics/threatintel/action/DeleteDatasourceAction.java b/src/main/java/org/opensearch/securityanalytics/threatintel/action/DeleteDatasourceAction.java new file mode 100644 index 000000000..98147bb9f --- /dev/null +++ b/src/main/java/org/opensearch/securityanalytics/threatintel/action/DeleteDatasourceAction.java @@ -0,0 +1,27 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.securityanalytics.threatintel.action; + +import org.opensearch.action.ActionType; +import org.opensearch.action.support.master.AcknowledgedResponse; + +/** + * Threat intel datasource delete action + */ +public class DeleteDatasourceAction extends ActionType { + /** + * Delete datasource action instance + */ + public static final DeleteDatasourceAction INSTANCE = new DeleteDatasourceAction(); + /** + * Delete datasource action name + */ + public static final String NAME = "cluster:admin/security_analytics/datasource/delete"; + + private DeleteDatasourceAction() { + super(NAME, AcknowledgedResponse::new); + } +} diff --git a/src/main/java/org/opensearch/securityanalytics/threatintel/action/DeleteDatasourceRequest.java b/src/main/java/org/opensearch/securityanalytics/threatintel/action/DeleteDatasourceRequest.java new file mode 100644 index 000000000..059f1be9c --- /dev/null +++ b/src/main/java/org/opensearch/securityanalytics/threatintel/action/DeleteDatasourceRequest.java @@ -0,0 +1,59 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.securityanalytics.threatintel.action; + +import lombok.AllArgsConstructor; +import lombok.Getter; +import lombok.Setter; +import org.opensearch.action.ActionRequest; +import org.opensearch.action.ActionRequestValidationException; +import org.opensearch.core.common.io.stream.StreamInput; +import org.opensearch.core.common.io.stream.StreamOutput; +import org.opensearch.securityanalytics.threatintel.common.ParameterValidator; + +import java.io.IOException; + +/** + * Threat intel datasource delete request + */ +@Getter +@Setter +@AllArgsConstructor +public class DeleteDatasourceRequest extends ActionRequest { + private static final ParameterValidator VALIDATOR = new ParameterValidator(); + /** + * @param name the datasource name + * @return the datasource name + */ + private String name; + + /** + * Constructor + * + * @param in the stream input + * @throws IOException IOException + */ + public DeleteDatasourceRequest(final StreamInput in) throws IOException { + super(in); + this.name = in.readString(); + } + + @Override + public ActionRequestValidationException validate() { + ActionRequestValidationException errors = null; + if (VALIDATOR.validateDatasourceName(name).isEmpty() == false) { + errors = new ActionRequestValidationException(); + errors.addValidationError("no such datasource exist"); + } + return errors; + } + + @Override + public void writeTo(final StreamOutput out) throws IOException { + super.writeTo(out); + out.writeString(name); + } +} diff --git a/src/main/java/org/opensearch/securityanalytics/threatintel/action/DeleteDatasourceTransportAction.java b/src/main/java/org/opensearch/securityanalytics/threatintel/action/DeleteDatasourceTransportAction.java new file mode 100644 index 000000000..2117bbf13 --- /dev/null +++ b/src/main/java/org/opensearch/securityanalytics/threatintel/action/DeleteDatasourceTransportAction.java @@ -0,0 +1,152 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.securityanalytics.threatintel.action; + +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; +import org.opensearch.OpenSearchStatusException; +import org.opensearch.ResourceNotFoundException; +import org.opensearch.action.support.ActionFilters; +import org.opensearch.action.support.HandledTransportAction; +import org.opensearch.action.support.master.AcknowledgedResponse; +import org.opensearch.common.inject.Inject; +import org.opensearch.core.action.ActionListener; +import org.opensearch.core.rest.RestStatus; + +import org.opensearch.ingest.IngestService; +import org.opensearch.securityanalytics.model.DetectorTrigger; +import org.opensearch.securityanalytics.threatintel.common.DatasourceState; +import org.opensearch.securityanalytics.threatintel.common.ThreatIntelLockService; +import org.opensearch.securityanalytics.threatintel.dao.DatasourceDao; +import org.opensearch.securityanalytics.threatintel.dao.ThreatIntelFeedDao; +import org.opensearch.securityanalytics.threatintel.jobscheduler.Datasource; +import org.opensearch.tasks.Task; +import org.opensearch.threadpool.ThreadPool; +import org.opensearch.transport.TransportService; + +import java.io.IOException; + +/** + * Transport action to delete datasource + */ +public class DeleteDatasourceTransportAction extends HandledTransportAction { + private static final Logger log = LogManager.getLogger(DetectorTrigger.class); + + private static final long LOCK_DURATION_IN_SECONDS = 300l; + private final ThreatIntelLockService lockService; + private final IngestService ingestService; + private final DatasourceDao datasourceDao; + private final ThreatIntelFeedDao threatIntelFeedDao; +// private final Ip2GeoProcessorDao ip2GeoProcessorDao; + private final ThreadPool threadPool; + + /** + * Constructor + * @param transportService the transport service + * @param actionFilters the action filters + * @param lockService the lock service + * @param ingestService the ingest service + * @param datasourceDao the datasource facade + */ + @Inject + public DeleteDatasourceTransportAction( + final TransportService transportService, + final ActionFilters actionFilters, + final ThreatIntelLockService lockService, + final IngestService ingestService, + final DatasourceDao datasourceDao, + final ThreatIntelFeedDao threatIntelFeedDao, +// final Ip2GeoProcessorDao ip2GeoProcessorDao, + final ThreadPool threadPool + ) { + super(DeleteDatasourceAction.NAME, transportService, actionFilters, DeleteDatasourceRequest::new); + this.lockService = lockService; + this.ingestService = ingestService; + this.datasourceDao = datasourceDao; + this.threatIntelFeedDao = threatIntelFeedDao; +// this.ip2GeoProcessorDao = ip2GeoProcessorDao; + this.threadPool = threadPool; + } + + /** + * We delete datasource regardless of its state as long as we can acquire a lock + * + * @param task the task + * @param request the request + * @param listener the listener + */ + @Override + protected void doExecute(final Task task, final DeleteDatasourceRequest request, final ActionListener listener) { + lockService.acquireLock(request.getName(), LOCK_DURATION_IN_SECONDS, ActionListener.wrap(lock -> { + if (lock == null) { + listener.onFailure( + new OpenSearchStatusException("Another processor is holding a lock on the resource. Try again later", RestStatus.BAD_REQUEST) + ); + log.error("Another processor is holding lock, BAD_REQUEST exception", RestStatus.BAD_REQUEST); + + return; + } + try { + // TODO: makes every sub-methods as async call to avoid using a thread in generic pool + threadPool.generic().submit(() -> { + try { + deleteDatasource(request.getName()); + lockService.releaseLock(lock); + listener.onResponse(new AcknowledgedResponse(true)); + } catch (Exception e) { + lockService.releaseLock(lock); + listener.onFailure(e); + log.error("delete data source failed",e); + } + }); + } catch (Exception e) { + lockService.releaseLock(lock); + listener.onFailure(e); + log.error("Internal server error", e); + } + }, exception -> { listener.onFailure(exception); })); + } + + protected void deleteDatasource(final String datasourceName) throws IOException { + Datasource datasource = datasourceDao.getDatasource(datasourceName); + if (datasource == null) { + throw new ResourceNotFoundException("no such datasource exist"); + } + DatasourceState previousState = datasource.getState(); +// setDatasourceStateAsDeleting(datasource); + + try { + threatIntelFeedDao.deleteThreatIntelDataIndex(datasource.getIndices()); + } catch (Exception e) { + if (previousState.equals(datasource.getState()) == false) { + datasource.setState(previousState); + datasourceDao.updateDatasource(datasource); + } + throw e; + } + datasourceDao.deleteDatasource(datasource); + } + +// private void setDatasourceStateAsDeleting(final Datasource datasource) { +// if (datasourceDao.getProcessors(datasource.getName()).isEmpty() == false) { +// throw new OpenSearchStatusException("datasource is being used by one of processors", RestStatus.BAD_REQUEST); +// } +// +// DatasourceState previousState = datasource.getState(); +// datasource.setState(DatasourceState.DELETING); +// datasourceDao.updateDatasource(datasource); +// +// // Check again as processor might just have been created. +// // If it fails to update the state back to the previous state, the new processor +// // will fail to convert an ip to a geo data. +// // In such case, user have to delete the processor and delete this datasource again. +// if (datasourceDao.getProcessors(datasource.getName()).isEmpty() == false) { +// datasource.setState(previousState); +// datasourceDao.updateDatasource(datasource); +// throw new OpenSearchStatusException("datasource is being used by one of processors", RestStatus.BAD_REQUEST); +// } +// } +} diff --git a/src/main/java/org/opensearch/securityanalytics/threatintel/action/GetDatasourceAction.java b/src/main/java/org/opensearch/securityanalytics/threatintel/action/GetDatasourceAction.java new file mode 100644 index 000000000..e7487226b --- /dev/null +++ b/src/main/java/org/opensearch/securityanalytics/threatintel/action/GetDatasourceAction.java @@ -0,0 +1,26 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.securityanalytics.threatintel.action; + +import org.opensearch.action.ActionType; + +/** + * Threat intel datasource get action + */ +public class GetDatasourceAction extends ActionType { + /** + * Get datasource action instance + */ + public static final GetDatasourceAction INSTANCE = new GetDatasourceAction(); + /** + * Get datasource action name + */ + public static final String NAME = "cluster:admin/security_analytics/datasource/get"; + + private GetDatasourceAction() { + super(NAME, GetDatasourceResponse::new); + } +} diff --git a/src/main/java/org/opensearch/securityanalytics/threatintel/action/GetDatasourceRequest.java b/src/main/java/org/opensearch/securityanalytics/threatintel/action/GetDatasourceRequest.java new file mode 100644 index 000000000..82db02414 --- /dev/null +++ b/src/main/java/org/opensearch/securityanalytics/threatintel/action/GetDatasourceRequest.java @@ -0,0 +1,66 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.securityanalytics.threatintel.action; + +import lombok.Getter; +import lombok.Setter; +import org.opensearch.action.ActionRequest; +import org.opensearch.action.ActionRequestValidationException; +import org.opensearch.core.common.io.stream.StreamInput; +import org.opensearch.core.common.io.stream.StreamOutput; + +import java.io.IOException; + +/** + * threat intel datasource get request + */ +@Getter +@Setter +public class GetDatasourceRequest extends ActionRequest { + /** + * @param names the datasource names + * @return the datasource names + */ + private String[] names; + + /** + * Constructs a new get datasource request with a list of datasources. + * + * If the list of datasources is empty or it contains a single element "_all", all registered datasources + * are returned. + * + * @param names list of datasource names + */ + public GetDatasourceRequest(final String[] names) { + this.names = names; + } + + /** + * Constructor with stream input + * @param in the stream input + * @throws IOException IOException + */ + public GetDatasourceRequest(final StreamInput in) throws IOException { + super(in); + this.names = in.readStringArray(); + } + + @Override + public ActionRequestValidationException validate() { + ActionRequestValidationException errors = null; + if (names == null) { + errors = new ActionRequestValidationException(); + errors.addValidationError("names should not be null"); + } + return errors; + } + + @Override + public void writeTo(final StreamOutput out) throws IOException { + super.writeTo(out); + out.writeStringArray(names); + } +} diff --git a/src/main/java/org/opensearch/securityanalytics/threatintel/action/GetDatasourceResponse.java b/src/main/java/org/opensearch/securityanalytics/threatintel/action/GetDatasourceResponse.java new file mode 100644 index 000000000..84c799c23 --- /dev/null +++ b/src/main/java/org/opensearch/securityanalytics/threatintel/action/GetDatasourceResponse.java @@ -0,0 +1,87 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.securityanalytics.threatintel.action; + +import lombok.EqualsAndHashCode; +import lombok.Getter; +import lombok.Setter; +import org.opensearch.core.ParseField; +import org.opensearch.core.action.ActionResponse; +import org.opensearch.core.common.io.stream.StreamInput; +import org.opensearch.core.common.io.stream.StreamOutput; +import org.opensearch.core.xcontent.ToXContentObject; +import org.opensearch.core.xcontent.XContentBuilder; +import org.opensearch.securityanalytics.threatintel.jobscheduler.Datasource; + +import java.io.IOException; +import java.time.Instant; +import java.util.List; + +/** + * threat intel datasource get request + */ +@Getter +@Setter +@EqualsAndHashCode(callSuper = false) +public class GetDatasourceResponse extends ActionResponse implements ToXContentObject { + private static final ParseField FIELD_NAME_DATASOURCES = new ParseField("datasources"); + private static final ParseField FIELD_NAME_NAME = new ParseField("name"); + private static final ParseField FIELD_NAME_STATE = new ParseField("state"); + private static final ParseField FIELD_NAME_ENDPOINT = new ParseField("endpoint"); + private static final ParseField FIELD_NAME_UPDATE_INTERVAL = new ParseField("update_interval_in_days"); + private static final ParseField FIELD_NAME_NEXT_UPDATE_AT = new ParseField("next_update_at_in_epoch_millis"); + private static final ParseField FIELD_NAME_NEXT_UPDATE_AT_READABLE = new ParseField("next_update_at"); + private static final ParseField FIELD_NAME_DATABASE = new ParseField("database"); + private static final ParseField FIELD_NAME_UPDATE_STATS = new ParseField("update_stats"); + private List datasources; + + /** + * Default constructor + * + * @param datasources List of datasources + */ + public GetDatasourceResponse(final List datasources) { + this.datasources = datasources; + } + + /** + * Constructor with StreamInput + * + * @param in the stream input + */ + public GetDatasourceResponse(final StreamInput in) throws IOException { + datasources = in.readList(Datasource::new); + } + + @Override + public void writeTo(final StreamOutput out) throws IOException { + out.writeList(datasources); + } + + @Override + public XContentBuilder toXContent(final XContentBuilder builder, final Params params) throws IOException { + builder.startObject(); + builder.startArray(FIELD_NAME_DATASOURCES.getPreferredName()); + for (Datasource datasource : datasources) { + builder.startObject(); + builder.field(FIELD_NAME_NAME.getPreferredName(), datasource.getName()); + builder.field(FIELD_NAME_STATE.getPreferredName(), datasource.getState()); + builder.field(FIELD_NAME_ENDPOINT.getPreferredName(), datasource.getEndpoint()); + builder.field(FIELD_NAME_UPDATE_INTERVAL.getPreferredName(), datasource.getSchedule()); //TODO + builder.timeField( + FIELD_NAME_NEXT_UPDATE_AT.getPreferredName(), + FIELD_NAME_NEXT_UPDATE_AT_READABLE.getPreferredName(), + datasource.getSchedule().getNextExecutionTime(Instant.now()).toEpochMilli() + ); + builder.field(FIELD_NAME_DATABASE.getPreferredName(), datasource.getDatabase()); + builder.field(FIELD_NAME_UPDATE_STATS.getPreferredName(), datasource.getUpdateStats()); + builder.endObject(); + } + builder.endArray(); + builder.endObject(); + return builder; + } +} diff --git a/src/main/java/org/opensearch/securityanalytics/threatintel/action/GetDatasourceTransportAction.java b/src/main/java/org/opensearch/securityanalytics/threatintel/action/GetDatasourceTransportAction.java new file mode 100644 index 000000000..1b1a3d9d7 --- /dev/null +++ b/src/main/java/org/opensearch/securityanalytics/threatintel/action/GetDatasourceTransportAction.java @@ -0,0 +1,79 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.securityanalytics.threatintel.action; + +import org.opensearch.OpenSearchException; +import org.opensearch.action.support.ActionFilters; +import org.opensearch.action.support.HandledTransportAction; +import org.opensearch.common.inject.Inject; +import org.opensearch.core.action.ActionListener; +import org.opensearch.index.IndexNotFoundException; +import org.opensearch.securityanalytics.threatintel.dao.DatasourceDao; +import org.opensearch.securityanalytics.threatintel.jobscheduler.Datasource; +import org.opensearch.tasks.Task; +import org.opensearch.transport.TransportService; + +import java.util.Collections; +import java.util.List; + +/** + * Transport action to get datasource + */ +public class GetDatasourceTransportAction extends HandledTransportAction { + private final DatasourceDao datasourceDao; + + /** + * Default constructor + * @param transportService the transport service + * @param actionFilters the action filters + * @param datasourceDao the datasource facade + */ + @Inject + public GetDatasourceTransportAction( + final TransportService transportService, + final ActionFilters actionFilters, + final DatasourceDao datasourceDao + ) { + super(GetDatasourceAction.NAME, transportService, actionFilters, GetDatasourceRequest::new); + this.datasourceDao = datasourceDao; + } + + @Override + protected void doExecute(final Task task, final GetDatasourceRequest request, final ActionListener listener) { + if (shouldGetAllDatasource(request)) { + // We don't expect too many data sources. Therefore, querying all data sources without pagination should be fine. + datasourceDao.getAllDatasources(newActionListener(listener)); + } else { + datasourceDao.getDatasources(request.getNames(), newActionListener(listener)); + } + } + + private boolean shouldGetAllDatasource(final GetDatasourceRequest request) { + if (request.getNames() == null) { + throw new OpenSearchException("names in a request should not be null"); + } + + return request.getNames().length == 0 || (request.getNames().length == 1 && "_all".equals(request.getNames()[0])); + } + + protected ActionListener> newActionListener(final ActionListener listener) { + return new ActionListener<>() { + @Override + public void onResponse(final List datasources) { + listener.onResponse(new GetDatasourceResponse(datasources)); + } + + @Override + public void onFailure(final Exception e) { + if (e instanceof IndexNotFoundException) { + listener.onResponse(new GetDatasourceResponse(Collections.emptyList())); + return; + } + listener.onFailure(e); + } + }; + } +} diff --git a/src/main/java/org/opensearch/securityanalytics/threatintel/action/PutDatasourceAction.java b/src/main/java/org/opensearch/securityanalytics/threatintel/action/PutDatasourceAction.java new file mode 100644 index 000000000..f111a0195 --- /dev/null +++ b/src/main/java/org/opensearch/securityanalytics/threatintel/action/PutDatasourceAction.java @@ -0,0 +1,27 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.securityanalytics.threatintel.action; + +import org.opensearch.action.ActionType; +import org.opensearch.action.support.master.AcknowledgedResponse; + +/** + * Threat intel datasource creation action + */ +public class PutDatasourceAction extends ActionType { + /** + * Put datasource action instance + */ + public static final PutDatasourceAction INSTANCE = new PutDatasourceAction(); + /** + * Put datasource action name + */ + public static final String NAME = "cluster:admin/security_analytics/datasource/put"; + + private PutDatasourceAction() { + super(NAME, AcknowledgedResponse::new); + } +} diff --git a/src/main/java/org/opensearch/securityanalytics/threatintel/action/PutDatasourceRequest.java b/src/main/java/org/opensearch/securityanalytics/threatintel/action/PutDatasourceRequest.java new file mode 100644 index 000000000..957275573 --- /dev/null +++ b/src/main/java/org/opensearch/securityanalytics/threatintel/action/PutDatasourceRequest.java @@ -0,0 +1,208 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.securityanalytics.threatintel.action; + +import java.io.IOException; +import java.net.MalformedURLException; +import java.net.URISyntaxException; +import java.net.URL; +import java.util.List; +import java.util.Locale; + +import lombok.Getter; +import lombok.Setter; + +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; +import org.opensearch.action.ActionRequest; +import org.opensearch.action.ActionRequestValidationException; +import org.opensearch.common.unit.TimeValue; +import org.opensearch.core.ParseField; +import org.opensearch.core.common.io.stream.StreamInput; +import org.opensearch.core.common.io.stream.StreamOutput; +import org.opensearch.core.xcontent.ObjectParser; +import org.opensearch.securityanalytics.model.DetectorTrigger; +import org.opensearch.securityanalytics.threatintel.common.DatasourceManifest; +import org.opensearch.securityanalytics.threatintel.common.ParameterValidator; + +/** + * Threat intel datasource creation request + */ +@Getter +@Setter +public class PutDatasourceRequest extends ActionRequest { + private static final Logger log = LogManager.getLogger(DetectorTrigger.class); + + public static final ParseField FEED_FORMAT_FIELD = new ParseField("feed_format"); + public static final ParseField ENDPOINT_FIELD = new ParseField("endpoint"); + public static final ParseField FEED_NAME_FIELD = new ParseField("feed_name"); + public static final ParseField DESCRIPTION_FIELD = new ParseField("description"); + public static final ParseField ORGANIZATION_FIELD = new ParseField("organization"); + public static final ParseField CONTAINED_IOCS_FIELD = new ParseField("contained_iocs_field"); + public static final ParseField UPDATE_INTERVAL_IN_DAYS_FIELD = new ParseField("update_interval_in_days"); + private static final ParameterValidator VALIDATOR = new ParameterValidator(); + + /** + * @param name the datasource name + * @return the datasource name + */ + private String name; + + private String feedFormat; + + /** + * @param endpoint url to a manifest file for a datasource + * @return url to a manifest file for a datasource + */ + private String endpoint; + + private String feedName; + + private String description; + + private String organization; + + private List contained_iocs_field; + + /** + * @param updateInterval update interval of a datasource + * @return update interval of a datasource + */ + private TimeValue updateInterval; + + /** + * Parser of a datasource + */ + public static final ObjectParser PARSER; + static { + PARSER = new ObjectParser<>("put_datasource"); + PARSER.declareString((request, val) -> request.setFeedFormat(val), FEED_FORMAT_FIELD); + PARSER.declareString((request, val) -> request.setEndpoint(val), ENDPOINT_FIELD); + PARSER.declareString((request, val) -> request.setFeedName(val), FEED_NAME_FIELD); + PARSER.declareString((request, val) -> request.setDescription(val), DESCRIPTION_FIELD); + PARSER.declareString((request, val) -> request.setOrganization(val), ORGANIZATION_FIELD); +// PARSER.declareStringArray((request, val[]) -> request.setContained_iocs_field(val), CONTAINED_IOCS_FIELD); + PARSER.declareLong((request, val) -> request.setUpdateInterval(TimeValue.timeValueDays(val)), UPDATE_INTERVAL_IN_DAYS_FIELD); + } + + /** + * Default constructor + * @param name name of a datasource + */ + public PutDatasourceRequest(final String name) { + this.name = name; + } + + /** + * Constructor with stream input + * @param in the stream input + * @throws IOException IOException + */ + public PutDatasourceRequest(final StreamInput in) throws IOException { + super(in); + this.name = in.readString(); + this.feedFormat = in.readString(); + this.endpoint = in.readString(); + this.feedName = in.readString(); + this.description = in.readString(); + this.organization = in.readString(); + this.contained_iocs_field = in.readStringList(); + this.updateInterval = in.readTimeValue(); + } + + @Override + public void writeTo(final StreamOutput out) throws IOException { + super.writeTo(out); + out.writeString(name); + out.writeString(feedFormat); + out.writeString(endpoint); + out.writeString(feedName); + out.writeString(description); + out.writeString(organization); + out.writeStringCollection(contained_iocs_field); + out.writeTimeValue(updateInterval); + } + + @Override + public ActionRequestValidationException validate() { + ActionRequestValidationException errors = new ActionRequestValidationException(); + List errorMsgs = VALIDATOR.validateDatasourceName(name); + if (errorMsgs.isEmpty() == false) { + errorMsgs.stream().forEach(msg -> errors.addValidationError(msg)); + } + validateEndpoint(errors); + validateUpdateInterval(errors); + return errors.validationErrors().isEmpty() ? null : errors; + } + + /** + * Conduct following validation on endpoint + * 1. endpoint format complies with RFC-2396 + * 2. validate manifest file from the endpoint + * + * @param errors the errors to add error messages + */ + private void validateEndpoint(final ActionRequestValidationException errors) { + try { + URL url = new URL(endpoint); + url.toURI(); // Validate URL complies with RFC-2396 + validateManifestFile(url, errors); + } catch (MalformedURLException | URISyntaxException e) { + log.info("Invalid URL[{}] is provided", endpoint, e); + errors.addValidationError("Invalid URL format is provided"); + } + } + + /** + * Conduct following validation on url + * 1. can read manifest file from the endpoint + * 2. the url in the manifest file complies with RFC-2396 + * 3. updateInterval is less than validForInDays value in the manifest file + * + * @param url the url to validate + * @param errors the errors to add error messages + */ + private void validateManifestFile(final URL url, final ActionRequestValidationException errors) { + DatasourceManifest manifest; + try { + manifest = DatasourceManifest.Builder.build(url); + } catch (Exception e) { + log.info("Error occurred while reading a file from {}", url, e); + errors.addValidationError(String.format(Locale.ROOT, "Error occurred while reading a file from %s: %s", url, e.getMessage())); + return; + } + + try { + new URL(manifest.getUrl()).toURI(); // Validate URL complies with RFC-2396 + } catch (MalformedURLException | URISyntaxException e) { + log.info("Invalid URL[{}] is provided for url field in the manifest file", manifest.getUrl(), e); + errors.addValidationError("Invalid URL format is provided for url field in the manifest file"); + return; + } + +// if (manifest.getValidForInDays() != null && updateInterval.days() >= manifest.getValidForInDays()) { +// errors.addValidationError( +// String.format( +// Locale.ROOT, +// "updateInterval %d should be smaller than %d", +// updateInterval.days(), +// manifest.getValidForInDays() +// ) +// ); +// } + } + + /** + * Validate updateInterval is equal or larger than 1 + * + * @param errors the errors to add error messages + */ + private void validateUpdateInterval(final ActionRequestValidationException errors) { + if (updateInterval.compareTo(TimeValue.timeValueDays(1)) < 0) { + errors.addValidationError("Update interval should be equal to or larger than 1 day"); + } + } +} diff --git a/src/main/java/org/opensearch/securityanalytics/threatintel/action/PutDatasourceTransportAction.java b/src/main/java/org/opensearch/securityanalytics/threatintel/action/PutDatasourceTransportAction.java new file mode 100644 index 000000000..6b8e72c0a --- /dev/null +++ b/src/main/java/org/opensearch/securityanalytics/threatintel/action/PutDatasourceTransportAction.java @@ -0,0 +1,182 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.securityanalytics.threatintel.action; + +import static org.opensearch.securityanalytics.threatintel.common.ThreatIntelLockService.LOCK_DURATION_IN_SECONDS; + +import java.time.Instant; +import java.util.ConcurrentModificationException; +import java.util.concurrent.atomic.AtomicReference; + +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; +import org.opensearch.ResourceAlreadyExistsException; +import org.opensearch.action.StepListener; +import org.opensearch.action.index.IndexResponse; +import org.opensearch.action.support.ActionFilters; +import org.opensearch.action.support.HandledTransportAction; +import org.opensearch.action.support.master.AcknowledgedResponse; +import org.opensearch.common.inject.Inject; +import org.opensearch.core.action.ActionListener; + +import org.opensearch.core.rest.RestStatus; +import org.opensearch.index.engine.VersionConflictEngineException; +import org.opensearch.jobscheduler.spi.LockModel; +import org.opensearch.securityanalytics.model.DetectorTrigger; +import org.opensearch.securityanalytics.threatintel.common.DatasourceState; +import org.opensearch.securityanalytics.threatintel.common.ThreatIntelLockService; +import org.opensearch.securityanalytics.threatintel.dao.DatasourceDao; +import org.opensearch.securityanalytics.threatintel.jobscheduler.Datasource; +import org.opensearch.securityanalytics.threatintel.jobscheduler.DatasourceUpdateService; +import org.opensearch.tasks.Task; +import org.opensearch.threadpool.ThreadPool; +import org.opensearch.transport.TransportService; + +/** + * Transport action to create datasource + */ +public class PutDatasourceTransportAction extends HandledTransportAction { + private static final Logger log = LogManager.getLogger(DetectorTrigger.class); + + private final ThreadPool threadPool; + private final DatasourceDao datasourceDao; + private final DatasourceUpdateService datasourceUpdateService; + private final ThreatIntelLockService lockService; + + /** + * Default constructor + * @param transportService the transport service + * @param actionFilters the action filters + * @param threadPool the thread pool + * @param datasourceDao the datasource facade + * @param datasourceUpdateService the datasource update service + * @param lockService the lock service + */ + @Inject + public PutDatasourceTransportAction( + final TransportService transportService, + final ActionFilters actionFilters, + final ThreadPool threadPool, + final DatasourceDao datasourceDao, + final DatasourceUpdateService datasourceUpdateService, + final ThreatIntelLockService lockService + ) { + super(PutDatasourceAction.NAME, transportService, actionFilters, PutDatasourceRequest::new); + this.threadPool = threadPool; + this.datasourceDao = datasourceDao; + this.datasourceUpdateService = datasourceUpdateService; + this.lockService = lockService; + } + + @Override + protected void doExecute(final Task task, final PutDatasourceRequest request, final ActionListener listener) { + lockService.acquireLock(request.getName(), LOCK_DURATION_IN_SECONDS, ActionListener.wrap(lock -> { + if (lock == null) { + listener.onFailure( + new ConcurrentModificationException("another processor is holding a lock on the resource. Try again later") + ); + log.error("another processor is a lock, BAD_REQUEST error", RestStatus.BAD_REQUEST); + return; + } + try { + internalDoExecute(request, lock, listener); + } catch (Exception e) { + lockService.releaseLock(lock); + listener.onFailure(e); + log.error("listener failed when executing", e); + } + }, exception -> { + listener.onFailure(exception); + log.error("execution failed", exception); + })); + } + + /** + * This method takes lock as a parameter and is responsible for releasing lock + * unless exception is thrown + */ + protected void internalDoExecute( + final PutDatasourceRequest request, + final LockModel lock, + final ActionListener listener + ) { + StepListener createIndexStep = new StepListener<>(); + datasourceDao.createIndexIfNotExists(createIndexStep); + createIndexStep.whenComplete(v -> { + Datasource datasource = Datasource.Builder.build(request); + datasourceDao.putDatasource(datasource, getIndexResponseListener(datasource, lock, listener)); + }, exception -> { + lockService.releaseLock(lock); + log.error("failed to release lock", exception); + listener.onFailure(exception); + }); + } + + /** + * This method takes lock as a parameter and is responsible for releasing lock + * unless exception is thrown + */ + protected ActionListener getIndexResponseListener( + final Datasource datasource, + final LockModel lock, + final ActionListener listener + ) { + return new ActionListener<>() { + @Override + public void onResponse(final IndexResponse indexResponse) { + // This is user initiated request. Therefore, we want to handle the first datasource update task in a generic thread + // pool. + threadPool.generic().submit(() -> { + AtomicReference lockReference = new AtomicReference<>(lock); + try { + createDatasource(datasource, lockService.getRenewLockRunnable(lockReference)); + } finally { + lockService.releaseLock(lockReference.get()); + } + }); + listener.onResponse(new AcknowledgedResponse(true)); + } + + @Override + public void onFailure(final Exception e) { + lockService.releaseLock(lock); + if (e instanceof VersionConflictEngineException) { + log.error("datasource already exists"); + listener.onFailure(new ResourceAlreadyExistsException("datasource [{}] already exists", datasource.getName())); + } else { + log.error("Internal server error"); + listener.onFailure(e); + } + } + }; + } + + protected void createDatasource(final Datasource datasource, final Runnable renewLock) { + if (DatasourceState.CREATING.equals(datasource.getState()) == false) { + log.error("Invalid datasource state. Expecting {} but received {}", DatasourceState.CREATING, datasource.getState()); + markDatasourceAsCreateFailed(datasource); + return; + } + + try { + datasourceUpdateService.updateOrCreateThreatIntelFeedData(datasource, renewLock); + } catch (Exception e) { + log.error("Failed to create datasource for {}", datasource.getName(), e); + markDatasourceAsCreateFailed(datasource); + } + } + + private void markDatasourceAsCreateFailed(final Datasource datasource) { + datasource.getUpdateStats().setLastFailedAt(Instant.now()); + datasource.setState(DatasourceState.CREATE_FAILED); + try { + datasourceDao.updateDatasource(datasource); + } catch (Exception e) { + log.error("Failed to mark datasource state as CREATE_FAILED for {}", datasource.getName(), e); + } + } +} + diff --git a/src/main/java/org/opensearch/securityanalytics/threatintel/action/RestDeleteDatasourceHandler.java b/src/main/java/org/opensearch/securityanalytics/threatintel/action/RestDeleteDatasourceHandler.java new file mode 100644 index 000000000..7b5555cc9 --- /dev/null +++ b/src/main/java/org/opensearch/securityanalytics/threatintel/action/RestDeleteDatasourceHandler.java @@ -0,0 +1,47 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.securityanalytics.threatintel.action; + +import org.opensearch.client.node.NodeClient; +import org.opensearch.rest.BaseRestHandler; +import org.opensearch.rest.RestRequest; +import org.opensearch.rest.action.RestToXContentListener; + +import java.util.List; +import java.util.Locale; + +import static org.opensearch.rest.RestRequest.Method.DELETE; + +/** + * Rest handler for threat intel datasource delete request + */ +public class RestDeleteDatasourceHandler extends BaseRestHandler { + private static final String ACTION_NAME = "threatintel_datasource_delete"; + private static final String PARAMS_NAME = "name"; + + @Override + public String getName() { + return ACTION_NAME; + } + + @Override + protected RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) { + final String name = request.param(PARAMS_NAME); + final DeleteDatasourceRequest deleteDatasourceRequest = new DeleteDatasourceRequest(name); + + return channel -> client.executeLocally( + DeleteDatasourceAction.INSTANCE, + deleteDatasourceRequest, + new RestToXContentListener<>(channel) + ); + } + + @Override + public List routes() { + String path = String.join("/", "/_plugins/_security_analytics", String.format(Locale.ROOT, "threatintel/datasource/{%s}", PARAMS_NAME)); + return List.of(new Route(DELETE, path)); + } +} diff --git a/src/main/java/org/opensearch/securityanalytics/threatintel/action/RestGetDatasourceHandler.java b/src/main/java/org/opensearch/securityanalytics/threatintel/action/RestGetDatasourceHandler.java new file mode 100644 index 000000000..dbb492f1e --- /dev/null +++ b/src/main/java/org/opensearch/securityanalytics/threatintel/action/RestGetDatasourceHandler.java @@ -0,0 +1,44 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.securityanalytics.threatintel.action; + +import org.opensearch.client.node.NodeClient; +import org.opensearch.core.common.Strings; +import org.opensearch.rest.BaseRestHandler; +import org.opensearch.rest.RestRequest; +import org.opensearch.rest.action.RestToXContentListener; + +import java.util.List; + +import static org.opensearch.rest.RestRequest.Method.GET; + +/** + * Rest handler for threat intel datasource get request + */ +public class RestGetDatasourceHandler extends BaseRestHandler { + private static final String ACTION_NAME = "threatintel_datasource_get"; + + @Override + public String getName() { + return ACTION_NAME; + } + + @Override + protected RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) { + final String[] names = request.paramAsStringArray("name", Strings.EMPTY_ARRAY); + final GetDatasourceRequest getDatasourceRequest = new GetDatasourceRequest(names); + + return channel -> client.executeLocally(GetDatasourceAction.INSTANCE, getDatasourceRequest, new RestToXContentListener<>(channel)); + } + + @Override + public List routes() { + return List.of( + new Route(GET, String.join("/", "/_plugins/_security_analytics", "threatintel/datasource")), + new Route(GET, String.join("/", "/_plugins/_security_analytics", "threatintel/datasource/{name}")) + ); + } +} diff --git a/src/main/java/org/opensearch/securityanalytics/threatintel/action/RestPutDatasourceHandler.java b/src/main/java/org/opensearch/securityanalytics/threatintel/action/RestPutDatasourceHandler.java new file mode 100644 index 000000000..d7b1e96d5 --- /dev/null +++ b/src/main/java/org/opensearch/securityanalytics/threatintel/action/RestPutDatasourceHandler.java @@ -0,0 +1,71 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.securityanalytics.threatintel.action; + +import org.opensearch.client.node.NodeClient; +import org.opensearch.common.settings.ClusterSettings; +import org.opensearch.common.unit.TimeValue; +import org.opensearch.core.xcontent.XContentParser; +import org.opensearch.rest.BaseRestHandler; +import org.opensearch.rest.RestRequest; +import org.opensearch.rest.action.RestToXContentListener; +import org.opensearch.securityanalytics.threatintel.common.ThreatIntelSettings; + +import java.io.IOException; +import java.util.List; + +import static org.opensearch.rest.RestRequest.Method.PUT; + +/** + * Rest handler for threat intel datasource creation + * + * This handler handles a request of + * PUT /_plugins/security_analytics/threatintel/datasource/{id} + * { + * "endpoint": {endpoint}, + * "update_interval_in_days": 3 + * } + * + * When request is received, it will create a datasource by downloading threat intel feed from the endpoint. + * After the creation of datasource is completed, it will schedule the next update task after update_interval_in_days. + * + */ +public class RestPutDatasourceHandler extends BaseRestHandler { + private static final String ACTION_NAME = "threatintel_datasource_put"; + private final ClusterSettings clusterSettings; + + public RestPutDatasourceHandler(final ClusterSettings clusterSettings) { + this.clusterSettings = clusterSettings; + } + + @Override + public String getName() { + return ACTION_NAME; + } + + @Override + protected RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException { + final PutDatasourceRequest putDatasourceRequest = new PutDatasourceRequest(request.param("name")); + if (request.hasContentOrSourceParam()) { + try (XContentParser parser = request.contentOrSourceParamParser()) { + PutDatasourceRequest.PARSER.parse(parser, putDatasourceRequest, null); + } + } + if (putDatasourceRequest.getEndpoint() == null) { + putDatasourceRequest.setEndpoint(clusterSettings.get(ThreatIntelSettings.DATASOURCE_ENDPOINT)); + } + if (putDatasourceRequest.getUpdateInterval() == null) { + putDatasourceRequest.setUpdateInterval(TimeValue.timeValueDays(clusterSettings.get(ThreatIntelSettings.DATASOURCE_UPDATE_INTERVAL))); + } + return channel -> client.executeLocally(PutDatasourceAction.INSTANCE, putDatasourceRequest, new RestToXContentListener<>(channel)); + } + + @Override + public List routes() { + String path = String.join("/", "/_plugins/_security_analytics", "threatintel/datasource/{name}"); + return List.of(new Route(PUT, path)); + } +} diff --git a/src/main/java/org/opensearch/securityanalytics/threatintel/action/RestUpdateDatasourceHandler.java b/src/main/java/org/opensearch/securityanalytics/threatintel/action/RestUpdateDatasourceHandler.java new file mode 100644 index 000000000..7d8e30438 --- /dev/null +++ b/src/main/java/org/opensearch/securityanalytics/threatintel/action/RestUpdateDatasourceHandler.java @@ -0,0 +1,50 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.securityanalytics.threatintel.action; + +import org.opensearch.client.node.NodeClient; +import org.opensearch.core.xcontent.XContentParser; +import org.opensearch.rest.BaseRestHandler; +import org.opensearch.rest.RestRequest; +import org.opensearch.rest.action.RestToXContentListener; + +import java.io.IOException; +import java.util.List; + +import static org.opensearch.rest.RestRequest.Method.PUT; + +/** + * Rest handler for threat intel datasource update request + */ +public class RestUpdateDatasourceHandler extends BaseRestHandler { + private static final String ACTION_NAME = "threatintel_datasource_update"; + + @Override + public String getName() { + return ACTION_NAME; + } + + @Override + protected RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException { + final UpdateDatasourceRequest updateDatasourceRequest = new UpdateDatasourceRequest(request.param("name")); + if (request.hasContentOrSourceParam()) { + try (XContentParser parser = request.contentOrSourceParamParser()) { + UpdateDatasourceRequest.PARSER.parse(parser, updateDatasourceRequest, null); + } + } + return channel -> client.executeLocally( + UpdateDatasourceAction.INSTANCE, + updateDatasourceRequest, + new RestToXContentListener<>(channel) + ); + } + + @Override + public List routes() { + String path = String.join("/", "/_plugins/_security_analytics", "threatintel/datasource/{name}/_settings"); + return List.of(new Route(PUT, path)); + } +} diff --git a/src/main/java/org/opensearch/securityanalytics/threatintel/action/UpdateDatasourceAction.java b/src/main/java/org/opensearch/securityanalytics/threatintel/action/UpdateDatasourceAction.java new file mode 100644 index 000000000..4d2066b92 --- /dev/null +++ b/src/main/java/org/opensearch/securityanalytics/threatintel/action/UpdateDatasourceAction.java @@ -0,0 +1,27 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.securityanalytics.threatintel.action; + +import org.opensearch.action.ActionType; +import org.opensearch.action.support.master.AcknowledgedResponse; + +/** + * Ip2Geo datasource update action + */ +public class UpdateDatasourceAction extends ActionType { + /** + * Update datasource action instance + */ + public static final UpdateDatasourceAction INSTANCE = new UpdateDatasourceAction(); + /** + * Update datasource action name + */ + public static final String NAME = "cluster:admin/geospatial/datasource/update"; + + private UpdateDatasourceAction() { + super(NAME, AcknowledgedResponse::new); + } +} diff --git a/src/main/java/org/opensearch/securityanalytics/threatintel/action/UpdateDatasourceRequest.java b/src/main/java/org/opensearch/securityanalytics/threatintel/action/UpdateDatasourceRequest.java new file mode 100644 index 000000000..f6e159ea4 --- /dev/null +++ b/src/main/java/org/opensearch/securityanalytics/threatintel/action/UpdateDatasourceRequest.java @@ -0,0 +1,176 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.securityanalytics.threatintel.action; + +import lombok.EqualsAndHashCode; +import lombok.Getter; +import lombok.Setter; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; +import org.opensearch.action.ActionRequest; +import org.opensearch.action.ActionRequestValidationException; +import org.opensearch.common.unit.TimeValue; +import org.opensearch.core.ParseField; +import org.opensearch.core.common.io.stream.StreamInput; +import org.opensearch.core.common.io.stream.StreamOutput; +import org.opensearch.core.xcontent.ObjectParser; +import org.opensearch.securityanalytics.model.DetectorTrigger; +import org.opensearch.securityanalytics.threatintel.common.DatasourceManifest; +import org.opensearch.securityanalytics.threatintel.common.ParameterValidator; + +import java.io.IOException; +import java.net.MalformedURLException; +import java.net.URISyntaxException; +import java.net.URL; +import java.util.Locale; + +/** + * threat intel datasource update request + */ +@Getter +@Setter +@EqualsAndHashCode(callSuper = false) +public class UpdateDatasourceRequest extends ActionRequest { + private static final Logger log = LogManager.getLogger(DetectorTrigger.class); + + public static final ParseField ENDPOINT_FIELD = new ParseField("endpoint"); + public static final ParseField UPDATE_INTERVAL_IN_DAYS_FIELD = new ParseField("update_interval_in_days"); + private static final int MAX_DATASOURCE_NAME_BYTES = 255; + private static final ParameterValidator VALIDATOR = new ParameterValidator(); + + /** + * @param name the datasource name + * @return the datasource name + */ + private String name; + /** + * @param endpoint url to a manifest file for a datasource + * @return url to a manifest file for a datasource + */ + private String endpoint; + /** + * @param updateInterval update interval of a datasource + * @return update interval of a datasource + */ + private TimeValue updateInterval; + + /** + * Parser of a datasource + */ + public static final ObjectParser PARSER; + static { + PARSER = new ObjectParser<>("update_datasource"); + PARSER.declareString((request, val) -> request.setEndpoint(val), ENDPOINT_FIELD); + PARSER.declareLong((request, val) -> request.setUpdateInterval(TimeValue.timeValueDays(val)), UPDATE_INTERVAL_IN_DAYS_FIELD); + } + + /** + * Constructor + * @param name name of a datasource + */ + public UpdateDatasourceRequest(final String name) { + this.name = name; + } + + /** + * Constructor + * @param in the stream input + * @throws IOException IOException + */ + public UpdateDatasourceRequest(final StreamInput in) throws IOException { + super(in); + this.name = in.readString(); + this.endpoint = in.readOptionalString(); + this.updateInterval = in.readOptionalTimeValue(); + } + + @Override + public void writeTo(final StreamOutput out) throws IOException { + super.writeTo(out); + out.writeString(name); + out.writeOptionalString(endpoint); + out.writeOptionalTimeValue(updateInterval); + } + + @Override + public ActionRequestValidationException validate() { + ActionRequestValidationException errors = new ActionRequestValidationException(); + if (VALIDATOR.validateDatasourceName(name).isEmpty() == false) { + errors.addValidationError("no such datasource exist"); + } + if (endpoint == null && updateInterval == null) { + errors.addValidationError("no values to update"); + } + + validateEndpoint(errors); + validateUpdateInterval(errors); + + return errors.validationErrors().isEmpty() ? null : errors; + } + + /** + * Conduct following validation on endpoint + * 1. endpoint format complies with RFC-2396 + * 2. validate manifest file from the endpoint + * + * @param errors the errors to add error messages + */ + private void validateEndpoint(final ActionRequestValidationException errors) { + if (endpoint == null) { + return; + } + + try { + URL url = new URL(endpoint); + url.toURI(); // Validate URL complies with RFC-2396 + validateManifestFile(url, errors); + } catch (MalformedURLException | URISyntaxException e) { + log.info("Invalid URL[{}] is provided", endpoint, e); + errors.addValidationError("Invalid URL format is provided"); + } + } + + /** + * Conduct following validation on url + * 1. can read manifest file from the endpoint + * 2. the url in the manifest file complies with RFC-2396 + * + * @param url the url to validate + * @param errors the errors to add error messages + */ + private void validateManifestFile(final URL url, final ActionRequestValidationException errors) { + DatasourceManifest manifest; + try { + manifest = DatasourceManifest.Builder.build(url); + } catch (Exception e) { + log.info("Error occurred while reading a file from {}", url, e); + errors.addValidationError(String.format(Locale.ROOT, "Error occurred while reading a file from %s: %s", url, e.getMessage())); + return; + } + + try { + new URL(manifest.getUrl()).toURI(); // Validate URL complies with RFC-2396 + } catch (MalformedURLException | URISyntaxException e) { + log.info("Invalid URL[{}] is provided for url field in the manifest file", manifest.getUrl(), e); + errors.addValidationError("Invalid URL format is provided for url field in the manifest file"); + } + } + + /** + * Validate updateInterval is equal or larger than 1 + * + * @param errors the errors to add error messages + */ + private void validateUpdateInterval(final ActionRequestValidationException errors) { + if (updateInterval == null) { + return; + } + + if (updateInterval.compareTo(TimeValue.timeValueDays(1)) < 0) { + errors.addValidationError("Update interval should be equal to or larger than 1 day"); + } + } +} diff --git a/src/main/java/org/opensearch/securityanalytics/threatintel/action/UpdateDatasourceTransportAction.java b/src/main/java/org/opensearch/securityanalytics/threatintel/action/UpdateDatasourceTransportAction.java new file mode 100644 index 000000000..0e4eb3288 --- /dev/null +++ b/src/main/java/org/opensearch/securityanalytics/threatintel/action/UpdateDatasourceTransportAction.java @@ -0,0 +1,179 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.securityanalytics.threatintel.action; + +import org.opensearch.OpenSearchStatusException; +import org.opensearch.ResourceNotFoundException; +import org.opensearch.action.support.ActionFilters; +import org.opensearch.action.support.HandledTransportAction; +import org.opensearch.action.support.master.AcknowledgedResponse; +import org.opensearch.common.inject.Inject; +import org.opensearch.core.action.ActionListener; +import org.opensearch.core.rest.RestStatus; +import org.opensearch.jobscheduler.spi.schedule.IntervalSchedule; +import org.opensearch.securityanalytics.threatintel.common.DatasourceState; +import org.opensearch.securityanalytics.threatintel.common.ThreatIntelLockService; +import org.opensearch.securityanalytics.threatintel.dao.DatasourceDao; +import org.opensearch.securityanalytics.threatintel.jobscheduler.Datasource; +import org.opensearch.securityanalytics.threatintel.jobscheduler.DatasourceTask; +import org.opensearch.securityanalytics.threatintel.jobscheduler.DatasourceUpdateService; +import org.opensearch.tasks.Task; +import org.opensearch.threadpool.ThreadPool; +import org.opensearch.transport.TransportService; + +import java.io.IOException; +import java.time.Instant; +import java.time.temporal.ChronoUnit; +import java.util.List; +import java.util.Locale; + +/** + * Transport action to update datasource + */ +public class UpdateDatasourceTransportAction extends HandledTransportAction { + private static final long LOCK_DURATION_IN_SECONDS = 300l; + private final ThreatIntelLockService lockService; + private final DatasourceDao datasourceDao; + private final DatasourceUpdateService datasourceUpdateService; + private final ThreadPool threadPool; + + /** + * Constructor + * + * @param transportService the transport service + * @param actionFilters the action filters + * @param lockService the lock service + * @param datasourceDao the datasource facade + * @param datasourceUpdateService the datasource update service + */ + @Inject + public UpdateDatasourceTransportAction( + final TransportService transportService, + final ActionFilters actionFilters, + final ThreatIntelLockService lockService, + final DatasourceDao datasourceDao, + final DatasourceUpdateService datasourceUpdateService, + final ThreadPool threadPool + ) { + super(UpdateDatasourceAction.NAME, transportService, actionFilters, UpdateDatasourceRequest::new); + this.lockService = lockService; + this.datasourceUpdateService = datasourceUpdateService; + this.datasourceDao = datasourceDao; + this.threadPool = threadPool; + } + + /** + * Get a lock and update datasource + * + * @param task the task + * @param request the request + * @param listener the listener + */ + @Override + protected void doExecute(final Task task, final UpdateDatasourceRequest request, final ActionListener listener) { + lockService.acquireLock(request.getName(), LOCK_DURATION_IN_SECONDS, ActionListener.wrap(lock -> { + if (lock == null) { + listener.onFailure( + new OpenSearchStatusException("Another processor is holding a lock on the resource. Try again later", RestStatus.BAD_REQUEST) + ); + return; + } + try { + // TODO: makes every sub-methods as async call to avoid using a thread in generic pool + threadPool.generic().submit(() -> { + try { + Datasource datasource = datasourceDao.getDatasource(request.getName()); + if (datasource == null) { + throw new ResourceNotFoundException("no such datasource exist"); + } + if (DatasourceState.AVAILABLE.equals(datasource.getState()) == false) { + throw new IllegalArgumentException( + String.format(Locale.ROOT, "data source is not in an [%s] state", DatasourceState.AVAILABLE) + ); + } + validate(request, datasource); + updateIfChanged(request, datasource); + lockService.releaseLock(lock); + listener.onResponse(new AcknowledgedResponse(true)); + } catch (Exception e) { + lockService.releaseLock(lock); + listener.onFailure(e); + } + }); + } catch (Exception e) { + lockService.releaseLock(lock); + listener.onFailure(e); + } + }, exception -> listener.onFailure(exception))); + } + + private void updateIfChanged(final UpdateDatasourceRequest request, final Datasource datasource) { + boolean isChanged = false; + if (isEndpointChanged(request, datasource)) { + datasource.setEndpoint(request.getEndpoint()); + isChanged = true; + } + if (isUpdateIntervalChanged(request)) { + datasource.setSchedule(new IntervalSchedule(Instant.now(), (int) request.getUpdateInterval().getDays(), ChronoUnit.DAYS)); + datasource.setTask(DatasourceTask.ALL); + isChanged = true; + } + + if (isChanged) { + datasourceDao.updateDatasource(datasource); + } + } + + /** + * Additional validation based on an existing datasource + * + * Basic validation is done in UpdateDatasourceRequest#validate + * In this method we do additional validation based on an existing datasource + * + * 1. Check the compatibility of new fields and old fields + * 2. Check the updateInterval is less than validForInDays in datasource + * + * This method throws exception if one of validation fails. + * + * @param request the update request + * @param datasource the existing datasource + * @throws IOException the exception + */ + private void validate(final UpdateDatasourceRequest request, final Datasource datasource) throws IOException { + validateFieldsCompatibility(request, datasource); + } + + private void validateFieldsCompatibility(final UpdateDatasourceRequest request, final Datasource datasource) throws IOException { + if (isEndpointChanged(request, datasource) == false) { + return; + } + + List fields = datasourceUpdateService.getHeaderFields(request.getEndpoint()); + if (datasource.isCompatible(fields) == false) { +// throw new IncompatibleDatasourceException( +// "new fields [{}] does not contain all old fields [{}]", +// fields.toString(), +// datasource.getDatabase().getFields().toString() +// ); + throw new OpenSearchStatusException("new fields does not contain all old fields", RestStatus.BAD_REQUEST); + } + } + + private boolean isEndpointChanged(final UpdateDatasourceRequest request, final Datasource datasource) { + return request.getEndpoint() != null && request.getEndpoint().equals(datasource.getEndpoint()) == false; + } + + /** + * Update interval is changed as long as user provide one because + * start time will get updated even if the update interval is same as current one. + * + * @param request the update datasource request + * @return true if update interval is changed, and false otherwise + */ + private boolean isUpdateIntervalChanged(final UpdateDatasourceRequest request) { + return request.getUpdateInterval() != null; + } +} diff --git a/src/main/java/org/opensearch/securityanalytics/threatintel/common/DatasourceManifest.java b/src/main/java/org/opensearch/securityanalytics/threatintel/common/DatasourceManifest.java index cd6b4b565..8cc4cfd36 100644 --- a/src/main/java/org/opensearch/securityanalytics/threatintel/common/DatasourceManifest.java +++ b/src/main/java/org/opensearch/securityanalytics/threatintel/common/DatasourceManifest.java @@ -1,4 +1,151 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ package org.opensearch.securityanalytics.threatintel.common; +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStreamReader; +import java.net.URL; +import java.net.URLConnection; +import java.nio.CharBuffer; +import java.security.AccessController; +import java.security.PrivilegedAction; + +import lombok.AllArgsConstructor; +import lombok.Getter; +import lombok.Setter; + +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; +import org.opensearch.SpecialPermission; +import org.opensearch.common.SuppressForbidden; +import org.opensearch.common.xcontent.json.JsonXContent; +import org.opensearch.core.ParseField; +import org.opensearch.core.rest.RestStatus; +import org.opensearch.core.xcontent.ConstructingObjectParser; +import org.opensearch.core.xcontent.DeprecationHandler; +import org.opensearch.core.xcontent.NamedXContentRegistry; +import org.opensearch.core.xcontent.XContentParser; +import org.opensearch.securityanalytics.model.DetectorTrigger; +import org.opensearch.securityanalytics.util.SecurityAnalyticsException; + +/** + * Threat intel datasource manifest file object + * + * Manifest file is stored in an external endpoint. OpenSearch read the file and store values it in this object. + */ +@Setter +@Getter +@AllArgsConstructor public class DatasourceManifest { + private static final Logger log = LogManager.getLogger(DetectorTrigger.class); + + private static final ParseField URL_FIELD = new ParseField("url"); + private static final ParseField DB_NAME_FIELD = new ParseField("db_name"); + private static final ParseField SHA256_HASH_FIELD = new ParseField("sha256_hash"); + private static final ParseField ORGANIZATION_FIELD = new ParseField("organization"); + private static final ParseField DESCRIPTION_FIELD = new ParseField("description"); + private static final ParseField UPDATED_AT_FIELD = new ParseField("updated_at_in_epoch_milli"); + + /** + * @param url URL of a ZIP file containing a database + * @return URL of a ZIP file containing a database + */ + private String url; + /** + * @param dbName A database file name inside the ZIP file + * @return A database file name inside the ZIP file + */ + private String dbName; + /** + * @param sha256Hash SHA256 hash value of a database file + * @return SHA256 hash value of a database file + */ + private String sha256Hash; + /** + * @param organization A database organization name + * @return A database organization name + */ + private String organization; + /** + * @param description A description of the database + * @return A description of a database + */ + private String description; + /** + * @param updatedAt A date when the database was updated + * @return A date when the database was updated + */ + private Long updatedAt; + + /** + * Datasource manifest parser + */ + public static final ConstructingObjectParser PARSER = new ConstructingObjectParser<>( + "datasource_manifest", + true, + args -> { + String url = (String) args[0]; + String dbName = (String) args[1]; + String sha256Hash = (String) args[2]; + String organization = (String) args[4]; + String description = (String) args[5]; + Long updatedAt = (Long) args[3]; + return new DatasourceManifest(url, dbName, sha256Hash, organization, description, updatedAt); + } + ); + static { + PARSER.declareString(ConstructingObjectParser.constructorArg(), URL_FIELD); + PARSER.declareString(ConstructingObjectParser.constructorArg(), DB_NAME_FIELD); + PARSER.declareString(ConstructingObjectParser.constructorArg(), SHA256_HASH_FIELD); + PARSER.declareString(ConstructingObjectParser.constructorArg(), ORGANIZATION_FIELD); + PARSER.declareString(ConstructingObjectParser.constructorArg(), DESCRIPTION_FIELD); + PARSER.declareLong(ConstructingObjectParser.constructorArg(), UPDATED_AT_FIELD); + } + + /** + * Datasource manifest builder + */ + public static class Builder { + private static final int MANIFEST_FILE_MAX_BYTES = 1024 * 8; //check this + + /** + * Build DatasourceManifest from a given url + * + * @param url url to downloads a manifest file + * @return DatasourceManifest representing the manifest file + */ + @SuppressForbidden(reason = "Need to connect to http endpoint to read manifest file") + public static DatasourceManifest build(final URL url) { + SpecialPermission.check(); + return AccessController.doPrivileged((PrivilegedAction) () -> { + try { + URLConnection connection = url.openConnection(); + return internalBuild(connection); + } catch (IOException e) { + log.error("Runtime exception", e); + throw new SecurityAnalyticsException("Runtime exception", RestStatus.INTERNAL_SERVER_ERROR, e); //TODO + } + }); + } + + @SuppressForbidden(reason = "Need to connect to http endpoint to read manifest file") + protected static DatasourceManifest internalBuild(final URLConnection connection) throws IOException { +// connection.addRequestProperty(Constants.USER_AGENT_KEY, Constants.USER_AGENT_VALUE); + InputStreamReader inputStreamReader = new InputStreamReader(connection.getInputStream()); + try (BufferedReader reader = new BufferedReader(inputStreamReader)) { + CharBuffer charBuffer = CharBuffer.allocate(MANIFEST_FILE_MAX_BYTES); + reader.read(charBuffer); + charBuffer.flip(); + XContentParser parser = JsonXContent.jsonXContent.createParser( + NamedXContentRegistry.EMPTY, + DeprecationHandler.IGNORE_DEPRECATIONS, + charBuffer.toString() + ); + return PARSER.parse(parser, null); + } + } + } } diff --git a/src/main/java/org/opensearch/securityanalytics/threatintel/common/DatasourceState.java b/src/main/java/org/opensearch/securityanalytics/threatintel/common/DatasourceState.java index eb8c7b9ca..f5de3861f 100644 --- a/src/main/java/org/opensearch/securityanalytics/threatintel/common/DatasourceState.java +++ b/src/main/java/org/opensearch/securityanalytics/threatintel/common/DatasourceState.java @@ -8,9 +8,9 @@ /** * Threat intel datasource state * - * When data source is created, it starts with CREATING state. Once the first threatIP data is generated, the state changes to AVAILABLE. - * Only when the first threatIP data generation failed, the state changes to CREATE_FAILED. - * Subsequent threatIP data failure won't change data source state from AVAILABLE to CREATE_FAILED. + * When data source is created, it starts with CREATING state. Once the first threat intel feed is generated, the state changes to AVAILABLE. + * Only when the first threat intel feed generation failed, the state changes to CREATE_FAILED. + * Subsequent threat intel feed failure won't change data source state from AVAILABLE to CREATE_FAILED. * When delete request is received, the data source state changes to DELETING. * * State changed from left to right for the entire lifecycle of a datasource diff --git a/src/main/java/org/opensearch/securityanalytics/threatintel/common/ParameterValidator.java b/src/main/java/org/opensearch/securityanalytics/threatintel/common/ParameterValidator.java new file mode 100644 index 000000000..51acfece4 --- /dev/null +++ b/src/main/java/org/opensearch/securityanalytics/threatintel/common/ParameterValidator.java @@ -0,0 +1,58 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.securityanalytics.threatintel.common; + +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.List; +import java.util.Locale; + +import org.apache.commons.lang3.StringUtils; +import org.opensearch.core.common.Strings; + +/** + * Parameter validator for TIF APIs + */ +public class ParameterValidator { + private static final int MAX_DATASOURCE_NAME_BYTES = 127; + + /** + * Validate datasource name and return list of error messages + * + * @param datasourceName datasource name + * @return Error messages. Empty list if there is no violation. + */ + public List validateDatasourceName(final String datasourceName) { + List errorMsgs = new ArrayList<>(); + if (StringUtils.isBlank(datasourceName)) { + errorMsgs.add("datasource name must not be empty"); + return errorMsgs; + } + + if (!Strings.validFileName(datasourceName)) { + errorMsgs.add( + String.format(Locale.ROOT, "datasource name must not contain the following characters %s", Strings.INVALID_FILENAME_CHARS) + ); + } + if (datasourceName.contains("#")) { + errorMsgs.add("datasource name must not contain '#'"); + } + if (datasourceName.contains(":")) { + errorMsgs.add("datasource name must not contain ':'"); + } + if (datasourceName.charAt(0) == '_' || datasourceName.charAt(0) == '-' || datasourceName.charAt(0) == '+') { + errorMsgs.add("datasource name must not start with '_', '-', or '+'"); + } + int byteCount = datasourceName.getBytes(StandardCharsets.UTF_8).length; + if (byteCount > MAX_DATASOURCE_NAME_BYTES) { + errorMsgs.add(String.format(Locale.ROOT, "datasource name is too long, (%d > %d)", byteCount, MAX_DATASOURCE_NAME_BYTES)); + } + if (datasourceName.equals(".") || datasourceName.equals("..")) { + errorMsgs.add("datasource name must not be '.' or '..'"); + } + return errorMsgs; + } +} diff --git a/src/main/java/org/opensearch/securityanalytics/threatintel/common/ThreatIntelExecutor.java b/src/main/java/org/opensearch/securityanalytics/threatintel/common/ThreatIntelExecutor.java index 7da2cbdae..7a887fc6d 100644 --- a/src/main/java/org/opensearch/securityanalytics/threatintel/common/ThreatIntelExecutor.java +++ b/src/main/java/org/opensearch/securityanalytics/threatintel/common/ThreatIntelExecutor.java @@ -16,7 +16,7 @@ * Provide a list of static methods related with executors for threat intel */ public class ThreatIntelExecutor { - private static final String THREAD_POOL_NAME = "_plugin_securityanalytics_threatintel_datasource_update"; + private static final String THREAD_POOL_NAME = "plugin_sap_datasource_update"; private final ThreadPool threadPool; public ThreatIntelExecutor(final ThreadPool threadPool) { diff --git a/src/main/java/org/opensearch/securityanalytics/threatintel/common/ThreatIntelLockService.java b/src/main/java/org/opensearch/securityanalytics/threatintel/common/ThreatIntelLockService.java index a66de589a..03bbcde02 100644 --- a/src/main/java/org/opensearch/securityanalytics/threatintel/common/ThreatIntelLockService.java +++ b/src/main/java/org/opensearch/securityanalytics/threatintel/common/ThreatIntelLockService.java @@ -51,7 +51,7 @@ public ThreatIntelLockService(final ClusterService clusterService, final Client /** * Wrapper method of LockService#acquireLockWithId * - * Datasource use its name as doc id in job scheduler. Therefore, we can use datasource name to acquire + * Datasource uses its name as doc id in job scheduler. Therefore, we can use datasource name to acquire * a lock on a datasource. * * @param datasourceName datasourceName to acquire lock on @@ -83,13 +83,15 @@ public void onResponse(final LockModel lockModel) { public void onFailure(final Exception e) { lockReference.set(null); countDownLatch.countDown(); + log.error("aquiring lock failed", e); } }); try { - countDownLatch.await(clusterService.getClusterSettings().get(ThreatIntelSettings.TIMEOUT).getSeconds(), TimeUnit.SECONDS); + countDownLatch.await(clusterService.getClusterSettings().get(ThreatIntelSettings.THREAT_INTEL_TIMEOUT).getSeconds(), TimeUnit.SECONDS); return Optional.ofNullable(lockReference.get()); } catch (InterruptedException e) { + log.error("Waiting for the count down latch failed", e); return Optional.empty(); } } @@ -124,15 +126,17 @@ public void onResponse(final LockModel lockModel) { @Override public void onFailure(final Exception e) { + log.error("failed to renew lock", e); lockReference.set(null); countDownLatch.countDown(); } }); try { - countDownLatch.await(clusterService.getClusterSettings().get(ThreatIntelSettings.TIMEOUT).getSeconds(), TimeUnit.SECONDS); + countDownLatch.await(clusterService.getClusterSettings().get(ThreatIntelSettings.THREAT_INTEL_TIMEOUT).getSeconds(), TimeUnit.SECONDS); return lockReference.get(); } catch (InterruptedException e) { + log.error("Interrupted exception", e); return null; } } @@ -155,6 +159,7 @@ public Runnable getRenewLockRunnable(final AtomicReference lockModel) } lockModel.set(renewLock(lockModel.get())); if (lockModel.get() == null) { + log.error("Exception: failed to renew a lock"); new OpenSearchException("failed to renew a lock [{}]", preLock); } }; diff --git a/src/main/java/org/opensearch/securityanalytics/threatintel/common/ThreatIntelSettings.java b/src/main/java/org/opensearch/securityanalytics/threatintel/common/ThreatIntelSettings.java index 2480f8518..e997af730 100644 --- a/src/main/java/org/opensearch/securityanalytics/threatintel/common/ThreatIntelSettings.java +++ b/src/main/java/org/opensearch/securityanalytics/threatintel/common/ThreatIntelSettings.java @@ -10,89 +10,94 @@ import java.net.URL; import java.util.List; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; import org.opensearch.common.settings.Setting; import org.opensearch.common.unit.TimeValue; +import org.opensearch.securityanalytics.model.DetectorTrigger; /** - * Settings for Ip2Geo datasource operations + * Settings for threat intel datasource operations */ public class ThreatIntelSettings { + private static final Logger log = LogManager.getLogger(DetectorTrigger.class); + + + /** + * Default endpoint to be used in threat intel feed datasource creation API + */ + public static final Setting DATASOURCE_ENDPOINT = Setting.simpleString( + "plugins.security_analytics.threatintel.datasource.endpoint", + "https://geoip.maps.opensearch.org/v1/geolite2-city/manifest.json", //TODO fix this endpoint + new DatasourceEndpointValidator(), + Setting.Property.NodeScope, + Setting.Property.Dynamic + ); -// /** -// * Default endpoint to be used in threatIP datasource creation API -// */ -// public static final Setting DATASOURCE_ENDPOINT = Setting.simpleString( -// "plugins.security_analytics.threatintel.datasource.endpoint", -// "https://geoip.maps.opensearch.org/v1/geolite2-city/manifest.json", -// new DatasourceEndpointValidator(), -// Setting.Property.NodeScope, -// Setting.Property.Dynamic -// ); -// -// /** -// * Default update interval to be used in Ip2Geo datasource creation API -// */ -// public static final Setting DATASOURCE_UPDATE_INTERVAL = Setting.longSetting( -// "plugins.security_analytics.threatintel.datasource.update_interval_in_days", -// 3l, -// 1l, -// Setting.Property.NodeScope, -// Setting.Property.Dynamic -// ); -// -// /** -// * Bulk size for indexing GeoIP data -// */ -// public static final Setting BATCH_SIZE = Setting.intSetting( -// "plugins.security_analytics.threatintel.datasource.batch_size", -// 10000, -// 1, -// Setting.Property.NodeScope, -// Setting.Property.Dynamic -// ); -// /** - * Timeout value for Ip2Geo processor + * Default update interval to be used in threat intel datasource creation API */ - public static final Setting TIMEOUT = Setting.timeSetting( - "plugins.security_analytics.index_timeout", + public static final Setting DATASOURCE_UPDATE_INTERVAL = Setting.longSetting( + "plugins.security_analytics.threatintel.datasource.update_interval_in_days", + 3l, + 1l, + Setting.Property.NodeScope, + Setting.Property.Dynamic + ); + + /** + * Bulk size for indexing threat intel feed data + */ + public static final Setting BATCH_SIZE = Setting.intSetting( + "plugins.security_analytics.threatintel.datasource.batch_size", + 10000, + 1, + Setting.Property.NodeScope, + Setting.Property.Dynamic + ); + + /** + * Timeout value for threat intel processor + */ + public static final Setting THREAT_INTEL_TIMEOUT = Setting.timeSetting( + "plugins.security_analytics.threat_intel_timeout", TimeValue.timeValueSeconds(30), TimeValue.timeValueSeconds(1), Setting.Property.NodeScope, Setting.Property.Dynamic ); -// /** -// * Max size for geo data cache -// */ -// public static final Setting CACHE_SIZE = Setting.longSetting( -// "plugins.security_analytics.threatintel.processor.cache_size", -// 1000, -// 0, -// Setting.Property.NodeScope, -// Setting.Property.Dynamic -// ); + /** + * Max size for threat intel feed cache + */ + public static final Setting CACHE_SIZE = Setting.longSetting( + "plugins.security_analytics.threatintel.processor.cache_size", + 1000, + 0, + Setting.Property.NodeScope, + Setting.Property.Dynamic + ); /** - * Return all settings of threatIntel feature - * @return a list of all settings for threatIntel feature + * Return all settings of threat intel feature + * @return a list of all settings for threat intel feature */ public static final List> settings() { -// return List.of(DATASOURCE_ENDPOINT, DATASOURCE_UPDATE_INTERVAL, BATCH_SIZE, TIMEOUT, CACHE_SIZE); - return List.of(TIMEOUT); + return List.of(DATASOURCE_ENDPOINT, DATASOURCE_UPDATE_INTERVAL, BATCH_SIZE, THREAT_INTEL_TIMEOUT); } -// /** -// * Visible for testing -// */ -// protected static class DatasourceEndpointValidator implements Setting.Validator { -// @Override -// public void validate(final String value) { -// try { -// new URL(value).toURI(); -// } catch (MalformedURLException | URISyntaxException e) { -// throw new IllegalArgumentException("Invalid URL format is provided"); -// } -// } -// } + /** + * Visible for testing + */ + protected static class DatasourceEndpointValidator implements Setting.Validator { + @Override + public void validate(final String value) { + try { + new URL(value).toURI(); + } catch (MalformedURLException | URISyntaxException e) { + log.error("Invalid URL format is provided", e); + throw new IllegalArgumentException("Invalid URL format is provided"); + } + } + } } diff --git a/src/main/java/org/opensearch/securityanalytics/threatintel/dao/DatasourceDao.java b/src/main/java/org/opensearch/securityanalytics/threatintel/dao/DatasourceDao.java index 7a712636d..5ec565df3 100644 --- a/src/main/java/org/opensearch/securityanalytics/threatintel/dao/DatasourceDao.java +++ b/src/main/java/org/opensearch/securityanalytics/threatintel/dao/DatasourceDao.java @@ -1,38 +1,67 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + package org.opensearch.securityanalytics.threatintel.dao; +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.nio.charset.StandardCharsets; +import java.time.Instant; +import java.util.Arrays; +import java.util.List; +import java.util.Objects; +import java.util.stream.Collectors; + +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; +import org.opensearch.OpenSearchException; +import org.opensearch.ResourceAlreadyExistsException; +import org.opensearch.ResourceNotFoundException; import org.opensearch.action.DocWriteRequest; +import org.opensearch.action.StepListener; +import org.opensearch.action.admin.indices.create.CreateIndexRequest; +import org.opensearch.action.admin.indices.create.CreateIndexResponse; +import org.opensearch.action.bulk.BulkRequest; +import org.opensearch.action.bulk.BulkResponse; +import org.opensearch.action.delete.DeleteResponse; import org.opensearch.action.get.GetRequest; import org.opensearch.action.get.GetResponse; +import org.opensearch.action.get.MultiGetItemResponse; +import org.opensearch.action.get.MultiGetResponse; +import org.opensearch.action.index.IndexRequest; import org.opensearch.action.index.IndexResponse; +import org.opensearch.action.search.SearchResponse; import org.opensearch.action.support.WriteRequest; import org.opensearch.client.Client; +import org.opensearch.cluster.routing.Preference; import org.opensearch.cluster.service.ClusterService; import org.opensearch.common.settings.ClusterSettings; import org.opensearch.common.xcontent.LoggingDeprecationHandler; import org.opensearch.common.xcontent.XContentFactory; import org.opensearch.common.xcontent.XContentHelper; - +import org.opensearch.core.action.ActionListener; +import org.opensearch.core.common.bytes.BytesReference; +import org.opensearch.core.rest.RestStatus; import org.opensearch.core.xcontent.NamedXContentRegistry; import org.opensearch.core.xcontent.ToXContent; import org.opensearch.core.xcontent.XContentParser; -import org.opensearch.index.IndexNotFoundException; import org.opensearch.securityanalytics.model.DetectorTrigger; -import org.opensearch.securityanalytics.threatintel.common.StashedThreadContext; import org.opensearch.securityanalytics.threatintel.common.ThreatIntelSettings; -import org.opensearch.securityanalytics.threatintel.jobscheduler.DatasourceExtension; - -import org.apache.logging.log4j.LogManager; -import org.apache.logging.log4j.Logger; import org.opensearch.securityanalytics.threatintel.jobscheduler.Datasource; +import org.opensearch.securityanalytics.threatintel.jobscheduler.DatasourceExtension; +import org.opensearch.securityanalytics.threatintel.common.StashedThreadContext; +import org.opensearch.index.IndexNotFoundException; +import org.opensearch.index.query.QueryBuilders; +import org.opensearch.search.SearchHit; +import org.opensearch.securityanalytics.util.SecurityAnalyticsException; -import java.io.BufferedReader; -import java.io.IOException; -import java.io.InputStream; -import java.io.InputStreamReader; -import java.nio.charset.StandardCharsets; -import java.time.Instant; -import java.util.stream.Collectors; - +/** + * Data access object for datasource + */ public class DatasourceDao { private static final Logger log = LogManager.getLogger(DetectorTrigger.class); @@ -47,36 +76,35 @@ public DatasourceDao(final Client client, final ClusterService clusterService) { this.clusterSettings = clusterService.getClusterSettings(); } -// /** -// * Create datasource index -// * -// * @param stepListener setup listener -// */ -// public void createIndexIfNotExists(final StepListener stepListener) { -// if (clusterService.state().metadata().hasIndex(DatasourceExtension.JOB_INDEX_NAME) == true) { -// stepListener.onResponse(null); -// return; -// } -// final CreateIndexRequest createIndexRequest = new CreateIndexRequest(DatasourceExtension.JOB_INDEX_NAME).mapping(getIndexMapping()) -// .settings(DatasourceExtension.INDEX_SETTING); -// -// StashedThreadContext.run(client, () -> client.admin().indices().create(createIndexRequest, new ActionListener<>() { -// @Override -// public void onResponse(final CreateIndexResponse createIndexResponse) { -// stepListener.onResponse(null); -// } -// -// @Override -// public void onFailure(final Exception e) { -// if (e instanceof ResourceAlreadyExistsException) { -// log.info("index[{}] already exist", DatasourceExtension.JOB_INDEX_NAME); -// stepListener.onResponse(null); -// return; -// } -// stepListener.onFailure(e); -// } -// })); -// } + /** + * Create datasource index + * + * @param stepListener setup listener + */ + public void createIndexIfNotExists(final StepListener stepListener) { + if (clusterService.state().metadata().hasIndex(DatasourceExtension.JOB_INDEX_NAME) == true) { + stepListener.onResponse(null); + return; + } + final CreateIndexRequest createIndexRequest = new CreateIndexRequest(DatasourceExtension.JOB_INDEX_NAME).mapping(getIndexMapping()) + .settings(DatasourceExtension.INDEX_SETTING); + StashedThreadContext.run(client, () -> client.admin().indices().create(createIndexRequest, new ActionListener<>() { + @Override + public void onResponse(final CreateIndexResponse createIndexResponse) { + stepListener.onResponse(null); + } + + @Override + public void onFailure(final Exception e) { + if (e instanceof ResourceAlreadyExistsException) { + log.info("index[{}] already exist", DatasourceExtension.JOB_INDEX_NAME); + stepListener.onResponse(null); + return; + } + stepListener.onFailure(e); + } + })); + } private String getIndexMapping() { try { @@ -86,7 +114,103 @@ private String getIndexMapping() { } } } catch (IOException e) { - throw new RuntimeException(e); + log.error("Runtime exception", e); + throw new SecurityAnalyticsException("Runtime exception", RestStatus.INTERNAL_SERVER_ERROR, e); //TODO + } + } + + /** + * Update datasource in an index {@code DatasourceExtension.JOB_INDEX_NAME} + * @param datasource the datasource + * @return index response + */ + public IndexResponse updateDatasource(final Datasource datasource) { + datasource.setLastUpdateTime(Instant.now()); + return StashedThreadContext.run(client, () -> { + try { + return client.prepareIndex(DatasourceExtension.JOB_INDEX_NAME) + .setId(datasource.getName()) + .setOpType(DocWriteRequest.OpType.INDEX) + .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE) + .setSource(datasource.toXContent(XContentFactory.jsonBuilder(), ToXContent.EMPTY_PARAMS)) + .execute() + .actionGet(clusterSettings.get(ThreatIntelSettings.THREAT_INTEL_TIMEOUT)); + } catch (IOException e) { + throw new SecurityAnalyticsException("Runtime exception", RestStatus.INTERNAL_SERVER_ERROR, e); //TODO + } + }); + } + + /** + * Update datasources in an index {@code DatasourceExtension.JOB_INDEX_NAME} + * @param datasources the datasources + * @param listener action listener + */ + public void updateDatasource(final List datasources, final ActionListener listener) { + BulkRequest bulkRequest = new BulkRequest(); + datasources.stream().map(datasource -> { + datasource.setLastUpdateTime(Instant.now()); + return datasource; + }).map(this::toIndexRequest).forEach(indexRequest -> bulkRequest.add(indexRequest)); + StashedThreadContext.run(client, () -> client.bulk(bulkRequest, listener)); + } + + private IndexRequest toIndexRequest(Datasource datasource) { + try { + IndexRequest indexRequest = new IndexRequest(); + indexRequest.index(DatasourceExtension.JOB_INDEX_NAME); + indexRequest.id(datasource.getName()); + indexRequest.opType(DocWriteRequest.OpType.INDEX); + indexRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE); + indexRequest.source(datasource.toXContent(XContentFactory.jsonBuilder(), ToXContent.EMPTY_PARAMS)); + return indexRequest; + } catch (IOException e) { + throw new SecurityAnalyticsException("Runtime exception", RestStatus.INTERNAL_SERVER_ERROR, e); //TODO + } + } + + /** + * Put datasource in an index {@code DatasourceExtension.JOB_INDEX_NAME} + * + * @param datasource the datasource + * @param listener the listener + */ + public void putDatasource(final Datasource datasource, final ActionListener listener) { + datasource.setLastUpdateTime(Instant.now()); + StashedThreadContext.run(client, () -> { + try { + client.prepareIndex(DatasourceExtension.JOB_INDEX_NAME) + .setId(datasource.getName()) + .setOpType(DocWriteRequest.OpType.CREATE) + .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE) + .setSource(datasource.toXContent(XContentFactory.jsonBuilder(), ToXContent.EMPTY_PARAMS)) + .execute(listener); + } catch (IOException e) { + throw new SecurityAnalyticsException("Runtime exception", RestStatus.INTERNAL_SERVER_ERROR, e); //TODO + } + }); + } + + /** + * Delete datasource in an index {@code DatasourceExtension.JOB_INDEX_NAME} + * + * @param datasource the datasource + * + */ + public void deleteDatasource(final Datasource datasource) { + DeleteResponse response = client.prepareDelete() + .setIndex(DatasourceExtension.JOB_INDEX_NAME) + .setId(datasource.getName()) + .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE) + .execute() + .actionGet(clusterSettings.get(ThreatIntelSettings.THREAT_INTEL_TIMEOUT)); + + if (response.status().equals(RestStatus.OK)) { + log.info("deleted datasource[{}] successfully", datasource.getName()); + } else if (response.status().equals(RestStatus.NOT_FOUND)) { + throw new ResourceNotFoundException("datasource[{}] does not exist", datasource.getName()); + } else { + throw new OpenSearchException("failed to delete datasource[{}] with status[{}]", datasource.getName(), response.status()); } } @@ -100,7 +224,7 @@ public Datasource getDatasource(final String name) throws IOException { GetRequest request = new GetRequest(DatasourceExtension.JOB_INDEX_NAME, name); GetResponse response; try { - response = StashedThreadContext.run(client, () -> client.get(request).actionGet(clusterSettings.get(ThreatIntelSettings.TIMEOUT))); + response = StashedThreadContext.run(client, () -> client.get(request).actionGet(clusterSettings.get(ThreatIntelSettings.THREAT_INTEL_TIMEOUT))); if (response.isExists() == false) { log.error("Datasource[{}] does not exist in an index[{}]", name, DatasourceExtension.JOB_INDEX_NAME); return null; @@ -119,25 +243,138 @@ public Datasource getDatasource(final String name) throws IOException { } /** - * Update datasource in an index {@code DatasourceExtension.JOB_INDEX_NAME} - * @param datasource the datasource - * @return index response + * Get datasource from an index {@code DatasourceExtension.JOB_INDEX_NAME} + * @param name the name of a datasource + * @param actionListener the action listener */ - public IndexResponse updateDatasource(final Datasource datasource) { - datasource.setLastUpdateTime(Instant.now()); - return StashedThreadContext.run(client, () -> { - try { - return client.prepareIndex(DatasourceExtension.JOB_INDEX_NAME) - .setId(datasource.getName()) - .setOpType(DocWriteRequest.OpType.INDEX) - .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE) - .setSource(datasource.toXContent(XContentFactory.jsonBuilder(), ToXContent.EMPTY_PARAMS)) + public void getDatasource(final String name, final ActionListener actionListener) { + GetRequest request = new GetRequest(DatasourceExtension.JOB_INDEX_NAME, name); + StashedThreadContext.run(client, () -> client.get(request, new ActionListener<>() { + @Override + public void onResponse(final GetResponse response) { + if (response.isExists() == false) { + actionListener.onResponse(null); + return; + } + + try { + XContentParser parser = XContentHelper.createParser( + NamedXContentRegistry.EMPTY, + LoggingDeprecationHandler.INSTANCE, + response.getSourceAsBytesRef() + ); + actionListener.onResponse(Datasource.PARSER.parse(parser, null)); + } catch (IOException e) { + actionListener.onFailure(e); + } + } + + @Override + public void onFailure(final Exception e) { + actionListener.onFailure(e); + } + })); + } + + /** + * Get datasources from an index {@code DatasourceExtension.JOB_INDEX_NAME} + * @param names the array of datasource names + * @param actionListener the action listener + */ + public void getDatasources(final String[] names, final ActionListener> actionListener) { + StashedThreadContext.run( + client, + () -> client.prepareMultiGet() + .add(DatasourceExtension.JOB_INDEX_NAME, names) + .execute(createGetDataSourceQueryActionLister(MultiGetResponse.class, actionListener)) + ); + } + + /** + * Get all datasources up to {@code MAX_SIZE} from an index {@code DatasourceExtension.JOB_INDEX_NAME} + * @param actionListener the action listener + */ + public void getAllDatasources(final ActionListener> actionListener) { + StashedThreadContext.run( + client, + () -> client.prepareSearch(DatasourceExtension.JOB_INDEX_NAME) + .setQuery(QueryBuilders.matchAllQuery()) + .setPreference(Preference.PRIMARY.type()) + .setSize(MAX_SIZE) + .execute(createGetDataSourceQueryActionLister(SearchResponse.class, actionListener)) + ); + } + + /** + * Get all datasources up to {@code MAX_SIZE} from an index {@code DatasourceExtension.JOB_INDEX_NAME} + */ + public List getAllDatasources() { + SearchResponse response = StashedThreadContext.run( + client, + () -> client.prepareSearch(DatasourceExtension.JOB_INDEX_NAME) + .setQuery(QueryBuilders.matchAllQuery()) + .setPreference(Preference.PRIMARY.type()) + .setSize(MAX_SIZE) .execute() - .actionGet(clusterSettings.get(ThreatIntelSettings.TIMEOUT)); - } catch (IOException e) { - throw new RuntimeException(e); + .actionGet(clusterSettings.get(ThreatIntelSettings.THREAT_INTEL_TIMEOUT)) + ); + + List bytesReferences = toBytesReferences(response); + return bytesReferences.stream().map(bytesRef -> toDatasource(bytesRef)).collect(Collectors.toList()); + } + + private ActionListener createGetDataSourceQueryActionLister( + final Class response, + final ActionListener> actionListener + ) { + return new ActionListener() { + @Override + public void onResponse(final T response) { + try { + List bytesReferences = toBytesReferences(response); + List datasources = bytesReferences.stream() + .map(bytesRef -> toDatasource(bytesRef)) + .collect(Collectors.toList()); + actionListener.onResponse(datasources); + } catch (Exception e) { + actionListener.onFailure(e); + } } - }); + + @Override + public void onFailure(final Exception e) { + actionListener.onFailure(e); + } + }; + } + + private List toBytesReferences(final Object response) { + if (response instanceof SearchResponse) { + SearchResponse searchResponse = (SearchResponse) response; + return Arrays.stream(searchResponse.getHits().getHits()).map(SearchHit::getSourceRef).collect(Collectors.toList()); + } else if (response instanceof MultiGetResponse) { + MultiGetResponse multiGetResponse = (MultiGetResponse) response; + return Arrays.stream(multiGetResponse.getResponses()) + .map(MultiGetItemResponse::getResponse) + .filter(Objects::nonNull) + .filter(GetResponse::isExists) + .map(GetResponse::getSourceAsBytesRef) + .collect(Collectors.toList()); + } else { + throw new OpenSearchException("No supported instance type[{}] is provided", response.getClass()); + } } + private Datasource toDatasource(final BytesReference bytesReference) { + try { + XContentParser parser = XContentHelper.createParser( + NamedXContentRegistry.EMPTY, + LoggingDeprecationHandler.INSTANCE, + bytesReference + ); + return Datasource.PARSER.parse(parser, null); + } catch (IOException e) { + throw new SecurityAnalyticsException("Runtime exception", RestStatus.INTERNAL_SERVER_ERROR, e); //TODO + } + } } diff --git a/src/main/java/org/opensearch/securityanalytics/threatintel/dao/ThreatIntelFeedDao.java b/src/main/java/org/opensearch/securityanalytics/threatintel/dao/ThreatIntelFeedDao.java new file mode 100644 index 000000000..f17cf5dda --- /dev/null +++ b/src/main/java/org/opensearch/securityanalytics/threatintel/dao/ThreatIntelFeedDao.java @@ -0,0 +1,351 @@ +/* + * Copyright OpenSearch Contributors + * SPDX-License-Identifier: Apache-2.0 + */ + +package org.opensearch.securityanalytics.threatintel.dao; + +import static org.opensearch.securityanalytics.threatintel.jobscheduler.Datasource.THREAT_INTEL_DATA_INDEX_NAME_PREFIX; + +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.net.URL; +import java.net.URLConnection; +import java.nio.charset.StandardCharsets; +import java.security.AccessController; +import java.security.PrivilegedAction; +import java.util.Arrays; +import java.util.Collections; +import java.util.Iterator; +import java.util.LinkedList; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Optional; +import java.util.Queue; +import java.util.stream.Collectors; +import java.util.zip.ZipEntry; +import java.util.zip.ZipInputStream; + +import lombok.NonNull; + +import org.apache.commons.csv.CSVFormat; +import org.apache.commons.csv.CSVParser; +import org.apache.commons.csv.CSVRecord; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; +import org.apache.logging.log4j.util.Strings; +import org.opensearch.OpenSearchException; +import org.opensearch.SpecialPermission; +import org.opensearch.action.DocWriteRequest; +import org.opensearch.action.admin.indices.create.CreateIndexRequest; +import org.opensearch.action.bulk.BulkRequest; +import org.opensearch.action.bulk.BulkResponse; +import org.opensearch.action.index.IndexRequest; +import org.opensearch.action.search.SearchResponse; +import org.opensearch.action.support.IndicesOptions; +import org.opensearch.action.support.master.AcknowledgedResponse; +import org.opensearch.client.Client; +import org.opensearch.client.Requests; +import org.opensearch.cluster.routing.Preference; +import org.opensearch.cluster.service.ClusterService; +import org.opensearch.common.SuppressForbidden; +import org.opensearch.common.settings.ClusterSettings; +import org.opensearch.common.unit.TimeValue; +import org.opensearch.common.xcontent.XContentFactory; +import org.opensearch.common.xcontent.XContentHelper; +import org.opensearch.common.xcontent.XContentType; +import org.opensearch.core.rest.RestStatus; +import org.opensearch.core.xcontent.XContentBuilder; +import org.opensearch.securityanalytics.model.DetectorTrigger; +import org.opensearch.securityanalytics.threatintel.common.DatasourceManifest; +import org.opensearch.securityanalytics.threatintel.common.ThreatIntelSettings; + +import org.opensearch.securityanalytics.threatintel.common.StashedThreadContext; +import org.opensearch.index.query.QueryBuilders; +import org.opensearch.securityanalytics.util.SecurityAnalyticsException; + +/** + * Data access object for threat intel feed data + */ +public class ThreatIntelFeedDao { + private static final Logger log = LogManager.getLogger(DetectorTrigger.class); + + private static final String IP_RANGE_FIELD_NAME = "_cidr"; + private static final String DATA_FIELD_NAME = "_data"; + private static final Map INDEX_SETTING_TO_CREATE = Map.of( + "index.number_of_shards", + 1, + "index.number_of_replicas", + 0, + "index.refresh_interval", + -1, + "index.hidden", + true + ); + private static final Map INDEX_SETTING_TO_FREEZE = Map.of( + "index.auto_expand_replicas", + "0-all", + "index.blocks.write", + true + ); + private final ClusterService clusterService; + private final ClusterSettings clusterSettings; + private final Client client; + + public ThreatIntelFeedDao(final ClusterService clusterService, final Client client) { + this.clusterService = clusterService; + this.clusterSettings = clusterService.getClusterSettings(); + this.client = client; + } + + /** + * Create an index for TIF data + * + * Index setting start with single shard, zero replica, no refresh interval, and hidden. + * Once the TIF data is indexed, do refresh and force merge. + * Then, change the index setting to expand replica to all nodes, and read only allow delete. + * See {@link #freezeIndex} + * + * @param indexName index name + */ + public void createIndexIfNotExists(final String indexName) { + if (clusterService.state().metadata().hasIndex(indexName) == true) { + return; + } + final CreateIndexRequest createIndexRequest = new CreateIndexRequest(indexName).settings(INDEX_SETTING_TO_CREATE) + .mapping(getIndexMapping()); + StashedThreadContext.run( + client, + () -> client.admin().indices().create(createIndexRequest).actionGet(clusterSettings.get(ThreatIntelSettings.THREAT_INTEL_TIMEOUT)) + ); + } + + private void freezeIndex(final String indexName) { + TimeValue timeout = clusterSettings.get(ThreatIntelSettings.THREAT_INTEL_TIMEOUT); + StashedThreadContext.run(client, () -> { + client.admin().indices().prepareForceMerge(indexName).setMaxNumSegments(1).execute().actionGet(timeout); + client.admin().indices().prepareRefresh(indexName).execute().actionGet(timeout); + client.admin() + .indices() + .prepareUpdateSettings(indexName) + .setSettings(INDEX_SETTING_TO_FREEZE) + .execute() + .actionGet(clusterSettings.get(ThreatIntelSettings.THREAT_INTEL_TIMEOUT)); + }); + } + + /** + * Generate XContentBuilder representing datasource database index mapping + * + * { + * "dynamic": false, + * "properties": { + * "_cidr": { + * "type": "ip_range", + * "doc_values": false + * } + * } + * } + * + * @return String representing datasource database index mapping + */ + private String getIndexMapping() { + try { + try (InputStream is = DatasourceDao.class.getResourceAsStream("/mappings/threat_intel_feed_mapping.json")) { + try (BufferedReader reader = new BufferedReader(new InputStreamReader(is, StandardCharsets.UTF_8))) { + return reader.lines().map(String::trim).collect(Collectors.joining()); + } + } + } catch (IOException e) { + log.error("Runtime exception", e); + throw new SecurityAnalyticsException("Runtime exception", RestStatus.INTERNAL_SERVER_ERROR, e); //TODO + } + } + + /** + * Create CSVParser of a threat intel feed + * + * @param manifest Datasource manifest + * @return CSVParser for threat intel feed + */ + @SuppressForbidden(reason = "Need to connect to http endpoint to read threat intel feed database file") + public CSVParser getDatabaseReader(final DatasourceManifest manifest) { + SpecialPermission.check(); + return AccessController.doPrivileged((PrivilegedAction) () -> { + try { + URL zipUrl = new URL(manifest.getUrl()); + return internalGetDatabaseReader(manifest, zipUrl.openConnection()); + } catch (IOException e) { + log.error("Exception: failed to read threat intel feed data from {}",manifest.getUrl(), e); + throw new OpenSearchException("failed to read threat intel feed data from {}", manifest.getUrl(), e); + } + }); + } + + @SuppressForbidden(reason = "Need to connect to http endpoint to read threat intel feed database file") + protected CSVParser internalGetDatabaseReader(final DatasourceManifest manifest, final URLConnection connection) throws IOException { +// connection.addRequestProperty(Constants.USER_AGENT_KEY, Constants.USER_AGENT_VALUE); + ZipInputStream zipIn = new ZipInputStream(connection.getInputStream()); + ZipEntry zipEntry = zipIn.getNextEntry(); + while (zipEntry != null) { + if (zipEntry.getName().equalsIgnoreCase(manifest.getDbName()) == false) { + zipEntry = zipIn.getNextEntry(); + continue; + } + return new CSVParser(new BufferedReader(new InputStreamReader(zipIn)), CSVFormat.RFC4180); + } + throw new IllegalArgumentException( + String.format(Locale.ROOT, "database file [%s] does not exist in the zip file [%s]", manifest.getDbName(), manifest.getUrl()) + ); + } + + /** + * Create a document to ingest in datasource database index + * + * It assumes the first field as ip_range. The rest is added under data field. + * + * Document example + * { + * "_cidr":"1.0.0.1/25", + * "_data":{ + * "country": "USA", + * "city": "Seattle", + * "location":"13.23,42.12" + * } + * } + * + * @param fields a list of field name + * @param values a list of values + * @return Document in json string format + * @throws IOException the exception + */ + public XContentBuilder createDocument(final String[] fields, final String[] values) throws IOException { + if (fields.length != values.length) { + throw new OpenSearchException("header[{}] and record[{}] length does not match", fields, values); + } + XContentBuilder builder = XContentFactory.jsonBuilder(); + builder.startObject(); + builder.field(IP_RANGE_FIELD_NAME, values[0]); + builder.startObject(DATA_FIELD_NAME); + for (int i = 1; i < fields.length; i++) { + if (Strings.isBlank(values[i])) { + continue; + } + builder.field(fields[i], values[i]); + } + builder.endObject(); + builder.endObject(); + builder.close(); + return builder; + } + + /** + * Query a given index using a given ip address to get TIF data + * + * @param indexName index + * @param ip ip address + * @return TIF data + */ + public Map getTIFData(final String indexName, final String ip) { + SearchResponse response = StashedThreadContext.run( + client, + () -> client.prepareSearch(indexName) + .setSize(1) + .setQuery(QueryBuilders.termQuery(IP_RANGE_FIELD_NAME, ip)) + .setPreference(Preference.LOCAL.type()) + .setRequestCache(true) + .get(clusterSettings.get(ThreatIntelSettings.THREAT_INTEL_TIMEOUT)) + ); + + if (response.getHits().getHits().length == 0) { + return Collections.emptyMap(); + } else { + return (Map) XContentHelper.convertToMap(response.getHits().getAt(0).getSourceRef(), false, XContentType.JSON) + .v2() + .get(DATA_FIELD_NAME); + } + } + + /** + * Puts TIF data from CSVRecord iterator into a given index in bulk + * + * @param indexName Index name to puts the TIF data + * @param fields Field name matching with data in CSVRecord in order + * @param iterator TIF data to insert + * @param renewLock Runnable to renew lock + */ + public void putTIFData( + @NonNull final String indexName, + @NonNull final String[] fields, + @NonNull final Iterator iterator, + @NonNull final Runnable renewLock + ) throws IOException { + TimeValue timeout = clusterSettings.get(ThreatIntelSettings.THREAT_INTEL_TIMEOUT); + Integer batchSize = clusterSettings.get(ThreatIntelSettings.BATCH_SIZE); + final BulkRequest bulkRequest = new BulkRequest(); + Queue requests = new LinkedList<>(); + for (int i = 0; i < batchSize; i++) { + requests.add(Requests.indexRequest(indexName)); + } + while (iterator.hasNext()) { + CSVRecord record = iterator.next(); + XContentBuilder document = createDocument(fields, record.values()); + IndexRequest indexRequest = (IndexRequest) requests.poll(); + indexRequest.source(document); + indexRequest.id(record.get(0)); + bulkRequest.add(indexRequest); + if (iterator.hasNext() == false || bulkRequest.requests().size() == batchSize) { + BulkResponse response = StashedThreadContext.run(client, () -> client.bulk(bulkRequest).actionGet(timeout)); + if (response.hasFailures()) { + throw new OpenSearchException( + "error occurred while ingesting threat intel feed data in {} with an error {}", + indexName, + response.buildFailureMessage() + ); + } + requests.addAll(bulkRequest.requests()); + bulkRequest.requests().clear(); + } + renewLock.run(); + } + freezeIndex(indexName); + } + + public void deleteThreatIntelDataIndex(final String index) { + deleteThreatIntelDataIndex(Arrays.asList(index)); + } + + public void deleteThreatIntelDataIndex(final List indices) { + if (indices == null || indices.isEmpty()) { + return; + } + + Optional invalidIndex = indices.stream() + .filter(index -> index.startsWith(THREAT_INTEL_DATA_INDEX_NAME_PREFIX) == false) + .findAny(); + if (invalidIndex.isPresent()) { + throw new OpenSearchException( + "the index[{}] is not threat intel data index which should start with {}", + invalidIndex.get(), + THREAT_INTEL_DATA_INDEX_NAME_PREFIX + ); + } + + AcknowledgedResponse response = StashedThreadContext.run( + client, + () -> client.admin() + .indices() + .prepareDelete(indices.toArray(new String[0])) + .setIndicesOptions(IndicesOptions.LENIENT_EXPAND_OPEN_CLOSED_HIDDEN) + .execute() + .actionGet(clusterSettings.get(ThreatIntelSettings.THREAT_INTEL_TIMEOUT)) + ); + + if (response.isAcknowledged() == false) { + throw new OpenSearchException("failed to delete data[{}] in datasource", String.join(",", indices)); + } + } +} diff --git a/src/main/java/org/opensearch/securityanalytics/threatintel/dao/ThreatIpDataDao.java b/src/main/java/org/opensearch/securityanalytics/threatintel/dao/ThreatIpDataDao.java deleted file mode 100644 index c6fcb0465..000000000 --- a/src/main/java/org/opensearch/securityanalytics/threatintel/dao/ThreatIpDataDao.java +++ /dev/null @@ -1,4 +0,0 @@ -package org.opensearch.securityanalytics.threatintel.dao; - -public class ThreatIpDataDao { -} diff --git a/src/main/java/org/opensearch/securityanalytics/threatintel/jobscheduler/Datasource.java b/src/main/java/org/opensearch/securityanalytics/threatintel/jobscheduler/Datasource.java index a95a392ab..9558a29f4 100644 --- a/src/main/java/org/opensearch/securityanalytics/threatintel/jobscheduler/Datasource.java +++ b/src/main/java/org/opensearch/securityanalytics/threatintel/jobscheduler/Datasource.java @@ -8,33 +8,28 @@ */ package org.opensearch.securityanalytics.threatintel.jobscheduler; +import lombok.*; import org.opensearch.core.ParseField; import org.opensearch.core.common.io.stream.StreamInput; import org.opensearch.core.common.io.stream.StreamOutput; import org.opensearch.core.common.io.stream.Writeable; import org.opensearch.core.xcontent.ConstructingObjectParser; +import org.opensearch.core.xcontent.ToXContent; import org.opensearch.jobscheduler.spi.ScheduledJobParameter; import org.opensearch.jobscheduler.spi.schedule.IntervalSchedule; import org.opensearch.jobscheduler.spi.schedule.Schedule; import org.opensearch.core.xcontent.XContentBuilder; import org.opensearch.jobscheduler.spi.schedule.ScheduleParser; -import lombok.AccessLevel; -import lombok.AllArgsConstructor; -import lombok.EqualsAndHashCode; -import lombok.Getter; -import lombok.Setter; -import lombok.ToString; - import java.io.IOException; import java.time.Instant; import java.time.temporal.ChronoUnit; -import java.util.ArrayList; -import java.util.List; -import java.util.Locale; +import java.util.*; import static org.opensearch.common.time.DateUtils.toInstant; +import org.opensearch.securityanalytics.threatintel.action.PutDatasourceRequest; +import org.opensearch.securityanalytics.threatintel.common.DatasourceManifest; import org.opensearch.securityanalytics.threatintel.common.DatasourceState; import org.opensearch.securityanalytics.threatintel.common.ThreatIntelLockService; @@ -47,7 +42,7 @@ public class Datasource implements Writeable, ScheduledJobParameter { /** * Prefix of indices having threatIntel data */ - public static final String THREATINTEL_DATA_INDEX_NAME_PREFIX = ".security_analytics-threatintel"; //.opensearch-sap-log-types-config + public static final String THREAT_INTEL_DATA_INDEX_NAME_PREFIX = ".opensearch-sap-threat-intel-config"; /** * Default fields for job scheduling @@ -67,16 +62,18 @@ public class Datasource implements Writeable, ScheduledJobParameter { /** * Additional fields for datasource */ + private static final ParseField FEED_NAME = new ParseField("feed_name"); + private static final ParseField FEED_FORMAT = new ParseField("feed_format"); private static final ParseField ENDPOINT_FIELD = new ParseField("endpoint"); + private static final ParseField DESCRIPTION = new ParseField("description"); + private static final ParseField ORGANIZATION = new ParseField("organization"); + private static final ParseField CONTAINED_IOCS_FIELD = new ParseField("contained_iocs_field"); private static final ParseField STATE_FIELD = new ParseField("state"); private static final ParseField CURRENT_INDEX_FIELD = new ParseField("current_index"); private static final ParseField INDICES_FIELD = new ParseField("indices"); private static final ParseField DATABASE_FIELD = new ParseField("database"); private static final ParseField UPDATE_STATS_FIELD = new ParseField("update_stats"); - private static final ParseField FEED_FORMAT = new ParseField("field_format"); - private static final ParseField DESCRIPTION = new ParseField("description"); - private static final ParseField ORGANIZATION = new ParseField("organization"); /** * Default variables for job scheduling @@ -93,12 +90,12 @@ public class Datasource implements Writeable, ScheduledJobParameter { */ private Instant lastUpdateTime; /** - * @param enabledTime Last time when a scheduling is enabled for a GeoIP data update + * @param enabledTime Last time when a scheduling is enabled for a threat intel feed data update * @return Last time when a scheduling is enabled for the job scheduler */ private Instant enabledTime; /** - * @param isEnabled Indicate if threatIP data update is scheduled or not + * @param isEnabled Indicate if threat intel feed data update is scheduled or not * @return Indicate if scheduling is enabled or not */ private boolean isEnabled; @@ -125,38 +122,60 @@ public class Datasource implements Writeable, ScheduledJobParameter { */ private String feedFormat; - private String description; - - private String organization; /** * @param endpoint URL of a manifest file * @return URL of a manifest file */ private String endpoint; + + /** + * @param feedName name of the threat intel feed + * @return name of the threat intel feed + */ + private String feedName; + + /** + * @param description description of the threat intel feed + * @return description of the threat intel feed + */ + private String description; + + /** + * @param organization organization of the threat intel feed + * @return organization of the threat intel feed + */ + private String organization; + + /** + * @param contained_iocs_field list of iocs contained in a given feed + * @return list of iocs contained in a given feed + */ + private List contained_iocs_field; + /** * @param state State of a datasource * @return State of a datasource */ private DatasourceState state; /** - * @param currentIndex the current index name having threatIP data - * @return the current index name having threatIP data + * @param currentIndex the current index name having threat intel feed data + * @return the current index name having threat intel feed data */ @Getter(AccessLevel.NONE) private String currentIndex; /** - * @param indices A list of indices having threatIP data including currentIndex - * @return A list of indices having threatIP data including currentIndex + * @param indices A list of indices having threat intel feed data including currentIndex + * @return A list of indices having threat intel feed data including currentIndex */ private List indices; /** - * @param database threatIP database information - * @return threatIP database information + * @param database threat intel feed database information + * @return threat intel feed database information */ private Database database; /** - * @param updateStats threatIP database update statistics - * @return threatIP database update statistics + * @param updateStats threat intel feed database update statistics + * @return threat intel feed database update statistics */ private UpdateStats updateStats; @@ -175,14 +194,16 @@ public class Datasource implements Writeable, ScheduledJobParameter { IntervalSchedule schedule = (IntervalSchedule) args[4]; DatasourceTask task = DatasourceTask.valueOf((String) args[6]); String feedFormat = (String) args[7]; - String description = (String) args[8]; - String organization = (String) args[9]; - String endpoint = (String) args[10]; - DatasourceState state = DatasourceState.valueOf((String) args[11]); - String currentIndex = (String) args[12]; - List indices = (List) args[13]; - Database database = (Database) args[14]; - UpdateStats updateStats = (UpdateStats) args[15]; + String endpoint = (String) args[8]; + String feedName = (String) args[9]; + String description = (String) args[10]; + String organization = (String) args[11]; + List contained_iocs_field = (List) args[12]; + DatasourceState state = DatasourceState.valueOf((String) args[13]); + String currentIndex = (String) args[14]; + List indices = (List) args[15]; + Database database = (Database) args[16]; + UpdateStats updateStats = (UpdateStats) args[17]; Datasource parameter = new Datasource( name, lastUpdateTime, @@ -191,9 +212,11 @@ public class Datasource implements Writeable, ScheduledJobParameter { schedule, task, feedFormat, + endpoint, + feedName, description, organization, - endpoint, + contained_iocs_field, state, currentIndex, indices, @@ -211,9 +234,11 @@ public class Datasource implements Writeable, ScheduledJobParameter { PARSER.declareObject(ConstructingObjectParser.constructorArg(), (p, c) -> ScheduleParser.parse(p), SCHEDULE_FIELD); PARSER.declareString(ConstructingObjectParser.constructorArg(), TASK_FIELD); PARSER.declareString(ConstructingObjectParser.constructorArg(), FEED_FORMAT); + PARSER.declareString(ConstructingObjectParser.constructorArg(), ENDPOINT_FIELD); + PARSER.declareString(ConstructingObjectParser.constructorArg(), FEED_NAME); PARSER.declareString(ConstructingObjectParser.constructorArg(), DESCRIPTION); PARSER.declareString(ConstructingObjectParser.constructorArg(), ORGANIZATION); - PARSER.declareString(ConstructingObjectParser.constructorArg(), ENDPOINT_FIELD); + PARSER.declareStringArray(ConstructingObjectParser.constructorArg(), CONTAINED_IOCS_FIELD); PARSER.declareString(ConstructingObjectParser.constructorArg(), STATE_FIELD); PARSER.declareString(ConstructingObjectParser.optionalConstructorArg(), CURRENT_INDEX_FIELD); PARSER.declareStringArray(ConstructingObjectParser.constructorArg(), INDICES_FIELD); @@ -222,10 +247,10 @@ public class Datasource implements Writeable, ScheduledJobParameter { } public Datasource() { - this(null, null, null, null, null, null); + this(null, null, null, null, null, null, null, null); } - public Datasource(final String name, final IntervalSchedule schedule, final String feedFormat, final String description, final String organization, final String endpoint ) { + public Datasource(final String name, final IntervalSchedule schedule, final String feedFormat, final String endpoint, final String feedName, final String description, final String organization, final List contained_iocs_field ) { this( name, Instant.now().truncatedTo(ChronoUnit.MILLIS), @@ -234,9 +259,11 @@ public Datasource(final String name, final IntervalSchedule schedule, final Stri schedule, DatasourceTask.ALL, feedFormat, + endpoint, + feedName, description, organization, - endpoint, + contained_iocs_field, DatasourceState.CREATING, null, new ArrayList<>(), @@ -253,9 +280,11 @@ public Datasource(final StreamInput in) throws IOException { schedule = new IntervalSchedule(in); task = DatasourceTask.valueOf(in.readString()); feedFormat = in.readString(); + endpoint = in.readString(); + feedName = in.readString(); description = in.readString(); organization = in.readString(); - endpoint = in.readString(); + contained_iocs_field = in.readStringList(); state = DatasourceState.valueOf(in.readString()); currentIndex = in.readOptionalString(); indices = in.readStringList(); @@ -271,9 +300,11 @@ public void writeTo(final StreamOutput out) throws IOException { schedule.writeTo(out); out.writeString(task.name()); out.writeString(feedFormat); + out.writeString(endpoint); + out.writeString(feedName); out.writeString(description); out.writeString(organization); - out.writeString(endpoint); + out.writeStringCollection(contained_iocs_field); out.writeString(state.name()); out.writeOptionalString(currentIndex); out.writeStringCollection(indices); @@ -300,10 +331,12 @@ public XContentBuilder toXContent(final XContentBuilder builder, final Params pa builder.field(ENABLED_FIELD.getPreferredName(), isEnabled); builder.field(SCHEDULE_FIELD.getPreferredName(), schedule); builder.field(TASK_FIELD.getPreferredName(), task.name()); - builder.field(FEED_FORMAT.getPreferredName(), endpoint); - builder.field(DESCRIPTION.getPreferredName(), endpoint); - builder.field(ORGANIZATION.getPreferredName(), endpoint); + builder.field(FEED_FORMAT.getPreferredName(), feedFormat); builder.field(ENDPOINT_FIELD.getPreferredName(), endpoint); + builder.field(FEED_NAME.getPreferredName(), feedName); + builder.field(DESCRIPTION.getPreferredName(), description); + builder.field(ORGANIZATION.getPreferredName(), organization); + builder.field(CONTAINED_IOCS_FIELD.getPreferredName(), contained_iocs_field); builder.field(STATE_FIELD.getPreferredName(), state.name()); if (currentIndex != null) { builder.field(CURRENT_INDEX_FIELD.getPreferredName(), currentIndex); @@ -346,7 +379,7 @@ public Long getLockDurationSeconds() { } /** - * Enable auto update of threatIP data + * Enable auto update of threat intel feed data */ public void enable() { if (isEnabled == true) { @@ -357,7 +390,7 @@ public void enable() { } /** - * Disable auto update of threatIP data + * Disable auto update of threat intel feed data */ public void disable() { enabledTime = null; @@ -373,6 +406,18 @@ public String currentIndexName() { return currentIndex; } + public void setSchedule(IntervalSchedule schedule) { + this.schedule = schedule; + } + + /** + * Reset database so that it can be updated in next run regardless there is new update or not + */ + public void resetDatabase() { + database.setUpdatedAt(null); + database.setSha256Hash(null); + } + /** * Index name for a datasource with given suffix * @@ -380,13 +425,253 @@ public String currentIndexName() { * @return index name for a datasource with given suffix */ public String newIndexName(final String suffix) { - return String.format(Locale.ROOT, "%s.%s.%s", THREATINTEL_DATA_INDEX_NAME_PREFIX, name, suffix); + return String.format(Locale.ROOT, "%s.%s.%s", THREAT_INTEL_DATA_INDEX_NAME_PREFIX, name, suffix); } - public void setSchedule(IntervalSchedule schedule) { - this.schedule = schedule; + /** + * Set database attributes with given input + * + * @param datasourceManifest the datasource manifest + * @param fields the fields + */ + public void setDatabase(final DatasourceManifest datasourceManifest, final List fields) { + this.database.setProvider(datasourceManifest.getOrganization()); + this.database.setSha256Hash(datasourceManifest.getSha256Hash()); + this.database.setUpdatedAt(Instant.ofEpochMilli(datasourceManifest.getUpdatedAt())); + this.database.setFields(fields); + } + + /** + * Checks if the database fields are compatible with the given set of fields. + * + * If database fields are null, it is compatible with any input fields + * as it hasn't been generated before. + * + * @param fields The set of input fields to check for compatibility. + * @return true if the database fields are compatible with the given input fields, false otherwise. + */ + public boolean isCompatible(final List fields) { + if (database.fields == null) { + return true; + } + + if (fields.size() < database.fields.size()) { + return false; + } + + Set fieldsSet = new HashSet<>(fields); + for (String field : database.fields) { + if (fieldsSet.contains(field) == false) { + return false; + } + } + return true; + } + + /** + * Database of a datasource + */ + @Getter + @Setter + @ToString + @EqualsAndHashCode + @AllArgsConstructor(access = AccessLevel.PRIVATE) + @NoArgsConstructor(access = AccessLevel.PRIVATE) + public static class Database implements Writeable, ToXContent { + private static final ParseField PROVIDER_FIELD = new ParseField("provider"); + private static final ParseField SHA256_HASH_FIELD = new ParseField("sha256_hash"); + private static final ParseField UPDATED_AT_FIELD = new ParseField("updated_at_in_epoch_millis"); + private static final ParseField UPDATED_AT_FIELD_READABLE = new ParseField("updated_at"); + private static final ParseField FIELDS_FIELD = new ParseField("fields"); + + /** + * @param provider A database provider name + * @return A database provider name + */ + private String provider; + /** + * @param sha256Hash SHA256 hash value of a database file + * @return SHA256 hash value of a database file + */ + private String sha256Hash; + /** + * @param updatedAt A date when the database was updated + * @return A date when the database was updated + */ + private Instant updatedAt; + /** + * @param fields A list of available fields in the database + * @return A list of available fields in the database + */ + private List fields; + + private static final ConstructingObjectParser PARSER = new ConstructingObjectParser<>( + "datasource_metadata_database", + true, + args -> { + String provider = (String) args[0]; + String sha256Hash = (String) args[1]; + Instant updatedAt = args[2] == null ? null : Instant.ofEpochMilli((Long) args[2]); + List fields = (List) args[3]; + return new Database(provider, sha256Hash, updatedAt, fields); + } + ); + static { + PARSER.declareString(ConstructingObjectParser.optionalConstructorArg(), PROVIDER_FIELD); + PARSER.declareString(ConstructingObjectParser.optionalConstructorArg(), SHA256_HASH_FIELD); + PARSER.declareLong(ConstructingObjectParser.optionalConstructorArg(), UPDATED_AT_FIELD); + PARSER.declareStringArray(ConstructingObjectParser.optionalConstructorArg(), FIELDS_FIELD); + } + + public Database(final StreamInput in) throws IOException { + provider = in.readOptionalString(); + sha256Hash = in.readOptionalString(); + updatedAt = toInstant(in.readOptionalVLong()); + fields = in.readOptionalStringList(); + } + + @Override + public void writeTo(final StreamOutput out) throws IOException { + out.writeOptionalString(provider); + out.writeOptionalString(sha256Hash); + out.writeOptionalVLong(updatedAt == null ? null : updatedAt.toEpochMilli()); + out.writeOptionalStringCollection(fields); + } + + @Override + public XContentBuilder toXContent(final XContentBuilder builder, final Params params) throws IOException { + builder.startObject(); + if (provider != null) { + builder.field(PROVIDER_FIELD.getPreferredName(), provider); + } + if (sha256Hash != null) { + builder.field(SHA256_HASH_FIELD.getPreferredName(), sha256Hash); + } + if (updatedAt != null) { + builder.timeField( + UPDATED_AT_FIELD.getPreferredName(), + UPDATED_AT_FIELD_READABLE.getPreferredName(), + updatedAt.toEpochMilli() + ); + } + if (fields != null) { + builder.startArray(FIELDS_FIELD.getPreferredName()); + for (String field : fields) { + builder.value(field); + } + builder.endArray(); + } + builder.endObject(); + return builder; + } + } + + /** + * Update stats of a datasource + */ + @Getter + @Setter + @ToString + @EqualsAndHashCode + @AllArgsConstructor(access = AccessLevel.PRIVATE) + @NoArgsConstructor(access = AccessLevel.PRIVATE) + public static class UpdateStats implements Writeable, ToXContent { + private static final ParseField LAST_SUCCEEDED_AT_FIELD = new ParseField("last_succeeded_at_in_epoch_millis"); + private static final ParseField LAST_SUCCEEDED_AT_FIELD_READABLE = new ParseField("last_succeeded_at"); + private static final ParseField LAST_PROCESSING_TIME_IN_MILLIS_FIELD = new ParseField("last_processing_time_in_millis"); + private static final ParseField LAST_FAILED_AT_FIELD = new ParseField("last_failed_at_in_epoch_millis"); + private static final ParseField LAST_FAILED_AT_FIELD_READABLE = new ParseField("last_failed_at"); + private static final ParseField LAST_SKIPPED_AT = new ParseField("last_skipped_at_in_epoch_millis"); + private static final ParseField LAST_SKIPPED_AT_READABLE = new ParseField("last_skipped_at"); + + /** + * @param lastSucceededAt The last time when threat intel feed data update was succeeded + * @return The last time when threat intel feed data update was succeeded + */ + private Instant lastSucceededAt; + /** + * @param lastProcessingTimeInMillis The last processing time when threat intel feed data update was succeeded + * @return The last processing time when threat intel feed data update was succeeded + */ + private Long lastProcessingTimeInMillis; + /** + * @param lastFailedAt The last time when threat intel feed data update was failed + * @return The last time when threat intel feed data update was failed + */ + private Instant lastFailedAt; + /** + * @param lastSkippedAt The last time when threat intel feed data update was skipped as there was no new update from an endpoint + * @return The last time when threat intel feed data update was skipped as there was no new update from an endpoint + */ + private Instant lastSkippedAt; + + private static final ConstructingObjectParser PARSER = new ConstructingObjectParser<>( + "datasource_metadata_update_stats", + true, + args -> { + Instant lastSucceededAt = args[0] == null ? null : Instant.ofEpochMilli((long) args[0]); + Long lastProcessingTimeInMillis = (Long) args[1]; + Instant lastFailedAt = args[2] == null ? null : Instant.ofEpochMilli((long) args[2]); + Instant lastSkippedAt = args[3] == null ? null : Instant.ofEpochMilli((long) args[3]); + return new UpdateStats(lastSucceededAt, lastProcessingTimeInMillis, lastFailedAt, lastSkippedAt); + } + ); + + static { + PARSER.declareLong(ConstructingObjectParser.optionalConstructorArg(), LAST_SUCCEEDED_AT_FIELD); + PARSER.declareLong(ConstructingObjectParser.optionalConstructorArg(), LAST_PROCESSING_TIME_IN_MILLIS_FIELD); + PARSER.declareLong(ConstructingObjectParser.optionalConstructorArg(), LAST_FAILED_AT_FIELD); + PARSER.declareLong(ConstructingObjectParser.optionalConstructorArg(), LAST_SKIPPED_AT); + } + + public UpdateStats(final StreamInput in) throws IOException { + lastSucceededAt = toInstant(in.readOptionalVLong()); + lastProcessingTimeInMillis = in.readOptionalVLong(); + lastFailedAt = toInstant(in.readOptionalVLong()); + lastSkippedAt = toInstant(in.readOptionalVLong()); + } + + @Override + public void writeTo(final StreamOutput out) throws IOException { + out.writeOptionalVLong(lastSucceededAt == null ? null : lastSucceededAt.toEpochMilli()); + out.writeOptionalVLong(lastProcessingTimeInMillis); + out.writeOptionalVLong(lastFailedAt == null ? null : lastFailedAt.toEpochMilli()); + out.writeOptionalVLong(lastSkippedAt == null ? null : lastSkippedAt.toEpochMilli()); + } + + @Override + public XContentBuilder toXContent(final XContentBuilder builder, final Params params) throws IOException { + builder.startObject(); + if (lastSucceededAt != null) { + builder.timeField( + LAST_SUCCEEDED_AT_FIELD.getPreferredName(), + LAST_SUCCEEDED_AT_FIELD_READABLE.getPreferredName(), + lastSucceededAt.toEpochMilli() + ); + } + if (lastProcessingTimeInMillis != null) { + builder.field(LAST_PROCESSING_TIME_IN_MILLIS_FIELD.getPreferredName(), lastProcessingTimeInMillis); + } + if (lastFailedAt != null) { + builder.timeField( + LAST_FAILED_AT_FIELD.getPreferredName(), + LAST_FAILED_AT_FIELD_READABLE.getPreferredName(), + lastFailedAt.toEpochMilli() + ); + } + if (lastSkippedAt != null) { + builder.timeField( + LAST_SKIPPED_AT.getPreferredName(), + LAST_SKIPPED_AT_READABLE.getPreferredName(), + lastSkippedAt.toEpochMilli() + ); + } + builder.endObject(); + return builder; + } } + /** * Builder class for Datasource */ @@ -398,8 +683,13 @@ public static Datasource build(final PutDatasourceRequest request) { (int) request.getUpdateInterval().days(), ChronoUnit.DAYS ); + String feedFormat = request.getFeedFormat(); String endpoint = request.getEndpoint(); - return new Datasource(id, schedule, endpoint); + String feedName = request.getFeedName(); + String description = request.getDescription(); + String organization = request.getOrganization(); + List contained_iocs_field = request.getContained_iocs_field(); + return new Datasource(id, schedule, feedFormat, endpoint, feedName, description, organization, contained_iocs_field); } } } \ No newline at end of file diff --git a/src/main/java/org/opensearch/securityanalytics/threatintel/jobscheduler/DatasourceExtension.java b/src/main/java/org/opensearch/securityanalytics/threatintel/jobscheduler/DatasourceExtension.java index 9cf21e88e..c010444e7 100644 --- a/src/main/java/org/opensearch/securityanalytics/threatintel/jobscheduler/DatasourceExtension.java +++ b/src/main/java/org/opensearch/securityanalytics/threatintel/jobscheduler/DatasourceExtension.java @@ -15,7 +15,14 @@ public class DatasourceExtension implements JobSchedulerExtension { /** * Job index name for a datasource */ - public static final String JOB_INDEX_NAME = ".scheduler-security_analytics-threatintel-datasource"; + public static final String JOB_INDEX_NAME = ".scheduler-security_analytics-threatintel-datasource"; //rename this... + + /** + * Job index setting + * + * We want it to be single shard so that job can be run only in a single node by job scheduler. + * We want it to expand to all replicas so that querying to this index can be done locally to reduce latency. + */ public static final Map INDEX_SETTING = Map.of("index.number_of_shards", 1, "index.number_of_replicas", "0-all", "index.hidden", true); @Override diff --git a/src/main/java/org/opensearch/securityanalytics/threatintel/jobscheduler/DatasourceRunner.java b/src/main/java/org/opensearch/securityanalytics/threatintel/jobscheduler/DatasourceRunner.java index 4374d68a1..e4252db27 100644 --- a/src/main/java/org/opensearch/securityanalytics/threatintel/jobscheduler/DatasourceRunner.java +++ b/src/main/java/org/opensearch/securityanalytics/threatintel/jobscheduler/DatasourceRunner.java @@ -18,12 +18,17 @@ import java.io.IOException; import java.util.Optional; import java.util.concurrent.atomic.AtomicReference; +import java.time.Instant; import org.opensearch.securityanalytics.threatintel.common.DatasourceState; import org.opensearch.securityanalytics.threatintel.common.ThreatIntelExecutor; import org.opensearch.securityanalytics.threatintel.common.ThreatIntelLockService; import org.opensearch.securityanalytics.threatintel.dao.DatasourceDao; - +/** + * Datasource update task + * + * This is a background task which is responsible for updating threat intel feed data + */ public class DatasourceRunner implements ScheduledJobRunner { private static final Logger log = LogManager.getLogger(DetectorTrigger.class); private static DatasourceRunner INSTANCE; @@ -43,7 +48,7 @@ public static DatasourceRunner getJobRunnerInstance() { private ClusterService clusterService; - // specialized part + // threat intel specific variables private DatasourceUpdateService datasourceUpdateService; private DatasourceDao datasourceDao; private ThreatIntelExecutor threatIntelExecutor; @@ -77,16 +82,16 @@ public void runJob(final ScheduledJobParameter jobParameter, final JobExecutionC log.info("Update job started for a datasource[{}]", jobParameter.getName()); if (jobParameter instanceof Datasource == false) { + log.error("Illegal state exception: job parameter is not instance of Datasource"); throw new IllegalStateException( "job parameter is not instance of Datasource, type: " + jobParameter.getClass().getCanonicalName() ); } - threatIntelExecutor.forDatasourceUpdate().submit(updateDatasourceRunner(jobParameter)); } /** - * Update threatIP data + * Update threat intel feed data * * Lock is used so that only one of nodes run this task. * @@ -130,19 +135,19 @@ protected void updateDatasource(final ScheduledJobParameter jobParameter, final if (DatasourceState.AVAILABLE.equals(datasource.getState()) == false) { log.error("Invalid datasource state. Expecting {} but received {}", DatasourceState.AVAILABLE, datasource.getState()); datasource.disable(); -// datasource.getUpdateStats().setLastFailedAt(Instant.now()); + datasource.getUpdateStats().setLastFailedAt(Instant.now()); datasourceDao.updateDatasource(datasource); return; } try { datasourceUpdateService.deleteUnusedIndices(datasource); if (DatasourceTask.DELETE_UNUSED_INDICES.equals(datasource.getTask()) == false) { - datasourceUpdateService.updateOrCreateGeoIpData(datasource, renewLock); + datasourceUpdateService.updateOrCreateThreatIntelFeedData(datasource, renewLock); } datasourceUpdateService.deleteUnusedIndices(datasource); } catch (Exception e) { log.error("Failed to update datasource for {}", datasource.getName(), e); -// datasource.getUpdateStats().setLastFailedAt(Instant.now()); + datasource.getUpdateStats().setLastFailedAt(Instant.now()); datasourceDao.updateDatasource(datasource); } } diff --git a/src/main/java/org/opensearch/securityanalytics/threatintel/jobscheduler/DatasourceUpdateService.java b/src/main/java/org/opensearch/securityanalytics/threatintel/jobscheduler/DatasourceUpdateService.java index 84747153d..9f44c05fc 100644 --- a/src/main/java/org/opensearch/securityanalytics/threatintel/jobscheduler/DatasourceUpdateService.java +++ b/src/main/java/org/opensearch/securityanalytics/threatintel/jobscheduler/DatasourceUpdateService.java @@ -24,12 +24,14 @@ import org.opensearch.cluster.service.ClusterService; import org.opensearch.common.settings.ClusterSettings; +import org.opensearch.core.rest.RestStatus; import org.opensearch.jobscheduler.spi.schedule.IntervalSchedule; import org.opensearch.securityanalytics.model.DetectorTrigger; import org.opensearch.securityanalytics.threatintel.common.DatasourceManifest; import org.opensearch.securityanalytics.threatintel.dao.DatasourceDao; -import org.opensearch.securityanalytics.threatintel.dao.ThreatIpDataDao; +import org.opensearch.securityanalytics.threatintel.dao.ThreatIntelFeedDao; import org.opensearch.securityanalytics.threatintel.common.DatasourceState; +import org.opensearch.securityanalytics.util.SecurityAnalyticsException; public class DatasourceUpdateService { @@ -40,21 +42,21 @@ public class DatasourceUpdateService { private final ClusterService clusterService; private final ClusterSettings clusterSettings; private final DatasourceDao datasourceDao; - private final ThreatIpDataDao threatIpDataDao; + private final ThreatIntelFeedDao threatIntelFeedDao; public DatasourceUpdateService( final ClusterService clusterService, final DatasourceDao datasourceDao, - final ThreatIpDataDao threatIpDataDao + final ThreatIntelFeedDao threatIntelFeedDao ) { this.clusterService = clusterService; this.clusterSettings = clusterService.getClusterSettings(); this.datasourceDao = datasourceDao; - this.threatIpDataDao = threatIpDataDao; + this.threatIntelFeedDao = threatIntelFeedDao; } /** - * Update threatIp data + * Update threat intel feed data * * The first column is ip range field regardless its header name. * Therefore, we don't store the first column's header name. @@ -64,12 +66,12 @@ public DatasourceUpdateService( * * @throws IOException */ - public void updateOrCreateGeoIpData(final Datasource datasource, final Runnable renewLock) throws IOException { + public void updateOrCreateThreatIntelFeedData(final Datasource datasource, final Runnable renewLock) throws IOException { URL url = new URL(datasource.getEndpoint()); DatasourceManifest manifest = DatasourceManifest.Builder.build(url); if (shouldUpdate(datasource, manifest) == false) { - log.info("Skipping GeoIP database update. Update is not required for {}", datasource.getName()); + log.info("Skipping threat intel feed database update. Update is not required for {}", datasource.getName()); datasource.getUpdateStats().setLastSkippedAt(Instant.now()); datasourceDao.updateDatasource(datasource); return; @@ -79,18 +81,19 @@ public void updateOrCreateGeoIpData(final Datasource datasource, final Runnable String indexName = setupIndex(datasource); String[] header; List fieldsToStore; - try (CSVParser reader = geoIpDataDao.getDatabaseReader(manifest)) { + try (CSVParser reader = threatIntelFeedDao.getDatabaseReader(manifest)) { CSVRecord headerLine = reader.iterator().next(); header = validateHeader(headerLine).values(); fieldsToStore = Arrays.asList(header).subList(1, header.length); if (datasource.isCompatible(fieldsToStore) == false) { + log.error("Exception: new fields does not contain all old fields"); throw new OpenSearchException( "new fields [{}] does not contain all old fields [{}]", fieldsToStore.toString(), datasource.getDatabase().getFields().toString() ); } - threatIpDataDao.putGeoIpData(indexName, header, reader.iterator(), renewLock); + threatIntelFeedDao.putTIFData(indexName, header, reader.iterator(), renewLock); } waitUntilAllShardsStarted(indexName, MAX_WAIT_TIME_FOR_REPLICATION_TO_COMPLETE_IN_MILLIS); @@ -98,9 +101,10 @@ public void updateOrCreateGeoIpData(final Datasource datasource, final Runnable updateDatasourceAsSucceeded(indexName, datasource, manifest, fieldsToStore, startTime, endTime); } + /** * We wait until all shards are ready to serve search requests before updating datasource metadata to - * point to a new index so that there won't be latency degradation during GeoIP data update + * point to a new index so that there won't be latency degradation during threat intel feed data update * * @param indexName the indexName */ @@ -118,24 +122,25 @@ protected void waitUntilAllShardsStarted(final String indexName, final int timeo MAX_WAIT_TIME_FOR_REPLICATION_TO_COMPLETE_IN_MILLIS ); } catch (InterruptedException e) { - throw new RuntimeException(e); + log.error("runtime exception", e); + throw new SecurityAnalyticsException("Runtime exception", RestStatus.INTERNAL_SERVER_ERROR, e); //TODO } } /** - * Return header fields of geo data with given url of a manifest file + * Return header fields of threat intel feed data with given url of a manifest file * * The first column is ip range field regardless its header name. * Therefore, we don't store the first column's header name. * * @param manifestUrl the url of a manifest file - * @return header fields of geo data + * @return header fields of ioc data */ public List getHeaderFields(String manifestUrl) throws IOException { URL url = new URL(manifestUrl); DatasourceManifest manifest = DatasourceManifest.Builder.build(url); - try (CSVParser reader = threatIpDataDao.getDatabaseReader(manifest)) { + try (CSVParser reader = threatIntelFeedDao.getDatabaseReader(manifest)) { String[] fields = reader.iterator().next().values(); return Arrays.asList(fields).subList(1, fields.length); } @@ -173,10 +178,7 @@ public void deleteUnusedIndices(final Datasource datasource) { */ public void updateDatasource(final Datasource datasource, final IntervalSchedule systemSchedule, final DatasourceTask task) { boolean updated = false; - if (datasource.getSystemSchedule().equals(systemSchedule) == false) { - datasource.setSystemSchedule(systemSchedule); - updated = true; - } + if (datasource.getTask().equals(task) == false) { datasource.setTask(task); updated = true; @@ -185,7 +187,7 @@ public void updateDatasource(final Datasource datasource, final IntervalSchedule if (updated) { datasourceDao.updateDatasource(datasource); } - } + } //TODO private List deleteIndices(final List indicesToDelete) { List deletedIndices = new ArrayList<>(indicesToDelete.size()); @@ -196,7 +198,7 @@ private List deleteIndices(final List indicesToDelete) { } try { - threatIpDataDao.deleteIp2GeoDataIndex(index); + threatIntelFeedDao.deleteThreatIntelDataIndex(index); deletedIndices.add(index); } catch (Exception e) { log.error("Failed to delete an index [{}]", index, e); @@ -216,10 +218,10 @@ private List deleteIndices(final List indicesToDelete) { */ private CSVRecord validateHeader(CSVRecord header) { if (header == null) { - throw new OpenSearchException("geoip database is empty"); + throw new OpenSearchException("threat intel feed database is empty"); } if (header.values().length < 2) { - throw new OpenSearchException("geoip database should have at least two fields"); + throw new OpenSearchException("threat intel feed database should have at least two fields"); } return header; } @@ -246,14 +248,14 @@ private void updateDatasourceAsSucceeded( datasource.setState(DatasourceState.AVAILABLE); datasourceDao.updateDatasource(datasource); log.info( - "threatIP database creation succeeded for {} and took {} seconds", + "threat intel feed database creation succeeded for {} and took {} seconds", datasource.getName(), Duration.between(startTime, endTime) ); } /*** - * Setup index to add a new threatIp data + * Setup index to add a new threat intel feed data * * @param datasource the datasource * @return new index name @@ -262,7 +264,7 @@ private String setupIndex(final Datasource datasource) { String indexName = datasource.newIndexName(UUID.randomUUID().toString()); datasource.getIndices().add(indexName); datasourceDao.updateDatasource(datasource); -// threatIpDataDao.createIndexIfNotExists(indexName); + threatIntelFeedDao.createIndexIfNotExists(indexName); return indexName; }