Small refactoring (#7149)

cvat/apps/engine/views.py

@@ -81,7 +81,7 @@
 from .log import ServerLogManager
 from cvat.apps.iam.permissions import (CloudStoragePermission,
     CommentPermission, IssuePermission, JobPermission, LabelPermission, ProjectPermission,
-    TaskPermission, UserPermission)
+    TaskPermission, UserPermission, PolicyEnforcer, IsAuthenticatedOrReadPublicResource)
 from cvat.apps.iam.filters import ORGANIZATION_OPEN_API_PARAMETERS
 from cvat.apps.engine.cache import MediaCache
 from cvat.apps.engine.view_utils import tus_chunk_action
@@ -2676,6 +2676,11 @@ class AssetsViewSet(
     def check_object_permissions(self, request, obj):
         super().check_object_permissions(request, obj.guide)
 
+    def get_permissions(self):
+        if self.action == 'retrieve':
+            return [IsAuthenticatedOrReadPublicResource(), PolicyEnforcer()]
+        return super().get_permissions()
+
     def get_serializer_class(self):
         if self.request.method in SAFE_METHODS:
             return AssetReadSerializer
@@ -2728,21 +2733,6 @@ def retrieve(self, request, *args, **kwargs):
         instance = self.get_object()
         return sendfile(request, os.path.join(settings.ASSETS_ROOT, str(instance.uuid), instance.filename))
 
-    # FIXME: It should be done in another way. It is better to introduce a "public resource" concept and handle it
-    # properly in PolicyEnfocer. Looks like PolicyEnfocer should handle rest_framework.permissions.IsAuthenticated internally.
-    @action(methods=['GET'], detail=True, url_path='public', permission_classes=[])
-    def public_retrieve(self, request, *args, **kwargs):
-        # Note: It is not a good approach to implement one more endpoint for receiving public assets
-        # but it separated to 2 endpoints for better server API specification.
-        # It could be implemented via overwriting get_permissions func,
-        # but in that case the specification would contain incorrect security information.
-        # Note: we cannot move this logic to OPA because OPA permissions
-        # don't imply that the user will be anonymous.
-        instance = self.get_object()
-        if instance.guide.is_public:
-            return sendfile(request, os.path.join(settings.ASSETS_ROOT, str(instance.uuid), instance.filename))
-        return Response(status=status.HTTP_204_NO_CONTENT)
-
     def perform_destroy(self, instance):
         full_path = os.path.join(instance.get_asset_dir(), instance.filename)
         if os.path.exists(full_path):

cvat/apps/iam/middleware.py

@@ -0,0 +1,59 @@
+# Copyright (C) 2023 CVAT.ai Corporation
+#
+# SPDX-License-Identifier: MIT
+
+from django.utils.functional import SimpleLazyObject
+from rest_framework.exceptions import ValidationError, NotFound
+from django.conf import settings
+
+
+def get_organization(request):
+    from cvat.apps.organizations.models import Organization
+
+    IAM_ROLES = {role: priority for priority, role in enumerate(settings.IAM_ROLES)}
+    groups = list(request.user.groups.filter(name__in=list(IAM_ROLES.keys())))
+    groups.sort(key=lambda group: IAM_ROLES[group.name])
+    privilege = groups[0] if groups else None
+
+    organization = None
+
+    try:
+        org_slug = request.GET.get('org')
+        org_id = request.GET.get('org_id')
+        org_header = request.headers.get('X-Organization')
+
+        if org_id is not None and (org_slug is not None or org_header is not None):
+            raise ValidationError('You cannot specify "org_id" query parameter with '
+                '"org" query parameter or "X-Organization" HTTP header at the same time.')
+
+        if org_slug is not None and org_header is not None and org_slug != org_header:
+            raise ValidationError('You cannot specify "org" query parameter and '
+                '"X-Organization" HTTP header with different values.')
+
+        org_slug = org_slug if org_slug is not None else org_header
+
+        if org_slug:
+            organization = Organization.objects.get(slug=org_slug)
+        elif org_id:
+            organization = Organization.objects.get(id=int(org_id))
+    except Organization.DoesNotExist:
+        raise NotFound(f'{org_slug or org_id} organization does not exist.')
+
+    context = {
+        "organization": organization,
+        "privilege": getattr(privilege, 'name', None)
+    }
+
+    return context
+
+
+class ContextMiddleware:
+    def __init__(self, get_response):
+        self.get_response = get_response
+
+    def __call__(self, request):
+
+        # https://stackoverflow.com/questions/26240832/django-and-middleware-which-uses-request-user-is-always-anonymous
+        request.iam_context = SimpleLazyObject(lambda: get_organization(request))
+
+        return self.get_response(request)

cvat/apps/iam/permissions.py

@@ -9,12 +9,11 @@
 from abc import ABCMeta, abstractmethod
 from collections import namedtuple
 from enum import Enum
-from importlib import import_module
-from typing import Any, Dict, List, Optional, Sequence, Union, cast
+from typing import Any, Dict, List, Optional, Sequence, Union, cast, TypeVar
 
 from attrs import define, field
 from django.conf import settings
-from django.db.models import Q
+from django.db.models import Q, Model
 from rest_framework.exceptions import PermissionDenied, ValidationError
 from rest_framework.permissions import BasePermission
 
@@ -85,18 +84,24 @@ def get_membership(request, organization):
         is_active=True
     ).first()
 
