From ec15e913e454a5119f937c8dd60783bc7d303855 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Laur=C3=A9line=20Gu=C3=A9rin?= <zebuline@entrouvert.com>
Date: Tue, 14 Sep 2021 10:55:19 +0200
Subject: [PATCH] api: add basic auth support to users/xxx and users/xxx/forms
 (#56908)

---
 tests/api/test_user.py | 183 ++++++++++++++++++++++++++++-------------
 wcs/api.py             |  16 ++--
 2 files changed, 138 insertions(+), 61 deletions(-)

diff --git a/tests/api/test_user.py b/tests/api/test_user.py
index fe43ce73..894cbc42 100644
--- a/tests/api/test_user.py
+++ b/tests/api/test_user.py
@@ -1,5 +1,6 @@
 import datetime
 import os
+from functools import partial
 
 import pytest
 from quixote import get_publisher
@@ -77,6 +78,27 @@ def admin_user():
     return user
 
 
+def _get_url(url, app, auth, access, user, **kwargs):
+    if auth == 'http-basic':
+        app.set_authorization(('Basic', ('test', '12345')))
+        return app.get(url, **kwargs)
+    else:
+        return app.get(
+            sign_uri(url, user=user, orig=access.access_identifier, key=access.access_key), **kwargs
+        )
+
+
+@pytest.fixture
+def access(pub):
+    ApiAccess.wipe()
+    access = ApiAccess()
+    access.name = 'test'
+    access.access_identifier = 'test'
+    access.access_key = '12345'
+    access.store()
+    return access
+
+
 def test_roles(pub, local_user):
     pub.role_class.wipe()
     role = pub.role_class(name='Hello World')
@@ -102,6 +124,30 @@ def test_roles(pub, local_user):
     assert resp.json['data'][0]['details'] == 'kouign amann'
 
 
+def test_user_api_with_restricted_access(pub, access):
+    role = pub.role_class(name='test')
+    role.store()
+
+    access.roles = [role]
+    access.store()
+
+    resp = get_app(pub).get(sign_uri('/api/user/', orig='test', key='12345'), status=403)
+    assert resp.json['err'] == 1
+    assert resp.json['err_desc'] == 'restricted API access'
+
+
+def test_users_api_with_restricted_access(pub, local_user, access):
+    role = pub.role_class(name='test')
+    role.store()
+
+    access.roles = [role]
+    access.store()
+
+    resp = get_app(pub).get(sign_uri('/api/users/', orig='test', key='12345'), status=403)
+    assert resp.json['err'] == 1
+    assert resp.json['err_desc'] == 'restricted API access'
+
+
 def test_users(pub, local_user):
     resp = get_app(pub).get('/api/users/', status=403)
 
@@ -199,27 +245,48 @@ def test_users_description(pub, local_user):
     )
 
 
-def test_user_by_nameid(pub, local_user):
-    resp = get_app(pub).get(sign_uri('/api/users/xyz/', user=local_user), status=404)
+@pytest.mark.parametrize('auth', ['signature', 'http-basic'])
+def test_user_by_nameid(pub, local_user, access, auth):
+    app = get_app(pub)
+    get_url = partial(_get_url, app=app, auth=auth, access=access, user=local_user)
+
+    resp = get_url('/api/users/xyz/', status=404)
     local_user.name_identifiers = ['xyz']
     local_user.store()
-    resp = get_app(pub).get(sign_uri('/api/users/xyz/', user=local_user))
+    resp = get_url('/api/users/xyz/')
     assert str(resp.json['id']) == str(local_user.id)
 
 
-def test_user_roles(pub, local_user):
-    local_user.name_identifiers = ['xyz']
-    local_user.store()
+def test_user_by_nameid_api_access_restrict_to_anonymised_data(pub, local_user, access):
+    app = get_app(pub)
+    get_url = partial(_get_url, app=app, auth='http-basic', access=access, user=local_user)
+
+    get_url('/api/users/%s/' % local_user.id)
+
+    access.restrict_to_anonymised_data = True
+    access.store()
+    resp = get_url('/api/users/%s/' % local_user.id, status=403)
+    assert resp.json['err'] == 1
+    assert resp.json['err_desc'] == 'restricted API access'
+
+
+@pytest.mark.parametrize('auth', ['signature', 'http-basic'])
+def test_user_roles(pub, local_user, access, auth):
     role = pub.role_class(name='Foo bar')
     role.store()
+    local_user.name_identifiers = ['xyz']
     local_user.roles = [role.id]
     local_user.store()
