from flask import Blueprint, request, jsonify, session from functools import wraps import subprocess import tempfile import os import time import uuid from datetime import datetime import docker import psutil from pymongo import MongoClient from activity_logger import log_user_activity, resolve_user_identity bp = Blueprint('coding', __name__) # MongoDB connection mongo_uri = os.getenv('MONGODB_URI', 'mongodb://localhost:27017/') client = MongoClient(mongo_uri) db = client.openlearnx def secure_execution_required(f): @wraps(f) def decorated_function(*args, **kwargs): # Check if user is in secure coding mode if not session.get('secure_coding_mode'): return jsonify({"error": "Secure coding mode required"}), 403 return f(*args, **kwargs) return decorated_function @bp.route("/start-session", methods=["POST"]) def start_coding_session(): """Start a secure coding session""" try: data = request.json course_id = data.get('course_id') lesson_id = data.get('lesson_id') session_id = str(uuid.uuid4()) session['coding_session_id'] = session_id session['secure_coding_mode'] = True session['start_time'] = datetime.now().isoformat() session['course_id'] = course_id session['lesson_id'] = lesson_id identity = resolve_user_identity(request, db) log_user_activity( db, identity.get("user_id"), "coding", "Started coding session", "Entered secure coding session", { "session_id": session_id, "course_id": course_id, "lesson_id": lesson_id, }, ) return jsonify({ "success": True, "session_id": session_id, "message": "Secure coding session started", "restrictions": { "copy_paste_disabled": True, "browser_locked": True, "extensions_blocked": True, "virtual_detection": True } }) except Exception as e: return jsonify({"error": str(e)}), 500 @bp.route("/execute", methods=["POST"]) @secure_execution_required def execute_code(): """Execute code securely in isolated environment""" try: data = request.json code = data.get('code') language = data.get('language', 'python') test_cases = data.get('test_cases', []) if not code: return jsonify({"error": "No code provided"}), 400 # Log coding attempt log_coding_attempt(session['coding_session_id'], code, language) # Execute code in secure container result = execute_in_container(code, language, test_cases) return jsonify(result) except Exception as e: return jsonify({"error": str(e)}), 500 @bp.route("/submit-test", methods=["POST"]) @secure_execution_required def submit_coding_test(): """Submit coding test for evaluation""" try: data = request.json code = data.get('code') problem_id = data.get('problem_id') # Validate against test cases test_result = validate_test_submission(code, problem_id) # Store submission submission_id = store_submission( session['coding_session_id'], session['course_id'], problem_id, code, test_result ) identity = resolve_user_identity(request, db) resolved_user_id = identity.get("user_id") if resolved_user_id: db.user_submissions.insert_one({ "user_id": resolved_user_id, "session_id": session.get('coding_session_id'), "course_id": session.get('course_id'), "problem_id": problem_id, "score": test_result.get('score', 0), "points_earned": int(test_result.get('score', 0)), "submitted_at": datetime.now(), "status": "submitted", }) log_user_activity( db, resolved_user_id, "coding", "Submitted coding solution", f"Submitted coding test for problem '{problem_id}'", { "submission_id": submission_id, "problem_id": problem_id, "score": test_result.get('score', 0), "passed": test_result.get('passed', 0), "total": test_result.get('total', 0), }, points_earned=int(test_result.get('score', 0)), ) return jsonify({ "success": True, "submission_id": submission_id, "score": test_result['score'], "passed_tests": test_result['passed'], "total_tests": test_result['total'], "feedback": test_result['feedback'] }) except Exception as e: return jsonify({"error": str(e)}), 500 def execute_in_container(code, language, test_cases): """Execute code in secure Docker container""" try: client = docker.from_env() # Language-specific container configuration containers = { 'python': 'python:3.9-alpine', 'java': 'openjdk:11-alpine', 'javascript': 'node:16-alpine' } if language not in containers: return {"error": "Unsupported language"} # Create temporary file with tempfile.NamedTemporaryFile(mode='w', suffix=f'.{get_file_extension(language)}', delete=False) as f: f.write(code) temp_file = f.name try: # Run container with security restrictions container = client.containers.run( containers[language], command=get_run_command(language, temp_file), volumes={os.path.dirname(temp_file): {'bind': '/app', 'mode': 'ro'}}, working_dir='/app', mem_limit='128m', cpu_period=100000, cpu_quota=50000, # 50% CPU limit network_mode='none', # No network access remove=True, timeout=10, # 10 second timeout detach=False ) output = container.decode('utf-8') # Run test cases if provided test_results = [] if test_cases: for test in test_cases: test_result = run_test_case(code, language, test) test_results.append(test_result) return { "success": True, "output": output, "test_results": test_results, "execution_time": "< 10s" } finally: os.unlink(temp_file) except docker.errors.ContainerError as e: return {"error": f"Runtime error: {e}"} except docker.errors.ImageNotFound: return {"error": "Language runtime not available"} except Exception as e: return {"error": f"Execution failed: {str(e)}"} def get_file_extension(language): extensions = { 'python': 'py', 'java': 'java', 'javascript': 'js' } return extensions.get(language, 'txt') def get_run_command(language, filename): commands = { 'python': f'python /app/{os.path.basename(filename)}', 'java': f'javac /app/{os.path.basename(filename)} && java -cp /app {os.path.splitext(os.path.basename(filename))[0]}', 'javascript': f'node /app/{os.path.basename(filename)}' } return commands.get(language) def log_coding_attempt(session_id, code, language): """Log all coding attempts for monitoring""" db.coding_logs.insert_one({ "session_id": session_id, "code": code, "language": language, "timestamp": datetime.now(), "ip_address": request.remote_addr, "user_agent": request.headers.get('User-Agent') }) def validate_test_submission(code, problem_id): """Validate code against predefined test cases""" # Load test cases for the problem test_cases = get_problem_test_cases(problem_id) passed = 0 total = len(test_cases) feedback = [] for i, test_case in enumerate(test_cases): result = run_test_case(code, 'python', test_case) if result['passed']: passed += 1 feedback.append(f"Test {i+1}: ✅ Passed") else: feedback.append(f"Test {i+1}: ❌ Failed - {result['error']}") score = (passed / total) * 100 return { "score": score, "passed": passed, "total": total, "feedback": feedback } def get_problem_test_cases(problem_id): """Get test cases for a specific problem""" # This would load from your database test_cases_db = { "python-basics-1": [ {"input": "hello", "expected_output": "HELLO"}, {"input": "world", "expected_output": "WORLD"} ], "java-oop-1": [ {"input": "5", "expected_output": "25"}, {"input": "10", "expected_output": "100"} ] } return test_cases_db.get(problem_id, [])