Django: QueryScreener, a record level ownership development auditor

During the development stage of a Django app I’m working on I was exploring how to best implement rowlevel user ownerships. There are several ways to overwrite methods on object managers and even the Django admin interface is properly configurable to take a ownership from “request.user”.

But since wrongfull data disclosure is absolutely unacceptable I was still afraid that I would miss something somewhere. A nice example I ran into was populating a dropdown list in a form, where all records were visible instead of only those owned by the logged in user.

That got me thinking and eventually I wrote this small but sweet piece of middleware. Further elaboration below the code.

from django.db import connection
import re

"""
QueryScreener is a middleware development tool. This tool helps to avoid
unwanted data disclosure once you go into production.

It monitors queries to the models in your model_list and warns you when queries
are executed that do not contain a ownership where clause. And thus can be a
potential data disclosure hazard.

It requires a owner attribute in your model definition, e.g:

    owner = models.ForeignKey(User, editable=False)

Edit the 'model_list' below for what models should be monitored. And add
QueryScreener to MIDDLEWARE_CLASSES in you settings.py

Note: This can/should only be used while running Django's testserver command
with e.g: ./manage.py runserver 192.168.1.81:8000
"""

class QueryScreener(object):

    model_list = ['myapp_customer', 'myapp_order', 'myapp_product']

    def process_view(self, request, view_func, view_args, view_kwargs):
        if len(connection.queries) > 0:
            query_parse(connection.queries, self.model_list, 'process_view')

    def process_response(self, request, response):
        if len(connection.queries) > 0:
            query_parse(connection.queries, self.model_list, 'process_response')
        return response

def query_parse(self, model_list, caller_process):

    for query in connection.queries:
        for modelname in model_list:
            modelstring = 'FROM `'+modelname

            if re.search(modelstring, query['sql']) and not 
                re.search(r'^SELECT.(1).AS', query['sql']):

                reg = re.compile(r'^SELECT.*WHERE.*owner.*(ORDER BY.*)?$',
                                    re.DOTALL)

                if not reg.search(query['sql']):
                    print ('<<< WARNING >>> Query execution without ownership '
                            'clause, called from "' + caller_process + '"')
                    print query['sql']

            if re.search(r'^SELECT.(1).AS.`a`.FROM.*WHERE.*$', query['sql']):
                print ('<<< Django Farted >>>')
#                print query['sql']

Update1: The ‘ORDER BY’ in the regex needs to be optional.
Update2: Django does a ‘try update’ in save_base() without owner (seperated the select statement)

The comment in the code above sums up how to get it working. What it does is print a warning and the query in question that does not respect ownership. If enabled while developing just keep track of your console output for:

<<< WARNING >>> Query execution without ownership clause, called from "process_response"

Should you  have suggestion, criticism, or words of admiration then please, do tell me 🙂

GrtzG