This commit is contained in:
Markos Gogoulos
2025-12-29 18:21:44 +02:00
parent b9899476b9
commit 06bc64b2c4
11 changed files with 249 additions and 104 deletions

View File

@@ -1,13 +1,11 @@
# Generated by Django 5.2.6 on 2025-12-24 15:08 # Generated by Django 5.2.6 on 2025-12-29 16:15
import django.db.models.deletion
from django.db import migrations, models from django.db import migrations, models
class Migration(migrations.Migration): class Migration(migrations.Migration):
dependencies = [ dependencies = [
('files', '0014_alter_subtitle_options_and_more'), ('files', '0014_alter_subtitle_options_and_more'),
('lti', '__first__'),
] ]
operations = [ operations = [
@@ -21,11 +19,4 @@ class Migration(migrations.Migration):
name='lti_context_id', name='lti_context_id',
field=models.CharField(blank=True, db_index=True, help_text='LTI context ID from platform', max_length=255), field=models.CharField(blank=True, db_index=True, help_text='LTI context ID from platform', max_length=255),
), ),
migrations.AddField(
model_name='category',
name='lti_platform',
field=models.ForeignKey(
blank=True, help_text='LTI Platform if this is an LTI course', null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='categories', to='lti.ltiplatform'
),
),
] ]

View File

@@ -0,0 +1,21 @@
# Generated by Django 5.2.6 on 2025-12-29 16:15
import django.db.models.deletion
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('files', '0015_category_is_lms_course_category_lti_context_id'),
('lti', '0001_initial'),
]
operations = [
migrations.AddField(
model_name='category',
name='lti_platform',
field=models.ForeignKey(
blank=True, help_text='LTI Platform if this is an LTI course', null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='categories', to='lti.ltiplatform'
),
),
]

View File

@@ -8,10 +8,15 @@ import json
from typing import Any, Dict, Optional from typing import Any, Dict, Optional
from django.core.cache import cache from django.core.cache import cache
from pylti1p3.message_launch import MessageLaunch
from pylti1p3.oidc_login import OIDCLogin
from pylti1p3.registration import Registration from pylti1p3.registration import Registration
from pylti1p3.request import Request from pylti1p3.request import Request
from pylti1p3.tool_config import ToolConfAbstract from pylti1p3.tool_config import ToolConfAbstract
from .keys import load_private_key
from .models import LTIPlatform
class DjangoRequest(Request): class DjangoRequest(Request):
"""Django request adapter for PyLTI1p3""" """Django request adapter for PyLTI1p3"""
@@ -57,8 +62,6 @@ class DjangoOIDCLogin:
def get_redirect(self, redirect_url): def get_redirect(self, redirect_url):
"""Get the redirect object for OIDC login""" """Get the redirect object for OIDC login"""
from pylti1p3.oidc_login import OIDCLogin
oidc_login = OIDCLogin(self.lti_request, self.tool_config, session_service=self.launch_data_storage, cookie_service=self.launch_data_storage) oidc_login = OIDCLogin(self.lti_request, self.tool_config, session_service=self.launch_data_storage, cookie_service=self.launch_data_storage)
return oidc_login.enable_check_cookies().redirect(redirect_url) return oidc_login.enable_check_cookies().redirect(redirect_url)
@@ -75,7 +78,6 @@ class DjangoMessageLaunch:
def validate(self): def validate(self):
"""Validate the LTI launch message""" """Validate the LTI launch message"""
from pylti1p3.message_launch import MessageLaunch
# Create custom MessageLaunch that properly implements _get_request_param # Create custom MessageLaunch that properly implements _get_request_param
class CustomMessageLaunch(MessageLaunch): class CustomMessageLaunch(MessageLaunch):
@@ -224,8 +226,6 @@ class DjangoToolConfig(ToolConfAbstract):
if config.get('auth_audience'): if config.get('auth_audience'):
registration.set_auth_audience(config.get('auth_audience')) registration.set_auth_audience(config.get('auth_audience'))
registration.set_key_set_url(config.get('key_set_url')) registration.set_key_set_url(config.get('key_set_url'))
if config.get('key_set'):
registration.set_key_set(config.get('key_set'))
return registration return registration
@@ -247,8 +247,6 @@ class DjangoToolConfig(ToolConfAbstract):
if config.get('auth_audience'): if config.get('auth_audience'):
registration.set_auth_audience(config.get('auth_audience')) registration.set_auth_audience(config.get('auth_audience'))
registration.set_key_set_url(config.get('key_set_url')) registration.set_key_set_url(config.get('key_set_url'))
if config.get('key_set'):
registration.set_key_set(config.get('key_set'))
return registration return registration
@@ -280,25 +278,26 @@ class DjangoToolConfig(ToolConfAbstract):
return self.find_registration_by_params(iss, client_id) return self.find_registration_by_params(iss, client_id)
def get_jwks(self, iss, client_id=None): def get_jwks(self, iss, client_id=None):
"""Get JWKS from configuration""" """Get JWKS from configuration - returns None to fetch from URL"""
if iss not in self._config: # No caching - PyLTI1p3 will fetch from key_set_url
return None return None
config_dict = self._config[iss]
if client_id and config_dict.get('client_id') != client_id:
return None
return config_dict.get('key_set')
def get_iss(self): def get_iss(self):
"""Get all issuers""" """Get all issuers"""
return list(self._config.keys()) return list(self._config.keys())
def get_jwk(self, iss=None, client_id=None):
"""
Get private JWK for signing Deep Linking responses
PyLTI1p3 calls this to get the tool's private key for signing
"""
# Return MediaCMS's private key (same for all platforms)
return load_private_key()
@classmethod @classmethod
def from_platform(cls, platform): def from_platform(cls, platform):
"""Create ToolConfig from LTIPlatform model instance""" """Create ToolConfig from LTIPlatform model instance"""
from .models import LTIPlatform
if isinstance(platform, LTIPlatform): if isinstance(platform, LTIPlatform):
config = {platform.platform_id: platform.get_lti_config()} config = {platform.platform_id: platform.get_lti_config()}
return cls(config) return cls(config)
@@ -307,10 +306,8 @@ class DjangoToolConfig(ToolConfAbstract):
@classmethod @classmethod
def from_all_platforms(cls): def from_all_platforms(cls):
"""Create ToolConfig with all active platforms""" """Create ToolConfig with all platforms"""
from .models import LTIPlatform platforms = LTIPlatform.objects.filter()
platforms = LTIPlatform.objects.filter(active=True)
config = {} config = {}
for platform in platforms: for platform in platforms:

