CVE-2024-37306 (Python, published 2024-06-13T15:15Z)
Description: value: Computer Vision Annotation Tool (CVAT) is an interactive video and image annotation tool for computer vision. Starting in version 2.2.0 and prior to version 2.14.3, if an attacker can trick a logged-in CVAT user into visiting a malicious URL, they can initiate a dataset export or a backup from a project, task or job that the victim user has permission to export into a cloud storage that the victim user has access to. The name of the resulting file can be chosen by the attacker. This implies that the attacker can overwrite arbitrary files in any cloud storage that the victim can access and, if the attacker has read access to the cloud storage used in the attack, they can obtain media files, annotations, settings and other information from any projects, tasks or jobs that the victim has permission to export. Version 2.14.3 contains a fix for the issue. No known workarounds are available.
Severity: nan (CVSS v3 base score: )
Weakness: NVD-CWE-noinfo \u2014 Insufficient Information
There is insufficient information about the issue to classify it; details are unkown or unspecified.
Repository: https://github.com/cvat-ai/cvat
Files changed:
changelog.d/20240523_192101_roman_export_backup_csrf.md
cvat/apps/dataset_manager/tests/test_rest_api_formats.py
cvat/apps/engine/backup.py
cvat/apps/engine/mixins.py
cvat/apps/engine/tests/test_rest_api_3D.py
cvat/apps/engine/views.py
tests/python/rest_api/test_csrf_workaround.py
tests/python/shared/utils/resource_import_export.py
Commit message: Merge pull request from GHSA-jpf9-646h-4px7

* Mitigate a CSRF vulnerability in export and backup-related endpoints

While Django has built-in CSRF protection (which we use), it does not cover
GET requests, and AFAICS, there is no way to force it to do that.
Unfortunately, the many endpoints that initiate dataset exports and backups
do accept GET requests _and_ initiate side effects, making them susceptible.

The proper fix for this issue would be to redesign those endpoints to use
POST requests, but a) that's more complicated, and b) we should still keep
the old endpoints for backwards compatibility.

So apply a less proper fix, which is to disable session authentication for
the affected endpoints. It's a bit complex, because in some cases
(particularly when `action=download`) we _need_ session authentication to
work, because the UI redirects the user to such endpoints.

In addition, modify the handling logic for these endpoints in order to
ensure that when `action=download`, no side effects are triggered.
Previously, `action=download` would still queue an RQ job if none existed.

Even after this, `action=download` will still have two small side effects:

* An existing RQ job will be deleted if its results are out of date.
  I don't think this is a problem, because such a job cannot be used anyway.

* A completed RQ job will be deleted too. This is a problematic design,
  but I don't think an attacker can achieve anything by exploiting this. If
  an attacker maliciously redirects the user to an `action=download` URL,
  then they'll just download the export/backup as usual.

Some tests were making export requests incorrectly, so fix them.

* Add test for the CSRF workaround

Vulnerable code:
        response = self._get_request_with_data(url, data, user)
                        "action": "download",
                        "action": "download",
                "action": "download",
                "action": "download",
                "action": "download",
                        "action": "download",
                        "action": "download",
                        "action": "download",
                        "action": "download",
                            "action": "download",
            "action": "download",
                    "action": "download",
                    "action": "download",
                    "action": "download",
    last_instance_update_time = timezone.localtime(db_instance.updated_date)
    timestamp = datetime.strftime(last_instance_update_time, "%Y_%m_%d_%H_%M_%S")
    location = location_conf.get('location')

        else:
            if rq_job.is_finished:
                if location == Location.LOCAL:
                    file_path = rq_job.return_value()

                    if not file_path:
                        return Response('A result for exporting job was not found for finished RQ job', status=status.HTTP_500_INTERNAL_SERVER_ERROR)

                    elif not os.path.exists(file_path):
                        return Response('The result file does not exist in export cache', status=status.HTTP_500_INTERNAL_SERVER_ERROR)

                    filename = filename or build_backup_file_name(
                        class_name=obj_type,
                        identifier=db_instance.name,
                        timestamp=timestamp,
                        extension=os.path.splitext(file_path)[1]
                    )

                    if action == "download":
                        rq_job.delete()
                        return sendfile(request, file_path, attachment=True,
                            attachment_filename=filename)

                    return Response(status=status.HTTP_201_CREATED)

                elif location == Location.CLOUD_STORAGE:
                    rq_job.delete()
                    return Response(status=status.HTTP_200_OK)
                else:
                    raise NotImplementedError()
            elif rq_job.is_failed:
                exc_info = rq_job.meta.get('formatted_exception', str(rq_job.exc_info))
                return Response(exc_info,
                    status=status.HTTP_500_INTERNAL_SERVER_ERROR)
                return Response(status=status.HTTP_202_ACCEPTED)
