Skip to content

Instantly share code, notes, and snippets.

Created February 25, 2019 23:21
  • Star 0 You must be signed in to star a gist
  • Fork 1 You must be signed in to fork a gist
Star You must be signed in to star a gist
What would you like to do?
import argparse
from datetime import datetime
from getpass import getpass
import requests
AAD_TENANT = "<tenant>"
AAD_CLIENT_ID = "<app ID>"
class MicrosoftGraphClient():
def __init__(self):
self._aad_client_secret = getpass("Client Secret for {}:".format(AAD_CLIENT_ID))
data = {
"client_id": AAD_CLIENT_ID,
"scope": "",
"client_secret": self._aad_client_secret,
"grant_type": "client_credentials",
url = "{}/oauth2/v2.0/token".format(AAD_TENANT)
self._session = requests.Session()
r =, data=data)
if r.status_code == 200:
self._token = r.json()['access_token']
raise Exception("Failed to get OAuth Token: {}".format(r.json()))
def _get(self, url, params={}, header={}):
header['Authorization'] = "Bearer {}".format(self._token)
return self._session.get(url, params=params, headers=header)
def _get_values(self, endpoint, params={}):
url = "{}{}".format(AAD_GRAPH_SERVICE, endpoint)
r = self._get(url, params)
if r.status_code != 200:
raise Exception("Status code {} returned error: {}".format(r.status_code, r.json()))
total_count = 0
while True:
count = len(r.json()['value'])
total_count = total_count + count
print("Processing {}/{}".format(count, total_count))
yield from r.json()['value']
if '@odata.nextLink' in r.json().keys():
url = r.json()['@odata.nextLink']
r = self._get(url)
def _save_value(self, endpoint, filename):
r = self._get(endpoint)
if r.status_code != 200:
raise Exception("Status code {} returned error: {}".format(r.status_code, r.json()))
with open(filename, 'wb') as f:
def download_eml(self, upn, subject):
params = {
"$select": "id,receivedDateTime,from,subject",
"$filter": "contains(subject,'{}')".format(subject),
for r in self._get_values("/v1.0/users/{}/messages".format(upn), params=params):
dt = datetime.strptime(r['receivedDateTime'], "%Y-%m-%dT%H:%M:%SZ")
r = self._save_value(
"{}/messages/{}/$value".format(upn, r['id']),
"{}-{}.eml".format(upn, dt.strftime("%Y-%m-%d_%H_%M_%S"))
if __name__ == "__main__":
parser = argparse.ArgumentParser(description='Download .eml message content from the Microsoft Graph API.')
help="The mailbox to search (usually the user's email address)"
help="The subject to search for (can be partial)"
args = parser.parse_args()
client = MicrosoftGraphClient()
client.download_eml(args.upn, args.subject)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment