✨(back) create item endpoint creating a WOPI session

In order to start a WOPI session, we have to fetch an endpoint on the
item viewset returing the access token, the access token ttl and the
wopi client launch url.

Author: lunika

src/backend/core/api/viewsets.py

@@ -22,6 +22,8 @@
 from rest_framework.permissions import AllowAny
 
 from core import enums, models
+from wopi.services import access as access_service
+from wopi.utils import get_wopi_client_config
 
 from . import permissions, serializers, utils
 from .filters import ItemFilter, ListItemFilter
@@ -966,6 +968,31 @@ def media_auth(self, request, *args, **kwargs):
 
         return drf.response.Response("authorized", headers=request.headers, status=200)
 
+    @drf.decorators.action(detail=True, methods=["get"], url_path="wopi")
+    def wopi(self, request, *args, **kwargs):
+        """
+        This view is used to generate an access token and access token ttl in order to start
+        a WOPI session for the item and the current user.
+        """
+        item = self.get_object()
+
+        if not (wopi_client := get_wopi_client_config(item)):
+            raise drf.exceptions.ValidationError(
+                {"detail": "This item does not suport WOPI integration."}
+            )
+
+        service = access_service.AccessUserItemService()
+        access_token, access_token_ttl = service.insert_new_access(item, request.user)
+
+        return drf.response.Response(
+            {
+                "access_token": access_token,
+                "access_token_ttl": access_token_ttl,
+                "launch_url": wopi_client["launch_url"],
+            },
+            status=drf.status.HTTP_200_OK,
+        )
+
 
 class ItemAccessViewSet(
     ResourceAccessViewsetMixin,

src/backend/core/models.py

@@ -681,6 +681,7 @@ def get_abilities(self, user, ancestors_links=None):
             "media_auth": can_get,
             "update": can_update,
             "upload_ended": is_owner_or_admin,
+            "wopi": can_get,
         }
 
     def send_email(self, subject, emails, context=None, language=None):

src/backend/core/tests/items/test_api_items_retrieve.py

@@ -44,6 +44,7 @@ def test_api_items_retrieve_anonymous_public_standalone():
             "tree": True,
             "update": item.link_role == "editor",
             "upload_ended": False,
+            "wopi": True,
         },
         "created_at": item.created_at.isoformat().replace("+00:00", "Z"),
         "creator": str(item.creator.id),
@@ -104,6 +105,7 @@ def test_api_items_retrieve_anonymous_public_parent():
             "tree": True,
             "update": grand_parent.link_role == "editor",
             "upload_ended": False,
+            "wopi": True,
         },
         "created_at": item.created_at.isoformat().replace("+00:00", "Z"),
         "creator": str(item.creator.id),
@@ -194,6 +196,7 @@ def test_api_items_retrieve_authenticated_unrelated_public_or_authenticated(reac
             "tree": True,
             "update": item.link_role == "editor",
             "upload_ended": False,
+            "wopi": True,
         },
         "created_at": item.created_at.isoformat().replace("+00:00", "Z"),
         "creator": str(item.creator.id),
@@ -259,6 +262,7 @@ def test_api_items_retrieve_authenticated_public_or_authenticated_parent(reach):
             "tree": True,
             "update": grand_parent.link_role == "editor",
             "upload_ended": False,
+            "wopi": True,
         },
         "created_at": item.created_at.isoformat().replace("+00:00", "Z"),
         "creator": str(item.creator.id),
@@ -438,6 +442,7 @@ def test_api_items_retrieve_authenticated_related_parent():
             "tree": True,
             "update": access.role != "reader",
             "upload_ended": access.role in ["administrator", "owner"],
+            "wopi": True,
         },
         "creator": str(item.creator.id),
         "created_at": item.created_at.isoformat().replace("+00:00", "Z"),

src/backend/core/tests/items/test_api_items_trashbin.py

@@ -84,6 +84,7 @@ def test_api_items_trashbin_format():
             "tree": True,
             "update": True,
             "upload_ended": True,
+            "wopi": True,
         },
         "created_at": item.created_at.isoformat().replace("+00:00", "Z"),
         "creator": str(item.creator.id),

src/backend/core/tests/items/test_api_items_wopi.py

