diff --git a/parsedmarc/mail/graph.py b/parsedmarc/mail/graph.py index b8c02bb5..9595a85d 100644 --- a/parsedmarc/mail/graph.py +++ b/parsedmarc/mail/graph.py @@ -93,6 +93,15 @@ def _generate_credential(auth_method: str, token_path: Path, **kwargs): class MSGraphConnection(MailboxConnection): + _WELL_KNOWN_FOLDERS = { + "inbox": "inbox", + "archive": "archive", + "drafts": "drafts", + "sentitems": "sentitems", + "deleteditems": "deleteditems", + "junkemail": "junkemail", + } + def __init__( self, auth_method: str, @@ -267,8 +276,25 @@ class MSGraphConnection(MailboxConnection): parent_folder_id = folder_id return self._find_folder_id_with_parent(path_parts[-1], parent_folder_id) else: + # Shared mailboxes can fail root listing; try well-known folders first. + well_known_folder_id = self._get_well_known_folder_id(folder_name) + if well_known_folder_id: + return well_known_folder_id return self._find_folder_id_with_parent(folder_name, None) + def _get_well_known_folder_id(self, folder_name: str) -> Optional[str]: + folder_key = folder_name.lower().replace(" ", "").replace("-", "") + alias = self._WELL_KNOWN_FOLDERS.get(folder_key) + if alias is None: + return None + + url = f"/users/{self.mailbox_name}/mailFolders/{alias}?$select=id,displayName" + folder_resp = self._client.get(url) + if folder_resp.status_code != 200: + return None + payload = folder_resp.json() + return payload.get("id") + def _find_folder_id_with_parent( self, folder_name: str, parent_folder_id: Optional[str] ): @@ -279,6 +305,10 @@ class MSGraphConnection(MailboxConnection): filter = f"?$filter=displayName eq '{folder_name}'" folders_resp = self._request_with_retries("get", url + filter) if folders_resp.status_code != 200: + if parent_folder_id is None: + well_known_folder_id = self._get_well_known_folder_id(folder_name) + if well_known_folder_id: + return well_known_folder_id raise RuntimeWarning(f"Failed to list folders.{folders_resp.json()}") folders: list = folders_resp.json()["value"] matched_folders = [ diff --git a/tests.py b/tests.py index 4ab6fb15..5941070c 100755 --- a/tests.py +++ b/tests.py @@ -468,6 +468,8 @@ hosts = localhost self.assertEqual(ctx.exception.code, 1) mock_save_aggregate.assert_called_once() mock_save_forensic_opensearch.assert_called_once() + + class _FakeGraphResponse: def __init__(self, status_code, payload=None, text=""): self.status_code = status_code @@ -477,6 +479,7 @@ class _FakeGraphResponse: def json(self): return self._payload + class _BreakLoop(BaseException): pass @@ -1210,5 +1213,52 @@ since = 2d self.assertEqual(system_exit.exception.code, 1) self.assertEqual(mock_watch_inbox.call_args.kwargs.get("since"), "2d") + +class _FakeGraphClient: + def get(self, url, params=None): + if "/mailFolders/inbox?$select=id,displayName" in url: + return _FakeGraphResponse(200, {"id": "inbox-id", "displayName": "Inbox"}) + + if "/mailFolders?$filter=displayName eq 'Inbox'" in url: + return _FakeGraphResponse( + 404, + { + "error": { + "code": "ErrorItemNotFound", + "message": "Default folder Root not found.", + } + }, + ) + + if "/mailFolders?$filter=displayName eq 'Custom'" in url: + return _FakeGraphResponse( + 404, + { + "error": { + "code": "ErrorItemNotFound", + "message": "Default folder Root not found.", + } + }, + ) + + return _FakeGraphResponse(404, {"error": {"code": "NotFound"}}) + + +class TestMSGraphFolderFallback(unittest.TestCase): + def testWellKnownFolderFallback(self): + connection = MSGraphConnection.__new__(MSGraphConnection) + connection.mailbox_name = "shared@example.com" + connection._client = _FakeGraphClient() + + folder_id = connection._find_folder_id_from_folder_path("Inbox") + self.assertEqual(folder_id, "inbox-id") + + def testUnknownFolderStillFails(self): + connection = MSGraphConnection.__new__(MSGraphConnection) + connection.mailbox_name = "shared@example.com" + connection._client = _FakeGraphClient() + + with self.assertRaises(RuntimeWarning): + connection._find_folder_id_from_folder_path("Custom") if __name__ == "__main__": unittest.main(verbosity=2)