""" Quiz Security Manager - Handles proctoring and malpractice detection """ import uuid import os import json from datetime import datetime from typing import Dict, List from flask import current_app, request from src.models import DatabaseContextManager from src.models.models import ( QuizSecurityLog, QuizScreenshot, QuizMalpracticeReport, QuizAttempt, ModuleQuiz, User, Course, CourseModule, CourseProgress ) from src.utils import custom_response, send_email_with_attachment class QuizSecurityManager: """Manages quiz security and proctoring features""" def log_security_event(self, payload: Dict) -> Dict: """Log a security event during quiz attempt""" try: with DatabaseContextManager() as ctx: attempt_id = payload.get('attempt_id') student_id = payload.get('student_id') quiz_id = payload.get('quiz_id') event_type = payload.get('event_type') # tab_switch, window_blur, copy_paste, etc. if not all([attempt_id, student_id, quiz_id, event_type]): return custom_response( success=False, data="Missing required fields: attempt_id, student_id, quiz_id, event_type", status_code=400 ) # Determine severity based on event type severity_map = { 'tab_switch': 'high', 'window_blur': 'high', 'copy_paste': 'critical', 'right_click': 'medium', 'fullscreen_exit': 'high', 'devtools_open': 'critical', 'screenshot_interval': 'low', 'page_reload': 'medium' } severity = severity_map.get(event_type, 'low') # Count existing violations for this attempt violation_count = ctx.session.query(QuizSecurityLog).filter( QuizSecurityLog.attempt_id == attempt_id, QuizSecurityLog.event_type.in_(['tab_switch', 'window_blur', 'copy_paste', 'fullscreen_exit', 'devtools_open']) ).count() # Determine if warning should be issued warning_issued = False warning_number = 0 action_taken = None if severity in ['high', 'critical']: violation_count += 1 warning_number = violation_count warning_issued = True if violation_count >= 3: # Third violation - send email to tutor action_taken = 'email_sent' self._send_malpractice_email(ctx, attempt_id, student_id, quiz_id, violation_count) else: action_taken = f'warning_{violation_count}' # Create security log entry security_log = QuizSecurityLog( id=str(uuid.uuid4()), attempt_id=attempt_id, student_id=student_id, quiz_id=quiz_id, event_type=event_type, severity=severity, ip_address=payload.get('ip_address') or request.remote_addr, user_agent=payload.get('user_agent') or request.headers.get('User-Agent'), browser_info=payload.get('browser_info'), os_info=payload.get('os_info'), screen_resolution=payload.get('screen_resolution'), event_data=json.dumps(payload.get('event_data', {})), previous_url=payload.get('previous_url'), duration_seconds=payload.get('duration_seconds'), warning_issued=warning_issued, warning_number=warning_number, action_taken=action_taken ) ctx.session.add(security_log) ctx.session.commit() return custom_response( success=True, data={ 'log_id': security_log.id, 'warning_issued': warning_issued, 'warning_number': warning_number, 'action_taken': action_taken, 'total_violations': violation_count }, status_code=201 ) except Exception as e: import traceback error_details = traceback.format_exc() current_app.logger.error(f"Error logging security event: {str(e)}\n{error_details}") return custom_response( success=False, data=f"Error logging security event: {str(e)}", status_code=500 ) def upload_screenshot(self, attempt_id: str, student_id: str, quiz_id: str, file) -> Dict: """Upload and store quiz screenshot""" try: with DatabaseContextManager() as ctx: # Try to find attempt, but don't fail if not found (may be in-progress) attempt = ctx.session.query(QuizAttempt).filter( QuizAttempt.id == attempt_id ).first() if not attempt: # Attempt not in DB yet - try to find any in-progress attempt for this student/quiz current_app.logger.warning(f"⚠️ Attempt {attempt_id} not found - looking for in-progress attempt") attempt = ctx.session.query(QuizAttempt).filter( QuizAttempt.student_id == student_id, QuizAttempt.quiz_id == quiz_id, QuizAttempt.is_completed == False ).order_by(QuizAttempt.started_at.desc()).first() # If still no attempt, create a temporary one if not attempt: current_app.logger.info(f"Creating temporary attempt record for screenshot") attempts_count = ctx.session.query(QuizAttempt).filter( QuizAttempt.quiz_id == quiz_id, QuizAttempt.student_id == student_id ).count() attempt = QuizAttempt( id=attempt_id, # Use the provided ID quiz_id=quiz_id, student_id=student_id, attempt_number=attempts_count + 1, started_at=datetime.utcnow(), is_completed=False, is_submitted=False, max_possible_score=100.0, # Temporary value to satisfy constraint total_score=0.0, percentage_score=0.0 ) ctx.session.add(attempt) ctx.session.flush() # Create screenshots directory server_base = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) screenshot_dir = os.path.join(server_base, 'uploads', 'quiz_screenshots', student_id) os.makedirs(screenshot_dir, exist_ok=True) # Generate unique filename timestamp = datetime.utcnow().strftime('%Y%m%d_%H%M%S') file_extension = 'png' filename = f"quiz_{quiz_id[:8]}_{timestamp}_{str(uuid.uuid4())[:8]}.{file_extension}" file_path = os.path.join(screenshot_dir, filename) # Save file if hasattr(file, 'save'): file.save(file_path) file_size = os.path.getsize(file_path) else: # Handle base64 encoded image import base64 image_data = file.split(',')[1] if ',' in file else file with open(file_path, 'wb') as f: f.write(base64.b64decode(image_data)) file_size = os.path.getsize(file_path) # Create screenshot record screenshot = QuizScreenshot( id=str(uuid.uuid4()), attempt_id=attempt_id, student_id=student_id, quiz_id=quiz_id, file_path=file_path, file_name=filename, file_size=file_size, capture_reason=request.form.get('capture_reason', 'interval'), question_number=request.form.get('question_number', type=int), time_remaining_seconds=request.form.get('time_remaining_seconds', type=int) ) ctx.session.add(screenshot) ctx.session.commit() return custom_response( success=True, data={'screenshot_id': screenshot.id, 'file_name': filename}, status_code=201 ) except Exception as e: import traceback error_details = traceback.format_exc() current_app.logger.error(f"Error uploading screenshot: {str(e)}\n{error_details}") return custom_response( success=False, data=f"Error uploading screenshot: {str(e)}", status_code=500 ) def get_violation_count(self, attempt_id: str) -> Dict: """Get current violation count for an attempt""" try: with DatabaseContextManager() as ctx: violation_count = ctx.session.query(QuizSecurityLog).filter( QuizSecurityLog.attempt_id == attempt_id, QuizSecurityLog.severity.in_(['high', 'critical']) ).count() return custom_response( success=True, data={'violation_count': violation_count}, status_code=200 ) except Exception as e: current_app.logger.error(f"Error getting violation count: {str(e)}") return custom_response( success=False, data=f"Error getting violation count: {str(e)}", status_code=500 ) def _send_malpractice_email(self, ctx, attempt_id: str, student_id: str, quiz_id: str, violation_count: int): """Send email to tutor about student malpractice (only once per attempt)""" try: # Check if email already sent for this attempt existing_report = ctx.session.query(QuizMalpracticeReport).filter( QuizMalpracticeReport.attempt_id == attempt_id, QuizMalpracticeReport.email_sent == True ).first() if existing_report: current_app.logger.info(f"📧 Email already sent for attempt {attempt_id} - skipping") return # Get attempt, student, quiz, and course details attempt = ctx.session.query(QuizAttempt).filter( QuizAttempt.id == attempt_id ).first() student = ctx.session.query(User).filter( User.id == student_id ).first() quiz = ctx.session.query(ModuleQuiz).filter( ModuleQuiz.id == quiz_id ).first() if not attempt: current_app.logger.warning(f"⚠️ Attempt not found in DB yet: {attempt_id} - skipping email") return if not student: current_app.logger.error(f"❌ Student not found: {student_id}") return if not quiz: current_app.logger.error(f"❌ Quiz not found: {quiz_id}") return module = ctx.session.query(CourseModule).filter( CourseModule.id == quiz.module_id ).first() if not module: current_app.logger.error("Module not found for quiz") return course = ctx.session.query(Course).filter( Course.id == module.course_id ).first() if not course: current_app.logger.error("Course not found for module") return # Get tutor (quiz creator) tutor = ctx.session.query(User).filter( User.id == quiz.created_by ).first() if not tutor or not tutor.email: current_app.logger.error("Tutor not found or has no email") return # Get all security logs for this attempt security_logs = ctx.session.query(QuizSecurityLog).filter( QuizSecurityLog.attempt_id == attempt_id, QuizSecurityLog.severity.in_(['high', 'critical']) ).all() # Get all screenshots for this attempt screenshots = ctx.session.query(QuizScreenshot).filter( QuizScreenshot.attempt_id == attempt_id ).all() current_app.logger.info(f"📸 Found {len(screenshots)} screenshots for attempt {attempt_id}") # Create violation summary violation_summary = "\n".join([ f"- {log.event_type} at {log.event_timestamp.strftime('%H:%M:%S')} (Severity: {log.severity})" for log in security_logs ]) # Create screenshots summary for email screenshots_summary = f"\n\nScreenshots Captured: {len(screenshots)}" if screenshots: screenshots_summary += "\n" + "\n".join([ f" - {s.file_name} (Question {s.question_number or 'N/A'}, {s.captured_at.strftime('%H:%M:%S')})" for s in screenshots[:10] # Show first 10 ]) if len(screenshots) > 10: screenshots_summary += f"\n ... and {len(screenshots) - 10} more" # Create email content email_subject = f"⚠️ Quiz Malpractice Alert - {student.first_name} {student.last_name} - {quiz.quiz_title}" email_body = f"""

