Diable API if auth is enabled (#1225)

* Disable api calls if auth is configured
* Remove leftover code
* Depricate WebSocket API

README.rst

@@ -104,8 +104,7 @@ Celery command and before Flower sub-command): ::
 API
 ---
 
-Flower API enables to manage the cluster via REST API, call tasks and
-receive task events in real-time via WebSockets.
+Flower API enables to manage the cluster via REST API.
 
 For example you can restart worker's pool by: ::
 
@@ -119,13 +118,6 @@ Or terminate executing task by: ::
 
     $ curl -X POST -d 'terminate=True' http://localhost:5555/api/task/revoke/8a4da87b-e12b-4547-b89a-e92e4d1f8efd
 
-Or receive task completion events in real-time: ::
-
-    var ws = new WebSocket("ws://localhost:5555/api/task/events/task-succeeded/");
-    ws.onmessage = function (event) {
-        console.log(event.data);
-    }
-
 For more info checkout `API Reference`_ and `examples`_.
 
 .. _API Reference: https://flower.readthedocs.io/en/latest/api.html

flower/api/__init__.py

@@ -1,25 +1,7 @@
-import tornado.websocket
+import tornado.web
+from ..views import BaseHandler
 
-
-class BaseWebSocketHandler(tornado.websocket.WebSocketHandler):
-    # listeners = [], should be created in derived class
-
-    def open(self):
-        listeners = self.listeners
-        listeners.append(self)
-
-    def on_message(self, message):
-        pass
-
-    def on_close(self):
-        listeners = self.listeners
-        if self in listeners:
-            listeners.remove(self)
-
-    @classmethod
-    def send_message(cls, message):
-        for l in cls.listeners:
-            l.write_message(message)
-
-    def check_origin(self, origin):
-        return True
+class BaseApiHandler(BaseHandler):
+    def prepare(self):
+        if self.application.options.basic_auth or self.application.options.auth:
+            raise tornado.web.HTTPError(405, "api is not available when auth is enabled")

flower/api/control.py

@@ -7,13 +7,13 @@
 from tornado import gen
 from tornado import util
 
-from ..views import BaseHandler
+from . import BaseApiHandler
 
 
 logger = logging.getLogger(__name__)
 
 
-class ControlHandler(BaseHandler):
+class ControlHandler(BaseApiHandler):
     def is_worker(self, workername):
         return workername and workername in self.application.workers
 

flower/api/tasks.py

@@ -15,7 +15,7 @@
 from celery.backends.base import DisabledBackend
 
 from ..utils import tasks
-from ..views import BaseHandler
+from . import BaseApiHandler
 from ..utils.broker import Broker
 from ..api.control import ControlHandler
 from collections import OrderedDict
@@ -24,7 +24,7 @@
 logger = logging.getLogger(__name__)
 
 
-class BaseTaskHandler(BaseHandler):
+class BaseTaskHandler(BaseApiHandler):
     DATE_FORMAT = '%Y-%m-%d %H:%M:%S.%f'
 
     def get_task_args(self):

flower/events.py

@@ -117,11 +117,6 @@ def event(self, event):
         if event_type == 'worker-offline':
             self.metrics.worker_online.labels(worker_name).set(0)
 
-        # Send event to api subscribers (via websockets)
-        classname = api.events.getClassName(event_type)
-        cls = getattr(api.events, classname, None)
-        if cls:
-            cls.send_message(event)
 
 
 class Events(threading.Thread):

flower/urls.py

@@ -2,7 +2,6 @@
 
 from tornado.web import StaticFileHandler, url
 
-from .api import events
 from .api import control
 from .api import tasks
 from .api import workers
@@ -57,15 +56,6 @@
     (r"/api/task/timeout/(.+)", control.TaskTimout),
     (r"/api/task/rate-limit/(.+)", control.TaskRateLimit),
     (r"/api/task/revoke/(.+)", control.TaskRevoke),
-    # Events WebSocket API
-    (r"/api/task/events/task-sent/(.*)", events.TaskSent),
-    (r"/api/task/events/task-received/(.*)", events.TaskReceived),
-    (r"/api/task/events/task-started/(.*)", events.TaskStarted),
-    (r"/api/task/events/task-succeeded/(.*)", events.TaskSucceeded),
-    (r"/api/task/events/task-failed/(.*)", events.TaskFailed),
-    (r"/api/task/events/task-revoked/(.*)", events.TaskRevoked),
-    (r"/api/task/events/task-retried/(.*)", events.TaskRetried),
-    (r"/api/task/events/task-custom/(.*)", events.TaskCustom),
     # Metrics
     (r"/metrics", monitor.Metrics),
     (r"/healthcheck", monitor.Healthcheck),

tests/unit/api/test_control.py

@@ -1,8 +1,9 @@
 from mock import MagicMock
-
+import mock
 from flower.api.control import ControlHandler
 
 from tests.unit import AsyncHTTPTestCase
+from tornado.options import options
 
 
 class UnknownWorkerControlTests(AsyncHTTPTestCase):
@@ -169,3 +170,12 @@ def test_terminate_signal(self):
         celery.control.revoke.assert_called_once_with('test',
                                                       terminate=True,
                                                       signal='SIGUSR1')
+
+
+class ControlAuthTests(WorkerControlTests):
+    def test_auth(self):
+        with mock.patch.object(options.mockable(), 'basic_auth', ['user1:password1']):
+            app = self._app.capp
+            app.control.broadcast = MagicMock()
+            r = self.post('/api/worker/shutdown/test', body={})
+            self.assertEqual(405, r.code)