Last active
April 26, 2024 08:59
-
-
Save chenlilyd/6cc298fd63b6a5f7bbae875760a9fad9 to your computer and use it in GitHub Desktop.
LTI tool implementation
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
# You can use this script to test LTI tool implementation on https://saltire.lti.app/platform | |
# To get a public domain, use: ngrok http 8000 | |
# | |
# Ref JWK: https://pyjwt.readthedocs.io/en/latest/ | |
import time | |
from flask import Flask, request, redirect | |
import jwt | |
from urllib.parse import urlencode | |
from uuid import uuid4 | |
from jwcrypto import jwk | |
from jwt import PyJWKClient | |
import time | |
# change this to the domain of your tool, ie., the domain this script is running | |
TOOL_DOMAIN = 'https://98d5-218-212-79-12.ngrok-free.app' | |
TOOL_CONFIG = { | |
# Note: platform actually supports configuring multiple Redirection URIs | |
# Even LTI named it `URI`, you should use complete URLs | |
'TOKEN_REDIRECT_URL': f'{TOOL_DOMAIN}/lti/callback', | |
} | |
TOOL_PRIVATE_KEY = {"alg":"RS256","d":"HXxLXil9ywZbFJbD9qsRepeVF2S7ai6lwImqnC3azBRfqNYfD1udLFxG8CTIGF8Okp9HLoYtdKNZpZxPewtDlxMymOvIG1usbpDSsDv_g7DiS0sRVOwC0WRsRqpOg2FtQ2rJEq4uQHN338tCfy_EjlwDPGhhYRCa3okMRXJV9KuxS6XleMMHgNYYSj7uuD-GO_KjQFCOysDq7n-PSSGOVTr7OEcW_FHXYIR8ae_ZOJVMGUkw-SFfX5Yv5lwC0bDg-Cf7_SApXEEA9XBkI5q6r5i52CB9Nf_YZ2sc8gnT54Gyp2CRXT7PuRyDxC3_IuQJpkWl5FFg6P8tQm9JdpJCoQ","dp":"wXUUFxXns43G8ZeIZm5IFfuRhUKYsJazJ5ER2tmruc-d5CfM6_OiBTQ1_BLmEf1naOwgc33RYZQhNDUkz55LrTJNeNjDqD9-xcGc1tZ3PG6jOyOUN_Pgn-8L0BXUAHjl_NewuAT8tGS4gLJexbQ2zC-vuMSgDhkj83FcxM2Smmk","dq":"tNd27942zR6DcML9hHL09PmoVZ6ce3-Iaki9YmI36xvDBCVXAbpTAfQ6sME94OgrxLn16VPmuoWHbbNod_rb3-ky-ZqBwg_hJ5XwUHeai1azlGipl2VX1C_s6brm7iyGlfoW4tGkqHmboCjPKH0FvXjcOd04mW6lnfUpsCm6MXU","e":"AQAB","kid":"lti-tool-kid-public","kty":"RSA","n":"w1lbrU-1K3s9q167txkonzz2wqk4mKOOhe_uwqA4KCS9_KFHQjeaWDwgNHnzTBBUGZ3pYE6R04IT5ntRc7NcgIm44t6XHS-3Y2jlbFlu3BbtyREzPbtB367IE6G5wUQ8KnGSaZGsh3IwUuU_hRRIagRjmF5Z5Bn7N0bGgjCI0znpGLJT8ULO4Wit_KZ1OqUZMMvR0f11UEejCXcQ1gPWOwoLNPnYq_k8FIbHlKmlg9dZWqEEyr-nFw9dgPX-MOzjXL9fOmubYBDHR9XmJg--zV9BnDn9W9Co2Ds9_Pu7ZrPzGQuIXUuJHOjoKtejR2HTveiW7-np_14nZUyugdoi_w","p":"8p2hZU57b8SPjO4hFgw9Wn-Kvz-lBBoofprd7v04ZuP8xASXLMSHXfCiK5p4ATFWZXYJqJPtdcieSm8C2tVjFH8RTt8Sq1XU25WchJi29DQA8wMuWZHR1ULyA8At4iQ5SPNv-YW67yA1e1_74qlDyomEkqG45xw8nEb5dL_LNfU","q":"ziAywOpmeMnjjifnK9uIZHA46lSM2G_J50Sb4YtXZ7YMILMRos9Nv3AV9pisVR6t9yN_LpssPX8ohYv_8Z_4teQtSKRS3ojhACEpFoBvto9XpQNmzfXBlc1NQVm3pUJVjJ3P6zyYB51aynVOklJ3kKZh1rTmclCMG8t_HrEWqKM","qi":"X16T3LH1nj-ezrA9AfUTKN3t9OH7ogfpCvwW7CzbKOsqnFZvUvpx84VvAHGvHfHSWYQ1hRjQyGylZ3Dxo87YjNn3Bnsa-vx2MJRTM18Uy_Ekw3JH562OqnGOQ3yRpKuZGO9iB_5SCgZnJ1eWDZfyIhGT5ar9Npik_qlYfZwZ6ws","use":"enc"} | |
TOOL_PUBLIC_KEY = """ | |
-----BEGIN PUBLIC KEY----- | |
MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAw1lbrU+1K3s9q167txko | |
nzz2wqk4mKOOhe/uwqA4KCS9/KFHQjeaWDwgNHnzTBBUGZ3pYE6R04IT5ntRc7Nc | |
gIm44t6XHS+3Y2jlbFlu3BbtyREzPbtB367IE6G5wUQ8KnGSaZGsh3IwUuU/hRRI | |
agRjmF5Z5Bn7N0bGgjCI0znpGLJT8ULO4Wit/KZ1OqUZMMvR0f11UEejCXcQ1gPW | |
OwoLNPnYq/k8FIbHlKmlg9dZWqEEyr+nFw9dgPX+MOzjXL9fOmubYBDHR9XmJg++ | |
zV9BnDn9W9Co2Ds9/Pu7ZrPzGQuIXUuJHOjoKtejR2HTveiW7+np/14nZUyugdoi | |
/wIDAQAB | |
-----END PUBLIC KEY----- | |
""" | |
# get from saltire | |
LTI_PLATFORM_CONFIG = dict( | |
kid = '6ox0b5ag4w', | |
message_hint="My LTI message hint!", # Message hint | |
mesage_url="https://saltire.lti.app/platform", # Message URL | |
deployment_id="cLWwj9cbmkSrCNsckEFBmA", # Platform Deployment ID | |
client_id ="saltire.lti.app", # Platform Client ID | |
auth_url = "https://saltire.lti.app/platform/auth", # Platform Authentication request URL: | |
token_url = "https://saltire.lti.app/platform/token/adb713a441ddbd2ea351a3c982a5b5b9", # Access Token service URL: | |
keyset_url = "https://saltire.lti.app/platform/jwks/adb713a441ddbd2ea351a3c982a5b5b9", # Public keyset URL: | |
issuer = "https://saltire.lti.app/platform" #Platform/Issuer ID | |
) | |
# don't use this in production | |
def insecure_ssl_context(): | |
import ssl | |
context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) | |
context.check_hostname = False | |
context.verify_mode = ssl.CERT_NONE | |
return context | |
def jwt_encode(payload: dict): | |
key = jwk.JWK(**TOOL_PRIVATE_KEY) | |
# Out[35]: {"kid":"lti-tool-kid-public","thumbprint":"ueHpt-S0lxBAzcj0pkK_Ods55y2P4W-p-lW4FlTh8Kc"} | |
return jwt.encode( | |
payload, | |
key.export_to_pem(private_key=True, password=None), | |
algorithm=TOOL_PRIVATE_KEY['alg'], | |
headers={'kid': TOOL_PRIVATE_KEY['kid']}) | |
def jwt_decode(token: str): | |
kid = LTI_PLATFORM_CONFIG['kid'] | |
headers = jwt.get_unverified_header(token) | |
if headers['kid'] != kid: # normally we would look up public keys settings by `kid` in tool configuration | |
raise Exception('unexpected kid') | |
url = LTI_PLATFORM_CONFIG['keyset_url'] | |
print(f'platform publick jwks url: {url}') | |
# insecure_ssl_context is used here, somehow in test environment, the ssl certificate is not valid | |
jwks_client = PyJWKClient(url, ssl_context=insecure_ssl_context()) | |
signing_key = jwks_client.get_signing_key(kid) | |
audience = LTI_PLATFORM_CONFIG['client_id'] | |
return jwt.decode( | |
token, | |
signing_key.key, | |
algorithms=["RS256"], | |
audience = audience | |
) | |
def generate_keypair(kid): | |
"""Creates an lti key pair""" | |
key = jwk.JWK.generate(kty='RSA', size=2048, alg='RS256', use='enc', kid=kid) | |
return (key.export_public(), key.export_private()) | |
app = Flask(__name__) | |
@app.route('/ping', methods=['GET']) | |
def ping(): | |
time.sleep(3) | |
return 'pong' | |
# this endpoint generates a redirect to the platform's authentication endpoint | |
@app.route('/lti/login', methods=['GET', 'POST']) | |
def lti_login(): | |
# https://www.imsglobal.org/spec/security/v1p0/#step-2-authentication-request | |
# Form data: | |
# {'iss': 'https://saltire.lti.app/platform', 'target_link_uri': 'https://saltire.lti.app/tool', 'login_hint': '29123', 'lti_message_hint': 'My LTI message hint!', 'client_id': 'saltire.lti.app', 'lti_deployment_id': 'cLWwj9cbmkSrCNsckEFBmA'} | |
# get as form data or query params | |
data = request.form or dict(request.args) | |
if not data: | |
raise Exception('failed to get request data') | |
nonce = str(uuid4()) | |
state = str(uuid4()) | |
client_id = data['client_id'] | |
issuer = data['iss'] | |
deployment_id = data['lti_deployment_id'] | |
params = dict( | |
response_type='id_token', | |
response_mode = 'form_post', | |
prompt = 'none', # hmm, is it string none, or it should just be None?? | |
scope='openid', | |
target_link_uri = data.get('target_link_uri', ''), | |
state = state, | |
nonce= nonce, | |
redirect_uri = TOOL_CONFIG['TOKEN_REDIRECT_URL'] , # https://openid.net/specs/openid-connect-core-1_0.html | |
client_id = client_id | |
) | |
# https://www.imsglobal.org/spec/lti/v1p3#lti_message_hint-login-parameter | |
# Similarly to the login_hint parameter, lti_message_hint value is opaque to | |
# the tool. If present in the login initiation request, the tool MUST include | |
# it back in the authentication request unaltered. | |
login_hint = data.get('login_hint', '') | |
lti_message_hint = data.get('lti_message_hint', '') | |
if login_hint: | |
params['login_hint'] = login_hint | |
if lti_message_hint: | |
params['lti_message_hint'] = lti_message_hint | |
query_string = urlencode(params) | |
# The platform will redirect to this link which is on the tool server | |
platform_auth_url = LTI_PLATFORM_CONFIG.get('auth_url', '') + '?' + query_string | |
# the client_id is unique in the view of the platform, in this sample we only support one platform (one issuer + client_id) | |
if client_id != LTI_PLATFORM_CONFIG['client_id']: | |
raise Exception('unexpected client_id') | |
if issuer != LTI_PLATFORM_CONFIG['issuer']: | |
raise Exception('unexpected issuer') | |
print(f'redirect to {platform_auth_url}') | |
# todo add state to cookie | |
# https://openid.net/specs/openid-connect-core-1_0.html#AuthRequest | |
# Typically, Cross-Site Request Forgery (CSRF, XSRF) mitigation is done by cryptographically binding the value of this parameter with a browser cookie. | |
resp = redirect(platform_auth_url) | |
resp.set_cookie('state', state) | |
return resp | |
# this endpoint returns/redirects the users to access a resource on the tool | |
@app.route('/lti/callback', methods=['POST', 'GET']) | |
def lti_launch(): | |
# Form Data: {'id_token': 'jwt_encoded_id_token', 'state': '317b5056-503a-40c5-92c8-bb66087bf394'} | |
data = request.form | |
if not data: | |
raise Exception('no data in lti launch') | |
if 'error' in data: | |
print(f'callback failed: {data}') | |
raise Exception(data['error'] + ': ' + data.get('error_description', '')) | |
# todo validate this `state` value in cookie | |
# https://openid.net/specs/openid-connect-core-1_0.html#AuthRequest | |
state = data['state'] | |
state_in_cookie = request.cookies.get('state') | |
if state != state_in_cookie: | |
raise Exception('state mismatch, possible CSRF attack') | |
id_token = data['id_token'] | |
decoded = jwt_decode(id_token) | |
message_type = decoded.get('https://purl.imsglobal.org/spec/lti/claim/message_type', '') | |
deep_link_return_url = None | |
if message_type == 'LtiResourceLinkRequest': | |
print('LtiResourceLinkRequest') | |
# Following LTI launch, the tool should redirect the user to the resource link URL | |
return dict(decoded=decoded, success=True, message_type=message_type, deep_link_return_url=deep_link_return_url) | |
elif message_type == 'LtiDeepLinkingRequest': | |
deep_link_return_url = decoded.get('https://purl.imsglobal.org/spec/lti-dl/claim/deep_linking_settings', {}).get('deep_link_return_url', '') | |
# Following LTI deep linking request, the tool should return a form containing content items | |
# After the user selects a content item, the tool should redirect the user with a POST request to the deep link return URL | |
# Here we just redirect the user to deep link return URL with some sample content items | |
print('LtiDeepLinkingRequest') | |
response_data = {'iss': decoded['aud'][0], | |
'aud': decoded['iss'], | |
'exp': int(time.time()) + 600, | |
'iat': int(time.time()), | |
'nonce': decoded['nonce'], | |
'https://purl.imsglobal.org/spec/lti/claim/deployment_id': decoded['https://purl.imsglobal.org/spec/lti/claim/deployment_id'], | |
'https://purl.imsglobal.org/spec/lti/claim/message_type': 'LtiDeepLinkingResponse', | |
"https://purl.imsglobal.org/spec/lti/claim/version": "1.3.0", | |
"https://purl.imsglobal.org/spec/lti-dl/claim/content_items": [ | |
{ | |
"type": "link", | |
"url": "https://www.youtube.com/watch?v=corV3-WsIro", | |
"embed": { | |
"html": | |
"<iframe width=\"560\" height=\"315\" src=\"https://www.youtube.com/embed/corV3-WsIro\" frameborder=\"0\" allow=\"autoplay; encrypted-media\" allowfullscreen></iframe>" | |
}, | |
"window": { | |
"targetName": "youtube-corV3-WsIro", | |
"windowFeatures": "height=560,width=315,menubar=no" | |
}, | |
"iframe": { | |
"width": 560, | |
"height": 315, | |
"src": "https://www.youtube.com/embed/corV3-WsIro" | |
} | |
}, | |
{ | |
"type": "ltiResourceLink", | |
"title": "A title", | |
"text": "This is a link to an activity that will be graded", | |
"url": "https://lti.example.com/launchMe", | |
"icon": { | |
"url": "https://lti.example.com/image.jpg", | |
"width": 100, | |
"height": 100 | |
}, | |
"thumbnail": { | |
"url": "https://lti.example.com/thumb.jpg", | |
"width": 90, | |
"height": 90 | |
}, | |
"lineItem": { | |
"scoreMaximum": 87, | |
"label": "Chapter 12 quiz", | |
"resourceId": "xyzpdq1234", | |
"tag": "originality", | |
"gradesReleased": True | |
}, | |
"available": { | |
"startDateTime": "2018-02-06T20:05:02Z", | |
"endDateTime": "2018-03-07T20:05:02Z" | |
}, | |
"submission": { | |
"endDateTime": "2018-03-06T20:05:02Z" | |
}, | |
"custom": { | |
"quiz_id": "az-123", | |
"duedate": "$ResourceLink.submission.endDateTime" | |
}, | |
"window": { | |
"targetName": "examplePublisherContent" | |
}, | |
"iframe": { | |
"height": 890 | |
} | |
},],} | |
return dict(decoded=decoded, success=True, message_type=message_type, deep_link_return_url=deep_link_return_url, response_data=response_data) | |
return dict(decoded=decoded, success=False) | |
@app.errorhandler(404) | |
def catch_all(e): | |
request_data = request.data or request.get_json(silent=True) | |
print("Request Body:", request_data) | |
return {'success': True} | |
@app.before_request | |
def log_request_info(): | |
print('=' * 30 + ">") | |
print('Headers: ', request.headers) | |
print('Body: ', request.get_data(as_text=True)) | |
print('Query params:', request.args) | |
# Log query parameters | |
if request.args: | |
print('Query Params: ', dict(request.args)) | |
if request.method == 'POST': | |
if request.form: | |
print('Form Data: ', dict(request.form)) | |
elif request.is_json: | |
print('Json Data: ', request.json) | |
@app.route('/tool/public-keys', methods=['GET']) | |
def public_keys(): | |
# Note | |
# 1. this is a list of public keys, not a single key | |
# 2. `kid` must be present | |
return { | |
"keys": [ | |
{ | |
"alg": "RS256", | |
"e": "AQAB", | |
"kid": "lti-tool-kid-public", | |
"kty": "RSA", | |
"n": "w1lbrU-1K3s9q167txkonzz2wqk4mKOOhe_uwqA4KCS9_KFHQjeaWDwgNHnzTBBUGZ3pYE6R04IT5ntRc7NcgIm44t6XHS-3Y2jlbFlu3BbtyREzPbtB367IE6G5wUQ8KnGSaZGsh3IwUuU_hRRIagRjmF5Z5Bn7N0bGgjCI0znpGLJT8ULO4Wit_KZ1OqUZMMvR0f11UEejCXcQ1gPWOwoLNPnYq_k8FIbHlKmlg9dZWqEEyr-nFw9dgPX-MOzjXL9fOmubYBDHR9XmJg--zV9BnDn9W9Co2Ds9_Pu7ZrPzGQuIXUuJHOjoKtejR2HTveiW7-np_14nZUyugdoi_w", | |
"use": "enc" | |
} | |
] | |
} | |
app.run(port=8000, debug=True) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment