Merge branch 'release/3.5.10'

Author: nusantara-self

CHANGELOG.md

@@ -1,5 +1,21 @@
 # Changelog
 
+## [3.5.10](https://github.com/TheHive-Project/Cortex-Analyzers/tree/3.5.10) (2025-04-11)
+
+[Full Changelog](https://github.com/TheHive-Project/Cortex-Analyzers/compare/3.5.9...3.5.10)
+
+**Closed issues:**
+
+- \[Bug\] MSEntraID - User Principal Names may produce invalid key [\#1346](https://github.com/TheHive-Project/Cortex-Analyzers/issues/1346)
+
+## [3.5.9](https://github.com/TheHive-Project/Cortex-Analyzers/tree/3.5.9) (2025-04-04)
+
+[Full Changelog](https://github.com/TheHive-Project/Cortex-Analyzers/compare/3.5.8...3.5.9)
+
+**Merged pull requests:**
+
+- Various improvements & fixes [\#1345](https://github.com/TheHive-Project/Cortex-Analyzers/pull/1345) ([nusantara-self](https://github.com/nusantara-self))
+
 ## [3.5.8](https://github.com/TheHive-Project/Cortex-Analyzers/tree/3.5.8) (2025-03-31)
 
 [Full Changelog](https://github.com/TheHive-Project/Cortex-Analyzers/compare/3.5.7...3.5.8)

analyzers/MSEntraID/MSEntraID.py

@@ -38,6 +38,26 @@ def authenticate(self):
 
         return token_r.json().get('access_token')
 
+    def resolve_user_guid(self, email, headers, base_url):
+        """Resolves a userPrincipalName (email) to an objectId (GUID) using direct lookup (most compatible)."""
+        url = f"{base_url}users/{email}?$select=id"
+        response = requests.get(url, headers=headers)
+
+        if response.status_code != 200:
+            self.error(f"Failed to resolve GUID for user {email}: {response.content}")
+
+        user = response.json()
+        user_id = user.get("id")
+        if not user_id:
+            self.error(f"ID not found in response for {email}")
+
+        return user_id
+
+    def ensure_user_guid(self, base_url, headers):
+        if "@" in self.user:
+            return self.resolve_user_guid(self.user, headers, base_url)
+        return self.user
+
     def handle_get_signins(self, headers, base_url):
         """
         Retrieve sign-in logs for a userPrincipalName within a specified time range.
@@ -49,14 +69,15 @@ def handle_get_signins(self, headers, base_url):
             self.user = self.get_data()
             if not self.user:
                 self.error("No user supplied")
-
+            self.guid = self.ensure_user_guid(base_url, headers)
+            
             # Build the filter time
             filter_time = datetime.utcnow() - timedelta(days=self.time_range)
             format_time = filter_time.strftime('%Y-%m-%dT00:00:00Z')
 
             # Query sign-in logs
             endpoint = (
-                f"auditLogs/signIns?$filter=startsWith(userPrincipalName,'{self.user}') "
+                f"auditLogs/signIns?$filter=userId eq '{self.guid}'"
                 f"and createdDateTime ge {format_time}&$top={self.lookup_limit}"
             )
             r = requests.get(base_url + endpoint, headers=headers)
@@ -168,13 +189,14 @@ def handle_get_userinfo(self, headers, base_url):
             self.user = self.get_data()
             if not self.user:
                 self.error("No user supplied")
-
+            
+            self.guid = self.ensure_user_guid(base_url, headers)
             # Use select to retrieve many user attributes. Adjust as needed.
             params = {
                         "$select": ",".join(self.params_list)
                     }
 
-            user_info_url = f"{base_url}users/{self.user}"
+            user_info_url = f"{base_url}users/{self.guid}"
 #            user_info_url = f"{base_url}users/{self.user}"
 
             user_response = requests.get(user_info_url, headers=headers, params=params)
@@ -211,7 +233,7 @@ def handle_get_userinfo(self, headers, base_url):
             }
 
             # Fetch user's manager
-            manager_url = f"{base_url}users/{self.user}/manager?$select=id,displayName,userPrincipalName"
+            manager_url = f"{base_url}users/{self.guid}/manager?$select=id,displayName,userPrincipalName"
             manager_resp = requests.get(manager_url, headers=headers)
             if manager_resp.status_code == 200:
                 manager_data = manager_resp.json()
@@ -224,7 +246,7 @@ def handle_get_userinfo(self, headers, base_url):
                     }
 
             # Fetch user's license details
-            license_url = f"{base_url}users/{self.user}/licenseDetails"
+            license_url = f"{base_url}users/{self.guid}/licenseDetails"
             license_resp = requests.get(license_url, headers=headers)
             if license_resp.status_code == 200:
                 license_data = license_resp.json().get("value", [])
@@ -238,7 +260,7 @@ def handle_get_userinfo(self, headers, base_url):
                     })
 
             # Fetch user's group memberships
-            member_of_url = f"{base_url}users/{self.user}/memberOf"
+            member_of_url = f"{base_url}users/{self.guid}/memberOf"
             member_of_response = requests.get(member_of_url, headers=headers)
             if member_of_response.status_code == 200:
                 memberships = member_of_response.json().get("value", [])
@@ -249,7 +271,7 @@ def handle_get_userinfo(self, headers, base_url):
                     })
 
             # MFA Methods
-            mfa_url = f"{base_url}users/{self.user}/authentication/methods"
+            mfa_url = f"{base_url}users/{self.guid}/authentication/methods"
             mfa_r = requests.get(mfa_url, headers=headers)
 
             if mfa_r.status_code == 200:
@@ -374,9 +396,11 @@ def handle_get_directoryAuditLogs(self, headers, base_url):
             self.error('Incorrect dataType. "mail" expected.')
         try:
             # Pull the userPrincipalName from the observable data (data_type=mail)
-            user_upn = self.get_data()
-            if not user_upn:
+            self.user = self.get_data()
+            if not self.user:
                 self.error("No user principal name supplied for directory audit logs")
+            self.guid = self.ensure_user_guid(base_url, headers)
+            
             # Calculate time range (past X days)
             filter_time = datetime.utcnow() - timedelta(days=self.time_range)
             filter_time_str = filter_time.strftime('%Y-%m-%dT%H:%M:%SZ')
@@ -386,7 +410,7 @@ def handle_get_directoryAuditLogs(self, headers, base_url):
             endpoint = (
                 "auditLogs/directoryAudits?"
                 f"$filter=activityDateTime ge {filter_time_str} "
-                f"and initiatedBy/user/userPrincipalName eq '{user_upn}'"
+                f"and initiatedBy/user/id eq '{self.guid}'"
                 f"&$top={self.lookup_limit}"
             )
 
@@ -435,16 +459,19 @@ def handle_get_devices(self, headers, base_url):
                 else:
                     self.error("No user UPN supplied")
 
-            # Build the appropriate endpoint based on the observable type
-            if self.data_type == 'hostname':
+            if self.data_type == 'mail':
+                # Resolve UPN to GUID and use exact match
+                self.user = query_value
+                guid = self.ensure_user_guid(base_url, headers)
                 endpoint = (
-                    "deviceManagement/managedDevices?"
-                    f"$filter=startswith(deviceName,'{query_value}')"
+                    f"deviceManagement/managedDevices?"
+                    f"$filter=userId eq '{guid}'"
                 )
-            elif self.data_type == 'mail':
+            else:
+                # Use startswith for partial hostname matches
                 endpoint = (
-                    "deviceManagement/managedDevices?"
-                    f"$filter=startswith(userPrincipalName,'{query_value}')"
+                    f"deviceManagement/managedDevices?"
+                    f"$filter=startswith(deviceName,'{query_value}')"
                 )
 
             # Perform the GET request

responders/MSEntraID/MSEntraID.py

@@ -32,8 +32,29 @@ def authenticate(self):
 
         return token_r.json().get('access_token')
 
+    def resolve_user_guid(self, email, headers, base_url):
+        """Resolves a userPrincipalName (email) to an objectId (GUID) using direct lookup (most compatible)."""
+        url = f"{base_url}users/{email}?$select=id"
+        response = requests.get(url, headers=headers)
+
+        if response.status_code != 200:
+            self.error(f"Failed to resolve GUID for user {email}: {response.content}")
+
+        user = response.json()
+        user_id = user.get("id")
+        if not user_id:
+            self.error(f"ID not found in response for {email}")
+
+        return user_id
+
+    def ensure_guid(self, headers, base_url):
+        if "@" in self.user:
+            self.guid = self.resolve_user_guid(self.user, headers, base_url)
+        else:
+            self.guid = self.user
+
     def check_user_status(self, user, headers, base_url):
-        r = requests.get(f"{base_url}{user}?$select=accountEnabled", headers=headers)
+        r = requests.get(f"{base_url}users/{user}?$select=accountEnabled", headers=headers)
 
         if r.status_code == 404:
             self.error(f'User {user} not found in Microsoft Entra ID')
@@ -58,16 +79,17 @@ def run(self):
         Responder.run(self)
         token = self.authenticate()
         headers = {'Authorization': f'Bearer {token}', 'Content-Type': 'application/json'}
-        base_url = 'https://graph.microsoft.com/v1.0/users/'
+        base_url = 'https://graph.microsoft.com/v1.0/'
 
         if self.service == "revokeSignInSessions":
             if self.get_param('data.dataType') == 'mail':
                 try:
                     self.user = self.get_param('data.data', None, 'No UPN supplied to revoke credentials for')
                     if not self.user:
                         self.error("No user supplied")
-                    
-                    r = requests.post(f"{base_url}{self.user}/revokeSignInSessions", headers=headers)
+                    self.ensure_guid(headers, base_url)
+
+                    r = requests.post(f"{base_url}users/{self.guid}/revokeSignInSessions", headers=headers)
 
                     if r.status_code != 200:
                         self.error(f'Failure to revoke access tokens of user {self.user}: {r.content}')
@@ -87,9 +109,10 @@ def run(self):
                     self.user = self.get_param('data.data', None, 'No UPN supplied for password reset')
                     if not self.user:
                         self.error("No user supplied")
-                    
+                    self.ensure_guid(headers, base_url)
+                                            
                     data = {"passwordProfile": {"forceChangePasswordNextSignIn": True}}
-                    r = requests.patch(f"{base_url}{self.user}", headers=headers, json=data)
+                    r = requests.patch(f"{base_url}users/{self.guid}", headers=headers, json=data)
 
                     if r.status_code != 204:
                         self.error(f'Failure to reset password for user {self.user}: {r.content}')
@@ -105,9 +128,10 @@ def run(self):
                 self.user = self.get_param('data.data', None, 'No UPN supplied for password reset with MFA')
                 if not self.user:
                     self.error("No user supplied")
-                
+                self.ensure_guid(headers, base_url)
+
                 data = {"passwordProfile": {"forceChangePasswordNextSignIn": True, "forceChangePasswordNextSignInWithMfa": True}}
-                r = requests.patch(f"{base_url}{self.user}", headers=headers, json=data)
+                r = requests.patch(f"{base_url}users/{self.guid}", headers=headers, json=data)
 
                 if r.status_code != 204:
                     self.error(f'Failure to reset password with MFA for user {self.user}: {r.content}')
@@ -122,14 +146,15 @@ def run(self):
                     self.user = self.get_param('data.data', None, 'No UPN supplied to enable user')
                     if not self.user:
                         self.error("No user supplied")
-                    
-                    user_status = self.check_user_status(self.user, headers, base_url)
+                    self.ensure_guid(headers, base_url)
+
+                    user_status = self.check_user_status(self.guid, headers, base_url)
                     if user_status is True:
                         self.report({"message": f"User {self.user} is already enabled"})
                         return
 
                     data = {"accountEnabled": True}
-                    r = requests.patch(f"{base_url}{self.user}", headers=headers, json=data)
+                    r = requests.patch(f"{base_url}users/{self.guid}", headers=headers, json=data)
 
                     if r.status_code != 204:
                         self.error(f'Failure to enable user {self.user}: {r.content}')
@@ -147,14 +172,15 @@ def run(self):
                     self.user = self.get_param('data.data', None, 'No UPN supplied to disable user')
                     if not self.user:
                         self.error("No user supplied")
-                    
-                    user_status = self.check_user_status(self.user, headers, base_url)
+                    self.ensure_guid(headers, base_url)
+
+                    user_status = self.check_user_status(self.guid, headers, base_url)
                     if user_status is False:
                         self.report({"message": f"User {self.user} is already disabled"})
                         return
 
                     data = {"accountEnabled": False}
-                    r = requests.patch(f"{base_url}{self.user}", headers=headers, json=data)
+                    r = requests.patch(f"{base_url}users/{self.guid}", headers=headers, json=data)
 
                     if r.status_code != 204:
                         self.error(f'Failure to disable user {self.user}: {r.content}')