⚠️ Quiz Malpractice Alert

Dear {tutor.first_name} {tutor.last_name},

We detected multiple suspicious activities during a quiz attempt. Here are the details:

Student Information

Name: {student.first_name} {student.last_name}

Email: {student.email}

Student ID: {getattr(student, 'student_id', 'N/A')}

Quiz Details

Course: {course.title} ({course.code})

Quiz: {quiz.quiz_title}

Attempt: #{attempt.attempt_number}

Started: {attempt.started_at.strftime('%Y-%m-%d %H:%M:%S')}

Violations Detected ({violation_count})

{violation_summary}

Screenshot Evidence ({len(screenshots)} captured)

Screenshots have been attached to this email for your review. They show the student's screen at various points during the quiz. 

{screenshots_summary}

⚠️ Action Required: Please review the quiz attempt and take appropriate action. The system has automatically flagged this attempt for your review.

This is an automated message from the Quiz Security System.
Timestamp: {datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S UTC')}

""" # Create malpractice report report = QuizMalpracticeReport( id=str(uuid.uuid4()), attempt_id=attempt_id, student_id=student_id, quiz_id=quiz_id, course_id=course.id, tutor_id=quiz.created_by, severity='high' if violation_count >= 3 else 'medium', total_violations=violation_count, violation_summary=violation_summary, email_recipient=tutor.email, email_subject=email_subject, email_body=email_body ) ctx.session.add(report) # Send email with screenshot attachments try: # Send email with first screenshot as attachment (send_email_with_attachment only supports one) attachment_path = None if screenshots: for screenshot in screenshots: if os.path.exists(screenshot.file_path): attachment_path = screenshot.file_path current_app.logger.info(f"📎 Attaching screenshot: {screenshot.file_name}") break # Send email with attachment success, result = send_email_with_attachment( sender_email="kisiwa@mutabletech.co.ke", sender_password=current_app.config.get('MAIL_PASSWORD', ''), receiver_email=tutor.email, subject=email_subject, message=email_body, attachment_path=attachment_path ) if success: report.email_sent = True report.email_sent_at = datetime.utcnow() current_app.logger.info(f"✅ Malpractice email sent to {tutor.email} with screenshot attachment") else: current_app.logger.error(f"❌ Failed to send email: {result}") except Exception as e: current_app.logger.error(f"❌ Failed to send email: {str(e)}") import traceback current_app.logger.error(traceback.format_exc()) # Continue even if email fails ctx.session.commit() except Exception as e: current_app.logger.error(f"Error in _send_malpractice_email: {str(e)}") def create_malpractice_report(self, payload: Dict) -> Dict: """Create a formal malpractice report""" try: with DatabaseContextManager() as ctx: attempt_id = payload.get('attempt_id') # Get attempt details attempt = ctx.session.query(QuizAttempt).filter( QuizAttempt.id == attempt_id ).first() if not attempt: return custom_response( success=False, data="Quiz attempt not found", status_code=404 ) # Check if report already exists existing_report = ctx.session.query(QuizMalpracticeReport).filter( QuizMalpracticeReport.attempt_id == attempt_id ).first() if existing_report: return custom_response( success=True, data={'report_id': existing_report.id, 'message': 'Report already exists'}, status_code=200 ) # Get violation count violation_count = ctx.session.query(QuizSecurityLog).filter( QuizSecurityLog.attempt_id == attempt_id, QuizSecurityLog.severity.in_(['high', 'critical']) ).count() # Get quiz and course quiz = ctx.session.query(ModuleQuiz).filter( ModuleQuiz.id == attempt.quiz_id ).first() from src.models.models import CourseModule module = ctx.session.query(CourseModule).filter( CourseModule.id == quiz.module_id ).first() # Create and send report self._send_malpractice_email(ctx, attempt_id, attempt.student_id, attempt.quiz_id, violation_count) ctx.session.commit() return custom_response( success=True, data={'message': 'Malpractice report created and email sent'}, status_code=201 ) except Exception as e: import traceback error_details = traceback.format_exc() current_app.logger.error(f"Error creating malpractice report: {str(e)}\n{error_details}") return custom_response( success=False, data=f"Error creating malpractice report: {str(e)}", status_code=500 ) def upload_webcam_video(self, attempt_id: str, student_id: str, quiz_id: str, video_file) -> Dict: """Upload webcam recording chunk or full video""" try: with DatabaseContextManager() as ctx: # Try to find attempt attempt = ctx.session.query(QuizAttempt).filter( QuizAttempt.id == attempt_id ).first() if not attempt: # Try to find any in-progress attempt current_app.logger.warning(f"⚠️ Webcam: Attempt {attempt_id} not found - looking for in-progress attempt") attempt = ctx.session.query(QuizAttempt).filter( QuizAttempt.student_id == student_id, QuizAttempt.quiz_id == quiz_id, QuizAttempt.is_completed == False ).order_by(QuizAttempt.started_at.desc()).first() # Create temporary attempt if needed if not attempt: current_app.logger.info(f"Creating temporary attempt for webcam recording") attempts_count = ctx.session.query(QuizAttempt).filter( QuizAttempt.quiz_id == quiz_id, QuizAttempt.student_id == student_id ).count() attempt = QuizAttempt( id=attempt_id, quiz_id=quiz_id, student_id=student_id, attempt_number=attempts_count + 1, started_at=datetime.utcnow(), is_completed=False, is_submitted=False, max_possible_score=100.0, # Temporary value to satisfy constraint total_score=0.0, percentage_score=0.0 ) ctx.session.add(attempt) ctx.session.flush() # Create webcam videos directory server_base = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) webcam_dir = os.path.join(server_base, 'uploads', 'quiz_webcam', student_id) os.makedirs(webcam_dir, exist_ok=True) # Check if recording already exists from src.models.models import QuizWebcamRecording recording = ctx.session.query(QuizWebcamRecording).filter( QuizWebcamRecording.attempt_id == attempt_id ).first() if not recording: # Create new recording record timestamp = datetime.utcnow().strftime('%Y%m%d_%H%M%S') filename = f"webcam_{quiz_id[:8]}_{timestamp}.webm" file_path = os.path.join(webcam_dir, filename) recording = QuizWebcamRecording( id=str(uuid.uuid4()), attempt_id=attempt_id, student_id=student_id, quiz_id=quiz_id, file_path=file_path, file_name=filename, recording_status='active' ) ctx.session.add(recording) ctx.session.flush() else: file_path = recording.file_path # Save or append video chunk if hasattr(video_file, 'save'): # File upload mode = 'ab' if os.path.exists(file_path) else 'wb' # Append if exists with open(file_path, mode) as f: video_file.save(f) else: # Base64 data import base64 video_data = video_file.split(',')[1] if ',' in video_file else video_file mode = 'ab' if os.path.exists(file_path) else 'wb' with open(file_path, mode) as f: f.write(base64.b64decode(video_data)) # Update recording metadata recording.file_size = os.path.getsize(file_path) recording.chunk_count = (recording.chunk_count or 0) + 1 recording.updated_at = datetime.utcnow() ctx.session.commit() return custom_response( success=True, data={ 'recording_id': recording.id, 'file_name': recording.file_name, 'chunk_count': recording.chunk_count }, status_code=201 ) except Exception as e: import traceback error_details = traceback.format_exc() current_app.logger.error(f"Error uploading webcam video: {str(e)}\n{error_details}") return custom_response( success=False, data=f"Error uploading webcam video: {str(e)}", status_code=500 ) def stop_webcam_recording(self, attempt_id: str) -> Dict: """Mark webcam recording as completed""" try: with DatabaseContextManager() as ctx: from src.models.models import QuizWebcamRecording recording = ctx.session.query(QuizWebcamRecording).filter( QuizWebcamRecording.attempt_id == attempt_id ).first() if not recording: return custom_response( success=False, data="Recording not found", status_code=404 ) recording.recording_status = 'completed' recording.stopped_at = datetime.utcnow() recording.duration_seconds = int((recording.stopped_at - recording.started_at).total_seconds()) ctx.session.commit() return custom_response( success=True, data={'message': 'Recording stopped successfully'}, status_code=200 ) except Exception as e: current_app.logger.error(f"Error stopping webcam recording: {str(e)}") return custom_response( success=False, data=f"Error stopping recording: {str(e)}", status_code=500 ) def get_attempt_security_summary(self, attempt_id: str) -> Dict: """Get security summary for a quiz attempt""" try: with DatabaseContextManager() as ctx: # Get all security logs logs = ctx.session.query(QuizSecurityLog).filter( QuizSecurityLog.attempt_id == attempt_id ).all() # Get screenshots screenshots = ctx.session.query(QuizScreenshot).filter( QuizScreenshot.attempt_id == attempt_id ).all() # Get malpractice reports reports = ctx.session.query(QuizMalpracticeReport).filter( QuizMalpracticeReport.attempt_id == attempt_id ).all() # Count violations by type violation_counts = {} for log in logs: violation_counts[log.event_type] = violation_counts.get(log.event_type, 0) + 1 high_severity_count = sum(1 for log in logs if log.severity in ['high', 'critical']) return custom_response( success=True, data={ 'total_events': len(logs), 'total_screenshots': len(screenshots), 'total_reports': len(reports), 'high_severity_violations': high_severity_count, 'violation_counts': violation_counts, 'warnings_issued': high_severity_count }, status_code=200 ) except Exception as e: current_app.logger.error(f"Error getting security summary: {str(e)}") return custom_response( success=False, data=f"Error getting security summary: {str(e)}", status_code=500 ) def upload_exam_evidence(self, attempt_id: str, student_id: str, quiz_id: str, screenshot_file, evidence_type: str = 'grade_result', evidence_data: str = None, portfolio_tag: str = 'exam_evidence', grade: str = None, max_grade: str = None, percentage: str = None) -> Dict: """Upload exam evidence (grade screenshot with portfolio metadata)""" try: current_app.logger.info(f"📸 Uploading exam evidence for attempt {attempt_id}") with DatabaseContextManager() as ctx: # Find the attempt attempt = ctx.session.query(QuizAttempt).filter( QuizAttempt.id == attempt_id ).first() if not attempt: current_app.logger.error(f"⚠️ Attempt {attempt_id} not found") return custom_response( success=False, data="Attempt not found", status_code=404 ) # Create upload directory if it doesn't exist evidence_dir = os.path.join(current_app.config.get('UPLOAD_FOLDER', 'uploads'), 'quiz_evidence') os.makedirs(evidence_dir, exist_ok=True) # Save screenshot file filename = f"quiz_evidence_{attempt_id}_{int(datetime.utcnow().timestamp())}.png" filepath = os.path.join(evidence_dir, filename) screenshot_file.save(filepath) current_app.logger.info(f"✅ Screenshot saved: {filepath}") # Create evidence record (store in quiz_screenshots table) screenshot_record = QuizScreenshot( id=str(uuid.uuid4()), attempt_id=attempt.id, screenshot_path=filepath, capture_reason=evidence_type, capture_timestamp=datetime.utcnow(), metadata=evidence_data if evidence_data else json.dumps({ 'grade': grade, 'max_grade': max_grade, 'percentage': percentage, 'portfolio_tag': portfolio_tag }) ) ctx.session.add(screenshot_record) ctx.session.commit() current_app.logger.info(f"✅ Evidence record created with ID: {screenshot_record.id}") return custom_response( success=True, data={ 'evidence_id': screenshot_record.id, 'filepath': filepath, 'attempt_id': attempt.id, 'evidence_type': evidence_type, 'grade': grade, 'max_grade': max_grade, 'percentage': percentage, 'portfolio_tag': portfolio_tag, 'message': 'Exam evidence uploaded successfully' }, status_code=200 ) except Exception as e: import traceback current_app.logger.error(f"❌ Error uploading exam evidence: {str(e)}") current_app.logger.error(traceback.format_exc()) return custom_response( success=False, data=f"Error uploading evidence: {str(e)}", status_code=500 )