-    resp = get_app(pub).get(sign_uri('/api/users/xyz/', user=local_user))
+
+    app = get_app(pub)
+    get_url = partial(_get_url, app=app, auth=auth, access=access, user=local_user)
+
+    resp = get_url('/api/users/xyz/')
     assert len(resp.json['user_roles']) == 1
     assert resp.json['user_roles'][0]['name'] == 'Foo bar'
 
 
-def test_user_forms(pub, local_user):
+def test_user_forms(pub, local_user, access):
     Workflow.wipe()
     workflow = Workflow.get_default_workflow()
     workflow.id = '2'
@@ -339,10 +406,6 @@ def test_user_forms(pub, local_user):
     role = pub.role_class(name='test')
     role.store()
 
-    access = ApiAccess()
-    access.name = 'test'
-    access.access_identifier = 'test'
-    access.access_key = '12345'
     access.roles = [role]
     access.store()
 
@@ -351,42 +414,6 @@ def test_user_forms(pub, local_user):
     assert resp.json['err_desc'] == 'restricted API access'
 
 
-def test_user_api_with_restricted_access(pub):
-    role = pub.role_class(name='test')
-    role.store()
-
-    access = ApiAccess()
-    access.name = 'test'
-    access.access_identifier = 'test'
-    access.access_key = '12345'
-    access.roles = [role]
-    access.store()
-
-    resp = get_app(pub).get(sign_uri('/api/user/', orig='test', key='12345'), status=403)
-    assert resp.json['err'] == 1
-    assert resp.json['err_desc'] == 'restricted API access'
-
-
-def test_users_api_with_restricted_access(pub, local_user):
-    role = pub.role_class(name='test')
-    role.store()
-
-    access = ApiAccess()
-    access.name = 'test'
-    access.access_identifier = 'test'
-    access.access_key = '12345'
-    access.roles = [role]
-    access.store()
-
-    resp = get_app(pub).get(sign_uri('/api/users/', orig='test', key='12345'), status=403)
-    assert resp.json['err'] == 1
-    assert resp.json['err_desc'] == 'restricted API access'
-
-    resp = get_app(pub).get(sign_uri('/api/users/%s/' % local_user.id, orig='test', key='12345'), status=403)
-    assert resp.json['err'] == 1
-    assert resp.json['err_desc'] == 'restricted API access'
-
-
 def test_user_forms_limit_offset(pub, local_user):
     if not pub.is_using_postgresql():
         pytest.skip('this requires SQL')
@@ -496,7 +523,8 @@ def test_user_forms_categories_filter(pub, local_user):
     assert len(resp.json['data']) == 5
 
 
-def test_user_forms_from_agent(pub, local_user):
+@pytest.mark.parametrize('auth', ['signature', 'http-basic'])
+def test_user_forms_from_agent(pub, local_user, access, auth):
     pub.role_class.wipe()
     role = pub.role_class(name='Foo bar')
     role.store()
@@ -525,7 +553,14 @@ def test_user_forms_from_agent(pub, local_user):
     formdata.jump_status('new')
     formdata.store()
 
-    resp = get_app(pub).get(sign_uri('/api/users/%s/forms' % local_user.id, user=agent_user))
+    if auth == 'http-basic':
+        access.roles = [role]
+        access.store()
+
+    app = get_app(pub)
+    get_url = partial(_get_url, app=app, auth=auth, access=access, user=agent_user)
+
+    resp = get_url('/api/users/%s/forms' % local_user.id)
     assert resp.json['err'] == 0
     assert len(resp.json['data']) == 1
     assert resp.json['data'][0]['form_name'] == 'test'
@@ -536,21 +571,46 @@ def test_user_forms_from_agent(pub, local_user):
     formdef.skip_from_360_view = True
     formdef.store()
 
-    resp = get_app(pub).get(sign_uri('/api/users/%s/forms' % local_user.id, user=agent_user))
+    resp = get_url('/api/users/%s/forms' % local_user.id)
     assert len(resp.json['data']) == 0
 
     formdef.workflow_roles = {'_receiver': str(role.id)}
     formdef.store()
     formdef.data_class().rebuild_security()
-    resp = get_app(pub).get(sign_uri('/api/users/%s/forms' % local_user.id, user=agent_user))
+    resp = get_url('/api/users/%s/forms' % local_user.id)
     assert len(resp.json['data']) == 1
 
     agent_user.roles = []
     agent_user.store()
