Add get_user to AWS datamodel

data_models/aws_cloudtrail_data_model.py

@@ -37,3 +37,38 @@ def load_ip_address(event):
         except ipaddress.AddressValueError:
             return None
     return source_ip
+
+
+# get actor user from correct field based on identity type
+# https://docs.aws.amazon.com/awscloudtrail/latest/userguide/cloudtrail-event-reference-user-identity.html#cloudtrail-event-reference-user-identity-fields
+def get_actor_user(event):
+    user_type = deep_get(event, "userIdentity", "type")
+    if event.get("eventType") == "AwsServiceEvent":
+        actor_user = deep_get(event, "userIdentity", "invokedBy", default="UnknownAwsServiceEvent")
+    elif user_type == "Root":
+        actor_user = deep_get(
+            event,
+            "userIdentity",
+            "userName",
+            default=deep_get(event, "userIdentity", "accountId", default="UnknownRootUser"),
+        )
+    elif user_type in ("IAMUser", "Directory", "Unknown", "SAMLUser", "WebIdentityUser"):
+        actor_user = deep_get(event, "userIdentity", "userName", default=f"Unknown{user_type}")
+    elif user_type in ("AssumedRole", "Role", "FederatedUser"):
+        actor_user = deep_get(
+            event,
+            "userIdentity",
+            "sessionContext",
+            "sessionIssuer",
+            "userName",
+            default=f"Unknown{user_type}",
+        )
+    elif user_type == "IdentityCenterUser":
+        actor_user = deep_get(
+            event, "additionalEventData", "UserName", default=f"Unknown{user_type}"
+        )
+    elif user_type in ("AWSService", "AWSAccount"):
+        actor_user = event.get("sourceIdentity", f"Unknown{user_type}")
+    else:
+        actor_user = "UnknownUser"
+    return actor_user

data_models/aws_cloudtrail_data_model.yml

@@ -7,7 +7,7 @@ Filename: aws_cloudtrail_data_model.py
 Enabled: true
 Mappings:
   - Name: actor_user
-    Path: $.userIdentity..userName
+    Method: get_actor_user
   - Name: event_type
     Method: get_event_type
   - Name: source_ip