Skip to content

use malicious event dao in main threat queries #3298

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Oct 20, 2025
Merged

use malicious event dao in main threat queries #3298

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 4 additions & 3 deletions apps/threat-detection-backend/src/main/java/com/akto/threat/backend/Main.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
import com.akto.kafka.KafkaProducerConfig;
import com.akto.kafka.Serializer;
import com.akto.log.LoggerMaker;
import com.akto.threat.backend.dao.MaliciousEventDao;
import com.akto.threat.backend.dao.ThreatDetectionDaoInit;
import com.akto.threat.backend.service.ApiDistributionDataService;
import com.akto.threat.backend.service.MaliciousEventService;
Expand Down Expand Up @@ -72,10 +73,10 @@ public static void main(String[] args) throws Exception {
new FlushMessagesToDB(internalKafkaConfig, threatProtectionMongo).run();

MaliciousEventService maliciousEventService =
new MaliciousEventService(internalKafkaConfig, threatProtectionMongo);
new MaliciousEventService(internalKafkaConfig, MaliciousEventDao.instance);
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

same why create dependency ? we can directly use MaliciousEventDao.instance ?


ThreatActorService threatActorService = new ThreatActorService(threatProtectionMongo);
ThreatApiService threatApiService = new ThreatApiService(threatProtectionMongo);
ThreatActorService threatActorService = new ThreatActorService(threatProtectionMongo, MaliciousEventDao.instance);
ThreatApiService threatApiService = new ThreatApiService(MaliciousEventDao.instance);
ApiDistributionDataService apiDistributionDataService = new ApiDistributionDataService(threatProtectionMongo);

new BackendVerticle(maliciousEventService, threatActorService, threatApiService, apiDistributionDataService).start();
Expand Down
15 changes: 15 additions & 0 deletions ...threat-detection-backend/src/main/java/com/akto/threat/backend/dao/MaliciousEventDao.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,12 @@

import com.akto.dto.threat_detection_backend.MaliciousEventDto;
import com.akto.threat.backend.constants.MongoDBCollection;
import com.mongodb.client.AggregateIterable;
import com.mongodb.client.MongoCollection;
import org.bson.Document;
import org.bson.conversions.Bson;

import java.util.List;

public class MaliciousEventDao extends AccountBasedDao<MaliciousEventDto> {

Expand All @@ -27,4 +32,14 @@ public void insertOne(String accountId, MaliciousEventDto event) {
public MongoCollection<MaliciousEventDto> getCollection(String accountId) {
return super.getCollection(accountId);
}

public AggregateIterable<Document> aggregateRaw(String accountId, List<Document> pipeline) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

simly MaliciosDao.instance.aggregate won't work ?

return getDatabase(accountId)
.getCollection(getCollectionName(), Document.class)
.aggregate(pipeline);
}

public long countDocuments(String accountId, Bson filter) {
return getCollection(accountId).countDocuments(filter);
}
}
79 changes: 26 additions & 53 deletions ...etection-backend/src/main/java/com/akto/threat/backend/service/MaliciousEventService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,30 +17,32 @@
import com.akto.proto.generated.threat_detection.service.malicious_alert_service.v1.RecordMaliciousEventRequest;
import com.akto.threat.backend.constants.KafkaTopic;
import com.akto.threat.backend.constants.MongoDBCollection;
import com.akto.threat.backend.dao.MaliciousEventDao;
import com.akto.threat.backend.utils.KafkaUtils;
import com.akto.threat.backend.utils.ThreatUtils;
import com.akto.threat.backend.constants.StatusConstants;
import com.mongodb.client.*;
import com.mongodb.client.model.Filters;
import com.mongodb.client.DistinctIterable;
import com.mongodb.client.MongoCursor;

import java.util.*;

import com.mongodb.client.model.Filters;
import com.mongodb.client.model.Updates;
import org.bson.Document;
import org.bson.conversions.Bson;

public class MaliciousEventService {

private final Kafka kafka;
private final MongoClient mongoClient;
private final MaliciousEventDao maliciousEventDao;
private static final LoggerMaker logger = new LoggerMaker(MaliciousEventService.class);

private static final HashMap<String, Boolean> shouldNotCreateIndexes = new HashMap<>();

public MaliciousEventService(
KafkaConfig kafkaConfig, MongoClient mongoClient) {
KafkaConfig kafkaConfig, MaliciousEventDao maliciousEventDao) {
this.kafka = new Kafka(kafkaConfig);
this.mongoClient = mongoClient;
this.maliciousEventDao = maliciousEventDao;
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why do we need the Dao as a variable now ? We should directly use MaliciousEventsDao.instance.. ?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

as discussed keeping this

}

// Convert string label to model Label enum
Expand Down Expand Up @@ -147,9 +149,9 @@ public void recordMaliciousEvent(String accountId, RecordMaliciousEventRequest r
KafkaTopic.ThreatDetection.INTERNAL_DB_MESSAGES);
}

private static <T> Set<T> findDistinctFields(
MongoCollection<MaliciousEventDto> coll, String fieldName, Class<T> tClass, Bson filters) {
DistinctIterable<T> r = coll.distinct(fieldName, filters, tClass);
private <T> Set<T> findDistinctFields(
String accountId, String fieldName, Class<T> tClass, Bson filters) {
DistinctIterable<T> r = maliciousEventDao.getCollection(accountId).distinct(fieldName, filters, tClass);
Set<T> result = new HashSet<>();
MongoCursor<T> cursor = r.cursor();
while (cursor.hasNext()) {
Expand All @@ -160,42 +162,28 @@ private static <T> Set<T> findDistinctFields(

public ThreatActorFilterResponse fetchThreatActorFilters(
String accountId, ThreatActorFilterRequest request) {
MongoCollection<MaliciousEventDto> coll =
this.mongoClient
.getDatabase(accountId)
.getCollection("malicious_events", MaliciousEventDto.class);

Set<String> latestAttack =
MaliciousEventService.<String>findDistinctFields(
coll, "filterId", String.class, Filters.empty());
this.findDistinctFields(accountId, "filterId", String.class, Filters.empty());

Set<String> countries =
MaliciousEventService.<String>findDistinctFields(
coll, "country", String.class, Filters.empty());
this.findDistinctFields(accountId, "country", String.class, Filters.empty());

Set<String> actorIds =
MaliciousEventService.<String>findDistinctFields(
coll, "actor", String.class, Filters.empty());
this.findDistinctFields(accountId, "actor", String.class, Filters.empty());

return ThreatActorFilterResponse.newBuilder().addAllSubCategories(latestAttack).addAllCountries(countries).addAllActorId(actorIds).build();
}

public FetchAlertFiltersResponse fetchAlertFilters(
String accountId, FetchAlertFiltersRequest request) {
MongoCollection<MaliciousEventDto> coll =
this.mongoClient
.getDatabase(accountId)
.getCollection("malicious_events", MaliciousEventDto.class);

Set<String> actors =
MaliciousEventService.<String>findDistinctFields(
coll, "actor", String.class, Filters.empty());
this.findDistinctFields(accountId, "actor", String.class, Filters.empty());
Set<String> urls =
MaliciousEventService.<String>findDistinctFields(
coll, "latestApiEndpoint", String.class, Filters.empty());
this.findDistinctFields(accountId, "latestApiEndpoint", String.class, Filters.empty());
Set<String> subCategories =
MaliciousEventService.<String>findDistinctFields(
coll, "filterId", String.class, Filters.empty());
this.findDistinctFields(accountId, "filterId", String.class, Filters.empty());

return FetchAlertFiltersResponse.newBuilder().addAllActors(actors).addAllUrls(urls).addAllSubCategory(subCategories).build();
}
Expand All @@ -216,12 +204,6 @@ public ListMaliciousRequestsResponse listMaliciousRequests(
return ListMaliciousRequestsResponse.newBuilder().build();
}

MongoCollection<MaliciousEventDto> coll =
this.mongoClient
.getDatabase(accountId)
.getCollection(
MongoDBCollection.ThreatDetection.MALICIOUS_EVENTS, MaliciousEventDto.class);

Document query = new Document();
if (!filter.getActorsList().isEmpty()) {
query.append("actor", new Document("$in", filter.getActorsList()));
Expand Down Expand Up @@ -283,9 +265,10 @@ public ListMaliciousRequestsResponse listMaliciousRequests(
applyLabelFilter(query, labelEnum);
}

long total = coll.countDocuments(query);
long total = maliciousEventDao.countDocuments(accountId, query);
try (MongoCursor<MaliciousEventDto> cursor =
coll.find(query)
maliciousEventDao.getCollection(accountId)
.find(query)
.sort(new Document("detectedAt", sort.getOrDefault("detectedAt", -1)))
.skip(skip)
.limit(limit)
Expand Down Expand Up @@ -327,13 +310,12 @@ public ListMaliciousRequestsResponse listMaliciousRequests(
}

public void createIndexIfAbsent(String accountId) {
ThreatUtils.createIndexIfAbsent(accountId, mongoClient);
ThreatUtils.createIndexIfAbsent(accountId, maliciousEventDao);
shouldNotCreateIndexes.put(accountId, true);
}

public int updateMaliciousEventStatus(String accountId, List<String> eventIds, Map<String, Object> filterMap, String status) {
try {
MongoCollection<MaliciousEventDto> coll = getMaliciousEventCollection(accountId);
MaliciousEventDto.Status eventStatus = MaliciousEventDto.Status.valueOf(status.toUpperCase());
Bson update = Updates.set("status", eventStatus.toString());

Expand All @@ -342,11 +324,11 @@ public int updateMaliciousEventStatus(String accountId, List<String> eventIds, M
return 0;
}

String logMessage = String.format("Updating events %s to status: %s",
String logMessage = String.format("Updating events %s to status: %s",
getQueryDescription(eventIds, filterMap), status);
logger.info(logMessage);
long modifiedCount = coll.updateMany(query, update).getModifiedCount();

long modifiedCount = maliciousEventDao.getCollection(accountId).updateMany(query, update).getModifiedCount();
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

With Dao it should directly be Dao.instance.updateMany ???

return (int) modifiedCount;
} catch (Exception e) {
logger.error("Error updating malicious event status", e);
Expand All @@ -356,33 +338,24 @@ public int updateMaliciousEventStatus(String accountId, List<String> eventIds, M

public int deleteMaliciousEvents(String accountId, List<String> eventIds, Map<String, Object> filterMap) {
try {
MongoCollection<MaliciousEventDto> coll = getMaliciousEventCollection(accountId);

Document query = buildQuery(eventIds, filterMap, "delete");
if (query == null) {
return 0;
}

String logMessage = "Deleting events " + getQueryDescription(eventIds, filterMap);
logger.info(logMessage);
long deletedCount = coll.deleteMany(query).getDeletedCount();

long deletedCount = maliciousEventDao.getCollection(accountId).deleteMany(query).getDeletedCount();
logger.info("Deleted " + deletedCount + " malicious events");

return (int) deletedCount;
} catch (Exception e) {
logger.error("Error deleting malicious events", e);
return 0;
}
}

private MongoCollection<MaliciousEventDto> getMaliciousEventCollection(String accountId) {
return this.mongoClient
.getDatabase(accountId)
.getCollection(
MongoDBCollection.ThreatDetection.MALICIOUS_EVENTS, MaliciousEventDto.class);
}

private Document buildQuery(List<String> eventIds, Map<String, Object> filterMap, String operation) {
if (eventIds != null && !eventIds.isEmpty()) {
// Query by event IDs
Expand Down
Loading
Loading