What is the best way to grant CORS for all subdomains dynamically?

[alvassin]

I need to provide CORS permissions for many subdomains (i do not know all), that are located on known list of domains. E.g. *.example.com, *.corp-example.com, etc.

To allow CORS for any origin, i can do the following:

import aiohttp_cors
from aiohttp.web_app import Application

app = Application()
aiohttp_cors.setup(app, defaults={
    "*": aiohttp_cors.ResourceOptions(
        allow_credentials=True,
        expose_headers="*",
        allow_headers="*",
    )
})

for handler in handlers:
    # Register handler
    route = app.router.add_route('*', handler.URL_PATH, handler)
    
    # Handler should be handled with CORS
    app['aiohttp_cors'].add(route)

It works, but is not secure. What is simplest way to check if client origin meets some requirements (e.g. that origin matches ^[a-z0-9_-]+.example.com$) and if does not - to deny request?

I supposed that it is possible to extend some basic method to add such checks, or to provide lambda to aiohttp_cors.setup defaults, that would accept origin as input parameter and return appropriate config.

[alvassin]

As i see, for preflight CORS requests it is enough to return non-200 response to deny user request.

It is possible to extend views from both aiohttp.web.View and aiohttp_cors.CorsViewMixin, and re-define _get_config(self, request, origin, request_method)method. If origin is not passing custom checks - raise KeyError exception, otherwise call super()._get_config(). KeyError would lead for 403 Forbidden response:

class BaseHandler(aiohttp.web.View, aiohttp_cors.CorsViewMixin):
    async def _get_config(self, request, origin, request_method):
        # Handle pre-flight requests
        if not origin_is_allowed(origin): 
            raise KeyError  # would cause 403 Forbidden response
        
        return super()._get_config(request, origin, request_method)

As for non-preflight requests, to deny user request we should not add CORS headers and i don't see easy implementation.

There is _CorsConfigImpl._on_response_prepare method, that is supposed to set appropriate CORS headers to allow request or do nothing (there are several return rows, that prevent CORS headers installation) to deny.

But it is internal object - it is hard-coded in CorsConfig.__init__ method.

It is possible to extend both CorsConfig and _CorsConfigImpl, add unified (both for preflight and for non-preflight requests) hook to _CorsConfigImpl._on_response_prepare.

What do you think? What is the best way to implement additional origin checks?

[asvetlov]

I think yes, overriding _get_config() is legitimate here.

[alvassin]

@asvetlov, thank you. And what is the correct way for preflight requests? Is the only way to implement checks - to extend _CorsConfigImpl._on_response_prepare() and CorsConfig.__init__() (to replace hard-coded _CorsConfigImpl)?

[asvetlov]

Hmm, hacking _CorsConfigImpl smells ugly.
Ideally _CorsConfigImpl._on_response_prepare() should be not changed but cors configuration.
Could you share your code to demonstrate why do you do it and what _on_response_prepare() lines are changed?

[alvassin]

Thank you for fast response.

My first idea was to do something like that:

class _CustomCorsConfigImpl(_CorsConfigImpl):
    @staticmethod
    def check_origin_allowed(origin: str) -> bool:
        # Do origin checks
        return True

    async def _on_response_prepare(self,
                                   request: web.Request,
                                   response: web.StreamResponse):
        # Terminate CORS if origin is not allowed
        if not self.check_origin_allowed(request.headers.get(hdrs.ORIGIN)):
            return
        return super()._on_response_prepare(request, response)

But later i found that it is much easier to extend ResourcesUrlDispatcherRouterAdapter class.
It can affect both preflight and non-preflight responses behavior:

class CustomResourcesUrlDispatcherRouterAdapter(ResourcesUrlDispatcherRouterAdapter):
    def __init__(self, *args, allowed_origin_patterns: Collection[re.Pattern], **kwargs):
        self.allowed_origin_patterns = allowed_origin_patterns
        super().__init__(*args, **kwargs)

    def check_origin_allowed(self, origin: str) -> bool:
        for allowed_origin_pattern in self.allowed_origin_patterns:
            if allowed_origin_pattern.match(origin):
                return True
        return False

    async def get_preflight_request_config(
            self,
            preflight_request: web.Request,
            origin: str,
            requested_method: str):
        # Do origin checks
        if not self.check_origin_allowed(origin):
            # raising web.HTTPForbidden seems not a very good idea here
            raise KeyError

        return super().get_preflight_request_config(preflight_request, origin,
                                                    requested_method)

    def get_non_preflight_request_config(self, request: web.Request):
        # Do origin checks and return empty config {} if checks are not passed
        if not self.check_origin_allowed(request.headers.get(hdrs.ORIGIN)):
            return {}

        return super().get_non_preflight_request_config(request)

also, it is possible to configure aiohttp_cors with custom ResourcesUrlDispatcherRouterAdapter class easily:

cors_defaults = {
    "*": aiohttp_cors.ResourceOptions(
        allow_credentials=True,
        expose_headers="*",
        allow_headers="*",
    ),
}
app['aiohttp_cors'] = CorsConfig(
    app,
    defaults=cors_defaults,
    router_adapter=CustomResourcesUrlDispatcherRouterAdapter(
        app.router, cors_defaults,
        allowed_origin_patterns=[re.compile('[a-z0-9_-]+.example.com$')],
    )
)

That seem be not so ugly and does not use internal api.

[asvetlov]

Extending adaptor looks better.
Sorry, the library is underpowered.
I know there are several bug reports that are not fixed yet.
The library initial author is retired, I have a limited spare time that I spend mostly on asyncio and aiohttp.

If somebody wants to maintain the project -- you are welcome!