wcs: add support for linking/displaying card file fields (#51994)

combo/apps/wcs/models.py

@@ -996,7 +996,7 @@ class WcsCardInfosCell(CardMixin, CellBase):
         if not card_id:
             return extra_context
         card_slug = self.carddef_reference.split(':')[1]
-        api_url = '/api/cards/%s/%s/' % (card_slug, card_id)
+        api_url = '/api/cards/%s/%s/?include-files-content=off' % (card_slug, card_id)
 
         wcs_site = get_wcs_services().get(self.wcs_site)
 

combo/apps/wcs/templates/combo/wcs/card-field-value.html

@@ -1,4 +1,4 @@
-{% spaceless %}
+{% load combo wcs %}{% spaceless %}
 {% if field.type == "text" and mode != 'inline' and value %}
 <div class="value">{{ field|format_text:value }}</div>
 {% else %}
@@ -7,6 +7,12 @@
   {{ value|date }}
 {% elif field.type == "bool" and value is not None %}
   {{ value|yesno }}
+{% elif field.type == 'file' and value %}
+  {% if value.content_type|startswith:"image/" %}
+    <img alt="" src="{% make_public_url url=value.url %}">
+  {% else %}
+    <a href="{% make_public_url url=value.url %}" download="{{ value.filename }}">{{ value.filename }}</a>
+  {% endif %}
 {% else %}
   {{ value|default:"" }}
 {% endif %}

combo/apps/wcs/templates/combo/wcs/card.html

@@ -35,7 +35,7 @@
 
 {% else %}
 {% for field in schema.fields %}
-  {% if 'varname' in field and field.varname and field.type != 'file' %}
+  {% if 'varname' in field and field.varname %}
   {% with card.fields|get:field.varname as value %}
   <div class="card--auto-field">
     <span class="label">{{ field.label }}</span>

combo/apps/wcs/templatetags/wcs.py

@@ -15,9 +15,14 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 from django import template
+from django.conf import settings
+from django.urls import reverse
+from django.utils.encoding import force_bytes
 from django.utils.html import escape
 from django.utils.safestring import mark_safe
 
+from combo.utils import aes_hex_encrypt
+
 register = template.Library()
 
 
@@ -71,3 +76,17 @@ def format_text(field, value):
     if field.get('pre'):
         return mark_safe('<pre>%s</pre>' % escape(value))
     return mark_safe('<p>' + '\n'.join([(escape(x) or '</p><p>') for x in value.splitlines()]) + '</p>')
+
+
+@register.simple_tag(takes_context=True)
+def make_public_url(context, url):
+    if not context.request.session.session_key:
+        context.request.session.cycle_key()
+    session_key = context.request.session.session_key
+    return reverse(
+        'wcs-redirect-crypto-url',
+        kwargs={
+            'session_key': session_key,
+            'crypto_url': aes_hex_encrypt(settings.SECRET_KEY, force_bytes(url)),
+        },
+    )

combo/apps/wcs/urls.py

@@ -16,9 +16,14 @@
 
 from django.conf.urls import url
 
-from .views import TrackingCodeView, tracking_code_search
+from .views import TrackingCodeView, redirect_crypto_url, tracking_code_search
 
 urlpatterns = [
     url(r'^tracking-code/$', TrackingCodeView.as_view(), name='wcs-tracking-code'),
     url(r'^api/search/tracking-code/$', tracking_code_search, name='wcs-tracking-code-search'),
+    url(
+        r'^api/wcs/file/(?P<session_key>\w+)/(?P<crypto_url>[\w,-]+)/$',
+        redirect_crypto_url,
+        name='wcs-redirect-crypto-url',
+    ),
 ]

combo/apps/wcs/views.py

@@ -21,14 +21,14 @@ import ratelimit.utils
 from django.conf import settings
 from django.contrib import messages
 from django.core.exceptions import DisallowedRedirect, PermissionDenied
-from django.http import HttpResponseBadRequest, HttpResponseRedirect, JsonResponse
+from django.http import HttpResponseBadRequest, HttpResponseForbidden, HttpResponseRedirect, JsonResponse
 from django.utils.http import urlquote
 from django.utils.translation import ugettext_lazy as _
 from django.views.decorators.csrf import csrf_exempt
 from django.views.generic import View
 
-from combo.utils import requests
-from combo.utils.misc import is_url_from_known_service
+from combo.utils import DecryptionError, aes_hex_decrypt, requests, sign_url
+from combo.utils.misc import get_known_service_for_url, is_url_from_known_service
 
 from .models import TrackingCodeInputCell
 from .utils import get_wcs_services
@@ -133,3 +133,19 @@ def tracking_code_search(request):
                     }
                 )
     return JsonResponse(response)
+
+
+def redirect_crypto_url(request, session_key, crypto_url):
+    if session_key != request.session.session_key:
+        return HttpResponseForbidden()
+    try:
+        real_url = aes_hex_decrypt(settings.SECRET_KEY, crypto_url)
+    except DecryptionError:
+        return HttpResponseForbidden('invalid crypto url')
+
+    service = get_known_service_for_url(real_url)
+    if '?' not in real_url:
+        real_url += '?'
+    real_url += '&orig=%s' % service['orig']
+    redirect_url = sign_url(real_url, service['secret'])
+    return HttpResponseRedirect(redirect_url)

