Created
June 12, 2017 16:20
-
-
Save dmahugh/bdd2abb9f00c34bdb1d8c1707aed4ba7 to your computer and use it in GitHub Desktop.
This sample is part of our exploration of various approaches to starter samples for working with Microsoft Graph. It requires Requests, Bottle, and Python 3.x. It's a work in progress - we'll be publishing a more comprehensive version with complete documentation soon, but posting this here for anyone who'd like to try it out.
This file contains 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
"""Copyright (c) Microsoft Corporation. All rights reserved. | |
Licensed under the MIT License. | |
""" | |
import base64 | |
import json | |
import os | |
import time | |
import urllib.parse | |
import uuid | |
import bottle | |
from bottle import redirect, request, route, template, view | |
import requests | |
class Connect(object): | |
"""Auth handler for AAD/MSA authentication.""" | |
def __init__(self): | |
self.app_id = 'your_app_id_here' | |
self.app_secret = 'your_app_secret_here' | |
self.state = '' | |
self.access_token = None | |
self.loggedin = False | |
self.loggedin_name = '' | |
self.loggedin_email = '' | |
self.loggedin_photo = None | |
def authorized(self): | |
"""Use received authorization code to request a token from Azure AD.""" | |
# Verify that this authorization attempt came from this app, by checking | |
# the received state, which should match the state (uuid) sent with our | |
# authorization request. | |
if self.state != request.query.state: | |
raise Exception(' -> SHUTTING DOWN: state mismatch' + \ | |
'\n\nState SENT: {0}\n\nState RECEIVED: {1}'. \ | |
format(str(self.state), str(request.query.state))) | |
self.state = '' # reset session state to prevent re-use | |
# try to fetch an access token | |
token_response = self.fetch_token(request.query.code) | |
if not token_response: | |
# no response | |
print(' -> auth.Connect: request for access token failed') | |
return redirect('/') | |
if not token_response.ok: | |
# error response | |
return token_response.text | |
# set properties for current authenticated user | |
me_response = self.get('me') | |
me_data = me_response.json() | |
if 'error' in me_data: | |
print(' -> auth.Connect: /me endpoint returned an error ... ' + \ | |
str(me_data)) | |
self.loggedin_name = me_data['displayName'] | |
self.loggedin_email = me_data['userPrincipalName'] | |
# save profile photo | |
profile_pic = self.get('me/photo/$value', stream=True) | |
if profile_pic.ok: | |
self.loggedin_photo = base64.b64encode(profile_pic.raw.read()) | |
else: | |
self.loggedin_photo = None # no profile photo available | |
return redirect('/') | |
def fetch_token(self, authcode): | |
"""attempt to fetch an access token, using specified authorization code. | |
""" | |
response = requests.post( \ | |
'https://login.microsoftonline.com/common/oauth2/v2.0/token', \ | |
data=dict(client_id=self.app_id, | |
client_secret=self.app_secret, | |
grant_type='authorization_code', | |
code=authcode, | |
redirect_uri='http://localhost:5000/login/authorized')) | |
if self.save_token(response): | |
return response # valid token returned and saved | |
else: | |
return None # token request failed | |
def get(self, endpoint, stream=False, jsononly=False): | |
"""Wrapper for authenticated HTTP GET from API endpoint. | |
endpoint = relative URL (e.g., "me/contacts") | |
stream = Requests stream argument; e.g., use True for image data | |
jsononly = if True, the JSON 'value' is returned instead of the response | |
object | |
""" | |
response = requests.get( \ | |
urllib.parse.urljoin('https://graph.microsoft.com/beta/',endpoint), | |
headers={'User-Agent' : 'graph-python-quickstart/1.0', | |
'Authorization' : 'Bearer {0}'.format(self.access_token), | |
'Accept' : 'application/json', | |
'Content-Type' : 'application/json', | |
'client-request-id' : str(uuid.uuid4()), # unique identifier | |
'return-client-request-id' : 'true'}, | |
stream=stream) | |
if jsononly: | |
return response.json().get('value', None) | |
else: | |
return response | |
def login(self, redirect_to): | |
"""Ask user to authenticate via web interface.""" | |
self.state = str(uuid.uuid4()) # used to verify source of auth request | |
print(' -> auth.Connect: asking user to authenticate') | |
redirect( \ | |
'https://login.microsoftonline.com/common/oauth2/v2.0/authorize/' + \ | |
'?response_type=code&client_id=' + self.app_id + \ | |
'&redirect_uri=http://localhost:5000/login/authorized' + \ | |
'&scope=Mail.Read%20User.Read' + \ | |
'&state=' + self.state, 302) | |
def logout(self): | |
"""Close current connection and redirect to specified route.""" | |
self.access_token = None | |
self.loggedin = False | |
redirect('/') | |
def save_token(self, response): | |
"""Save a new access token and related metadata. | |
response = response object from the token request | |
""" | |
jsondata = response.json() | |
if not 'access_token' in jsondata: | |
# no access token found in the response | |
print(' -> auth.Connect: request for access token failed') | |
self.logout() | |
return False | |
self.access_token = jsondata['access_token'] | |
self.loggedin = True | |
print(' -> auth.Connect: access token acquired ({0} bytes)'. \ | |
format(len(self.access_token))) | |
return True | |
def render_table(dataset, columns): | |
"""Convert list of dictionaries to HTML string for display.""" | |
coltuples = [column.split(':') for column in columns.split(' ')] | |
html_strings = ['<table class = "sample">'] | |
html_strings.append('<tr>') | |
for fldname, _ in coltuples: | |
html_strings.append('<th>' + fldname.capitalize() + '</th>') | |
html_strings.append('</tr>') | |
for entity in dataset: | |
html_strings.append('<tr>') | |
for fldname, nchars in coltuples: | |
fldsize = int(nchars) | |
if fldname in entity and entity[fldname]: | |
html_strings.append( \ | |
'<td>' + str(entity[fldname])[:fldsize].ljust(fldsize) + '</td>') | |
else: | |
html_strings.append('<td></td>') # missing value | |
html_strings.append('</tr>') | |
html_strings.append('</table>') | |
return ''.join(html_strings) | |
msgraph = Connect() | |
@route('/') | |
@view('homepage') | |
def home(): | |
"""Render the home page.""" | |
if msgraph.loggedin: | |
response = msgraph.get('me/mailFolders/Inbox/messages') | |
jsondata = response.json().get('value', None) | |
emails = [] | |
for email in jsondata: | |
email_date = email['receivedDateTime'][:10] | |
sender = '{0} ({1})'.format(email['from']['emailAddress']['name'], | |
email['from']['emailAddress']['address']) | |
subject = email['subject'] | |
emails.append(dict(date=email_date, sender=sender, subject=subject)) | |
return dict(sample='INBOX', | |
sampledata=render_table(emails, 'date:10 sender:35 subject:35')) | |
else: | |
return dict(sample=None, sampledata=None) | |
@route('/login') | |
def login(): | |
"""Prompt user to authenticate.""" | |
msgraph.login() | |
@route('/login/authorized') | |
def authorized(): | |
"""Fetch access token for authenticated user.""" | |
return msgraph.authorized() | |
@route('/logout') | |
def logout(): | |
"""Log out from MS Graph connection.""" | |
msgraph.logout() | |
if __name__ == '__main__': | |
@bottle.route('/static/<filepath:path>') | |
def server_static(filepath): | |
"""Handler for static files, used with the development server.""" | |
PROJECT_ROOT = os.path.abspath(os.path.dirname(__file__)) | |
STATIC_ROOT = os.path.join(PROJECT_ROOT, 'static').replace('\\', '/') | |
return bottle.static_file(filepath, root=STATIC_ROOT) | |
bottle.run(app=bottle.app(), server='wsgiref', host='localhost', port=5000) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment