✨(wopi) verify wopi signature in Wopi viewset

We can now implement the logic to verify a wopi request signature.
If a wopi configuration have proof keys then the request must provide a
signature and a valid timestamp.

CHANGELOG.md

@@ -20,6 +20,7 @@ and this project adheres to
 - ✨(backend) add a quota_excluded flag on items
 - ✨(backend) apply per-audience attributes to external api items
 - ✨(backend) add a grant_unlimited_storage command
+- ✨(wopi) verify the WOPI request proof signature
 
 ### Changed
 

src/backend/wopi/tests/conftest.py

@@ -4,9 +4,38 @@ from django.core.cache import cache
 
 import pytest
 
+from wopi.tasks.configure_wopi import WOPI_CONFIGURATION_CACHE_KEY
+
 
 @pytest.fixture(autouse=True)
 def clear_cache():
     """Fixture to clear the cache before each test."""
     yield
     cache.clear()
+
+
+@pytest.fixture
+def configure_wopi_clients():
+    """Configure wopi clients."""
+
+    wopi_configuration = {
+        "mimetypes": {
+            "text/plain": {
+                "launch_url": "http://localhost:9980/browser/0968141f2c/cool.html?",
+                "client": "vendorA",
+            }
+        },
+        "extensions": {
+            "txt": {
+                "launch_url": "http://localhost:9980/browser/0968141f2c/cool.html?",
+                "client": "vendorA",
+            }
+        },
+        "vendorA": {
+            "proof_keys": {"public_key": b"public_proof_key\n"},
+        },
+    }
+    cache.set(WOPI_CONFIGURATION_CACHE_KEY, wopi_configuration)
+
+    yield wopi_configuration
+    cache.delete(WOPI_CONFIGURATION_CACHE_KEY)

src/backend/wopi/tests/viewset/test_check_file_info.py

@@ -1,17 +1,22 @@
 """Testing the check file info endpoint"""
 
+from datetime import timedelta
 from io import BytesIO
+from unittest import mock
 
 from django.contrib.auth.models import AnonymousUser
 from django.core.cache import cache
 from django.core.files.storage import default_storage
+from django.utils.timezone import now
 
 import pytest
 from rest_framework.test import APIClient
 
 from core import factories, models
+from wopi.exceptions import WopiRequestSignatureError
 from wopi.services.access import AccessUserItemService
 from wopi.tasks.configure_wopi import WOPI_CONFIGURATION_CACHE_KEY
+from wopi.utils import signature as signature_utils
 
 from drive.settings import Base
 
@@ -278,6 +283,7 @@ def test_check_file_info_supports_rename_override(settings, monkeypatch):
                     "client": "collabora",
                 },
             },
+            "collabora": {"proof_keys": {}},
         },
     )
 
@@ -304,3 +310,244 @@ def test_check_file_info_non_existing_access_token():
         HTTP_AUTHORIZATION="Bearer not_existing_token",
     )
     assert response.status_code == 403
