• Step 1 - Allow for access only from the owner
• Step 2 - search filter
• Step 3 - Create permissions
• Extras - Additional exercises

The goal of this tutorial is to implement record access permissions in simple and complicated cases.

Prerequisites:

1. previous steps with owner field

2. at least two different users

   (my-site) $ my-site users create admin@test.ch -a --password=123456 # create admin user ID 1
   (my-site) $ my-site users create manager@test.ch -a --password=123456 # create admin user ID 2
   (my-site) $ my-site users create visitor@test.ch -a --password=123456 # create visitor user ID 3

3. at least two records

   curl -k --header "Content-Type: application/json" --request POST --data '{"title":"My test record", "contributors": [{"name": "Doe, John"}], "owner": 1}' https://localhost:5000/api/records/?prettyprint=1
   
   curl -k --header "Content-Type: application/json" --request POST --data '{"title":"Second test record", "contributors": [{"name": "Copernicus, Mikolaj"}], "owner": 2}' https://localhost:5000/api/records/?prettyprint=1

Restrict the access to read, edit and delete action for the record only to its owner.

1. We implement the permission factory. The permission requires a need to be fulfilled by a user for a record. In this case we remember that:

       "owner": {
       "type": "integer"
       },

   so the permission factory requires users to provide their ID as stored in the the "owner" field of the record. Add the following my_site/records/permissions.py file:

   from invenio_access import Permission, any_user
   +from flask_principal import UserNeed
   
   
   def files_permission_factory(obj, action=None):
       """Permissions factory for buckets."""
       return Permission(any_user)
   +
   +def owner_permission_factory(record=None):
   +    """Permission factory with owner access to the record."""
   +    return Permission(UserNeed(record["owner"]))

2. We use the permission factory in the configuration file to let the application know that this endpoint has a permission requirement (RUD). Edit my_site/records/config.py:

   +from my_site.records.permissions import owner_permission_factory
   
   RECORDS_REST_ENDPOINTS = {
       'recid': dict(
           pid_type='recid',
           pid_minter='recid',
           pid_fetcher='recid',
           default_endpoint_prefix=True,
           search_class=RecordsSearch,
           indexer_class=RecordIndexer,
           search_index='records',
           search_type=None,
           record_serializers={
               'application/json': ('my_site.records.serializers'
                                   ':json_v1_response'),
           },
           search_serializers={
               'application/json': ('my_site.records.serializers'
                                   ':json_v1_search'),
           },
           record_loaders={
               'application/json': ('my_site.records.loaders'
                                   ':json_v1'),
           },
           list_route='/records/',
           item_route='/records/<pid(recid):pid_value>',
           default_media_type='application/json',
           max_result_window=10000,
           error_handlers=dict(),
           create_permission_factory_imp=allow_all,
   -       read_permission_factory_imp=check_elasticsearch,
   -       update_permission_factory_imp=allow_all,
   -       delete_permission_factory_imp=allow_all,
   +       read_permission_factory_imp=owner_permission_factory,
   +       update_permission_factory_imp=owner_permission_factory,
   +       delete_permission_factory_imp=owner_permission_factory,
           list_permission_factory_imp=allow_all
       ),
   }
   """REST API for my-site."""

3. log in as manager user

4. visit /api/records/<id> (first record)

   {
   "message": "You don't have the permission to access the requested resource. It is either read-protected or not readable by the server.",
   "status": 403
   }

5. visit /records/<id>

   Record still not protected!

