✨(backend) build short-lived WOPI source URLs

OnlyOffice must fetch the source bytes from Drive during server-side conversion.

Use short-lived WOPI access URLs so conversion does not depend on direct object
storage reachability.

docker/onlyoffice/local-development.json

@@ -18,5 +18,12 @@
     "converter": {
       "maxDownloadBytes": 209715200
     }
+  },
+  "services": {
+    "CoAuthoring": {
+      "request-filtering-agent": {
+        "allowPrivateIPAddress": true
+      }
+    }
   }
 }

src/backend/drive/settings.py

@@ -1395,6 +1395,9 @@ class Base(Configuration):
     WOPI_ACCESS_TOKEN_TIMEOUT = values.IntegerValue(
         60 * 60 * 10, environ_name="WOPI_ACCESS_TOKEN_TIMEOUT", environ_prefix=None
     )
+    WOPI_CONVERSION_SOURCE_TOKEN_TIMEOUT = values.IntegerValue(
+        120, environ_name="WOPI_CONVERSION_SOURCE_TOKEN_TIMEOUT", environ_prefix=None
+    )
     WOPI_LOCK_TIMEOUT = values.IntegerValue(
         30 * 60, environ_name="WOPI_LOCK_TIMEOUT", environ_prefix=None
     )

src/backend/wopi/conversion/source_url.py

@@ -0,0 +1,22 @@
+"""Build the short-lived WOPI URL OnlyOffice uses to fetch the source bytes."""
+
+from django.conf import settings
+
+from wopi.conversion.exceptions import ConversionMisconfigured
+from wopi.services.access import AccessUserItemService
+
+
+def build_source_url(item, user):
+    """Return a short-lived WOPI GetFile URL pointing at the item for the user."""
+    base_url = settings.WOPI_SRC_BASE_URL
+
+    if not base_url:
+        raise ConversionMisconfigured("Missing WOPI_SRC_BASE_URL for conversion source URL")
+
+    access_token, _ttl_ms = AccessUserItemService().insert_new_access(
+        item, user, ttl=settings.WOPI_CONVERSION_SOURCE_TOKEN_TIMEOUT
+    )
+    return (
+        f"{base_url.rstrip('/')}/api/{settings.API_VERSION}"
+        f"/wopi/files/{item.id}/contents/?access_token={access_token}"
+    )

src/backend/wopi/services/access.py

@@ -68,22 +68,25 @@ class AccessUserItemService:
         """Generate a random access token"""
         return token_urlsafe()
 
-    def insert_new_access(self, item: Item, user: AbstractUser) -> tuple[str, int]:
+    def insert_new_access(
+        self, item: Item, user: AbstractUser, ttl: int | None = None
+    ) -> 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
+        access_token_ttl must be a timestamp in milliseconds.
+
+        ttl overrides the default WOPI_ACCESS_TOKEN_TIMEOUT lifetime. Pass a short
+        ttl for one-shot uses (server-to-server conversion source download, etc.).
         """
         abilities = item.get_abilities(user)
         if not abilities["retrieve"]:
             raise AccessUserItemNotAllowed()
+
+        effective_ttl = ttl if ttl is not None else settings.WOPI_ACCESS_TOKEN_TIMEOUT
         token = self.generate_token()
         access_user_item = AccessUserItem(item=item, user=user)
-        token_eol = timezone.now() + timedelta(seconds=settings.WOPI_ACCESS_TOKEN_TIMEOUT)
-        cache.set(
-            token,
-            access_user_item.to_dict(),
-            timeout=settings.WOPI_ACCESS_TOKEN_TIMEOUT,
-        )
+        token_eol = timezone.now() + timedelta(seconds=effective_ttl)
+        cache.set(token, access_user_item.to_dict(), timeout=effective_ttl)
         return token, int(round(token_eol.timestamp())) * 1000
 
     def get_access_user_item(self, token: str) -> AccessUserItem:

src/backend/wopi/tests/conversion/test_source_url.py

@@ -0,0 +1,64 @@
+"""Tests for the WOPI source-URL builder used by the conversion service."""
+
+from unittest import mock
+
+import pytest
+
+from core import factories
+from wopi.conversion import exceptions
+from wopi.conversion.source_url import build_source_url
+
+
+@pytest.fixture
+def _access_service():
+    """Mock the AccessUserItemService used by build_source_url."""
+    service = mock.Mock()
+    service.return_value.insert_new_access.return_value = ("tok-abc", 1_700_000_000_000)
+    with mock.patch("wopi.conversion.source_url.AccessUserItemService", service):
+        yield service
+
+
+def test_build_source_url_uses_configured_wopi_base_url(settings, _access_service):
+    """Start the built URL with WOPI_SRC_BASE_URL and embed the access token."""
+    settings.WOPI_SRC_BASE_URL = "https://drive.example"
+    item = factories.ItemFactory.build()
+    user = factories.UserFactory.build()
+
+    url = build_source_url(item, user)
+
+    assert url == (
+        f"https://drive.example/api/v1.0/wopi/files/{item.id}/contents/?access_token=tok-abc"
+    )
+
+
+def test_build_source_url_strips_trailing_slash_on_base_url(settings, _access_service):
+    """Strip a trailing slash on the base URL to avoid a double slash."""
+    settings.WOPI_SRC_BASE_URL = "https://drive.example/"
+    item = factories.ItemFactory.build()
+    user = factories.UserFactory.build()
+
+    url = build_source_url(item, user)
+
+    assert "//api" not in url
+
+
+def test_build_source_url_raises_when_base_url_is_missing(settings, _access_service):
+    """Raise when WOPI_SRC_BASE_URL is missing."""
+    settings.WOPI_SRC_BASE_URL = None
+    item = factories.ItemFactory.build()
+    user = factories.UserFactory.build()
+
+    with pytest.raises(exceptions.ConversionMisconfigured, match="Missing WOPI_SRC_BASE_URL"):
+        build_source_url(item, user)
+
+
+def test_build_source_url_delegates_to_access_user_item_service(settings, _access_service):
+    """Issue a short-lived WOPI access token via the existing service."""
+    settings.WOPI_SRC_BASE_URL = "https://drive.example"
+    settings.WOPI_CONVERSION_SOURCE_TOKEN_TIMEOUT = 90
+    item = factories.ItemFactory.build()
+    user = factories.UserFactory.build()
+
+    build_source_url(item, user)
+
+    _access_service.return_value.insert_new_access.assert_called_once_with(item, user, ttl=90)