check_task_attempt_access

Author: PeeachPie

app/api/task_attempts.py

@@ -3,7 +3,7 @@
 from bson import ObjectId
 from flask import session, request, Blueprint
 
-from app.check_access import check_access
+from app.check_access import check_access, check_task_attempt_access
 from app.lti_session_passback.auth_checkers import check_auth, is_admin
 from app.mongo_odm import TaskAttemptsDBManager, TasksDBManager
 from app.utils import check_arguments_are_convertible_to_object_id, check_argument_is_convertible_to_object_id
@@ -91,7 +91,7 @@ def get_task_attempt(task_attempt_id) -> tuple[dict, int]:
         a dictionary with an explanation and 404 HTTP return code if a task attempt was not found, or
         an empty dictionary with 404 HTTP return code if access was denied.
     """
-    if not check_access({'_id': ObjectId(task_attempt_id)}):
+    if not check_task_attempt_access(task_attempt_id):
         return {}, 404
 
     task_attempt_db = TaskAttemptsDBManager().get_task_attempt(task_attempt_id)

app/api/trainings.py

@@ -10,7 +10,7 @@
 
 from app.audio import Audio
 from app.check_access import check_access
-from app.lti_session_passback.auth_checkers import is_admin, check_auth, check_admin
+from app.lti_session_passback.auth_checkers import is_admin, check_auth, check_admin, is_logged_in
 from app.mongo_models import Trainings
 from app.mongo_odm import TrainingsDBManager, TaskAttemptsDBManager, TasksDBManager, DBManager
 from app.filters import GetAllTrainingsFilterManager
@@ -439,8 +439,7 @@ def get_all_trainings() -> (dict, int):
 
     print(number_page, count_items)
 
-    authorized = check_auth() is not None
-    if not (check_admin() or (authorized and [session.get('session_id')] == username)):
+    if not (check_admin() or (is_logged_in() and [session.get('session_id')] == username)):
         return {}, 404
 
     trainings = GetAllTrainingsFilterManager().query_with_filters(filters, number_page, count_items)

app/check_access.py

@@ -1,6 +1,6 @@
 from flask import session
 
-from app.mongo_odm import SessionsDBManager, TrainingsDBManager
+from app.mongo_odm import SessionsDBManager, TrainingsDBManager, TaskAttemptsDBManager
 from app.utils import is_testing_active
 
 
@@ -25,3 +25,25 @@ def _check_access(filters: dict) -> bool:
 
 def _check_access_testing(filters: dict) -> bool:
     return True
+
+def _check_task_attempt_access(task_attempt_id: str) -> bool:
+    username = session.get('session_id', default=None)
+    consumer_key = session.get('consumer_key', default=None)
+    user_session = SessionsDBManager().get_session(username, consumer_key)
+
+    if not user_session:
+        return False
+    if user_session.is_admin:
+        return True
+
+    task_attempt = TaskAttemptsDBManager().get_task_attempt(task_attempt_id)
+    return task_attempt.username == username
+
+def check_task_attempt_access(task_attempt_id: str) -> bool:
+    if not is_testing_active():
+        return _check_task_attempt_access(task_attempt_id)
+    else:
+        return _check_task_attempt_access_testing(task_attempt_id)
+    
+def _check_task_attempt_access_testing(task_attempt_id: str) -> bool:
+    return True

app/routes/task_attempts.py

@@ -5,13 +5,19 @@
 from app.localisation import *
 
 from app.api.task_attempts import get_task_attempt
-from app.check_access import check_access
+from app.check_access import check_access, check_task_attempt_access
 from app.lti_session_passback.auth_checkers import check_admin, check_auth
 from app.utils import check_arguments_are_convertible_to_object_id
 
 routes_task_attempts = Blueprint('routes_task_attempts', __name__)
 logger = get_root_logger()
 
+def ch(tr_id):
+    print(tr_id)
+    print(check_access({'_id': ObjectId(tr_id)}), sep=" ")
+    print()
+    return check_access({'_id': ObjectId(tr_id)})
+
 @check_arguments_are_convertible_to_object_id
 @routes_task_attempts.route('/task_attempts/<task_attempt_id>/', methods=['GET'])
 def view_task_attempt(task_attempt_id: str):
@@ -22,11 +28,9 @@ def view_task_attempt(task_attempt_id: str):
     :return: Page or an empty dictionary with 404 HTTP code if access was denied.
     """
 
-    if not check_access({'_id': ObjectId(task_attempt_id)}):
+    if not check_task_attempt_access(task_attempt_id):
         return {}, 404
 
-    # Нужна ли проверка авторизации?
-    
     task_attempt, task_attempt_status_code = get_task_attempt(task_attempt_id)
 
     if task_attempt.get('message') != 'OK':

app/routes/trainings.py

@@ -12,7 +12,7 @@
 from app.check_access import check_access
 from app.criteria_pack import CriteriaPackFactory
 from app.feedback_evaluator import FeedbackEvaluatorFactory
-from app.lti_session_passback.auth_checkers import check_admin, check_auth
+from app.lti_session_passback.auth_checkers import check_admin, check_auth, is_logged_in
 from app.mongo_odm import CriterionPackDBManager, TasksDBManager, TaskAttemptsDBManager
 from app.status import TrainingStatus, AudioStatus, PresentationStatus
 from app.utils import check_arguments_are_convertible_to_object_id
@@ -141,8 +141,7 @@ def view_all_trainings():
     except:
         pass
 
-    authorized = check_auth() is not None
-    if not (check_admin() or (authorized and session.get('session_id') == username)):
+    if not (check_admin() or (is_logged_in() and session.get('session_id') == username)):
         return {}, 404
 
     raw_filters = request.args.getlist('f')

app/static/js/task_attempts.js

@@ -42,7 +42,7 @@ function createTableRowElement(trainingInfo, trainingId) {
     tableRowIdElement.innerHTML = `<a href="/trainings/statistics/${trainingId}/">${trainingId}</a>`;
 
     const tableRowScoreElement = document.createElement("td");
-    tableRowScoreElement.innerHTML = trainingInfo.score.toFixed(2) || "none";
+    tableRowScoreElement.innerHTML = trainingInfo.score ? trainingInfo.score.toFixed(2) : "none";
 
     const tableRowStatusElement = document.createElement("td");
     tableRowStatusElement.innerHTML = trainingInfo.passedBackStatus || "none";