Repository: cylc/cylc-uiserver (language: python)

Issue:
Define how the authorization in the UI server will work with other components
JupyterHub has authentication, and also a thin layer of authorization in its reverse proxy. But the Cylc UI server will need to redefine parts of its authentication.

This issue is to record discussions on authorization for the UI server, which will interface with JupyterHub and probably be used by Cylc Web as a bridge to communicate with cylc and JupyterHub.

See related discussion in cylc/cylc: https://github.com/cylc/cylc/issues/1901

See related discussion in cylc/web: https://github.com/cylc/cylc-web/issues/60

Cross-user operation - umbrella issue
(And related topics, such as how UI Servers behave when not looked at, etc.). See also:
- #10 (authorization)
- #27 ("list other UI Servers running")

This is to record an @oliver-sanders "brain dump" on the topic, in response to questions from @jarich, from Element chat, as that seems to be the current best record of earlier discussions (primarily the Feb 2020 workshop).  To be refined and replaced by more focused Issues as we go...

> Brain dump time (apologies in advance), contains some extra details and contents of discussions not currently documented...
>
>>    If 2 or more users want to view Bob's workflows, will there be only one UI server running for Bob?
>>
>>    If I attempt to spawn a UI server from the command line for Bob, and there is already a UI server running for Bob, is that a no-op?
>
>    - There will only ever be one UI server (UIS) per "owner" running under one hub.
>    - When you visit the URL for that server Jupyterhub will spawn the UIS if it is not running then redirect, otherwise I think it is a simple redirect at the reverse-poxy (very efficient).
>    - The UIS knows about open connections so is able to tell when it is not in use. We anticipate building in a configurable shutdown timeout.
>    - An idle UIS should have a low footprint and starting up a UIS should be cheap.
>    - We currently anticipate that startup/shutdown for the UIS will be efficient and that aggressively shutting-down unused UIServers is likely a good idea.
>    - This will help to keep the UIServer flock on the latest version.
>    - UIServers can be manually shutdown through the web interface.
>
> Additionally:
>
>   - The websocket connections between the UIS and UI have a "heartbeat" which gives us a robust way of telling when connections are inactive from the UIS end.
>   - We have a plan for the UI to detect inactivity and suspend connections, freezing the UI but maintaining its state until the user interacts with it again.
>   - When this happens the UI would probably go grey with a message in the middle saying "paused, click to resume" or something like that.
>   - The time it takes to sync a workflow through to the UI is pretty good, even for larger flows so it should be fast to resume.
>
>> If the UI Server is not being run as a user with a password, then it will need to persist even with no connections. Or at least some part of it will have to. This could potentially be done with a flag on start up?
>
> We will want to make the "shutdown on inactivity" feature configurable anyway so should be good there, however, if the UIS went down for some other reason it would have to be brought back up manually.
>
>>    what is the pathway the user makes?
>
>   - The URL for the UI contains the username, these URLs can be shared and should work for other users.
    Visiting a URL for someone else's UIS will cause you to (spawn if necessary and) connect to their UIS (providing you are authorised).
>   - The UI loads to a default dashboard page.
>   - In the top left-hand corner of the UI are two boxes for manually entering a username and hub address to facilitate changing between users and potentially hubs.
 >  - We can do user-name discovery from the user's authorisation configuration file (i.e. list user names that the user has themselves configured, low security impact).