@@ -0,0 +1,209 @@
+"""Test for items API endpoint managing wopi init request."""
+
+from django.contrib.auth.models import AnonymousUser
+from django.utils import timezone
+
+import pytest
+from rest_framework.test import APIClient
+
+from core import factories, models
+
+pytestmark = pytest.mark.django_db
+
+
+@pytest.fixture
+def timestamp_now():
+    """Timestamp now in milliseconds."""
+    return int(round(timezone.now().timestamp())) * 1000
+
+
+@pytest.fixture
+def valid_mimetype():
+    """Valid mimetype for testing."""
+    return "application/vnd.oasis.opendocument.text"
+
+
+@pytest.fixture
+def valid_wopi_launch_url():
+    """Valid WOPI launch URL for testing."""
+    return "https://vendorA.com/launch_url"
+
+
+@pytest.fixture(autouse=True)
+def configure_wopi_settings(settings, valid_mimetype, valid_wopi_launch_url):
+    settings.WOPI_CLIENTS = ["vendorA"]
+    settings.WOPI_CLIENTS_CONFIGURATION = {
+        "vendorA": {
+            "launch_url": valid_wopi_launch_url,
+            "mimetypes": [valid_mimetype],
+        }
+    }
+
+
+def test_api_items_wopi_not_existing_item():
+    """
+    Anonymous user cannot generate wopi access token for non-existing item.
+    """
+
+    client = APIClient()
+    response = client.get("/api/v1.0/items/00000000-0000-0000-0000-000000000000/wopi/")
+
+    assert response.status_code == 404
+
+
+def test_api_items_wopi_anonymous_user_item_public(
+    timestamp_now, valid_mimetype, valid_wopi_launch_url
+):
+    """
+    Anonymous user can generate wopi access token for public item.
+    """
+
+    item = factories.ItemFactory(
+        link_reach=models.LinkReachChoices.PUBLIC,
+        type=models.ItemTypeChoices.FILE,
+        mimetype=valid_mimetype,
+    )
+    item.upload_state = models.ItemUploadStateChoices.UPLOADED
+    item.save()
+
+    client = APIClient()
+    response = client.get(f"/api/v1.0/items/{item.id!s}/wopi/")
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["access_token"] is not None
+    assert data["access_token_ttl"] > timestamp_now
+    assert data["launch_url"] == valid_wopi_launch_url
+
+
+@pytest.mark.parametrize(
+    "link_reach",
+    [models.LinkReachChoices.AUTHENTICATED, models.LinkReachChoices.RESTRICTED],
+)
+def test_api_items_wopi_anonymous_user_item_not_public(link_reach):
+    """Anymous user can not access not public item."""
+    item = factories.ItemFactory(link_reach=link_reach)
+
+    client = APIClient()
+    response = client.get(f"/api/v1.0/items/{item.id!s}/wopi/")
+
+    assert response.status_code == 401
+
+
+def test_api_items_wopi_anonymous_user_not_item_file():
+    """Anymous user can not access item that is not a file."""
+    item = factories.ItemFactory(
+        type=models.ItemTypeChoices.FOLDER, link_reach=models.LinkReachChoices.PUBLIC
+    )
+
+    client = APIClient()
+    response = client.get(f"/api/v1.0/items/{item.id!s}/wopi/")
+
+    assert response.status_code == 400
+    assert response.json() == {"detail": "This item does not suport WOPI integration."}
+
+
+def test_api_items_wopi_anonymous_item_file_mimetype_not_supported():
+    """Anymous user can not access item file with mimetype not supported."""
+    item = factories.ItemFactory(
+        type=models.ItemTypeChoices.FILE,
+        mimetype="image/png",
+        link_reach=models.LinkReachChoices.PUBLIC,
+    )
+    item.upload_state = models.ItemUploadStateChoices.UPLOADED
+    item.save()
+
+    client = APIClient()
+    response = client.get(f"/api/v1.0/items/{item.id!s}/wopi/")
+
+    assert response.status_code == 400
+    assert response.json() == {"detail": "This item does not suport WOPI integration."}
+
+
+def test_api_items_wopi_anonymous_user_item_not_uploaded():
+    """Anymous user can not access item file that is not uploaded."""
+    item = factories.ItemFactory(
+        type=models.ItemTypeChoices.FILE,
+        mimetype="image/png",
+        link_reach=models.LinkReachChoices.PUBLIC,
+    )
+
+    client = APIClient()
+    response = client.get(f"/api/v1.0/items/{item.id!s}/wopi/")
+
+    assert response.status_code == 400
+    assert response.json() == {"detail": "This item does not suport WOPI integration."}
+
+
+def test_api_items_wopi_authenticated_user_item_not_accessible():
+    """Authenticated user can not access item that is not accessible."""
+    user = factories.UserFactory()
+    item = factories.ItemFactory(link_reach=models.LinkReachChoices.RESTRICTED)
+
+    client = APIClient()
+    client.force_login(user)
+    response = client.get(f"/api/v1.0/items/{item.id!s}/wopi/")
+
+    assert response.status_code == 403
+
+
+def test_api_items_wopi_authenticated_can_access_retricted_item(
+    timestamp_now, valid_mimetype, valid_wopi_launch_url
+):
+    """Authenticated user can access item that is accessible."""
+    user = factories.UserFactory()
+    item = factories.ItemFactory(
+        link_reach=models.LinkReachChoices.RESTRICTED,
+        type=models.ItemTypeChoices.FILE,
+        mimetype=valid_mimetype,
+    )
+    item.upload_state = models.ItemUploadStateChoices.UPLOADED
+    item.save()
+    factories.UserItemAccessFactory(user=user, item=item)
+
+    client = APIClient()
+    client.force_login(user)
+    response = client.get(f"/api/v1.0/items/{item.id!s}/wopi/")
+
+    assert response.status_code == 200
+    data = response.json()
+    assert data["access_token"] is not None
+    assert data["access_token_ttl"] > timestamp_now
+    assert data["launch_url"] == valid_wopi_launch_url
+
+
+def test_api_items_wopi_authenticated_user_item_not_file():
+    """Authenticated user can not access item that is not a file."""
+    user = factories.UserFactory()
+    item = factories.ItemFactory(
+        link_reach=models.LinkReachChoices.RESTRICTED,
+        type=models.ItemTypeChoices.FOLDER,
+    )
+    factories.UserItemAccessFactory(user=user, item=item)
+
+    client = APIClient()
+    client.force_login(user)
+    response = client.get(f"/api/v1.0/items/{item.id!s}/wopi/")
+
+    assert response.status_code == 400
+    assert response.json() == {"detail": "This item does not suport WOPI integration."}
+
+
+def test_api_items_wopi_authenticated_user_item_mimetype_not_supported():
+    """Authenticated user can not access item that mimetype is not supported."""
+    user = factories.UserFactory()
+    item = factories.ItemFactory(
+        link_reach=models.LinkReachChoices.RESTRICTED,
+        type=models.ItemTypeChoices.FILE,
+        mimetype="image/png",
+    )
+    item.upload_state = models.ItemUploadStateChoices.UPLOADED
+    item.save()
+    factories.UserItemAccessFactory(user=user, item=item)
+
+    client = APIClient()
+    client.force_login(user)
+    response = client.get(f"/api/v1.0/items/{item.id!s}/wopi/")
+
+    assert response.status_code == 400
+    assert response.json() == {"detail": "This item does not suport WOPI integration."}

