tests/api/test_formdata.py

@@ -106,8 +106,39 @@ def ics_data(local_user):
     formdata.store()
 
 
-def test_formdata(pub, local_user):
+@pytest.mark.parametrize('user', ['query-email', 'api-access'])
+@pytest.mark.parametrize('auth', ['signature', 'http-basic'])
+def test_formdata(pub, local_user, user, auth):
     NamedDataSource.wipe()
+
+    app = get_app(pub)
+
+    if user == 'api-access':
+        ApiAccess.wipe()
+        access = ApiAccess()
+        access.name = 'test'
+        access.access_identifier = 'test'
+        access.access_key = '12345'
+        access.store()
+
+        if auth == 'http-basic':
+
+            def get_url(url, **kwargs):
+                app.set_authorization(('Basic', ('test', '12345')))
+                return app.get(url, **kwargs)
+
+        else:
+
+            def get_url(url, **kwargs):
+                return app.get(sign_uri(url, orig=access.access_identifier, key=access.access_key), **kwargs)
+
+    else:
+        if auth == 'http-basic':
+            pytest.skip('http basic authentication requires ApiAccess')
+
+        def get_url(url, **kwargs):
+            return app.get(sign_uri(url, user=local_user), **kwargs)
+
     data_source = NamedDataSource(name='foobar')
     data_source.data_source = {
         'type': 'jsonvalue',
@@ -189,11 +220,16 @@ def test_formdata(pub, local_user):
     formdata.geolocations = {'base': {'lon': 10, 'lat': -12}}
     formdata.store()
 
-    resp = get_app(pub).get(sign_uri('/api/forms/test/%s/' % formdata.id, user=local_user), status=403)
+    resp = get_url('/api/forms/test/%s/' % formdata.id, status=403)
 
-    local_user.roles = [role.id]
-    local_user.store()
-    resp = get_app(pub).get(sign_uri('/api/forms/test/%s/' % formdata.id, user=local_user), status=200)
+    if user == 'api-access':
+        access.roles = [role]
+        access.store()
+    else:
+        local_user.roles = [role.id]
+        local_user.store()
+
+    resp = get_url('/api/forms/test/%s/' % formdata.id, status=200)
 
     assert datetime.datetime.strptime(resp.json['last_update_time'], '%Y-%m-%dT%H:%M:%S')
     assert datetime.datetime.strptime(resp.json['receipt_time'], '%Y-%m-%dT%H:%M:%S')
@@ -210,6 +246,7 @@ def test_formdata(pub, local_user):
     assert resp.json['fields']['file']['content_type'] == 'text/plain'
     assert resp.json['fields']['file']['url'].startswith('http://example.net/test/1/download?hash=')
     assert 'thumbnail_url' not in resp.json['fields']['file']
+    get_url(resp.json['fields']['file']['url'], status=200)
     assert resp.json['fields']['item'] == 'foo'
     assert resp.json['fields']['item_raw'] == '1'
     assert resp.json['fields']['item_structured'] == {'id': '1', 'text': 'foo', 'more': 'XXX'}
@@ -251,14 +288,12 @@ def test_formdata(pub, local_user):
     formdata.jump_status('st1')
     assert formdata.status == 'wf-st1'
 
-    resp = get_app(pub).get(sign_uri('/api/forms/test/%s/' % formdata.id, user=local_user), status=200)
+    resp = get_url('/api/forms/test/%s/' % formdata.id, status=200)
     assert resp.json['workflow']['status'] == {'id': 'new', 'name': 'New'}
     assert resp.json['workflow']['real_status'] == {'id': 'st1', 'name': 'Status1'}
 
     # check ?include-files-content=off
-    resp = get_app(pub).get(
-        sign_uri('/api/forms/test/%s/?include-files-content=off' % formdata.id, user=local_user), status=200
-    )
+    resp = get_url('/api/forms/test/%s/?include-files-content=off' % formdata.id, status=200)
     assert 'content' not in resp.json['fields']['file']
     assert resp.json['fields']['file']['url']
     assert resp.json['fields']['file']['filename'] == 'test.txt'
@@ -391,7 +426,7 @@ def test_formdata_submission_fields(pub, local_user):
 
 
 def test_formdata_backoffice_fields(pub, local_user):
-    test_formdata(pub, local_user)
+    test_formdata(pub, local_user, 'query-email', 'signature')
     Workflow.wipe()
     workflow = Workflow(name='foo')
     workflow.backoffice_fields_formdef = WorkflowBackofficeFieldsFormDef(workflow)

wcs/audit.py

@@ -46,6 +46,9 @@ class Audit(sql.Audit):
             user = get_publisher().user_class.get(audit.user_id, ignore_errors=True)
         elif request:
             user = request.get_user()
+            if user and user.is_api_user:
+                # do not audit API calls
+                return
             if user and hasattr(user, 'id'):
                 audit.user_id = user.id
         if request:

wcs/forms/common.py

@@ -507,11 +507,10 @@ class FormStatusPage(Directory, FormTemplateMixin):
         )
 
     def check_receiver(self):
-        session = get_session()
-        if not session or not session.user:
+        user = get_request().user
+        if not user:
             if not self.filled.formdef.is_user_allowed_read(None, self.filled):
                 raise errors.AccessUnauthorizedError()
-        user = get_request().user
         if self.filled.formdef is None:
             raise errors.AccessForbiddenError()
         if not self.filled.formdef.is_user_allowed_read(user, self.filled):