View File

@@ -10,6 +10,7 @@ from .models import (
LTIPlatform, LTIPlatform,
LTIResourceLink, LTIResourceLink,
LTIRoleMapping, LTIRoleMapping,
LTIToolKeys,
LTIUserMapping, LTIUserMapping,
) )
from .services import LTINRPSClient from .services import LTINRPSClient
@@ -19,28 +20,20 @@ from .services import LTINRPSClient
class LTIPlatformAdmin(admin.ModelAdmin): class LTIPlatformAdmin(admin.ModelAdmin):
"""Admin for LTI Platforms (Moodle instances)""" """Admin for LTI Platforms (Moodle instances)"""
list_display = ['name', 'platform_id', 'client_id', 'active_badge', 'nrps_enabled', 'deep_linking_enabled', 'created_at'] list_display = ['name', 'platform_id', 'client_id', 'nrps_enabled', 'deep_linking_enabled', 'created_at']
list_filter = ['active', 'enable_nrps', 'enable_deep_linking', 'created_at'] list_filter = ['enable_nrps', 'enable_deep_linking', 'created_at']
search_fields = ['name', 'platform_id', 'client_id'] search_fields = ['name', 'platform_id', 'client_id']
readonly_fields = ['created_at', 'updated_at', 'key_set_updated'] readonly_fields = ['created_at', 'updated_at']
fieldsets = ( fieldsets = (
('Basic Information', {'fields': ('name', 'platform_id', 'client_id', 'active')}), ('Basic Information', {'fields': ('name', 'platform_id', 'client_id')}),
('OIDC Endpoints', {'fields': ('auth_login_url', 'auth_token_url', 'auth_audience')}), ('OIDC Endpoints', {'fields': ('auth_login_url', 'auth_token_url', 'auth_audience')}),
('JWK Configuration', {'fields': ('key_set_url', 'key_set', 'key_set_updated'), 'classes': ('collapse',)}), ('JWK Configuration', {'fields': ('key_set_url',), 'classes': ('collapse',)}),
('Deployment & Features', {'fields': ('deployment_ids', 'enable_nrps', 'enable_deep_linking')}), ('Deployment & Features', {'fields': ('deployment_ids', 'enable_nrps', 'enable_deep_linking')}),
('Auto-Provisioning Settings', {'fields': ('auto_create_categories', 'auto_create_users', 'auto_sync_roles', 'remove_from_groups_on_unenroll')}), ('Auto-Provisioning Settings', {'fields': ('remove_from_groups_on_unenroll',)}),
('Timestamps', {'fields': ('created_at', 'updated_at'), 'classes': ('collapse',)}), ('Timestamps', {'fields': ('created_at', 'updated_at'), 'classes': ('collapse',)}),
) )
def active_badge(self, obj):
"""Display active status as badge"""
if obj.active:
return format_html('<span style="color: green;">✓ Active</span>')
return format_html('<span style="color: red;">✗ Inactive</span>')
active_badge.short_description = 'Status'
def nrps_enabled(self, obj): def nrps_enabled(self, obj):
return '' if obj.enable_nrps else '' return '' if obj.enable_nrps else ''
@@ -171,14 +164,14 @@ class LTIRoleMappingAdmin(admin.ModelAdmin):
class LTILaunchLogAdmin(admin.ModelAdmin): class LTILaunchLogAdmin(admin.ModelAdmin):
"""Admin for LTI Launch Logs""" """Admin for LTI Launch Logs"""
list_display = ['created_at', 'platform', 'user_link', 'launch_type', 'success_badge', 'ip_address'] list_display = ['created_at', 'platform', 'user_link', 'launch_type', 'success_badge']
list_filter = ['success', 'launch_type', 'platform', 'created_at'] list_filter = ['success', 'launch_type', 'platform', 'created_at']
search_fields = ['user__username', 'ip_address', 'error_message'] search_fields = ['user__username', 'error_message']
readonly_fields = ['created_at', 'claims'] readonly_fields = ['created_at', 'claims']
date_hierarchy = 'created_at' date_hierarchy = 'created_at'
fieldsets = ( fieldsets = (
('Launch Info', {'fields': ('platform', 'user', 'resource_link', 'launch_type', 'success', 'ip_address', 'created_at')}), ('Launch Info', {'fields': ('platform', 'user', 'resource_link', 'launch_type', 'success', 'created_at')}),
('Error Details', {'fields': ('error_message',), 'classes': ('collapse',)}), ('Error Details', {'fields': ('error_message',), 'classes': ('collapse',)}),
('Claims Data', {'fields': ('claims',), 'classes': ('collapse',)}), ('Claims Data', {'fields': ('claims',), 'classes': ('collapse',)}),
) )
@@ -204,3 +197,43 @@ class LTILaunchLogAdmin(admin.ModelAdmin):
def has_change_permission(self, request, obj=None): def has_change_permission(self, request, obj=None):
"""Make launch logs read-only""" """Make launch logs read-only"""
return False return False
@admin.register(LTIToolKeys)
class LTIToolKeysAdmin(admin.ModelAdmin):
"""Admin for LTI Tool RSA Keys"""
list_display = ['key_id', 'created_at', 'updated_at']
readonly_fields = ['key_id', 'created_at', 'updated_at', 'public_key_display']
fieldsets = (
('Key Information', {'fields': ('key_id', 'created_at', 'updated_at')}),
('Public Key (for JWKS)', {'fields': ('public_key_display',)}),
('Private Key (Keep Secure!)', {'fields': ('private_key_jwk',), 'classes': ('collapse',), 'description': '⚠️ This is your private signing key. Do not share it!'}),
)
actions = ['regenerate_keys']
def public_key_display(self, obj):
"""Display public key in readable format"""
import json
return format_html('<pre>{}</pre>', json.dumps(obj.public_key_jwk, indent=2))
public_key_display.short_description = 'Public Key (JWK)'
def regenerate_keys(self, request, queryset):
"""Regenerate keys for selected instances"""
for key_obj in queryset:
key_obj.generate_keys()
self.message_user(request, f"Keys regenerated for {key_obj.key_id}", messages.SUCCESS)
regenerate_keys.short_description = 'Regenerate RSA keys'
def has_add_permission(self, request):
"""Only allow one key pair - disable manual add if exists"""
return not LTIToolKeys.objects.exists()
def has_delete_permission(self, request, obj=None):
"""Prevent accidental deletion of keys"""
return False

View File

@@ -1,5 +1,7 @@
from django.apps import AppConfig from django.apps import AppConfig
from .keys import ensure_keys_exist
class LtiConfig(AppConfig): class LtiConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField' default_auto_field = 'django.db.models.BigAutoField'
@@ -7,6 +9,10 @@ class LtiConfig(AppConfig):
verbose_name = 'LTI 1.3 Integration' verbose_name = 'LTI 1.3 Integration'
def ready(self): def ready(self):
"""Import signal handlers when app is ready""" """Initialize LTI app - ensure keys exist"""
# Import any signals here if needed in the future # Ensure RSA key pair exists for signing Deep Linking responses
pass try:
ensure_keys_exist()
except Exception:
# Don't block startup if key generation fails
pass

View File

@@ -4,6 +4,8 @@ LTI Deep Linking 2.0 for MediaCMS
Allows instructors to select media from MediaCMS library and embed in Moodle courses Allows instructors to select media from MediaCMS library and embed in Moodle courses
""" """
import traceback
from django.contrib.auth.decorators import login_required from django.contrib.auth.decorators import login_required
from django.http import JsonResponse from django.http import JsonResponse
from django.shortcuts import render from django.shortcuts import render
@@ -11,6 +13,8 @@ from django.urls import reverse
from django.utils.decorators import method_decorator from django.utils.decorators import method_decorator
from django.views import View from django.views import View
from django.views.decorators.csrf import csrf_exempt from django.views.decorators.csrf import csrf_exempt
from pylti1p3.deep_link import DeepLink
from pylti1p3.deep_link_resource import DeepLinkResource
from files.models import Media from files.models import Media
from files.views.media import MediaList from files.views.media import MediaList
@@ -125,9 +129,6 @@ class SelectMediaView(View):
""" """
Create JWT response for deep linking using PyLTI1p3 Create JWT response for deep linking using PyLTI1p3
""" """
from pylti1p3.deep_link import DeepLink
from pylti1p3.deep_link_resource import DeepLinkResource
try: try:
platform_id = deep_link_data['platform_id'] platform_id = deep_link_data['platform_id']
platform = LTIPlatform.objects.get(id=platform_id) platform = LTIPlatform.objects.get(id=platform_id)
@@ -182,7 +183,5 @@ class SelectMediaView(View):
except Exception as e: except Exception as e:
# Log error for debugging # Log error for debugging
import traceback
traceback.print_exc() traceback.print_exc()
raise ValueError(f"Failed to create Deep Linking JWT: {str(e)}") raise ValueError(f"Failed to create Deep Linking JWT: {str(e)}")

View File

@@ -260,9 +260,7 @@ def apply_lti_roles(user, platform, lti_roles, rbac_group):
if role_global: if role_global:
global_role = get_higher_privilege_global(global_role, role_global) global_role = get_higher_privilege_global(global_role, role_global)
# Apply global role if auto_sync_roles is enabled user.set_role_from_mapping(global_role)
if platform.auto_sync_roles:
user.set_role_from_mapping(global_role)
# Determine group role # Determine group role
group_role = 'member' group_role = 'member'

38
lti/keys.py Normal file
View File

@@ -0,0 +1,38 @@
"""
LTI Key Management for MediaCMS
Manages RSA keys for signing Deep Linking responses (stored in database)
"""
def load_private_key():
"""Load private key from database"""
from .models import LTIToolKeys
key_obj = LTIToolKeys.get_or_create_keys()
return key_obj.private_key_jwk
def load_public_key():
"""Load public key from database"""
from .models import LTIToolKeys
key_obj = LTIToolKeys.get_or_create_keys()
return key_obj.public_key_jwk
def get_jwks():
"""
Get JWKS (JSON Web Key Set) for public keys
Returns public keys in JWKS format for the /lti/jwks/ endpoint
"""
public_key = load_public_key()
return {'keys': [public_key]}
def ensure_keys_exist():
"""Ensure key pair exists in database, generate if not"""
from .models import LTIToolKeys
LTIToolKeys.get_or_create_keys()

View File

