diff --git a/.gitlab-ci.yml b/.gitlab-ci.yml
index bdbfb3682..05bf86b75 100644
--- a/.gitlab-ci.yml
+++ b/.gitlab-ci.yml
@@ -33,6 +33,7 @@ test:
POSTGRES_PASSWORD: kobo
POSTGRES_DB: kobocat_test
SERVICE_ACCOUNT_BACKEND_URL: redis://redis_cache:6379/4
+ ENKETO_REDIS_MAIN_URL: redis://redis_cache:6379/0
services:
- name: postgis/postgis:14-3.2
alias: postgres
diff --git a/onadata/apps/api/tests/fixtures/formList_w_require_auth.xml b/onadata/apps/api/tests/fixtures/formList_w_require_auth.xml
new file mode 100644
index 000000000..ec3985681
--- /dev/null
+++ b/onadata/apps/api/tests/fixtures/formList_w_require_auth.xml
@@ -0,0 +1,2 @@
+
+transportation_2011_07_25transportation_2011_07_25md5:%(hash)stransportation_2011_07_25http://testserver/forms/%(pk)s/form.xmlhttp://testserver/xformsManifest/%(pk)s
diff --git a/onadata/apps/api/tests/viewsets/test_abstract_viewset.py b/onadata/apps/api/tests/viewsets/test_abstract_viewset.py
index eca65dc9b..883e1672b 100644
--- a/onadata/apps/api/tests/viewsets/test_abstract_viewset.py
+++ b/onadata/apps/api/tests/viewsets/test_abstract_viewset.py
@@ -9,7 +9,6 @@
)
from django.test import TestCase
from django.test.client import Client
-from django_digest.test import Client as DigestClient
from django_digest.test import DigestAuth
from kobo_service_account.utils import get_request_headers
from rest_framework.reverse import reverse
@@ -170,7 +169,6 @@ def _create_user_profile(self, extra_post_data={}):
organization=self.profile_data['organization'],
home_page=self.profile_data['home_page'],
twitter=self.profile_data['twitter'],
- require_auth=False
)
return new_profile
@@ -290,12 +288,3 @@ def _add_form_metadata(
response = self._post_form_metadata(data, test)
return response
-
- def _get_digest_client(self):
- self.user.profile.require_auth = True
- self.user.profile.save()
- client = DigestClient()
- client.set_authorization(self.profile_data['username'],
- self.profile_data['password1'],
- 'Digest')
- return client
diff --git a/onadata/apps/api/tests/viewsets/test_attachment_viewset.py b/onadata/apps/api/tests/viewsets/test_attachment_viewset.py
index 61e32d099..49259fcd5 100644
--- a/onadata/apps/api/tests/viewsets/test_attachment_viewset.py
+++ b/onadata/apps/api/tests/viewsets/test_attachment_viewset.py
@@ -43,6 +43,8 @@ def setUp(self):
alice_profile = self._create_user_profile(alice_profile_data)
self.alice = alice_profile.user
+ # re-assign `self.user` and `self.profile_data` to bob
+ self._login_user_and_profile(self.default_profile_data.copy())
def _retrieve_view(self, auth_headers):
self._submit_transport_instance_w_attachment()
diff --git a/onadata/apps/api/tests/viewsets/test_briefcase_api.py b/onadata/apps/api/tests/viewsets/test_briefcase_api.py
index 36e6adbaa..ab789a239 100644
--- a/onadata/apps/api/tests/viewsets/test_briefcase_api.py
+++ b/onadata/apps/api/tests/viewsets/test_briefcase_api.py
@@ -33,15 +33,10 @@ def setUp(self):
self.form_def_path = os.path.join(
self.main_directory, 'fixtures', 'transportation',
'transportation.xml')
- self._submission_list_url = reverse(
- 'view-submission-list', kwargs={'username': self.user.username})
- self._submission_url = reverse(
- 'submissions', kwargs={'username': self.user.username})
- self._download_submission_url = reverse(
- 'view-download-submission',
- kwargs={'username': self.user.username})
- self._form_upload_url = reverse(
- 'form-upload', kwargs={'username': self.user.username})
+ self._submission_list_url = reverse('view-submission-list')
+ self._submission_url = reverse('submissions')
+ self._download_submission_url = reverse('view-download-submission')
+ self._form_upload_url = reverse('form-upload')
def test_view_submission_list(self):
view = BriefcaseApi.as_view({'get': 'list'})
@@ -50,11 +45,11 @@ def test_view_submission_list(self):
request = self.factory.get(
self._submission_list_url,
data={'formId': self.xform.id_string})
- response = view(request, username=self.user.username)
+ response = view(request)
self.assertEqual(response.status_code, 401)
auth = DigestAuth(self.login_username, self.login_password)
request.META.update(auth(request.META, response))
- response = view(request, username=self.user.username)
+ response = view(request)
self.assertEqual(response.status_code, 200)
submission_list_path = os.path.join(
self.main_directory, 'fixtures', 'transportation',
@@ -71,6 +66,21 @@ def test_view_submission_list(self):
'{{resumptionCursor}}', str(last_index))
self.assertContains(response, expected_submission_list)
+ def test_cannot_view_submission_list_with_username(self):
+ view = BriefcaseApi.as_view({'get': 'list'})
+ self._publish_xml_form()
+ self._make_submissions()
+
+ request = self.factory.get(
+ self._submission_list_url, data={'formId': self.xform.id_string}
+ )
+ response = view(request, username=self.user.username)
+ self.assertEqual(response.status_code, 401)
+ auth = DigestAuth(self.login_username, self.login_password)
+ request.META.update(auth(request.META, response))
+ response = view(request, username=self.user.username)
+ self.assertEqual(response.status_code, 403)
+
def test_view_submission_list_w_deleted_submission(self):
view = BriefcaseApi.as_view({'get': 'list'})
self._publish_xml_form()
@@ -80,11 +90,11 @@ def test_view_submission_list_w_deleted_submission(self):
request = self.factory.get(
self._submission_list_url,
data={'formId': self.xform.id_string})
- response = view(request, username=self.user.username)
+ response = view(request)
self.assertEqual(response.status_code, 401)
auth = DigestAuth(self.login_username, self.login_password)
request.META.update(auth(request.META, response))
- response = view(request, username=self.user.username)
+ response = view(request)
self.assertEqual(response.status_code, 200)
submission_list_path = os.path.join(
self.main_directory, 'fixtures', 'transportation',
@@ -102,18 +112,18 @@ def test_view_submission_list_w_deleted_submission(self):
self.assertContains(response, expected_submission_list)
view = BriefcaseApi.as_view({'get': 'retrieve'})
- formId = '%(formId)s[@version=null and @uiVersion=null]/' \
- '%(formId)s[@key=uuid:%(instanceId)s]' % {
- 'formId': self.xform.id_string,
- 'instanceId': uuid}
- params = {'formId': formId}
- request = self.factory.get(
- self._download_submission_url, data=params)
- response = view(request, username=self.user.username)
+ form_id = (
+ '%(formId)s[@version=null and @uiVersion=null]/'
+ '%(formId)s[@key=uuid:%(instanceId)s]'
+ % {'formId': self.xform.id_string, 'instanceId': uuid}
+ )
+ params = {'formId': form_id}
+ request = self.factory.get(self._download_submission_url, data=params)
+ response = view(request)
self.assertEqual(response.status_code, 401)
auth = DigestAuth(self.login_username, self.login_password)
request.META.update(auth(request.META, response))
- response = view(request, username=self.user.username)
+ response = view(request)
self.assertTrue(response.status_code, 404)
def test_view_submission_list_other_user(self):
@@ -132,10 +142,10 @@ def test_view_submission_list_other_user(self):
request = self.factory.get(
self._submission_list_url,
data={'formId': self.xform.id_string})
- response = view(request, username=self.user.username)
+ response = view(request)
self.assertEqual(response.status_code, 401)
request.META.update(auth(request.META, response))
- response = view(request, username=self.user.username)
+ response = view(request)
self.assertEqual(response.status_code, 404)
def test_view_submission_list_num_entries(self):
@@ -169,10 +179,10 @@ def get_last_index(xform, last_index=None):
request = self.factory.get(
self._submission_list_url,
data=params)
- response = view(request, username=self.user.username)
+ response = view(request)
self.assertEqual(response.status_code, 401)
request.META.update(auth(request.META, response))
- response = view(request, username=self.user.username)
+ response = view(request)
self.assertEqual(response.status_code, 200)
if index > 2:
last_index = get_last_index(self.xform, last_index)
@@ -208,10 +218,10 @@ def test_view_download_submission(self):
auth = DigestAuth(self.login_username, self.login_password)
request = self.factory.get(
self._download_submission_url, data=params)
- response = view(request, username=self.user.username)
+ response = view(request)
self.assertEqual(response.status_code, 401)
request.META.update(auth(request.META, response))
- response = view(request, username=self.user.username)
+ response = view(request)
download_submission_path = os.path.join(
self.main_directory, 'fixtures', 'transportation',
'view', 'downloadSubmission.xml')
@@ -239,10 +249,10 @@ def test_view_download_submission_no_xmlns(self):
auth = DigestAuth(self.login_username, self.login_password)
request = self.factory.get(
self._download_submission_url, data=params)
- response = view(request, username=self.user.username)
+ response = view(request)
self.assertEqual(response.status_code, 401)
request.META.update(auth(request.META, response))
- response = view(request, username=self.user.username)
+ response = view(request)
download_submission_path = os.path.join(
self.main_directory, 'fixtures', 'transportation',
'view', 'downloadSubmission.xml')
@@ -259,9 +269,9 @@ def test_view_download_submission_no_xmlns(self):
with override_settings(SUPPORT_BRIEFCASE_SUBMISSION_DATE=False):
request = self.factory.get(
self._download_submission_url, data=params)
- response = view(request, username=self.user.username)
+ response = view(request)
request.META.update(auth(request.META, response))
- response = view(request, username=self.user.username)
+ response = view(request)
response.render()
self.assertNotIn(
'transportation id="transportation_2011_07_25" '
@@ -293,19 +303,19 @@ def test_view_download_submission_other_user(self):
auth = DigestAuth('alice', 'alicealice')
url = self._download_submission_url # aliasing long name
request = self.factory.get(url, data=params)
- response = view(request, username=self.user.username)
+ response = view(request)
self.assertEqual(response.status_code, 401)
# Rewind the file to avoid the xml parser to get an empty string
# and throw and parsing error
request = request = self.factory.get(url, data=params)
request.META.update(auth(request.META, response))
- response = view(request, username=self.user.username)
+ response = view(request)
self.assertEqual(response.status_code, 404)
def test_publish_xml_form_other_user(self):
view = BriefcaseApi.as_view({'post': 'create'})
- # deno cannot publish form to bob's account
+ # alice cannot publish form to bob's account
alice_data = {
'username': 'alice',
'password1': 'alicealice',
@@ -336,7 +346,7 @@ def test_publish_xml_form_where_filename_is_not_id_string(self):
params = {'form_def_file': f, 'dataFile': ''}
auth = DigestAuth(self.login_username, self.login_password)
request = self.factory.post(self._form_upload_url, data=params)
- response = view(request, username=self.user.username)
+ response = view(request)
self.assertEqual(response.status_code, 401)
# Rewind the file to avoid the xml parser to get an empty string
@@ -345,7 +355,7 @@ def test_publish_xml_form_where_filename_is_not_id_string(self):
# Create a new requests to avoid request.FILES to be empty
request = self.factory.post(self._form_upload_url, data=params)
request.META.update(auth(request.META, response))
- response = view(request, username=self.user.username)
+ response = view(request)
self.assertEqual(XForm.objects.count(), count + 1)
self.assertContains(
response, "successfully published.", status_code=201)
@@ -358,7 +368,7 @@ def _publish_xml_form(self, auth=None):
params = {'form_def_file': f, 'dataFile': ''}
auth = auth or DigestAuth(self.login_username, self.login_password)
request = self.factory.post(self._form_upload_url, data=params)
- response = view(request, username=self.user.username)
+ response = view(request)
self.assertEqual(response.status_code, 401)
# Rewind the file to avoid the xml parser to get an empty string
@@ -367,7 +377,7 @@ def _publish_xml_form(self, auth=None):
# Create a new requests to avoid request.FILES to be empty
request = self.factory.post(self._form_upload_url, data=params)
request.META.update(auth(request.META, response))
- response = view(request, username=self.user.username)
+ response = view(request)
self.assertEqual(XForm.objects.count(), count + 1)
self.assertContains(
response, "successfully published.", status_code=201)
@@ -381,7 +391,7 @@ def test_form_upload(self):
params = {'form_def_file': f, 'dataFile': ''}
auth = DigestAuth(self.login_username, self.login_password)
request = self.factory.post(self._form_upload_url, data=params)
- response = view(request, username=self.user.username)
+ response = view(request)
self.assertEqual(response.status_code, 401)
# Rewind the file to avoid the xml parser to get an empty string
@@ -390,7 +400,7 @@ def test_form_upload(self):
# Create a new requests to avoid request.FILES to be empty
request = self.factory.post(self._form_upload_url, data=params)
request.META.update(auth(request.META, response))
- response = view(request, username=self.user.username)
+ response = view(request)
self.assertEqual(response.status_code, 400)
# SQLite returns `UNIQUE constraint failed` whereas PostgreSQL
# returns 'duplicate key ... violates unique constraint'
@@ -404,10 +414,10 @@ def test_upload_head_request(self):
auth = DigestAuth(self.login_username, self.login_password)
request = self.factory.head(self._form_upload_url)
- response = view(request, username=self.user.username)
+ response = view(request)
self.assertEqual(response.status_code, 401)
request.META.update(auth(request.META, response))
- response = view(request, username=self.user.username)
+ response = view(request)
self.assertEqual(response.status_code, 204)
self.assertTrue(response.has_header('X-OpenRosa-Version'))
self.assertTrue(
@@ -438,7 +448,7 @@ def test_submission_with_instance_id_on_root_node(self):
# Create a new requests to avoid request.FILES to be empty
request = self.factory.post(self._submission_list_url, post_data)
request.META.update(auth(request.META, response))
- response = view(request, username=self.user.username)
+ response = view(request)
self.assertContains(response, message, status_code=201)
self.assertContains(response, instanceId, status_code=201)
self.assertEqual(Instance.objects.count(), count + 1)
diff --git a/onadata/apps/api/tests/viewsets/test_connect_viewset.py b/onadata/apps/api/tests/viewsets/test_connect_viewset.py
index faab4f105..d2cd0c8d5 100644
--- a/onadata/apps/api/tests/viewsets/test_connect_viewset.py
+++ b/onadata/apps/api/tests/viewsets/test_connect_viewset.py
@@ -32,7 +32,7 @@ def setUp(self):
'website': user_profile_data['website'],
'twitter': user_profile_data['twitter'],
'gravatar': user_profile_data['gravatar'],
- 'require_auth': False,
+ 'require_auth': True,
'api_token': self.user.auth_token.key,
'temp_token': self.client.session.session_key,
}
diff --git a/onadata/apps/api/tests/viewsets/test_data_viewset.py b/onadata/apps/api/tests/viewsets/test_data_viewset.py
index bfe639a79..3eba3e109 100644
--- a/onadata/apps/api/tests/viewsets/test_data_viewset.py
+++ b/onadata/apps/api/tests/viewsets/test_data_viewset.py
@@ -145,11 +145,6 @@ def test_data_with_service_account(self):
response.data)
def test_data_anon(self):
- # By default, `_create_user_and_login()` creates users without
- # authentication required. We force it when submitting data to persist
- # collector's username because this test expects it.
- # See `_submitted_data` in `data` below
- self._set_require_auth(True)
self._make_submissions()
view = DataViewSet.as_view({'get': 'list'})
request = self.factory.get('/')
@@ -341,15 +336,18 @@ def test_cannot_get_enketo_edit_url_without_require_auth(self):
less-bad option is to reject edit requests with an explicit error
message when anonymous submissions are enabled.
"""
- self.assertFalse(self.user.profile.require_auth)
+ self.xform.require_auth = False
+ self.xform.save(update_fields=['require_auth'])
+ self.assertFalse(self.xform.require_auth)
self._make_submissions()
+
for view_ in ['enketo', 'enketo_edit']:
view = DataViewSet.as_view({'get': view_})
formid = self.xform.pk
dataid = self.xform.instances.all().order_by('id')[0].pk
request = self.factory.get(
'/',
- data={'return_url': "http://test.io/test_url"},
+ data={'return_url': 'http://test.io/test_url'},
**self.extra
)
response = view(request, pk=formid, dataid=dataid)
@@ -357,14 +355,12 @@ def test_cannot_get_enketo_edit_url_without_require_auth(self):
self.assertTrue(
response.data[0].startswith(
'Cannot edit submissions while "Require authentication '
- 'to see forms and submit data" is disabled for your '
- 'account'
+ 'to see form and submit data" is disabled for your '
+ 'project'
)
)
def test_get_enketo_edit_url(self):
- self.user.profile.require_auth = True
- self.user.profile.save()
self._make_submissions()
for view_ in ['enketo', 'enketo_edit']:
# ensure both legacy `/enketo` and the new `/enketo_edit` endpoints
@@ -448,7 +444,6 @@ def test_get_form_public_data(self):
response_first_element)
def test_data_w_attachment(self):
- self._set_require_auth(auth=True)
self._submit_transport_instance_w_attachment()
view = DataViewSet.as_view({'get': 'list'})
diff --git a/onadata/apps/api/tests/viewsets/test_user.py b/onadata/apps/api/tests/viewsets/test_user.py
index bc8f8f78e..0856b8081 100644
--- a/onadata/apps/api/tests/viewsets/test_user.py
+++ b/onadata/apps/api/tests/viewsets/test_user.py
@@ -9,6 +9,7 @@
from rest_framework.reverse import reverse
from kobo_service_account.utils import get_request_headers
+from onadata.apps.logger.models.xform import XForm
from .test_abstract_viewset import TestAbstractViewSet
@@ -117,7 +118,6 @@ def test_service_account_can_delete_user(self):
def test_only_open_rosa_endpoints_allowed_with_not_validated_password(self):
# log in as bob
self._login_user_and_profile()
- self.user.profile.require_auth = True
self.user.profile.validated_password = True
self.user.profile.save()
@@ -272,6 +272,8 @@ def _access_endpoints(self, access_granted: bool, headers: dict = {}):
)
assert response.status_code == status.HTTP_200_OK
+ # Need to deactivate auth on XForm when using OpenRosa endpoints with username
+ XForm.objects.filter(pk=xform_id).update(require_auth=False)
response = self.client.get(
reverse('manifest-url', kwargs={'pk': xform_id, 'username': 'bob'}),
**headers,
diff --git a/onadata/apps/api/tests/viewsets/test_xform_list_api.py b/onadata/apps/api/tests/viewsets/test_xform_list_api.py
index 047c58a8a..4bdd41c1e 100644
--- a/onadata/apps/api/tests/viewsets/test_xform_list_api.py
+++ b/onadata/apps/api/tests/viewsets/test_xform_list_api.py
@@ -1,27 +1,214 @@
# coding: utf-8
import os
+import re
from django.conf import settings
from django_digest.test import DigestAuth
from guardian.shortcuts import assign_perm
+from rest_framework.reverse import reverse
-from onadata.apps.api.tests.viewsets.test_abstract_viewset import\
+from onadata.apps.api.tests.viewsets.test_abstract_viewset import (
TestAbstractViewSet
+)
from onadata.apps.api.viewsets.xform_list_api import XFormListApi
from onadata.libs.constants import (
CAN_ADD_SUBMISSIONS,
CAN_VIEW_XFORM
)
+from onadata.apps.logger.models.xform import XForm
+
+class TestXFormListApiBase(TestAbstractViewSet):
-class TestXFormListApi(TestAbstractViewSet):
def setUp(self):
super().setUp()
self.view = XFormListApi.as_view({
- "get": "list"
+ 'get': 'list'
})
self.publish_xls_form()
+ def _load_metadata(self, xform=None):
+ data_value = "screenshot.png"
+ data_type = 'media'
+ fixture_dir = os.path.join(
+ settings.ONADATA_DIR, "apps", "main", "tests", "fixtures",
+ "transportation"
+ )
+ path = os.path.join(fixture_dir, data_value)
+ xform = xform or self.xform
+
+ self._add_form_metadata(xform, data_type, data_value, path)
+
+
+class TestXFormListApiWithoutAuthRequired(TestXFormListApiBase):
+
+ """
+ Tests should point to `https://kc//*`
+ """
+
+ def setUp(self):
+ super().setUp()
+ self.xform_without_auth = self.xform
+ self.xform_without_auth.require_auth = False
+ self.xform_without_auth.save(update_fields=['require_auth'])
+
+ data = {
+ 'owner': self.user.username,
+ 'public': False,
+ 'public_data': False,
+ 'description': 'transportation_with_attachment',
+ 'downloadable': True,
+ 'encrypted': False,
+ 'id_string': 'transportation_with_attachment',
+ 'title': 'transportation_with_attachment',
+ }
+
+ path = os.path.join(
+ settings.ONADATA_DIR,
+ 'apps',
+ 'main',
+ 'tests',
+ 'fixtures',
+ 'transportation',
+ 'transportation_with_attachment.xls',
+ )
+ self.publish_xls_form(data=data, path=path)
+ self.assertNotEqual(self.xform.pk, self.xform_without_auth)
+ self.assertEqual(XForm.objects.all().count(), 2)
+ self.assertEqual(XForm.objects.filter(require_auth=False).count(), 1)
+
+ def test_get_xform_list_as_anonymous_user(self):
+
+ request = self.factory.get('/')
+ response = self.view(request, username=self.user.username)
+ self.assertEqual(response.status_code, 200)
+
+ path = os.path.join(
+ os.path.dirname(__file__), '..', 'fixtures', 'formList.xml'
+ )
+ # Response should contain only xform
+ with open(path, 'r') as f:
+ form_list_xml = f.read().strip()
+ data = {
+ 'hash': self.xform_without_auth.md5_hash,
+ 'pk': self.xform_without_auth.pk,
+ }
+ content = response.render().content
+ self.assertEqual(content.decode('utf-8'), form_list_xml % data)
+ self.assertTrue(response.has_header('X-OpenRosa-Version'))
+ self.assertTrue(
+ response.has_header('X-OpenRosa-Accept-Content-Length'))
+ self.assertTrue(response.has_header('Date'))
+ self.assertEqual(response['Content-Type'],
+ 'text/xml; charset=utf-8')
+
+ def test_get_xform_list_as_owner(self):
+
+ """
+ Same test as `test_get_xform_list_as_anonymous_user()` except
+ we want the user to be authenticated right away. Do not use Digest, but
+ session auth.
+ User should only see their projects that allow data submission without
+ authentication (like anonymous user)
+ """
+
+ # Use session auth
+ response = self.client.get(
+ reverse('form-list', kwargs={'username': self.user.username})
+ )
+ self.assertEqual(response.status_code, 200)
+
+ path = os.path.join(
+ os.path.dirname(__file__), '..', 'fixtures', 'formList.xml'
+ )
+ # Response should contain only xform
+ with open(path, 'r') as f:
+ form_list_xml = f.read().strip()
+ data = {
+ 'hash': self.xform_without_auth.md5_hash,
+ 'pk': self.xform_without_auth.pk,
+ }
+ content = response.render().content
+ self.assertEqual(content.decode('utf-8'), form_list_xml % data)
+ self.assertTrue(response.has_header('X-OpenRosa-Version'))
+ self.assertTrue(
+ response.has_header('X-OpenRosa-Accept-Content-Length'))
+ self.assertTrue(response.has_header('Date'))
+ self.assertEqual(response['Content-Type'],
+ 'text/xml; charset=utf-8')
+
+ def test_retrieve_xform_manifest_as_owner(self):
+
+ self._load_metadata(self.xform_without_auth)
+ self.view = XFormListApi.as_view({
+ 'get': 'manifest'
+ })
+ request = self.factory.get('/')
+ response = self.view(request, pk=self.xform_without_auth.pk)
+ self.assertEqual(response.status_code, 401)
+ response = self.view(
+ request, pk=self.xform_without_auth.pk, username=self.user.username
+ )
+ self.assertEqual(response.status_code, 200)
+
+ manifest_xml = (
+ '\n'
+ ''
+ ' '
+ ' screenshot.png'
+ ' %(hash)s'
+ ' http://testserver/bob/xformsMedia/%(xform)s/%(pk)s.png'
+ ' '
+ ''
+ )
+
+ manifest_xml = re.sub(r'> +<', '><', manifest_xml).strip()
+
+ data = {
+ 'hash': self.metadata.md5_hash,
+ 'pk': self.metadata.pk,
+ 'xform': self.xform_without_auth.pk,
+ }
+
+ content = response.render().content.decode('utf-8').strip()
+ self.assertEqual(content, manifest_xml % data)
+ self.assertTrue(response.has_header('X-OpenRosa-Version'))
+ self.assertTrue(
+ response.has_header('X-OpenRosa-Accept-Content-Length'))
+ self.assertTrue(response.has_header('Date'))
+ self.assertEqual(response['Content-Type'], 'text/xml; charset=utf-8')
+
+ def test_retrieve_xform_media_as_anonymous_user(self):
+
+ self._load_metadata(self.xform_without_auth)
+ self.view = XFormListApi.as_view({
+ 'get': 'media'
+ })
+ request = self.factory.get('/')
+ response = self.view(
+ request,
+ pk=self.xform_without_auth.pk,
+ metadata=self.metadata.pk,
+ format='png',
+ )
+ self.assertEqual(response.status_code, 401)
+
+ response = self.view(
+ request,
+ pk=self.xform_without_auth.pk,
+ username=self.user.username,
+ metadata=self.metadata.pk,
+ format='png',
+ )
+ self.assertEqual(response.status_code, 200)
+
+
+class TestXFormListApiWithAuthRequired(TestXFormListApiBase):
+
+ """
+ Tests should point to `https://kc/*`
+ """
+
def test_head_xform_list(self):
request = self.factory.head('/')
response = self.view(request)
@@ -42,7 +229,10 @@ def test_get_xform_list(self):
path = os.path.join(
os.path.dirname(__file__),
- '..', 'fixtures', 'formList.xml')
+ '..',
+ 'fixtures',
+ 'formList_w_require_auth.xml',
+ )
with open(path, 'r') as f:
form_list_xml = f.read().strip()
@@ -78,37 +268,11 @@ def test_get_xform_list_inactive_form(self):
self.assertEqual(response['Content-Type'],
'text/xml; charset=utf-8')
- def test_get_xform_list_anonymous_user(self):
- request = self.factory.get('/')
- response = self.view(request)
- self.assertEqual(response.status_code, 401)
- response = self.view(request, username=self.user.username)
- self.assertEqual(response.status_code, 200)
-
- path = os.path.join(
- os.path.dirname(__file__),
- '..', 'fixtures', 'formList.xml')
-
- with open(path, 'r') as f:
- form_list_xml = f.read().strip()
- data = {"hash": self.xform.md5_hash, "pk": self.xform.pk}
- content = response.render().content
- self.assertEqual(content.decode('utf-8'), form_list_xml % data)
- self.assertTrue(response.has_header('X-OpenRosa-Version'))
- self.assertTrue(
- response.has_header('X-OpenRosa-Accept-Content-Length'))
- self.assertTrue(response.has_header('Date'))
- self.assertEqual(response['Content-Type'],
- 'text/xml; charset=utf-8')
-
- def test_get_xform_list_anonymous_user_require_auth(self):
- self.user.profile.require_auth = True
- self.user.profile.save()
+ def test_get_xform_list_as_anonymous_user(self):
request = self.factory.get('/')
+ # Get formList without username requires auth unconditionally
response = self.view(request)
self.assertEqual(response.status_code, 401)
- response = self.view(request, username=self.user.username)
- self.assertEqual(response.status_code, 401)
def test_get_xform_list_other_user_with_no_role(self):
request = self.factory.get('/')
@@ -194,7 +358,10 @@ def test_get_xform_list_other_user_with_dataentry_role(self):
path = os.path.join(
os.path.dirname(__file__),
- '..', 'fixtures', 'formList.xml')
+ '..',
+ 'fixtures',
+ 'formList_w_require_auth.xml',
+ )
with open(path, 'r') as f:
form_list_xml = f.read().strip()
@@ -214,16 +381,27 @@ def test_get_xform_list_with_formid_parameter(self):
"""
# Test unrecognized `formID`
request = self.factory.get('/', {'formID': 'unrecognizedID'})
- response = self.view(request, username=self.user.username)
+ response = self.view(request)
+ self.assertEqual(response.status_code, 401)
+ auth = DigestAuth('bob', 'bobbob')
+ request.META.update(auth(request.META, response))
+ response = self.view(request)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.data, [])
# Test a valid `formID`
request = self.factory.get('/', {'formID': self.xform.id_string})
- response = self.view(request, username=self.user.username)
+ response = self.view(request)
+ self.assertEqual(response.status_code, 401)
+ auth = DigestAuth('bob', 'bobbob')
+ request.META.update(auth(request.META, response))
+ response = self.view(request)
self.assertEqual(response.status_code, 200)
path = os.path.join(
- os.path.dirname(__file__), '..', 'fixtures', 'formList.xml'
+ os.path.dirname(__file__),
+ '..',
+ 'fixtures',
+ 'formList_w_require_auth.xml',
)
with open(path) as f:
@@ -234,7 +412,7 @@ def test_get_xform_list_with_formid_parameter(self):
def test_retrieve_xform_xml(self):
self.view = XFormListApi.as_view({
- "get": "retrieve"
+ 'get': 'retrieve'
})
request = self.factory.head('/')
response = self.view(request, pk=self.xform.pk)
@@ -257,39 +435,14 @@ def test_retrieve_xform_xml(self):
with open(path) as f:
form_xml = f.read().strip()
- data = {"form_uuid": self.xform.uuid}
+ data = {'form_uuid': self.xform.uuid}
content = response.render().content.decode('utf-8').strip()
self.assertEqual(content, form_xml % data)
- def _load_metadata(self, xform=None):
- data_value = "screenshot.png"
- data_type = 'media'
- fixture_dir = os.path.join(
- settings.ONADATA_DIR, "apps", "main", "tests", "fixtures",
- "transportation"
- )
- path = os.path.join(fixture_dir, data_value)
- xform = xform or self.xform
-
- self._add_form_metadata(xform, data_type, data_value, path)
-
- def test_head_xform_manifest(self):
- self._load_metadata(self.xform)
- self.view = XFormListApi.as_view({
- "get": "manifest"
- })
- request = self.factory.head('/')
- response = self.view(request, pk=self.xform.pk)
- self.assertEqual(response.status_code, 401)
- auth = DigestAuth('bob', 'bobbob')
- request.META.update(auth(request.META, response))
- response = self.view(request, pk=self.xform.pk)
- self.validate_openrosa_head_response(response)
-
def test_retrieve_xform_manifest(self):
self._load_metadata(self.xform)
self.view = XFormListApi.as_view({
- "get": "manifest"
+ 'get': 'manifest'
})
request = self.factory.head('/')
response = self.view(request, pk=self.xform.pk)
@@ -299,10 +452,24 @@ def test_retrieve_xform_manifest(self):
response = self.view(request, pk=self.xform.pk)
self.assertEqual(response.status_code, 200)
- manifest_xml = """
-screenshot.png%(hash)shttp://testserver/bob/xformsMedia/%(xform)s/%(pk)s.png""" # noqa
- data = {"hash": self.metadata.md5_hash, "pk": self.metadata.pk,
- "xform": self.xform.pk}
+ manifest_xml = (
+ '\n'
+ ''
+ ' '
+ ' screenshot.png'
+ ' %(hash)s'
+ ' http://testserver/xformsMedia/%(xform)s/%(pk)s.png'
+ ' '
+ ''
+ )
+
+ manifest_xml = re.sub(r'> +<', '><', manifest_xml).strip()
+
+ data = {
+ 'hash': self.metadata.md5_hash,
+ 'pk': self.metadata.pk,
+ 'xform': self.xform.pk,
+ }
content = response.render().content.decode('utf-8').strip()
self.assertEqual(content, manifest_xml % data)
self.assertTrue(response.has_header('X-OpenRosa-Version'))
@@ -311,48 +478,36 @@ def test_retrieve_xform_manifest(self):
self.assertTrue(response.has_header('Date'))
self.assertEqual(response['Content-Type'], 'text/xml; charset=utf-8')
- def test_retrieve_xform_manifest_anonymous_user(self):
+ def test_retrieve_xform_manifest_as_anonymous(self):
self._load_metadata(self.xform)
self.view = XFormListApi.as_view({
- "get": "manifest"
+ 'get': 'manifest'
})
request = self.factory.get('/')
- response = self.view(request, pk=self.xform.pk)
- self.assertEqual(response.status_code, 401)
- response = self.view(request, pk=self.xform.pk,
- username=self.user.username)
- self.assertEqual(response.status_code, 200)
+ response = self.view(request, pk=self.xform.pk, username=self.user.username)
+ # The project (self.xform) requires auth by default. Anonymous user cannot
+ # access it. It is also true for the project manifest, thus anonymous
+ # should receive a 404.
+ # See `TestXFormListApiWithoutAuthRequired.test_retrieve_xform_manifest()`
+ self.assertEqual(response.status_code, 404)
- manifest_xml = """
-screenshot.png%(hash)shttp://testserver/bob/xformsMedia/%(xform)s/%(pk)s.png""" # noqa
- data = {"hash": self.metadata.md5_hash, "pk": self.metadata.pk,
- "xform": self.xform.pk}
- content = response.render().content.decode('utf-8').strip()
- self.assertEqual(content, manifest_xml % data)
- self.assertTrue(response.has_header('X-OpenRosa-Version'))
- self.assertTrue(
- response.has_header('X-OpenRosa-Accept-Content-Length'))
- self.assertTrue(response.has_header('Date'))
- self.assertEqual(response['Content-Type'], 'text/xml; charset=utf-8')
-
- def test_retrieve_xform_manifest_anonymous_user_require_auth(self):
- self.user.profile.require_auth = True
- self.user.profile.save()
+ def test_head_xform_manifest(self):
self._load_metadata(self.xform)
self.view = XFormListApi.as_view({
- "get": "manifest"
+ 'get': 'manifest'
})
- request = self.factory.get('/')
+ request = self.factory.head('/')
response = self.view(request, pk=self.xform.pk)
self.assertEqual(response.status_code, 401)
- response = self.view(request, pk=self.xform.pk,
- username=self.user.username)
- self.assertEqual(response.status_code, 401)
+ auth = DigestAuth('bob', 'bobbob')
+ request.META.update(auth(request.META, response))
+ response = self.view(request, pk=self.xform.pk)
+ self.validate_openrosa_head_response(response)
def test_head_xform_media(self):
self._load_metadata(self.xform)
self.view = XFormListApi.as_view({
- "get": "media"
+ 'get': 'media'
})
request = self.factory.head('/')
response = self.view(request, pk=self.xform.pk,
@@ -380,29 +535,13 @@ def test_retrieve_xform_media(self):
metadata=self.metadata.pk, format='png')
self.assertEqual(response.status_code, 200)
- def test_retrieve_xform_media_anonymous_user(self):
+ def test_retrieve_xform_media_as_anonymous(self):
self._load_metadata(self.xform)
self.view = XFormListApi.as_view({
- "get": "media"
+ 'get': 'media'
})
request = self.factory.get('/')
- response = self.view(request, pk=self.xform.pk,
- metadata=self.metadata.pk, format='png')
- self.assertEqual(response.status_code, 401)
-
- response = self.view(request, pk=self.xform.pk,
- username=self.user.username,
- metadata=self.metadata.pk, format='png')
- self.assertEqual(response.status_code, 200)
-
- def test_retrieve_xform_media_anonymous_user_require_auth(self):
- self.user.profile.require_auth = True
- self.user.profile.save()
- self._load_metadata(self.xform)
- self.view = XFormListApi.as_view({
- "get": "media"
- })
- request = self.factory.get('/')
- response = self.view(request, pk=self.xform.pk,
- metadata=self.metadata.pk, format='png')
+ response = self.view(
+ request, pk=self.xform.pk, metadata=self.metadata.pk, format='png'
+ )
self.assertEqual(response.status_code, 401)
diff --git a/onadata/apps/api/tests/viewsets/test_xform_submission_api.py b/onadata/apps/api/tests/viewsets/test_xform_submission_api.py
index ac7155131..f279c46c7 100644
--- a/onadata/apps/api/tests/viewsets/test_xform_submission_api.py
+++ b/onadata/apps/api/tests/viewsets/test_xform_submission_api.py
@@ -40,38 +40,64 @@ def test_head_response(self):
self.validate_openrosa_head_response(response)
def test_post_submission_anonymous(self):
+
+ self.xform.require_auth = False
+ self.xform.save(update_fields=['require_auth'])
+
s = self.surveys[0]
- media_file = "1335783522563.jpg"
- path = os.path.join(self.main_directory, 'fixtures',
- 'transportation', 'instances', s, media_file)
+ media_file = '1335783522563.jpg'
+ path = os.path.join(
+ self.main_directory,
+ 'fixtures',
+ 'transportation',
+ 'instances',
+ s,
+ media_file,
+ )
with open(path, 'rb') as f:
- f = InMemoryUploadedFile(f, 'media_file', media_file, 'image/jpg',
- os.path.getsize(path), None)
+ f = InMemoryUploadedFile(
+ f,
+ 'media_file',
+ media_file,
+ 'image/jpg',
+ os.path.getsize(path),
+ None,
+ )
submission_path = os.path.join(
- self.main_directory, 'fixtures',
- 'transportation', 'instances', s, s + '.xml')
+ self.main_directory,
+ 'fixtures',
+ 'transportation',
+ 'instances',
+ s,
+ s + '.xml',
+ )
with open(submission_path) as sf:
data = {'xml_submission_file': sf, 'media_file': f}
request = self.factory.post(
- '/%s/submission' % self.user.username, data)
+ f'/{self.user.username}/submission', data
+ )
request.user = AnonymousUser()
response = self.view(request, username=self.user.username)
- self.assertContains(response, 'Successful submission',
- status_code=201)
+ self.assertContains(
+ response, 'Successful submission', status_code=201
+ )
self.assertTrue(response.has_header('X-OpenRosa-Version'))
self.assertTrue(
- response.has_header('X-OpenRosa-Accept-Content-Length'))
+ response.has_header('X-OpenRosa-Accept-Content-Length')
+ )
self.assertTrue(response.has_header('Date'))
- self.assertEqual(response['Content-Type'],
- 'text/xml; charset=utf-8')
- self.assertEqual(response['Location'],
- 'http://testserver/%s/submission'
- % self.user.username)
+ self.assertEqual(
+ response['Content-Type'], 'text/xml; charset=utf-8'
+ )
+ self.assertEqual(
+ response['Location'],
+ f'http://testserver/{self.user.username}/submission',
+ )
def test_post_submission_authenticated(self):
s = self.surveys[0]
- media_file = "1335783522563.jpg"
+ media_file = '1335783522563.jpg'
path = os.path.join(self.main_directory, 'fixtures',
'transportation', 'instances', s, media_file)
with open(path, 'rb') as f:
@@ -91,6 +117,7 @@ def test_post_submission_authenticated(self):
# rewind the file and redo the request since they were
# consumed
sf.seek(0)
+ f.seek(0)
request = self.factory.post('/submission', data)
auth = DigestAuth('bob', 'bobbob')
request.META.update(auth(request.META, response))
@@ -204,8 +231,6 @@ def test_post_submission_authenticated_bad_json(self):
'http://testserver/submission')
def test_post_submission_require_auth(self):
- self.user.profile.require_auth = True
- self.user.profile.save()
count = Attachment.objects.count()
s = self.surveys[0]
media_file = "1335783522563.jpg"
@@ -246,8 +271,6 @@ def test_post_submission_require_auth(self):
'http://testserver/submission')
def test_post_submission_require_auth_anonymous_user(self):
- self.user.profile.require_auth = True
- self.user.profile.save()
count = Attachment.objects.count()
s = self.surveys[0]
media_file = "1335783522563.jpg"
@@ -269,8 +292,6 @@ def test_post_submission_require_auth_anonymous_user(self):
self.assertEqual(count, Attachment.objects.count())
def test_post_submission_require_auth_other_user(self):
- self.user.profile.require_auth = True
- self.user.profile.save()
alice_data = {
'username': 'alice',
@@ -310,8 +331,6 @@ def test_post_submission_require_auth_other_user(self):
self.assertContains(response, 'Forbidden', status_code=403)
def test_post_submission_require_auth_data_entry_role(self):
- self.user.profile.require_auth = True
- self.user.profile.save()
alice_data = {
'username': 'alice',
@@ -374,9 +393,6 @@ def test_edit_submission_with_service_account(self):
"""
# Ensure only authenticated users can submit data
- self.user.profile.require_auth = True
- self.user.profile.save(update_fields=['require_auth'])
-
path = os.path.join(
os.path.dirname(os.path.abspath(__file__)),
'..',
@@ -447,6 +463,11 @@ def test_submission_blocking_flag(self):
# submission do fail with the flag set
self.xform.user.profile.metadata['submissions_suspended'] = True
self.xform.user.profile.save()
+
+ # No need auth for this test
+ self.xform.require_auth = False
+ self.xform.save(update_fields=['require_auth'])
+
s = self.surveys[0]
username = self.user.username
media_file = '1335783522563.jpg'
diff --git a/onadata/apps/api/viewsets/briefcase_api.py b/onadata/apps/api/viewsets/briefcase_api.py
index 452f559dd..e7d6617a3 100644
--- a/onadata/apps/api/viewsets/briefcase_api.py
+++ b/onadata/apps/api/viewsets/briefcase_api.py
@@ -50,11 +50,11 @@ def _extract_uuid(text):
return text
-def _extract_id_string(formId):
- if isinstance(formId, str):
- return formId[0:formId.find('[')]
+def _extract_id_string(form_id):
+ if isinstance(form_id, str):
+ return form_id[0:form_id.find('[')]
- return formId
+ return form_id
def _parse_int(num):
@@ -103,42 +103,45 @@ def __init__(self, *args, **kwargs):
]
def get_object(self):
- formId = self.request.GET.get('formId', '')
- id_string = _extract_id_string(formId)
- uuid = _extract_uuid(formId)
- username = self.kwargs.get('username')
+ form_id = self.request.GET.get('formId', '')
+ id_string = _extract_id_string(form_id)
+ uuid = _extract_uuid(form_id)
- obj = get_instance_or_404(xform__user__username__iexact=username,
- xform__id_string__exact=id_string,
- uuid=uuid)
+ obj = get_instance_or_404(
+ xform__id_string=id_string,
+ uuid=uuid,
+ )
self.check_object_permissions(self.request, obj.xform)
return obj
def filter_queryset(self, queryset):
username = self.kwargs.get('username')
- if username is None and self.request.user.is_anonymous:
- # raises a permission denied exception, forces authentication
- self.permission_denied(self.request)
-
- if username is not None and self.request.user.is_anonymous:
- profile = get_object_or_404(
- UserProfile, user__username=username.lower())
-
- if profile.require_auth:
+ if username is None:
+ # Briefcase does not allow anonymous access, user should always be
+ # authenticated
+ if self.request.user.is_anonymous:
# raises a permission denied exception, forces authentication
self.permission_denied(self.request)
else:
- queryset = queryset.filter(user=profile.user)
+ # Return all the forms the currently-logged-in user can access,
+ # including those shared by other users
+ queryset = super().filter_queryset(queryset)
else:
- queryset = super().filter_queryset(queryset)
+ # With the advent of project-level anonymous access in #904, Briefcase
+ # requests must no longer use endpoints whose URLs contain usernames.
+ # Ideally, Briefcase would display error messages returned by this method,
+ # but sadly that is not the case.
+ # Raise an empty PermissionDenied since it's impossible to have
+ # Briefcase display any guidance.
+ raise exceptions.PermissionDenied()
- formId = self.request.GET.get('formId', '')
+ form_id = self.request.GET.get('formId', '')
- if formId.find('[') != -1:
- formId = _extract_id_string(formId)
+ if form_id.find('[') != -1:
+ form_id = _extract_id_string(form_id)
- xform = get_object_or_404(queryset, id_string__exact=formId)
+ xform = get_object_or_404(queryset, id_string=form_id)
self.check_object_permissions(self.request, xform)
instances = Instance.objects.filter(xform=xform).order_by('pk')
num_entries = self.request.GET.get('numEntries')
@@ -154,11 +157,11 @@ def filter_queryset(self, queryset):
if instances.count():
last_instance = instances[instances.count() - 1]
- self.resumptionCursor = last_instance.pk
+ self.resumption_cursor = last_instance.pk
elif instances.count() == 0 and cursor:
- self.resumptionCursor = cursor
+ self.resumption_cursor = cursor
else:
- self.resumptionCursor = 0
+ self.resumption_cursor = 0
return instances
@@ -170,23 +173,19 @@ def create(self, request, *args, **kwargs):
xform_def = request.FILES.get('form_def_file', None)
response_status = status.HTTP_201_CREATED
- username = kwargs.get('username')
- form_user = (username and get_object_or_404(User, username=username)) \
- or request.user
-
- if not request.user.has_perm(
- 'can_add_xform',
- UserProfile.objects.get_or_create(user=form_user)[0]
- ):
- raise exceptions.PermissionDenied(
- detail=t("User %(user)s has no permission to add xforms to "
- "account %(account)s" %
- {'user': request.user.username,
- 'account': form_user.username}))
+ # With the advent of project-level anonymous access in #904, Briefcase
+ # requests must no longer use endpoints whose URLs contain usernames.
+ # Ideally, Briefcase would display error messages returned by this method,
+ # but sadly that is not the case.
+ # Raise an empty PermissionDenied since it's impossible to have
+ # Briefcase display any guidance.
+ if kwargs.get('username'):
+ raise exceptions.PermissionDenied()
+
data = {}
if isinstance(xform_def, File):
- do_form_upload = DoXmlFormUpload(xform_def, form_user)
+ do_form_upload = DoXmlFormUpload(xform_def, request.user)
dd = publish_form(do_form_upload.publish)
if isinstance(dd, XForm):
@@ -205,24 +204,27 @@ def create(self, request, *args, **kwargs):
template_name=self.template_name)
def list(self, request, *args, **kwargs):
- self.object_list = self.filter_queryset(self.get_queryset())
+ object_list = self.filter_queryset(self.get_queryset())
- data = {'instances': self.object_list,
- 'resumptionCursor': self.resumptionCursor}
+ data = {
+ 'instances': object_list,
+ 'resumptionCursor': self.resumption_cursor,
+ }
- return Response(data,
- headers=self.get_openrosa_headers(request,
- location=False),
- template_name='submissionList.xml')
+ return Response(
+ data,
+ headers=self.get_openrosa_headers(request, location=False),
+ template_name='submissionList.xml',
+ )
def retrieve(self, request, *args, **kwargs):
- self.object = self.get_object()
+ instance = self.get_object()
- submission_xml_root_node = self.object.get_root_node()
+ submission_xml_root_node = instance.get_root_node()
submission_xml_root_node.setAttribute(
- 'instanceID', 'uuid:%s' % self.object.uuid)
+ 'instanceID', 'uuid:%s' % instance.uuid)
submission_xml_root_node.setAttribute(
- 'submissionDate', self.object.date_created.isoformat()
+ 'submissionDate', instance.date_created.isoformat()
)
# Added this because of https://github.com/onaio/onadata/pull/2139
@@ -236,7 +238,7 @@ def retrieve(self, request, *args, **kwargs):
data = {
'submission_data': submission_xml_root_node.toxml(),
- 'media_files': Attachment.objects.filter(instance=self.object),
+ 'media_files': Attachment.objects.filter(instance=instance),
'host': request.build_absolute_uri().replace(
request.get_full_path(), '')
}
@@ -248,21 +250,23 @@ def retrieve(self, request, *args, **kwargs):
@action(detail=True, methods=['GET'])
def manifest(self, request, *args, **kwargs):
- self.object = self.get_object()
+ xform = self.get_object()
object_list = MetaData.objects.filter(
- data_type__in=MetaData.MEDIA_FILES_TYPE, xform=self.object
+ data_type__in=MetaData.MEDIA_FILES_TYPE, xform=xform
)
context = self.get_serializer_context()
- serializer = XFormManifestSerializer(object_list, many=True,
- context=context)
+ serializer = XFormManifestSerializer(
+ object_list, many=True, context=context
+ )
- return Response(serializer.data,
- headers=self.get_openrosa_headers(request,
- location=False))
+ return Response(
+ serializer.data,
+ headers=self.get_openrosa_headers(request, location=False),
+ )
@action(detail=True, methods=['GET'])
def media(self, request, *args, **kwargs):
- self.object = self.get_object()
+ xform = self.get_object()
pk = kwargs.get('metadata')
if not pk:
@@ -271,7 +275,7 @@ def media(self, request, *args, **kwargs):
meta_obj = get_object_or_404(
MetaData,
data_type__in=MetaData.MEDIA_FILES_TYPE,
- xform=self.object,
+ xform=xform,
pk=pk,
)
diff --git a/onadata/apps/api/viewsets/data_viewset.py b/onadata/apps/api/viewsets/data_viewset.py
index 68d22de6d..cbf97a69d 100644
--- a/onadata/apps/api/viewsets/data_viewset.py
+++ b/onadata/apps/api/viewsets/data_viewset.py
@@ -648,21 +648,6 @@ def enketo(self, request, *args, **kwargs):
permission_classes=[EnketoSubmissionEditPermissions],
)
def enketo_edit(self, request, *args, **kwargs):
- """
- Trying to edit in Enketo while `profile.require_auth == False` leads to
- an infinite authentication loop because Enketo never sends credentials
- unless it receives a 401 response to an unauthenticated HEAD request.
- There's no way to send such a response for editing only while
- simultaneously allowing anonymous submissions to the same endpoint.
- Avoid the infinite loop by blocking doomed requests here and returning
- a helpful error message.
- """
- profile = UserProfile.objects.get_or_create(user=request.user)[0]
- if not profile.require_auth:
- raise ValidationError(t(
- 'Cannot edit submissions while "Require authentication to see '
- 'forms and submit data" is disabled for your account'
- ))
return self._enketo_request(request, action_='edit', *args, **kwargs)
@action(
@@ -683,6 +668,20 @@ def _enketo_request(self, request, action_, *args, **kwargs):
if not return_url and not action_ == 'view':
raise ParseError(t('`return_url` not provided.'))
+ if not object_.xform.require_auth and action_ == 'edit':
+ # Trying to edit in Enketo while `xform.require_auth == False`
+ # leads to an infinite authentication loop because Enketo never
+ # sends credentials unless it receives a 401 response to an
+ # unauthenticated HEAD request.
+ # There's no way to send such a response for editing only while
+ # simultaneously allowing anonymous submissions to the same endpoint.
+ # Avoid the infinite loop by blocking doomed requests here and
+ # returning a helpful error message.
+ raise ValidationError(t(
+ 'Cannot edit submissions while "Require authentication to '
+ 'see form and submit data" is disabled for your project'
+ ))
+
try:
data['url'] = get_enketo_submission_url(
request, object_, return_url, action=action_
diff --git a/onadata/apps/api/viewsets/xform_list_api.py b/onadata/apps/api/viewsets/xform_list_api.py
index 5a4ce1bdf..32bb14536 100644
--- a/onadata/apps/api/viewsets/xform_list_api.py
+++ b/onadata/apps/api/viewsets/xform_list_api.py
@@ -15,7 +15,6 @@
from onadata.apps.api.tools import get_media_file_response
from onadata.apps.logger.models.xform import XForm
from onadata.apps.main.models.meta_data import MetaData
-from onadata.apps.main.models.user_profile import UserProfile
from onadata.libs import filters
from onadata.libs.authentication import DigestAuthentication
from onadata.libs.renderers.renderers import MediaFileContentNegotiation
@@ -46,8 +45,9 @@ def __init__(self, *args, **kwargs):
DigestAuthentication,
]
self.authentication_classes = authentication_classes + [
- auth_class for auth_class in self.authentication_classes
- if auth_class not in authentication_classes
+ auth_class
+ for auth_class in self.authentication_classes
+ if auth_class not in authentication_classes
]
def get_openrosa_headers(self):
@@ -76,6 +76,7 @@ def get_renderers(self):
def filter_queryset(self, queryset):
username = self.kwargs.get('username')
+
if username is None:
# If no username is specified, the request must be authenticated
if self.request.user.is_anonymous:
@@ -86,25 +87,12 @@ def filter_queryset(self, queryset):
# including those shared by other users
queryset = super().filter_queryset(queryset)
else:
- profile = get_object_or_404(
- UserProfile, user__username=username.lower()
+ # Only return projects that allow anonymous submissions when path
+ # starts with a username
+ queryset = queryset.filter(
+ user__username=username.lower(), require_auth=False
)
- # Include only the forms belonging to the specified user
- queryset = queryset.filter(user=profile.user)
- if profile.require_auth:
- # The specified has user ticked "Require authentication to see
- # forms and submit data"; reject anonymous requests
- if self.request.user.is_anonymous:
- # raises a permission denied exception, forces
- # authentication
- self.permission_denied(self.request)
- else:
- # Someone has logged in, but they are not necessarily
- # allowed to access the forms belonging to the specified
- # user. Filter again to consider object-level permissions
- queryset = super().filter_queryset(
- queryset
- )
+
try:
# https://docs.getodk.org/openrosa-form-list/#form-list-api says:
# `formID`: If specified, the server MUST return information for
@@ -118,21 +106,27 @@ def filter_queryset(self, queryset):
return queryset
def list(self, request, *args, **kwargs):
- self.object_list = self.filter_queryset(self.get_queryset())
+
+ object_list = self.filter_queryset(self.get_queryset())
if request.method == 'HEAD':
return self.get_response_for_head_request()
- serializer = self.get_serializer(self.object_list, many=True)
+ serializer = self.get_serializer(
+ object_list, many=True, require_auth=not bool(kwargs.get('username'))
+ )
return Response(serializer.data, headers=self.get_openrosa_headers())
def retrieve(self, request, *args, **kwargs):
- self.object = self.get_object()
- return Response(self.object.xml_with_disclaimer, headers=self.get_openrosa_headers())
+ xform = self.get_object()
+
+ return Response(
+ xform.xml_with_disclaimer, headers=self.get_openrosa_headers()
+ )
@action(detail=True, methods=['GET'])
def manifest(self, request, *args, **kwargs):
- self.object = self.get_object()
+ xform = self.get_object()
media_files = {}
expired_objects = False
@@ -141,7 +135,7 @@ def manifest(self, request, *args, **kwargs):
# Retrieve all media files for the current form
queryset = MetaData.objects.filter(
- data_type__in=MetaData.MEDIA_FILES_TYPE, xform=self.object
+ data_type__in=MetaData.MEDIA_FILES_TYPE, xform=xform
)
object_list = queryset.all()
@@ -168,15 +162,18 @@ def manifest(self, request, *args, **kwargs):
# > "A new version of this form has been downloaded"
media_files = dict(sorted(media_files.items()))
context = self.get_serializer_context()
- serializer = XFormManifestSerializer(media_files.values(),
- many=True,
- context=context)
+ serializer = XFormManifestSerializer(
+ media_files.values(),
+ many=True,
+ context=context,
+ require_auth=not bool(kwargs.get('username')),
+ )
return Response(serializer.data, headers=self.get_openrosa_headers())
@action(detail=True, methods=['GET'])
def media(self, request, *args, **kwargs):
- self.object = self.get_object()
+ xform = self.get_object()
pk = kwargs.get('metadata')
if not pk:
@@ -185,7 +182,7 @@ def media(self, request, *args, **kwargs):
meta_obj = get_object_or_404(
MetaData,
data_type__in=MetaData.MEDIA_FILES_TYPE,
- xform=self.object,
+ xform=xform,
pk=pk,
)
diff --git a/onadata/apps/api/viewsets/xform_submission_api.py b/onadata/apps/api/viewsets/xform_submission_api.py
index 50faf4777..741220dbe 100644
--- a/onadata/apps/api/viewsets/xform_submission_api.py
+++ b/onadata/apps/api/viewsets/xform_submission_api.py
@@ -20,7 +20,6 @@
from rest_framework.renderers import BrowsableAPIRenderer, JSONRenderer
from onadata.apps.logger.models import Instance
-from onadata.apps.main.models.user_profile import UserProfile
from onadata.libs import filters
from onadata.libs.authentication import DigestAuthentication
from onadata.libs.mixins.openrosa_headers_mixin import OpenRosaHeadersMixin
@@ -60,7 +59,6 @@ def create_instance_from_xml(username, request):
xml_file_list = request.FILES.pop('xml_submission_file', [])
xml_file = xml_file_list[0] if len(xml_file_list) else None
media_files = request.FILES.values()
-
return safe_create_instance(username, xml_file, media_files, None, request)
@@ -170,16 +168,14 @@ def __init__(self, *args, **kwargs):
def create(self, request, *args, **kwargs):
username = self.kwargs.get('username')
+
if self.request.user.is_anonymous:
- if username is None:
+ if not username:
# Authentication is mandatory when username is omitted from the
# submission URL
raise NotAuthenticated
else:
- user = get_object_or_404(User, username=username.lower())
- profile, created = UserProfile.objects.get_or_create(user=user)
- if profile.require_auth:
- raise NotAuthenticated
+ _ = get_object_or_404(User, username=username.lower())
elif not username:
# get the username from the user if not set
username = request.user and get_real_user(request).username
diff --git a/onadata/apps/logger/management/commands/import_briefcase.py b/onadata/apps/logger/management/commands/import_briefcase.py
index 7d4714332..a8f01c66a 100644
--- a/onadata/apps/logger/management/commands/import_briefcase.py
+++ b/onadata/apps/logger/management/commands/import_briefcase.py
@@ -23,7 +23,6 @@ def add_arguments(self, parser):
parser.add_argument('--to',
help="username in this server")
-
def handle(self, *args, **kwargs):
url = kwargs.get('url')
username = kwargs.get('username')
diff --git a/onadata/apps/logger/management/commands/update_attachment_storage_bytes.py b/onadata/apps/logger/management/commands/update_attachment_storage_bytes.py
index ad7ff24bb..88533418f 100644
--- a/onadata/apps/logger/management/commands/update_attachment_storage_bytes.py
+++ b/onadata/apps/logger/management/commands/update_attachment_storage_bytes.py
@@ -4,7 +4,6 @@
from django.contrib.auth import get_user_model
from django.core.management.base import BaseCommand
from django.db.models import Sum, OuterRef, Subquery
-from django.db.models.functions import Coalesce
from onadata.apps.logger.models.attachment import Attachment
from onadata.apps.logger.models.xform import XForm
diff --git a/onadata/apps/logger/migrations/0034_set_require_auth_at_project_level.py b/onadata/apps/logger/migrations/0034_set_require_auth_at_project_level.py
new file mode 100644
index 000000000..652bc7dd6
--- /dev/null
+++ b/onadata/apps/logger/migrations/0034_set_require_auth_at_project_level.py
@@ -0,0 +1,172 @@
+# Generated by Django 3.2.15 on 2023-11-02 15:30
+from itertools import islice
+from urllib.parse import urlparse
+
+from django.conf import settings
+from django.db import migrations
+from django_redis import get_redis_connection
+
+
+CHUNK_SIZE = 2000
+
+
+def restore_open_rosa_server_in_redis(apps, schema_editor):
+
+ if settings.SKIP_HEAVY_MIGRATIONS:
+ return
+
+ print(
+ """
+ This migration might take a while. If it is too slow, you may want to
+ re-run migrations with SKIP_HEAVY_MIGRATIONS=True and apply this one
+ manually from the django shell.
+ """
+ )
+
+ XForm = apps.get_model('logger', 'XForm') # noqa
+
+ parsed_url = urlparse(settings.KOBOCAT_URL)
+ server_url = settings.KOBOCAT_URL.rstrip('/')
+
+ xforms_iter = (
+ XForm.objects.filter(require_auth=False)
+ .values('id_string', 'user__username')
+ .iterator(chunk_size=CHUNK_SIZE)
+ )
+
+ while True:
+ xforms = list(islice(xforms_iter, CHUNK_SIZE))
+ if not xforms:
+ break
+ keys = []
+ for xform in xforms:
+ username = xform['user__username']
+ id_string = xform['id_string']
+ keys.append(f"or:{parsed_url.netloc}/{username},{id_string}|{username}")
+
+ lua_keys = '", "'.join(keys)
+
+ lua_script = f"""
+ local keys = {{"{lua_keys}"}}
+ for _, key in ipairs(keys) do
+ local redis_real_key = string.sub(key, 1, string.find(key, '|') - 1)
+ local username = string.sub(key, string.find(key, '|') + 1, string.len(key))
+ local ee_id = redis.call('get', redis_real_key)
+ if ee_id then
+ redis.call('hset', 'id:' .. ee_id, 'openRosaServer', '{server_url}/' .. username)
+ end
+ end
+ """
+ redis_client = get_redis_connection('enketo_redis_main')
+ pipeline = redis_client.pipeline()
+ pipeline.eval(lua_script, 0)
+ pipeline.execute()
+
+
+def restore_require_auth_at_profile_level(apps, schema_editor):
+
+ print(
+ """
+ Restoring authentication at the profile level cannot be done
+ automatically.
+
+ This is an example of what can be done:
+ ⚠️ WARNING ⚠️ The example below makes all projects publicly
+ accessible when profile level is restored even if, at least, one project
+ was publicly accessible at project level.
+
+ ```python
+ UserProfile.objects.filter(
+ user_id__in=XForm.objects.filter(require_auth=False).values_list(
+ 'user_id'
+ )
+ ).update(require_auth=False)
+ XForm.objects.filter(require_auth=True).update(require_auth=False)
+ ```
+ """
+ )
+
+
+def set_require_auth_at_project_level(apps, schema_editor):
+
+ if settings.SKIP_HEAVY_MIGRATIONS:
+ return
+
+ print(
+ """
+ This migration might take a while. If it is too slow, you may want to
+ re-run migrations with SKIP_HEAVY_MIGRATIONS=True and apply this one
+ manually from the django shell.
+ """
+ )
+
+ XForm = apps.get_model('logger', 'XForm') # noqa
+ UserProfile = apps.get_model('main', 'UserProfile') # noqa
+
+ XForm.objects.all().update(require_auth=True)
+ XForm.objects.filter(
+ user_id__in=UserProfile.objects.filter(require_auth=False).values_list(
+ 'user_id'
+ )
+ ).update(require_auth=False)
+
+
+def update_open_rosa_server_in_redis(apps, schema_editor):
+
+ if settings.SKIP_HEAVY_MIGRATIONS:
+ return
+
+ XForm = apps.get_model('logger', 'XForm') # noqa
+
+ parsed_url = urlparse(settings.KOBOCAT_URL)
+ server_url = settings.KOBOCAT_URL.strip('/')
+
+ xforms_iter = (
+ XForm.objects.filter(user__profile__require_auth=False)
+ .values('id_string', 'user__username')
+ .iterator(chunk_size=CHUNK_SIZE)
+ )
+
+ while True:
+ xforms = list(islice(xforms_iter, CHUNK_SIZE))
+ if not xforms:
+ break
+ keys = []
+ for xform in xforms:
+ username = xform['user__username']
+ id_string = xform['id_string']
+ keys.append(f"or:{parsed_url.netloc}/{username},{id_string}")
+
+ lua_keys = '", "'.join(keys)
+
+ lua_script = f"""
+ local keys = {{"{lua_keys}"}}
+ for _, key in ipairs(keys) do
+ local ee_id = redis.call('get', key)
+ if ee_id then
+ redis.call('hset', 'id:' .. ee_id, 'openRosaServer', '{server_url}')
+ end
+ end
+ """
+ redis_client = get_redis_connection('enketo_redis_main')
+ pipeline = redis_client.pipeline()
+ pipeline.eval(lua_script, 0)
+ pipeline.execute()
+
+
+class Migration(migrations.Migration):
+
+ dependencies = [
+ ('logger', '0033_add_deleted_at_field_to_attachment'),
+ ]
+
+ operations = [
+ migrations.RunPython(
+ set_require_auth_at_project_level,
+ restore_require_auth_at_profile_level,
+ ),
+ migrations.RunPython(
+ update_open_rosa_server_in_redis,
+ restore_open_rosa_server_in_redis,
+ ),
+ ]
diff --git a/onadata/apps/logger/models/xform.py b/onadata/apps/logger/models/xform.py
index 84995d3ec..966543ca9 100644
--- a/onadata/apps/logger/models/xform.py
+++ b/onadata/apps/logger/models/xform.py
@@ -68,7 +68,10 @@ class XForm(BaseModel):
xml = models.TextField()
user = models.ForeignKey(User, related_name='xforms', null=True, on_delete=models.CASCADE)
- require_auth = models.BooleanField(default=False)
+ require_auth = models.BooleanField(
+ default=True,
+ verbose_name=t('Require authentication to see form and submit data'),
+ )
shared = models.BooleanField(default=False)
shared_data = models.BooleanField(default=False)
downloadable = models.BooleanField(default=True)
@@ -123,10 +126,10 @@ def file_name(self):
def url(self):
return reverse(
- "download_xform",
+ 'download_xform',
kwargs={
- "username": self.user.username,
- "id_string": self.id_string
+ 'username': self.user.username,
+ 'pk': self.pk
}
)
diff --git a/onadata/apps/logger/tests/models/test_instance.py b/onadata/apps/logger/tests/models/test_instance.py
index 9d8f9d9f0..66eda9d37 100644
--- a/onadata/apps/logger/tests/models/test_instance.py
+++ b/onadata/apps/logger/tests/models/test_instance.py
@@ -47,10 +47,6 @@ def test_json_stores_user_attribute(self, mock_time):
mock_time.return_value = datetime.utcnow().replace(tzinfo=utc)
self._publish_transportation_form()
- # make account require phone auth
- self.user.profile.require_auth = True
- self.user.profile.save()
-
# submit instance with a request user
path = os.path.join(
self.this_directory, 'fixtures', 'transportation', 'instances',
diff --git a/onadata/apps/logger/tests/test_briefcase_client.py b/onadata/apps/logger/tests/test_briefcase_client.py
index 464e0a3cf..f393914e2 100644
--- a/onadata/apps/logger/tests/test_briefcase_client.py
+++ b/onadata/apps/logger/tests/test_briefcase_client.py
@@ -1,20 +1,17 @@
# coding: utf-8
import os.path
from io import BytesIO
-from urllib.parse import urljoin
import requests
from django.contrib.auth import authenticate
from django.core.files.storage import default_storage as storage
from django.core.files.uploadedfile import UploadedFile
-from django.urls import reverse
from django.test import RequestFactory
from django_digest.test import Client as DigestClient
from httmock import urlmatch, HTTMock
from onadata.apps.api.viewsets.xform_list_api import XFormListApi
from onadata.apps.logger.models import Instance, XForm
-from onadata.apps.logger.views import download_xform
from onadata.apps.main.models import MetaData
from onadata.apps.main.tests.test_base import TestBase
from onadata.libs.utils.briefcase_client import BriefcaseClient
@@ -27,6 +24,11 @@ def formList(*args, **kwargs): # noqa
response.render()
return response
+def xformsDownload(*args, **kwargs): # noqa
+ view = XFormListApi.as_view({'get': 'retrieve'})
+ response = view(*args, **kwargs)
+ response.render()
+ return response
def xformsManifest(*args, **kwargs): # noqa
view = XFormListApi.as_view({'get': 'manifest'})
@@ -46,8 +48,6 @@ def form_list_xml(url, request, **kwargs):
factory = RequestFactory()
req = factory.get(url.path)
req.user = authenticate(username='bob', password='bob')
- req.user.profile.require_auth = False
- req.user.profile.save()
id_string = 'transportation_2011_07_25'
# Retrieve XForm pk for user bob.
# SQLite resets PK to 1 every time the table is truncated (i.e. after
@@ -59,20 +59,20 @@ def form_list_xml(url, request, **kwargs):
.last()
)
if url.path.endswith('formList'):
- res = formList(req, username='bob')
+ res = formList(req)
elif url.path.endswith('form.xml'):
- res = download_xform(req, username='bob', id_string=id_string)
+ res = xformsDownload(req, pk=xform_id)
elif url.path.find('xformsManifest') > -1:
- res = xformsManifest(req, username='bob', pk=xform_id)
+ res = xformsManifest(req, pk=xform_id)
elif url.path.find('xformsMedia') > -1:
filename = url.path[url.path.rfind('/') + 1:]
metadata_id, _ = os.path.splitext(filename)
res = xformsMedia(
- req, username='bob', pk=xform_id, metadata=metadata_id
+ req, pk=xform_id, metadata=metadata_id
)
response._content = get_streaming_content(res)
else:
- res = formList(req, username='bob')
+ res = formList(req)
response.status_code = 200
if not response._content:
response._content = res.content
@@ -116,15 +116,12 @@ def setUp(self):
count = MetaData.objects.count()
MetaData.media_upload(self.xform, uf)
self.assertEqual(MetaData.objects.count(), count + 1)
- url = urljoin(
- self.base_url,
- reverse('user_profile', kwargs={'username': self.user.username})
- )
self._logout()
self._create_user_and_login('deno', 'deno')
+
self.bc = BriefcaseClient(
username='bob', password='bob',
- url=url,
+ url=self.base_url,
user=self.user
)
@@ -189,6 +186,7 @@ def test_push(self):
self.bc.download_xforms()
with HTTMock(instances_xml):
self.bc.download_instances(self.xform.id_string)
+
XForm.objects.all().delete()
xforms = XForm.objects.filter(
user=self.user, id_string=self.xform.id_string)
diff --git a/onadata/apps/logger/tests/test_digest_authentication.py b/onadata/apps/logger/tests/test_digest_authentication.py
index 7701a3aef..0e7f13bdd 100644
--- a/onadata/apps/logger/tests/test_digest_authentication.py
+++ b/onadata/apps/logger/tests/test_digest_authentication.py
@@ -22,36 +22,34 @@ def test_authenticated_submissions(self):
self.this_directory, 'fixtures',
'transportation', 'instances', s, s + '.xml'
)
- self._set_require_auth()
auth = DigestAuth(self.login_username, self.login_password)
self._make_submission(xml_submission_file_path, add_uuid=True,
auth=auth)
self.assertEqual(self.response.status_code, 201)
- def _set_require_auth(self, auth=True):
- profile, created = \
- UserProfile.objects.get_or_create(user=self.user)
- profile.require_auth = auth
- profile.save()
-
def test_fail_authenticated_submissions_to_wrong_account(self):
username = 'dennis'
- # set require_auth b4 we switch user
- self._set_require_auth()
self._create_user_and_login(username=username, password=username)
- self._set_require_auth()
s = self.surveys[0]
xml_submission_file_path = os.path.join(
self.this_directory, 'fixtures',
'transportation', 'instances', s, s + '.xml'
)
-
- self._make_submission(xml_submission_file_path, add_uuid=True,
- auth=DigestAuth('alice', 'alice'))
+ self._make_submission(
+ xml_submission_file_path,
+ add_uuid=True,
+ auth=DigestAuth('alice', 'alice'),
+ assert_success=False,
+ )
# Authentication required
self.assertEqual(self.response.status_code, 401)
+
auth = DigestAuth('dennis', 'dennis')
- self._make_submission(xml_submission_file_path, add_uuid=True,
- auth=auth)
+ self._make_submission(
+ xml_submission_file_path,
+ add_uuid=True,
+ auth=auth,
+ assert_success=False,
+ )
# Not allowed
self.assertEqual(self.response.status_code, 403)
diff --git a/onadata/apps/logger/tests/test_encrypted_submissions.py b/onadata/apps/logger/tests/test_encrypted_submissions.py
index faff66209..f4a5496ff 100644
--- a/onadata/apps/logger/tests/test_encrypted_submissions.py
+++ b/onadata/apps/logger/tests/test_encrypted_submissions.py
@@ -3,9 +3,9 @@
from django.urls import reverse
from django.contrib.auth import authenticate
+from django_digest.test import DigestAuth
from rest_framework.test import APIRequestFactory
-from onadata.apps.api.viewsets.xform_submission_api import XFormSubmissionApi
from onadata.apps.main.tests.test_base import TestBase
from onadata.apps.logger.models import Attachment
from onadata.apps.logger.models import Instance
@@ -39,16 +39,26 @@ def test_encrypted_submissions(self):
count = Instance.objects.count()
attachments_count = Attachment.objects.count()
+ username = self.user.username # bob
+
with open(files['submission.xml.enc'], 'rb') as ef:
with open(files['submission.xml'], 'rb') as f:
post_data = {
'xml_submission_file': f,
'submission.xml.enc': ef}
self.factory = APIRequestFactory()
+ auth = DigestAuth(username, username)
+ request = self.factory.post(self._submission_url, post_data)
+ request.user = authenticate(username=auth.username, password=auth.password)
+ response = self.submission_view(request, username=username)
+ self.assertEqual(response.status_code, 401)
+
+ f.seek(0)
+ ef.seek(0)
+
request = self.factory.post(self._submission_url, post_data)
- request.user = authenticate(username='bob',
- password='bob')
- response = self.submission_view(request, username=self.user.username)
+ request.META.update(auth(request.META, response))
+ response = self.submission_view(request, username=username)
self.assertContains(response, message, status_code=201)
self.assertEqual(Instance.objects.count(), count + 1)
self.assertEqual(Attachment.objects.count(), attachments_count + 1)
diff --git a/onadata/apps/logger/tests/test_form_list.py b/onadata/apps/logger/tests/test_form_list.py
deleted file mode 100644
index 23c23dd0a..000000000
--- a/onadata/apps/logger/tests/test_form_list.py
+++ /dev/null
@@ -1,50 +0,0 @@
-# coding: utf-8
-from django.contrib.auth.models import AnonymousUser
-from django.test import RequestFactory
-from django_digest.test import DigestAuth
-
-from onadata.apps.api.viewsets.xform_list_api import XFormListApi
-from onadata.apps.main.tests.test_base import TestBase
-
-
-def formList(*args, **kwargs): # noqa
- view = XFormListApi.as_view({'get': 'list'})
- return view(*args, **kwargs)
-
-
-class TestFormList(TestBase):
- def setUp(self):
- super().setUp()
- self.factory = RequestFactory()
-
- def test_returns_200_for_owner(self):
- self._set_require_auth()
- request = self.factory.get('/')
- auth = DigestAuth('bob', 'bob')
- response = formList(request, username=self.user.username)
- request.META.update(auth(request.META, response))
- response = formList(request, username=self.user.username)
- self.assertEqual(response.status_code, 200)
-
- def test_return_401_for_anon_when_require_auth_true(self):
- self._set_require_auth()
- request = self.factory.get('/')
- response = formList(request, username=self.user.username)
- self.assertEqual(response.status_code, 401)
-
- def test_returns_200_for_authenticated_non_owner(self):
- self._set_require_auth()
- credentials = ('alice', 'alice',)
- self._create_user(*credentials)
- auth = DigestAuth('alice', 'alice')
- request = self.factory.get('/')
- response = formList(request, username=self.user.username)
- request.META.update(auth(request.META, response))
- response = formList(request, username=self.user.username)
- self.assertEqual(response.status_code, 200)
-
- def test_show_for_anon_when_require_auth_false(self):
- request = self.factory.get('/')
- request.user = AnonymousUser()
- response = formList(request, username=self.user.username)
- self.assertEqual(response.status_code, 200)
diff --git a/onadata/apps/logger/tests/test_form_submission.py b/onadata/apps/logger/tests/test_form_submission.py
index 958afe143..eba2687c8 100644
--- a/onadata/apps/logger/tests/test_form_submission.py
+++ b/onadata/apps/logger/tests/test_form_submission.py
@@ -2,16 +2,12 @@
import os
import re
-import pytest
-from django.conf import settings
from django.http import Http404
from django_digest.test import DigestAuth
from django_digest.test import Client as DigestClient
from guardian.shortcuts import assign_perm
-from kobo_service_account.utils import get_request_headers
from mock import patch
-from onadata.apps.api.viewsets.xform_viewset import XFormViewSet
from onadata.apps.main.models.user_profile import UserProfile
from onadata.apps.main.tests.test_base import TestBase
from onadata.apps.logger.models import Instance
@@ -90,50 +86,48 @@ def test_fail_with_ioerror_wsgi(self, mock_pop):
def test_submission_to_require_auth_anon(self):
"""
- test submission to a private form by non-owner without perm is
- forbidden.
+ test submission anonymous cannot submit to a private form
"""
- view = XFormViewSet.as_view({
- 'patch': 'partial_update'
- })
- data = {'require_auth': True}
- self.assertFalse(self.xform.require_auth)
- service_account_meta = self.get_meta_from_headers(
- get_request_headers(self.user.username)
- )
- service_account_meta['HTTP_HOST'] = settings.TEST_HTTP_HOST
- request = self.factory.patch('/', data=data, **service_account_meta)
- view(request, pk=self.xform.id)
- self.xform.reload()
- self.assertTrue(self.xform.require_auth)
+ xml_submission_file_path = os.path.join(
+ os.path.dirname(os.path.abspath(__file__)),
+ "../fixtures/tutorial/instances/tutorial_2012-06-27_11-27-53.xml"
+ )
+
+ # Anonymous should authenticate when submit data to `//submission`
+ self._make_submission(
+ xml_submission_file_path, auth=False, assert_success=False
+ )
+ self.assertEqual(self.response.status_code, 401)
+
+ # …or `/submission`
+ self._make_submission(
+ xml_submission_file_path,
+ username='',
+ auth=False,
+ assert_success=False,
+ )
+ self.assertEqual(self.response.status_code, 401)
+
+ def test_submission_to_not_required_auth_as_anonymous_user(self):
+ self.xform.require_auth = False
+ self.xform.save(update_fields=['require_auth'])
xml_submission_file_path = os.path.join(
os.path.dirname(os.path.abspath(__file__)),
"../fixtures/tutorial/instances/tutorial_2012-06-27_11-27-53.xml"
)
- self._make_submission(xml_submission_file_path)
- self.assertEqual(self.response.status_code, 403)
+ # Anonymous should be able to submit data
+ self._make_submission(
+ xml_submission_file_path, auth=False, assert_success=False
+ )
+ self.assertEqual(self.response.status_code, 201)
def test_submission_to_require_auth_without_perm(self):
"""
test submission to a private form by non-owner without perm is
forbidden.
"""
- view = XFormViewSet.as_view({
- 'patch': 'partial_update'
- })
- data = {'require_auth': True}
- self.assertFalse(self.xform.require_auth)
- service_account_meta = self.get_meta_from_headers(
- get_request_headers(self.user.username)
- )
- service_account_meta['HTTP_HOST'] = settings.TEST_HTTP_HOST
- request = self.factory.patch('/', data=data, **service_account_meta)
- view(request, pk=self.xform.id)
- self.xform.reload()
- self.assertTrue(self.xform.require_auth)
-
xml_submission_file_path = os.path.join(
os.path.dirname(os.path.abspath(__file__)),
'../fixtures/tutorial/instances/tutorial_2012-06-27_11-27-53.xml'
@@ -143,31 +137,13 @@ def test_submission_to_require_auth_without_perm(self):
username = 'alice'
self._create_user(username, username)
self._make_submission(
- xml_submission_file_path, auth=DigestAuth('alice', 'alice')
+ xml_submission_file_path,
+ auth=DigestAuth('alice', 'alice'),
+ assert_success=False,
)
self.assertEqual(self.response.status_code, 403)
- @pytest.mark.skip(reason='Send authentication challenge when xform.require_auth is set')
def test_submission_to_require_auth_with_perm(self):
- """
- test submission to a private form by non-owner is forbidden.
-
- TODO send authentication challenge when xform.require_auth is set.
- This is non-trivial because we do not know the xform until we have
- parsed the XML.
- """
-
- view = XFormViewSet.as_view({
- 'patch': 'partial_update'
- })
- data = {'require_auth': True}
- self.assertFalse(self.xform.require_auth)
- request = self.factory.patch('/', data=data, **{
- 'HTTP_AUTHORIZATION': 'Token %s' % self.user.auth_token})
- view(request, pk=self.xform.id)
- self.xform.reload()
- self.assertTrue(self.xform.require_auth)
-
# create a new user
username = 'alice'
alice = self._create_user(username, username)
@@ -189,7 +165,9 @@ def test_form_post_to_missing_form(self):
"../fixtures/tutorial/instances/"
"tutorial_invalid_id_string_2012-06-27_11-27-53.xml"
)
- self._make_submission(path=xml_submission_file_path)
+ self._make_submission(
+ path=xml_submission_file_path, assert_success=False
+ )
self.assertEqual(self.response.status_code, 404)
def test_duplicate_submissions(self):
@@ -220,8 +198,6 @@ def test_unicode_submission(self):
"..", "fixtures", "tutorial", "instances",
"tutorial_unicode_submission.xml"
)
- self.user.profile.require_auth = True
- self.user.profile.save()
# create a new user
alice = self._create_user('alice', 'alice')
@@ -365,7 +341,9 @@ def test_fail_submission_if_bad_id_string(self):
"..", "fixtures", "tutorial", "instances",
"tutorial_2012-06-27_11-27-53_bad_id_string.xml"
)
- self._make_submission(path=xml_submission_file_path)
+ self._make_submission(
+ path=xml_submission_file_path, assert_success=False
+ )
self.assertEqual(self.response.status_code, 404)
def test_edit_updated_geopoint_cache(self):
@@ -406,9 +384,6 @@ def test_edit_updated_geopoint_cache(self):
self.assertEqual(float(gps[1]), float(cached_geopoint[1]))
def test_submission_when_requires_auth(self):
- self.user.profile.require_auth = True
- self.user.profile.save()
-
# create a new user
alice = self._create_user('alice', 'alice')
@@ -425,9 +400,6 @@ def test_submission_when_requires_auth(self):
self.assertEqual(self.response.status_code, 201)
def test_submission_linked_to_reporter(self):
- self.user.profile.require_auth = True
- self.user.profile.save()
-
# create a new user
alice = self._create_user('alice', 'alice')
UserProfile.objects.create(user=alice)
@@ -466,18 +438,24 @@ def test_anonymous_cannot_edit_submissions(self):
"..", "fixtures", "tutorial", "instances",
"tutorial_2012-06-27_11-27-53_w_uuid_edited.xml"
)
- # …without "Require authentication to see forms and submit data"
- self.assertFalse(self.user.profile.require_auth)
- self._make_submission(xml_submission_file_path, auth=False)
+ # …without "Require authentication to see form and submit data"
+ self.xform.require_auth = False
+ self.xform.save(update_fields=['require_auth'])
+
+ self._make_submission(
+ xml_submission_file_path, auth=False, assert_success=False
+ )
self.assertEqual(self.response.status_code, 401)
self.assertEqual(
Instance.objects.order_by('pk').last().xml_hash,
created_instance.xml_hash,
)
# …now with "Require authentication to…"
- self.user.profile.require_auth = True
- self.user.profile.save()
- self._make_submission(xml_submission_file_path, auth=False)
+ self.xform.require_auth = True
+ self.xform.save(update_fields=['require_auth'])
+ self._make_submission(
+ xml_submission_file_path, auth=False, assert_success=False
+ )
self.assertEqual(self.response.status_code, 401)
self.assertEqual(
Instance.objects.order_by('pk').last().xml_hash,
@@ -497,8 +475,6 @@ def test_authorized_user_can_edit_submissions_without_require_auth(self):
submissions at the same time.
"""
- self.assertFalse(self.user.profile.require_auth)
-
xml_submission_file_path = os.path.join(
os.path.dirname(os.path.abspath(__file__)),
"..", "fixtures", "tutorial", "instances",
@@ -535,8 +511,6 @@ def test_authorized_user_can_edit_submissions_without_require_auth(self):
)
def test_authorized_user_can_edit_submissions_with_require_auth(self):
- self.user.profile.require_auth = True
- self.user.profile.save()
xml_submission_file_path = os.path.join(
os.path.dirname(os.path.abspath(__file__)),
@@ -599,18 +573,23 @@ def test_unauthorized_cannot_edit_submissions(self):
"..", "fixtures", "tutorial", "instances",
"tutorial_2012-06-27_11-27-53_w_uuid_edited.xml"
)
- # …without "Require authentication to see forms and submit data"
- self.assertFalse(self.user.profile.require_auth)
- self._make_submission(xml_submission_file_path, auth=auth)
+ # …without "Require authentication to see form and submit data"
+ self.xform.require_auth = False
+ self.xform.save(update_fields=['require_auth'])
+ self._make_submission(
+ xml_submission_file_path, auth=auth, assert_success=False
+ )
self.assertEqual(self.response.status_code, 403)
self.assertEqual(
Instance.objects.order_by('pk').last().xml_hash,
created_instance.xml_hash,
)
# …now with "Require authentication to…"
- self.user.profile.require_auth = True
- self.user.profile.save()
- self._make_submission(xml_submission_file_path, auth=auth)
+ self.xform.require_auth = True
+ self.xform.save(update_fields=['require_auth'])
+ self._make_submission(
+ xml_submission_file_path, auth=auth, assert_success=False
+ )
self.assertEqual(self.response.status_code, 403)
self.assertEqual(
Instance.objects.order_by('pk').last().xml_hash,
diff --git a/onadata/apps/logger/tests/test_instance_creation.py b/onadata/apps/logger/tests/test_instance_creation.py
index 49c7e2e11..031a3735c 100644
--- a/onadata/apps/logger/tests/test_instance_creation.py
+++ b/onadata/apps/logger/tests/test_instance_creation.py
@@ -45,9 +45,7 @@ class TestInstanceCreation(TestCase):
def setUp(self):
self.user = User.objects.create(username="bob")
- profile, c = UserProfile.objects.get_or_create(user=self.user)
- profile.require_auth = False
- profile.save()
+ _ = UserProfile.objects.get_or_create(user=self.user)
absolute_path = get_absolute_path("forms")
open_forms = open_all_files(absolute_path)
diff --git a/onadata/apps/logger/tests/test_publish_xls.py b/onadata/apps/logger/tests/test_publish_xls.py
index e6d12d964..b4a0e1725 100644
--- a/onadata/apps/logger/tests/test_publish_xls.py
+++ b/onadata/apps/logger/tests/test_publish_xls.py
@@ -21,7 +21,7 @@ def test_publish_xls(self):
call_command('publish_xls', xls_file_path, self.user.username)
self.assertEqual(XForm.objects.count(), count + 1)
form = XForm.objects.get()
- self.assertFalse(form.require_auth)
+ self.assertTrue(form.require_auth)
def test_publish_xls_replacement(self):
count = XForm.objects.count()
diff --git a/onadata/apps/logger/tests/test_simple_submission.py b/onadata/apps/logger/tests/test_simple_submission.py
index ae875f10f..a796e9af5 100644
--- a/onadata/apps/logger/tests/test_simple_submission.py
+++ b/onadata/apps/logger/tests/test_simple_submission.py
@@ -4,12 +4,12 @@
from pyxform import SurveyElementBuilder
from onadata.apps.logger.xform_instance_parser import DuplicateInstance
+from onadata.apps.main.models.user_profile import UserProfile
from onadata.apps.viewer.models.data_dictionary import DataDictionary
from onadata.libs.utils.logger_tools import (
create_instance, safe_create_instance
)
-
class TempFileProxy:
"""
create_instance will be looking for a file object,
@@ -51,6 +51,8 @@ def setUp(self):
self.user = User.objects.create(
username="admin", email="sample@example.com")
self.user.set_password("pass")
+ UserProfile.objects.get_or_create(user=self.user)
+
self.xform1 = DataDictionary()
self.xform1.user = self.user
self.xform1.json = '{"id_string": "yes_or_no", "children": [{"name": '\
@@ -72,6 +74,11 @@ def test_start_time_boolean_properly_set(self):
self.assertTrue(self.xform2.has_start_time)
def test_simple_yes_submission(self):
+ # Set require_auth to False to avoid permissions check.
+ # Other tests cover them.
+ self.xform1.require_auth = False
+ self.xform1.save(update_fields=['require_auth'])
+
self.assertEqual(0, self.xform1.instances.count())
self._submit_simple_yes()
@@ -88,6 +95,11 @@ def test_start_time_submissions(self):
*with start_time available* are marked as duplicates when the XML is a
direct match.
"""
+ # Set require_auth to False to avoid permissions check.
+ # Other tests cover them.
+ self.xform2.require_auth = False
+ self.xform2.save(update_fields=['require_auth'])
+
self.assertEqual(0, self.xform2.instances.count())
self._submit_at_hour(11)
self.assertEqual(1, self.xform2.instances.count())
diff --git a/onadata/apps/logger/views.py b/onadata/apps/logger/views.py
index 7139e9b68..79c5bd9fe 100644
--- a/onadata/apps/logger/views.py
+++ b/onadata/apps/logger/views.py
@@ -148,33 +148,6 @@ def bulksubmission_form(request, username=None):
return HttpResponseRedirect('/%s' % request.user.username)
-def download_xform(request, username, id_string):
- user = get_object_or_404(User, username__iexact=username)
- xform = get_object_or_404(XForm,
- user=user, id_string__exact=id_string)
- profile, created = UserProfile.objects.get_or_create(user=user)
-
- if (
- profile.require_auth
- and (digest_response := digest_authentication(request))
- ):
- return digest_response
-
- audit = {
- "xform": xform.id_string
- }
- audit_log(
- Actions.FORM_XML_DOWNLOADED, request.user, xform.user,
- t("Downloaded XML for form '%(id_string)s'.") %
- {
- "id_string": xform.id_string
- }, audit, request)
- response = response_with_mimetype_and_name('xml', id_string,
- show_date=False)
- response.content = xform.xml
- return response
-
-
def download_xlsform(request, username, id_string):
xform = get_object_or_404(XForm,
user__username__iexact=username,
diff --git a/onadata/apps/main/models/user_profile.py b/onadata/apps/main/models/user_profile.py
index 7737ddae3..83e742f01 100644
--- a/onadata/apps/main/models/user_profile.py
+++ b/onadata/apps/main/models/user_profile.py
@@ -27,12 +27,9 @@ class UserProfile(models.Model):
home_page = models.CharField(max_length=255, blank=True)
twitter = models.CharField(max_length=255, blank=True)
description = models.CharField(max_length=255, blank=True)
- require_auth = models.BooleanField(
- default=False,
- verbose_name=gettext_lazy(
- "Require authentication to see forms and submit data"
- )
- )
+ # TODO Remove this field (`require_auth`) in the next release following the one where
+ # this commit has been deployed to production
+ require_auth = models.BooleanField(default=True)
address = models.CharField(max_length=255, blank=True)
phonenumber = models.CharField(max_length=30, blank=True)
num_of_submissions = models.IntegerField(default=0)
@@ -87,20 +84,6 @@ def set_object_permissions(sender, instance=None, created=False, **kwargs):
dispatch_uid='set_object_permissions')
-def default_user_profile_require_auth(
- sender, instance, created, raw, **kwargs):
- if raw or not created:
- return
- instance.require_auth = \
- settings.REQUIRE_AUTHENTICATION_TO_SEE_FORMS_AND_SUBMIT_DATA_DEFAULT
- instance.save()
-
-
-post_save.connect(default_user_profile_require_auth,
- sender=UserProfile,
- dispatch_uid='default_user_profile_require_auth')
-
-
def get_anonymous_user_instance(User):
"""
Force `AnonymousUser` to be saved with `pk` == `ANONYMOUS_USER_ID`
diff --git a/onadata/apps/main/tests/test_base.py b/onadata/apps/main/tests/test_base.py
index 5e3365dcd..9eaa9b163 100644
--- a/onadata/apps/main/tests/test_base.py
+++ b/onadata/apps/main/tests/test_base.py
@@ -78,10 +78,8 @@ def _create_user_and_login(self, username="bob", password="bob"):
self.login_password = password
self.user = self._create_user(username, password)
- # create user profile and set require_auth to false for tests
- profile, created = UserProfile.objects.get_or_create(user=self.user)
- profile.require_auth = False
- profile.save()
+ # create user profile if it does not exist
+ UserProfile.objects.get_or_create(user=self.user)
self.client = self._login(username, password)
self.anon = Client()
@@ -222,14 +220,3 @@ def _get_response_content(self, response):
def _set_mock_time(self, mock_time):
current_time = timezone.now()
mock_time.return_value = current_time
-
- def _set_require_auth(self, auth=True):
- profile, created = UserProfile.objects.get_or_create(user=self.user)
- profile.require_auth = auth
- profile.save()
-
- def _get_digest_client(self):
- self._set_require_auth(True)
- client = DigestClient()
- client.set_authorization('bob', 'bob', 'Digest')
- return client
diff --git a/onadata/apps/main/tests/test_form_show.py b/onadata/apps/main/tests/test_form_show.py
index 28e06a521..625042403 100644
--- a/onadata/apps/main/tests/test_form_show.py
+++ b/onadata/apps/main/tests/test_form_show.py
@@ -9,7 +9,6 @@
from onadata.apps.logger.views import (
download_xlsform,
download_jsonform,
- download_xform,
)
from onadata.libs.utils.logger_tools import publish_xml_form
from onadata.libs.utils.user_auth import http_auth_string
@@ -107,34 +106,6 @@ def test_dl_json_for_cors_options(self):
self.assertEqual(response['Access-Control-Allow-Methods'], 'GET')
self.assertEqual(response['Access-Control-Allow-Origin'], '*')
- def test_dl_xform_to_anon_if_public(self):
- self.xform.shared = True
- self.xform.save()
- response = self.anon.get(reverse(download_xform, kwargs={
- 'username': self.user.username,
- 'id_string': self.xform.id_string
- }))
- self.assertEqual(response.status_code, 200)
-
- def test_dl_xform_for_basic_auth(self):
- extra = {
- 'HTTP_AUTHORIZATION':
- http_auth_string(self.login_username, self.login_password)
- }
- response = self.anon.get(reverse(download_xform, kwargs={
- 'username': self.user.username,
- 'id_string': self.xform.id_string
- }), **extra)
- self.assertEqual(response.status_code, 200)
-
- def test_dl_xform_for_authenticated_non_owner(self):
- self._create_user_and_login('alice', 'alice')
- response = self.client.get(reverse(download_xform, kwargs={
- 'username': 'bob',
- 'id_string': self.xform.id_string
- }))
- self.assertEqual(response.status_code, 200)
-
def test_publish_xml_xlsform_download(self):
count = XForm.objects.count()
path = os.path.join(
diff --git a/onadata/apps/main/tests/test_process.py b/onadata/apps/main/tests/test_process.py
index a921f81b5..b29a74be9 100644
--- a/onadata/apps/main/tests/test_process.py
+++ b/onadata/apps/main/tests/test_process.py
@@ -5,7 +5,6 @@
import os
import re
import unittest
-from io import BytesIO
from xml.dom import Node
from defusedxml import minidom
@@ -13,7 +12,6 @@
from django.conf import settings
from django_digest.test import Client as DigestClient
from django.core.files.uploadedfile import UploadedFile
-from openpyxl import load_workbook
from onadata.apps.main.models import MetaData
from onadata.apps.logger.models import XForm
diff --git a/onadata/apps/main/urls.py b/onadata/apps/main/urls.py
index 11869ae1d..5c310443b 100644
--- a/onadata/apps/main/urls.py
+++ b/onadata/apps/main/urls.py
@@ -25,7 +25,6 @@
from onadata.apps.logger.views import (
bulksubmission,
bulksubmission_form,
- download_xform,
download_xlsform,
download_jsonform,
)
@@ -68,16 +67,16 @@
),
# briefcase api urls
- re_path(r"^(?P\w+)/view/submissionList$",
+ re_path(r"^view/submissionList$",
BriefcaseApi.as_view({'get': 'list', 'head': 'list'}),
name='view-submission-list'),
- re_path(r"^(?P\w+)/view/downloadSubmission$",
+ re_path(r"^view/downloadSubmission$",
BriefcaseApi.as_view({'get': 'retrieve', 'head': 'retrieve'}),
name='view-download-submission'),
- re_path(r"^(?P\w+)/formUpload$",
+ re_path(r"^formUpload$",
BriefcaseApi.as_view({'post': 'create', 'head': 'create'}),
name='form-upload'),
- re_path(r"^(?P\w+)/upload$",
+ re_path(r"^upload$",
BriefcaseApi.as_view({'post': 'create', 'head': 'create'}),
name='upload'),
@@ -126,11 +125,12 @@
bulksubmission),
re_path(r"^(?P\w+)/bulk-submission-form$",
bulksubmission_form),
- re_path(r"^(?P\w+)/forms/(?P[\d+^/]+)/form\.xml$",
+ re_path(r'^forms/(?P[\d+^/]+)/form\.xml$',
XFormListApi.as_view({'get': 'retrieve'}),
- name="download_xform"),
- re_path(r"^(?P\w+)/forms/(?P[^/]+)/form\.xml$",
- download_xform, name="download_xform"),
+ name='download_xform'),
+ re_path(r'^(?P\w+)/forms/(?P[\d+^/]+)/form\.xml$',
+ XFormListApi.as_view({'get': 'retrieve'}),
+ name='download_xform'),
re_path(r"^(?P\w+)/forms/(?P[^/]+)/form\.xls$",
download_xlsform,
name="download_xlsform"),
diff --git a/onadata/apps/viewer/tests/test_attachment_url.py b/onadata/apps/viewer/tests/test_attachment_url.py
index 57dcb237c..37cec310b 100644
--- a/onadata/apps/viewer/tests/test_attachment_url.py
+++ b/onadata/apps/viewer/tests/test_attachment_url.py
@@ -1,12 +1,9 @@
# coding: utf-8
-import requests
from django.urls import reverse
-from django_digest.test import DigestAuth
from django_digest.test import Client as DigestClient
from onadata.apps.main.tests.test_base import TestBase
from onadata.apps.logger.models import Attachment
-from onadata.apps.viewer.views import attachment_url
from onadata.libs.utils.storage import rmdir
diff --git a/onadata/apps/viewer/tests/test_pandas_mongo_bridge.py b/onadata/apps/viewer/tests/test_pandas_mongo_bridge.py
index e64195dee..124e28871 100644
--- a/onadata/apps/viewer/tests/test_pandas_mongo_bridge.py
+++ b/onadata/apps/viewer/tests/test_pandas_mongo_bridge.py
@@ -551,11 +551,7 @@ def test_csv_export_with_df_size_limit(self):
os.unlink(temp_file.name)
def test_csv_column_indices_in_groups_within_repeats(self):
- # By default, `_create_user_and_login()` creates users without
- # authentication required. We force it when submitting data to persist
- # collector's username because this test expects it.
# See `_submitted_data` in `expected_data_0` below
- self._set_require_auth(auth=True)
self._publish_xls_fixture_set_xform("groups_in_repeats")
self._submit_fixture_instance("groups_in_repeats", "01")
dd = self.xform.data_dictionary()
diff --git a/onadata/libs/serializers/xform_serializer.py b/onadata/libs/serializers/xform_serializer.py
index d3629d273..c4c4a5865 100644
--- a/onadata/libs/serializers/xform_serializer.py
+++ b/onadata/libs/serializers/xform_serializer.py
@@ -105,6 +105,10 @@ class XFormListSerializer(serializers.Serializer):
downloadUrl = serializers.SerializerMethodField('get_url')
manifestUrl = serializers.SerializerMethodField('get_manifest_url')
+ def __init__(self, *args, **kwargs):
+ self._require_auth = kwargs.pop('require_auth', None)
+ super().__init__(*args, **kwargs)
+
class Meta:
fields = '__all__'
@@ -133,14 +137,19 @@ def get_hash(self, obj):
@check_obj
def get_url(self, obj):
- kwargs = {'pk': obj.pk, 'username': obj.user.username}
+ kwargs = {'pk': obj.pk}
+ if not self._require_auth:
+ kwargs['username'] = obj.user.username
+
request = self.context.get('request')
return reverse('download_xform', kwargs=kwargs, request=request)
@check_obj
def get_manifest_url(self, obj):
- kwargs = {'pk': obj.pk, 'username': obj.user.username}
+ kwargs = {'pk': obj.pk}
+ if not self._require_auth:
+ kwargs['username'] = obj.user.username
request = self.context.get('request')
return reverse('manifest-url', kwargs=kwargs, request=request)
@@ -152,6 +161,10 @@ class XFormManifestSerializer(serializers.Serializer):
hash = serializers.SerializerMethodField()
downloadUrl = serializers.SerializerMethodField('get_url')
+ def __init__(self, *args, **kwargs):
+ self._require_auth = kwargs.pop('require_auth', None)
+ super().__init__(*args, **kwargs)
+
def get_filename(self, obj):
# If file has been synchronized from KPI and it is a remote URL,
# manifest.xml should return only the name, not the full URL.
@@ -165,9 +178,10 @@ def get_filename(self, obj):
@check_obj
def get_url(self, obj):
- kwargs = {'pk': obj.xform.pk,
- 'username': obj.xform.user.username,
- 'metadata': obj.pk}
+ kwargs = {'pk': obj.xform.pk, 'metadata': obj.pk}
+ if not self._require_auth:
+ kwargs['username'] = obj.xform.user.username
+
request = self.context.get('request')
_, extension = os.path.splitext(obj.filename)
# if `obj` is a remote url, it is possible it does not have any
diff --git a/onadata/libs/tests/mixins/make_submission_mixin.py b/onadata/libs/tests/mixins/make_submission_mixin.py
index 269578d48..f21c7a77c 100644
--- a/onadata/libs/tests/mixins/make_submission_mixin.py
+++ b/onadata/libs/tests/mixins/make_submission_mixin.py
@@ -8,6 +8,7 @@
from django.contrib.auth import authenticate
from django_digest.test import DigestAuth
from kobo_service_account.utils import get_request_headers
+from rest_framework import status
from rest_framework.test import APIRequestFactory
from onadata.apps.api.viewsets.xform_submission_api import XFormSubmissionApi
@@ -46,6 +47,7 @@ def _make_submission(
auth: Union[DigestAuth, bool] = None,
media_file: 'io.BufferedReader' = None,
use_service_account: bool = False,
+ assert_success: bool = True,
):
"""
Pass `auth=False` for an anonymous request, or omit `auth` to perform
@@ -84,12 +86,23 @@ def _make_submission(
password=auth.password)
self.response = None # Reset in case error in viewset below
self.response = self.submission_view(request, username=username)
+
if auth and self.response.status_code == 401:
f.seek(0)
+ if media_file is not None:
+ media_file.seek(0)
+
request = self.factory.post(url, post_data)
request.META.update(auth(request.META, self.response))
self.response = self.submission_view(request, username=username)
+ if assert_success:
+ assert self.response.status_code in [
+ status.HTTP_200_OK,
+ status.HTTP_201_CREATED,
+ status.HTTP_202_ACCEPTED,
+ ]
+
if forced_submission_time:
instance = Instance.objects.order_by('-pk').all()[0]
instance.date_created = forced_submission_time
diff --git a/onadata/libs/utils/briefcase_client.py b/onadata/libs/utils/briefcase_client.py
index 12feec715..5922d58c3 100644
--- a/onadata/libs/utils/briefcase_client.py
+++ b/onadata/libs/utils/briefcase_client.py
@@ -282,6 +282,7 @@ def __init__(self, xml_file, user):
def publish_xform(self):
return publish_xml_form(self.xml_file, self.user)
+
xml_file = default_storage.open(path)
xml_file.name = file_name
k = PublishXForm(xml_file, self.user)
@@ -341,7 +342,7 @@ def push(self):
for form_dir in dirs:
dir_path = os.path.join(self.forms_path, form_dir)
form_dirs, form_files = default_storage.listdir(dir_path)
- form_xml = '%s.xml' % form_dir
+ form_xml = f'{form_dir}.xml'
if form_xml in form_files:
form_xml_path = os.path.join(dir_path, form_xml)
x = self._upload_xform(form_xml_path, form_xml)
diff --git a/onadata/libs/utils/logger_tools.py b/onadata/libs/utils/logger_tools.py
index 7bf88bf66..273893e2a 100644
--- a/onadata/libs/utils/logger_tools.py
+++ b/onadata/libs/utils/logger_tools.py
@@ -36,6 +36,7 @@
from modilabs.utils.subprocess_timeout import ProcessTimedOut
from pyxform.errors import PyXFormError
from pyxform.xform2json import create_survey_element_from_xml
+from rest_framework.exceptions import NotAuthenticated
from xml.dom import Node
from wsgiref.util import FileWrapper
@@ -97,25 +98,22 @@ def check_submission_permissions(
"""
Check that permission is required and the request user has permission.
- The user does no have permissions iff:
- * the user is authed,
- * either the profile or the form require auth,
- * the xform user is not submitting.
-
- Since we have a username, the Instance creation logic will
- handle checking for the forms existence by its id_string.
+ If the form does not require auth, anyone can submit, regardless of whether
+ they are authenticated. Otherwise, if the form does require auth, the
+ user must be the owner or have CAN_ADD_SUBMISSIONS.
:returns: None.
:raises: PermissionDenied based on the above criteria.
"""
- profile = UserProfile.objects.get_or_create(user=xform.user)[0]
+ if not xform.require_auth:
+ # Anonymous submissions are allowed!
+ return
+
+ if request and request.user.is_anonymous:
+ raise NotAuthenticated
+
if (
request
- and (
- profile.require_auth
- or xform.require_auth
- or request.path == '/submission'
- )
and xform.user != request.user
and not request.user.has_perm('report_xform', xform)
):
@@ -762,7 +760,7 @@ def _get_instance(
`update_xform_submission_count()` from doing anything, which avoids locking
any rows in `logger_xform` or `main_userprofile`.
"""
- # check if its an edit submission
+ # check if it is an edit submission
old_uuid = get_deprecated_uuid_from_xml(xml)
instances = Instance.objects.filter(uuid=old_uuid)
diff --git a/onadata/libs/utils/viewer_tools.py b/onadata/libs/utils/viewer_tools.py
index 8ba9076e6..a8c2d804c 100644
--- a/onadata/libs/utils/viewer_tools.py
+++ b/onadata/libs/utils/viewer_tools.py
@@ -146,8 +146,9 @@ def enketo_url(
if action == 'view':
url = f'{url}/view'
- req = requests.post(url, data=values,
- auth=(settings.ENKETO_API_TOKEN, ''), verify=False)
+ req = requests.post(
+ url, data=values, auth=(settings.ENKETO_API_TOKEN, ''), verify=False
+ )
if req.status_code in [200, 201]:
try:
diff --git a/onadata/settings/base.py b/onadata/settings/base.py
index 8bad1a5ad..4654324a9 100644
--- a/onadata/settings/base.py
+++ b/onadata/settings/base.py
@@ -203,11 +203,11 @@ def skip_suspicious_operations(record):
'django_digest',
'corsheaders',
'oauth2_provider',
+ 'onadata.apps.logger.LoggerAppConfig',
'rest_framework',
'rest_framework.authtoken',
'taggit',
'readonly',
- 'onadata.apps.logger.LoggerAppConfig',
'onadata.apps.viewer',
'onadata.apps.main',
'onadata.apps.restservice',
@@ -394,7 +394,10 @@ def skip_suspicious_operations(record):
CACHES = {
# Set CACHE_URL to override. Only redis is supported.
- 'default': env.cache(default='redis://redis_cache:6380/3'),
+ 'default': env.cache_url(default='redis://redis_cache:6380/3'),
+ 'enketo_redis_main': env.cache_url(
+ 'ENKETO_REDIS_MAIN_URL', default='redis://change-me.invalid/0'
+ ),
}
DIGEST_NONCE_BACKEND = 'onadata.apps.django_digest_backends.cache.RedisCacheNonceStorage'
@@ -473,12 +476,6 @@ def skip_suspicious_operations(record):
os.environ.get('KOBOCAT_PUBLIC_SUBDOMAIN', 'kc'),
os.environ.get('PUBLIC_DOMAIN_NAME', 'kobotoolbox.org'))
-# Default value for the `UserProfile.require_auth` attribute
-REQUIRE_AUTHENTICATION_TO_SEE_FORMS_AND_SUBMIT_DATA_DEFAULT = env.bool(
- 'REQUIRE_AUTHENTICATION_TO_SEE_FORMS_AND_SUBMIT_DATA_DEFAULT',
- False
-)
-
OAUTH2_PROVIDER = {
# this is the list of available scopes
'SCOPES': {