0002-api-add-roles-based-access-restrictions-48752.patch

Frédéric Péters, 27 avril 2021 07:56

Voir les différences: en ligne côte à côte

Subject: [PATCH 2/2] api: add roles-based access restrictions (#48752)

 tests/api/test_carddef.py  | 44 ++++++++++++++++++++++++++++++
 tests/api/test_user.py     | 32 ++++++++++++++++++++++
 tests/api/test_workflow.py | 56 ++++++++++++++++++++++++++++++++++++++
 wcs/api.py                 |  9 ++++--
 wcs/api_access.py          | 13 +++++++++
 wcs/api_utils.py           | 12 ++++++--
 6 files changed, 162 insertions(+), 4 deletions(-)

     from quixote import get_publisher
     from wcs import fields, qommon
     from wcs.api_access import ApiAccess
     from wcs.api_utils import sign_url
     from wcs.carddef import CardDef
     from wcs.categories import CardDefCategory
-...
         assert resp.json['data']['creation_time'] <= resp.json['data']['completion_time']
     def test_cards_retricted_api(pub, local_user):
         pub.role_class.wipe()
         role = pub.role_class(name='test')
         role.store()
         CardDef.wipe()
         carddef = CardDef()
         carddef.name = 'test'
         carddef.fields = [fields.StringField(id='0', label='foobar', varname='foo')]
         carddef.workflow_roles = {'_viewer': role.id}
         carddef.store()
         carddef.data_class().wipe()
         formdata = carddef.data_class()()
         formdata.data = {'0': 'blah'}
         formdata.just_created()
         formdata.store()
         access = ApiAccess()
         access.name = 'test'
         access.access_identifier = 'test'
         access.access_key = '12345'
         access.store()
         # no role restrictions, get it
         resp = get_app(pub).get(sign_uri('/api/cards/test/list', orig='test', key='12345'))
         assert len(resp.json['data']) == 1
         # restricted to the correct role, get it
         access.roles = [role]
         access.store()
         resp = get_app(pub).get(sign_uri('/api/cards/test/list', orig='test', key='12345'))
         assert len(resp.json['data']) == 1
         # retricted to another role, do not get it
         role2 = pub.role_class(name='second')
         role2.store()
         access.roles = [role2]
         access.store()
         resp = get_app(pub).get(sign_uri('/api/cards/test/list', orig='test', key='12345'), status=403)
         assert resp.json['err_desc'] == 'unsufficient roles'
     def test_post_invalid_json(pub, local_user):
         resp = get_app(pub).post(
             '/api/cards/test/submit', params='not a json payload', content_type='application/json', status=400

     from wcs import fields
     from wcs.admin.settings import UserFieldsFormDef
     from wcs.api_access import ApiAccess
     from wcs.formdef import FormDef
     from wcs.qommon.http_request import HTTPRequest
     from wcs.qommon.ident.password_accounts import PasswordAccount
-...
         assert resp2.json['data'][0] == resp.json['data'][1]
         assert resp2.json['data'][1] == resp.json['data'][0]
         # check there is no access with roles-limited API users
         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/forms', orig='test', key='12345'), status=403)
         assert resp.json['err'] == 1
         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_user_forms_limit_offset(pub, local_user):
         if not pub.is_using_postgresql():

     from quixote import get_publisher
     from wcs import fields
     from wcs.api_access import ApiAccess
     from wcs.formdef import FormDef
     from wcs.qommon.http_request import HTTPRequest
     from wcs.qommon.ident.password_accounts import PasswordAccount
-...
         get_app(pub).post(sign_uri(formdata.get_url() + 'jump/trigger/XXX'), status=200)
         assert formdef.data_class().get(formdata.id).status == 'wf-st2'
         assert formdef.data_class().get(formdata.id).evolution[-1].who is None
         # check with trailing slash
         formdata.store()  # reset
-...
         assert formdef.data_class().get(formdata.id).status == 'wf-st3'
     def test_workflow_trigger_api_access(pub, local_user):
         pub.role_class.wipe()
         role = pub.role_class(name='xxx')
         role.store()
         role2 = pub.role_class(name='xxx2')
         role2.store()
         workflow = Workflow(name='test')
         st1 = workflow.add_status('Status1', 'st1')
         jump = JumpWorkflowStatusItem()
         jump.trigger = 'XXX'
         jump.status = 'st2'
         st1.items.append(jump)
         jump.parent = st1
         workflow.add_status('Status2', 'st2')
         workflow.store()
         FormDef.wipe()
         formdef = FormDef()
         formdef.name = 'test'
         formdef.fields = []
         formdef.workflow_id = workflow.id
         formdef.store()
         formdef.data_class().wipe()
         formdata = formdef.data_class()()
         formdata.just_created()
         formdata.store()
         jump.by = [role.id]
         workflow.store()
         access = ApiAccess()
         access.name = 'test'
         access.access_identifier = 'test'
         access.access_key = '12345'
         access.roles = [role2]
         access.store()
         get_app(pub).post(
             sign_uri(formdata.get_url() + 'jump/trigger/XXX/', orig='test', key='12345'), status=403
+        )
         assert formdef.data_class().get(formdata.id).status == 'wf-st1'  # no change
         access.roles = [role]
         access.store()
         get_app(pub).post(
             sign_uri(formdata.get_url() + 'jump/trigger/XXX/', orig='test', key='12345'), status=200
+        )
         assert formdef.data_class().get(formdata.id).status == 'wf-st2'
         assert formdef.data_class().get(formdata.id).evolution[-1].who is None
     def test_workflow_global_webservice_trigger(pub, local_user):
         workflow = Workflow(name='test')
         workflow.add_status('Status1', 'st1')

+                        )
                 if formdata_user:
                     formdata.user_id = formdata_user[0].id
             else:
             elif isinstance(user, get_publisher().user_class):
                 formdata.user_id = user.id
             formdata.store()
-...
+                        )
                 if formdata_user:
                     formdata.user_id = formdata_user[0].id
             elif user:
             elif isinstance(user, get_publisher().user_class):
                 formdata.user_id = user.id
             if json_input.get('context'):
                 formdata.submission_context = json_input['context']
                 formdata.submission_channel = formdata.submission_context.pop('channel', None)