+def build_iam_context(request, organization: Optional[Organization], membership: Optional[Membership]):
+    return {
+        'user_id': request.user.id,
+        'group_name': request.iam_context['privilege'],
+        'org_id': getattr(organization, 'id', None),
+        'org_slug': getattr(organization, 'slug', None),
+        'org_owner_id': getattr(organization.owner, 'id', None)
+            if organization else None,
+        'org_role': getattr(membership, 'role', None),
+    }
 
-def get_iam_context(request, obj):
+
+def get_iam_context(request, obj) -> Dict[str, Any]:
     organization = get_organization(request, obj)
     membership = get_membership(request, organization)
 
-    iam_context = dict()
-    for builder_func_path in settings.IAM_CONTEXT_BUILDERS:
-        package, attr = builder_func_path.rsplit('.', 1)
-        builder_func = getattr(import_module(package), attr)
-        iam_context.update(builder_func(request, organization, membership))
+    return build_iam_context(request, organization, membership)
 
-    return iam_context
 
 class OpenPolicyAgentPermission(metaclass=ABCMeta):
     url: str
@@ -1994,22 +1999,27 @@ def get_resource(self):
 
         return data
 
+T = TypeVar('T', bound=Model)
+
+def is_public_obj(obj: T) -> bool:
+    return getattr(obj, "is_public", False)
 
 class PolicyEnforcer(BasePermission):
     # pylint: disable=no-self-use
-    def check_permission(self, request, view, obj):
-        permissions: List[OpenPolicyAgentPermission] = []
-
-        iam_context = get_iam_context(request, obj)
-
+    def check_permission(self, request, view, obj) -> bool:
         # DRF can send OPTIONS request. Internally it will try to get
         # information about serializers for PUT and POST requests (clone
         # request and replace the http method). To avoid handling
         # ('POST', 'metadata') and ('PUT', 'metadata') in every request,
         # the condition below is enough.
-        if not self.is_metadata_request(request, view):
-            for perm in OpenPolicyAgentPermission.__subclasses__():
-                permissions.extend(perm.create(request, view, obj, iam_context))
+        if self.is_metadata_request(request, view) or obj and is_public_obj(obj):
+            return True
+
+        permissions: List[OpenPolicyAgentPermission] = []
+        iam_context = get_iam_context(request, obj)
+
+        for perm in OpenPolicyAgentPermission.__subclasses__():
+            permissions.extend(perm.create(request, view, obj, iam_context))
 
         allow = True
         for perm in permissions:
@@ -2032,6 +2042,14 @@ def is_metadata_request(request, view):
         return request.method == 'OPTIONS' \
             or (request.method == 'POST' and view.action == 'metadata' and len(request.data) == 0)
 
+class IsAuthenticatedOrReadPublicResource(BasePermission):
+    def has_object_permission(self, request, view, obj) -> bool:
+        return bool(
+            request.user and request.user.is_authenticated or
+            request.method == 'GET' and is_public_obj(obj)
+        )
+
+
 class AnalyticsReportPermission(OpenPolicyAgentPermission):
     class Scopes(StrEnum):
         LIST = 'list'

cvat/apps/iam/signals.py

@@ -26,10 +26,9 @@ def create_user(sender, instance, created, **kwargs):
                 EmailAddress.objects.get_or_create(user=instance,
                     email=instance.email, primary=True, verified=True)
         else: # don't need to add default groups for superuser
-            if created:
-                for role in settings.GET_IAM_DEFAULT_ROLES(instance):
-                    db_group = Group.objects.get(name=role)
-                    instance.groups.add(db_group)
+            if created and not getattr(instance, 'skip_group_assigning', None):
+                db_group = Group.objects.get(name=settings.IAM_DEFAULT_ROLE)
+                instance.groups.add(db_group)
 
 elif settings.IAM_TYPE == 'LDAP':
     def create_user(sender, user=None, ldap_user=None, **kwargs):

cvat/apps/iam/utils.py

@@ -8,29 +8,13 @@ def create_opa_bundle():
     if bundle_path.is_file():
         bundle_path.unlink()
 
-    rules_paths = [Path(settings.BASE_DIR) / 'cvat/apps/iam/rules']
-    # FIXME: Let's have OPA_RULES_PATH instead for the list of directories.
-    if getattr(settings, 'EXTRA_RULES_PATHS', None):
-        rules_paths.extend([Path(settings.BASE_DIR) / p for p in settings.EXTRA_RULES_PATHS])
+    rules_paths = [Path(settings.BASE_DIR) / rel_path for rel_path in settings.IAM_OPA_RULES_PATH.strip(':').split(':')]
 
     with tarfile.open(bundle_path, 'w:gz') as tar:
         for p in rules_paths:
             for f in p.glob('*[!.gen].rego'):
                 tar.add(name=f, arcname=f.relative_to(p.parent))
 
-
-def build_iam_context(request, organization, membership):
-    return {
-        'user_id': request.user.id,
-        'group_name': request.iam_context['privilege'],
-        'org_id': getattr(organization, 'id', None),
-        'org_slug': getattr(organization, 'slug', None),
-        'org_owner_id': getattr(organization.owner, 'id', None)
-            if organization else None,
-        'org_role': getattr(membership, 'role', None),
-    }
-
-
 def get_dummy_user(email):
     from allauth.account.models import EmailAddress
     from allauth.account import app_settings

cvat/apps/iam/views.py

@@ -6,10 +6,9 @@
 import functools
 import hashlib
 
-from django.utils.functional import SimpleLazyObject
 from django.http import Http404, HttpResponseBadRequest, HttpResponseRedirect
 from rest_framework import views, serializers
-from rest_framework.exceptions import ValidationError, NotFound
+from rest_framework.exceptions import ValidationError
 from rest_framework.permissions import AllowAny
 from django.conf import settings
 from django.http import HttpResponse
@@ -29,56 +28,6 @@
 
 from .authentication import Signer
 
-def get_organization(request):
-    from cvat.apps.organizations.models import Organization
-
-    IAM_ROLES = {role: priority for priority, role in enumerate(settings.IAM_ROLES)}
-    groups = list(request.user.groups.filter(name__in=list(IAM_ROLES.keys())))
-    groups.sort(key=lambda group: IAM_ROLES[group.name])
-    privilege = groups[0] if groups else None
-
-    organization = None
-
-    try:
-        org_slug = request.GET.get('org')
-        org_id = request.GET.get('org_id')
-        org_header = request.headers.get('X-Organization')
-
-        if org_id is not None and (org_slug is not None or org_header is not None):
-            raise ValidationError('You cannot specify "org_id" query parameter with '
-                '"org" query parameter or "X-Organization" HTTP header at the same time.')
-
-        if org_slug is not None and org_header is not None and org_slug != org_header:
-            raise ValidationError('You cannot specify "org" query parameter and '
-                '"X-Organization" HTTP header with different values.')
-
-        org_slug = org_slug if org_slug is not None else org_header
-
-        if org_slug:
-            organization = Organization.objects.get(slug=org_slug)
-        elif org_id:
-            organization = Organization.objects.get(id=int(org_id))
-    except Organization.DoesNotExist:
-        raise NotFound(f'{org_slug or org_id} organization does not exist.')
-
-    context = {
-        "organization": organization,
-        "privilege": getattr(privilege, 'name', None)
-    }
-
-    return context
-
-class ContextMiddleware:
-    def __init__(self, get_response):
-        self.get_response = get_response
-
-    def __call__(self, request):
-
-        # https://stackoverflow.com/questions/26240832/django-and-middleware-which-uses-request-user-is-always-anonymous
-        request.iam_context = SimpleLazyObject(lambda: get_organization(request))
-
-        return self.get_response(request)
-
 @extend_schema(tags=['auth'])
 @extend_schema_view(post=extend_schema(
     summary='This method signs URL for access to the server',

cvat/schema.yml

@@ -176,32 +176,6 @@ paths:
       responses:
         '204':
           description: The asset has been deleted
-  /api/assets/{uuid}/public:
-    get:
-      operationId: assets_retrieve_public
-      parameters:
-      - in: path
-        name: uuid
-        schema:
-          type: string
-          format: uuid
-        description: A UUID string identifying this asset.
-        required: true
-      tags:
-      - assets
-      security:
-      - sessionAuth: []
-        csrfAuth: []
-        tokenAuth: []
-      - signatureAuth: []
-      - basicAuth: []
-      responses:
-        '200':
-          content:
-            application/vnd.cvat+json:
-              schema:
-                $ref: '#/components/schemas/AssetRead'
-          description: ''
   /api/auth/login:
     post:
       operationId: auth_create_login

cvat/settings/base.py

@@ -197,7 +197,7 @@ def generate_secret_key():
     'django.contrib.messages.middleware.MessageMiddleware',
     'django.middleware.clickjacking.XFrameOptionsMiddleware',
     'dj_pagination.middleware.PaginationMiddleware',
-    'cvat.apps.iam.views.ContextMiddleware',
+    'cvat.apps.iam.middleware.ContextMiddleware',
 ]
 
 UI_URL = ''
@@ -230,25 +230,18 @@ def generate_secret_key():
 # IAM settings
 IAM_TYPE = 'BASIC'
 IAM_BASE_EXCEPTION = None # a class which will be used by IAM to report errors
-
-# FIXME: There are several ways to "replace" default IAM role.
-# One of them is to assign groups when you create a user inside Crowdsourcing plugin and don't add more groups
-# if user.groups field isn't empty.
-# the function should be in uppercase to able get access from django.conf.settings
-def GET_IAM_DEFAULT_ROLES(user) -> list:
-    return ['user']
+IAM_DEFAULT_ROLE = 'user'
 
 IAM_ADMIN_ROLE = 'admin'
 # Index in the list below corresponds to the priority (0 has highest priority)
 IAM_ROLES = [IAM_ADMIN_ROLE, 'business', 'user', 'worker']
 IAM_OPA_HOST = 'http://opa:8181'
 IAM_OPA_DATA_URL = f'{IAM_OPA_HOST}/v1/data'
+IAM_OPA_RULES_PATH = 'cvat/apps/iam/rules:'
 LOGIN_URL = 'rest_login'
 LOGIN_REDIRECT_URL = '/'
 
 OBJECTS_NOT_RELATED_WITH_ORG = ['user', 'function', 'request', 'server',]
-# FIXME: It looks like an internal function of IAM app.
-IAM_CONTEXT_BUILDERS = ['cvat.apps.iam.utils.build_iam_context',]
 
 # ORG settings
 ORG_INVITATION_CONFIRM = 'No'
@@ -702,8 +695,6 @@ class CVAT_QUEUES(Enum):
 
 SMOKESCREEN_ENABLED = True
 
-EXTRA_RULES_PATHS = []
-
 # By default, email backend is django.core.mail.backends.smtp.EmailBackend
 # But it won't work without additional configuration, so we set it to None
 # to check configuration and throw ImproperlyConfigured if thats a case