Skip to content

Commit bdcb41e

Browse files
nsgupta1nsgupta1
committed
CrowdStrike API call and retry mechanism
Signed-off-by: nsgupta1 <nsgupta1@users.noreply.github.com>
1 parent 52a1e6d commit bdcb41e

20 files changed

Lines changed: 1106 additions & 31 deletions

File tree

‎data-prepper-plugins/saas-source-plugins/atlassian-commons/src/main/java/org/opensearch/dataprepper/plugins/source/atlassian/AtlassianSourceConfig.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -50,4 +50,9 @@ public String getAuthType() {
5050
}
5151

5252
public abstract String getOauth2UrlContext();
53+
54+
@Override
55+
public int getNumberOfWorkers() {
56+
return DEFAULT_NUMBER_OF_WORKERS;
57+
}
5358
}

‎data-prepper-plugins/saas-source-plugins/crowdstrike-source/src/main/java/org/opensearch/dataprepper/plugins/source/crowdstrike/CrowdStrikeService.java‎

Lines changed: 89 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,89 @@
1+
package org.opensearch.dataprepper.plugins.source.crowdstrike;
2+
import io.micrometer.core.instrument.Timer;
3+
import org.opensearch.dataprepper.metrics.PluginMetrics;
4+
import org.opensearch.dataprepper.plugins.source.crowdstrike.models.CrowdStrikeApiResponse;
5+
import org.opensearch.dataprepper.plugins.source.crowdstrike.models.CrowdStrikeIndicatorResult;
6+
import org.opensearch.dataprepper.plugins.source.crowdstrike.rest.CrowdStrikeRestClient;
7+
import org.opensearch.dataprepper.plugins.source.crowdstrike.utils.CrowdStrikeNextLinkValidator;
8+
import org.slf4j.Logger;
9+
import org.slf4j.LoggerFactory;
10+
import org.springframework.http.ResponseEntity;
11+
import org.springframework.web.util.UriComponentsBuilder;
12+
import javax.inject.Named;
13+
import java.net.URI;
14+
import java.net.URLEncoder;
15+
import java.nio.charset.StandardCharsets;
16+
import static org.opensearch.dataprepper.plugins.source.crowdstrike.utils.Constants.BATCH_SIZE;
17+
import static org.opensearch.dataprepper.plugins.source.crowdstrike.utils.Constants.FILTER_KEY;
18+
import static org.opensearch.dataprepper.plugins.source.crowdstrike.utils.Constants.LAST_UPDATED;
19+
import static org.opensearch.dataprepper.plugins.source.crowdstrike.utils.Constants.LIMIT_KEY;
20+
21+
/**
22+
* This service manages the interaction with CrowdStrike's API endpoints, handles pagination
23+
* and provides methods for fetching indicators from CrowdStrike.
24+
*/
25+
@Named
26+
public class CrowdStrikeService {
27+
28+
private final CrowdStrikeRestClient crowdStrikeRestClient;
29+
private static final Logger log = LoggerFactory.getLogger(CrowdStrikeService.class);
30+
private static final String BASE_URL = "https://api.crowdstrike.com/";
31+
private static final String COMBINED_URL = "https://api.crowdstrike.com/intel/combined/indicators/v1";
32+
private final Timer searchCallLatencyTimer;
33+
34+
35+
public CrowdStrikeService(CrowdStrikeRestClient crowdStrikeRestClient, PluginMetrics pluginMetrics) {
36+
this.crowdStrikeRestClient = crowdStrikeRestClient;
37+
this.searchCallLatencyTimer = pluginMetrics.timer("searchCallLatencyTimer");
38+
}
39+
40+
/**
41+
* Retrieves all indicator data from CrowdStrike in a paginated fashion.
42+
* @param startTime The start timestamp (inclusive) for filtering indicators.
43+
* @param endTime The end timestamp (exclusive) for filtering indicators.
44+
* @param paginationLink An optional pagination URL suffix (used when fetching next pages).
45+
* @return A {@link CrowdStrikeApiResponse} containing response body and headers.
46+
*/
47+
public CrowdStrikeApiResponse getAllContent(Long startTime, Long endTime, String paginationLink) {
48+
URI uri = buildCrowdStrikeUri(startTime, endTime, paginationLink);
49+
50+
return searchCallLatencyTimer.record(() -> {
51+
try {
52+
log.debug("Calling CrowdStrike API with URI: {}", uri);
53+
ResponseEntity<CrowdStrikeIndicatorResult> responseEntity = crowdStrikeRestClient.invokeGetApi(uri, CrowdStrikeIndicatorResult.class);
54+
55+
CrowdStrikeApiResponse response = new CrowdStrikeApiResponse();
56+
response.setBody(responseEntity.getBody());
57+
response.setHeaders(responseEntity.getHeaders());
58+
return response;
59+
} catch (Exception e) {
60+
log.error("Error fetching CrowdStrike content from URI: {}", uri, e);
61+
throw new RuntimeException("CrowdStrike API call failed", e);
62+
}
63+
});
64+
}
65+
66+
protected URI buildCrowdStrikeUri(Long startTime, Long endTime, String paginationLink) {
67+
try {
68+
if (paginationLink != null) {
69+
String urlString = BASE_URL + paginationLink;
70+
urlString = CrowdStrikeNextLinkValidator.validateAndSanitizeURL(urlString);
71+
return new URI(urlString);
72+
} else {
73+
// Manually construct and encode the query string
74+
String filter1 = URLEncoder.encode(LAST_UPDATED + ":>=" + startTime, StandardCharsets.UTF_8);
75+
String filter2 = URLEncoder.encode(LAST_UPDATED + ":<" + endTime, StandardCharsets.UTF_8);
76+
String encodedFilter = filter1 + "%2B" + filter2; // Use literal '+' // ensure literal '+'
77+
78+
UriComponentsBuilder builder = UriComponentsBuilder
79+
.fromHttpUrl(COMBINED_URL)
80+
.queryParam(FILTER_KEY, encodedFilter)
81+
.queryParam(LIMIT_KEY, BATCH_SIZE);
82+
83+
return builder.build(true).toUri();
84+
}
85+
} catch (Exception e) {
86+
throw new RuntimeException("Failed to construct CrowdStrike request URI", e);
87+
}
88+
}
89+
}

‎data-prepper-plugins/saas-source-plugins/crowdstrike-source/src/main/java/org/opensearch/dataprepper/plugins/source/crowdstrike/CrowdStrikeSourceConfig.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,6 @@ public class CrowdStrikeSourceConfig implements CrawlerSourceConfig {
2727
@Min(1)
2828
@Max(50)
2929
@Valid
30-
private int numWorkers = DEFAULT_NUMBER_OF_WORKERS;
30+
private int numberOfWorkers = DEFAULT_NUMBER_OF_WORKERS;
3131

3232
}

‎data-prepper-plugins/saas-source-plugins/crowdstrike-source/src/main/java/org/opensearch/dataprepper/plugins/source/crowdstrike/models/CrowdStrikeApiResponse.java‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,29 @@
1+
package org.opensearch.dataprepper.plugins.source.crowdstrike.models;
2+
3+
import lombok.Getter;
4+
import lombok.Setter;
5+
import org.springframework.util.CollectionUtils;
6+
import java.util.Collections;
7+
import java.util.List;
8+
import java.util.Map;
9+
10+
/**
11+
* Represents the response returned from a CrowdStrike API call.
12+
*/
13+
@Getter @Setter
14+
public class CrowdStrikeApiResponse {
15+
16+
private CrowdStrikeIndicatorResult body;
17+
private Map<String, List<String>> headers;
18+
19+
// Convenience method to get a specific header
20+
public List<String> getHeader(String headerName) {
21+
return headers.getOrDefault(headerName, Collections.emptyList());
22+
}
23+
24+
// Convenience method to get the first value of a specific header
25+
public String getFirstHeaderValue(String headerName) {
26+
List<String> values = getHeader(headerName);
27+
return CollectionUtils.isEmpty(values) ? null : values.get(0);
28+
}
29+
}

‎data-prepper-plugins/saas-source-plugins/crowdstrike-source/src/main/java/org/opensearch/dataprepper/plugins/source/crowdstrike/models/CrowdStrikeIndicatorResult.java‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,18 @@
1+
package org.opensearch.dataprepper.plugins.source.crowdstrike.models;
2+
3+
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
4+
import com.fasterxml.jackson.annotation.JsonProperty;
5+
import lombok.Getter;
6+
import java.util.List;
7+
8+
/**
9+
* The result of Falcon query search.
10+
*/
11+
@Getter
12+
@JsonIgnoreProperties(ignoreUnknown = true)
13+
public class CrowdStrikeIndicatorResult {
14+
15+
@JsonProperty("resources")
16+
private List<ThreatIndicator> results = null;
17+
18+
}

