From b57af0df37afd16fe86370358dd05e6481747d2c 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 | 112 +++++++++++++++++++++++++++-----------
 wcs/api.py                |  37 +++++++------
 2 files changed, 100 insertions(+), 49 deletions(-)

diff --git a/tests/api/test_formdef.py b/tests/api/test_formdef.py
index 79c47f73..b802237a 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()
@@ -492,22 +526,20 @@ def test_formdef_submit(pub, local_user):
     formdef = FormDef()
     formdef.name = 'test'
     formdef.fields = [fields.StringField(id='0', label='foobar')]
+    formdef.backoffice_submission_roles = [role.id]
     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'
+    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)
 
-    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') :]
+    if auth == 'http-basic':
+        resp = post_json_url('/api/formdefs/test/submit', params={'data': {}}, status=403)
+        access.roles = [role]
+        access.store()
 
-    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 +547,78 @@ 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.backoffice_submission_roles = []
     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 +696,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..28f8d200 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()()
 
@@ -549,7 +549,7 @@ class ApiFormdefDirectory(Directory):
         formdata.data = posted_json_data_to_formdata_data(self.formdef, data)
 
         meta = json_input.get('meta') or {}
-        if meta.get('backoffice-submission'):
+        if meta.get('backoffice-submission') or user and user.is_api_user:
             if not user:
                 raise AccessForbiddenError('no user set for backoffice submission')
             if not self.formdef.backoffice_submission_roles:
@@ -557,21 +557,22 @@ class ApiFormdefDirectory(Directory):
             if not set(user.get_roles()).intersection(self.formdef.backoffice_submission_roles):
                 raise AccessForbiddenError('not cleared for backoffice submit')
             formdata.backoffice_submission = True
-        elif 'user' in json_input:
-            formdata_user = None
-            for name_id in json_input['user'].get('NameID') or []:
-                formdata_user = get_publisher().user_class.get_users_with_name_identifier(name_id)
+        if not meta.get('backoffice-submission'):
+            if 'user' in json_input:
+                formdata_user = None
+                for name_id in json_input['user'].get('NameID') or []:
+                    formdata_user = get_publisher().user_class.get_users_with_name_identifier(name_id)
+                    if formdata_user:
+                        break
+                else:
+                    if json_input['user'].get('email'):
+                        formdata_user = get_publisher().user_class.get_users_with_email(
+                            json_input['user'].get('email')
+                        )
                 if formdata_user:
-                    break
-            else:
-                if json_input['user'].get('email'):
-                    formdata_user = get_publisher().user_class.get_users_with_email(
-                        json_input['user'].get('email')
-                    )
-            if formdata_user:
-                formdata.user_id = formdata_user[0].id
-        elif user and not user.is_api_user:
-            formdata.user_id = user.id
+                    formdata.user_id = formdata_user[0].id
+            elif user and not user.is_api_user:
+                formdata.user_id = user.id
 
         if json_input.get('context'):
             formdata.submission_context = json_input['context']
@@ -584,7 +585,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.33.0

