Source code for mdvtools.auth.project_auth

from flask import jsonify, session, current_app
from datetime import datetime, timedelta
from mdvtools.auth.authutils import user_project_cache
from mdvtools.dbutils.dbservice import ProjectService
from mdvtools.logging_config import get_logger

[docs] logger = get_logger(__name__)
[docs] TIMESTAMP_UPDATE_INTERVAL = timedelta(hours=1)
[docs] def project_pre_dispatch_checks(project_id: str, options: dict): """ Performs checks before dispatching a request to a project route. This function is intended to be used as a 'before_request' hook. - Checks if the project exists. - Updates the project's 'accessed_timestamp'. - Performs authentication and authorization checks based on the route's options. :param project_id: The ID of the project being accessed. :param options: The options dictionary for the matched route (e.g., {'access_level': 'editable'}). :return: A Flask Response object to short-circuit the request in case of an error, or None to continue. """ project = ProjectService.get_project_by_id(project_id) if not project: logger.error(f"In pre-dispatch check: Project with ID {project_id} not found.") return jsonify({"error": f"Project with ID {project_id} not found"}), 404 # Update the accessed timestamp only if the last update was more than TIMESTAMP_UPDATE_INTERVAL ago if not project.accessed_timestamp or (datetime.now() - project.accessed_timestamp > TIMESTAMP_UPDATE_INTERVAL): try: ProjectService.set_project_accessed_timestamp(project_id) except Exception as e: logger.exception(f"Error updating accessed timestamp for project {project_id}: {e}") return jsonify({"error": "Failed to update project accessed timestamp"}), 500 # If auth is not enabled for the app, we can skip the user-specific checks if not current_app.config.get("ENABLE_AUTH", False): if options.get('access_level') == 'editable' and ( not project.access_level or project.access_level != 'editable' ): return jsonify({"error": "This project is read-only and cannot be modified."}), 403 return None # No more checks needed # --- Auth-enabled checks below --- user = session.get('user') user_id = user.get("id") if user else None if not user_id: return jsonify({"error": "Authentication required."}), 401 try: project_permissions = user_project_cache.get(user_id, {}).get(int(project_id)) except ValueError as e: # project_id is passed as a string, conversion to int may fail logger.exception(f"Error getting project permissions for user {user_id} and project {project_id}: {e}") return jsonify({"error": "Failed to get project permissions"}), 500 # Check for access level only if specified in options if options.get('access_level') == 'editable': # Check permission from the database model if project.access_level != 'editable': return jsonify({"error": "This project is read-only and cannot be modified."}), 403 # Check user-specific write permissions if not project_permissions or not (project_permissions.get("is_owner") or project_permissions.get("can_write")): return jsonify({"error": "User does not have the required write permissions for this project."}), 403 # On successful write access, update the project's update timestamp ProjectService.set_project_update_timestamp(project_id) return None # All checks passed