combo/utils/misc.py

@@ -41,13 +41,18 @@ def flatten_context(context):
     return flat_context
 
 
+def get_known_service_for_url(url):
+    netloc = urllib.parse.urlparse(url).netloc
+    for services in settings.KNOWN_SERVICES.values():
+        for service in services.values():
+            remote_url = service.get('url')
+            if urllib.parse.urlparse(remote_url).netloc == netloc:
+                return service
+    return None
+
+
 def is_url_from_known_service(url):
     netloc = urllib.parse.urlparse(url).netloc
     if not netloc:
         return True
-    for service_id in settings.KNOWN_SERVICES or {}:
-        for service_key in settings.KNOWN_SERVICES[service_id]:
-            service = settings.KNOWN_SERVICES[service_id][service_key]
-            if urllib.parse.urlparse(service.get('url')).netloc == netloc:
-                return True
-    return False
+    return bool(get_known_service_for_url(url))

combo/utils/requests_wrapper.py

@@ -27,6 +27,7 @@ from requests import Response
 from requests import Session as RequestsSession
 from requests.auth import AuthBase
 
+from .misc import get_known_service_for_url
 from .signature import sign_url
 
 
@@ -59,23 +60,11 @@ class Requests(RequestsSession):
         self.cookies.clear()
 
         if remote_service == 'auto':
-            remote_service = None
-            scheme, netloc, path, params, query, fragment = urllib.parse.urlparse(url)
-            for services in settings.KNOWN_SERVICES.values():
-                for service in services.values():
-                    remote_url = service.get('url')
-                    remote_scheme, remote_netloc, dummy, dummy, dummy, dummy = urllib.parse.urlparse(
-                        remote_url
-                    )
-                    if remote_scheme == scheme and remote_netloc == netloc:
-                        remote_service = service
-                        break
-                else:
-                    continue
-                break
+            remote_service = get_known_service_for_url(url)
             if remote_service:
                 # only keeps the path (URI) in url parameter, scheme and netloc are
                 # in remote_service
+                scheme, netloc, path, params, query, fragment = urllib.parse.urlparse(url)
                 url = urllib.parse.urlunparse(('', '', path, params, query, fragment))
             else:
                 logging.warning('service not found in settings.KNOWN_SERVICES for %s', url)

tests/test_wcs.py

@@ -2,6 +2,7 @@ import copy
 import json
 import re
 import urllib.parse
+from importlib import import_module
 from unittest import mock
 
 import pytest
@@ -195,7 +196,7 @@ WCS_CARD_DATA = {
         'fielda': 'a',
         'fieldb': True,
         'fieldc': '2020-09-28',
-        'fieldd': {'filename': 'file.pdf', 'url': 'http://some-url.com/download?f=42'},
+        'fieldd': {'filename': 'file.pdf', 'url': 'http://127.0.0.1:8999/download?f=42'},
         'fielde': 'lorem<strong>ipsum\n\nhello world',
         'fieldf': 'lorem<strong>ipsum\n\nhello world',
         'related': 'Foo Bar',
@@ -279,7 +280,8 @@ def mocked_requests_send(request, **kwargs):
 def context():
     ctx = {'request': RequestFactory().get('/')}
     ctx['request'].user = None
-    ctx['request'].session = {}
+    session_engine = import_module(settings.SESSION_ENGINE)
+    ctx['request'].session = session_engine.SessionStore()
     return ctx
 
 
@@ -1843,7 +1845,7 @@ def test_card_cell_render(mock_send, context):
     assert PyQuery(result).find('span.label:contains("Related") + span').text() == 'Foo Bar'
     assert 'related_raw' not in result
     assert 'related_structured' not in result
-    assert 'Field D' not in result
+    assert PyQuery(result).find('span.label:contains("Field D") + span a').text() == 'file.pdf'
 
     context.pop('title')
     cell.title_type = 'manual'
@@ -2561,3 +2563,26 @@ def test_search_external_forms_links(mock_send, context):
     request.user = AnonymousUser()
     hits = search_site(request, 'form')
     assert len(hits) == 2
+
+
+@mock.patch('combo.apps.wcs.utils.requests.send', side_effect=mocked_requests_send)
+def test_card_file_redirection(mock_send, app):
+    page = Page(title='One', slug='one', template_name='standard')
+    page.save()
+    cell = WcsCardInfosCell(page=page, placeholder='content', order=0)
+    cell.carddef_reference = 'default:card_model_1'
+    cell.card_id = '11'
+    cell.save()
+    resp = app.get('/one/')
+    ajax_cell_url = PyQuery(resp.text).find('[data-ajax-cell-url]').attr['data-ajax-cell-url']
+    resp = app.get(ajax_cell_url)
+    file_url = PyQuery(resp.text).find('[download]').attr['href']
+    resp = app.get(file_url)
+    assert 'download?f=42' in resp.location
+    assert '&signature=' in resp.location
+
+    # invalid crypto
+    resp = app.get(file_url[:-2] + 'X/', status=403)
+
+    # invalid session key
+    resp = app.get(file_url.replace('file/', 'file/X'), status=403)