"""Blueprint Flask del módulo de evaluación RA3 CDT1. Adaptador web (capa infraestructura): valida entradas con allow-list y límites de longitud, gestiona sesiones por token, registra incidentes de trampa y delega toda la lógica de calificación al dominio puro (ra3_grader). Rutas: GET /ra3-cdt1 Página de la evaluación. POST /iniciar-ra3 Crea sesión (token) tras identificar al estudiante. POST /incidente-ra3 Registra un incidente anti-trampa (-5 pts c/u). POST /guardar-ra3 Califica server-side y guarda el CSV final. """ import csv import os import secrets import threading from datetime import datetime from flask import Blueprint, jsonify, render_template, request import ra3_grader bp = Blueprint("ra3", __name__, template_folder="templates") DIR_ACTUAL = os.path.dirname(os.path.abspath(__file__)) CSV_RESULTADOS = os.path.join(DIR_ACTUAL, "resultados_ra3_cdt1.csv") CSV_INCIDENCIAS = os.path.join(DIR_ACTUAL, "incidencias_ra3_cdt1.csv") ENCABEZADO_RESULTADOS = ["No", "Apellidos", "Nombres", "Carrera", "Nota"] ENCABEZADO_INCIDENCIAS = ["fecha", "token", "estudiante", "tipo"] CARRERAS_PERMITIDAS = ("Mecánica", "Electrónica") TIPOS_INCIDENTE_VALIDOS = ( "pantalla_completa", "pestana_oculta", "ventana_ajena", "copiar", "pegar", "atajo_teclado", "devtools", "clic_derecho", ) MAX_LONGITUD_TEXTO_ABIERTO = 500 MAX_LONGITUD_DATO = 60 _sesiones = {} _bloqueo = threading.Lock() def _texto_valido(valor, maximo): """Devuelve texto saneado o None si viola tipo/longitud.""" if not isinstance(valor, str): return None limpio = valor.strip() if not limpio or len(limpio) > maximo: return None return limpio def _obtener_sesion(token): with _bloqueo: return _sesiones.get(token) if isinstance(token, str) else None # --------------------------------------------------------------------------- # Escritura segura de CSV (UTF-8 BOM + protección contra inyección de fórmulas) # --------------------------------------------------------------------------- def _escapar_celda(valor): """Neutraliza inyección de fórmulas CSV (=,+,-,@ inicial) y saltos.""" texto = str(valor) texto = texto.replace("\n", " ").replace("\r", " ") if texto.startswith(("=", "+", "-", "@")): texto = "'" + texto return texto def _agregar_fila(ruta_csv, encabezado, fila): existe = os.path.exists(ruta_csv) with open(ruta_csv, "a", newline="", encoding="utf-8-sig") as archivo: escritor = csv.writer(archivo) if not existe: escritor.writerow(encabezado) escritor.writerow([_escapar_celda(celda) for celda in fila]) def _siguiente_correlativo(): """No = número de filas de datos ya presentes (encabezado excluido).""" if not os.path.exists(CSV_RESULTADOS): return 1 with open(CSV_RESULTADOS, encoding="utf-8-sig") as archivo: filas = [linea for linea in archivo if linea.strip()] return max(1, len(filas)) # con encabezado presente, len(filas)-1+1 = len(filas) # --------------------------------------------------------------------------- # Rutas # --------------------------------------------------------------------------- @bp.route("/ra3-cdt1") def pagina_ra3(): return render_template("ra3-cdt1.html") @bp.route("/iniciar-ra3", methods=["POST"]) def iniciar_ra3(): datos = request.get_json(silent=True) or {} apellidos = _texto_valido(datos.get("apellidos"), 100) nombres = _texto_valido(datos.get("nombres"), 100) carrera = datos.get("carrera") if not apellidos or not nombres: return jsonify({"ok": False, "error": "Identificación incompleta."}), 400 if carrera not in CARRERAS_PERMITIDAS: return jsonify({"ok": False, "error": "Carrera no permitida."}), 400 token = secrets.token_hex(16) with _bloqueo: _sesiones[token] = { "apellidos": apellidos, "nombres": nombres, "carrera": carrera, "incidentes": [], "enviado": False, "creado": datetime.now(), } return jsonify({"ok": True, "token": token}) @bp.route("/incidente-ra3", methods=["POST"]) def registrar_incidente(): datos = request.get_json(silent=True) or {} sesion = _obtener_sesion(datos.get("token")) if sesion is None: return jsonify({"ok": False, "error": "Sesión inválida."}), 400 tipo = datos.get("tipo") if tipo not in TIPOS_INCIDENTE_VALIDOS: return jsonify({"ok": False, "error": "Tipo de incidente desconocido."}), 400 with _bloqueo: sesion["incidentes"].append(tipo) total = len(sesion["incidentes"]) estudiante = f"{sesion['apellidos']}, {sesion['nombres']}" _agregar_fila( CSV_INCIDENCIAS, ENCABEZADO_INCIDENCIAS, [datetime.now().strftime("%Y-%m-%d %H:%M:%S"), str(datos.get("token"))[:8], estudiante, tipo], ) return jsonify({"ok": True, "incidentes": total, "descuento": 5 * total}) # --------------------------------------------------------------------------- # Validación de respuestas (defense in depth: allow-list en el boundary) # --------------------------------------------------------------------------- def _dict_limpio(entrada, claves_permitidas, maximo): """Sanea un dict tomando solo claves permitidas con textos acotados.""" if not isinstance(entrada, dict): return {} limpio = {} for clave in claves_permitidas: valor = entrada.get(clave) if isinstance(valor, str) and len(valor) <= maximo: limpio[clave] = valor return limpio def _sanear_pasos(entrada): """Valida la lista ordenada de pasos contra el allow-list del dominio.""" if not isinstance(entrada, list): return [] return [paso for paso in entrada if paso in ra3_grader.PASOS_CORRECTOS] def _sanear_respuestas(datos): """Construye el payload confiable que se entrega al dominio.""" r = datos.get("respuestas") or {} metas_entrada = datos.get("metas") or {} metas_limpias = {} if isinstance(metas_entrada, dict): for clave, meta in metas_entrada.items(): if isinstance(clave, str) and clave.startswith("s3_") \ and isinstance(meta, dict) and len(meta) <= 8: metas_limpias[clave] = { "tiempo_s": str(meta.get("tiempo_s", ""))[:10], "teclas": str(meta.get("teclas", ""))[:7], "correcciones": str(meta.get("correcciones", ""))[:7], } letras = {"0": None, "1": None, "2": None, "3": None, "4": None} serie1 = {k: v for k, v in _dict_limpio(r.get("serie1"), letras, 2).items() if v in ("V", "F")} serie2 = {k: v for k, v in _dict_limpio(r.get("serie2"), letras, 2).items() if v in ("A", "B", "C", "D")} serie3 = _dict_limpio(r.get("serie3"), letras, MAX_LONGITUD_TEXTO_ABIERTO) serie4 = _dict_limpio( r.get("serie4"), {"0", "1", "2", "3", "porc_h", "porc_o", "5", "6"}, MAX_LONGITUD_DATO, ) serie5 = _dict_limpio(r.get("serie5"), {"v1", "v2", "v3"}, MAX_LONGITUD_DATO) serie5["pasos"] = _sanear_pasos((r.get("serie5") or {}).get("pasos")) return { "serie1": serie1, "serie2": serie2, "serie3": serie3, "serie4": serie4, "serie5": serie5, }, metas_limpias @bp.route("/guardar-ra3", methods=["POST"]) def guardar_ra3(): datos = request.get_json(silent=True) or {} sesion = _obtener_sesion(datos.get("token")) if sesion is None: return jsonify({"ok": False, "error": "Sesión inválida."}), 400 with _bloqueo: if sesion["enviado"]: return jsonify({"ok": False, "error": "Esta evaluación ya fue enviada."}), 409 respuestas, metas = _sanear_respuestas(datos) with _bloqueo: n_incidentes = len(sesion["incidentes"]) resultado = ra3_grader.calificar_examen(respuestas, metas, n_incidentes) with _bloqueo: numero = _siguiente_correlativo() sesion["enviado"] = True _agregar_fila( CSV_RESULTADOS, ENCABEZADO_RESULTADOS, [numero, sesion["apellidos"], sesion["nombres"], sesion["carrera"], resultado["nota_final"]], ) return jsonify({"ok": True, "resultado": resultado})