+
+
+def test_check_file_info_connected_user_with_access_with_valid_signature(
+    configure_wopi_clients,
+):
+    """Check signature validation for a connected user with access."""
+    wopi_configuration = configure_wopi_clients
+    folder = factories.ItemFactory(
+        type=models.ItemTypeChoices.FOLDER,
+    )
+    item = factories.ItemFactory(
+        parent=folder,
+        type=models.ItemTypeChoices.FILE,
+        filename="wopi_test.txt",
+        update_upload_state=models.ItemUploadStateChoices.READY,
+        link_reach=models.LinkReachChoices.RESTRICTED,
+        link_role=models.LinkRoleChoices.EDITOR,
+        mimetype="text/plain",
+    )
+    user = factories.UserFactory()
+    factories.UserItemAccessFactory(item=item, user=user, role=models.RoleChoices.EDITOR)
+
+    default_storage.connection.meta.client.put_object(
+        Bucket=default_storage.bucket_name,
+        Key=item.file_key,
+        Body=BytesIO(b"my prose"),
+        ContentType="text/plain",
+    )
+
+    service = AccessUserItemService()
+    access_token, _ = service.insert_new_access(item, user)
+
+    wopi_timestamp = (
+        signature_utils.DOTNET_EPOCH_TICKS + now().timestamp() * signature_utils.TICKS_PER_SECOND
+    )
+
+    client = APIClient()
+    with mock.patch.object(signature_utils, "verify_wopi_proof") as mock_verify_wopi_proof:
+        mock_verify_wopi_proof.return_value = True
+        response = client.get(
+            f"/api/v1.0/wopi/files/{item.id}/",
+            HTTP_AUTHORIZATION=f"Bearer {access_token}",
+            HTTP_X_WOPI_PROOF="valid_signature",
+            HTTP_X_WOPI_TIMESTAMP=wopi_timestamp,
+        )
+
+    assert response.status_code == 200
+
+    mock_verify_wopi_proof.assert_called_once_with(
+        wopi_configuration["vendorA"]["proof_keys"],
+        "valid_signature",
+        None,
+        mock.ANY,
+    )
+
+
+def test_check_file_info_connected_user_with_access_with_invalid_signature(
+    configure_wopi_clients,
+):
+    """Check signature validation for a connected user with access."""
+    wopi_configuration = configure_wopi_clients
+    folder = factories.ItemFactory(
+        type=models.ItemTypeChoices.FOLDER,
+    )
+    item = factories.ItemFactory(
+        parent=folder,
+        type=models.ItemTypeChoices.FILE,
+        filename="wopi_test.txt",
+        update_upload_state=models.ItemUploadStateChoices.READY,
+        link_reach=models.LinkReachChoices.RESTRICTED,
+        link_role=models.LinkRoleChoices.EDITOR,
+        mimetype="text/plain",
+    )
+    user = factories.UserFactory()
+    factories.UserItemAccessFactory(item=item, user=user, role=models.RoleChoices.EDITOR)
+
+    default_storage.connection.meta.client.put_object(
+        Bucket=default_storage.bucket_name,
+        Key=item.file_key,
+        Body=BytesIO(b"my prose"),
+        ContentType="text/plain",
+    )
+
+    service = AccessUserItemService()
+    access_token, _ = service.insert_new_access(item, user)
+
+    wopi_timestamp = (
+        signature_utils.DOTNET_EPOCH_TICKS + now().timestamp() * signature_utils.TICKS_PER_SECOND
+    )
+
+    client = APIClient()
+    with mock.patch.object(signature_utils, "verify_wopi_proof") as mock_verify_wopi_proof:
+        mock_verify_wopi_proof.return_value = False
+        with pytest.raises(WopiRequestSignatureError, match="Invalid request signature"):
+            client.get(
+                f"/api/v1.0/wopi/files/{item.id}/",
+                HTTP_AUTHORIZATION=f"Bearer {access_token}",
+                HTTP_X_WOPI_PROOF="invalid_signature",
+                HTTP_X_WOPI_TIMESTAMP=wopi_timestamp,
+            )
+
+    mock_verify_wopi_proof.assert_called_once_with(
+        wopi_configuration["vendorA"]["proof_keys"],
+        "invalid_signature",
+        None,
+        mock.ANY,
+    )
+
+
+def test_check_file_info_connected_user_with_access_with_expired_wopi_timestamp(
+    configure_wopi_clients,  # pylint: disable=unused-argument
+):
+    """Check signature validation for a connected user with access with expired wopi timestamp."""
+    folder = factories.ItemFactory(
+        type=models.ItemTypeChoices.FOLDER,
+    )
+    item = factories.ItemFactory(
+        parent=folder,
+        type=models.ItemTypeChoices.FILE,
+        filename="wopi_test.txt",
+        update_upload_state=models.ItemUploadStateChoices.READY,
+        link_reach=models.LinkReachChoices.RESTRICTED,
+        link_role=models.LinkRoleChoices.EDITOR,
+        mimetype="text/plain",
+    )
+    user = factories.UserFactory()
+    factories.UserItemAccessFactory(item=item, user=user, role=models.RoleChoices.EDITOR)
+
+    default_storage.connection.meta.client.put_object(
+        Bucket=default_storage.bucket_name,
+        Key=item.file_key,
+        Body=BytesIO(b"my prose"),
+        ContentType="text/plain",
+    )
+
+    service = AccessUserItemService()
+    access_token, _ = service.insert_new_access(item, user)
+
+    wopi_timestamp = (
+        signature_utils.DOTNET_EPOCH_TICKS
+        + (now() - timedelta(minutes=20)).timestamp() * signature_utils.TICKS_PER_SECOND
+    )
+
+    client = APIClient()
+    with mock.patch.object(signature_utils, "verify_wopi_proof") as mock_verify_wopi_proof:
+        with pytest.raises(
+            WopiRequestSignatureError, match="Timestamp is too old, request rejected"
+        ):
+            client.get(
+                f"/api/v1.0/wopi/files/{item.id}/",
+                HTTP_AUTHORIZATION=f"Bearer {access_token}",
+                HTTP_X_WOPI_PROOF="invalid_signature",
+                HTTP_X_WOPI_TIMESTAMP=wopi_timestamp,
+            )
+
+    mock_verify_wopi_proof.assert_not_called()
+
+
+def test_check_file_info_connected_user_with_access_proof_keys_configured_but_no_signature_provided(
+    configure_wopi_clients,  # pylint: disable=unused-argument
+):
+    """Check signature validation for a connected user with access with no signature provided."""
+    folder = factories.ItemFactory(
+        type=models.ItemTypeChoices.FOLDER,
+    )
+    item = factories.ItemFactory(
+        parent=folder,
+        type=models.ItemTypeChoices.FILE,
+        filename="wopi_test.txt",
+        update_upload_state=models.ItemUploadStateChoices.READY,
+        link_reach=models.LinkReachChoices.RESTRICTED,
+        link_role=models.LinkRoleChoices.EDITOR,
+        mimetype="text/plain",
+    )
+    user = factories.UserFactory()
+    factories.UserItemAccessFactory(item=item, user=user, role=models.RoleChoices.EDITOR)
+
+    default_storage.connection.meta.client.put_object(
+        Bucket=default_storage.bucket_name,
+        Key=item.file_key,
+        Body=BytesIO(b"my prose"),
+        ContentType="text/plain",
+    )
+
+    service = AccessUserItemService()
+    access_token, _ = service.insert_new_access(item, user)
+
+    client = APIClient()
+    with mock.patch.object(signature_utils, "verify_wopi_proof") as mock_verify_wopi_proof:
+        with pytest.raises(
+            WopiRequestSignatureError, match="No signature provided, request rejected"
+        ):
+            client.get(
+                f"/api/v1.0/wopi/files/{item.id}/",
+                HTTP_AUTHORIZATION=f"Bearer {access_token}",
+            )
+
+    mock_verify_wopi_proof.assert_not_called()
+
+
+def test_check_file_info_connected_user_with_access_proof_keys_configured_but_no_timestamp_provided(
+    configure_wopi_clients,  # pylint: disable=unused-argument
+):
+    """Check signature validation for a connected user with access with no timestamp provided."""
+    folder = factories.ItemFactory(
+        type=models.ItemTypeChoices.FOLDER,
+    )
+    item = factories.ItemFactory(
+        parent=folder,
+        type=models.ItemTypeChoices.FILE,
+        filename="wopi_test.txt",
+        update_upload_state=models.ItemUploadStateChoices.READY,
+        link_reach=models.LinkReachChoices.RESTRICTED,
+        link_role=models.LinkRoleChoices.EDITOR,
+        mimetype="text/plain",
+    )
+    user = factories.UserFactory()
+    factories.UserItemAccessFactory(item=item, user=user, role=models.RoleChoices.EDITOR)
+
+    default_storage.connection.meta.client.put_object(
+        Bucket=default_storage.bucket_name,
+        Key=item.file_key,
+        Body=BytesIO(b"my prose"),
+        ContentType="text/plain",
+    )
+
+    service = AccessUserItemService()
+    access_token, _ = service.insert_new_access(item, user)
+
+    client = APIClient()
+    with mock.patch.object(signature_utils, "verify_wopi_proof") as mock_verify_wopi_proof:
+        with pytest.raises(
+            WopiRequestSignatureError, match="No timestamp provided, request rejected"
+        ):
+            client.get(
+                f"/api/v1.0/wopi/files/{item.id}/",
+                HTTP_AUTHORIZATION=f"Bearer {access_token}",
+                HTTP_X_WOPI_PROOF="invalid_signature",
+            )
+
+    mock_verify_wopi_proof.assert_not_called()