-...
             user = self.user or get_user_from_api_query_string() or get_request().user
             if not user:
                 raise AccessForbiddenError('no user specified')
             if not isinstance(user, get_publisher().user_class):
                 raise AccessForbiddenError('restricted API access')
             user_info = user.get_substitution_variables(prefix='')
             del user_info['user']
             user_info['id'] = user.id
-...
                 return json.dumps({'err': 1, 'err_desc': 'unknown NameID', 'data': []})
             if not user:
                 return json.dumps({'err': 1, 'err_desc': 'no user specified', 'data': []})
             if not isinstance(user, get_publisher().user_class):
                 raise AccessForbiddenError('restricted API access')
             forms = self.get_user_forms(user)

                     if role_name:
                         criterias.append(Equal('name', role_name))
             return get_publisher().role_class.select([Or(criterias)], order_by='name')
         def get_as_api_user(self):
             class RestrictedApiUser:
                 # kept as inner class so cannot be pickled
                 id = Ellipsis  # make sure it fails all over the place if used
                 is_admin = False
                 def get_roles(self):
                     return self.roles
             user = RestrictedApiUser()
             user.roles = [x.id for x in self.get_roles()]
             return user

     def get_user_from_api_query_string(api_name=None):
         # check signature or auth header
         if not is_url_signed():
             if api_name:
                 check_http_basic_auth(api_name)
             else:
                 return None
         # Signature or auth header are ok.
         # Look for the user, by email/NameID.
         # check access restriction defined in API access object
         orig = get_request().form.get('orig')
         if orig:
             api_access = ApiAccess.get_by_identifier(orig)
             if api_access and api_access.get_roles():
                 return api_access.get_as_api_user()
         # get user reference from query string
         user = None
         if get_request().form.get('email'):
             email = get_request().form.get('email')
+    -

Projet

Général

Profil

Produits Entr'ouvert » w.c.s.

0002-api-add-roles-based-access-restrictions-48752.patch