@@ -1,4 +1,4 @@
# Generated by Django 5.2.6 on 2025-12-24 15:18 # Generated by Django 5.2.6 on 2025-12-29 16:15
import django.db.models.deletion import django.db.models.deletion
from django.conf import settings from django.conf import settings
@@ -9,11 +9,27 @@ class Migration(migrations.Migration):
initial = True initial = True
dependencies = [ dependencies = [
('files', '0015_category_is_lms_course_category_lti_context_id'),
('rbac', '0003_alter_rbacgroup_members'), ('rbac', '0003_alter_rbacgroup_members'),
migrations.swappable_dependency(settings.AUTH_USER_MODEL), migrations.swappable_dependency(settings.AUTH_USER_MODEL),
] ]
operations = [ operations = [
migrations.CreateModel(
name='LTIToolKeys',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('key_id', models.CharField(default='mediacms-lti-key', help_text='Key identifier', max_length=255, unique=True)),
('private_key_jwk', models.JSONField(help_text='Private key in JWK format (for signing)')),
('public_key_jwk', models.JSONField(help_text='Public key in JWK format (for JWKS endpoint)')),
('created_at', models.DateTimeField(auto_now_add=True)),
('updated_at', models.DateTimeField(auto_now=True)),
],
options={
'verbose_name': 'LTI Tool Keys',
'verbose_name_plural': 'LTI Tool Keys',
},
),
migrations.CreateModel( migrations.CreateModel(
name='LTIPlatform', name='LTIPlatform',
fields=[ fields=[
@@ -25,18 +41,12 @@ class Migration(migrations.Migration):
('auth_token_url', models.URLField(help_text='OAuth2 token endpoint URL')), ('auth_token_url', models.URLField(help_text='OAuth2 token endpoint URL')),
('auth_audience', models.URLField(blank=True, help_text='OAuth2 audience (optional)', null=True)), ('auth_audience', models.URLField(blank=True, help_text='OAuth2 audience (optional)', null=True)),
('key_set_url', models.URLField(help_text="Platform's public JWK Set URL")), ('key_set_url', models.URLField(help_text="Platform's public JWK Set URL")),
('key_set', models.JSONField(blank=True, help_text='Cached JWK Set (auto-fetched)', null=True)),
('key_set_updated', models.DateTimeField(blank=True, help_text='Last time JWK Set was fetched', null=True)),
('deployment_ids', models.JSONField(default=list, help_text='List of deployment IDs for this platform')), ('deployment_ids', models.JSONField(default=list, help_text='List of deployment IDs for this platform')),
('enable_nrps', models.BooleanField(default=True, help_text='Enable Names and Role Provisioning Service')), ('enable_nrps', models.BooleanField(default=True, help_text='Enable Names and Role Provisioning Service')),
('enable_deep_linking', models.BooleanField(default=True, help_text='Enable Deep Linking 2.0')), ('enable_deep_linking', models.BooleanField(default=True, help_text='Enable Deep Linking 2.0')),
('auto_create_categories', models.BooleanField(default=True, help_text='Automatically create categories for courses')),
('auto_create_users', models.BooleanField(default=True, help_text='Automatically create users on first launch')),
('auto_sync_roles', models.BooleanField(default=True, help_text='Automatically sync user roles from LTI')),
('remove_from_groups_on_unenroll', models.BooleanField(default=False, help_text="Remove users from RBAC groups when they're no longer in the course")), ('remove_from_groups_on_unenroll', models.BooleanField(default=False, help_text="Remove users from RBAC groups when they're no longer in the course")),
('created_at', models.DateTimeField(auto_now_add=True)), ('created_at', models.DateTimeField(auto_now_add=True)),
('updated_at', models.DateTimeField(auto_now=True)), ('updated_at', models.DateTimeField(auto_now=True)),
('active', models.BooleanField(default=True, help_text='Whether this platform is currently active')),
], ],
options={ options={
'verbose_name': 'LTI Platform', 'verbose_name': 'LTI Platform',
@@ -80,7 +90,6 @@ class Migration(migrations.Migration):
('success', models.BooleanField(db_index=True, default=True, help_text='Whether the launch was successful')), ('success', models.BooleanField(db_index=True, default=True, help_text='Whether the launch was successful')),
('error_message', models.TextField(blank=True, help_text='Error message if launch failed')), ('error_message', models.TextField(blank=True, help_text='Error message if launch failed')),
('claims', models.JSONField(help_text='Sanitized LTI claims from the launch')), ('claims', models.JSONField(help_text='Sanitized LTI claims from the launch')),
('ip_address', models.GenericIPAddressField(blank=True, help_text='IP address of the user', null=True)),
('created_at', models.DateTimeField(auto_now_add=True, db_index=True)), ('created_at', models.DateTimeField(auto_now_add=True, db_index=True)),
( (
'user', 'user',
@@ -111,13 +120,7 @@ class Migration(migrations.Migration):
'global_role', 'global_role',
models.CharField( models.CharField(
blank=True, blank=True,
choices=[ choices=[('advancedUser', 'Advanced User'), ('editor', 'MediaCMS Editor'), ('manager', 'MediaCMS Manager'), ('admin', 'MediaCMS Administrator')],
('user', 'Authenticated User'),
('advancedUser', 'Advanced User'),
('editor', 'MediaCMS Editor'),
('manager', 'MediaCMS Manager'),
('admin', 'MediaCMS Administrator'),
],
help_text='MediaCMS global role to assign', help_text='MediaCMS global role to assign',
max_length=20, max_length=20,
), ),

View File

@@ -1,4 +1,10 @@
import json
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from django.db import models from django.db import models
from jwcrypto import jwk
class LTIPlatform(models.Model): class LTIPlatform(models.Model):
@@ -13,22 +19,16 @@ class LTIPlatform(models.Model):
auth_audience = models.URLField(blank=True, null=True, help_text="OAuth2 audience (optional)") auth_audience = models.URLField(blank=True, null=True, help_text="OAuth2 audience (optional)")
key_set_url = models.URLField(help_text="Platform's public JWK Set URL") key_set_url = models.URLField(help_text="Platform's public JWK Set URL")
key_set = models.JSONField(blank=True, null=True, help_text="Cached JWK Set (auto-fetched)")
key_set_updated = models.DateTimeField(null=True, blank=True, help_text="Last time JWK Set was fetched")
deployment_ids = models.JSONField(default=list, help_text="List of deployment IDs for this platform") deployment_ids = models.JSONField(default=list, help_text="List of deployment IDs for this platform")
enable_nrps = models.BooleanField(default=True, help_text="Enable Names and Role Provisioning Service") enable_nrps = models.BooleanField(default=True, help_text="Enable Names and Role Provisioning Service")
enable_deep_linking = models.BooleanField(default=True, help_text="Enable Deep Linking 2.0") enable_deep_linking = models.BooleanField(default=True, help_text="Enable Deep Linking 2.0")
auto_create_categories = models.BooleanField(default=True, help_text="Automatically create categories for courses")
auto_create_users = models.BooleanField(default=True, help_text="Automatically create users on first launch")
auto_sync_roles = models.BooleanField(default=True, help_text="Automatically sync user roles from LTI")
remove_from_groups_on_unenroll = models.BooleanField(default=False, help_text="Remove users from RBAC groups when they're no longer in the course") remove_from_groups_on_unenroll = models.BooleanField(default=False, help_text="Remove users from RBAC groups when they're no longer in the course")
# Timestamps # Timestamps
created_at = models.DateTimeField(auto_now_add=True) created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True) updated_at = models.DateTimeField(auto_now=True)
active = models.BooleanField(default=True, help_text="Whether this platform is currently active")
class Meta: class Meta:
verbose_name = 'LTI Platform' verbose_name = 'LTI Platform'
@@ -47,7 +47,6 @@ class LTIPlatform(models.Model):
'auth_token_url': self.auth_token_url, 'auth_token_url': self.auth_token_url,
'auth_audience': self.auth_audience, 'auth_audience': self.auth_audience,
'key_set_url': self.key_set_url, 'key_set_url': self.key_set_url,
'key_set': self.key_set,
'deployment_ids': self.deployment_ids, 'deployment_ids': self.deployment_ids,
} }
@@ -149,7 +148,6 @@ class LTILaunchLog(models.Model):
error_message = models.TextField(blank=True, help_text="Error message if launch failed") error_message = models.TextField(blank=True, help_text="Error message if launch failed")
claims = models.JSONField(help_text="Sanitized LTI claims from the launch") claims = models.JSONField(help_text="Sanitized LTI claims from the launch")
ip_address = models.GenericIPAddressField(null=True, blank=True, help_text="IP address of the user")
created_at = models.DateTimeField(auto_now_add=True, db_index=True) created_at = models.DateTimeField(auto_now_add=True, db_index=True)
class Meta: class Meta:
@@ -165,3 +163,75 @@ class LTILaunchLog(models.Model):
status = "" if self.success else "" status = "" if self.success else ""
user_str = self.user.username if self.user else "Unknown" user_str = self.user.username if self.user else "Unknown"
return f"{status} {user_str} @ {self.platform.name} ({self.created_at.strftime('%Y-%m-%d %H:%M')})" return f"{status} {user_str} @ {self.platform.name} ({self.created_at.strftime('%Y-%m-%d %H:%M')})"
class LTIToolKeys(models.Model):
"""
Stores MediaCMS's RSA key pair for signing LTI responses (e.g., Deep Linking)
Only one instance should exist (singleton pattern)
"""
key_id = models.CharField(max_length=255, unique=True, default='mediacms-lti-key', help_text='Key identifier')
# JWK format keys
private_key_jwk = models.JSONField(help_text='Private key in JWK format (for signing)')
public_key_jwk = models.JSONField(help_text='Public key in JWK format (for JWKS endpoint)')
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
class Meta:
verbose_name = 'LTI Tool Keys'
verbose_name_plural = 'LTI Tool Keys'
def __str__(self):
return f"LTI Keys ({self.key_id})"
@classmethod
def get_or_create_keys(cls):
"""Get or create the default key pair"""
key_obj, created = cls.objects.get_or_create(
key_id='mediacms-lti-key',
defaults={'private_key_jwk': {}, 'public_key_jwk': {}}, # Will be populated by save()
)
# If keys are empty, generate them
if created or not key_obj.private_key_jwk or not key_obj.public_key_jwk:
key_obj.generate_keys()
return key_obj
def generate_keys(self):
"""Generate new RSA key pair"""
# Generate RSA key pair
private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048, backend=default_backend())
public_key = private_key.public_key()
# Convert to PEM
private_pem = private_key.private_bytes(encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.PKCS8, encryption_algorithm=serialization.NoEncryption())
public_pem = public_key.public_bytes(encoding=serialization.Encoding.PEM, format=serialization.PublicFormat.SubjectPublicKeyInfo)
# Convert to JWK
private_jwk = jwk.JWK.from_pem(private_pem)
public_jwk = jwk.JWK.from_pem(public_pem)
# Add metadata
private_jwk_dict = json.loads(private_jwk.export())
private_jwk_dict['kid'] = self.key_id
private_jwk_dict['alg'] = 'RS256'
private_jwk_dict['use'] = 'sig'
public_jwk_dict = json.loads(public_jwk.export_public())
public_jwk_dict['kid'] = self.key_id
public_jwk_dict['alg'] = 'RS256'
public_jwk_dict['use'] = 'sig'
# Save to database
self.private_key_jwk = private_jwk_dict
self.public_key_jwk = public_jwk_dict
self.save()
return private_jwk_dict, public_jwk_dict

View File

@@ -40,6 +40,7 @@ from .handlers import (
provision_lti_user, provision_lti_user,
validate_lti_session, validate_lti_session,
) )
from .keys import get_jwks
from .models import LTILaunchLog, LTIPlatform, LTIResourceLink from .models import LTILaunchLog, LTIPlatform, LTIResourceLink
from .services import LTINRPSClient from .services import LTINRPSClient
@@ -82,7 +83,7 @@ class OIDCLoginView(View):
return JsonResponse({'error': 'Missing required OIDC parameters'}, status=400) return JsonResponse({'error': 'Missing required OIDC parameters'}, status=400)
# Get platform configuration # Get platform configuration
platform = get_object_or_404(LTIPlatform, platform_id=iss, client_id=client_id, active=True) platform = get_object_or_404(LTIPlatform, platform_id=iss, client_id=client_id)
# Create tool config for this platform # Create tool config for this platform
tool_config = DjangoToolConfig.from_platform(platform) tool_config = DjangoToolConfig.from_platform(platform)
@@ -168,7 +169,7 @@ class LaunchView(View):
aud = unverified.get('aud') aud = unverified.get('aud')
# Get platform # Get platform
platform = get_object_or_404(LTIPlatform, platform_id=iss, client_id=aud, active=True) platform = get_object_or_404(LTIPlatform, platform_id=iss, client_id=aud)
# Create tool config # Create tool config
tool_config = DjangoToolConfig.from_platform(platform) tool_config = DjangoToolConfig.from_platform(platform)
@@ -193,7 +194,6 @@ class LaunchView(View):
claims = self.sanitize_claims(launch_data) claims = self.sanitize_claims(launch_data)
# Extract key claims # Extract key claims
sub = launch_data.get('sub')
resource_link = launch_data.get('https://purl.imsglobal.org/spec/lti/claim/resource_link', {}) resource_link = launch_data.get('https://purl.imsglobal.org/spec/lti/claim/resource_link', {})
resource_link_id = resource_link.get('id', 'default') resource_link_id = resource_link.get('id', 'default')
roles = launch_data.get('https://purl.imsglobal.org/spec/lti/claim/roles', []) roles = launch_data.get('https://purl.imsglobal.org/spec/lti/claim/roles', [])
@@ -205,17 +205,7 @@ class LaunchView(View):
# Deep linking request - handle separately # Deep linking request - handle separately
return self.handle_deep_linking_launch(request, message_launch, platform, launch_data) return self.handle_deep_linking_launch(request, message_launch, platform, launch_data)
# Provision user user = provision_lti_user(platform, launch_data)
if platform.auto_create_users:
user = provision_lti_user(platform, launch_data)
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")
user = mapping.user
# Provision context (category + RBAC group) # Provision context (category + RBAC group)
if 'https://purl.imsglobal.org/spec/lti/claim/context' in launch_data: if 'https://purl.imsglobal.org/spec/lti/claim/context' in launch_data:
@@ -231,7 +221,7 @@ class LaunchView(View):
create_lti_session(request, user, message_launch, platform) create_lti_session(request, user, message_launch, platform)
# Log successful launch # 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)) LTILaunchLog.objects.create(platform=platform, user=user, resource_link=resource_link_obj, launch_type='resource_link', success=True, claims=claims)
# Determine where to redirect # Determine where to redirect
redirect_url = self.determine_redirect(launch_data, resource_link_obj) redirect_url = self.determine_redirect(launch_data, resource_link_obj)
@@ -245,7 +235,7 @@ class LaunchView(View):
# Log failed launch # Log failed launch
if platform: if platform:
LTILaunchLog.objects.create(platform=platform, user=user, launch_type='resource_link', success=False, error_message=error_message, claims=claims, ip_address=get_client_ip(request)) LTILaunchLog.objects.create(platform=platform, user=user, launch_type='resource_link', success=False, error_message=error_message, claims=claims)
return render(request, 'lti/launch_error.html', {'error': 'LTI Launch Failed', 'message': error_message}, status=400) return render(request, 'lti/launch_error.html', {'error': 'LTI Launch Failed', 'message': error_message}, status=400)
@@ -311,14 +301,13 @@ class JWKSView(View):
""" """
JWKS Endpoint - Provides tool's public keys JWKS Endpoint - Provides tool's public keys
Used by Moodle to validate signatures from MediaCMS Used by Moodle to validate signatures from MediaCMS (e.g., Deep Linking responses)
""" """
def get(self, request): def get(self, request):
"""Return tool's public JWK Set""" """Return tool's public JWK Set"""
# For now, return empty JWKS since we're not signing responses # Return public keys for signature validation
# In the future, we can generate and store keys for signing deep linking responses jwks = get_jwks()
jwks = {"keys": []}
return JsonResponse(jwks, content_type='application/json') return JsonResponse(jwks, content_type='application/json')
@@ -390,7 +379,7 @@ class ManualSyncView(APIView):
"""Manually trigger NRPS sync""" """Manually trigger NRPS sync"""
try: try:
# Get platform # Get platform
platform = get_object_or_404(LTIPlatform, id=platform_id, active=True) platform = get_object_or_404(LTIPlatform, id=platform_id)
# Find resource link by context # Find resource link by context
resource_link = LTIResourceLink.objects.filter(platform=platform, context_id=context_id).first() resource_link = LTIResourceLink.objects.filter(platform=platform, context_id=context_id).first()