Canvas LTI<>Metabase glue layer

lti_config.json

{  
   "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"
}

views.py

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)