feat(stream): Microsoft 365 directory-audit event type and deserializer #1603
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -3,6 +3,7 @@ | |
| public enum DataEnrichmentServiceType { | ||
|
|
||
| INTEGRATED_TOOLS_EVENTS, | ||
| RMM_RESULTS, | ||
| PRE_ENRICHED | ||
|
|
||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,145 @@ | ||
| package com.openframe.stream.deserializer; | ||
|
|
||
| import com.fasterxml.jackson.databind.JsonNode; | ||
| import com.fasterxml.jackson.databind.ObjectMapper; | ||
| import com.fasterxml.jackson.databind.node.ObjectNode; | ||
| import com.openframe.data.cassandra.model.enums.UnifiedEventType; | ||
| import com.openframe.data.model.enums.MessageType; | ||
| import com.openframe.kafka.model.debezium.CommonDebeziumMessage; | ||
| import com.openframe.stream.mapping.EventTypeMapper; | ||
| import com.openframe.stream.model.fleet.debezium.DeserializedDebeziumMessage; | ||
| import lombok.RequiredArgsConstructor; | ||
| import lombok.extern.slf4j.Slf4j; | ||
| import org.apache.commons.lang3.StringUtils; | ||
| import org.springframework.stereotype.Component; | ||
|
|
||
| import java.time.Instant; | ||
| import java.time.ZoneId; | ||
| import java.time.format.DateTimeFormatter; | ||
| import java.util.Optional; | ||
|
|
||
| /** | ||
| * Deserializes Google Workspace directory audit events polled from the Admin SDK Reports API | ||
| * {@code activities.list("admin")} endpoint. Events are hand-built by the poller (not CDC): a | ||
| * single polled activity's {@code events[]} array is fanned out by the poller into one Kafka | ||
| * record per event, so this deserializer stays 1:1 with {@link KafkaMessageDeserializer#deserialize} | ||
| * (that interface is single-result). The poller writes a FLAT {@code after} object per event with | ||
| * fields {@code uniqueQualifier}, {@code eventIndex}, {@code activityTime}, {@code eventType}, | ||
| * {@code eventName}, {@code actorEmail}, {@code ipAddress}, {@code event} (nested JSON carrying the | ||
| * raw event's {@code parameters}), plus tenant/organization passthrough fields ({@code tenantId}, | ||
| * {@code organizationId}, {@code organizationName}) and multi-connection fields | ||
| * ({@code connectionId}, {@code connectionName}) — hence | ||
| * {@link com.openframe.data.model.enums.DataEnrichmentServiceType#PRE_ENRICHED}. {@code toolEventId} | ||
| * is {@code uniqueQualifier + "-" + eventIndex}: Reports API activities are uniquely identified by | ||
| * {@code uniqueQualifier}, but a single activity can carry multiple events, so the pair keeps | ||
| * replays from the poller's cursor overlap window upsert idempotent per event. Events carry no | ||
| * agent reference. | ||
| * {@code connectionId}/{@code connectionName} (multi-connection orgs) are passed through into details. | ||
| * <p> | ||
| * {@code event.parameters[]} is an UNTOUCHED passthrough of the Reports API parameter union: each | ||
| * entry always has {@code name}, but its value key varies — {@code value} (string), | ||
| * {@code boolValue}, {@code intValue}, {@code multiValue} (array of strings) or | ||
| * {@code multiMessageValue}. The write-audit publisher (saas-lib | ||
| * {@code GoogleWorkspaceWriteAuditPublisher}) emits only the string {@code {name,value}} member of | ||
| * that union. Consumers rendering details must read | ||
| * {@code value ?? boolValue ?? intValue ?? multiValue} for the scalar/array members; nothing may | ||
| * assume {@code value} alone. {@code multiMessageValue} is deliberately NOT part of that fallback | ||
| * chain — it is an array of nested {@code {parameter: [{name, value, ...}]}} objects, so a consumer | ||
| * that needs it must render it recursively rather than coerce it to a scalar. | ||
| */ | ||
| @Slf4j | ||
| @Component | ||
| @RequiredArgsConstructor | ||
| public class GoogleWorkspaceAuditEventDeserializer implements KafkaMessageDeserializer { | ||
|
|
||
| private static final DateTimeFormatter DAY_FORMATTER = | ||
| DateTimeFormatter.ofPattern("yyyy-MM-dd").withZone(ZoneId.of("UTC")); | ||
| // Reports API does not carry a structured success/failure result field; failure is signalled | ||
| // by eventName itself (e.g. LOGIN_FAILURE, login_failure). | ||
| private static final String FAILURE_MARKER = "_FAILURE"; | ||
| private static final String UNKNOWN = "unknown"; | ||
|
|
||
| private final ObjectMapper mapper; | ||
|
|
||
| @Override | ||
| public MessageType getType() { | ||
| return MessageType.GOOGLE_WORKSPACE_AUDIT_EVENT; | ||
| } | ||
|
|
||
| @Override | ||
| public DeserializedDebeziumMessage deserialize(CommonDebeziumMessage debeziumMessage, MessageType messageType) { | ||
| JsonNode after = debeziumMessage.getPayload().getAfter(); | ||
| if (after == null || after.isNull()) { | ||
| return null; | ||
| } | ||
| long eventTimestamp = getEventTimestamp(after) | ||
| .orElse(debeziumMessage.getPayload().getTimestamp()); | ||
| String eventType = textField(after, "eventType").orElse(UNKNOWN); | ||
|
|
||
| return DeserializedDebeziumMessage.builder() | ||
| .payload(debeziumMessage.getPayload()) | ||
| .agentId(null) | ||
| .ingestDay(DAY_FORMATTER.format(Instant.ofEpochMilli(eventTimestamp))) | ||
| .sourceEventType(eventType) | ||
| .toolEventId(buildToolEventId(after)) | ||
| .unifiedEventType(resolveEventType(after, eventType)) | ||
| .message(textField(after, "eventName").orElse(null)) | ||
| .integratedToolType(messageType.getIntegratedToolType()) | ||
| .debeziumMessage(after.toString()) | ||
| .details(buildDetails(after)) | ||
| .eventTimestamp(eventTimestamp) | ||
| .skipProcessing(false) | ||
| .isVisible(true) | ||
| .tenantId(textField(after, "tenantId").orElse(null)) | ||
| .organizationId(textField(after, "organizationId").orElse(null)) | ||
| .organizationName(textField(after, "organizationName").orElse(null)) | ||
| .userId(textField(after, "actorEmail").orElse(null)) | ||
| .build(); | ||
| } | ||
|
|
||
| private UnifiedEventType resolveEventType(JsonNode after, String eventType) { | ||
| String eventName = textField(after, "eventName").orElse(""); | ||
| if (StringUtils.containsIgnoreCase(eventName, FAILURE_MARKER)) { | ||
| return UnifiedEventType.GWS_AUDIT_FAILURE; | ||
| } | ||
| UnifiedEventType mapped = EventTypeMapper.mapToUnifiedType(getType().getIntegratedToolType(), eventType); | ||
| return mapped == UnifiedEventType.UNKNOWN ? UnifiedEventType.GWS_AUDIT_OTHER : mapped; | ||
| } | ||
|
|
||
| private String buildToolEventId(JsonNode after) { | ||
| return textField(after, "uniqueQualifier") | ||
| .map(uniqueQualifier -> uniqueQualifier + "-" + textField(after, "eventIndex").orElse("0")) | ||
| .orElse(null); | ||
| } | ||
|
|
||
| private Optional<Long> getEventTimestamp(JsonNode after) { | ||
| return textField(after, "activityTime") | ||
| .flatMap(value -> { | ||
| try { | ||
| return Optional.of(Instant.parse(value).toEpochMilli()); | ||
| } catch (Exception e) { | ||
| log.warn("Unparseable activityTime '{}', falling back to processing timestamp", value); | ||
| return Optional.empty(); | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| private String buildDetails(JsonNode after) { | ||
| ObjectNode details = mapper.createObjectNode(); | ||
| JsonNode event = after.get("event"); | ||
| if (event != null && !event.isNull()) { | ||
| details.set("event", event); | ||
| } | ||
| textField(after, "ipAddress").ifPresent(value -> details.put("ipAddress", value)); | ||
| textField(after, "connectionId").ifPresent(value -> details.put("connectionId", value)); | ||
| textField(after, "connectionName").ifPresent(value -> details.put("connectionName", value)); | ||
| return details.toString(); | ||
| } | ||
|
|
||
| private Optional<String> textField(JsonNode node, String fieldName) { | ||
| return Optional.ofNullable(node.get(fieldName)) | ||
| .filter(field -> !field.isNull()) | ||
| .map(JsonNode::asText) | ||
| .filter(StringUtils::isNotBlank); | ||
| } | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,131 @@ | ||
| package com.openframe.stream.deserializer; | ||
|
|
||
| import com.fasterxml.jackson.databind.JsonNode; | ||
| import com.fasterxml.jackson.databind.ObjectMapper; | ||
| import com.fasterxml.jackson.databind.node.ObjectNode; | ||
| import com.openframe.data.cassandra.model.enums.UnifiedEventType; | ||
| import com.openframe.data.model.enums.MessageType; | ||
| import com.openframe.kafka.model.debezium.CommonDebeziumMessage; | ||
| import com.openframe.stream.mapping.EventTypeMapper; | ||
| import com.openframe.stream.model.fleet.debezium.DeserializedDebeziumMessage; | ||
| import lombok.RequiredArgsConstructor; | ||
| import lombok.extern.slf4j.Slf4j; | ||
| import org.apache.commons.lang3.StringUtils; | ||
| import org.springframework.stereotype.Component; | ||
|
|
||
| import java.time.Instant; | ||
| import java.time.ZoneId; | ||
| import java.time.format.DateTimeFormatter; | ||
| import java.util.Optional; | ||
| import java.util.Set; | ||
|
|
||
| /** | ||
| * Deserializes Microsoft 365 Entra directory audit events polled from Graph | ||
| * {@code auditLogs/directoryAudits}. Events are hand-built by the poller (not CDC), arrive | ||
| * pre-enriched with tenant/organization fields in the payload and carry no agent reference — | ||
| * hence {@link com.openframe.data.model.enums.DataEnrichmentServiceType#PRE_ENRICHED}. | ||
| * {@code toolEventId} is the Graph audit record id, so replays from the poller's cursor | ||
| * overlap window upsert idempotently. | ||
| * {@code connectionId}/{@code connectionName} (multi-connection orgs) are passed through into details. | ||
| */ | ||
| @Slf4j | ||
| @Component | ||
| @RequiredArgsConstructor | ||
| public class Microsoft365AuditEventDeserializer implements KafkaMessageDeserializer { | ||
|
|
||
| private static final DateTimeFormatter DAY_FORMATTER = | ||
| DateTimeFormatter.ofPattern("yyyy-MM-dd").withZone(ZoneId.of("UTC")); | ||
| // Graph result values that mean the operation did not succeed (result can be | ||
| // success | failure | timeout | unknownFutureValue). | ||
| private static final Set<String> FAILED_RESULTS = Set.of("failure", "timeout"); | ||
| private static final String UNKNOWN = "unknown"; | ||
|
|
||
| private final ObjectMapper mapper; | ||
|
|
||
| @Override | ||
| public MessageType getType() { | ||
| return MessageType.MICROSOFT_365_AUDIT_EVENT; | ||
| } | ||
|
|
||
| @Override | ||
| public DeserializedDebeziumMessage deserialize(CommonDebeziumMessage debeziumMessage, MessageType messageType) { | ||
| JsonNode after = debeziumMessage.getPayload().getAfter(); | ||
| if (after == null || after.isNull()) { | ||
| return null; | ||
| } | ||
| long eventTimestamp = getEventTimestamp(after) | ||
| .orElse(debeziumMessage.getPayload().getTimestamp()); | ||
| String category = textField(after, "category").orElse(UNKNOWN); | ||
|
|
||
| return DeserializedDebeziumMessage.builder() | ||
| .payload(debeziumMessage.getPayload()) | ||
| .agentId(null) | ||
| .ingestDay(DAY_FORMATTER.format(Instant.ofEpochMilli(eventTimestamp))) | ||
| .sourceEventType(category) | ||
| .toolEventId(textField(after, "auditId").orElse(null)) | ||
|
coderabbitai[bot] marked this conversation as resolved.
|
||
| .unifiedEventType(resolveEventType(after, category)) | ||
| .message(textField(after, "activityDisplayName").orElse(null)) | ||
| .integratedToolType(messageType.getIntegratedToolType()) | ||
| .debeziumMessage(after.toString()) | ||
| .details(buildDetails(after)) | ||
| .eventTimestamp(eventTimestamp) | ||
| .skipProcessing(false) | ||
| .isVisible(true) | ||
| .tenantId(textField(after, "tenantId").orElse(null)) | ||
| .organizationId(textField(after, "organizationId").orElse(null)) | ||
| .organizationName(textField(after, "organizationName").orElse(null)) | ||
| .userId(textPath(after.path("initiatedBy").path("user"), "userPrincipalName")) | ||
| .build(); | ||
| } | ||
|
|
||
| private UnifiedEventType resolveEventType(JsonNode after, String category) { | ||
| if (textField(after, "result").map(String::toLowerCase).filter(FAILED_RESULTS::contains).isPresent()) { | ||
| return UnifiedEventType.M365_AUDIT_FAILURE; | ||
| } | ||
| UnifiedEventType mapped = EventTypeMapper.mapToUnifiedType(getType().getIntegratedToolType(), category); | ||
| return mapped == UnifiedEventType.UNKNOWN ? UnifiedEventType.M365_AUDIT_OTHER : mapped; | ||
| } | ||
|
|
||
| private Optional<Long> getEventTimestamp(JsonNode after) { | ||
| return textField(after, "activityDateTime") | ||
| .flatMap(value -> { | ||
| try { | ||
| return Optional.of(Instant.parse(value).toEpochMilli()); | ||
| } catch (Exception e) { | ||
| log.warn("Unparseable activityDateTime '{}', falling back to processing timestamp", value); | ||
| return Optional.empty(); | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| private String buildDetails(JsonNode after) { | ||
| ObjectNode details = mapper.createObjectNode(); | ||
| JsonNode initiatedBy = after.get("initiatedBy"); | ||
| if (initiatedBy != null && !initiatedBy.isNull()) { | ||
| details.set("initiatedBy", initiatedBy); | ||
| } | ||
| JsonNode targetResources = after.get("targetResources"); | ||
| if (targetResources != null && !targetResources.isNull()) { | ||
| details.set("targetResources", targetResources); | ||
| } | ||
| JsonNode additionalDetails = after.get("additionalDetails"); | ||
| if (additionalDetails != null && !additionalDetails.isNull()) { | ||
| details.set("additionalDetails", additionalDetails); | ||
| } | ||
| textField(after, "connectionId").ifPresent(value -> details.put("connectionId", value)); | ||
| textField(after, "connectionName").ifPresent(value -> details.put("connectionName", value)); | ||
| return details.toString(); | ||
|
coderabbitai[bot] marked this conversation as resolved.
|
||
| } | ||
|
|
||
| private Optional<String> textField(JsonNode node, String fieldName) { | ||
| return Optional.ofNullable(node.get(fieldName)) | ||
| .filter(field -> !field.isNull()) | ||
| .map(JsonNode::asText) | ||
| .filter(StringUtils::isNotBlank); | ||
| } | ||
|
|
||
| private String textPath(JsonNode node, String fieldName) { | ||
| String value = node.path(fieldName).asText(null); | ||
| return StringUtils.isNotBlank(value) ? value : null; | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@Arsenii-Malov seams we get more and more events non related to the tools.
If we need to refactor events?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think that all of them are events and it doesn't really matter whether they are tools or not
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
🥇