Search code examples
pythonpyramid

Pyramid: resource tree in URL Dispatch(hybrid) application


I am trying to implement dynamic ACLs(including row-level security) in URL Dispatch app. Defining Root factory only doesn't seem to be sufficient as I need to perform individual authz checks for each secured endpoint. My setup looks as follows(I was using pyramid docs and mmerickel's tutorials as a guide):

config.py

...
settings = config.registry.settings
config = Configurator(settings=settings, root_factory=RootPermissionFactory)
config.set_authentication_policy(CustomAuthenticationPolicy(settings))
config.set_authorization_policy(ACLAuthorizationPolicy())
...

views.py

#import ...

@view_defaults(renderer='json', permission='secured')
class RecordsView(object):
    ...

   @view_config(request_method='GET', route_name='records_by_id')
   def get(self):
        record = self.request.context.data
        if not record:
            return HTTPNotFound()
        return record

    @view_config(request_method='GET', route_name='records')
    def get_by_owners(self):
        owner_uids = self.request.params.mixed()['owner_uids']
        return records_service.get_records(owner_uids=owner_uids)

def includeme(config):
    config.add_route('records', '/records', factory=RecordsResource)
    config.add_route('records_by_id', 'records/{record_id}', factory=RecordFactory, traverse='{record_id}')

authorization.py

class RootPermissionFactory(object):
    __name__ = None
    __parent__ = None

    def __acl__(self):
        return [
            (Allow, 'authenticated_principal', 'secured'),
        ]

    def __init__(self, request):
        self.request = request


class RecordFactory(object):
    def __init__(self, request):
        self.request = request

    def __getitem__(self, key):
        record_data = records_service.get_record(key)
        owner = record_data.get('owner_uid')
        return RecordContext(self.request, owner, record_data)


class RecordContext(object):
    def __acl__(self):    
        owner_principal = 'u:{owner}'.format(owner=self.owner)
        return [
            (Allow, owner_principal, 'secured'),
        ]

    def __init__(self, request, owner, record_data):
        self.request = request
        self.owner = owner
        self.data = record_data


class RecordsResource(object):
    def __acl__(self):
        request_params = self.request.params.mixed()
        request_owner_uids = request_params['owner_uids']
        authorized_owner_uids = owners_api_service.get_authorized_owners(self.request.user['auth_data'])
        return [(Allow, 'authenticated_principal', 'secured')]\
            if set(authorized_owner_uids) == set(request_owner_uids) else []

    def __init__(self, request):
        self.request = request

My questions are following:

  • is it acceptable to utilize row-level security checks without having Model layer? i.e. there is no ORM set up for Records data and there is also no plain Model to put persisted data into, so I have to use 'fake' RecordContext class to attach __acl__ rules and pass data to the view
  • is it acceptable to treat /records endpoint as a Resource despite the fact it is not a resource from Traversal perspective and relies on query parameters rather than path sections?

Solution

  • I think the answer to both of your question is: if it works for you then it's totally acceptable. I've found a lot of success treating URLs as resources as a general pattern to the extent that I have a some tooling to avoid using route_name. For example:

    config.add_route('records', '/records', factory=RecordsResource, use_global_views=True)
    config.add_route('records_by_id', 'records/{record_id}', factory=RecordFactory, traverse='{record_id}', use_global_views=True)
    
    @view_config(context=RecordsResource, renderer='json')
    def records_view(request):
        return {}
    
    @view_config(context=RecordContext, renderer='json')
    def record_view(request):
        return {}