src/backend/wopi/utils/signature.py

@@ -14,6 +14,14 @@ from cryptography.hazmat.primitives.asymmetric import padding
 
 # ---------- HELPERS ----------
 
+# .NET DateTime epoch: January 1, 0001 00:00:00
+# Unix datetime epoch: January 1, 1970 00:00:00 UTC
+# Ticks between these epochs: 621,355,968,000,000,000
+DOTNET_EPOCH_TICKS = 621355968000000000
+
+# Each tick is 100 nanoseconds = 0.0000001 seconds
+TICKS_PER_SECOND = 10000000
+
 
 def ticks_to_datetime(ticks: int) -> datetime:
     """Convert a .NET DateTime ticks to a datetime object.
@@ -27,16 +35,9 @@ def ticks_to_datetime(ticks: int) -> datetime:
     Returns:
         datetime: Python datetime object corresponding to the ticks
     """
-    # .NET DateTime epoch: January 1, 0001 00:00:00
-    # Unix datetime epoch: January 1, 1970 00:00:00 UTC
-    # Ticks between these epochs: 621,355,968,000,000,000
-    dotnet_epoch_ticks = 621355968000000000
-
-    # Each tick is 100 nanoseconds = 0.0000001 seconds
-    ticks_per_second = 10000000
 
     # Convert .NET ticks to Unix timestamp (seconds since 1970-01-01)
