Google App Engine - Securing url of cron python

Question

I'm a newbie to google app engine. I want the security restriction for url of cron so that it shouldn't be accessible by url directly. For this I've already read the docs and some of Q&As ([Google app engine: security of cron jobs).

I implemented the login : admin solution suggested in this link. But I failed to implement security as self.request.headers.get('X-AppEngine-Cron') is always None, whether it is cron or accessed via url directly.

So I don't know from where is the request coming (from cron or direct access)

def cron_method(BaseRestHandler):
  def check_if_cron(self, *args, **kwargs):
    if self.request.headers.get('X-AppEngine-Cron') is None:
        logging.info("error-- not cron")
        self.UNAUTH = "cron"
        self.raise_exception()
    else:
        return BaseRestHandler(self, *args, **kwargs)

return check_if_cron

I used customized handler BaseRestHandler for other authentications.

@cron_method
def put(self):
    logging.info("inside put--")

This is called via taskqueue from the get method of the class. The problem is I didn't get header X-AppEngine-Cron Any other logic or method will be appreciated.

Thanks In Advance.

Answer 1

It seems you attempted to make the check a decorator.

But your code shows the decorator applied to a put() method, not a get() method - the cron executes only on a get().

Also your decorator doesn't look quite right to me. Shouldn't a decorator take as argument a function and return some locally defined function which executes (not returns) the function received as argument?

I'd suggest you go back to basics - try to make the header check in the get method of the handler itself and only after you get that working consider further, more complex changes like the pulling the check in a decorator.

It is more likely that your decorator is not working than GAE's documented infra to not be working. Keeping things simple (at first) would at least help your investigation effort be pointed in a better direction.

Try this:

def cron_method(handler_method):

    def check_if_cron(self, *args, **kwargs):

        if self.request.headers.get('X-AppEngine-Cron') is None:
            logging.info("error-- not cron")
            self.UNAUTH = "cron"
            self.raise_exception()
        else:
            handler_method(self, *args, **kwargs)

    return check_if_cron

As for the invocations from the task queue - those requests are no longer cron requests, even if the tasks are created and enqueued by a cron request.

From Securing task handler URLs:

If a task performs sensitive operations (such as modifying data), you might want to secure its worker URL to prevent a malicious external user from calling it directly. You can prevent users from accessing task URLs by restricting access to App Engine administrators. Task requests themselves are issued by App Engine and can always target restricted URL.

You can restrict a URL by adding the login: admin element to the handler configuration in your app.yaml file.

If you want to also prevent manual access to those URLs (i.e. restrict it only to task queue requests) you can perform header checks similar to the cron one. The header values are listed in Reading request headers. Personally I picked X-AppEngine-TaskName.