diff --git a/controlpanel/api/models/apps3bucket.py b/controlpanel/api/models/apps3bucket.py
index 9d12b6582..9cd5d44d4 100644
--- a/controlpanel/api/models/apps3bucket.py
+++ b/controlpanel/api/models/apps3bucket.py
@@ -2,9 +2,8 @@
 from django.db import models
 
 # First-party/Local
-from controlpanel.api import cluster
+from controlpanel.api import cluster, tasks
 from controlpanel.api.models.access_to_s3bucket import AccessToS3Bucket
-from controlpanel.api import tasks
 
 
 class AppS3Bucket(AccessToS3Bucket):
@@ -43,11 +42,6 @@ def __repr__(self):
 
     def grant_bucket_access(self):
         tasks.S3BucketGrantToApp(self, self.current_user).create_task()
-        # cluster.App(self.app).grant_bucket_access(
-        #     self.s3bucket.arn,
-        #     self.access_level,
-        #     self.resources,
-        # )
 
     def revoke_bucket_access(self):
-        cluster.App(self.app).revoke_bucket_access(self.s3bucket.arn)
+        tasks.S3BucketRevokeAppAccess(self, self.current_user).create_task()
diff --git a/controlpanel/api/models/s3bucket.py b/controlpanel/api/models/s3bucket.py
index 36efdfe91..90de336f3 100644
--- a/controlpanel/api/models/s3bucket.py
+++ b/controlpanel/api/models/s3bucket.py
@@ -10,10 +10,9 @@
 from django_extensions.db.models import TimeStampedModel
 
 # First-party/Local
-from controlpanel.api import cluster, validators
+from controlpanel.api import cluster, tasks, validators
 from controlpanel.api.models.apps3bucket import AppS3Bucket
 from controlpanel.api.models.users3bucket import UserS3Bucket
-from controlpanel.api import tasks
 
 
 def s3bucket_console_url(name):
@@ -129,24 +128,27 @@ def access_level(self, user):
 
     def save(self, *args, **kwargs):
         is_create = not self.pk
-
         super().save(*args, **kwargs)
-
-        if is_create:
-            bucket_owner = kwargs.pop("bucket_owner", self.bucket_owner)
-
-            # self.cluster.create(bucket_owner)
-            tasks.S3BucketCreate(self, self.created_by).create_task()
-
-            # XXX created_by is always set if model is saved by the API view
-            if self.created_by:
-                UserS3Bucket.objects.create(
-                    user=self.created_by,
-                    current_user=self.created_by,
-                    s3bucket=self,
-                    is_admin=True,
-                    access_level=UserS3Bucket.READWRITE,
-                )
+        if not is_create:
+            return self
+
+        tasks.S3BucketCreate(
+            entity=self,
+            user=self.created_by,
+            extra_data={
+                "bucket_owner": kwargs.pop("bucket_owner", self.bucket_owner),
+            }
+        ).create_task()
+
+        # created_by should always be set, but this is a failsafe
+        if self.created_by:
+            UserS3Bucket.objects.create(
+                user=self.created_by,
+                current_user=self.created_by,
+                s3bucket=self,
+                is_admin=True,
+                access_level=UserS3Bucket.READWRITE,
+            )
 
         return self
 
diff --git a/controlpanel/api/models/users3bucket.py b/controlpanel/api/models/users3bucket.py
index 4c5b87c44..7e5cfbec3 100644
--- a/controlpanel/api/models/users3bucket.py
+++ b/controlpanel/api/models/users3bucket.py
@@ -2,9 +2,8 @@
 from django.db import models
 
 # First-party/Local
-from controlpanel.api import cluster
+from controlpanel.api import cluster, tasks
 from controlpanel.api.models.access_to_s3bucket import AccessToS3Bucket
-from controlpanel.api import tasks
 
 
 class UserS3Bucket(AccessToS3Bucket):
@@ -46,22 +45,10 @@ def __repr__(self):
         )
 
     def grant_bucket_access(self):
-        if self.s3bucket.is_folder:
-            return cluster.User(self.user).grant_folder_access(
-                root_folder_path=self.s3bucket.name,
-                access_level=self.access_level,
-                paths=self.paths,
-            )
         tasks.S3BucketGrantToUser(self, self.current_user).create_task()
-        # cluster.User(self.user).grant_bucket_access(
-        #     self.s3bucket.arn,
-        #     self.access_level,
-        #     self.resources,
-        # )
 
     def revoke_bucket_access(self):