-    unix_seconds = (ticks - dotnet_epoch_ticks) / ticks_per_second
+    unix_seconds = (ticks - DOTNET_EPOCH_TICKS) / TICKS_PER_SECOND
 
     # Create datetime from Unix timestamp (UTC)
     return datetime.fromtimestamp(unix_seconds, timezone.utc)

src/backend/wopi/viewsets.py

@@ -2,6 +2,7 @@
 
 import logging
 import uuid
+from datetime import timedelta
 from os.path import splitext
 
 from django.conf import settings
@@ -10,6 +11,7 @@ from django.core.files.base import ContentFile
 from django.core.files.storage import default_storage
 from django.db import transaction
 from django.http import StreamingHttpResponse
+from django.utils.timezone import now
 
 from lasuite.malware_detection import malware_detection
 from rest_framework import viewsets
@@ -19,10 +21,16 @@ from sentry_sdk import capture_exception
 
 from core.api.utils import get_item_file_head_object
 from core.models import Item
-from wopi.authentication import WopiAccessTokenAuthentication
+from wopi.authentication import WopiAccessTokenAuthentication, get_access_token
+from wopi.exceptions import WopiRequestSignatureError
 from wopi.permissions import AccessTokenPermission
 from wopi.services.lock import LockService
-from wopi.utils import get_wopi_client_config, get_wopi_item_version
+from wopi.utils import (
+    get_wopi_client_config,
+    get_wopi_client_proof_keys,
+    get_wopi_item_version,
+    signature,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -30,6 +38,9 @@ logger = logging.getLogger(__name__)
 HTTP_X_WOPI_LOCK = "HTTP_X_WOPI_LOCK"
 HTTP_X_WOPI_OLD_LOCK = "HTTP_X_WOPI_OLDLOCK"
 HTTP_X_WOPI_OVERRIDE = "HTTP_X_WOPI_OVERRIDE"
+HTTP_X_WOPI_TIMESTAMP = "HTTP_X_WOPI_TIMESTAMP"
+HTTP_X_WOPI_PROOF = "HTTP_X_WOPI_PROOF"
+HTTP_X_WOPI_PROOFOLD = "HTTP_X_WOPI_PROOFOLD"
 
 X_WOPI_INVALIDFILENAMERROR = "X-WOPI-InvalidFileNameError"
 X_WOPI_ITEMVERSION = "X-WOPI-ItemVersion"
@@ -60,6 +71,48 @@ class WopiViewSet(viewsets.ViewSet):
         """Get the file id from the URL path."""
         return uuid.UUID(self.kwargs.get("pk"))
 
+    def _verify_request_signature(self, request):
+        """Verify the request signature."""
+        proof_keys = get_wopi_client_proof_keys(request.auth.item, request.user)
+
+        if not proof_keys:
+            # The proof key is not provided by the wopi client,
+            # so we can't verify the request signature and the request is accepted
+            return
+
+        request_signature = request.META.get(HTTP_X_WOPI_PROOF)
+        request_signature_old = request.META.get(HTTP_X_WOPI_PROOFOLD)
+
+        if not request_signature:
+            raise WopiRequestSignatureError("No signature provided, request rejected")
+
+        string_timestamp = request.META.get(HTTP_X_WOPI_TIMESTAMP)
+        if not string_timestamp:
+            raise WopiRequestSignatureError("No timestamp provided, request rejected")
+        try:
+            timestamp = int(string_timestamp)
+        except ValueError as e:
+            raise WopiRequestSignatureError("Invalid timestamp provided") from e
+
+        datetime_timestamp = signature.ticks_to_datetime(timestamp)
+        if datetime_timestamp < now() - timedelta(minutes=20):
+            raise WopiRequestSignatureError("Timestamp is too old, request rejected")
+
+        access_token = get_access_token(request)
+        expected_proof = signature.build_expected_proof(
+            access_token,
+            request.build_absolute_uri(),
+            timestamp,
+        )
+
+        if not signature.verify_wopi_proof(
+            proof_keys,
+            request_signature,
+            request_signature_old,
+            expected_proof,
+        ):
+            raise WopiRequestSignatureError("Invalid request signature")
+
     # pylint: disable=unused-argument
     def retrieve(self, request, pk=None):
         """
@@ -69,6 +122,8 @@ class WopiViewSet(viewsets.ViewSet):
         item = request.auth.item
         abilities = item.get_abilities(request.user)
 
+        self._verify_request_signature(request)
+
         head_object = get_item_file_head_object(item)
         wopi_client = get_wopi_client_config(item, request.user)
         client_options = {}
@@ -111,6 +166,7 @@ class WopiViewSet(viewsets.ViewSet):
         """
         Operations to get or put the file content.
         """
+        self._verify_request_signature(request)
         if request.method == "GET":
             return self._get_file_content(request, pk)
         if request.method == "POST":
@@ -214,6 +270,9 @@ class WopiViewSet(viewsets.ViewSet):
         """
         if not request.META.get(HTTP_X_WOPI_OVERRIDE) in self.detail_post_actions:
             return Response(status=404)
+
+        self._verify_request_signature(request)
+
         item = request.auth.item
         abilities = item.get_abilities(request.user)