Periodically check auth status

Closes #4

Author: Marco Paolini

pushpull/config.py

@@ -53,3 +53,7 @@ def get_cors_allow_credentials():
 
 def get_ws_autoping_timeout():
     return 15
+
+
+def get_periodic_auth_check_timeout():
+    return 150

pushpull/websocket/gateway.py

@@ -19,8 +19,7 @@
 async def websocket_rabbitmq_gateway(request):
     authorization = decode_auth_querystring_param(request.GET)
     try:
-        # TODO: reuse amqp channel: here we open one, we close it and then we reopen another one with the
-        # Exchanger a few lines down
+        # TODO: reuse amqp channel and maybe share it with the Exchanger
         user_info = await auth.get_user_info(authorization)
     except auth.NotAuthorized as exc:
         raise aiohttp.web_exceptions.HTTPUnauthorized() from exc
@@ -29,8 +28,7 @@ async def websocket_rabbitmq_gateway(request):
     except auth.AuthTimeout as exc:
         logger.warning('auth backend timeout')
         raise aiohttp.web_exceptions.HTTPInternalServerError(text='auth backend timeout') from exc
-    else:
-        name = user_info.id
+    name = user_info.id
     ws = aiohttp.web.WebSocketResponse()
     client_id = request.GET.get('client-id')
     if not client_id:
@@ -45,7 +43,8 @@ async def websocket_rabbitmq_gateway(request):
                 asyncio.ensure_future(coro) for coro in [
                     send_from_amqp_to_websocket(amqp_receiver, ws),
                     send_from_websocket_to_amqp(ws, amqp_sender),
-                    send_ping_to_websocket(ws, config.get_ws_autoping_timeout())
+                    send_ping_to_websocket(ws, config.get_ws_autoping_timeout()),
+                    check_auth_periodic(authorization, delay=config.get_periodic_auth_check_timeout())
                 ]
             ]
             try:
@@ -61,9 +60,9 @@ async def websocket_rabbitmq_gateway(request):
                     except Exception as exc:
                         logger.exception('client id %s coroutine %r done due to exception: %r', client_id, coro, exc)
                     else:
-                        logger.info('client id %s coroutine %r done, result: %r', client_id, coro, result)
+                        logger.debug('client id %s coroutine %r done, result: %r', client_id, coro, result)
                 for coro in pending:
-                    logger.info('client id %s cancelling pending coroutine %r', client_id, coro)
+                    logger.debug('client id %s cancelling pending coroutine %r', client_id, coro)
                     coro.cancel()
                     await asyncio.sleep(0)
     except Exception:
@@ -100,6 +99,18 @@ async def send_ping_to_websocket(ws, timeout):
         await asyncio.sleep(timeout)
 
 
+async def check_auth_periodic(authorization, delay):
+    while True:
+        await asyncio.sleep(delay)
+        logger.debug('periodic auth check start for %s', authorization)
+        try:
+            # TODO: reuse amqp channel and maybe share it with the Exchanger
+            await auth.get_user_info(authorization)
+        except auth.AuthTimeout as exc:
+            logger.warning('periodic auth check timeout for %s', authorization)
+        logger.debug('periodic auth check OK for %s', authorization)
+
+
 async def echo_websocket(ws):
     """For testing only """
     async for msg in ws: