From 7e884e68b3de5f9901f63af4a8063ea4ea83898d Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Laur=C3=A9line=20Gu=C3=A9rin?= <zebuline@entrouvert.com>
Date: Tue, 14 Sep 2021 15:49:48 +0200
Subject: [PATCH] api: add basic auth support to forms submit (#55858)

---
 tests/api/test_formdef.py | 107 ++++++++++++++++++++++++++------------
 wcs/api.py                |   6 +--
 2 files changed, 78 insertions(+), 35 deletions(-)

diff --git a/tests/api/test_formdef.py b/tests/api/test_formdef.py
index 79c47f73..fa0b9194 100644
--- a/tests/api/test_formdef.py
+++ b/tests/api/test_formdef.py
@@ -5,6 +5,7 @@ import json
 import os
 import time
 import urllib.parse
+from functools import partial
 from unittest import mock
 
 import pytest
@@ -12,6 +13,7 @@ from django.utils.encoding import force_text
 from quixote import get_publisher
 
 from wcs import fields, qommon
+from wcs.api_access import ApiAccess
 from wcs.api_utils import sign_url
 from wcs.categories import Category
 from wcs.data_sources import NamedDataSource
@@ -66,6 +68,37 @@ def local_user():
     return user
 
 
+def _post_url(url, app, auth, access, user, **kwargs):
+    if auth == 'http-basic':
+        app.set_authorization(('Basic', ('test', '12345')))
+        return app.post(url, **kwargs)
+    else:
+        return app.post(
+            sign_uri(url, user=user, orig=access.access_identifier, key=access.access_key), **kwargs
+        )
+
+
+def _post_json_url(url, app, auth, access, user, **kwargs):
+    if auth == 'http-basic':
+        app.set_authorization(('Basic', ('test', '12345')))
+        return app.post_json(url, **kwargs)
+    else:
+        return app.post_json(
+            sign_uri(url, user=user, orig=access.access_identifier, key=access.access_key), **kwargs
+        )
+
+
+@pytest.fixture
+def access(pub):
+    ApiAccess.wipe()
+    access = ApiAccess()
+    access.name = 'test'
+    access.access_identifier = 'test'
+    access.access_key = '12345'
+    access.store()
+    return access
+
+
 def test_formdef_list(pub):
     pub.role_class.wipe()
     role = pub.role_class(name='Foo bar')
@@ -481,7 +514,8 @@ def test_post_invalid_json(pub, local_user):
     assert resp.json['err_class'] == 'Invalid request'
 
 
-def test_formdef_submit(pub, local_user):
+@pytest.mark.parametrize('auth', ['signature', 'http-basic'])
+def test_formdef_submit(pub, local_user, access, auth):
     pub.role_class.wipe()
     role = pub.role_class(name='test')
     role.store()
@@ -495,19 +529,11 @@ def test_formdef_submit(pub, local_user):
     formdef.store()
     data_class = formdef.data_class()
 
-    resp = get_app(pub).post_json('/api/formdefs/test/submit', {'data': {}}, status=403)
-    assert resp.json['err'] == 1
-    assert resp.json['err_desc'] == 'unsigned API call'
-
-    def url():
-        signed_url = sign_url(
-            'http://example.net/api/formdefs/test/submit'
-            + '?format=json&orig=coucou&email=%s' % urllib.parse.quote(local_user.email),
-            '1234',
-        )
-        return signed_url[len('http://example.net') :]
+    app = get_app(pub)
+    post_url = partial(_post_url, app=app, auth=auth, access=access, user=local_user)
+    post_json_url = partial(_post_json_url, app=app, auth=auth, access=access, user=local_user)
 
-    resp = get_app(pub).post_json(url(), {'data': {}})
+    resp = post_json_url('/api/formdefs/test/submit', params={'data': {}})
     assert resp.json['err'] == 0
     assert resp.json['data']['url'] == ('http://example.net/test/%s/' % resp.json['data']['id'])
     assert resp.json['data']['backoffice_url'] == (
@@ -515,56 +541,77 @@ def test_formdef_submit(pub, local_user):
     )
     assert resp.json['data']['api_url'] == ('http://example.net/api/forms/test/%s/' % resp.json['data']['id'])
     assert data_class.get(resp.json['data']['id']).status == 'wf-new'
-    assert data_class.get(resp.json['data']['id']).user_id == str(local_user.id)
+    if auth == 'signature':
+        assert data_class.get(resp.json['data']['id']).user_id == str(local_user.id)
+    else:
+        assert data_class.get(resp.json['data']['id']).user_id is None
     assert data_class.get(resp.json['data']['id']).tracking_code is None
 
     local_user2 = get_publisher().user_class()
     local_user2.name = 'Test'
     local_user2.email = 'foo@localhost'
     local_user2.store()
-    resp = get_app(pub).post_json(url(), {'data': {}, 'user': {'NameID': [], 'email': local_user2.email}})
+    resp = post_json_url(
+        '/api/formdefs/test/submit', params={'data': {}, 'user': {'NameID': [], 'email': local_user2.email}}
+    )
     assert data_class.get(resp.json['data']['id']).user.email == local_user2.email
 
-    resp = get_app(pub).post(
-        url(), json.dumps({'data': {}}), status=400
+    resp = post_url(
+        '/api/formdefs/test/submit', params=json.dumps({'data': {}}), status=400
     )  # missing Content-Type: application/json header
     assert resp.json['err_desc'] == 'expected JSON but missing appropriate content-type'
 
     # check qualified content type are recognized
-    resp = get_app(pub).post(url(), json.dumps({'data': {}}), content_type='application/json; charset=utf-8')
+    resp = post_url(
+        '/api/formdefs/test/submit',
+        params=json.dumps({'data': {}}),
+        content_type='application/json; charset=utf-8',
+    )
     assert resp.json['data']['url']
 
     formdef.disabled = True
     formdef.store()
-    resp = get_app(pub).post_json(url(), {'data': {}}, status=403)
+    resp = post_json_url('/api/formdefs/test/submit', params={'data': {}}, status=403)
     assert resp.json['err'] == 1
     assert resp.json['err_desc'] == 'disabled form'
 
     formdef.disabled = False
     formdef.store()
-    resp = get_app(pub).post_json(url(), {'meta': {'backoffice-submission': True}, 'data': {}}, status=403)
+    resp = post_json_url(
+        '/api/formdefs/test/submit', params={'meta': {'backoffice-submission': True}, 'data': {}}, status=403
+    )
     formdef.backoffice_submission_roles = ['xx']
     formdef.store()
-    resp = get_app(pub).post_json(url(), {'meta': {'backoffice-submission': True}, 'data': {}}, status=403)
+    resp = post_json_url(
+        '/api/formdefs/test/submit', params={'meta': {'backoffice-submission': True}, 'data': {}}, status=403
+    )
     formdef.backoffice_submission_roles = [role.id]
     formdef.store()
-    resp = get_app(pub).post_json(url(), {'meta': {'backoffice-submission': True}, 'data': {}})
+    if auth == 'http-basic':
+        access.roles = [role]
+        access.store()
+    resp = post_json_url(
+        '/api/formdefs/test/submit', params={'meta': {'backoffice-submission': True}, 'data': {}}
+    )
     assert data_class.get(resp.json['data']['id']).status == 'wf-new'
     assert data_class.get(resp.json['data']['id']).backoffice_submission is True
     assert data_class.get(resp.json['data']['id']).user_id is None
-    assert data_class.get(resp.json['data']['id']).submission_agent_id == str(local_user.id)
+    if auth == 'signature':
+        assert data_class.get(resp.json['data']['id']).submission_agent_id == str(local_user.id)
+    else:
+        assert data_class.get(resp.json['data']['id']).submission_agent_id is None
 
     formdef.enable_tracking_codes = True
     formdef.store()
-    resp = get_app(pub).post_json(url(), {'data': {}})
+    resp = post_json_url('/api/formdefs/test/submit', params={'data': {}})
     assert data_class.get(resp.json['data']['id']).tracking_code
 
-    resp = get_app(pub).post_json(url(), {'meta': {'draft': True}, 'data': {}})
+    resp = post_json_url('/api/formdefs/test/submit', params={'meta': {'draft': True}, 'data': {}})
     assert data_class.get(resp.json['data']['id']).status == 'draft'
 
-    resp = get_app(pub).post_json(
-        url(),
-        {
+    resp = post_json_url(
+        '/api/formdefs/test/submit',
+        params={
             'meta': {'backoffice-submission': True},
             'data': {},
             'context': {'channel': 'mail', 'comments': 'blah'},
@@ -642,10 +689,6 @@ def test_formdef_submit_with_varname(pub, local_user):
     formdef.store()
     data_class = formdef.data_class()
 
-    resp = get_app(pub).post_json('/api/formdefs/test/submit', {'data': {}}, status=403)
-    assert resp.json['err'] == 1
-    assert resp.json['err_desc'] == 'unsigned API call'
-
     signed_url = sign_url(
         'http://example.net/api/formdefs/test/submit'
         + '?format=json&orig=coucou&email=%s' % urllib.parse.quote(local_user.email),
diff --git a/wcs/api.py b/wcs/api.py
index 70f2b0bc..affd1056 100644
--- a/wcs/api.py
+++ b/wcs/api.py
@@ -525,9 +525,9 @@ class ApiFormdefDirectory(Directory):
         get_response().set_content_type('application/json')
         if self.formdef.is_disabled():
             raise AccessForbiddenError('disabled form')
-        if not is_url_signed():
-            raise AccessForbiddenError('unsigned API call')
         user = get_user_from_api_query_string()
+        if user and user.is_api_user:
+            pass  # API users are ok
         json_input = get_request().json
         formdata = self.formdef.data_class()()
 
@@ -584,7 +584,7 @@ class ApiFormdefDirectory(Directory):
             if [x for x in user_forms if not x.is_draft()]:
                 raise AccessForbiddenError('only one formdata by user is allowed')
 
-        if meta.get('backoffice-submission'):
+        if meta.get('backoffice-submission') and not user.is_api_user:
             # keep track of the agent that did the submit
             formdata.submission_agent_id = str(user.id)
 
-- 
2.30.2