-        if self.s3bucket.is_folder:
-            return cluster.User(self.user).revoke_folder_access(
-                root_folder_path=self.s3bucket.name
-            )
-        cluster.User(self.user).revoke_bucket_access(self.s3bucket.arn)
+        # TODO when soft delete is added, this should be updated to use the user that
+        # has deleted the parent S3bucket to ensure we store the user that has sent the
+        # task in the case of cascading deletes
+        tasks.S3BucketRevokeUserAccess(self, self.current_user).create_task()
diff --git a/controlpanel/api/serializers.py b/controlpanel/api/serializers.py
index e33b21f14..ac17f4114 100644
--- a/controlpanel/api/serializers.py
+++ b/controlpanel/api/serializers.py
@@ -404,8 +404,10 @@ def _process_existing_env_settings(self, app_auth_settings, auth_settings_status
             env_data["variables"] = sorted(var_data.values(), key=lambda  x: x["name"])
             env_data["auth_required"] = auth_required
 
-
     def _process_redundant_envs(self, app_auth_settings, auth_settings_status):
+        # NB. if earlier call to get app_auth_settings failed, this will have been
+        # passed into serializer as an empty dict. Which results in all env details
+        # being marked as redundant mistakenly
         redundant_envs = list(set(auth_settings_status.keys()) -
                               set(app_auth_settings.keys()))
         for env_name in redundant_envs:
diff --git a/controlpanel/api/tasks/__init__.py b/controlpanel/api/tasks/__init__.py
index dd777c893..19e653ff1 100644
--- a/controlpanel/api/tasks/__init__.py
+++ b/controlpanel/api/tasks/__init__.py
@@ -1,7 +1,10 @@
 
-from controlpanel.api.tasks.app import AppCreateRole, AppCreateAuth
+# First-party/Local
+from controlpanel.api.tasks.app import AppCreateAuth, AppCreateRole
 from controlpanel.api.tasks.s3bucket import (
     S3BucketCreate,
-    S3BucketGrantToUser,
     S3BucketGrantToApp,
+    S3BucketGrantToUser,
+    S3BucketRevokeAppAccess,
+    S3BucketRevokeUserAccess,
 )
diff --git a/controlpanel/api/tasks/app.py b/controlpanel/api/tasks/app.py
index 0df460f9d..94468d322 100644
--- a/controlpanel/api/tasks/app.py
+++ b/controlpanel/api/tasks/app.py
@@ -1,4 +1,7 @@
+# Third-party
 from django.conf import settings
+
+# First-party/Local
 from controlpanel.api.tasks.task_base import TaskBase
 
 
diff --git a/controlpanel/api/tasks/handlers/__init__.py b/controlpanel/api/tasks/handlers/__init__.py
index 14190b744..454fd9094 100644
--- a/controlpanel/api/tasks/handlers/__init__.py
+++ b/controlpanel/api/tasks/handlers/__init__.py
@@ -1,13 +1,18 @@
+# First-party/Local
 from controlpanel import celery_app
-
+from controlpanel.api.tasks.handlers.app import CreateAppAuthSettings, CreateAppAWSRole
 from controlpanel.api.tasks.handlers.s3 import (
-    CreateS3Bucket, GrantAppS3BucketAccess, GrantUserS3BucketAccess,
+    CreateS3Bucket,
+    GrantAppS3BucketAccess,
+    GrantUserS3BucketAccess,
+    S3BucketRevokeAppAccess,
+    S3BucketRevokeUserAccess,
 )
-from controlpanel.api.tasks.handlers.app import CreateAppAuthSettings, CreateAppAWSRole
-
 
 create_app_aws_role = celery_app.register_task(CreateAppAWSRole())
 create_s3bucket = celery_app.register_task(CreateS3Bucket())
 grant_app_s3bucket_access = celery_app.register_task(GrantAppS3BucketAccess())
 grant_user_s3bucket_access = celery_app.register_task(GrantUserS3BucketAccess())
 create_app_auth_settings = celery_app.register_task(CreateAppAuthSettings())
+revoke_user_s3bucket_access = celery_app.register_task(S3BucketRevokeUserAccess())
+revoke_app_s3bucket_access = celery_app.register_task(S3BucketRevokeAppAccess())
diff --git a/controlpanel/api/tasks/handlers/app.py b/controlpanel/api/tasks/handlers/app.py
index 8b302954e..accc6d68b 100644
--- a/controlpanel/api/tasks/handlers/app.py
+++ b/controlpanel/api/tasks/handlers/app.py
@@ -1,22 +1,21 @@
+# First-party/Local
 from controlpanel.api import cluster
-from controlpanel.api.models import App
+from controlpanel.api.models import App, User
 from controlpanel.api.tasks.handlers.base import BaseModelTaskHandler
 
 
 class CreateAppAuthSettings(BaseModelTaskHandler):
     model = App
     name = "create_app_auth_settings"
-    permission_required = "api.create_app"
 
-    def has_permission(self, user, obj=None):
-        if not super().has_permission(user, obj):
-            return False
+    def handle(self, envs, disable_authentication, connections):
+        task_user = User.objects.filter(pk=self.task_user_pk).first()
+        if not task_user or not task_user.github_api_token:
+            # TODO maybe log this as something has gone wrong?
+            return self.complete()
 
-        return user.github_api_token
-
-    def run_task(self, app, user, envs, disable_authentication, connections):
         for env in envs:
-            cluster.App(app, user.github_api_token).create_auth_settings(
+            cluster.App(self.object, task_user.github_api_token).create_auth_settings(
                 env_name=env,
                 disable_authentication=disable_authentication,
                 connections=connections,
@@ -27,8 +26,7 @@ def run_task(self, app, user, envs, disable_authentication, connections):
 class CreateAppAWSRole(BaseModelTaskHandler):
     model = App
     name = "create_app_aws_role"
-    permission_required = "api.create_app"
 
-    def run_task(self, app, user):
-        cluster.App(app).create_iam_role()
+    def handle(self):
+        cluster.App(self.object).create_iam_role()
         self.complete()
diff --git a/controlpanel/api/tasks/handlers/base.py b/controlpanel/api/tasks/handlers/base.py
index 543a7e3b0..4f1455566 100644
--- a/controlpanel/api/tasks/handlers/base.py
+++ b/controlpanel/api/tasks/handlers/base.py
@@ -1,13 +1,34 @@
+# Third-party
 from celery import Task as CeleryTask
 
-from controlpanel.api.models import User, Task
+# First-party/Local
+from controlpanel.api.models import Task, User
 
 
-class BaseModelTaskHandler(CeleryTask):
+class BaseTaskHandler(CeleryTask):
+
+    def complete(self):
+        task = Task.objects.filter(task_id=self.request.id).first()
+        if task:
+            task.completed = True
+            task.save()
+
+    def run(self, *args, **kwargs):
+        self.handle(*args, **kwargs)
+
+    def handle(self, *args, **kwargs):
+        """
+        Should contain the logic to run the task, and will be called after the run
+        method has been successfully called.
+        """
+        raise NotImplementedError("Task logic not implemented")
+
+
+class BaseModelTaskHandler(BaseTaskHandler):
     name = None
     model = None
-    permission_required = None
-
+    object = None
+    task_user_pk = None
     # can be applied to project settings also
     # these settings mean that messages are only removed from the queue (acknowledged)
     # when returned. if an error occurs, they remain in the queue, and will be resent
@@ -24,58 +45,14 @@ def get_object(self, pk):
             # added back to the queue as could be due to a race condition
             raise exc
 
-    def get_user(self, pk):
-        """
-        Try to find the user, then check they have the correct permission required to
-        run the task action.
-        """
-        try:
-            return User.objects.get(pk=pk)
-        except User.DoesNotExist as exc:
-            # if the user is found, this should be a hard fail? So suggest log the error
-            # and then mark as complete to stop task being rerun?
-            return None
-
-    def has_permission(self, user, obj=None):
+    def run(self, obj_pk, task_user_pk, *args, **kwargs):
         """
-        Check that the user has permission to run the task on the given object.
-        Override on the subclass for further permission checks.
+        Default method that a celery Task object requires to be defined, and will be
+        called by the worker when a message is received by the queue. This will look up
+        the instance of the model, and store the PK of the user running the task to use
+        to look up the user later if required. The `handle` method is then called
+        with any other args and kwargs sent.
         """
-        if not self.permission_required:
-            raise NotImplementedError("Must define a permission to check")
-
-        if not user.has_perm(self.permission_required, obj=obj):
-            # log that the user did not have permission?
-            return False
-
-        return True
-
-    def complete(self):
-        task = Task.objects.filter(task_id=self.request.id).first()
-        if task:
-            task.completed = True
-            task.save()
-
-    def run(self, obj_pk, user_pk, *args, **kwargs):
-        """
-        Default message that a celery Task object requires to be defined, and will be
-        called by the worker when a message is received by the queue. This runs some
-        lookups and validates the user, and if these pass calls `run_task` which must
-        be defined on any subclass of BaseTaskHandler.
-        """
-        obj = self.get_object(obj_pk)
-        user = self.get_user(user_pk)
-        if not user:
-            return self.complete()
-
-        if not self.has_permission(user, obj):
-            return self.complete()
-
-        self.run_task(obj, user, *args, **kwargs)
-
-    def run_task(self, *args, **kwargs):
-        """
-        Should contain the logic to run the task, and will be called after the run
-        method has been successfully called.
-        """
-        raise NotImplementedError("Task logic not implemented")
+        self.object = self.get_object(obj_pk)
+        self.task_user_pk = task_user_pk
+        self.handle(*args, **kwargs)
diff --git a/controlpanel/api/tasks/handlers/s3.py b/controlpanel/api/tasks/handlers/s3.py
index 87928514a..fd8fe922e 100644
--- a/controlpanel/api/tasks/handlers/s3.py
+++ b/controlpanel/api/tasks/handlers/s3.py
@@ -1,28 +1,31 @@
+# Third-party
+from celery import Task as CeleryTask
+
+# First-party/Local
 from controlpanel.api import cluster
-from controlpanel.api.models import S3Bucket, UserS3Bucket, AppS3Bucket
-from controlpanel.api.tasks.handlers.base import BaseModelTaskHandler
+from controlpanel.api.models import App, AppS3Bucket, S3Bucket, User, UserS3Bucket
+from controlpanel.api.tasks.handlers.base import BaseModelTaskHandler, BaseTaskHandler
 
 
 class CreateS3Bucket(BaseModelTaskHandler):
     model = S3Bucket
     name = "create_s3bucket"
-    permission_required = "api.create_s3bucket"
 
-    def run_task(self, bucket, user, bucket_owner="APP"):
-        bucket.cluster.create(owner=bucket_owner)
+    def handle(self, bucket_owner=None):
+        bucket_owner = bucket_owner or "USER"
+        self.object.cluster.create(owner=bucket_owner)
         self.complete()
 
 
 class GrantAppS3BucketAccess(BaseModelTaskHandler):
     model = AppS3Bucket
     name = 'grant_app_s3bucket_access'
-    permission_required = 'api.create_apps3bucket'
 
-    def run_task(self, app_bucket, user):
-        cluster.App(app_bucket.app).grant_bucket_access(
-            app_bucket.s3bucket.arn,
-            app_bucket.access_level,
-            app_bucket.resources,
+    def handle(self):
+        cluster.App(self.object.app).grant_bucket_access(
+            self.object.s3bucket.arn,
+            self.object.access_level,
+            self.object.resources,
         )
         self.complete()
 
@@ -30,12 +33,43 @@ def run_task(self, app_bucket, user):
 class GrantUserS3BucketAccess(BaseModelTaskHandler):
     model = UserS3Bucket
     name = "grant_user_s3bucket_access"
-    permission_required = "api.create_users3bucket"
 
-    def run_task(self, user_bucket, user):
-        cluster.User(user_bucket.user).grant_bucket_access(
-            user_bucket.s3bucket.arn,
-            user_bucket.access_level,
-            user_bucket.resources,
-        )
+    def handle(self):
+        if self.object.s3bucket.is_folder:
+            cluster.User(self.object.user).grant_folder_access(
+                root_folder_path=self.object.s3bucket.name,
+                access_level=self.object.access_level,
+                paths=self.object.paths,
+            )
+        else:
+            cluster.User(self.object.user).grant_bucket_access(
+                bucket_arn=self.object.s3bucket.arn,
+                access_level=self.object.access_level,
+                path_arns=self.object.resources,
+            )
+        self.complete()
+
+
+class S3BucketRevokeUserAccess(BaseTaskHandler):
+    name = "revoke_user_s3bucket_access"
+
+    def handle(self, bucket_identifier, bucket_user_pk, is_folder):
+        bucket_user = User.objects.get(pk=bucket_user_pk)
+        if is_folder:
+            cluster.User(bucket_user).revoke_folder_access(bucket_identifier)
+        else:
+            cluster.User(bucket_user).revoke_bucket_access(bucket_identifier)
+        self.complete()
+
+
+class S3BucketRevokeAppAccess(BaseTaskHandler):
+    name = "revoke_app_s3bucket_access"
+
+    def handle(self, bucket_arn, app_pk):
+        try:
+            app = App.objects.get(pk=app_pk)
+        except App.DoesNotExist:
+            # if the app doesnt exist, nothing to revoke, so mark completed
+            self.complete()
+        cluster.App(app).revoke_bucket_access(bucket_arn)
         self.complete()
diff --git a/controlpanel/api/tasks/s3bucket.py b/controlpanel/api/tasks/s3bucket.py
index 21b3a2286..c59f4070d 100644
--- a/controlpanel/api/tasks/s3bucket.py
+++ b/controlpanel/api/tasks/s3bucket.py
@@ -1,4 +1,7 @@
+# Third-party
 from django.conf import settings
+
+# First-party/Local
 from controlpanel.api.tasks.task_base import TaskBase
 
 
@@ -14,34 +17,70 @@ def task_name(self):
     def task_description(self):
         return "creating s3 bucket"
 
+    def _get_args_list(self):
+        return [
+            self.entity.id,
+            self.user.id if self.user else 'None',
+            self.extra_data.get('bucket_owner'),
+        ]
 
-class S3BucketGrantToUser(TaskBase):
-    ENTITY_CLASS = "UserS3Bucket"
+
+class S3AccessMixin:
+    ACTION = None
+    ROLE = None
 
     @property
     def task_name(self):
-        return "grant_user_s3bucket_access"
+        return "{action}_{role}_s3bucket_access".format(
+            action=self.ACTION.lower(),
+            role=self.ROLE.lower()
+        )
 
     @property
     def task_description(self):
-        return "granting access to the user"
+        return "{action} access to the {role}".format(
+            action=self.ACTION.lower(),
+            role=self.ROLE.lower()
+        )
 
     @property
     def entity_description(self):
         return self.entity.s3bucket.name
 
 
-class S3BucketGrantToApp(TaskBase):
+class S3BucketGrantToUser(S3AccessMixin, TaskBase):
+    ENTITY_CLASS = "UserS3Bucket"
+    ACTION = "GRANT"
+    ROLE = "USER"
+
+
+class S3BucketGrantToApp(S3AccessMixin, TaskBase):
     ENTITY_CLASS = "AppS3Bucket"
+    ACTION = "GRANT"
+    ROLE = "APP"
 
-    @property
-    def task_name(self):
-        return "grant_app_s3bucket_access"
 
-    @property
-    def task_description(self):
-        return "granting access to the app"
+class S3BucketRevokeUserAccess(S3AccessMixin, TaskBase):
+    ENTITY_CLASS = "UserS3Bucket"
+    ACTION = "REVOKE"
+    ROLE = "USER"
 
-    @property
-    def entity_description(self):
-        return self.entity.s3bucket.name
+    def _get_args_list(self):
+        bucket = self.entity.s3bucket
+        return [
+            bucket.name if bucket.is_folder else bucket.arn,
+            self.entity.user.pk,
+            bucket.is_folder,
+        ]
+
+
+class S3BucketRevokeAppAccess(S3AccessMixin, TaskBase):
+    ENTITY_CLASS = "AppS3Bucket"
+    ACTION = "REVOKE"
+    ROLE = "APP"
+
+    def _get_args_list(self):
+        return [
+            self.entity.s3bucket.arn,
+            self.entity.app.pk,
+        ]
diff --git a/controlpanel/api/tasks/task_base.py b/controlpanel/api/tasks/task_base.py
index 96e79c7f1..8b124ed54 100644
--- a/controlpanel/api/tasks/task_base.py
+++ b/controlpanel/api/tasks/task_base.py
@@ -1,8 +1,13 @@
+# Standard library
 import uuid
+
+# Third-party
 from django.conf import settings
 
+# First-party/Local
 from controlpanel.api.message_broker import (
-    MessageBrokerClient, LocalMessageBrokerClient
+    LocalMessageBrokerClient,
+    MessageBrokerClient,
 )
 from controlpanel.api.models.task import Task
 
diff --git a/controlpanel/cli/management/commands/migrating_customers.py b/controlpanel/cli/management/commands/migrating_customers.py
index 9d978b8d0..ed92e993a 100644
--- a/controlpanel/cli/management/commands/migrating_customers.py
+++ b/controlpanel/cli/management/commands/migrating_customers.py
@@ -10,7 +10,10 @@
 
 
 class Command(BaseCommand):
-    help = "Copy the customers of an application from old auth client to new auth clients"
+    help = """Copy the customers of an application from old auth client to new auth
+        clients. This is idempotent, meaning that rerunning for the apps will not
+        problems, all copied customers will remain but any un-copied customers will be
+        migrated."""
 
     DEFAULT_ENVS = ["dev", "prod"]
 
diff --git a/controlpanel/frontend/views/app.py b/controlpanel/frontend/views/app.py
index c8a271fd9..d009bcd74 100644
--- a/controlpanel/frontend/views/app.py
+++ b/controlpanel/frontend/views/app.py
@@ -79,12 +79,17 @@ def _get_all_app_settings(self, app):
         access_repo_error_msg = None
         github_settings_access_error_msg = None
         try:
+            # NB: if this call fails....
             deployment_env_names = app_manager_ins.get_deployment_envs()
         except requests.exceptions.HTTPError as ex:
             access_repo_error_msg = ex.__str__()
+            github_settings_access_error_msg = ex.__str__()
+            # ...this is set to empty list...
             deployment_env_names = []
+        # ...which means this will remain empty dict...
         deployments_settings = {}
         auth0_connections = app.auth0_connections_by_env()
+        # ...so no call to get secrets/variables is made
         try:
             for env_name in deployment_env_names:
                 deployments_settings[env_name] = {
@@ -94,7 +99,7 @@ def _get_all_app_settings(self, app):
                 }
         except requests.exceptions.HTTPError as ex:
             github_settings_access_error_msg = ex.__str__()
-
+        # ...knock on effect is in serializers.py these envs will be marked as redundant
         return deployments_settings, access_repo_error_msg, \
             github_settings_access_error_msg
 
@@ -111,6 +116,8 @@ def get_context_data(self, **kwargs):
             exclude_connected=True,
         )
 
+        # If auth settings not returned, all envs marked redundant in the serializer.
+        # Should hide them instead?
         auth_settings, access_repo_error_msg, github_settings_access_error_msg \
             = self._get_all_app_settings(app)
         auth0_clients_status = app.auth0_clients_status()
@@ -305,6 +312,11 @@ class RevokeAppAccess(OIDCLoginRequiredMixin, PermissionRequiredMixin, DeleteVie
     model = AppS3Bucket
     permission_required = "api.remove_app_bucket"
 
+    def get_object(self, queryset=None):
+        obj = super().get_object(queryset=queryset)
+        obj.current_user = self.request.user
+        return obj
+
     def get_success_url(self):
         messages.success(self.request, "Successfully disconnected data source")
         return reverse_lazy("manage-app", kwargs={"pk": self.object.app.id})
diff --git a/controlpanel/frontend/views/datasource.py b/controlpanel/frontend/views/datasource.py
index e6ef13dfb..5099089ea 100644
--- a/controlpanel/frontend/views/datasource.py
+++ b/controlpanel/frontend/views/datasource.py
@@ -15,7 +15,7 @@
 from rules.contrib.views import PermissionRequiredMixin
 
 # First-party/Local
-from controlpanel.api import cluster
+from controlpanel.api import cluster, tasks
 from controlpanel.api.elasticsearch import bucket_hits_aggregation
 from controlpanel.api.models import (
     IAMManagedPolicy,
@@ -174,16 +174,15 @@ def form_valid(self, form):
         datasource_type = self.request.GET.get("type")
 
         try:
-            with transaction.atomic():
-                self.object = S3Bucket.objects.create(
-                    name=name,
-                    created_by=self.request.user,
-                    is_data_warehouse=datasource_type == "warehouse",
-                )
-                messages.success(
-                    self.request,
-                    f"Successfully created {name} {datasource_type} data source",
-                )
+            self.object = S3Bucket.objects.create(
+                name=name,
+                created_by=self.request.user,
+                is_data_warehouse=datasource_type == "warehouse",
+            )
+            messages.success(
+                self.request,
+                f"Successfully created {name} {datasource_type} data source",
+            )
         except Exception as ex:
             form.add_error("name", str(ex))
             return FormMixin.form_invalid(self, form)
@@ -294,6 +293,11 @@ class RevokeAccess(OIDCLoginRequiredMixin, PermissionRequiredMixin, DeleteView):
     model = UserS3Bucket
     permission_required = "api.destroy_users3bucket"
 
+    def get_object(self, queryset=None):
+        obj = super().get_object(queryset=queryset)
+        obj.current_user = self.request.user
+        return obj
+
     def get_success_url(self):
         messages.success(self.request, "Successfully revoked access")
         return reverse_lazy("manage-datasource", kwargs={"pk": self.object.s3bucket.id})
diff --git a/tests/api/models/test_apps3bucket.py b/tests/api/models/test_apps3bucket.py
index a0e0c75ec..6402c2a47 100644
--- a/tests/api/models/test_apps3bucket.py
+++ b/tests/api/models/test_apps3bucket.py
@@ -46,14 +46,16 @@ def test_aws_permissions(app, bucket, sqs, helpers):
 
     apps3bucket.save()
     messages = helpers.retrieve_messages(sqs)
-    helpers.validate_task_with_sqs_messages(messages, AppS3Bucket.__name__, apps3bucket.id)
+    helpers.validate_task_with_sqs_messages(
+        messages, AppS3Bucket.__name__, apps3bucket.id
+    )
 
 
 @pytest.mark.django_db
 def test_delete_revoke_permissions(app, bucket):
-    with patch("controlpanel.api.aws.AWSRole.grant_bucket_access"), \
-            patch("controlpanel.api.cluster.AWSRole.revoke_bucket_access") \
-                    as revoke_bucket_access_action:
+    with patch(
+        "controlpanel.api.tasks.S3BucketRevokeAppAccess"
+    ) as revoke_bucket_access_task:
         apps3bucket = mommy.make(
             "api.AppS3Bucket",
             app=app,
@@ -63,7 +65,8 @@ def test_delete_revoke_permissions(app, bucket):
 
         apps3bucket.delete()
 
-        revoke_bucket_access_action.assert_called_with(
-            apps3bucket.iam_role_name,
-            bucket.arn,
+        revoke_bucket_access_task.assert_called_with(
+            apps3bucket,
+            None
         )
+        revoke_bucket_access_task.return_value.create_task.assert_called()
diff --git a/tests/api/models/test_s3bucket.py b/tests/api/models/test_s3bucket.py
index a466c1af6..9d9d89f95 100644
--- a/tests/api/models/test_s3bucket.py
+++ b/tests/api/models/test_s3bucket.py
@@ -23,20 +23,16 @@ def bucket():
 
 
 def test_delete_revokes_permissions(bucket):
-    with patch("controlpanel.api.aws.AWSRole.grant_bucket_access"), \
-            patch("controlpanel.api.cluster.AWSRole.revoke_bucket_access") \
-            as revoke_bucket_access_action:
-        users3bucket = mommy.make("api.UserS3Bucket", s3bucket=bucket)
-        apps3bucket = mommy.make("api.AppS3Bucket", s3bucket=bucket)
-
+    with patch("controlpanel.api.models.AppS3Bucket.revoke_bucket_access") as app_revoke_bucket_access, \
+            patch("controlpanel.api.models.UserS3Bucket.revoke_bucket_access") as user_revoke_user_bucket_access:
+        # link the bucket with an UserS3Bucket and AppS3Bucket
+        mommy.make("api.UserS3Bucket", s3bucket=bucket)
+        mommy.make("api.AppS3Bucket", s3bucket=bucket)
+        # delete the source S3Bucket
         bucket.delete()
-
-        revoke_bucket_access_action.assert_has_calls(
-            [
-                call(apps3bucket.iam_role_name, bucket.arn),
-                call(users3bucket.iam_role_name, bucket.arn),
-            ]
-        )
+        # check that related objects revoke access methods were called
+        app_revoke_bucket_access.assert_called_once()
+        user_revoke_user_bucket_access.assert_called_once()
 
 
 def test_delete_marks_bucket_for_archival(bucket):
diff --git a/tests/api/models/test_users3bucket.py b/tests/api/models/test_users3bucket.py
index 8a55d89b5..7d8effcda 100644
--- a/tests/api/models/test_users3bucket.py
+++ b/tests/api/models/test_users3bucket.py
@@ -7,8 +7,8 @@
 from model_mommy import mommy
 
 # First-party/Local
-from controlpanel.api.models.access_to_s3bucket import AccessToS3Bucket
 from controlpanel.api.models import UserS3Bucket
+from controlpanel.api.models.access_to_s3bucket import AccessToS3Bucket
 
 
 @pytest.fixture
@@ -46,35 +46,33 @@ def test_aws_create_bucket(user, bucket, sqs, helpers):
         access_level=AccessToS3Bucket.READONLY
     )
     messages = helpers.retrieve_messages(sqs)
-    helpers.validate_task_with_sqs_messages(messages, UserS3Bucket.__name__, users3bucket.id)
+    helpers.validate_task_with_sqs_messages(
+        messages, UserS3Bucket.__name__, users3bucket.id
+    )
 
 
 @pytest.mark.django_db
-@patch("controlpanel.api.cluster.AWSRole.grant_folder_access")
-def test_aws_create_folder(grant_folder_access, user, bucket):
+@patch("controlpanel.api.models.users3bucket.tasks")
+def test_aws_create_folder(tasks, user, bucket):
     with patch.object(
         bucket.__class__, "is_folder", new_callable=PropertyMock(return_value=True)
     ):
-        user.users3buckets.create(
+        user_bucket = user.users3buckets.create(
             s3bucket=bucket,
             access_level=AccessToS3Bucket.READONLY,
+            current_user=user
         )
-        grant_folder_access.assert_called_with(
-            role_name=user.iam_role_name,
-            root_folder_path=bucket.name,
-            access_level=AccessToS3Bucket.READONLY,
-            paths=[],
+        tasks.S3BucketGrantToUser.assert_called_once_with(
+            user_bucket, user
         )
+        tasks.S3BucketGrantToUser.return_value.create_task.assert_called_once()
 
 
 @pytest.mark.django_db
-def test_delete_revoke_permissions(user, bucket, users3bucket):
+def test_delete_revoke_permissions(bucket, users3bucket):
     with patch(
-        "controlpanel.api.cluster.AWSRole.revoke_bucket_access"
-    ) as revoke_bucket_access_action:
+        "controlpanel.api.tasks.S3BucketRevokeUserAccess"
+    ) as revoke_user_access_task:
         users3bucket.delete()
-        revoke_bucket_access_action.assert_called_with(
-            user.iam_role_name,
-            bucket.arn,
-        )
-        # TODO get policy from call and assert bucket ARN removed
+        revoke_user_access_task.assert_called_with(users3bucket, None)
+        revoke_user_access_task.return_value.create_task.assert_called()
diff --git a/tests/api/tasks/test_create_app_auth_settings.py b/tests/api/tasks/test_create_app_auth_settings.py
index d52fa59f3..f2f5acfce 100644
--- a/tests/api/tasks/test_create_app_auth_settings.py
+++ b/tests/api/tasks/test_create_app_auth_settings.py
@@ -1,8 +1,11 @@
-import pytest
-from unittest.mock import patch, PropertyMock
+# Standard library
+from unittest.mock import PropertyMock, patch
 
+# Third-party
+import pytest
 from model_mommy import mommy
 
+# First-party/Local
 from controlpanel.api.models import App
 from controlpanel.api.tasks.handlers import create_app_auth_settings
 
@@ -21,12 +24,14 @@ def test_cluster_not_called_without_valid_app(cluster, complete, users):
 @pytest.mark.django_db
 @patch("controlpanel.api.tasks.handlers.base.BaseModelTaskHandler.complete")
 @patch("controlpanel.api.tasks.handlers.app.cluster")
-@patch("controlpanel.api.models.user.User.github_api_token", new=PropertyMock(return_value=None))
+@patch("controlpanel.api.models.user.User.github_api_token", new=PropertyMock(return_value=None))  # noqa
 def test_cluster_not_called_without_github_api_token(cluster, complete, users):
     app = mommy.make("api.App")
 
     user = users["superuser"]
-    create_app_auth_settings(app.pk, user.pk)
+    create_app_auth_settings(
+        app.pk, user.pk, "envs", "disable_authentication", "connections",
+    )
 
     # role should not be created
     cluster.App.assert_not_called()
@@ -37,7 +42,7 @@ def test_cluster_not_called_without_github_api_token(cluster, complete, users):
 @pytest.mark.django_db
 @patch("controlpanel.api.tasks.handlers.base.BaseModelTaskHandler.complete")
 @patch("controlpanel.api.tasks.handlers.app.cluster")
-@patch("controlpanel.api.models.user.User.github_api_token", new=PropertyMock(return_value="dummy-token"))
+@patch("controlpanel.api.models.user.User.github_api_token", new=PropertyMock(return_value="dummy-token"))  # noqa
 def test_valid_app_and_user(cluster, complete, users):
     app = mommy.make("api.App")
 
diff --git a/tests/api/tasks/test_create_app_aws_role.py b/tests/api/tasks/test_create_app_aws_role.py
index 8b719c01a..a882a9845 100644
--- a/tests/api/tasks/test_create_app_aws_role.py
+++ b/tests/api/tasks/test_create_app_aws_role.py
@@ -1,10 +1,16 @@
-import pytest
+# Standard library
 from unittest.mock import patch
 
+# Third-party
+import pytest
 from model_mommy import mommy
 
+# First-party/Local
 from controlpanel.api.models import App
-from controlpanel.api.tasks.handlers import create_app_aws_role, create_app_auth_settings
+from controlpanel.api.tasks.handlers import (
+    create_app_auth_settings,
+    create_app_aws_role,
+)
 
 
 @pytest.mark.django_db
@@ -18,21 +24,6 @@ def test_cluster_not_called_without_valid_app(cluster, complete, users):
         complete.assert_not_called()
 
 
-@pytest.mark.django_db
-@patch("controlpanel.api.tasks.handlers.base.BaseModelTaskHandler.complete")
-@patch("controlpanel.api.tasks.handlers.app.cluster")
-def test_cluster_not_called_without_valid_user(cluster, complete, users):
-    app = mommy.make("api.App")
-
-    user = users["normal_user"]
-    create_app_aws_role(app.pk, user.pk)
-
-    # role should not be created
-    cluster.App.assert_not_called()
-    # task should be complete so that it does not run again, as user not valid
-    complete.assert_called_once()
-
-
 @pytest.mark.django_db
 @patch("controlpanel.api.tasks.handlers.base.BaseModelTaskHandler.complete")
 @patch("controlpanel.api.tasks.handlers.app.cluster")
diff --git a/tests/api/tasks/test_s3.py b/tests/api/tasks/test_s3.py
index babd83877..dc275accf 100644
--- a/tests/api/tasks/test_s3.py
+++ b/tests/api/tasks/test_s3.py
@@ -1,11 +1,16 @@
+# Third-party
 import pytest
 from mock import patch
 from model_mommy import mommy
 
-from controlpanel.api import cluster
-from controlpanel.api.models import S3Bucket, AppS3Bucket, UserS3Bucket
+# First-party/Local
+from controlpanel.api.models import AppS3Bucket, S3Bucket, UserS3Bucket
 from controlpanel.api.tasks.handlers import (
-    create_s3bucket, grant_app_s3bucket_access, grant_user_s3bucket_access,
+    create_s3bucket,
+    grant_app_s3bucket_access,
+    grant_user_s3bucket_access,
+    revoke_app_s3bucket_access,
+    revoke_user_s3bucket_access,
 )
 
 
@@ -15,7 +20,7 @@
     (grant_app_s3bucket_access, AppS3Bucket),
     (grant_user_s3bucket_access, UserS3Bucket),
 ])
-@patch("controlpanel.api.tasks.handlers.base.BaseModelTaskHandler.complete")
+@patch("controlpanel.api.tasks.handlers.base.BaseTaskHandler.complete")
 def test_exception_raised_when_called_without_valid_app(
     complete, users, task_method, model
 ):
@@ -26,15 +31,19 @@ def test_exception_raised_when_called_without_valid_app(
 
 
 @pytest.mark.django_db
-@patch("controlpanel.api.tasks.handlers.base.BaseModelTaskHandler.complete")
+@patch("controlpanel.api.tasks.handlers.base.BaseTaskHandler.complete")
 @patch("controlpanel.api.models.s3bucket.cluster")
 def test_bucket_created(cluster, complete, users):
-    s3bucket = mommy.make("api.S3Bucket")
+    s3bucket = mommy.make("api.S3Bucket", bucket_owner="APP")
 
-    create_s3bucket(s3bucket.pk, users["superuser"].pk, bucket_owner="APP")
+    create_s3bucket(
+        s3bucket.pk, users["superuser"].pk, bucket_owner=s3bucket.bucket_owner
+    )
 
     cluster.S3Bucket.assert_called_once_with(s3bucket)
-    cluster.S3Bucket.return_value.create.assert_called_once_with(owner="APP")
+    cluster.S3Bucket.return_value.create.assert_called_once_with(
+        owner=s3bucket.bucket_owner
+    )
     complete.assert_called_once()
 
 
@@ -43,7 +52,7 @@ def test_bucket_created(cluster, complete, users):
     (grant_app_s3bucket_access, "api.AppS3Bucket", "App"),
     (grant_user_s3bucket_access, "api.UserS3Bucket", "User"),
 ])
-@patch("controlpanel.api.tasks.handlers.base.BaseModelTaskHandler.complete")
+@patch("controlpanel.api.tasks.handlers.base.BaseTaskHandler.complete")
 @patch("controlpanel.api.tasks.handlers.s3.cluster")
 def test_access_granted(
     cluster, complete, users, task_method, model_class, cluster_class
@@ -56,3 +65,50 @@ def test_access_granted(
     cluster_obj.assert_called_once()
     cluster_obj.return_value.grant_bucket_access.assert_called_once()
     complete.assert_called_once()
+
+
+@pytest.mark.django_db
+@pytest.mark.parametrize("bucket_name, is_folder", [
+    ("example-bucket", False),
+    ("example-bucket/folder", True),
+])
+@patch("controlpanel.api.tasks.handlers.base.BaseTaskHandler.complete")
+@patch("controlpanel.api.tasks.handlers.s3.cluster")
+def test_revoke_user_access(cluster, complete, bucket_name, is_folder):
+    user_bucket_access = mommy.make("api.UserS3Bucket", s3bucket__name=bucket_name)
+    s3bucket = user_bucket_access.s3bucket
+    bucket_identifier = s3bucket.name if is_folder else s3bucket.arn
+    revoke_user_s3bucket_access(
+        bucket_identifier=bucket_identifier,
+        bucket_user_pk=user_bucket_access.user.pk,
+        is_folder=is_folder
+    )
+
+    cluster.User.assert_called_once_with(user_bucket_access.user)
+    if is_folder:
+        cluster.User.return_value.revoke_folder_access.assert_called_once_with(
+            bucket_identifier
+        )
+    else:
+        cluster.User.return_value.revoke_bucket_access.assert_called_once_with(
+            bucket_identifier
+        )
+
+    complete.assert_called_once()
+
+
+@pytest.mark.django_db
+@patch("controlpanel.api.tasks.handlers.base.BaseTaskHandler.complete")
+@patch("controlpanel.api.tasks.handlers.s3.cluster")
+def test_revoke_app_access(cluster, complete):
+    app_bucket_access = mommy.make("api.AppS3Bucket")
+    revoke_app_s3bucket_access(
+        bucket_arn=app_bucket_access.s3bucket.arn,
+        app_pk=app_bucket_access.app.pk,
+    )
+
+    cluster.App.assert_called_once_with(app_bucket_access.app)
+    cluster.App.return_value.revoke_bucket_access.assert_called_once_with(
+        app_bucket_access.s3bucket.arn
+    )
+    complete.assert_called_once()
diff --git a/tests/api/views/test_apps3bucket.py b/tests/api/views/test_apps3bucket.py
index 8a7a45479..c61df0bac 100644
--- a/tests/api/views/test_apps3bucket.py
+++ b/tests/api/views/test_apps3bucket.py
@@ -60,16 +60,12 @@ def test_detail(client, apps3buckets):
 
 def test_delete(client, apps3buckets):
     with patch(
-        "controlpanel.api.aws.AWSRole.revoke_bucket_access"
+        "controlpanel.api.models.AppS3Bucket.revoke_bucket_access"
     ) as revoke_bucket_access:
         response = client.delete(reverse("apps3bucket-detail", (apps3buckets[1].id,)))
         assert response.status_code == status.HTTP_204_NO_CONTENT
 
-        revoke_bucket_access.assert_called_with(
-            apps3buckets[1].iam_role_name,
-            apps3buckets[1].s3bucket.arn,
-        )
-        # TODO get policy document JSON from call and assert bucket ARN not present
+        revoke_bucket_access.assert_called()
 
         response = client.get(reverse("apps3bucket-detail", (apps3buckets[1].id,)))
         assert response.status_code == status.HTTP_404_NOT_FOUND
@@ -85,7 +81,9 @@ def test_create(client, apps, buckets, sqs, helpers):
     assert response.status_code == status.HTTP_201_CREATED
     apps3bucket = AppS3Bucket.objects.get(app=apps[1], s3bucket=buckets[3])
     messages = helpers.retrieve_messages(sqs)
-    helpers.validate_task_with_sqs_messages(messages, AppS3Bucket.__name__, apps3bucket.id)
+    helpers.validate_task_with_sqs_messages(
+        messages, AppS3Bucket.__name__, apps3bucket.id
+    )
 
 
 def test_update(client, apps, apps3buckets, buckets, sqs, helpers):
@@ -103,7 +101,9 @@ def test_update(client, apps, apps3buckets, buckets, sqs, helpers):
     assert response.data["access_level"] == data["access_level"]
     apps3bucket = AppS3Bucket.objects.get(app=apps[1], s3bucket=buckets[1])
     messages = helpers.retrieve_messages(sqs)
-    helpers.validate_task_with_sqs_messages(messages, AppS3Bucket.__name__, apps3bucket.id)
+    helpers.validate_task_with_sqs_messages(
+        messages, AppS3Bucket.__name__, apps3bucket.id
+    )
 
 
 def test_update_bad_requests(client, apps, apps3buckets, buckets):
diff --git a/tests/api/views/test_users3bucket.py b/tests/api/views/test_users3bucket.py
index 62610a9ef..5677b1579 100644
--- a/tests/api/views/test_users3bucket.py
+++ b/tests/api/views/test_users3bucket.py
@@ -62,16 +62,12 @@ def test_create(client, buckets, users, sqs, helpers):
 
 def test_delete(client, users3buckets):
     with patch(
-        "controlpanel.api.aws.AWSRole.revoke_bucket_access"
+        "controlpanel.api.models.UserS3Bucket.revoke_bucket_access"
     ) as revoke_bucket_access:
         response = client.delete(reverse("users3bucket-detail", (users3buckets[1].id,)))
         assert response.status_code == status.HTTP_204_NO_CONTENT
 
-        revoke_bucket_access.assert_called_with(
-            users3buckets[1].user.iam_role_name,
-            users3buckets[1].s3bucket.arn,
-        )
-        # TODO get policy from call and assert bucket ARN not contained
+        revoke_bucket_access.assert_called()
 
         response = client.get(reverse("users3bucket-detail", (users3buckets[1].id,)))
         assert response.status_code == status.HTTP_404_NOT_FOUND
diff --git a/tests/frontend/views/test_datasource.py b/tests/frontend/views/test_datasource.py
index 4851663a5..1f43f02a0 100644
--- a/tests/frontend/views/test_datasource.py
+++ b/tests/frontend/views/test_datasource.py
@@ -274,7 +274,11 @@ def test_list_other_datasources_admins(client, buckets, users):
     assert other_datasources_admins[buckets["other"].id] == []
 
 
-def test_bucket_creator_has_readwrite_and_admin_access(client, users):
+@patch("controlpanel.api.models.users3bucket.tasks.S3BucketGrantToUser")
+@patch("controlpanel.api.models.s3bucket.tasks.S3BucketCreate")
+def test_bucket_creator_has_readwrite_and_admin_access(
+    create_bucket_task, grant_user_task, client, users
+):
     user = users["normal_user"]
     client.force_login(user)
     create(client)
@@ -282,6 +286,8 @@ def test_bucket_creator_has_readwrite_and_admin_access(client, users):
     ub = user.users3buckets.all()[0]
     assert ub.access_level == UserS3Bucket.READWRITE
     assert ub.is_admin
+    create_bucket_task.assert_called_once()
+    grant_user_task.assert_called_once()
 
 
 @pytest.mark.parametrize(
@@ -303,31 +309,44 @@ def test_create_get_form_class(rf, folders_enabled, datasource_type, form_class)
         assert view.get_form_class() == form_class
 
 
+@patch("controlpanel.api.models.users3bucket.tasks.S3BucketGrantToUser")
+@patch("controlpanel.api.models.s3bucket.tasks.S3BucketCreate")
 @patch("django.conf.settings.features.s3_folders.enabled", True)
-def test_create_folders(client, users, root_folder_bucket):
+@pytest.mark.parametrize("user", ["superuser", "normal_user", "other_user"])
+def test_create_folders(
+    create_bucket_task, grant_user_task, user, client, users, root_folder_bucket
+):
     """
     Check that all users can create a folder datasource
     """
-    for _, user in users.items():
-        client.force_login(user)
-        folder_name = f"test-{user.username}-folder"
-        response = create(client, name=folder_name)
-
-        # redirect expected on success
-        assert response.status_code == 302
-        assert user.users3buckets.filter(
-            s3bucket__name=f"{root_folder_bucket.name}/{folder_name}"
-        ).exists()
-
-        # create another folder to catch any errors updating IAM policy
-        folder_name = f"test-{user.username}-folder-2"
-        response = create(client, name=folder_name)
-
-        # redirect expected on success
-        assert response.status_code == 302
-        assert user.users3buckets.filter(
-            s3bucket__name=f"{root_folder_bucket.name}/{folder_name}"
-        ).exists()
+    user = users[user]
+    client.force_login(user)
+    folder_name = f"test-{user.username}-folder"
+    response = create(client, name=folder_name)
+
+    # redirect expected on success
+    assert response.status_code == 302
+    assert user.users3buckets.filter(
+        s3bucket__name=f"{root_folder_bucket.name}/{folder_name}"
+    ).exists()
+    # make sure tasks are sent
+    create_bucket_task.assert_called_once()
+    grant_user_task.assert_called_once()
+
+    # create another folder to catch any errors updating IAM policy
+    create_bucket_task.reset_mock()
+    grant_user_task.reset_mock()
+    folder_name = f"test-{user.username}-folder-2"
+    response = create(client, name=folder_name)
+
+    # redirect expected on success
+    assert response.status_code == 302
+    assert user.users3buckets.filter(
+        s3bucket__name=f"{root_folder_bucket.name}/{folder_name}"
+    ).exists()
+    # tasks to create are sent again for the second folder
+    create_bucket_task.assert_called_once()
+    grant_user_task.assert_called_once()
 
 
 @patch("django.conf.settings.features.s3_folders.enabled", False)