‎data-prepper-plugins/saas-source-plugins/crowdstrike-source/src/main/java/org/opensearch/dataprepper/plugins/source/crowdstrike/models/ThreatIndicator.java‎

Lines changed: 59 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,59 @@
1+
package org.opensearch.dataprepper.plugins.source.crowdstrike.models;
2+
3+
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
4+
import com.fasterxml.jackson.annotation.JsonProperty;
5+
import lombok.Getter;
6+
import lombok.Setter;
7+
8+
/**
9+
* Represents a threat intelligence indicator from CrowdStrike's API.
10+
* This class encapsulates information about potential security threats,
11+
* including indicators of compromise (IoCs) and associated metadata.
12+
*
13+
* <p>A threat indicator may include various attributes such as:
14+
* <ul>
15+
* <li>Type of the indicator (IP, Domain, Hash, etc.)</li>
16+
* <li>Value of the indicator</li>
17+
* <li>Confidence score</li>
18+
* <li>Associated timestamps</li>
19+
* </ul>
20+
* </p>
21+
*/
22+
@Setter
23+
@Getter
24+
@JsonIgnoreProperties(ignoreUnknown = true)
25+
public class ThreatIndicator {
26+
/**
27+
* The ID of the IOC.
28+
*/
29+
@JsonProperty("id")
30+
private String id = null;
31+
32+
/**
33+
* The type of the IOC.
34+
*/
35+
@JsonProperty("type")
36+
private String type = null;
37+
38+
/**
39+
* The value of the IOC.
40+
*/
41+
@JsonProperty("indicator")
42+
private String indicator = null;
43+
44+
/**
45+
* The epoch timestamp of the creation date of IOC.
46+
*/
47+
@JsonProperty("published_date")
48+
private long publishedDate = 0L;
49+
50+
@JsonProperty("malicious_confidence")
51+
private String maliciousConfidence = null;
52+
53+
/**
54+
* The epoch timestamp of last updated date of the IOC.
55+
*/
56+
@JsonProperty("last_updated")
57+
private long lastUpdated = 0L;
58+
59+
}

‎data-prepper-plugins/saas-source-plugins/crowdstrike-source/src/main/java/org/opensearch/dataprepper/plugins/source/crowdstrike/rest/CrowdStrikeAuthClient.java‎

Lines changed: 50 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
import lombok.Getter;
44
import org.opensearch.dataprepper.plugins.source.crowdstrike.CrowdStrikeSourceConfig;
55
import org.opensearch.dataprepper.plugins.source.crowdstrike.configuration.AuthenticationConfig;
6+
import org.opensearch.dataprepper.plugins.source.source_crawler.exception.UnauthorizedException;
67
import org.slf4j.Logger;
78
import org.slf4j.LoggerFactory;
89
import org.springframework.http.HttpEntity;
@@ -16,6 +17,9 @@
1617
import java.util.Map;
1718
import javax.inject.Named;
1819
import static org.opensearch.dataprepper.logging.DataPrepperMarkers.NOISY;
20+
import static org.opensearch.dataprepper.plugins.source.crowdstrike.utils.Constants.MAX_RETRIES;
21+
import static org.opensearch.dataprepper.plugins.source.crowdstrike.utils.Constants.RETRY_ATTEMPT_SLEEP_TIME;
22+
import static org.opensearch.dataprepper.plugins.source.crowdstrike.utils.Constants.SLEEP_TIME_MULTIPLIER;
1923

2024

2125
/**
@@ -37,7 +41,7 @@ public class CrowdStrikeAuthClient {
3741
private static final String OAUTH_TOKEN_URL = "https://api.crowdstrike.com/oauth2/token";
3842
private static final String ACCESS_TOKEN = "access_token";
3943
private static final String EXPIRE_IN = "expires_in";
40-
44+
private final Object tokenRenewLock = new Object();
4145

4246

4347
public CrowdStrikeAuthClient(final CrowdStrikeSourceConfig sourceConfig) {
@@ -48,48 +52,74 @@ public CrowdStrikeAuthClient(final CrowdStrikeSourceConfig sourceConfig) {
4852

4953

5054
/**
51-
* Initializes the credentials by obtaining an authentication token.
55+
* Initializes the credentials by getting an authentication token.
5256
*/
5357
public void initCredentials() {
5458
log.info("Getting CrowdStrike Authentication Token");
5559
getAuthToken();
5660
}
5761

