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:
RecordContext
class to attach __acl__
rules and pass data to the view/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?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 {}