from typing import Optional, Callable, Dict, Any
        response = self._get_request_with_data(url, data, user)
                        "action": "download",
                        "action": "download",
                        "action": "download",
                        "action": "download",
from typing import Optional, Any, Dict, List, cast, Callable
from cvat.apps.engine.mixins import PartialUpdateModelMixin, UploadMixin, AnnotationMixin, SerializeMixin
    PartialUpdateModelMixin, UploadMixin, AnnotationMixin, SerializeMixin
        url_path=r'dataset/?$', parser_classes=_UPLOAD_PARSER_CLASSES)
        serializer_class=LabeledDataSerializer)
    @action(methods=['GET'], detail=True, url_path='backup')
    PartialUpdateModelMixin, UploadMixin, AnnotationMixin, SerializeMixin
    @action(methods=['GET'], detail=True, url_path='backup')
        serializer_class=None, parser_classes=_UPLOAD_PARSER_CLASSES)
        url_path='dataset')
    UploadMixin, AnnotationMixin
        serializer_class=LabeledDataSerializer, parser_classes=_UPLOAD_PARSER_CLASSES)
        url_path='dataset')
    instance_update_time = timezone.localtime(db_instance.updated_date)
    if isinstance(db_instance, Project):
        tasks_update = list(map(lambda db_task: timezone.localtime(db_task.updated_date), db_instance.tasks.all()))
        instance_update_time = max(tasks_update + [instance_update_time])
        rq_request = rq_job.meta.get('request', None)
        request_time = rq_request.get('timestamp', None) if rq_request else None
        if request_time is None or request_time < instance_update_time:
            # The result is outdated, need to restart the export.
            # Cancel the current job.
            if rq_job.is_finished:
                if location == Location.CLOUD_STORAGE:
                    rq_job.delete()
                    return Response(status=status.HTTP_200_OK)

                elif location == Location.LOCAL:
                    file_path = rq_job.return_value()

                    if not file_path:
                        return Response(
                            'A result for exporting job was not found for finished RQ job',
                            status=status.HTTP_500_INTERNAL_SERVER_ERROR
                        )

                    with dm.util.get_export_cache_lock(
                        file_path, ttl=60, # request timeout
                    ):
                        if action == "download":
                            if not osp.exists(file_path):
                                return Response(
                                    "The exported file has expired, please retry exporting",
                                    status=status.HTTP_404_NOT_FOUND
                                )

                            filename = filename or \
                                build_annotations_file_name(
                                    class_name=db_instance.__class__.__name__,
                                    identifier=db_instance.name if isinstance(db_instance, (Task, Project)) else db_instance.id,
                                    timestamp=instance_timestamp,
                                    format_name=format_name,
                                    is_annotation_file=is_annotation_file,
                                    extension=osp.splitext(file_path)[1]
                                )

                            rq_job.delete()
                            return sendfile(request, file_path, attachment=True, attachment_filename=filename)
                        else:
                            if osp.exists(file_path):
                                # Update last update time to prolong the export lifetime
                                # as the last access time is not available on every filesystem
                                os.utime(file_path, None)

                                return Response(status=status.HTTP_201_CREATED)
                            else:
                                # Cancel and reenqueue the job.
                                # The new attempt will be made after the last existing job.
                                # In the case the server is configured with ONE_RUNNING_JOB_IN_QUEUE_PER_USER
                                # we have to enqueue dependent jobs after canceling one.
                                rq_job.cancel(enqueue_dependents=settings.ONE_RUNNING_JOB_IN_QUEUE_PER_USER)
                                rq_job.delete()
                else:
                    raise NotImplementedError(f"Export to {location} location is not implemented yet")
            elif rq_job.is_failed:
                exc_info = rq_job.meta.get('formatted_exception', str(rq_job.exc_info))
                rq_job.delete()
                return Response(exc_info, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
            elif rq_job.is_deferred and rq_id not in queue.deferred_job_registry.get_job_ids():
                # Sometimes jobs can depend on outdated jobs in the deferred jobs registry.
                # They can be fetched by their specific ids, but are not listed by get_job_ids().
                # Supposedly, this can happen because of the server restarts
                # (potentially, because the redis used for the queue is inmemory).
                # Another potential reason is canceling without enqueueing dependents.
                # Such dependencies are never removed or finished,
                # as there is no TTL for deferred jobs,
                # so the current job can be blocked indefinitely.

                # Cancel the current job and then reenqueue it, considering the current situation.
                # The new attempt will be made after the last existing job.
                # In the case the server is configured with ONE_RUNNING_JOB_IN_QUEUE_PER_USER
                # we have to enqueue dependent jobs after canceling one.
                rq_job.cancel(enqueue_dependents=settings.ONE_RUNNING_JOB_IN_QUEUE_PER_USER)
                rq_job.delete()
            else:
                return Response(status=status.HTTP_202_ACCEPTED)
            assert status in (HTTPStatus.CREATED, HTTPStatus.ACCEPTED)
            response = get_method(user, f"{obj}/{obj_id}/{resource}", action="download", **kwargs)

Fixed code:
### Security

- Mitigated a CSRF vulnerability in backup and export-related endpoints
  (<https://github.com/cvat-ai/cvat/security/advisories/GHSA-jpf9-646h-4px7>)
        response = self._get_request_with_data(url, {**data, "action": "download"}, user)
    last_instance_update_time = timezone.localtime(db_instance.updated_date)

            rq_job = None

    timestamp = datetime.strftime(last_instance_update_time, "%Y_%m_%d_%H_%M_%S")
    location = location_conf.get('location')

    if action == "download":
        if location != Location.LOCAL:
            return Response('Action "download" is only supported for a local backup location', status=status.HTTP_400_BAD_REQUEST)

        if not rq_job or not rq_job.is_finished:
            return Response('Backup has not finished', status=status.HTTP_400_BAD_REQUEST)

        file_path = rq_job.return_value()

        if not file_path:
            return Response('A result for exporting job was not found for finished RQ job', status=status.HTTP_500_INTERNAL_SERVER_ERROR)

        elif not os.path.exists(file_path):
            return Response('The result file does not exist in export cache', status=status.HTTP_500_INTERNAL_SERVER_ERROR)

        filename = filename or build_backup_file_name(
            class_name=obj_type,
            identifier=db_instance.name,
            timestamp=timestamp,
            extension=os.path.splitext(file_path)[1]
        )

        rq_job.delete()
        return sendfile(request, file_path, attachment=True,
            attachment_filename=filename)

    if rq_job:
        if rq_job.is_finished:
            if location == Location.LOCAL:
                return Response(status=status.HTTP_201_CREATED)

            elif location == Location.CLOUD_STORAGE:
                return Response(status=status.HTTP_200_OK)
                raise NotImplementedError()
        elif rq_job.is_failed:
            exc_info = rq_job.meta.get('formatted_exception', str(rq_job.exc_info))
            rq_job.delete()
            return Response(exc_info,
                status=status.HTTP_500_INTERNAL_SERVER_ERROR)
        else:
            return Response(status=status.HTTP_202_ACCEPTED)
from typing import Optional, Callable, Dict, Any, Mapping
from rest_framework.authentication import SessionAuthentication
from rest_framework.views import APIView

class CsrfWorkaroundMixin(APIView):
    """
    Disables session authentication for GET/HEAD requests
    for which csrf_workaround_is_needed returns True.

    csrf_workaround_is_needed is supposed to be overridden by each view.

    This only exists to mitigate CSRF attacks on several known endpoints that
    perform side effects in response to GET requests. Do not use this in
    new code: instead, make sure that all endpoints with side effects use
    a method other than GET/HEAD. Then Django's built-in CSRF protection
    will cover them.
    """

    @staticmethod
    def csrf_workaround_is_needed(query_params: Mapping[str, str]) -> bool:
        return False

    def get_authenticators(self):
        authenticators = super().get_authenticators()

        if (
            self.request and
            # Don't apply the workaround for requests from unit tests, since
            # they can only use session authentication.
            not getattr(self.request, "_dont_enforce_csrf_checks", False) and
            self.request.method in ("GET", "HEAD") and
            self.csrf_workaround_is_needed(self.request.GET)
        ):
            authenticators = [a for a in authenticators if not isinstance(a, SessionAuthentication)]

        return authenticators
        response = self._get_request_with_data(url, {**data, "action": "download"}, user)
from typing import Optional, Any, Dict, List, cast, Callable, Mapping
from cvat.apps.engine.mixins import (
    PartialUpdateModelMixin, UploadMixin, AnnotationMixin, SerializeMixin, CsrfWorkaroundMixin
)
def csrf_workaround_is_needed_for_backup(query_params: Mapping[str, str]) -> bool:
    return query_params.get('action') != 'download'

def csrf_workaround_is_needed_for_export(query_params: Mapping[str, str]) -> bool:
    return 'format' in query_params and query_params.get('action') != 'download'

    PartialUpdateModelMixin, UploadMixin, AnnotationMixin, SerializeMixin, CsrfWorkaroundMixin
        url_path=r'dataset/?$', parser_classes=_UPLOAD_PARSER_CLASSES,
        csrf_workaround_is_needed=lambda qp:
            csrf_workaround_is_needed_for_export(qp) and qp.get("action") != "import_status")
        serializer_class=LabeledDataSerializer,
        csrf_workaround_is_needed=csrf_workaround_is_needed_for_export)
    @action(methods=['GET'], detail=True, url_path='backup',
        csrf_workaround_is_needed=csrf_workaround_is_needed_for_backup)
    PartialUpdateModelMixin, UploadMixin, AnnotationMixin, SerializeMixin, CsrfWorkaroundMixin
    @action(methods=['GET'], detail=True, url_path='backup',
        csrf_workaround_is_needed=csrf_workaround_is_needed_for_backup)
        serializer_class=None, parser_classes=_UPLOAD_PARSER_CLASSES,
        csrf_workaround_is_needed=csrf_workaround_is_needed_for_export)
        url_path='dataset', csrf_workaround_is_needed=csrf_workaround_is_needed_for_export)
    UploadMixin, AnnotationMixin, CsrfWorkaroundMixin
        serializer_class=LabeledDataSerializer, parser_classes=_UPLOAD_PARSER_CLASSES,
        csrf_workaround_is_needed=csrf_workaround_is_needed_for_export)
        url_path='dataset', csrf_workaround_is_needed=csrf_workaround_is_needed_for_export)
    instance_update_time = timezone.localtime(db_instance.updated_date)
    if isinstance(db_instance, Project):
        tasks_update = list(map(lambda db_task: timezone.localtime(db_task.updated_date), db_instance.tasks.all()))
        instance_update_time = max(tasks_update + [instance_update_time])

    if rq_job:
        rq_request = rq_job.meta.get('request', None)
        request_time = rq_request.get('timestamp', None) if rq_request else None
        if request_time is None or request_time < instance_update_time:
            # The result is outdated, need to restart the export.
            # Cancel the current job.
            # The new attempt will be made after the last existing job.
            # In the case the server is configured with ONE_RUNNING_JOB_IN_QUEUE_PER_USER
            # we have to enqueue dependent jobs after canceling one.
            rq_job.cancel(enqueue_dependents=settings.ONE_RUNNING_JOB_IN_QUEUE_PER_USER)
            rq_job.delete()
            rq_job = None

    REQUEST_TIMEOUT = 60

    if action == "download":
        if location != Location.LOCAL:
            return Response('Action "download" is only supported for a local export location',
                status=status.HTTP_400_BAD_REQUEST)

        if not rq_job or not rq_job.is_finished:
            return Response('Export has not finished', status=status.HTTP_400_BAD_REQUEST)

        file_path = rq_job.return_value()

        if not file_path:
            return Response(
                'A result for exporting job was not found for finished RQ job',
                status=status.HTTP_500_INTERNAL_SERVER_ERROR
            )

        with dm.util.get_export_cache_lock(file_path, ttl=REQUEST_TIMEOUT):
            if not osp.exists(file_path):
                return Response(
                    "The exported file has expired, please retry exporting",
                    status=status.HTTP_404_NOT_FOUND
                )

            filename = filename or \
                build_annotations_file_name(
                    class_name=db_instance.__class__.__name__,
                    identifier=db_instance.name if isinstance(db_instance, (Task, Project)) else db_instance.id,
                    timestamp=instance_timestamp,
                    format_name=format_name,
                    is_annotation_file=is_annotation_file,
                    extension=osp.splitext(file_path)[1]
                )

            rq_job.delete()
            return sendfile(request, file_path, attachment=True, attachment_filename=filename)


        if rq_job.is_finished:
            if location == Location.CLOUD_STORAGE:
                rq_job.delete()
                return Response(status=status.HTTP_200_OK)

            elif location == Location.LOCAL:
                file_path = rq_job.return_value()

                if not file_path:
                    return Response(
                        'A result for exporting job was not found for finished RQ job',
                        status=status.HTTP_500_INTERNAL_SERVER_ERROR
                    )

                with dm.util.get_export_cache_lock(file_path, ttl=REQUEST_TIMEOUT):
                    if osp.exists(file_path):
                        # Update last update time to prolong the export lifetime
                        # as the last access time is not available on every filesystem
                        os.utime(file_path, None)

                        return Response(status=status.HTTP_201_CREATED)
                    else:
                        # Cancel and reenqueue the job.
                        # The new attempt will be made after the last existing job.
                        # In the case the server is configured with ONE_RUNNING_JOB_IN_QUEUE_PER_USER
                        # we have to enqueue dependent jobs after canceling one.
                        rq_job.cancel(enqueue_dependents=settings.ONE_RUNNING_JOB_IN_QUEUE_PER_USER)
                        rq_job.delete()
            else:
                raise NotImplementedError(f"Export to {location} location is not implemented yet")
        elif rq_job.is_failed:
            exc_info = rq_job.meta.get('formatted_exception', str(rq_job.exc_info))
            rq_job.delete()
            return Response(exc_info, status=status.HTTP_500_INTERNAL_SERVER_ERROR)
        elif rq_job.is_deferred and rq_id not in queue.deferred_job_registry.get_job_ids():
            # Sometimes jobs can depend on outdated jobs in the deferred jobs registry.
            # They can be fetched by their specific ids, but are not listed by get_job_ids().
            # Supposedly, this can happen because of the server restarts
            # (potentially, because the redis used for the queue is inmemory).
            # Another potential reason is canceling without enqueueing dependents.
            # Such dependencies are never removed or finished,
            # as there is no TTL for deferred jobs,
            # so the current job can be blocked indefinitely.

            # Cancel the current job and then reenqueue it, considering the current situation.
            return Response(status=status.HTTP_202_ACCEPTED)
# Copyright (C) 2024 CVAT.ai Corporation
#
# SPDX-License-Identifier: MIT

import pytest
from cvat_sdk.api_client import ApiClient, Configuration, models
from cvat_sdk.api_client.exceptions import UnauthorizedException

from shared.utils.config import BASE_URL, USER_PASS


class TestCsrfWorkaround:
    """
    Test that session authentication does not work with endpoints
    that respond to GET requests and produce side effects.
    """

    @pytest.fixture(autouse=True)
    def setup(self, admin_user):
        self.client = ApiClient(Configuration(host=BASE_URL))

        # Don't store the result - we only want the session cookie.
        self.client.auth_api.create_login(
            models.LoginSerializerExRequest(username=admin_user, password=USER_PASS)
        )

        # Test that session authentication works in general.
        user, _ = self.client.users_api.retrieve_self()
        assert user.username == admin_user

    def test_project(self, projects):
        actual_project, _ = self.client.projects_api.retrieve(next(iter(projects))["id"])

        with pytest.raises(UnauthorizedException):
            self.client.projects_api.retrieve_backup(actual_project.id)
        with pytest.raises(UnauthorizedException):
            self.client.projects_api.retrieve_annotations(id=actual_project.id, format="COCO 1.0")
        with pytest.raises(UnauthorizedException):
            self.client.projects_api.retrieve_dataset(id=actual_project.id, format="COCO 1.0")

    def test_task(self, tasks):
        actual_task, _ = self.client.tasks_api.retrieve(next(iter(tasks))["id"])

        with pytest.raises(UnauthorizedException):
            self.client.tasks_api.retrieve_backup(actual_task.id)
        with pytest.raises(UnauthorizedException):
            self.client.tasks_api.retrieve_annotations(id=actual_task.id, format="COCO 1.0")
        with pytest.raises(UnauthorizedException):
            self.client.tasks_api.retrieve_dataset(id=actual_task.id, format="COCO 1.0")

    def test_job(self, jobs):
        actual_job, _ = self.client.jobs_api.retrieve(next(iter(jobs))["id"])

        with pytest.raises(UnauthorizedException):
            self.client.jobs_api.retrieve_annotations(id=actual_job.id, format="COCO 1.0")
        with pytest.raises(UnauthorizedException):
            self.client.jobs_api.retrieve_dataset(id=actual_job.id, format="COCO 1.0")
            assert status == HTTPStatus.ACCEPTED
            response = get_method(user, f"{obj}/{obj_id}/{resource}", **kwargs)