62+
5863
/**
5964
* Retrieves a new authentication token from the CrowdStrike API.
6065
* The token is stored in the {@code bearerToken} field, and its expiration time is updated.
6166
*
62-
* @throws RuntimeException if the token cannot be retrieved.
67+
* @throws UnauthorizedException Runtime exception if the token cannot be retrieved.
6368
*/
64-
private void getAuthToken() {
69+
protected void getAuthToken() {
6570
log.info(NOISY, "You are trying to access token");
6671
HttpHeaders headers = new HttpHeaders();
6772
headers.setContentType(MediaType.APPLICATION_FORM_URLENCODED);
6873
headers.setBasicAuth(this.clientId, this.clientSecret);
6974
HttpEntity<String> request = new HttpEntity<>(headers);
70-
try {
71-
ResponseEntity<Map> response = restTemplate.postForEntity(OAUTH_TOKEN_URL, request, Map.class);
72-
Map tokenData = response.getBody();
73-
this.bearerToken = (String) tokenData.get(ACCESS_TOKEN);
74-
this.expireTime = Instant.now().plusSeconds((Integer) tokenData.get(EXPIRE_IN));
75-
log.info("Access token acquired successfully");
76-
} catch (HttpClientErrorException ex) {
77-
this.expireTime = Instant.ofEpochMilli(0);
78-
HttpStatus statusCode = ex.getStatusCode();
79-
log.error("Failed to acquire access token. Status code: {}, Error Message: {}",
80-
statusCode, ex.getMessage());
81-
throw new RuntimeException("Error while requesting token:" + ex.getMessage(), ex);
75+
int retryCount = 0;
76+
while(retryCount < MAX_RETRIES) {
77+
try {
78+
ResponseEntity<Map> response = restTemplate.postForEntity(OAUTH_TOKEN_URL, request, Map.class);
79+
Map tokenData = response.getBody();
80+
this.bearerToken = (String) tokenData.get(ACCESS_TOKEN);
81+
this.expireTime = Instant.now().plusSeconds((Integer) tokenData.get(EXPIRE_IN));
82+
log.info("Access token acquired successfully");
83+
return;
84+
} catch (HttpClientErrorException ex) {
85+
this.expireTime = Instant.ofEpochMilli(0);
86+
HttpStatus statusCode = ex.getStatusCode();
87+
String statusMessage = ex.getMessage();
88+
log.error("Failed to acquire access token. Status code: {}, Error Message: {}",
89+
statusCode, statusMessage);
90+
try {
91+
Thread.sleep((long) RETRY_ATTEMPT_SLEEP_TIME.get(retryCount) * SLEEP_TIME_MULTIPLIER);
92+
} catch (InterruptedException e) {
93+
throw new RuntimeException("Sleep in the retry attempt got interrupted", e);
94+
}
95+
}
96+
retryCount++;
8297
}
98+
String errorMessage = String.format("Failed to acquire access token even after %s retry attempts", MAX_RETRIES);
99+
log.error(errorMessage);
100+
throw new UnauthorizedException(errorMessage);
83101
}
84102

85-
public boolean isTokenExpired() {
86-
return this.bearerToken == null || Instant.now().isAfter(this.expireTime);
103+
protected boolean isTokenValid() {
104+
Instant currentTime = Instant.now();
105+
return this.bearerToken != null && currentTime.isBefore(this.expireTime);
87106
}
88107

89108
/**
90109
* Refreshes the bearer token by retrieving a new one from CrowdStrike.
91110
*/
92111
public void refreshToken() {
93-
112+
if (isTokenValid()) {
113+
//There is still time to renew, or someone else must have already renewed it
114+
return;
115+
}
116+
synchronized (tokenRenewLock) {
117+
if (isTokenValid()) {
118+
//Someone else must have already renewed it
119+
return;
120+
}
121+
log.info("Renewing authentication token for CrowdStrike Connector.");
122+
getAuthToken();
123+
}
94124
}
95-
}
125+
}

0 commit comments

Comments
 (0)