-    get_app(pub).get(sign_uri('/api/users/%s/forms' % local_user.id, user=agent_user), status=403)
+    if auth == 'http-basic':
+        access.roles = []
+        access.store()
+        resp = get_url('/api/users/%s/forms' % local_user.id)
+        assert len(resp.json['data']) == 0
+    else:
+        get_url('/api/users/%s/forms' % local_user.id, status=403)
+
+
+def test_user_forms_api_access_restrict_to_anonymised_data(pub, local_user, access):
+    FormDef.wipe()
+    formdef = FormDef()
+    formdef.name = 'test'
+    formdef.fields = []
+    formdef.store()
+
+    app = get_app(pub)
+    get_url = partial(_get_url, app=app, auth='http-basic', access=access, user=local_user)
+
+    get_url('/api/users/%s/forms' % local_user.id)
+
+    access.restrict_to_anonymised_data = True
+    access.store()
+    resp = get_url('/api/users/%s/forms' % local_user.id, status=403)
+    assert resp.json['err'] == 1
+    assert resp.json['err_desc'] == 'restricted API access'
 
 
-def test_user_forms_include_accessible(pub, local_user):
+def test_user_forms_include_accessible(pub, local_user, access):
     if not pub.is_using_postgresql():
         pytest.skip('this requires SQL')
         return
@@ -611,8 +671,10 @@ def test_user_forms_include_accessible(pub, local_user):
     formdata4.jump_status('new')
     formdata4.store()
 
+    app = get_app(pub)
+
     def get_ids(url):
-        resp = get_app(pub).get(url)
+        resp = app.get(url)
         return {int(x['form_number_raw']) for x in resp.json['data']}
 
     resp = get_ids(sign_uri('/api/user/forms', user=local_user))
@@ -628,6 +690,17 @@ def test_user_forms_include_accessible(pub, local_user):
     resp = get_ids(sign_uri('/api/users/%s/forms?include-accessible=on' % local_user.id, user=agent_user))
     assert resp == {formdata1.id, formdata3.id}
 
+    # an api access gets the same results
+    access.roles = [role]
+    access.store()
+    app.set_authorization(('Basic', ('test', '12345')))
+
+    resp = get_ids('/api/users/%s/forms' % local_user.id)
+    assert resp == {formdata1.id}
+
+    resp = get_ids('/api/users/%s/forms?include-accessible=on' % local_user.id)
+    assert resp == {formdata1.id, formdata3.id}
+
 
 def test_user_drafts(pub, local_user):
     FormDef.wipe()
diff --git a/wcs/api.py b/wcs/api.py
index 70f2b0bc..4ae086f4 100644
--- a/wcs/api.py
+++ b/wcs/api.py
@@ -968,15 +968,18 @@ class ApiUserDirectory(Directory):
         if user.is_api_user:
             raise AccessForbiddenError('restricted API access')
 
+        query_user = get_user_from_api_query_string() or get_request().user
+        if query_user and query_user.is_api_user and query_user.api_access.restrict_to_anonymised_data:
+            raise AccessForbiddenError('restricted API access')
+
         forms = self.get_user_forms(user)
 
         if self.user:
             # call to /api/users/<id>/forms, this returns the forms of the
             # given user filtered according to the permissions of the caller
             # (from query string or session).
-            query_user = get_user_from_api_query_string() or get_request().user
             if query_user and query_user.id != self.user.id:
-                if not query_user.can_go_in_backoffice():
+                if not query_user.is_api_user and not query_user.can_go_in_backoffice():
                     raise AccessForbiddenError('user not allowed to query data from others')
                 # mark forms that are readable by querying user
                 user_roles = set(query_user.get_roles())
@@ -1081,12 +1084,13 @@ class ApiUsersDirectory(Directory):
         return json.dumps({'data': data, 'err': 0}, cls=misc.JSONEncoder)
 
     def _q_lookup(self, component):
-        if not (is_url_signed() or (get_request().user and get_request().user.can_go_in_admin())):
-            raise AccessForbiddenError('unsigned request or user has no access to backoffice')
-
         api_user = get_user_from_api_query_string()
         if api_user and api_user.is_api_user:
-            raise AccessForbiddenError('restricted API access')
+            # API users are ok except if they are restricted to anonymised data
+            if api_user.api_access.restrict_to_anonymised_data:
+                raise AccessForbiddenError('restricted API access')
+        elif not (is_url_signed() or (get_request().user and get_request().user.can_go_in_admin())):
+            raise AccessForbiddenError('unsigned request or user has no access to backoffice')
 
         user_class = get_publisher().user_class
         try:
-- 
2.33.0

