This commit is contained in:
Markos Gogoulos
2025-12-29 14:13:45 +02:00
parent 9667e6b0ad
commit 0585513439
6 changed files with 43 additions and 230 deletions

View File

@@ -26,7 +26,6 @@ class DjangoRequest(Request):
"""Get parameter from GET or POST"""
# Check both POST and GET, POST takes priority
value = self._request.POST.get(key) or self._request.GET.get(key)
print(f"DjangoRequest.get_param('{key}') = {value}", flush=True)
return value
def get_cookie(self, key):
@@ -105,12 +104,8 @@ class DjangoSessionService:
def save_launch_data(self, key, data):
"""Save launch data to session"""
session_key = self._session_key_prefix + key
print(f"Saving launch data: key={key}, session_key={session_key}, data={data}", flush=True)
print(f"Session ID before save: {self.request.session.session_key}", flush=True)
self.request.session[session_key] = json.dumps(data)
self.request.session.modified = True
print(f"Session ID after save: {self.request.session.session_key}", flush=True)
print("Data saved successfully", flush=True)
return True
def check_launch_data_storage_exists(self, key):
@@ -121,40 +116,31 @@ class DjangoSessionService:
def check_state_is_valid(self, state, nonce):
"""Check if state is valid - state is for CSRF protection, nonce is validated separately by JWT"""
state_key = f'state-{state}'
print(f"Checking state validity: state={state}", flush=True)
print(f"Looking for state_key: {state_key}", flush=True)
state_data = self.get_launch_data(state_key)
print(f"State data found: {state_data}", flush=True)
if not state_data:
print("ERROR: State data not found in session!", flush=True)
return False
# State exists - that's sufficient for CSRF protection
# Nonce validation is handled by PyLTI1p3 through JWT signature and claims validation
print("State is valid!", flush=True)
return True
def check_nonce(self, nonce):
"""Check if nonce is valid (not used before) and mark it as used"""
nonce_key = f'nonce-{nonce}'
print(f"Checking nonce: {nonce}", flush=True)
# Check if nonce was already used
if self.check_launch_data_storage_exists(nonce_key):
print(f"ERROR: Nonce {nonce} was already used!", flush=True)
return False
# Mark nonce as used
self.save_launch_data(nonce_key, {'used': True})
print(f"Nonce {nonce} is valid and marked as used", flush=True)
return True
def set_state_valid(self, state, id_token_hash):
"""Mark state as valid and associate it with the id_token_hash"""
state_key = f'state-{state}'
print(f"Setting state valid: state={state}, id_token_hash={id_token_hash}", flush=True)
self.save_launch_data(state_key, {'valid': True, 'id_token_hash': id_token_hash})
return True
@@ -225,12 +211,9 @@ class DjangoToolConfig(ToolConfAbstract):
def find_registration_by_issuer(self, iss, *args, **kwargs):
"""Find registration by issuer"""
print(f"DjangoToolConfig.find_registration_by_issuer('{iss}')", flush=True)
if iss not in self._config:
print(" -> Not found in config", flush=True)
return None
config = self._config[iss]
print(f" -> Found: {config.get('client_id')}", flush=True)
# Create Registration object from config dict
registration = Registration()
@@ -248,18 +231,13 @@ class DjangoToolConfig(ToolConfAbstract):
def find_registration_by_params(self, iss, client_id, *args, **kwargs):
"""Find registration by issuer and client ID"""
print(f"DjangoToolConfig.find_registration_by_params('{iss}', '{client_id}')", flush=True)
if iss not in self._config:
print(" -> Issuer not found", flush=True)
return None
config = self._config[iss]
if config.get('client_id') != client_id:
print(f" -> Client ID mismatch: expected {client_id}, got {config.get('client_id')}", flush=True)
return None
print(" -> Match found", flush=True)
# Create Registration object from config dict
registration = Registration()
registration.set_issuer(iss)

View File

@@ -12,6 +12,7 @@ from .models import (
LTIRoleMapping,
LTIUserMapping,
)
from .services import LTINRPSClient
@admin.register(LTIPlatform)
@@ -85,72 +86,42 @@ class LTIResourceLinkAdmin(admin.ModelAdmin):
def sync_course_members(self, request, queryset):
"""Sync course members from LMS using NRPS"""
import traceback
from .services import LTINRPSClient
print("=" * 80, flush=True)
print("ADMIN ACTION: Sync course members started", flush=True)
print(f"User: {request.user.username}", flush=True)
print(f"Number of resource links selected: {queryset.count()}", flush=True)
synced_count = 0
failed_count = 0
for resource_link in queryset:
print(f"\n--- Processing: {resource_link.context_title} (ID: {resource_link.id}) ---", flush=True)
try:
# Check if NRPS is enabled
print(f"Platform: {resource_link.platform.name}", flush=True)
print(f"NRPS enabled: {resource_link.platform.enable_nrps}", flush=True)
if not resource_link.platform.enable_nrps:
print("ERROR: NRPS is disabled", flush=True)
messages.warning(request, f'NRPS is disabled for platform: {resource_link.platform.name}')
failed_count += 1
continue
# Check if RBAC group exists
print(f"RBAC group: {resource_link.rbac_group}", flush=True)
if not resource_link.rbac_group:
print("ERROR: No RBAC group", flush=True)
messages.warning(request, f'No RBAC group for: {resource_link.context_title}')
failed_count += 1
continue
# Get last successful launch for NRPS endpoint
print("Looking for last successful launch...", flush=True)
last_launch = LTILaunchLog.objects.filter(platform=resource_link.platform, resource_link=resource_link, success=True).order_by('-created_at').first()
if not last_launch:
print("ERROR: No launch data found", flush=True)
messages.warning(request, f'No launch data for: {resource_link.context_title}')
failed_count += 1
continue
print(f"Found launch from: {last_launch.created_at}", flush=True)
print("Creating NRPS client...", flush=True)
# Perform NRPS sync
nrps_client = LTINRPSClient(resource_link.platform, last_launch.claims)
print("Calling sync_members_to_rbac_group...", flush=True)
result = nrps_client.sync_members_to_rbac_group(resource_link.rbac_group)
print(f"Sync result: {result}", flush=True)
synced_count += result.get('synced', 0)
messages.success(request, f'Synced {result.get("synced", 0)} members for: {resource_link.context_title}')
except Exception as e:
print(f"ERROR during sync: {str(e)}", flush=True)
print(f"Traceback:\n{traceback.format_exc()}", flush=True)
messages.error(request, f'Error syncing {resource_link.context_title}: {str(e)}')
failed_count += 1
# Summary message
print("\n=== Sync Complete ===", flush=True)
print(f"Total synced: {synced_count}", flush=True)
print(f"Failed: {failed_count}", flush=True)
print("=" * 80, flush=True)
if synced_count > 0:
self.message_user(request, f'Successfully synced members from {queryset.count() - failed_count} course(s). Total members: {synced_count}', messages.SUCCESS)
if failed_count > 0:

View File