6. Set permission factory also for UI endpoints in my_site/records/config.py:

   RECORDS_UI_ENDPOINTS = dict(
       recid=dict(
           pid_type='recid',
           route='/records/<pid_value>',
           template='records/record.html',
           record_class='invenio_records_files.api:Record',
   +       permission_factory_imp='my_site.records.permissions:owner_permission_factory',
       ),

7. visit /records/<id>

The details pages of records are now protected. But if we visit /search?page=1&size=20&q=, all the records are still visible in the search page. The same is true for the REST API: /api/records/. We would like to hide the results from search if they are not owned by the current user.

1. We implement a search filter that will display records in the search results only to their owner. Let' s create my_site/records/search.py:

   from elasticsearch_dsl import Q
   from flask_security import current_user
   
   
   def owner_permission_filter():
       """Search filter with permission."""
       return [Q('match', owner=current_user.get_id())]

2. We implement a search class that uses the implemented filter (also in search.py).

   from elasticsearch_dsl import Q
   from flask_security import current_user
   +from invenio_search.api import DefaultFilter, RecordsSearch
   
   
   def owner_permission_filter():
       """Search filter with permission."""
       return [Q('match', owner=current_user.get_id())]
   
   
   +class OwnerRecordsSearch(RecordsSearch):
   +    """Class providing permission search filter."""
   +
   +    class Meta:
   +        index = 'records'
   +        default_filter = DefaultFilter(owner_permission_filter)
   +        doc_types = None
   

3. We add the search class to the configuration in my_site/records/config.py:

   +from my_site.records.search import OwnerRecordsSearch
   
   RECORDS_REST_ENDPOINTS = {
       'recid': dict(
           pid_type='recid',
           pid_minter='recid',
           pid_fetcher='recid',
           default_endpoint_prefix=True,
   +       search_class=OwnerRecordsSearch,
           indexer_class=RecordIndexer,
           search_index='records',
           search_type=None,
           record_serializers={
               'application/json': ('my_site.records.serializers'
                                   ':json_v1_response'),
           },
           search_serializers={
               'application/json': ('my_site.records.serializers'
                                   ':json_v1_search'),
           },
           record_loaders={
               'application/json': ('my_site.records.loaders'
                                   ':json_v1'),
           },
           list_route='/records/',
           item_route='/records/<pid(recid):pid_value>',
           default_media_type='application/json',
           max_result_window=10000,
           error_handlers=dict(),
           create_permission_factory_imp=allow_all,
           read_permission_factory_imp=owner_permission_factory,
           update_permission_factory_imp=owner_permission_factory,
           delete_permission_factory_imp=owner_permission_factory,
           list_permission_factory_imp=allow_all
       ),
   }
   """REST API for my-site."""

4. Go to the API search page https://127.0.0.1:5000/api/records/?prettyprint=1 and check that it displays only the records owned by the current user

5. Go to the UI search page https://127.0.0.1:5000/search?page=1&size=20&q= and check that it displays only the records owned by the current user

1. Implement the permission factory in my_site/records/permissions.py

   from invenio_access import Permission, authenticated_user
   
   def authenticated_user_permission(record=None):
       """Return an object that evaluates if the current user is authenticated."""
       return Permission(authenticated_user)

2. Add the permission factory to the configuration of the records REST endpoints in my_site/records/config.py

   -from my_site.records.permissions import owner_permission_factory
   +from my_site.records.permissions import owner_permission_factory, \
   +   authenticated_user_permission
   
   RECORDS_REST_ENDPOINTS = {
       'recid': dict(
           pid_type='recid',
           pid_minter='recid',
           pid_fetcher='recid',
           default_endpoint_prefix=True,
           search_class=OwnerRecordsSearch,
           indexer_class=RecordIndexer,
           search_index='records',
           search_type=None,
           record_serializers={
               'application/json': ('my_site.records.serializers'
                                   ':json_v1_response'),
           },
           search_serializers={
               'application/json': ('my_site.records.serializers'
                                   ':json_v1_search'),
           },
           record_loaders={
               'application/json': ('my_site.records.loaders'
                                   ':json_v1'),
           },
           list_route='/records/',
           item_route='/records/<pid(recid):pid_value>',
           default_media_type='application/json',
           max_result_window=10000,
           error_handlers=dict(),
   -       create_permission_factory_imp=allow_all
   +       create_permission_factory_imp=authenticated_user_permission,
   

3. Perform a POST request by using curl to test permission to create records as an unauthenticated user (should fail)

   curl -k --header "Content-Type: application/json" --request POST --data '{"title":"Second test record", "contributors": [{"name": "Copernicus, Mikolaj"}], "owner": 2}' https://localhost:5000/api/records/?prettyprint=1

Use case: We would like to allow our site's managers to edit and delete records

 NOTE: we have existing records already, we would not like to add the group access one by one to each record.

1. Create a managers role (group)

   (my-site)$ my-site roles create managers

2. Connect manager user with created role

   (my-site)$ my-site roles add manager@test.ch managers

3. Create the permission factory for role and owner

   from invenio_access import Permission
   from flask_principal import UserNeed, RoleNeed
   
   
   def owner_manager_permission_factory(record=None):
       """Returns permission for managers group."""
       return Permission(UserNeed(record["owner"]), RoleNeed('managers'))

4. Implement search filter for role and owner

   from elasticsearch_dsl import Q
   from flask_security import current_user
   from invenio_search.api import DefaultFilter, RecordsSearch
   
   
   def owner_manager_permission_filter():
       """Search filter with permission."""
       if current_user.has_role('managers'):
           return [Q(name_or_query='match_all')]
       else:
           return [Q('match', owner=current_user.get_id())]
   
   
   class OwnerManagerRecordsSearch(RecordsSearch):
       """Class providing permission search filter."""
   
       class Meta:
           index = 'records'
           default_filter = DefaultFilter(owner_manager_permission_filter)
           doc_types = None

5. Update the configuration file with your new filter and factory

   from my_site.records.permissions import owner_permission_factory, \
       authenticated_user_permission, owner_manager_permission_factory
   from my_site.records.search import OwnerManagerRecordsSearch
   
   RECORDS_REST_ENDPOINTS = {
       'recid': dict(
           pid_type='recid',
           pid_minter='recid',
           pid_fetcher='recid',
           default_endpoint_prefix=True,
           search_class=OwnerManagerRecordsSearch,
           indexer_class=RecordIndexer,
           search_index='records',
           search_type=None,
           record_serializers={
               'application/json': ('my_site.records.serializers'
                                   ':json_v1_response'),
           },
           search_serializers={
               'application/json': ('my_site.records.serializers'
                                   ':json_v1_search'),
           },
           record_loaders={
               'application/json': ('my_site.records.loaders'
                                   ':json_v1'),
           },
           list_route='/records/',
           item_route='/records/<pid(recid):pid_value>',
           default_media_type='application/json',
           max_result_window=10000,
           error_handlers=dict(),
           create_permission_factory_imp=authenticated_user_permission,
           read_permission_factory_imp=owner_manager_permission_factory,
           update_permission_factory_imp=owner_manager_permission_factory,
           delete_permission_factory_imp=owner_manager_permission_factory,
           list_permission_factory_imp=allow_all
       ),
   }
   """REST API for my-site."""

6. Visit https://127.0.0.1:5000/search?page=1&size=20&q= and https://127.0.0.1:5000/api/records/?prettyprint=1 as manager user and check if all the records are listed.

1. Implement access management for the record having in mind the structure below

   {
       "_access": {
           "read": {
               "systemroles": ["campus_user"]
           },
           "update": {
               "users": [1],
               "roles": ["curators"]
           }
       }
   }