src/backend/core/tests/test_models_items.py

@@ -137,6 +137,7 @@ def test_models_items_get_abilities_forbidden(
         "tree": False,
         "update": False,
         "upload_ended": False,
+        "wopi": False,
     }
     nb_queries = 1 if is_authenticated else 0
     with django_assert_num_queries(nb_queries):
@@ -180,6 +181,7 @@ def test_models_items_get_abilities_reader(
         "tree": True,
         "update": False,
         "upload_ended": False,
+        "wopi": True,
     }
     nb_queries = 1 if is_authenticated else 0
     with django_assert_num_queries(nb_queries):
@@ -223,6 +225,7 @@ def test_models_items_get_abilities_editor(
         "tree": True,
         "update": True,
         "upload_ended": False,
+        "wopi": True,
     }
     nb_queries = 1 if is_authenticated else 0
     with django_assert_num_queries(nb_queries):
@@ -253,6 +256,7 @@ def test_models_items_get_abilities_owner(django_assert_num_queries):
         "tree": True,
         "update": True,
         "upload_ended": True,
+        "wopi": True,
     }
     with django_assert_num_queries(1):
         assert item.get_abilities(user) == expected_abilities
@@ -283,6 +287,7 @@ def test_models_items_get_abilities_administrator(django_assert_num_queries):
         "tree": True,
         "update": True,
         "upload_ended": True,
+        "wopi": True,
     }
     with django_assert_num_queries(1):
         assert item.get_abilities(user) == expected_abilities
@@ -312,6 +317,7 @@ def test_models_items_get_abilities_editor_user(django_assert_num_queries):
         "tree": True,
         "update": True,
         "upload_ended": False,
+        "wopi": True,
     }
     with django_assert_num_queries(1):
         assert item.get_abilities(user) == expected_abilities
@@ -342,6 +348,7 @@ def test_models_items_get_abilities_reader_user(django_assert_num_queries):
         "tree": True,
         "update": access_from_link,
         "upload_ended": False,
+        "wopi": True,
     }
     with django_assert_num_queries(1):
         assert item.get_abilities(user) == expected_abilities
@@ -375,6 +382,7 @@ def test_models_items_get_abilities_preset_role(django_assert_num_queries):
         "tree": True,
         "update": False,
         "upload_ended": False,
+        "wopi": True,
     }
 
 

src/backend/wopi/services/access.py

@@ -70,7 +70,7 @@ def generate_token():
         """Generate a rancom access token"""
         return token_urlsafe()
 
-    def insert_new_access(self, item: Item, user: AbstractUser) -> str:
+    def insert_new_access(self, item: Item, user: AbstractUser) -> tuple[str, int]:
         """
         Insert a new access token for the user and item. Return an access_token and access_token_ttl
         access_token_ttl must be a timestamp in milliseconds

src/backend/wopi/utils/__init__.py

@@ -19,3 +19,19 @@ def is_item_wopi_supported(item):
         if item.mimetype in client_config["mimetypes"]:
             return True
     return False
+
+
+def get_wopi_client_config(item):
+    """
+    Get the WOPI client configuration for an item.
+    """
+    if item.type != models.ItemTypeChoices.FILE:
+        return False
+
+    if item.upload_state != models.ItemUploadStateChoices.UPLOADED:
+        return False
+
+    for _, client_config in settings.WOPI_CLIENTS_CONFIGURATION.items():
+        if item.mimetype in client_config["mimetypes"]:
+            return client_config
+    return None