CopyBot/mediacms
1
0
Fork 0
mirror of https://github.com/mediacms-io/mediacms.git synced 2026-01-20 15:22:58 -05:00
Code Issues Packages Projects Releases Wiki Activity

wtv

Browse Source
This commit is contained in:
Markos Gogoulos
2025-12-30 18:38:07 +02:00
parent d3c858173f
commit 5070050fa4
2 changed files with 46 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
lti/adapters.py
View File

@@ -4,8 +4,10 @@ PyLTI1p3 Django adapters for MediaCMS
Provides Django-specific implementations for PyLTI1p3 interfaces
"""
import hashlib
import json
import time
import uuid
from typing import Any, Dict, Optional
import jwt
@@ -192,8 +194,10 @@ class DjangoServiceConnector(ServiceConnector):
self._access_token_expires = 0
def get_access_token(self, scopes):
if self._access_token and time.time() < self._access_token_expires:
return self._access_token
cache_key = 'lti_access_token_' + self._registration.get_issuer() + '_' + hashlib.sha1(' '.join(scopes).encode('utf-8')).hexdigest()
token_data = cache.get(cache_key)
if token_data:
return token_data['access_token']
key_obj = LTIToolKeys.get_or_create_keys()
jwk_obj = jwk.JWK(**key_obj.private_key_jwk)
@@ -201,13 +205,17 @@ class DjangoServiceConnector(ServiceConnector):
private_key = serialization.load_pem_private_key(pem_bytes, password=None, backend=default_backend())
now = int(time.time())
# Moodle can be picky about audience. Including both token URL and issuer is safer.
audience = [self._registration.get_auth_token_url(), self._registration.get_issuer()]
payload = {
'iss': self._registration.get_client_id(),
'sub': self._registration.get_client_id(),
'aud': self._registration.get_auth_token_url(),
'aud': audience,
'iat': now,
'exp': now + 300,
'jti': str(time.time()),
'jti': str(uuid.uuid4()),
}
client_assertion = jwt.encode(payload, private_key, algorithm='RS256', headers={'kid': key_obj.private_key_jwk['kid']})
@@ -220,15 +228,24 @@ class DjangoServiceConnector(ServiceConnector):
'scope': ' '.join(scopes),
}
print(f"LTI Service: Requesting access token from {token_url} with scopes: {scopes}")
response = requests.post(token_url, data=data, timeout=10)
response.raise_for_status()
token_data = response.json()
self._access_token = token_data['access_token']
expires_in = token_data.get('expires_in', 3600)
self._access_token_expires = time.time() + expires_in - 10
try:
response.raise_for_status()
token_data = response.json()
print(f"LTI Service: Successfully received access token. Expires in: {token_data.get('expires_in', 'N/A')}")
return self._access_token
expires_in = token_data.get('expires_in', 3600)
cache.set(cache_key, token_data, timeout=expires_in - 10)
return token_data['access_token']
except requests.exceptions.HTTPError as e:
print(f"LTI Service Error: Failed to get access token. Status: {e.response.status_code}, Response: {e.response.text}")
raise
except json.JSONDecodeError:
print(f"LTI Service Error: Failed to decode JSON from token endpoint. Response: {response.text}")
raise
def make_service_request(self, scopes, url, is_post=False, data=None, **kwargs):
access_token = self.get_access_token(scopes)