aiohttp security authorization for StaticResource (or any aiohttp resource)

demo.py

from aiohttp import web
from functools import wraps
from logging import getLogger
from types import MethodType


_LOGGER = getLogger(__name__)


AUTH_USER = 'authenticated_user'


def permits(user, required_permission):
    if user == 123:
        return True
    else:
        return False


def require_permission(required_permission):
    def authorize_handler_decorator(handler):
        @wraps(handler)
        async def new_handler(request):
            if AUTH_USER not in request:
                _LOGGER.error("Please install the authorization middleware.")
                raise web.HTTPInternalServerError
            if permits(request[AUTH_USER], required_permission):
                return await handler(request)
            else:
                raise web.HTTPForbidden

        return new_handler
    return authorize_handler_decorator


def authorize_resource(resource, required_permission):
    """Modify the resource to ensure authorization against the requested permissions."""
    orig_resolve = resource.resolve
    async def new_resolve(self, request):
        resolved = await orig_resolve(request)
        url_mapping, allowed = resolved
        if url_mapping and authorization:
            # Wrap the mapping's _handler
            url_mapping.route._handler = require_permission(required_permission)(
                    url_mapping.route._handler)
        return url_mapping, allowed
    resource.resolve = MethodType(new_resolve, resource)
    return resource


@require_permission('view')
async def test(request):
    print('Handler function called')
    return web.Response(text="Hello")


@web.middleware
async def auth_middleware(request, handler):
    request[AUTH_USER] = 123
    response = await handler(request)
    return response


app = web.Application(middlewares=[auth_middleware])
app.router.add_get('/', test)

resource = web.StaticResource('/static', './', show_index=True)
resource = authorize_resource(resource, required_permission='view_static')
app.router.register_resource(resource)

web.run_app(app)

The fact that one would almost always use aiohttp-session to do the authentication is not really that important here. I wanted to see how the authorization looked - especially for web.StaticResource instances.