@@ -4,8 +4,6 @@ LTI Deep Linking 2.0 for MediaCMS
Allows instructors to select media from MediaCMS library and embed in Moodle courses
"""
import logging
from django.conf import settings
from django.contrib.auth.decorators import login_required
from django.http import JsonResponse
@@ -17,10 +15,9 @@ from django.views.decorators.csrf import csrf_exempt
from files.models import Media
from .adapters import DjangoToolConfig
from .models import LTIPlatform
logger = logging.getLogger(__name__)
@method_decorator(login_required, name='dispatch')
class SelectMediaView(View):
@@ -112,7 +109,6 @@ class SelectMediaView(View):
content_items.append(content_item)
except Media.DoesNotExist:
logger.warning(f"Media {media_id} not found during deep linking")
continue
if not content_items:
@@ -141,8 +137,6 @@ class SelectMediaView(View):
# For now, return a placeholder
try:
from .adapters import DjangoToolConfig
platform_id = deep_link_data['platform_id']
platform = LTIPlatform.objects.get(id=platform_id)
@@ -156,10 +150,7 @@ class SelectMediaView(View):
# 2. Call launch.get_deep_link()
# 3. Call deep_link.output_response_form(content_items)
logger.warning("Deep linking JWT creation not fully implemented")
return "JWT_TOKEN_PLACEHOLDER"
except Exception as e:
logger.error(f"Error creating deep link JWT: {str(e)}", exc_info=True)
except Exception:
return "ERROR_CREATING_JWT"

View File

@@ -9,7 +9,6 @@ Provides functions to:
"""
import hashlib
import logging
from allauth.account.models import EmailAddress
from django.conf import settings
@@ -22,9 +21,6 @@ from users.models import User
from .models import LTIResourceLink, LTIRoleMapping, LTIUserMapping
logger = logging.getLogger(__name__)
# Default LTI role mappings
DEFAULT_LTI_ROLE_MAPPINGS = {
'Instructor': {'global_role': 'advancedUser', 'group_role': 'manager'},
@@ -49,10 +45,6 @@ def provision_lti_user(platform, claims):
Pattern: Similar to saml_auth.adapter.perform_user_actions()
"""
print("\n" + "=" * 80, flush=True)
print("LTI USER PROVISIONING - User data from Moodle:", flush=True)
print("=" * 80, flush=True)
lti_user_id = claims.get('sub')
if not lti_user_id:
raise ValueError("Missing 'sub' claim in LTI launch")
@@ -62,33 +54,6 @@ def provision_lti_user(platform, claims):
family_name = claims.get('family_name', '')
name = claims.get('name', f"{given_name} {family_name}").strip()
print(f"LTI User ID (sub): {lti_user_id}", flush=True)
print(f"Email: {email if email else 'NOT PROVIDED'}", flush=True)
print(f"Given Name: {given_name if given_name else 'NOT PROVIDED'}", flush=True)
print(f"Family Name: {family_name if family_name else 'NOT PROVIDED'}", flush=True)
print(f"Full Name: {name if name else 'NOT PROVIDED'}", flush=True)
# Check what username would be generated
test_username = generate_username_from_lti(lti_user_id, email, given_name, family_name)
print(f"\nGenerated username: {test_username}", flush=True)
# Explain why this username was chosen
if email and '@' in email:
email_part = email.split('@')[0]
if len(email_part) >= 4:
print(f" -> Using email-based username (from '{email}')", flush=True)
else:
print(f" -> Email part '{email_part}' too short (< 4 chars), trying name...", flush=True)
if given_name and family_name and len(f"{given_name}.{family_name}") >= 4:
print(f" -> Using name-based username ('{given_name}.{family_name}')", flush=True)
else:
print(" -> Name too short or missing, using fallback (hashed user ID)", flush=True)
elif given_name and family_name and len(f"{given_name}.{family_name}") >= 4:
print(f" -> No email, using name-based username ('{given_name}.{family_name}')", flush=True)
else:
print(" -> No usable email or name, using fallback (hashed user ID)", flush=True)
print("=" * 80 + "\n", flush=True)
# Check for existing mapping
mapping = LTIUserMapping.objects.filter(platform=platform, lti_user_id=lti_user_id).select_related('user').first()
@@ -123,8 +88,6 @@ def provision_lti_user(platform, claims):
mapping.email = email
mapping.save(update_fields=['email'])
logger.info(f"Updated LTI user: {user.username} (platform: {platform.name})")
else:
# Create new user
username = generate_username_from_lti(lti_user_id, email, given_name, family_name)
@@ -140,14 +103,12 @@ def provision_lti_user(platform, claims):
if email:
try:
EmailAddress.objects.create(user=user, email=email, verified=True, primary=True)
except Exception as e:
logger.warning(f"Could not create EmailAddress for LTI user: {e}")
except Exception:
pass
# Create mapping
LTIUserMapping.objects.create(platform=platform, lti_user_id=lti_user_id, user=user, email=email, given_name=given_name, family_name=family_name, name=name)
logger.info(f"Created new LTI user: {user.username} (platform: {platform.name})")
return user
@@ -214,7 +175,7 @@ def provision_lti_context(platform, claims, resource_link_id):
)
if created:
logger.info(f"Created category for LTI context: {category.title} (uid: {uid})")
pass
else:
# Update title if changed
if context_title and category.title != context_title:
@@ -230,13 +191,9 @@ def provision_lti_context(platform, claims, resource_link_id):
},
)
if created:
logger.info(f"Created RBAC group for LTI context: {rbac_group.name}")
# Link category to RBAC group
if category not in rbac_group.categories.all():
rbac_group.categories.add(category)
logger.info(f"Linked category {category.title} to RBAC group {rbac_group.name}")
# Get or create resource link
resource_link, created = LTIResourceLink.objects.get_or_create(
@@ -316,7 +273,6 @@ def apply_lti_roles(user, platform, lti_roles, rbac_group):
# Apply global role if auto_sync_roles is enabled
if platform.auto_sync_roles:
user.set_role_from_mapping(global_role)
logger.info(f"Applied global role '{global_role}' to user {user.username}")
# Determine group role
group_role = 'member'
@@ -329,11 +285,6 @@ def apply_lti_roles(user, platform, lti_roles, rbac_group):
# Create or update RBAC membership
membership, created = RBACMembership.objects.update_or_create(user=user, rbac_group=rbac_group, defaults={'role': group_role})
if created:
logger.info(f"Added user {user.username} to RBAC group {rbac_group.name} as {group_role}")
else:
logger.info(f"Updated user {user.username} in RBAC group {rbac_group.name} to {group_role}")
return global_role, group_role
@@ -394,8 +345,6 @@ def create_lti_session(request, user, launch_data, platform):
timeout = getattr(settings, 'LTI_SESSION_TIMEOUT', 3600)
request.session.set_expiry(timeout)
logger.info(f"Created LTI session for user {user.username} (expires in {timeout}s)")
return True

View File

@@ -4,18 +4,19 @@ LTI Names and Role Provisioning Service (NRPS) Client
Fetches course membership from Moodle via NRPS and syncs to MediaCMS RBAC groups
"""
import logging
import hashlib
from allauth.account.models import EmailAddress
from django.utils import timezone
from pylti1p3.names_roles import NamesRolesProvisioningService
from rbac.models import RBACMembership
from users.models import User
from .adapters import DjangoToolConfig
from .handlers import apply_lti_roles, generate_username_from_lti
from .models import LTIUserMapping
logger = logging.getLogger(__name__)
class LTINRPSClient:
"""Client for Names and Role Provisioning Service"""
@@ -37,16 +38,13 @@ class LTINRPSClient:
def can_sync(self):
"""Check if NRPS sync is available"""
if not self.platform.enable_nrps:
logger.warning(f"NRPS disabled for platform {self.platform.name}")
return False
if not self.nrps_claim:
logger.warning("NRPS claim missing in launch data")
return False
service_url = self.nrps_claim.get('context_memberships_url')
if not service_url:
logger.warning("NRPS context_memberships_url missing")
return False
return True
@@ -62,28 +60,19 @@ class LTINRPSClient:
return []
try:
print(f"NRPS claim data: {self.nrps_claim}", flush=True)
# Use PyLTI1p3's NRPS service
# Note: This requires proper configuration in the tool config
from .adapters import DjangoToolConfig
tool_config = DjangoToolConfig.from_platform(self.platform)
# Pass the entire NRPS claim as service_data, not just the URL
nrps = NamesRolesProvisioningService(tool_config, self.nrps_claim)
# Fetch members
print("Calling nrps.get_members()...", flush=True)
members = nrps.get_members()
print(f"Fetched {len(members)} members from NRPS", flush=True)
logger.info(f"Fetched {len(members)} members from NRPS for platform {self.platform.name}")
return members
except Exception as e:
print(f"NRPS fetch error: {str(e)}", flush=True)
logger.error(f"NRPS fetch error: {str(e)}", exc_info=True)
except Exception:
return []
def sync_members_to_rbac_group(self, rbac_group):
@@ -99,7 +88,6 @@ class LTINRPSClient:
members = self.fetch_members()
if not members:
logger.warning("No members fetched from NRPS")
return {'synced': 0, 'removed': 0, 'synced_at': timezone.now().isoformat()}
processed_users = set()
@@ -121,26 +109,19 @@ class LTINRPSClient:
synced_count += 1
except Exception as e:
logger.error(f"Error syncing NRPS member {member.get('user_id')}: {str(e)}")
except Exception:
continue
# Remove unenrolled users if configured
removed_count = 0
if self.platform.remove_from_groups_on_unenroll:
from rbac.models import RBACMembership
removed = RBACMembership.objects.filter(rbac_group=rbac_group).exclude(user_id__in=processed_users)
removed_count = removed.count()
removed.delete()
logger.info(f"Removed {removed_count} unenrolled users from RBAC group {rbac_group.name}")
result = {'synced': synced_count, 'removed': removed_count, 'synced_at': timezone.now().isoformat()}
logger.info(f"NRPS sync complete for {rbac_group.name}: {result}")
return result
def _get_or_create_user_from_nrps(self, member):
@@ -155,7 +136,6 @@ class LTINRPSClient:
"""
user_id = member.get('user_id')
if not user_id:
logger.warning("NRPS member missing user_id")
return None
# Get user details from NRPS data
@@ -176,27 +156,22 @@ class LTINRPSClient:
if email and user.email != email:
user.email = email
update_fields.append('email')
print(f"Updating email for {user.username}: {user.email} -> {email}", flush=True)
# Update name fields if changed
if given_name and user.first_name != given_name:
user.first_name = given_name
update_fields.append('first_name')
print(f"Updating first_name for {user.username}: {user.first_name} -> {given_name}", flush=True)
if family_name and user.last_name != family_name:
user.last_name = family_name
update_fields.append('last_name')
print(f"Updating last_name for {user.username}: {user.last_name} -> {family_name}", flush=True)
if name and user.name != name:
user.name = name
update_fields.append('name')
print(f"Updating name for {user.username}: {user.name} -> {name}", flush=True)
if update_fields:
user.save(update_fields=update_fields)
logger.info(f"Updated user details for {user.username} via NRPS sync")
# Update mapping cache
mapping_update_fields = []
@@ -225,8 +200,6 @@ class LTINRPSClient:
# Check if username exists
if User.objects.filter(username=username).exists():
import hashlib
username = f"{username}_{hashlib.md5(user_id.encode()).hexdigest()[:6]}"
# Create user
@@ -238,12 +211,8 @@ class LTINRPSClient:
# Mark email as verified
if email:
try:
from allauth.account.models import EmailAddress
EmailAddress.objects.create(user=user, email=email, verified=True, primary=True)
except Exception as e:
logger.warning(f"Could not create EmailAddress for NRPS user: {e}")
logger.info(f"Created user {username} from NRPS data")
except Exception:
pass
return user

View File

@@ -10,8 +10,11 @@ Implements the LTI 1.3 / LTI Advantage flow:
- Manual NRPS Sync
"""
import logging
import json
import uuid
from urllib.parse import urlencode
import jwt
from django.http import HttpResponseRedirect, JsonResponse
from django.shortcuts import get_object_or_404, render
from django.urls import reverse
@@ -38,11 +41,9 @@ from .handlers import (
provision_lti_user,
validate_lti_session,
)
from .models import LTILaunchLog, LTIPlatform, LTIResourceLink
from .models import LTILaunchLog, LTIPlatform, LTIResourceLink, LTIUserMapping
from .services import LTINRPSClient
logger = logging.getLogger(__name__)
def get_client_ip(request):
"""Get client IP address from request"""
@@ -70,13 +71,7 @@ class OIDCLoginView(View):
def handle_oidc_login(self, request):
"""Handle OIDC login initiation"""
print("=== OIDC Login Started ===", flush=True)
logger.info("=== OIDC Login Started ===")
try:
# Get all request parameters for debugging
all_params = dict(request.GET.items()) if request.method == 'GET' else dict(request.POST.items())
print(f"All OIDC request params: {all_params}", flush=True)
# Get target_link_uri and other OIDC params
target_link_uri = request.GET.get('target_link_uri') or request.POST.get('target_link_uri')
iss = request.GET.get('iss') or request.POST.get('iss')
@@ -84,23 +79,14 @@ class OIDCLoginView(View):
login_hint = request.GET.get('login_hint') or request.POST.get('login_hint')
lti_message_hint = request.GET.get('lti_message_hint') or request.POST.get('lti_message_hint')
print(f"OIDC params - iss: {iss}, client_id: {client_id}, target: {target_link_uri}", flush=True)
print(f"login_hint: {login_hint}, lti_message_hint: {lti_message_hint}", flush=True)
logger.info(f"OIDC params - iss: {iss}, client_id: {client_id}, target: {target_link_uri}")
if not all([target_link_uri, iss, client_id]):
print("ERROR: Missing OIDC parameters", flush=True)
logger.error("Missing OIDC parameters")
return JsonResponse({'error': 'Missing required OIDC parameters'}, status=400)
# Get platform configuration
platform = get_object_or_404(LTIPlatform, platform_id=iss, client_id=client_id, active=True)
print(f"Found platform: {platform.name}", flush=True)
logger.info(f"Found platform: {platform.name}")
# Create tool config for this platform
tool_config = DjangoToolConfig.from_platform(platform)
print(f"Tool config: {tool_config._config}", flush=True)
# Wrap Django request for PyLTI1p3
lti_request = DjangoRequest(request)
@@ -109,30 +95,15 @@ class OIDCLoginView(View):
session_service = DjangoSessionService(request)
cookie_service = DjangoSessionService(request) # Using same service for cookies
print("Creating OIDCLogin...", flush=True)
oidc_login = OIDCLogin(lti_request, tool_config, session_service=session_service, cookie_service=cookie_service)
print("OIDCLogin created successfully", flush=True)
# Redirect to platform's authorization endpoint
print(f"Target link URI: {target_link_uri}", flush=True)
print(f"Auth login URL: {platform.auth_login_url}", flush=True)
try:
print("Calling enable_check_cookies()...", flush=True)
oidc_with_cookies = oidc_login.enable_check_cookies()
print(f"Calling redirect({target_link_uri})...", flush=True)
redirect_url = oidc_with_cookies.redirect(target_link_uri)
print(f"Redirect returned: '{redirect_url}'", flush=True)
print(f"OIDC redirect URL type: {type(redirect_url)}", flush=True)
print(f"OIDC redirecting to: {redirect_url}", flush=True)
logger.info(f"OIDC redirecting to: {redirect_url}")
if not redirect_url:
print("PyLTI1p3 redirect failed, building URL manually...", flush=True)
# Manual OIDC redirect construction with all required OAuth 2.0 parameters
import uuid
from urllib.parse import urlencode
state = str(uuid.uuid4())
nonce = str(uuid.uuid4())
@@ -157,21 +128,14 @@ class OIDCLoginView(View):
params['lti_message_hint'] = lti_message_hint
redirect_url = f"{platform.auth_login_url}?{urlencode(params)}"
print(f"Manually built redirect URL: {redirect_url}", flush=True)
return HttpResponseRedirect(redirect_url)
except Exception as e:
print(f"ERROR in OIDC redirect: {str(e)}", flush=True)
import traceback
traceback.print_exc()
except Exception:
raise
except LtiException as e:
logger.error(f"LTI OIDC Login Error: {str(e)}")
return render(request, 'lti/launch_error.html', {'error': 'OIDC Login Failed', 'message': str(e)}, status=400)
except Exception as e:
logger.error(f"OIDC Login Error: {str(e)}", exc_info=True)
except Exception:
return JsonResponse({'error': 'Internal server error during OIDC login'}, status=500)
@@ -186,8 +150,6 @@ class LaunchView(View):
def post(self, request):
"""Handle LTI launch with JWT validation"""
print("=== LTI Launch Started ===", flush=True)
logger.info("=== LTI Launch Started ===")
platform = None
user = None
error_message = ''
@@ -200,16 +162,12 @@ class LaunchView(View):
raise ValueError("Missing id_token in launch request")
# Decode JWT to get issuer (without validation first)
import jwt
unverified = jwt.decode(id_token, options={"verify_signature": False})
iss = unverified.get('iss')
aud = unverified.get('aud')
# Get platform
platform = get_object_or_404(LTIPlatform, platform_id=iss, client_id=aud, active=True)
print(f"Launch from platform: {platform.name}", flush=True)
logger.info(f"Launch from platform: {platform.name}")
# Create tool config
tool_config = DjangoToolConfig.from_platform(platform)
@@ -233,6 +191,28 @@ class LaunchView(View):
launch_data = message_launch.get_launch_data()
claims = self.sanitize_claims(launch_data)
# Print all JWT information for debugging
print("\n" + "=" * 80)
print("LTI JWT DECRYPTED - ALL CLAIMS:")
print("=" * 80)
print(f"Issuer (iss): {launch_data.get('iss')}")
print(f"Subject (sub): {launch_data.get('sub')}")
print(f"Email: {launch_data.get('email')}")
print(f"Given Name: {launch_data.get('given_name')}")
print(f"Family Name: {launch_data.get('family_name')}")
print(f"Full Name: {launch_data.get('name')}")
print(f"Roles: {launch_data.get('https://purl.imsglobal.org/spec/lti/claim/roles')}")
context = launch_data.get('https://purl.imsglobal.org/spec/lti/claim/context', {})
print(f"Context ID: {context.get('id')}")
print(f"Context Title: {context.get('title')}")
print(f"Context Label: {context.get('label')}")
resource_link = launch_data.get('https://purl.imsglobal.org/spec/lti/claim/resource_link', {})
print(f"Resource Link ID: {resource_link.get('id')}")
print(f"Resource Link Title: {resource_link.get('title')}")
print("\nFull Launch Data:")
print(json.dumps(launch_data, indent=2))
print("=" * 80 + "\n")
# Extract key claims
sub = launch_data.get('sub')
resource_link = launch_data.get('https://purl.imsglobal.org/spec/lti/claim/resource_link', {})
@@ -247,16 +227,10 @@ class LaunchView(View):
return self.handle_deep_linking_launch(request, message_launch, platform, launch_data)
# Provision user
print(f"Provisioning user, sub: {sub}", flush=True)
logger.info(f"Provisioning user, sub: {sub}")
if platform.auto_create_users:
user = provision_lti_user(platform, launch_data)
print(f"User provisioned: {user.username}", flush=True)
logger.info(f"User provisioned: {user.username}")
else:
# Must find existing user
from .models import LTIUserMapping
mapping = LTIUserMapping.objects.filter(platform=platform, lti_user_id=sub).first()
if not mapping:
raise ValueError("User auto-creation disabled and no existing mapping found")
@@ -264,38 +238,29 @@ class LaunchView(View):
# Provision context (category + RBAC group)
if 'https://purl.imsglobal.org/spec/lti/claim/context' in launch_data:
logger.info("Provisioning context...")
category, rbac_group, resource_link_obj = provision_lti_context(platform, launch_data, resource_link_id)
logger.info(f"Context provisioned: category={category.title if category else None}")
# Apply roles
apply_lti_roles(user, platform, roles, rbac_group)
logger.info(f"Roles applied: {roles}")
else:
# No context - might be a direct media embed
resource_link_obj = None
# Create session
create_lti_session(request, user, message_launch, platform)
logger.info("LTI session created")
# Log successful launch
LTILaunchLog.objects.create(platform=platform, user=user, resource_link=resource_link_obj, launch_type='resource_link', success=True, claims=claims, ip_address=get_client_ip(request))
logger.info("Launch logged")
# Determine where to redirect
redirect_url = self.determine_redirect(launch_data, resource_link_obj)
print(f"=== Launch Success - Redirecting to: {redirect_url} ===", flush=True)
logger.info(f"=== Launch Success - Redirecting to: {redirect_url} ===")
return HttpResponseRedirect(redirect_url)
except LtiException as e:
error_message = f"LTI Launch Error: {str(e)}"
logger.error(error_message)
except Exception as e:
error_message = f"Launch Error: {str(e)}"
logger.error(error_message, exc_info=True)
# Log failed launch
if platform:
@@ -372,24 +337,15 @@ class MyMediaLTIView(View):
def get(self, request):
"""Display my media page"""
print(f"=== My Media LTI View - User: {request.user} ===", flush=True)
logger.info(f"=== My Media LTI View - User: {request.user} ===")
# Validate LTI session
lti_session = validate_lti_session(request)
print(f"LTI session valid: {bool(lti_session)}", flush=True)
logger.info(f"LTI session valid: {bool(lti_session)}")
if not lti_session:
print("ERROR: LTI session validation failed", flush=True)
logger.error("LTI session validation failed")
return JsonResponse({'error': 'Not authenticated via LTI'}, status=403)
# Redirect to user's profile page
# The existing user profile page is already iframe-compatible
profile_url = f"/user/{request.user.username}"
print(f"Redirecting to profile: {profile_url}", flush=True)
logger.info(f"Redirecting to profile: {profile_url}")
return HttpResponseRedirect(profile_url)
@@ -484,5 +440,4 @@ class ManualSyncView(APIView):
)
except Exception as e:
logger.error(f"NRPS sync error: {str(e)}", exc_info=True)
return Response({'error': 'Sync failed', 'message': str(e)}, status=status.HTTP_500_INTERNAL_SERVER_ERROR)