This commit is contained in:
Markos Gogoulos
2025-12-30 18:12:49 +02:00
parent 2e57164831
commit 9370706097
4 changed files with 39 additions and 5 deletions

View File

@@ -6,6 +6,7 @@ Provides Django-specific implementations for PyLTI1p3 interfaces
import json
import time
import uuid
from typing import Any, Dict, Optional
import jwt
@@ -192,6 +193,8 @@ class DjangoServiceConnector(ServiceConnector):
self._access_token_expires = 0
def get_access_token(self, scopes):
print(f"Requesting access token for scopes: {scopes}")
if self._access_token and time.time() < self._access_token_expires:
return self._access_token
@@ -200,14 +203,19 @@ class DjangoServiceConnector(ServiceConnector):
pem_bytes = jwk_obj.export_to_pem(private_key=True, password=None)
private_key = serialization.load_pem_private_key(pem_bytes, password=None, backend=default_backend())
# Use configured audience if available, otherwise auth_token_url
# Moodle expects auth_token_url as audience usually, but some platforms differ
aud = self._registration.get_auth_audience() if self._registration.get_auth_audience() else self._registration.get_auth_token_url()
print(f"Using audience for token request: {aud}")
now = int(time.time())
payload = {
'iss': self._registration.get_client_id(),
'sub': self._registration.get_client_id(),
'aud': self._registration.get_auth_token_url(),
'aud': aud,
'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,7 +228,10 @@ class DjangoServiceConnector(ServiceConnector):
'scope': ' '.join(scopes),
}
print(f"Posting to token URL: {token_url}")
response = requests.post(token_url, data=data, timeout=10)
if not response.ok:
print(f"Token request failed: {response.status_code} {response.text}")
response.raise_for_status()
token_data = response.json()
@@ -240,11 +251,14 @@ class DjangoServiceConnector(ServiceConnector):
if 'accept' in kwargs:
headers['Accept'] = kwargs['accept']
print(f"Making service request to: {url}")
if is_post:
response = requests.post(url, json=data, headers=headers, timeout=10)
else:
response = requests.get(url, headers=headers, timeout=10)
print(f"Service response status: {response.status_code}")
response.raise_for_status()
try: