Skip to content

Commit

Permalink
Added dummy search when creating detector on the given indicies (#197)
Browse files Browse the repository at this point in the history 
Signed-off-by: Stevan Buzejic <[email protected]>
  • Loading branch information
@stevanbz
stevanbz authored Jan 5, 2023
1 parent 95b4f95 commit 51a176b
Show file tree
Hide file tree
Showing 5 changed files with 817 additions and 456 deletions.
46 changes: 36 additions & 10 deletions src/main/java/org/opensearch/securityanalytics/transport/TransportIndexDetectorAction.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -173,8 +173,35 @@ protected void doExecute(Task task, IndexDetectorRequest request, ActionListener
return;
}

AsyncIndexDetectorsAction asyncAction = new AsyncIndexDetectorsAction(user, task, request, listener);
asyncAction.start();
checkIndicesAndExecute(task, request, listener, user);
}

private void checkIndicesAndExecute(
Task task,
IndexDetectorRequest request,
ActionListener<IndexDetectorResponse> listener,
User user
) {
String [] detectorIndices = request.getDetector().getInputs().stream().flatMap(detectorInput -> detectorInput.getIndices().stream()).toArray(String[]::new);
SearchRequest searchRequest = new SearchRequest(detectorIndices).source(SearchSourceBuilder.searchSource().size(1).query(QueryBuilders.matchAllQuery()));;
client.search(searchRequest, new ActionListener<>() {
@Override
public void onResponse(SearchResponse searchResponse) {
AsyncIndexDetectorsAction asyncAction = new AsyncIndexDetectorsAction(user, task, request, listener);
asyncAction.start();
}

@Override
public void onFailure(Exception e) {
if (e instanceof OpenSearchStatusException) {
listener.onFailure(SecurityAnalyticsException.wrap(
new OpenSearchStatusException(String.format(Locale.getDefault(), "User doesn't have read permissions for one or more configured index %s", detectorIndices), RestStatus.FORBIDDEN)
));
} else {
listener.onFailure(e);
}
}
});
}

private void createMonitorFromQueries(String index, List<Pair<String, Rule>> rulesById, Detector detector, ActionListener<List<IndexMonitorResponse>> listener, WriteRequest.RefreshPolicy refreshPolicy) throws SigmaError, IOException {
Expand All @@ -192,7 +219,7 @@ private void createMonitorFromQueries(String index, List<Pair<String, Rule>> rul
monitorRequests.addAll(buildBucketLevelMonitorRequests(Pair.of(index, bucketLevelRules), detector, refreshPolicy, Monitor.NO_ID, Method.POST));
}
// Do nothing if detector doesn't have any monitor
if(monitorRequests.isEmpty()){
if (monitorRequests.isEmpty()){
listener.onResponse(Collections.emptyList());
return;
}
Expand All @@ -206,7 +233,7 @@ private void createMonitorFromQueries(String index, List<Pair<String, Rule>> rul
addFirstMonitorStep.whenComplete(addedFirstMonitorResponse -> {
monitorResponses.add(addedFirstMonitorResponse);
int numberOfUnprocessedResponses = monitorRequests.size() - 1;
if(numberOfUnprocessedResponses == 0){
if (numberOfUnprocessedResponses == 0){
listener.onResponse(monitorResponses);
} else {
GroupedActionListener<IndexMonitorResponse> monitorResponseListener = new GroupedActionListener(
Expand Down Expand Up @@ -251,7 +278,7 @@ private void updateMonitorFromQueries(String index, List<Pair<String, Rule>> rul

for (Pair<String, Rule> query: bucketLevelRules) {
Rule rule = query.getRight();
if(rule.getAggregationQueries() != null){
if (rule.getAggregationQueries() != null){
// Detect if the monitor should be added or updated
if (monitorPerRule.containsKey(rule.getId())) {
String monitorId = monitorPerRule.get(rule.getId());
Expand Down Expand Up @@ -321,15 +348,15 @@ private void updateAlertingMonitors(
executeMonitorActionRequest(monitorsToBeAdded, addNewMonitorsStep);
// 1. Add new alerting monitors (for the rules that didn't exist previously)
addNewMonitorsStep.whenComplete(addNewMonitorsResponse -> {
if(addNewMonitorsResponse != null && !addNewMonitorsResponse.isEmpty()) {
if (addNewMonitorsResponse != null && !addNewMonitorsResponse.isEmpty()) {
updatedMonitors.addAll(addNewMonitorsResponse);
}

StepListener<List<IndexMonitorResponse>> updateMonitorsStep = new StepListener<>();
executeMonitorActionRequest(monitorsToBeUpdated, updateMonitorsStep);
// 2. Update existing alerting monitors (based on the common rules)
updateMonitorsStep.whenComplete(updateMonitorResponse -> {
if(updateMonitorResponse!=null && !updateMonitorResponse.isEmpty()) {
if (updateMonitorResponse!=null && !updateMonitorResponse.isEmpty()) {
updatedMonitors.addAll(updateMonitorResponse);
}

Expand Down Expand Up @@ -413,7 +440,7 @@ private List<IndexMonitorRequest> buildBucketLevelMonitorRequests(Pair<String, L
Rule rule = query.getRight();

// Creating bucket level monitor per each aggregation rule
if(rule.getAggregationQueries() != null){
if (rule.getAggregationQueries() != null){
monitorRequests.add(createBucketLevelMonitorRequest(
query.getRight(),
logIndexToQueries.getLeft(),
Expand Down Expand Up @@ -490,7 +517,7 @@ public void executeMonitorActionRequest(
ActionListener<List<IndexMonitorResponse>> listener) {

// In the case of not provided monitors, just return empty list
if(indexMonitors == null || indexMonitors.isEmpty()) {
if (indexMonitors == null || indexMonitors.isEmpty()) {
listener.onResponse(new ArrayList<>());
return;
}
Expand Down Expand Up @@ -595,7 +622,6 @@ class AsyncIndexDetectorsAction {

void start() {
try {

TransportIndexDetectorAction.this.threadPool.getThreadContext().stashContext();

if (!detectorIndices.detectorIndexExists()) {
Expand Down
98 changes: 95 additions & 3 deletions src/test/java/org/opensearch/securityanalytics/SecurityAnalyticsRestTestCase.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -72,7 +72,7 @@
import static org.opensearch.action.admin.indices.create.CreateIndexRequest.MAPPINGS;
import static org.opensearch.securityanalytics.TestHelpers.sumAggregationTestRule;
import static org.opensearch.securityanalytics.TestHelpers.productIndexAvgAggRule;
import static org.opensearch.securityanalytics.util.RuleTopicIndices.ruleTopicIndexMappings;
import static org.opensearch.securityanalytics.TestHelpers.windowsIndexMapping;
import static org.opensearch.securityanalytics.util.RuleTopicIndices.ruleTopicIndexSettings;

public class SecurityAnalyticsRestTestCase extends OpenSearchRestTestCase {
Expand Down Expand Up @@ -108,6 +108,22 @@ protected void createRuleTopicIndex(String detectorType, String additionalMappin
}
}

protected final List<String> clusterPermissions = List.of(
"cluster:admin/opensearch/securityanalytics/detector/*",
"cluster:admin/opendistro/alerting/alerts/*",
"cluster:admin/opendistro/alerting/findings/*",
"cluster:admin/opensearch/securityanalytics/mapping/*",
"cluster:admin/opensearch/securityanalytics/rule/*"
);

protected final List<String> indexPermissions = List.of(
"indices:admin/mappings/get",
"indices:admin/mapping/put",
"indices:data/read/search"
);

protected static String TEST_HR_ROLE = "hr_role";

protected String createTestIndex(String index, String mapping) throws IOException {
createTestIndex(index, mapping, Settings.EMPTY);
return index;
Expand Down Expand Up @@ -375,6 +391,20 @@ public static SearchResponse executeSearchRequest(String indexName, String query
return SearchResponse.fromXContent(parser);
}

public static SearchResponse executeSearchRequest(RestClient client, String indexName, String queryJson) throws IOException {

Request request = new Request("GET", indexName + "/_search");
request.setJsonEntity(queryJson);
Response response = client.performRequest(request);

XContentParser parser = JsonXContent.jsonXContent.createParser(
new NamedXContentRegistry(ClusterModule.getNamedXWriteables()),
DeprecationHandler.THROW_UNSUPPORTED_OPERATION,
response.getEntity().getContent()
);
return SearchResponse.fromXContent(parser);
}

protected HttpEntity toHttpEntity(Detector detector) throws IOException {
return new StringEntity(toJsonString(detector), ContentType.APPLICATION_JSON);
}
Expand Down Expand Up @@ -997,6 +1027,41 @@ protected RestClient buildClient(Settings settings, HttpHost[] hosts) throws IOE

}

protected void createIndexRole(String name, List<String> clusterPermissions, List<String> indexPermission, List<String> indexPatterns) throws IOException {
Response response;
try {
response = client().performRequest(new Request("GET", String.format(Locale.getDefault(), "/_plugins/_security/api/roles/%s", name)));
} catch (ResponseException ex) {
response = ex.getResponse();
}
// Role already exists
if(response.getStatusLine().getStatusCode() == RestStatus.OK.getStatus()) {
return;
}

Request request = new Request("PUT", String.format(Locale.getDefault(), "/_plugins/_security/api/roles/%s", name));
String clusterPermissionsStr = clusterPermissions.stream().map(p -> "\"" + p + "\"").collect(Collectors.joining(","));
String indexPermissionsStr = indexPermission.stream().map(p -> "\"" + p + "\"").collect(Collectors.joining(","));
String indexPatternsStr = indexPatterns.stream().map(p -> "\"" + p + "\"").collect(Collectors.joining(","));

String entity = "{\n" +
"\"cluster_permissions\": [\n" +
"" + clusterPermissionsStr + "\n" +
"], \n" +
"\"index_permissions\": [\n" +
"{" +
"\"fls\": [], " +
"\"masked_fields\": [], " +
"\"allowed_actions\": [" + indexPermissionsStr + "], " +
"\"index_patterns\": [" + indexPatternsStr + "]" +
"}" +
"], " +
"\"tenant_permissions\": []" +
"}";

request.setJsonEntity(entity);
client().performRequest(request);
}

protected void createCustomRole(String name, String clusterPermissions) throws IOException {
Request request = new Request("PUT", String.format(Locale.getDefault(), "/_plugins/_security/api/roles/%s", name));
Expand All @@ -1009,7 +1074,7 @@ protected void createCustomRole(String name, String clusterPermissions) throws I
client().performRequest(request);
}

protected void createUser(String name, String passwd, String[] backendRoles) throws IOException {
public void createUser(String name, String passwd, String[] backendRoles) throws IOException {
Request request = new Request("PUT", String.format(Locale.getDefault(), "/_plugins/_security/api/internalusers/%s", name));
String broles = String.join(",", backendRoles);
//String roles = String.join(",", customRoles);
Expand Down Expand Up @@ -1048,19 +1113,46 @@ protected void createUserWithDataAndCustomRole(String userName, String userPass
createUserRolesMapping(roleName, users);
}

protected void createUserWithData(String userName, String userPasswd, String roleName, String[] backendRoles ) throws IOException {
protected void createUserWithDataAndCustomRole(String userName, String userPasswd, String roleName, String[] backendRoles, List<String> clusterPermissions, List<String> indexPermissions, List<String> indexPatterns) throws IOException {
String[] users = {userName};
createUser(userName, userPasswd, backendRoles);
createIndexRole(roleName, clusterPermissions, indexPermissions, indexPatterns);
createUserRolesMapping(roleName, users);
}

protected void createUserWithData(String userName, String userPasswd, String roleName, String[] backendRoles ) throws IOException {
String[] users = {userName};
createUser(userName, userPasswd, backendRoles);
createUserRolesMapping(roleName, users);
}

public void createUserWithTestData(String user, String index, String role, String [] backendRoles, List<String> indexPermissions) throws IOException{
String[] users = {user};
createUser(user, user, backendRoles);
createTestIndex(client(), index, windowsIndexMapping(), Settings.EMPTY);
createIndexRole(role, Collections.emptyList(), indexPermissions, List.of(index));
createUserRolesMapping(role, users);
}

protected void deleteUser(String name) throws IOException {
Request request = new Request("DELETE", String.format(Locale.getDefault(), "/_plugins/_security/api/internalusers/%s", name));
client().performRequest(request);
}

protected void tryDeletingRole(String name) throws IOException{
Response response;
try {
response = client().performRequest(new Request("GET", String.format(Locale.getDefault(), "/_plugins/_security/api/roles/%s", name)));
} catch (ResponseException ex) {
response = ex.getResponse();
}
// Role already exists
if(response.getStatusLine().getStatusCode() == RestStatus.OK.getStatus()) {
Request request = new Request("DELETE", String.format(Locale.getDefault(), "/_plugins/_security/api/roles/%s", name));
client().performRequest(request);
}
}

@Override
protected boolean preserveIndicesUponCompletion() {
return true;
Expand Down
Loading

0 comments on commit 51a176b

Please sign in to comment.