>
>>    If UI servers run on machines that are not running the/a hub, how does the hub find them? Can we load balance UI servers.
>
>   - This comes down to the Jupyterhub "spawner". The spawner is the thing that starts the UIServers in the Cylc Hub (notebooks in regular JupyterHub).
>   - We would like to build a special spawner which uses the scheduler-distribution logic from cylc-flow allowing load balancing of UIServers on startup.
>   - This load-balancing system can rank hosts by [psutil metrics](https://psutil.readthedocs.io/en/latest/) e.g. cpu, memory, server load, etc as well as setting hard limits.
>   - This plugin would just do the load balancing then defer to a regular distributed-spawner plugin to start the workflow (i.e. it wraps a spawner of your choice).
>
>>    I'm going to use unix permissions here, to simplify things. So read (can view the workflow), write (can do edit runs, change anything we can change through the interface about how the workflow runs) and execute (can stop, start, restart, pause, release etc the workflow).
>
>   - That's what we were thinking.
>   - There isn't much written up though there is something on [cylc-admin](https://cylc.github.io/cylc-admin/feb2020-workshop-notes)
>   - I think the plan was to provide a two-tier system consisting of a site and user configuration.
>   - This would enable the site to control the level of permissions that users are able to give away.
>   - Permissions would be grouped into read/write/execute, the main use case would be for a user to give a certain level of permissions to another user.
>   - However we would like these permissions to be fine grained for more particular use cases.
>   - E.G. Could give a user read access but also the ability to start the workflow (execute level) without giving them write (or other execute) permissions in the process.
>
>>    There is a mechanism to start a workflow from the command line that will spawn a UI server (or at least a workflow daemon?) for the user who owns it
>
> - There shouldn't be any need to start a UIS when a Scheduler starts, the UIS can be auto-started whenever its URL is visited.
>
>>   list of UI servers that all users can find after they log in
>
> There isn't a list per-se, the user would either need to:
> -  Know the user name of the person whose UIS they want to connect to and type it into the username box (top left of the UI).
> - OR configure this user name in their authorisation configuration (which should cause the UI to list it).
> - OR visit the URL of the users UIServer (derived from the hub address and user name).
>
>>   Bonus points if I can also interact with the UI server of another user from the command line for at least read access (potentially subject to constraints)
>
> For that I point you to this [cylc-admin proposal](https://cylc.github.io/cylc-admin/proposal-universal-id.html) and this [cylc-flow issue](https://github.com/cylc/cylc-flow/issues/3592).
> 
> Long-story short we plan to achieve multi-workflow and cross-user functionality on the command line by going through the UIS, examples:
```bash
# stop a single flow (can be done without the UIS in cases where the user can interact with the flow directly)
$ cylc stop myflow//

# stop multiple flows in one command
$ cylc stop myflow1// myflow2// myflow3//

# stop all my flows
$ cylc stop '*'

# stop another users flow (subject to authorisation)
$ cylc stop ~otheruser/theirflow//
```
> Not a top-priority right now, however, the new syntax has been sorted and a [tokeniser that is able to back-support the old syntax in tangent with the new system](https://github.com/cylc/cylc-flow/pull/3931) has been written. Hopefully we can deliver this for Cylc 8.0.0.
>
> Note that there will be a simple bash CLI completion to fill in flow//cycle/task names making this syntax more convenient.



Discussion / hints:
Discussion on Discourse with good user requirements: https://cylc.discourse.group/t/the-role-of-the-ui-server-in-production-from-a-user-and-permissions-point-of-view-cylc8/55
While testing the LDAP Authenticator today (#51), I started looking for the groups. I tried to understand the concept of groups in JupyterHub, related to Authenticators.

An authenticator and a spawner may use the system groups to authenticate the user, or when forking a spawner process to use the user/group IDs.

But after that, the REST API endpoint `/groups` and the SQLite database table `groups`, both seem to be more related to JupyterHub, and not to the system groups.

So even though my Linux user has over 5 groups (`kinow`, `cdrom`, `adm`, `docker`, etc), after my first log in, the groups table is empty.

I can use the REST API to add groups to my existing user, but these groups won't be really related to my OS users.

This may be useful in the future in case users want to have authorization based on their existing groups in Active Directory, FreeIPA, etc. I couldn't find an existing way to sync groups between OS and JupyterHub; though it should be doable either by extending the authenticator, or by using a [service](https://jupyterhub.readthedocs.io/en/stable/reference/services.html).
Kubernetes has a good model for RBAC: https://kubernetes.io/docs/reference/access-authn-authz/rbac/. Airflow appears to have used it as reference too (found the link in their docs).

Jenkins too, and it's configurable (using multiple strategies, like their matrix authorization strategy: https://plugins.jenkins.io/matrix-auth/)

I think JupyterHub RBAC might look similar to the Kubernetes model. In which case, we could use their authorization too.

(commenting here because the issue I found for Cylc Flow appears to be more for the CLI https://github.com/cylc/cylc-flow/issues/2729, and whatever is defined here would drive the UI too I think)
Comments on authorization here:

- https://github.com/cylc/cylc-uiserver/issues/27#issuecomment-668478159
- https://github.com/cylc/cylc-uiserver/issues/27#issuecomment-668478637
Authorization, from Feb 2020 workshop notes:

 - need site and user config
 - user can ramp up authorization levels as far as the site allows
 - groups:
    - (1) locally (site and user) defined lists
    - (2) access to Unix OS groups
    - (3) LDAP groups? (Does the Hub authentication plugin return this info? Probably not?)
          -  or query getent()?
          - (JR) OS groups would do, as we link OS and LDAP groups
     -  do (1) and (2) initially, (3) etc. after 8.0
- auth config file needs to be .cylc format

- authorization config files:
![image](https://user-images.githubusercontent.com/2362137/106984083-c8cf8f80-67cb-11eb-8ae8-2ee93525fb6f.png)

Plus:
- authorization methods and config, as a UIS sub-service


 
Further to this, and harking back to https://cylc.discourse.group/t/the-role-of-the-ui-server-in-production-from-a-user-and-permissions-point-of-view-cylc8/55/8 in some ways, our usage of Cylc provides read-only views into production versus read-execute, and read-write-execute views.

I understand that the Jupyter UI Servers expect the person logging in to have shell accounts to the machine that the UI Server is running on; and of course it would make sense for the standard use-case for the Cylc 8 UI Servers to have shell accounts too (for their own workflows).  However, in the case where we want to provide read-only views into production, we almost certainly do not want to provide those users with shell accounts.

It would be useful if there was a capability for having shell=false for most/all users who log in, while being able to see the list of other users' workflows they are authorized to see.  That is, my lack of shell or workflows, should not mean I can't see the workflows I have authorization to see.

> I understand that the Jupyter UI Servers expect the person logging in to have shell accounts to the machine that the UI Server is running on ...

No, not necessarily.  The (privileged) hub has to be able to start a UI Server as the target user (i.e. the owner of the workflows), not the authenticated user, if there is not already one running there.  The UIS will then authorize restricted access to its schedulers, according to what rights the owner has granted the authenticated user.

  
removing the question label as we have an accepted cylc-admin proposal - https://github.com/cylc/cylc-admin/blob/master/docs/proposal-multi-user-approach.md#configuration
And earlier, from me:
> Trying to recall the plan off the top off my head ...
>
> -    Cylc 8 schedulers run as the user will only accept connections from the user, so no authorization needed there.
> -    (They can be started from the command line or the UI Server)
> -    UI Servers run as the user too, and my UI Server will only interact with my schedulers
> -    UI Servers however can accept connections from other users, and will control authorized access to its scheudlers
> -    To view or interact with other users' schedulers, my UI will connect to the other users' UI Servers
>       - Note in general this requires the UI/hub to be able to spawn UI Servers on other user accounts, not just mine, in case there's no UI Server running there already. It can do this, because it is the only privileged part of the system (not technically any different than spawning my own UI Server). We just have to ensure that it can spawn ONLY Cylc UI Servers.
>       -  I'm sure it is technically possible to connect to existing UI Servers rather than spawn them, but [I'm not sure of] the details of that, and besides (IMO at least) a UIS running as me is a UIS running as me, it shouldn't matter who spawned it.


To summarise the above brain dump into some rough pathway:

1. Investigate authorising to other users UI servers.

   * Can we do this through the hub?
   * Do we need to configure jupyterhub some special way?
   * Proof of concept.

2. Develop a basic POC authorisation config file.

   * This is to allow one user to permit another to access their UIS.
   * We will need to add a permissions model later.
   * For a POC a simple yes/no with authorisation performed at the relevant end points will do.

3. Develop fine grained authorisation.

   * We would like a unix-like read/write/execute model.
   * Similar to the GitHub OAuth model it would be nice to be able to assign more fine-grained permissions independently of this (e.g. authorise a user to stop a workflow but without giving them full execute or write priv).
   * Auth file plans outlined here - https://github.com/cylc/cylc-uiserver/issues/10
Update:

* Above steps completed.
* #219 will provide standalone hub-less functionality and integrate with the hub better for multi-user setups.
* #204 will provide authorisation with fine grained configurability for mutations (i.e. commands).

Tagged against 0.6.0.

Fix patch:
diff --git a/.mailmap b/.mailmap
index 35da8c12..ca221fa8 100644
--- a/.mailmap
+++ b/.mailmap
@@ -3,3 +3,4 @@ Bruno Kinoshita <kinow@users.noreply.github.com>
 David Matthews <david.matthews@metoffice.gov.uk>
 Hilary Oliver <hilary.j.oliver@gmail.com>
 Sadie Bartholomew <30274190+sadielbartholomew@users.noreply.github.com>
+Mel Hall <37735232+datamel@users.noreply.github.com> Melanie Hall
diff --git a/CHANGES.md b/CHANGES.md
index dc5fbb2b..a22c00bb 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -18,6 +18,9 @@ Multi-user functionality implemented.
 
 ### Enhancements
 
+[#204](https://github.com/cylc/cylc-uiserver/pull/**204**) -
+Implementation of configurable multi user authorisation.
+
 [#230](https://github.com/cylc/cylc-uiserver/pull/230) -
 Convert the UI Server to a jupyter_server extension.
 
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index 908ec375..68c74345 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -42,6 +42,7 @@ below.
  - Stephen McVeigh
  - Ronnie Dutta
  - David Matthews
+ - Mel Hall
  <!-- end-shortlog -->
 
 (All contributors are identifiable with email addresses in the git version
diff --git a/README.md b/README.md
index 9e8221be..b7e27ace 100644
--- a/README.md
+++ b/README.md
@@ -168,6 +168,8 @@ Currently the UI Server accepts these configurations:
 * `c.CylcUIServer.ui_build_dir`
 * `c.CylcUIServer.ui_version`
 * `c.CylcUIServer.scan_iterval`
+* `c.CylcUIServer.site_authorization`
+* `c.CylcUIServer.user_authorization`
 
 See the `cylc.uiserver.app.UIServer` file for details.
 
diff --git a/cylc/uiserver/app.py b/cylc/uiserver/app.py
index a8a3a9b2..b1596d32 100644
--- a/cylc/uiserver/app.py
+++ b/cylc/uiserver/app.py
@@ -14,6 +14,7 @@
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
+import getpass
 import os
 from pathlib import Path, PurePath
 import sys
@@ -23,6 +24,7 @@
 from tornado import ioloop
 from tornado.web import RedirectHandler
 from traitlets import (
+    Dict,
     Float,
     TraitError,
     TraitType,
@@ -38,10 +40,13 @@
 from cylc.flow.network.graphql import (
     CylcGraphQLBackend, IgnoreFieldMiddleware
 )
-
 from cylc.uiserver import (
     __file__ as uis_pkg
 )
+from cylc.uiserver.authorise import (
+    Authorization,
+    AuthorizationMiddleware
+)
 from cylc.uiserver.data_store_mgr import DataStoreMgr
 from cylc.uiserver.handlers import (
     CylcStaticHandler,
@@ -102,6 +107,103 @@ class CylcUIServer(ExtensionApp):
             ]
         )
     )
+    # TODO: Add a link to the access group table mappings in cylc documentation
+    AUTH_DESCRIPTION = '''
+            Authorization can be granted at operation (mutation) level, i.e.
+            specifically grant user access to execute Cylc commands, e.g.
+            ``play``, ``pause``, ``edit``, ``trigger`` etc. For your
+            convenience, these operations have been mapped to access groups
+            ``READ``, ``CONTROL`` and ``ALL``.
+
+            To remove permissions, prepend the access group or operation with
+            ``!``.
+
+            Permissions are additive but negated permissions take precedence
+            above additions e.g. ``CONTROL, !stop`` will permit all operations
+            in the ``CONTROL`` group except for ``stop``.
+
+            .. note::
+
+               Any authorization permissions granted to a user will be
+               applied to all workflows.
+    '''
+
+    site_authorization = Dict(
+        config=True,
+        help='''
+            Dictionary containing site limits and defaults for authorization.
+
+            This configuration should be placed only in the site set
+            configuration file and not the user configuration file (use
+            ``c.CylcUIServer.user_authorization`` for user defined
+            authorization).
+
+            If this configuration is empty, site authorization defaults to no
+            configurable authorization and users will be unable to set any
+            authorization.
+
+            ''' + AUTH_DESCRIPTION + '''
+
+        .. rubric:: Example Configuration:
+
+        .. code-block:: python
+
+           c.CylcUIServer.site_authorization = {
+               "*": {  # For all ui-server owners,
+                   "*": {  # Any authenticated user
+                       "default": "READ",  # Will have default read-only access
+                   },
+                   "user1": {  # user1
+                       "default": ["!ALL"],  # No privileges for all ui-server
+                       # owners.
+                   },  # No limit set, so all ui-server owners
+               },  # limit is also "!ALL" for user1
+               "server_owner_1": {  # For specific UI Server owner,
+                   "group:group_a": {  # Any user who is a member of group_a
+                       "default": "READ",  # Will have default read-only access
+                       "limit": ["ALL", "!play"],  # server_owner_1 is able to
+                   },  # grant All privileges, except play.
+               },
+               "group:grp_of_svr_owners": {  # Group of owners of UI Servers
+                   "group:group_b": {
+                       "limit": [  # can grant groupB users up to READ and
+                           "READ",  # CONTROL privileges, without stop and
+                           "CONTROL",  # kill
+                           "!stop",
+                           "!kill",  # No default, so default is no access
+                       ],
+                   },
+               },
+           }
+
+        ''')
+
+    user_authorization = Dict(
+        config=True,
+        help='''
+            Dictionary containing authorized users and permission levels for
+            authorization.
+
+            Use this setting to share control of your workflows
+            with other users.
+            Note that you are only permitted to give away permissions up to
+            your limit for each user, as defined in the site_authorization
+            configuration.
+            ''' + AUTH_DESCRIPTION + '''
+            Example configuration, residing in
+            ``~/.cylc/hub/jupyter_config.py``:
+
+        .. code-block:: python
+
+           c.CylcUIServer.user_authorization = {
+               "*": ["READ"],  # any authenticated user has READ access
+               "group:group2": ["ALL"],  # Any user in system group2 has access
+                                         # to all operations
+               "userA": ["ALL", "!stop"],  # userA has ALL operations, not stop
+           }
+
+        '''
+    )
 
     ui_path = PathType(
         config=False,
@@ -189,6 +291,14 @@ def _check_ui_build_dir_exists(self, proposed):
             return proposed['value']
         raise TraitError(f'ui_build_dir does not exist: {proposed["value"]}')
 
+    @validate('site_authorization')
+    def _check_site_auth_dict_correct_format(self, proposed):
+        # TODO: More advanced auth dict validating
+        if isinstance(proposed['value'], dict):
+            return proposed['value']
+        raise TraitError(
+            f'Error in site authorization config: {proposed["value"]}')
+
     @staticmethod
     def _list_ui_versions(path: Path) -> List[str]:
         """Return a list of UI build versions detected in self.ui_path."""
@@ -243,12 +353,6 @@ def __init__(self, *args, **kwargs):
             log=self.log,
             workflows_mgr=self.workflows_mgr,
         )
-        self.subscription_server = TornadoSubscriptionServer(
-            schema,
-            backend=CylcGraphQLBackend(),
-            middleware=[IgnoreFieldMiddleware],
-        )
-
         ioloop.IOLoop.current().add_callback(
             self.workflows_mgr.update
         )
@@ -277,8 +381,15 @@ def initialize_settings(self):
         ).start()
 
     def initialize_handlers(self):
+        self.authobj = self.set_auth()
+        self.set_sub_server()
+
         self.handlers.extend([
-            ('cylc/version', CylcVersionHandler),
+            (
+                'cylc/version',
+                CylcVersionHandler,
+                {'auth': self.authobj}
+            ),
             (
                 'cylc/graphql',
                 UIServerGraphQLHandler,
@@ -286,7 +397,11 @@ def initialize_handlers(self):
                     'schema': schema,
                     'resolvers': self.resolvers,
                     'backend': CylcGraphQLBackend(),
-                    'middleware': [IgnoreFieldMiddleware],
+                    'middleware': [
+                        AuthorizationMiddleware,
+                        IgnoreFieldMiddleware
+                    ],
+                    'auth': self.authobj,
                 }
             ),
             (
@@ -296,8 +411,12 @@ def initialize_handlers(self):
                     'schema': schema,
                     'resolvers': self.resolvers,
                     'backend': CylcGraphQLBackend(),
-                    'middleware': [IgnoreFieldMiddleware],
+                    'middleware': [
+                        AuthorizationMiddleware,
+                        IgnoreFieldMiddleware
+                    ],
                     'batch': True,
+                    'auth': self.authobj,
                 }
             ),
             (
@@ -311,6 +430,7 @@ def initialize_handlers(self):
             (
                 'cylc/userprofile',
                 UserProfileHandler,
+                {'auth': self.authobj}
             ),
             (
                 'cylc/(.*)?',
@@ -330,6 +450,27 @@ def initialize_handlers(self):
             )
         ])
 
+    def set_sub_server(self):
+        self.subscription_server = TornadoSubscriptionServer(
+            schema,
+            backend=CylcGraphQLBackend(),
+            middleware=[
+                IgnoreFieldMiddleware,
+                AuthorizationMiddleware,
+            ],
+            auth=self.authobj
+        )
+
+    def set_auth(self):
+        """Create authorization object.
+        One for the lifetime of the UIServer.
+        """
+        return Authorization(
+            getpass.getuser(),
+            self.config.CylcUIServer.user_authorization,
+            self.config.CylcUIServer.site_authorization
+        )
+
     def initialize_templates(self):
         """Change the jinja templating environment."""
 
diff --git a/cylc/uiserver/authorise.py b/cylc/uiserver/authorise.py
new file mode 100644
index 00000000..8f1799d3
--- /dev/null
+++ b/cylc/uiserver/authorise.py
@@ -0,0 +1,542 @@
+# Copyright (C) NIWA & British Crown (Met Office) & Contributors.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+from contextlib import suppress
+from functools import lru_cache
+import graphene
+import grp
+from typing import List, Dict, Optional, Union, Any, Sequence, Set
+from inspect import iscoroutinefunction
+import logging
+import os
+import re
+from tornado import web
+from traitlets.config.loader import LazyConfigValue
+from traitlets.traitlets import Bool
+
+from cylc.uiserver.schema import UISMutations
+
+LOG = logging.getLogger(__name__)
+
+
+def constant(func):
+    """Decorator preventing reassignment"""
+
+    def fset(self, value):
+        raise TypeError
+
+    def fget():
+        return func()
+
+    return property(fget, fset)
+
+
+class Authorization:
+    """Authorization Information Class
+    One instance for the life of the UI Server. If authorization settings
+    change they will need to re-start the UI Server.
+    Authorization has access groups: `READ`, `CONTROL`, `ALL` - along with
+    their negations, `!READ`, `!CONTROL` and `!ALL` which indicate removal of
+    the permission groups.
+
+    """
+
+    # config literals
+    DEFAULT = "default"
+    LIMIT = "limit"
+    GRP_IDENTIFIER = "group:"
+
+    # Operations
+
+    ##########################################################################
+    #                               !WARNING!                                #
+    #                                                                        #
+    #   Beware of changing these permission groups. Users may be relying on  #
+    #   these settings. Changes should be widely publicised to users.        #
+    #                                                                        #
+    #   If adding/removing operations, ensure documentation is updated.      #
+    #                                                                        #
+    ##########################################################################
+
+    READ_OPERATION = "read"
+
+    # Access group identifiers (used in config)
+
+    READ = "READ"
+    CONTROL = "CONTROL"
+    ALL = "ALL"
+    NOT_READ = "!READ"
+    NOT_CONTROL = "!CONTROL"
+    NOT_ALL = "!ALL"
+
+    # Access Groups
+
+    READ_OPS = [
+        "ping",
+        READ_OPERATION,
+    ]
+
+    ASYNC_OPS = {"query", "mutation"}
+    READ_AUTH_OPS = {"query", "subscription"}
+
+    @staticmethod
+    @constant
+    def ALL_OPS() -> List[str]:
+        """ALL OPS constant, returns list of all mutations."""
+        return get_list_of_mutations()
+
+    @staticmethod
+    @constant
+    def CONTROL_OPS() -> List[str]:
+        """CONTROL OPS constant, returns list of all control mutations."""
+        return get_list_of_mutations(control=True)
+
+    def __init__(self, owner, owner_auth_conf, site_auth_conf) -> None:
+        self.owner = owner
+        self.owner_auth_conf = self.set_auth_conf(owner_auth_conf)
+        self.site_auth_config = self.set_auth_conf(site_auth_conf)
+        self.owner_user_info = {
+            "user": self.owner,
+            "user_groups": get_groups(self.owner),
+        }
+        self.owner_dict = self.build_owner_site_auth_conf()
+
+    @staticmethod
+    def expand_and_process_access_groups(permission_set: set) -> set:
+        """Process a permission set.
+
+        Takes a permission set, e.g. limits, defaults.
+        Expands the access groups and removes negated operations.
+
+        Args:
+            permission_set: set of permissions
+
+        Returns:
+            permission_set: processed permission set.
+        """
+        for action_group, expansion in {
+            Authorization.CONTROL: Authorization.CONTROL_OPS.fget(),
+            Authorization.ALL: Authorization.ALL_OPS.fget(),
+            Authorization.READ: Authorization.READ_OPS,
+        }.items():
+            if action_group in permission_set:
+                permission_set.remove(action_group)
+                permission_set.update(expansion)
+        # Expand negated permissions
+        for action_group, expansion in {
+                Authorization.NOT_CONTROL: [
+                    f"!{x}" for x in Authorization.CONTROL_OPS.fget()],
+                Authorization.NOT_ALL: [
+                    f"!{x}" for x in Authorization.ALL_OPS.fget()],
+                Authorization.NOT_READ: [
+                    f"!{x}" for x in Authorization.READ_OPS]}.items():
+            if action_group in permission_set:
+                permission_set.remove(action_group)
+                permission_set.update(expansion)
+        # Remove negated permissions
+        remove = set()
+
+        for perm in permission_set:
+            if perm.startswith("!"):
+                remove.add(perm.lstrip("!"))
+                remove.add(perm)
+        permission_set.difference_update(remove)
+        permission_set.discard("")
+        return permission_set
+
+    @staticmethod
+    def set_auth_conf(auth_conf: Union[LazyConfigValue, dict]) -> dict:
+        """Resolve lazy config where empty
+
+        Args:
+            auth_conf: Authorization configuration from a jupyter_config.py
+
+        Returns:
+            Valid configuration dictionary
+        """
+        if isinstance(auth_conf, LazyConfigValue) and not auth_conf.to_dict():
+            # no owner user auth - return empty dict
+            return {}
+        return auth_conf
+
+    def get_owner_site_limits_for_access_user(
+        self, access_user: Dict[str, Union[str, Sequence[Any]]]
+    ) -> Set:
+        """Returns limits owner can give to given access_user
+
+        Args:
+            access_user: Dictionary containing info about access user and their
+            membership of system groups.
+
+        Returns:
+            Set of limits that the uiserver owner is allowed to give away
+            for given access user.
+        """
+        limits: Set[str] = set()
+        if not self.owner_dict:
+            return limits
+        items_to_check = ["*", access_user["access_username"]]
+        items_to_check.extend(access_user["access_user_groups"])
+        for item in items_to_check:
+            permission: Union[str, List] = ""
+            default = ""
+            with suppress(KeyError):
+                default = self.owner_dict[item].get(Authorization.DEFAULT, "")
+            with suppress(KeyError):
+                permission = self.owner_dict[item].get(
+                    Authorization.LIMIT, default)
+            if permission == []:
+                raise_auth_config_exception("site")
+            if isinstance(permission, str):
+                limits.add(permission)
+            else:
+                limits.update(permission)
+        limits.discard("")
+        return limits
+
+    def get_access_user_permissions_from_owner_conf(
+        self, access_user: Dict[str, Union[str, Sequence[Any]]]
+    ) -> set:
+        """
+        Returns set of operations specific to access user from owner user conf.
+
+        Args:
+            access_user: Dictionary containing info about access user and their
+            membership of system groups. Defaults to None.
+        """
+        items_to_check = ["*", access_user["access_username"]]
+        items_to_check.extend(access_user["access_user_groups"])
+        allowed_operations = set()
+        for item in items_to_check:
+            permission = self.owner_auth_conf.get(item, "")
+            # Specifiying empty list equates to removing of all permissions.
+            if permission == []:
+                raise_auth_config_exception("user")
+            if isinstance(permission, str):
+                allowed_operations.add(permission)
+            else:
+                allowed_operations.update(permission)
+        allowed_operations.discard("")
+        return allowed_operations
+
+    @lru_cache(maxsize=128)
+    def get_permitted_operations(self, access_user: str):
+        """Return permitted operations for given access_user.
+
+        Cached for efficiency.
+        Checks:
+        - site config to ensure owner is permitted to give away permissions
+        - user config for authorised operations related to access_user and
+          their groups
+        - if user not in user config, then returns defaults from site config.
+
+        Args:
+            access_user: username to check for permitted operations
+
+        Returns:
+            Set of operations permitted by given access user for this UI Server
+        """
+        # For use in the ui, owner permissions (ALL operations) are set
+        if access_user == self.owner:
+            return set(Authorization.ALL_OPS.fget())
+        # Otherwise process permissions for (non-uiserver owner) access_user
+
+        access_user_dict = {
+            "access_username": access_user,
+            "access_user_groups": get_groups(access_user),
+        }
+        limits_owner_can_give = self.get_owner_site_limits_for_access_user(
+            access_user=access_user_dict)
+        user_conf_permitted_ops = (
+            self.get_access_user_permissions_from_owner_conf(
+                access_user=access_user_dict)
+        )
+        # If not explicit permissions for access user in owner conf then revert
+        # to site defaults
+        if len(user_conf_permitted_ops) == 0:
+            user_conf_permitted_ops = (
+                self.return_site_auth_defaults_for_access_user(
+                    access_user=access_user_dict
+                )
+            )
+        user_conf_permitted_ops = self.expand_and_process_access_groups(
+            user_conf_permitted_ops
+        )
+        limits_owner_can_give = self.expand_and_process_access_groups(
+            limits_owner_can_give
+        )
+        allowed_operations = limits_owner_can_give.intersection(
+            user_conf_permitted_ops
+        )
+        LOG.info(
+            f"User {access_user} authorized permissions: "
+            f"{sorted(allowed_operations)}"
+        )
+        return allowed_operations
+
+    def is_permitted(self, access_user: str, operation: str) -> Bool:
+        """Checks if user is permitted to action operation.
+
+        Args:
+            access_user: User attempting to action given operation.
+            operation: operation name
+
+        Returns:
+            True if access_user permitted to action operation, otherwise,
+            False.
+        """
+        if access_user == self.owner_user_info["user"]:
+            return True
+        # re.sub needed for snake/camel case
+        if re.sub(
+            r'(?<!^)(?=[A-Z])', '_', operation
+                ).lower() in self.get_permitted_operations(access_user):
+            LOG.info(f"{access_user}: authorized to {operation}")
+            return True
+        LOG.info(f"{access_user}: not authorized to {operation}")
+
+        return False
+
+    def build_owner_site_auth_conf(self):
+        """Build UI Server owner permissions dictionary.
+        Creates a reduced site auth dictionary for the ui-server owner.
+
+        """
+        owner_dict = {}
+        items_to_check = ["*", self.owner_user_info["user"]]
+        items_to_check.extend(self.owner_user_info["user_groups"])
+
+        # dict containing user info applying to the current ui_server owner
+        for uis_owner_conf, access_user_dict in self.site_auth_config.items():
+            if uis_owner_conf in items_to_check:
+                # acc_user = access_user
+                for acc_user_conf, acc_user_perms in access_user_dict.items():
+                    existing_user_conf = owner_dict.get(acc_user_conf)
+                    if existing_user_conf:
+                        # process limits and defaults and update dictionary
+                        existing_default = existing_user_conf.get(
+                            Authorization.DEFAULT, '')
+                        existing_limit = existing_user_conf.get(
+                            Authorization.LIMIT, existing_default)
+                        new_default = acc_user_perms.get(
+                            Authorization.DEFAULT, '')
+                        new_limit = acc_user_perms.get(
+                            Authorization.LIMIT, new_default)
+                        set_defs = set()
+                        for conf in [existing_default, new_default]:
+                            if isinstance(conf, list):
+                                set_defs.update(conf)
+                            else:
+                                set_defs.add(conf)
+                        set_lims = set()
+                        for conf in [existing_limit, new_limit]:
+                            if isinstance(conf, list):
+                                set_lims.update(conf)
+                            else:
+                                set_lims.add(conf)
+                        # update and continue
+                        owner_dict[
+                            acc_user_conf][Authorization.LIMIT] = list(
+                            set_lims)
+                        owner_dict[
+                            acc_user_conf][Authorization.DEFAULT] = list(
+                            set_defs)
+                        continue
+                    owner_dict.update(access_user_dict)
+        # Now we have a reduced site auth dictionary for the current owner
+        return owner_dict
+
+    def return_site_auth_defaults_for_access_user(
+        self, access_user: Dict[str, Union[str, Sequence[Any]]]
+    ) -> Set:
+        """Return site authorization defaults for given access user.
+        Args:
+            access_user: access_user dictionary, in the form
+                        {'access_username': username
+                         'access_user_group: [group1, group2,...]'
+                        }
+        Returns:
+            Set of default operations permitted
+
+        """
+        defaults: Set[str] = set()
+        if not self.owner_dict:
+            return defaults
+        items_to_check = ["*", access_user["access_username"]]
+        items_to_check.extend(access_user["access_user_groups"])
+        for item in items_to_check:
+            permission: Union[str, List] = ""
+            with suppress(KeyError):
+                permission = self.owner_dict[item].get(
+                    Authorization.DEFAULT, ""
+                )
+            if permission == []:
+                raise_auth_config_exception("site")
+            if isinstance(permission, str):
+                defaults.add(permission)
+            else:
+                defaults.update(permission)
+        defaults.discard("")
+        return defaults
+
+
+# GraphQL middleware
+class AuthorizationMiddleware:
+    """Authorization Middleware for authorization checking GraphQL.
+
+    Mutations are checked against permissions from config files.
+
+    Raises:
+        web.HTTPError: Unauthorized requests.
+    """
+
+    auth = None
+
+    def resolve(self, next_, root, info, **args):
+        current_user = info.context["current_user"]
+        # We won't be re-checking auth for return variables
+        if len(info.path) > 1:
+            return next_(root, info, **args)
+        op_name = self.get_op_name(info.field_name, info.operation.operation)
+        # It shouldn't get here but worth checking for zero trust
+        if not op_name:
+            self.auth_failed(current_user, op_name, 400,
+                             "Operation not in schema.")
+        try:
+            authorised = self.auth.is_permitted(current_user, op_name)
+        except Exception:
+            # Fail secure
+            authorised = False
+        if not authorised:
+            self.auth_failed(current_user, op_name, 403)
+        if (info.operation.operation in Authorization.ASYNC_OPS
+                or iscoroutinefunction(next_)):
+            return self.async_resolve(next_, root, info, **args)
+        return next_(root, info, **args)
+
+    @staticmethod
+    def auth_failed(current_user: str, op_name: str,
+                    http_code: int, message: Optional[str] = None):
+        """
+        Raise authorization error
+        Args:
+            current_user: username accessing operation
+            op_name: operation name
+            http_code: http error code to raise
+            message: Message to log Defaults to None.
+
+        Raises:
+            web.HTTPError
+        """
+        log_message = (f"Authorization failed for {current_user}"
+                       f":requested to {op_name}.")
+        if message:
+            log_message = log_message + " " + message
+        LOG.warning(log_message)
+        raise web.HTTPError(http_code, reason=message)
+
+    def get_op_name(self, field_name: str, operation: str) -> Union[None, str]:
+        """
+        Returns operation name required for authorization.
+        Converts queries and subscriptions to read operations.
+        Args:
+            field_name: Field name e.g. play
+            operation: operation type
+
+        Returns:
+            operation name
+        """
+        if operation in Authorization.READ_AUTH_OPS:
+            return Authorization.READ_OPERATION
+        else:
+            # Check it is a mutation in our schema
+            if self.auth and re.sub(
+                r'(?<!^)(?=[A-Z])', '_', field_name
+                    ).lower() in Authorization.ALL_OPS.fget():
+                return field_name
+        return None
+
+    async def async_resolve(self, next_, root, info, **args):
+        """Return awaited coroutine"""
+        return await next_(root, info, **args)
+
+
+def get_groups(username: str) -> List[str]:
+    """Return list of system groups for given user
+
+    Args:
+        username: username used to check system groups.
+
+    Returns:
+        list: system groups for username given
+    """
+    groupmax = os.NGROUPS_MAX  # type: ignore
+    group_ids = os.getgrouplist(username, groupmax)
+    group_ids.remove(groupmax)
+    # turn list of group_ids into group names with group identifier prepended
+    return parse_group_ids(group_ids)
+
+
+def parse_group_ids(group_ids: List) -> List:
+    """Returns list of groups in the correct format for authorisation.
+
+    Args:
+        group_ids: List of users groups, in number format
+
+    Returns:
+        List: List of users groups, in id format with group identifier
+        prepended.
+    """
+    group_list = []
+    for x in group_ids:
+        try:
+            group_list.append(
+                f"{Authorization.GRP_IDENTIFIER}{grp.getgrgid(x).gr_name}"
+            )
+        except OverflowError:
+            continue
+    return group_list
+
+
+def get_list_of_mutations(control=False) -> List[str]:
+    """Gets list of mutations"""
+    list_of_mutations = [
+        attr for attr in dir(UISMutations)
+        if isinstance(getattr(UISMutations, attr), graphene.Field)
+    ]
+    if control:
+        # Ping is a READ mutation
+        list_of_mutations.remove("ping")
+        # Broadcast is an ALL mutation
+        list_of_mutations.remove("broadcast")
+        return list_of_mutations
+    else:
+        # 'read' is used soley for authorization and is not a UISMutation
+        list_of_mutations.append(Authorization.READ_OPERATION)
+        return list_of_mutations
+
+
+def raise_auth_config_exception(config_type: str):
+    """Error raise for empty list in auth config.
+
+    Args:
+        config_type: Either site or user.
+    """
+    raise Exception(
+        f'Error in {config_type} config: '
+        f'`c.CylcUIServer.{config_type}_authorization`. '
+        f'"[]" is not supported. Use "{Authorization.NOT_ALL}" to remove all'
+        ' permissions.'
+    )
diff --git a/cylc/uiserver/handlers.py b/cylc/uiserver/handlers.py
index cb4ccd79..e4514f97 100644
--- a/cylc/uiserver/handlers.py
+++ b/cylc/uiserver/handlers.py
@@ -32,6 +32,7 @@
     list_plugins as list_cylc_plugins,
 )
 
+from cylc.uiserver.authorise import Authorization, AuthorizationMiddleware
 from cylc.uiserver.websockets import authenticated as websockets_authenticated
 
 
@@ -71,10 +72,10 @@ def _inner(
             isinstance(user, str)  # token authenticated
             or (
                 isinstance(user, dict)
-                and _authorise(handler, user['name'], '')
+                and _authorise(handler, user['name'])
             )
         ):
-            raise web.HTTPError(403, reason='authorisation insufficient')
+            raise web.HTTPError(403, reason='authorization insufficient')
         return fun(handler, *args, **kwargs)
     return _inner
 
@@ -113,19 +114,37 @@ def is_token_authenticated(
 
 def _authorise(
     handler: 'CylcAppHandler',
-    username: str,
-    action: str = 'READ'
+    username: str
 ) -> bool:
     """Authorises a user to perform an action.
 
-    Currently this returns False unless the authenticated user is the same
-    as the user this server is running under.
+    Returns True if the user is the UIServer owner, or the user has read
+    permissions.
+
     """
-    if username != ME:
-        # auth provided by the hub, check the user name
-        handler.log.warning(f'Authorisation failed for {username}')
+    if username == ME or handler.auth.is_permitted(
+            username, Authorization.READ_OPERATION):
+        return True
+    else:
+        handler.log.warning(f'Authorization failed for {username}')
         return False
-    return True
+
+
+def parse_current_user(current_user):
+    """Standardises and returns the current user."""
+    if isinstance(current_user, dict):
+        # the server is running with authentication services provided
+        # by a hub
+        current_user = dict(current_user)  # make a copy for safety
+        return current_user
+    else:
+        # the server is running using a token
+        # authentication is provided by jupyter server
+        return {
+            'kind': 'user',
+            'name': ME,
+            'server': socket.gethostname()
+        }
 
 
 class CylcAppHandler(JupyterHandler):
@@ -184,8 +203,11 @@ class CylcVersionHandler(CylcAppHandler):
     Equivalent to running `cylc version --long` in the UIS environment.
     """
 
-    @web.authenticated
+    def initialize(self, auth):
+        self.auth = auth
+
     @authorised
+    @web.authenticated
     def get(self):
         self.write(
             '<pre>'
@@ -196,6 +218,28 @@ def get(self):
         )
 
 
+def snake_to_camel(snake):
+    """Converts snake_case to camelCase
+        Examples:
+        >>> snake_to_camel('foo_bar_baz')
+        'fooBarBaz'
+        >>> snake_to_camel('')
+        ''
+        >>> snake_to_camel(None)
+        Traceback (most recent call last):
+        TypeError: <class 'NoneType'>
+        >>> snake_to_camel('ping')
+        'ping'
+
+    """
+    if isinstance(snake, str):
+        if not snake:
+            return ''
+        components = snake.split('_')
+        return components[0] + ''.join(x.title() for x in components[1:])
+    raise TypeError(type(snake))
+
+
 class UserProfileHandler(CylcAppHandler):
     """Provides information about the user in JSON format.
 
@@ -205,6 +249,9 @@ class UserProfileHandler(CylcAppHandler):
     would have returned.
     """
 
+    def initialize(self, auth):
+        self.auth = auth
+
     def set_default_headers(self) -> None:
         super().set_default_headers()
         self.set_header("Content-Type", 'application/json')
@@ -214,24 +261,19 @@ def set_default_headers(self) -> None:
     def get(self):
         user_info = self.get_current_user()
 
-        if isinstance(user_info, dict):
-            # the server is running with authentication services provided
-            # by a hub
-            user_info = dict(user_info)  # make a copy for safety
-        else:
-            # the server is running using a token
-            # authentication is provided by jupyter server
-            user_info = {
-                'kind': 'user',
-                'name': ME,
-                'server': socket.gethostname()
-            }
+        user_info = parse_current_user(user_info)
 
         # add an entry for the workflow owner
         # NOTE: when running behind a hub this may be different from the
         # authenticated user
         user_info['owner'] = ME
 
+        # Make user permissions available to the ui
+        user_info['permissions'] = [
+            snake_to_camel(perm) for perm in (
+                self.auth.get_permitted_operations(user_info['name']))
+                ]
+
         self.write(json.dumps(user_info))
 
 
@@ -243,6 +285,8 @@ class UIServerGraphQLHandler(CylcAppHandler, TornadoGraphQLHandler):
     somehow need to pass the request info (headers, username ...etc) in also
     """
 
+    # No authorization decorators here, auth handled in AuthorizationMiddleware
+
     # Declare extra attributes
     resolvers = None
 
@@ -251,12 +295,17 @@ def set_default_headers(self) -> None:
 
     def initialize(self, schema=None, executor=None, middleware=None,
                    root_value=None, graphiql=False, pretty=False,
-                   batch=False, backend=None, **kwargs):
+                   batch=False, backend=None, auth=None, **kwargs):
         super(TornadoGraphQLHandler, self).initialize()
-
+        self.auth = auth
         self.schema = schema
+
         if middleware is not None:
             self.middleware = list(self.instantiate_middleware(middleware))
+        # Make authorization info available to auth middleware
+        for mw in self.middleware:
+            if isinstance(mw, AuthorizationMiddleware):
+                mw.auth = self.auth
         self.executor = executor
         self.root_value = root_value
         self.pretty = pretty
@@ -270,20 +319,21 @@ def initialize(self, schema=None, executor=None, middleware=None,
 
     @property
     def context(self):
-        wider_context = {
+        """The GraphQL context passed to resolvers (incl middleware)."""
+        return {
             'graphql_params': self.graphql_params,
             'request': self.request,
             'resolvers': self.resolvers,
+            'current_user': parse_current_user(
+                self.get_current_user()
+            ).get('name'),
         }
-        return wider_context
 
     @web.authenticated
-    @authorised
     def prepare(self):
         super().prepare()
 
     @web.authenticated
-    @authorised
     async def execute(self, *args, **kwargs):
         # Use own backend, and TornadoGraphQLHandler already does validation.
         return await self.schema.execute(
@@ -295,14 +345,13 @@ async def execute(self, *args, **kwargs):
         )
 
     @web.authenticated
-    @authorised
     async def run(self, *args, **kwargs):
         await TornadoGraphQLHandler.run(self, *args, **kwargs)
 
 
 class SubscriptionHandler(CylcAppHandler, websocket.WebSocketHandler):
     """Endpoint for performing GraphQL subscriptions."""
-
+    # No authorization decorators here, auth handled in AuthorizationMiddleware
     def initialize(self, sub_server, resolvers):
         self.queue = Queue(100)
         self.subscription_server = sub_server
@@ -312,13 +361,11 @@ def select_subprotocol(self, subprotocols):
         return GRAPHQL_WS
 
     @websockets_authenticated
-    @authorised
     def get(self, *args, **kwargs):
         # forward this call so we can authenticate/authorise it
         return websocket.WebSocketHandler.get(self, *args, **kwargs)
 
     @websockets_authenticated  # noqa: A003
-    @authorised
     def open(self, *args, **kwargs):  # noqa: A003
         IOLoop.current().spawn_callback(
             self.subscription_server.handle,
@@ -337,8 +384,11 @@ def recv_nowait(self):
 
     @property
     def context(self):
-        wider_context = {
+        """The GraphQL context passed to resolvers (incl middleware)."""
+        return {
             'request': self.request,
             'resolvers': self.resolvers,
+            'current_user': parse_current_user(
+                self.get_current_user()
+            ).get('name'),
         }
-        return wider_context
diff --git a/cylc/uiserver/resolvers.py b/cylc/uiserver/resolvers.py
index 06392126..f5435907 100644
--- a/cylc/uiserver/resolvers.py
+++ b/cylc/uiserver/resolvers.py
@@ -15,7 +15,6 @@
 
 """GraphQL resolvers for use in data accessing and mutation of workflows."""
 
-import getpass
 import os
 from subprocess import Popen, PIPE, DEVNULL
 
@@ -124,14 +123,9 @@ async def play(cls, workflows, args, workflows_mgr, log):
             return cls._error(exc)
 
         # start each requested flow
-        me = getpass.getuser()
-        for user, flow, _ in workflows:
+        for _user, flow, _ in workflows:
             try:
-                if user and user != me:
-                    # TODO: multi-user support
-                    # forward this request to the other users UIS
-                    raise ValueError(f'Cannot start flow for user: {user}')
-
+                # Note: authorisation has already taken place.
                 # add the workflow to the command
                 cmd = [*cmd, flow]
 
@@ -202,7 +196,9 @@ async def mutator(self, info, *m_args):
             'request_string': req_str,
             'variables': variables,
         }
-        return self.workflows_mgr.multi_request('graphql', w_ids, graphql_args)
+        return await self.workflows_mgr.multi_request(
+            'graphql', w_ids, graphql_args
+        )
 
     async def service(self, info, *m_args):
         return await Services.play(
@@ -227,5 +223,6 @@ async def nodes_mutator(self, info, *m_args):
             'variables': variables,
         }
         multi_args = {w_id: graphql_args for w_id in w_ids}
-        return self.workflows_mgr.multi_request(
-            'graphql', w_ids, multi_args=multi_args)
+        return await self.workflows_mgr.multi_request(
+            'graphql', w_ids, multi_args=multi_args
+        )
diff --git a/cylc/uiserver/websockets/tornado.py b/cylc/uiserver/websockets/tornado.py
index 013fe78c..9fc3f765 100644
--- a/cylc/uiserver/websockets/tornado.py
+++ b/cylc/uiserver/websockets/tornado.py
@@ -4,7 +4,9 @@
 # The file was copied from this revision:
 # https://github.com/graphql-python/graphql-ws/blob/cf560b9a5d18d4a3908dc2cfe2199766cc988fef/graphql_ws/tornado.py
 
+import getpass
 from inspect import isawaitable, isclass
+import socket
 
 from asyncio import create_task, gather, wait, shield, sleep
 from asyncio.queues import QueueEmpty
@@ -21,6 +23,7 @@
 
 from typing import Union, Awaitable, Any, List, Tuple, Dict, Optional
 
+from cylc.uiserver.authorise import AuthorizationMiddleware
 setup_observable_extension()
 
 NO_MSG_DELAY = 1.0
@@ -47,10 +50,18 @@ async def close(self, code):
 
 
 class TornadoSubscriptionServer(BaseSubscriptionServer):
-    def __init__(self, schema, keep_alive=True, loop=None, backend=None, middleware=None):
+    def __init__(
+        self, schema,
+        keep_alive=True,
+        loop=None,
+        backend=None,
+        middleware=None,
+        auth=None
+    ):
         self.loop = loop
         self.backend = backend or None
         self.middleware = middleware
+        self.auth = auth
         super().__init__(schema, keep_alive)
 
     @staticmethod
@@ -72,6 +83,9 @@ def get_graphql_params(self, *args, **kwargs):
             )
         else:
             middleware = self.middleware
+        for mw in self.middleware:
+            if mw == AuthorizationMiddleware:
+                mw.auth = self.auth
         return dict(
             params,
             return_promise=True,


Test patch:
diff --git a/cylc/uiserver/tests/conftest.py b/cylc/uiserver/tests/conftest.py
index e0162923..02520f2a 100644
--- a/cylc/uiserver/tests/conftest.py
+++ b/cylc/uiserver/tests/conftest.py
@@ -15,6 +15,7 @@
 """Test code and fixtures."""
 
 import asyncio
+from cylc.uiserver.handlers import UIServerGraphQLHandler
 from getpass import getuser
 import inspect
 import logging
@@ -24,6 +25,7 @@
 from tempfile import TemporaryDirectory
 
 import pytest
+from tornado.web import HTTPError
 from traitlets.config import Config
 import zmq
 
@@ -189,11 +191,20 @@ def _read(self):
 def authorisation_true(monkeypatch):
     """Disabled request authorisation for test purposes."""
     monkeypatch.setattr(
-        'cylc.uiserver.handlers._authorised',
+        'cylc.uiserver.handlers._authorise',
         lambda x: True
     )
 
 
+@pytest.fixture
+def authorisation_false(monkeypatch):
+    """Disabled request authorisation for test purposes."""
+    monkeypatch.setattr(
+        'cylc.uiserver.handlers._authorise',
+        lambda x: False
+    )
+
+
 @pytest.fixture
 def mock_authentication(monkeypatch):
 
@@ -203,11 +214,38 @@ def _mock_authentication(user=None, server=None, none=False):
             'server': server or gethostname()
         }
         if none:
-            ret = 'anonymous'
+            ret = None
+            monkeypatch.setattr(
+                'cylc.uiserver.handlers.parse_current_user',
+                lambda x: {
+                    'kind': 'user',
+                    'name': None,
+                    'server': 'some_server'
+                }
+            )
+
+            def mock_redirect(*args):
+                # normally tornado would attempt to redirect us to the login
+                # page - for testing purposes we will skip this and raise
+                # a 403 with an explanatory reason
+                raise HTTPError(
+                    403,
+                    reason='login redirect replaced by 403 for test purposes'
+                )
+
+            monkeypatch.setattr(
+                'cylc.uiserver.handlers.CylcAppHandler.redirect',
+                mock_redirect
+            )
+
         monkeypatch.setattr(
             'cylc.uiserver.handlers.CylcAppHandler.get_current_user',
             lambda x: ret
         )
+        monkeypatch.setattr(
+            'cylc.uiserver.handlers.CylcAppHandler.get_login_url',
+            lambda x: "http://cylc"
+        )
 
     _mock_authentication()
 
diff --git a/cylc/uiserver/tests/test_auth.py b/cylc/uiserver/tests/test_auth.py
index 6c21376c..c66eea94 100644
--- a/cylc/uiserver/tests/test_auth.py
+++ b/cylc/uiserver/tests/test_auth.py
@@ -101,7 +101,7 @@ async def test_authorised_and_authenticated(
         pytest.param(
             ('cylc', 'graphql'),
             403,
-            'Forbidden',
+            'login redirect replaced by 403 for test purposes',
             None,
             id='cylc/graphql',
         ),
@@ -115,7 +115,7 @@ async def test_authorised_and_authenticated(
         pytest.param(
             ('cylc', 'userprofile'),
             403,
-            'Forbidden',
+            'login redirect replaced by 403 for test purposes',
             None,
             id='cylc/userprofile',
         )
@@ -138,23 +138,25 @@ async def test_unauthenticated(
     'endpoint,code,message,body',
     [
         pytest.param(
+            # should pass through authentication but fail as there is no query
             ('cylc', 'graphql'),
-            403,
-            'authorisation insufficient',
+            400,
+            'Bad Request',
             None,
             id='cylc/graphql',
         ),
         pytest.param(
+            # should pass through authentication but fail as there is no query
             ('cylc', 'subscriptions'),
-            403,
-            'authorisation insufficient',
+            400,
+            'Bad Request',
             None,
             id='cylc/subscriptions',
         ),
         pytest.param(
             ('cylc', 'userprofile'),
             403,
-            'authorisation insufficient',
+            'authorization insufficient',
             None,
             id='cylc/userprofile',
         )
@@ -171,6 +173,27 @@ async def test_unauthorised(
     await _test(jp_fetch, endpoint, code, message, body)
 
 
+@pytest.fixture
+def authorisation_middleware_instances(monkeypatch):
+    """Captures instances of the AuthorizationMiddleware class.
+
+    Returns a list which is updated with instances of AuthorizationMiddleware
+    created within the lifetime of the test function.
+    """
+    instances = []
+
+    def _init(self):
+        nonlocal instances
+        instances.append(self)
+
+    monkeypatch.setattr(
+        'cylc.uiserver.authorise.AuthorizationMiddleware.__init__',
+        _init
+    )
+
+    return instances
+
+
 async def _test(jp_fetch, endpoint, code, message, body):
     """Test 400 HTTP response (upgrade to websocket)."""
     fetch = partial(jp_fetch, *endpoint)
@@ -185,7 +208,7 @@ async def _test(jp_fetch, endpoint, code, message, body):
         if body:
             assert body in exc.response.body
     else:
-        # succees cases, test the response
+        # success cases, test the response
         response = await fetch()
         assert code == response.code
         if message:
diff --git a/cylc/uiserver/tests/test_authorise.py b/cylc/uiserver/tests/test_authorise.py
new file mode 100644
index 00000000..c83d953a
--- /dev/null
+++ b/cylc/uiserver/tests/test_authorise.py
@@ -0,0 +1,375 @@
+# Copyright (C) NIWA & British Crown (Met Office) & Contributors.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import pytest
+from pytest import MonkeyPatch
+from unittest.mock import patch
+
+from cylc.uiserver.authorise import (
+    Authorization,
+    AuthorizationMiddleware,
+    get_list_of_mutations,
+)
+
+CONTROL_OPS = [
+    "ext_trigger",
+    "hold",
+    "kill",
+    "message",
+    "pause",
+    "play",
+    "poll",
+    "release",
+    "release_hold_point",
+    "reload",
+    "remove",
+    "resume",
+    "set_graph_window_extent",
+    "set_hold_point",
+    "set_outputs",
+    "set_verbosity",
+    "stop",
+    "trigger",
+]
+
+ALL_OPS = [
+    "read",
+    "broadcast",
+    "ext_trigger",
+    "hold",
+    "kill",
+    "message",
+    "pause",
+    "ping",
+    "play",
+    "poll",
+    "release",
+    "release_hold_point",
+    "reload",
+    "remove",
+    "resume",
+    "set_graph_window_extent",
+    "set_hold_point",
+    "set_outputs",
+    "set_verbosity",
+    "stop",
+    "trigger",
+]
+
+FAKE_SITE_CONF = {
+    "*": {
+        "*": {
+            "default": "READ",
+        },
+        "user1": {
+            "limit": ["!ALL"],
+        },
+    },
+    "server_owner_1": {
+        "*": {"default": ["READ", "message", "!ping"],
+              "limit": ["READ", "CONTROL"]},
+        "user1": {"default": ["READ", "play", "pause"], "limit": ["ALL"]},
+    },
+    "server_owner_2": {
+        "user2": {"limit": "ALL"},
+        "group:groupA": {"default": ["READ", "CONTROL"]},
+    },
+    "group:grp_of_svr_owners": {
+        "group:group2": {
+            "default": "READ",
+            "limit": ["READ", "CONTROL", "!stop", "!kill"],
+        },
+    },
+}
+
+FAKE_USER_CONF = {
+    # Specified interactions granted to user1
+    "user1": ["READ", "pause", "trigger", "message"],
+    # Access to all operations workflows
+    "group:group1": ["ALL"],
+    # granted to users in group1
+    # All READ and CONTROL, except trigger and edit
+    "user2": ["READ", "CONTROL", "!trigger", "!edit"],
+    # granted to user2
+    "user3": ["READ"],
+    "user4": ["!ALL"],
+    "user5": ["play", "pause", "!ping"],
+    "group:group2": ["READ", "CONTROL", "!reload"],
+}
+
+
+@pytest.mark.parametrize(
+    "expected_operations, owner_name, owner_groups, user_name, user_groups",
+    [
+        pytest.param(
+            {
+                "message",
+                "play",
+                "trigger",
+                "resume",
+                "set_verbosity",
+                "set_graph_window_extent",
+                "ping",
+                "read",
+                "poll",
+                "hold",
+                "remove",
+                "set_hold_point",
+                "release_hold_point",
+                "ext_trigger",
+                "pause",
+                "pause",
+                "set_outputs",
+                "release",
+            },
+            "server_owner_2",
+            ["group:grp_of_svr_owners"],
+            "user7",
+            ["group:group2"],
+            id="user in * and groups, owner in * and groups",
+        ),
+        pytest.param(
+            {"read", "ping"},
+            "server_owner_3",
+            [""],
+            "user6",
+            [""],
+            id="user only in *, owner only in *",
+        ),
+        pytest.param(
+            {"read", "ping"},
+            "server_owner_2",
+            ["group:grp_of_svr_owners"],
+            "user-not-in-user-conf",
+            ["group:non-existent-group"],
+            id="user only in group and *",
+        ),
+        pytest.param(
+            set(),
+            "server_owner_1",
+            [""],
+            "user1",
+            [""],
+            id="user1 forbidden in site conf",
+        ),
+        pytest.param(
+            set(),
+            "server_owner_2",
+            ["group:grp_of_svr_owners"],
+            "user4",
+            ["group:group2"],
+            id="owner only in group and *",
+        )
+    ],
+)
+@patch("cylc.uiserver.authorise.get_groups")
+def test_get_permitted_operations(
+    mocked_get_groups,
+    expected_operations,
+    owner_name,
+    owner_groups,
+    user_name,
+    user_groups,
+):
+    mocked_get_groups.side_effect = [owner_groups, user_groups]
+    auth_obj = Authorization(
+        owner_name, FAKE_USER_CONF, FAKE_SITE_CONF
+    )
+    actual_operations = auth_obj.get_permitted_operations(
+        access_user=user_name
+    )
+    assert actual_operations == expected_operations
+
+
+@pytest.mark.parametrize(
+    "expected_operations, access_user_dict, owner_auth_conf,",
+    [
+        pytest.param(
+            {"!kill", "READ", "kill", "!stop", "pause", "play"},
+            {
+                "access_username": "access_user_1",
+                "access_user_groups": ["group:group1", "group:group2"],
+            },
+            {
+                "*": ["READ", "!kill"],
+                "access_user_1": ["READ", "pause", "kill", "play", "!stop"],
+            },
+            id="Check username in user conf and in *",
+        ),
+        pytest.param(
+            {"!ping", "READ"},
+            {
+                "access_username": "access_user_2",
+                "access_user_groups": ["group:group1", "group:group2"],
+            },
+            {
+                "*": ["READ", "!ping"],
+                "access_user_1": ["READ", "pause", "kill", "play", "!stop"],
+            },
+            id="Check user just in *",
+        ),
+        pytest.param(
+            {"!ping", "pause", "kill", "!stop", "READ", "CONTROL"},
+            {
+                "access_username": "access_user_1",
+                "access_user_groups": ["group:group1", "group:group2"],
+            },
+            {
+                "*": ["READ", "!ping"],
+                "access_user_1": ["READ", "pause", "kill", "!stop"],
+                "group:group1": ["CONTROL"],
+            },
+            id="Check group access permissions are added",
+        ),
+    ],
+)
+@patch("cylc.uiserver.authorise.get_groups")
+def test_get_access_user_permissions_from_owner_conf(
+    mocked_get_groups, expected_operations, access_user_dict, owner_auth_conf
+):
+    """Test the un-processed permissions of owner conf."""
+    mocked_get_groups.return_value = ["group:blah"]
+    authobj = Authorization(
+        "some_user", owner_auth_conf, {"fake": "config"}
+    )
+    permitted_operations = authobj.get_access_user_permissions_from_owner_conf(
+        access_user_dict
+    )
+    assert permitted_operations == expected_operations
+
+
+@pytest.mark.parametrize(
+    "permission_set, expected",
+    [
+        pytest.param(
+            {"!kill", "READ", "kill", "!stop", "!ping", "play"},
+            {"play", "read"},
+            id="Expansion with additions and negations",
+        ),
+        pytest.param(
+            {"!CONTROL", "ALL"},
+            {"broadcast", "ping", "read"},
+            id="Check expansion and negation",
+        ),
+        pytest.param(
+            {"READ", "!ALL", "broadcast", "CONTROL"},
+            set(),
+            id="Check expansion for !ALL with additions",
+        ),
+    ],
+)
+def test_expand_and_process_access_groups(permission_set, expected):
+    authobj = Authorization(
+        "some_user",
+        {"fake": "config"},
+        {"fake": "config"}
+    )
+    actual = authobj.expand_and_process_access_groups(permission_set)
+    assert actual == expected
+
+
+@pytest.mark.parametrize(
+    "mut_field_name, operation, expected_op_name",
+    [
+        pytest.param(
+            "play",
+            "mutation",
+            "play",
+            id="Mutation operation, play field, returns play",
+        ),
+        pytest.param(
+            "_schema",
+            "subscription",
+            "read",
+            id="Subscription operation, _schema field, returns read",
+        ),
+        pytest.param(
+            "blah",
+            "mutation",
+            None,
+            id="Mutation operation, invalid field, returns None",
+        ),
+        pytest.param(
+            "workflows",
+            "query",
+            "read",
+            id="Query operation, workflows field, returns read",
+        ),
+    ],
+)
+def test_get_op_name(mut_field_name, operation, expected_op_name):
+    mock_authobj = Authorization(
+        "some_user", {"fake": "config"},
+        {"fake": "config"}
+    )
+    auth_middleware = AuthorizationMiddleware
+    auth_middleware.auth = mock_authobj
+    actual_op_name = auth_middleware.get_op_name(
+        auth_middleware, mut_field_name, operation
+    )
+    assert actual_op_name == expected_op_name
+
+
+@pytest.mark.parametrize(
+    "owner_name,  user_name, get_permitted_operations_is_called, expected",
+    [
+        pytest.param(
+            "mel",
+            "mel",
+            False,
+            True,
+            id="Owner user always permitted"
+        ),
+        pytest.param(
+            "mel",
+            "tim",
+            True,
+            True,
+            id="User is not owner calls get_permitted_operations",
+        ),
+    ],
+)
+@patch("cylc.uiserver.authorise.Authorization.get_permitted_operations")
+@patch("cylc.uiserver.authorise.get_groups")
+def test_is_permitted(
+    mocked_get_groups,
+    mocked_get_permitted_operations,
+    owner_name,
+    user_name,
+    get_permitted_operations_is_called,
+    expected,
+):
+    mocked_get_groups.side_effect = [[""], [""]]
+    mocked_get_permitted_operations.return_value = ["fake_operation"]
+    auth_obj = Authorization(owner_name, FAKE_USER_CONF, FAKE_SITE_CONF)
+    actual = auth_obj.is_permitted(
+        access_user=user_name, operation="fake_operation"
+    )
+    if get_permitted_operations_is_called:
+        mocked_get_permitted_operations.assert_called_with(user_name)
+    assert actual == expected
+
+
+@pytest.mark.parametrize(
+    "control, expected",
+    [
+        pytest.param(False, ALL_OPS, id="All ops returned"),
+        pytest.param(True, CONTROL_OPS, id="Control ops returned"),
+    ],
+)
+def test_get_list_of_mutations(control, expected):
+    """Test ALL_OPS are returned"""
+    actual = get_list_of_mutations(control=control)
+    assert set(actual) == set(expected)
diff --git a/cylc/uiserver/tests/test_handlers.py b/cylc/uiserver/tests/test_handlers.py
index 5a15dc1e..0a472edb 100644
--- a/cylc/uiserver/tests/test_handlers.py
+++ b/cylc/uiserver/tests/test_handlers.py
@@ -15,7 +15,9 @@
 
 from functools import partial
 from getpass import getuser
+from unittest import mock
 from unittest.mock import MagicMock
+import pytest
 
 from graphql_ws.constants import GRAPHQL_WS
 from tornado.httputil import HTTPServerRequest
@@ -49,16 +51,20 @@ def _create_handler(self, logged_in=True):
         request.connection = MagicMock()
         handler = SubscriptionHandler(application=app, request=request,
                                       sub_server=None, resolvers=None)
+
         if logged_in:
-            handler.get_current_user = lambda: {'name': getuser()}
+            handler.get
+            _current_user = lambda: {'name': getuser()}
         else:
             handler.get_current_user = lambda: None
         return handler
 
+    @pytest.mark.usefixtures("mock_authentication")
     def test_websockets_subprotocol(self):
         handler = self._create_handler()
         assert handler.select_subprotocol(subprotocols=[]) == GRAPHQL_WS
 
+    @pytest.mark.usefixtures("mock_authentication")
     def test_websockets_check_origin_accepts_same_origin(self):
         """A request that includes the Host header must use the same
         value as the server host, or an error is raised.
@@ -73,6 +79,7 @@ def test_websockets_check_origin_accepts_same_origin(self):
         handler.request.headers['Host'] = host_header
         assert handler.check_origin(f'http://{host_header}')
 
+    @pytest.mark.usefixtures("mock_authentication")
     def test_websockets_check_origin_rejects_different_origin(self):
         """A request from a different Host MUST be blocked to prevent
         CORS attacks.
@@ -86,11 +93,13 @@ def test_websockets_check_origin_rejects_different_origin(self):
         handler.request.headers['Origin'] = 'ui.notcylc'
         assert not handler.check_origin('http://ui.notcylc')
 
+    @pytest.mark.usefixtures("mock_authentication")
     def test_websockets_context(self):
         handler = self._create_handler()
         assert handler.request == handler.context['request']
         assert 'resolvers' in handler.context
 
+    @pytest.mark.usefixtures("mock_authentication")
     def test_websockets_queue(self):
         handler = self._create_handler()
         message = 'a message'
@@ -102,6 +111,7 @@ def test_websockets_queue(self):
                                                 get_async_test_timeout())
         assert handler.queue.empty()
 
+    @pytest.mark.usefixtures("mock_authentication")
     def test_assert_callback_handler_gets_called(self):
         handler = self._create_handler()
         handler.subscription_server = MagicMock()