Skip to content

Instantly share code, notes, and snippets.

@brettcvz
Created March 5, 2024 18:29
Show Gist options
  • Save brettcvz/9af8fb681b8bc6aa436de98765ae2dde to your computer and use it in GitHub Desktop.
Save brettcvz/9af8fb681b8bc6aa436de98765ae2dde to your computer and use it in GitHub Desktop.
Canvas LTI<>Metabase glue layer
{
"title":"SPS Canvas -> Metabase Link",
"scopes":[],
"extensions":[
{
"domain":"spstools.summitps.org",
"tool_id":"sps-canvas-metabase",
"platform":"canvas.instructure.com",
"settings":{
"text":"SPS Analytics",
"icon_url":"https://metabase.summitps.org/app/assets/img/apple-touch-icon.png",
"placements":[
{
"text":"SPS Course Analytics",
"enabled":true,
"icon_url":"https://metabase.summitps.org/app/assets/img/apple-touch-icon.png",
"placement":"course_navigation",
"message_type":"LtiResourceLinkRequest",
"target_link_uri":"https://metabase.summitps.org/dashboard/196-canvas-test",
"visibility": "admins",
"selection_height":1400,
"custom_fields": {
"user_sis_id": "$Canvas.user.sisIntegrationId",
"user_email": "$Person.email.primary",
"user_id": "$Canvas.user.sisSourceId",
"course_sis_id": "$Canvas.course.sisSourceId",
"course_section_sis_id": "$Canvas.course.sectionSisSourceIds"
}
}
]
}
}
],
"public_jwk":{
"kty":"RSA",
"alg":"RS256",
"e":"AQAB",
"kid":"8f796169-0ac4-48a3-a202-fa4f3d814fcd",
"n":"nZD7QWmIwj-3N_RZ1qJjX6CdibU87y2l02yMay4KunambalP9g0fU9yZLwLX9WYJINcXZDUf6QeZ-SSbblET-h8Q4OvfSQ7iuu0WqcvBGy8M0qoZ7I-NiChw8dyybMJHgpiP_AyxpCQnp3bQ6829kb3fopbb4cAkOilwVRBYPhRLboXma0cwcllJHPLvMp1oGa7Ad8osmmJhXhM9qdFFASg_OCQdPnYVzp8gOFeOGwlXfSFEgt5vgeU25E-ycUOREcnP7BnMUk7wpwYqlE537LWGOV5z_1Dqcqc9LmN-z4HmNV7b23QZW4_mzKIOY4IqjmnUGgLU9ycFj5YGDCts7Q",
"use":"sig"
},
"description":"Link from Canvas into SPS Metabase",
"target_link_uri":"https://metabase.summitps.org/",
"oidc_initiation_url":"https://spstools.summitps.org/lti/metabase/initiate"
}
from django.shortcuts import render, redirect
from django.http import HttpResponse
from django.core.exceptions import SuspiciousOperation
from django.views.decorators.csrf import csrf_exempt
from django.conf import settings as django_settings
import secrets
from urllib.parse import urlencode
from jwcrypto.jwk import JWK
import jwt
import base64
import json
import requests
import time
import re
ISSUER_TO_AUTH_URL_MAP = {
"https://canvas.instructure.com": "https://sso.canvaslms.com/api/lti/authorize_redirect",
"https://canvas.beta.instructure.com": "https://sso.beta.canvaslms.com/api/lti/authorize_redirect",
"https://canvas.test.instructure.com": "https://sso.test.canvaslms.com/api/lti/authorize_redirect",
}
ISSUER_TO_PUBLIC_KEY_MAP = {
"https://canvas.instructure.com": "https://sso.canvaslms.com/api/lti/security/jwks",
"https://canvas.beta.instructure.com": "https://sso.beta.canvaslms.com/api/lti/security/jwks",
"https://canvas.test.instructure.com": "https://sso.test.canvaslms.com/api/lti/security/jwks",
}
@csrf_exempt
def initiate(request):
if request.method != "POST":
raise SuspiciousOperation("Invalid lti request - method")
if request.POST.get("iss") not in ISSUER_TO_AUTH_URL_MAP:
raise SuspiciousOperation("Invalid lti request - iss")
client_id = request.POST["client_id"]
request.session["lti_client_id"] = client_id
request.session["lti_state"] = secrets.token_urlsafe()
request.session["lti_nonce"] = secrets.token_urlsafe()
request.session.modified = True
auth_url_base = ISSUER_TO_AUTH_URL_MAP[request.POST["iss"]]
query_string = urlencode({
"scope": "openid",
"response_type": "id_token",
"client_id": client_id,
"login_hint": request.POST["login_hint"],
"lti_message_hint": request.POST["lti_message_hint"],
"response_mode": "form_post",
"prompt": "none",
"state": request.session["lti_state"],
"nonce": request.session["lti_nonce"],
"redirect_uri": django_settings.LTI_REDIRECT_URI,
})
auth_url = f"{auth_url_base}?{query_string}"
return redirect(auth_url)
def urlsafe_b64decode(val):
remainder = len(val) % 4
if remainder > 0:
padlen = 4 - remainder
val = val + ("=" * padlen)
tmp = val.translate(str.maketrans("-_", "+/"))
return base64.b64decode(tmp).decode("utf-8")
def verify_token(id_token):
jwt_parts = id_token.split(".")
header = json.loads(urlsafe_b64decode(jwt_parts[0]))
body = json.loads(urlsafe_b64decode(jwt_parts[1]))
# Fetch the public keys for Canvas
issuer = body.get("iss")
if issuer not in ISSUER_TO_PUBLIC_KEY_MAP:
return body, False
keyset_req = requests.get(ISSUER_TO_PUBLIC_KEY_MAP[issuer])
keyset = keyset_req.json()
# Find which key we want to use based on what was in the token header
for key in keyset["keys"]:
if header["kid"] == key["kid"] and header["alg"] == key["alg"]:
jwk_obj = JWK(**key)
public_key = jwk_obj.export_to_pem()
try:
jwt.decode(id_token, public_key, algorithms=[header["alg"]], options={"verify_aud": False})
return body, True
except jwt.InvalidTokenException as e:
print("Invalid token", e)
return
print("No matching key found", header, keys)
return None, False
def generate_metabase_embed_link(target_url, user_info):
dashboard_id_match = re.match(r"https://metabase.summitps.org/dashboard/(\d+).*", target_url)
if not dashboard_id_match:
raise SuspiciousOperation("Invalid lti request - target_url")
dashboard_id = int(dashboard_id_match.group(1))
print("Loading metabase dashboard", dashboard_id, "with params", user_info)
payload = {
"resource": {"dashboard": dashboard_id},
#"params": user_info,
"params": {},
"exp": round(time.time()) + (60 * 10) # 10 minute expiration
}
token = jwt.encode(payload, django_settings.METABASE_SECRET_KEY, algorithm="HS256")
url = f"{django_settings.METABASE_SITE_URL}/embed/dashboard/{token}#bordered=false&titled=false"
return url
@csrf_exempt
def authenticate(request):
if request.method != "POST":
raise SuspiciousOperation("Invalid lti request - method")
if "lti_state" not in request.session or request.POST.get("state") != request.session["lti_state"]:
raise SuspiciousOperation("Invalid lti request - state")
del request.session["lti_state"]
if "id_token" not in request.POST:
raise SuspiciousOperation("Invalid lti request - token")
id_token = request.POST["id_token"]
parsed_token, valid = verify_token(id_token)
if not valid:
raise SuspiciousOperation("Invalid lti request - token")
if "lti_nonce" not in request.session or parsed_token["nonce"] != request.session["lti_nonce"]:
raise SuspiciousOperation("Invalid lti request - nonce")
if "lti_client_id" not in request.session or parsed_token["aud"] != request.session["lti_client_id"]:
raise SuspiciousOperation("Invalid lti request - aud")
del request.session["lti_nonce"]
#print(json.dumps(parsed_token, indent=2))
target_url = parsed_token["https://purl.imsglobal.org/spec/lti/claim/target_link_uri"]
user_info = parsed_token["https://purl.imsglobal.org/spec/lti/claim/custom"]
metabase_url = generate_metabase_embed_link(target_url, user_info)
return redirect(metabase_url)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment