diff --git a/common-lib/src/main/scala/com/gu/mediaservice/lib/BaseStore.scala b/common-lib/src/main/scala/com/gu/mediaservice/lib/BaseStore.scala
index 9c6aa257b5c..cd7df655a29 100644
--- a/common-lib/src/main/scala/com/gu/mediaservice/lib/BaseStore.scala
+++ b/common-lib/src/main/scala/com/gu/mediaservice/lib/BaseStore.scala
@@ -1,49 +1,25 @@
 package com.gu.mediaservice.lib
 
-import org.apache.pekko.actor.{Cancellable, Scheduler}
-import com.gu.mediaservice.lib.aws.S3
+import com.gu.mediaservice.lib.aws.{S3, S3Bucket}
 import com.gu.mediaservice.lib.config.CommonConfig
 import com.gu.mediaservice.lib.logging.GridLogging
+import org.apache.pekko.actor.{Cancellable, Scheduler}
 import org.joda.time.DateTime
-import software.amazon.awssdk.services.s3.model.{GetObjectRequest, ListObjectsV2Request}
 
 import java.util.concurrent.atomic.AtomicReference
-import java.io.InputStream
-import scala.jdk.CollectionConverters._
 import scala.concurrent.ExecutionContext
 import scala.concurrent.duration._
 import scala.util.control.NonFatal
 
 
-abstract class BaseStore[TStoreKey, TStoreVal](bucket: String, config: CommonConfig)(implicit ec: ExecutionContext)
+abstract class BaseStore[TStoreKey, TStoreVal](bucket: S3Bucket, config: CommonConfig, s3: S3)(implicit ec: ExecutionContext)
   extends GridLogging {
 
-  val s3 = new S3(config)
-
   protected val store: AtomicReference[Map[TStoreKey, TStoreVal]] = new AtomicReference(Map.empty)
   protected val lastUpdated: AtomicReference[DateTime] = new AtomicReference(DateTime.now())
 
   protected def getS3Object(key: String): Option[String] = s3.getObjectAsString(bucket, key)
 
-  protected def getLatestS3Stream: Option[InputStream] = {
-    val objects = s3.client.listObjectsV2(ListObjectsV2Request.builder().bucket(bucket).build())
-      .contents().asScala.toList
-      .filterNot(_.key() == "AMAZON_SES_SETUP_NOTIFICATION")
-
-    if (objects.nonEmpty) {
-      val obj = objects.maxBy(_.lastModified())
-      logger.info(s"Latest key ${obj.key} in bucket $bucket")
-
-      val stream = s3.client.getObject(
-        GetObjectRequest.builder().key(obj.key()).bucket(bucket).build()
-      )
-      Some(stream)
-    } else {
-      logger.error(s"Bucket $bucket is empty")
-      None
-    }
-  }
-
   private var cancellable: Option[Cancellable] = None
 
   def scheduleUpdates(scheduler: Scheduler): Unit = {
diff --git a/common-lib/src/main/scala/com/gu/mediaservice/lib/ImageIngestOperations.scala b/common-lib/src/main/scala/com/gu/mediaservice/lib/ImageIngestOperations.scala
index ef96e09cf12..3bff1d70d4a 100644
--- a/common-lib/src/main/scala/com/gu/mediaservice/lib/ImageIngestOperations.scala
+++ b/common-lib/src/main/scala/com/gu/mediaservice/lib/ImageIngestOperations.scala
@@ -1,17 +1,15 @@
 package com.gu.mediaservice.lib
 
 
-import com.gu.mediaservice.lib.aws.S3Object
+import com.gu.mediaservice.lib.aws.{S3Bucket, S3Object}
 import com.gu.mediaservice.lib.config.CommonConfig
 import com.gu.mediaservice.lib.logging.LogMarker
 import com.gu.mediaservice.model.{Instance, MimeType}
 import com.typesafe.scalalogging.StrictLogging
 import org.joda.time.DateTime
-import software.amazon.awssdk.services.s3.model.{Delete, DeleteObjectsRequest, ObjectIdentifier}
 
 import java.io.File
 import scala.concurrent.Future
-import scala.jdk.CollectionConverters._
 
 object ImageIngestOperations {
   def fileKeyFromId(id: String)(implicit instance: Instance): String = instance.id + "/" + snippetForId(id)
@@ -21,7 +19,7 @@ object ImageIngestOperations {
   private def snippetForId(id: String) = id.take(6).mkString("/") + "/" + id
 }
 
-class ImageIngestOperations(imageBucket: String, thumbnailBucket: String, config: CommonConfig, isVersionedS3: Boolean = false)
+class ImageIngestOperations(imageBucket: S3Bucket, thumbnailBucket: S3Bucket, config: CommonConfig, isVersionedS3: Boolean = false)
   extends S3ImageStorage(config) with StrictLogging {
 
   import ImageIngestOperations.{fileKeyFromId, optimisedPngKeyFromId}
@@ -36,7 +34,7 @@ class ImageIngestOperations(imageBucket: String, thumbnailBucket: String, config
   private def storeOriginalImage(storableImage: StorableOriginalImage)
                         (implicit logMarker: LogMarker): Future[S3Object] = {
     val instanceSpecificKey = instanceAwareOriginalImageKey(storableImage)
-    logger.info(s"Storing original image to instance specific key:$imageBucket / $instanceSpecificKey")
+    logger.info(s"Storing original image to instance specific key:${imageBucket.name} / $instanceSpecificKey")
     storeImage(imageBucket, instanceSpecificKey, storableImage.file, Some(storableImage.mimeType),
       storableImage.meta, overwrite = false)
   }
@@ -44,7 +42,7 @@ class ImageIngestOperations(imageBucket: String, thumbnailBucket: String, config
   private def storeThumbnailImage(storableImage: StorableThumbImage)
                                  (implicit logMarker: LogMarker): Future[S3Object] = {
     val instanceSpecificKey = instanceAwareThumbnailImageKey(storableImage)
-    logger.info(s"Storing thumbnail to instance specific key: $thumbnailBucket / $instanceSpecificKey")
+    logger.info(s"Storing thumbnail to instance specific key: ${thumbnailBucket.name} / $instanceSpecificKey")
     storeImage(thumbnailBucket, instanceSpecificKey, storableImage.file, Some(storableImage.mimeType),
       overwrite = true)
   }
@@ -52,29 +50,15 @@ class ImageIngestOperations(imageBucket: String, thumbnailBucket: String, config
   private def storeOptimisedImage(storableImage: StorableOptimisedImage)
                                  (implicit logMarker: LogMarker): Future[S3Object] = {
     val instanceSpecificKey = optimisedPngKeyFromId(storableImage.id)(storableImage.instance)
-    logger.info(s"Storing optimised image to instance specific key: $thumbnailBucket / $instanceSpecificKey")
+    logger.info(s"Storing optimised image to instance specific key: ${thumbnailBucket.name} / $instanceSpecificKey")
     storeImage(imageBucket, instanceSpecificKey, storableImage.file, Some(storableImage.mimeType),
       overwrite = true)
   }
 
-
-  private def bulkDelete(bucket: String, keys: List[String]): Future[Map[String, Boolean]] = keys match {
+  private def bulkDelete(bucket: S3Bucket, keys: List[String]): Future[Map[String, Boolean]] = keys match {
     case Nil => Future.successful(Map.empty)
     case _ => Future {
-      val objects = keys.map { key =>
-        ObjectIdentifier.builder()
-          .key(key)
-          .build()
-      }.asJava
-      val response = client.deleteObjects(
-        DeleteObjectsRequest.builder().bucket(bucket)
-          .delete(Delete.builder().objects(objects).build())
-          .build()
-      )
-      val errorKeys = response.errors().asScala.toList.map(_.key())
-      keys.map { key =>
-        key -> !errorKeys.contains(key)
-      }.toMap
+      deleteObjects(bucket, keys)
     }
   }
 
@@ -106,7 +90,7 @@ sealed trait ImageWrapper {
   val instance: Instance
 }
 sealed trait StorableImage extends ImageWrapper {
-  def toProjectedS3Object(thumbBucket: String): S3Object = S3Object(
+  def toProjectedS3Object(thumbBucket: S3Bucket): S3Object = S3Object(
     thumbBucket,
     ImageIngestOperations.fileKeyFromId(id)(instance),
     file,
@@ -118,7 +102,7 @@ sealed trait StorableImage extends ImageWrapper {
 
 case class StorableThumbImage(id: String, file: File, mimeType: MimeType, meta: Map[String, String] = Map.empty, instance: Instance) extends StorableImage
 case class StorableOriginalImage(id: String, file: File, mimeType: MimeType, lastModified: DateTime, meta: Map[String, String] = Map.empty, instance: Instance) extends StorableImage {
-  override def toProjectedS3Object(thumbBucket: String): S3Object = S3Object(
+  override def toProjectedS3Object(thumbBucket: S3Bucket): S3Object = S3Object(
     thumbBucket,
     ImageIngestOperations.fileKeyFromId(id)(instance),
     file,
@@ -128,7 +112,7 @@ case class StorableOriginalImage(id: String, file: File, mimeType: MimeType, las
   )
 }
 case class StorableOptimisedImage(id: String, file: File, mimeType: MimeType, meta: Map[String, String] = Map.empty, instance: Instance) extends StorableImage {
-  override def toProjectedS3Object(thumbBucket: String): S3Object = S3Object(
+  override def toProjectedS3Object(thumbBucket: S3Bucket): S3Object = S3Object(
     thumbBucket,
     ImageIngestOperations.optimisedPngKeyFromId(id)(instance),
     file,
diff --git a/common-lib/src/main/scala/com/gu/mediaservice/lib/ImageQuarantineOperations.scala b/common-lib/src/main/scala/com/gu/mediaservice/lib/ImageQuarantineOperations.scala
index 0cc3a146a0e..eaf6b82979c 100644
--- a/common-lib/src/main/scala/com/gu/mediaservice/lib/ImageQuarantineOperations.scala
+++ b/common-lib/src/main/scala/com/gu/mediaservice/lib/ImageQuarantineOperations.scala
@@ -2,13 +2,13 @@ package com.gu.mediaservice.lib
 
 import java.io.File
 import com.gu.mediaservice.lib.config.CommonConfig
-import com.gu.mediaservice.lib.aws.S3Object
+import com.gu.mediaservice.lib.aws.{S3Bucket, S3Object}
 import com.gu.mediaservice.lib.logging.LogMarker
 import com.gu.mediaservice.model.{Instance, MimeType}
 
 import scala.concurrent.Future
 
-class ImageQuarantineOperations(quarantineBucket: String, config: CommonConfig, isVersionedS3: Boolean = false)
+class ImageQuarantineOperations(quarantineBucket: S3Bucket, config: CommonConfig, isVersionedS3: Boolean = false)
   extends S3ImageStorage(config) {
 
   def storeQuarantineImage(id: String, file: File, mimeType: Option[MimeType], meta: Map[String, String] = Map.empty)
diff --git a/common-lib/src/main/scala/com/gu/mediaservice/lib/ImageStorage.scala b/common-lib/src/main/scala/com/gu/mediaservice/lib/ImageStorage.scala
index 1e57a3513af..1369b5e1eb4 100644
--- a/common-lib/src/main/scala/com/gu/mediaservice/lib/ImageStorage.scala
+++ b/common-lib/src/main/scala/com/gu/mediaservice/lib/ImageStorage.scala
@@ -6,7 +6,7 @@ import java.io.File
 import scala.concurrent.{ExecutionContext, Future}
 import scala.concurrent.duration._
 import scala.language.postfixOps
-import com.gu.mediaservice.lib.aws.S3Object
+import com.gu.mediaservice.lib.aws.{S3Bucket, S3Object}
 import com.gu.mediaservice.lib.logging.LogMarker
 import com.gu.mediaservice.model.MimeType
 
@@ -37,9 +37,9 @@ trait ImageStorage {
   /** Store a copy of the given file and return the URI of that copy.
     * The file can safely be deleted afterwards.
     */
-  def storeImage(bucket: String, id: String, file: File, mimeType: Option[MimeType],
+  def storeImage(bucket: S3Bucket, id: String, file: File, mimeType: Option[MimeType],
                  meta: Map[String, String] = Map.empty, overwrite: Boolean)
                 (implicit logMarker: LogMarker): Future[S3Object]
 
-  def deleteImage(bucket: String, id: String)(implicit logMarker: LogMarker): Future[Unit]
+  def deleteImage(bucket: S3Bucket, id: String)(implicit logMarker: LogMarker): Future[Unit]
 }
diff --git a/common-lib/src/main/scala/com/gu/mediaservice/lib/S3ImageStorage.scala b/common-lib/src/main/scala/com/gu/mediaservice/lib/S3ImageStorage.scala
index acaf5503db6..6af69154965 100644
--- a/common-lib/src/main/scala/com/gu/mediaservice/lib/S3ImageStorage.scala
+++ b/common-lib/src/main/scala/com/gu/mediaservice/lib/S3ImageStorage.scala
@@ -1,10 +1,10 @@
 package com.gu.mediaservice.lib
 
-import com.gu.mediaservice.lib.aws.S3
+import com.gu.mediaservice.lib.aws.{S3, S3Bucket}
 import com.gu.mediaservice.lib.config.CommonConfig
 import com.gu.mediaservice.lib.logging.{GridLogging, LogMarker}
 import com.gu.mediaservice.model.MimeType
-import software.amazon.awssdk.services.s3.model.{DeleteObjectRequest, HeadObjectRequest, ListObjectsV2Request}
+import software.amazon.awssdk.services.s3.model.ListObjectsV2Request
 
 import java.io.File
 import scala.concurrent.Future
@@ -14,10 +14,10 @@ import scala.jdk.CollectionConverters._
 class S3ImageStorage(config: CommonConfig) extends S3(config) with ImageStorage with GridLogging {
 
   private val cacheSetting = Some(cacheForever)
-  def storeImage(bucket: String, id: String, file: File, mimeType: Option[MimeType],
+  def storeImage(bucket: S3Bucket, id: String, file: File, mimeType: Option[MimeType],
                  meta: Map[String, String] = Map.empty, overwrite: Boolean)
                 (implicit logMarker: LogMarker) = {
-    logger.info(logMarker, s"bucket: $bucket, id: $id, meta: $meta")
+    logger.info(logMarker, s"bucket: ${bucket.name}, id: $id, meta: $meta")
     val eventualObject = if (overwrite) {
       store(bucket, id, file, mimeType, meta, cacheSetting)
     } else {
@@ -27,27 +27,23 @@ class S3ImageStorage(config: CommonConfig) extends S3(config) with ImageStorage
     eventualObject
   }
 
-  def deleteImage(bucket: String, key: String)(implicit logMarker: LogMarker) = Future {
-    logger.info(logMarker, s"Deleted image $key from bucket $bucket")
-    client.deleteObject(
-      DeleteObjectRequest.builder().bucket(bucket).key(key).build())
+  def deleteImage(bucket: S3Bucket, key: String)(implicit logMarker: LogMarker) = Future {
+    deleteObject(bucket, key)
+    logger.info(logMarker, s"Deleted image $key from bucket ${bucket.name}")
   }
 
-  def deleteVersionedImage(bucket: String, id: String)(implicit logMarker: LogMarker) = Future {
-    val objectVersion = client.headObject(HeadObjectRequest.builder().bucket(bucket).key(id).build()).versionId()
-    client.deleteObject(DeleteObjectRequest.builder().bucket(bucket).key(id).versionId(objectVersion).build())
-    logger.info(logMarker, s"Deleted image $id from bucket $bucket (version: $objectVersion)")
+  def deleteVersionedImage(bucket: S3Bucket, id: String)(implicit logMarker: LogMarker) = Future {
+    val objectVersion = getMetadata(bucket, id).objectVersion.getOrElse(
+      throw new IllegalStateException(s"No version id found for $id in bucket ${bucket.name}")
+    )
+    deleteVersion(bucket, id, objectVersion)
+    logger.info(logMarker, s"Deleted image $id from bucket ${bucket.name} (version: $objectVersion)")
   }
 
-  def deleteFolder(bucket: String, id: String)(implicit logMarker: LogMarker) = Future {
-    val files = client.listObjectsV2(
-      ListObjectsV2Request.builder().bucket(bucket).prefix(id).build()
-    ).contents().asScala.toList
+  def deleteFolder(bucket: S3Bucket, id: String)(implicit logMarker: LogMarker): Future[Unit] = list(bucket, id).map { files =>
     logger.info(s"Found ${files.size} files to delete in folder $id")
-    files.foreach(file => client.deleteObject(
-      DeleteObjectRequest.builder().bucket(bucket).key(file.key()).build()
-    ))
-		logger.info(logMarker, s"Deleting images in folder $id from bucket $bucket")
-	}
+    files.foreach(file => deleteObject(bucket, bucket.keyFromURL(file.uri)))
+    logger.info(logMarker, s"Deleting images in folder $id from bucket $bucket")
+  }
 
 }
diff --git a/common-lib/src/main/scala/com/gu/mediaservice/lib/auth/KeyStore.scala b/common-lib/src/main/scala/com/gu/mediaservice/lib/auth/KeyStore.scala
index 2bd89378de8..fdf7c30d966 100644
--- a/common-lib/src/main/scala/com/gu/mediaservice/lib/auth/KeyStore.scala
+++ b/common-lib/src/main/scala/com/gu/mediaservice/lib/auth/KeyStore.scala
@@ -1,15 +1,15 @@
 package com.gu.mediaservice.lib.auth
 
 import com.gu.mediaservice.lib.BaseStore
+import com.gu.mediaservice.lib.aws.{S3, S3Bucket}
 import com.gu.mediaservice.lib.config.CommonConfig
 import com.gu.mediaservice.model.Instance
-import software.amazon.awssdk.services.s3.model.ListObjectsV2Request
 
-import scala.jdk.CollectionConverters._
-import scala.concurrent.ExecutionContext
+import scala.concurrent.duration._
+import scala.concurrent.{Await, ExecutionContext}
 
-class KeyStore(bucket: String, config: CommonConfig)(implicit ec: ExecutionContext)
-  extends BaseStore[String, ApiAccessor](bucket, config)(ec) {
+class KeyStore(bucket: S3Bucket, config: CommonConfig, s3: S3)(implicit ec: ExecutionContext)
+  extends BaseStore[String, ApiAccessor](bucket, config, s3)(ec) {
 
   def lookupIdentity(key: String)(implicit instance: Instance): Option[ApiAccessor] = store.get().get(instance.id + "/" + key)
 
@@ -20,9 +20,9 @@ class KeyStore(bucket: String, config: CommonConfig)(implicit ec: ExecutionConte
   }
 
   private def fetchAll: Map[String, ApiAccessor] = {
-    val contents = s3.client.listObjectsV2(ListObjectsV2Request.builder().bucket(bucket).build())
-      .contents().asScala.toList
-    val keys = contents.map(_.key())
+    val objects = Await.result(s3.listPaginating(bucket, None), 10.seconds)
+    logger.info(s"fetchAll found ${objects.size} objects")
+    val keys = objects.map( s3Object => bucket.keyFromURL(s3Object.uri))
     keys.flatMap(k => getS3Object(k).map(k -> ApiAccessor(_))).toMap
   }
 }
diff --git a/common-lib/src/main/scala/com/gu/mediaservice/lib/aws/S3.scala b/common-lib/src/main/scala/com/gu/mediaservice/lib/aws/S3.scala
index c8c264a2e5c..a0ec134d539 100644
--- a/common-lib/src/main/scala/com/gu/mediaservice/lib/aws/S3.scala
+++ b/common-lib/src/main/scala/com/gu/mediaservice/lib/aws/S3.scala
@@ -9,28 +9,27 @@ import software.amazon.awssdk.core.sync.RequestBody
 import software.amazon.awssdk.regions.Region
 import software.amazon.awssdk.services.s3.model._
 import software.amazon.awssdk.services.s3.presigner.S3Presigner
-import software.amazon.awssdk.services.s3.presigner.model.GetObjectPresignRequest
+import software.amazon.awssdk.services.s3.presigner.model.{GetObjectPresignRequest, PresignedPutObjectRequest, PutObjectPresignRequest}
 import software.amazon.awssdk.services.s3.{S3Client, S3Configuration}
 
 import java.io.File
 import java.net.{URI, URL}
 import java.nio.charset.StandardCharsets
 import java.time.Duration
+import java.util
+import scala.annotation.tailrec
 import scala.concurrent.{ExecutionContext, Future}
 import scala.jdk.CollectionConverters._
+import scala.util.Try
 
 case class S3Object(uri: URI, size: Long, metadata: S3Metadata)
 
 object S3Object {
-  def objectUrl(bucket: String, key: String): URI = {
-    val bucketUrl = s"$bucket.${S3Ops.s3Endpoint}"
-    new URI("http", bucketUrl, s"/$key", null)
-  }
 
-  def apply(bucket: String, key: String, size: Long, metadata: S3Metadata): S3Object =
-    apply(objectUrl(bucket, key), size, metadata)
+  def apply(bucket: S3Bucket, key: String, size: Long, metadata: S3Metadata): S3Object =
+    apply(bucket.objectUrl(key), size, metadata)
 
-  def apply(bucket: String, key: String, file: File, mimeType: Option[MimeType], lastModified: Option[DateTime],
+  def apply(bucket: S3Bucket, key: String, file: File, mimeType: Option[MimeType], lastModified: Option[DateTime],
             meta: Map[String, String] = Map.empty, cacheControl: Option[String] = None): S3Object = {
     S3Object(
       bucket,
@@ -48,17 +47,22 @@ object S3Object {
   }
 }
 
-case class S3Metadata(userMetadata: Map[String, String], objectMetadata: S3ObjectMetadata)
+case class S3Metadata(userMetadata: Map[String, String], objectMetadata: S3ObjectMetadata, objectVersion: Option[String] = None)
 
 object S3Metadata {
   def apply(meta: HeadObjectResponse): S3Metadata = {
+    val maybeMineType = Try {
+      Option(meta.contentType()).filterNot(_.toLowerCase == "application/octet-stream").map(MimeType.apply)
+    }.toOption.flatten
+
     S3Metadata(
       meta.metadata().asScala.toMap,
       S3ObjectMetadata(
-        contentType = Option(meta.contentType()).filterNot(_.toLowerCase == "application/octet-stream").map(MimeType.apply),
+        contentType = maybeMineType,
         cacheControl = Option(meta.cacheControl()),
         lastModified = Option(meta.lastModified()).map(l => new DateTime(l.toEpochMilli).withZone(DateTimeZone.UTC))
-      )
+      ),
+      objectVersion = Option(meta.versionId())
     )
   }
 }
@@ -66,22 +70,16 @@ object S3Metadata {
 case class S3ObjectMetadata(contentType: Option[MimeType], cacheControl: Option[String], lastModified: Option[DateTime])
 
 class S3(config: CommonConfig) extends GridLogging with ContentDisposition with RoundedExpiration {
-  type Bucket = String
   type Key = String
   type UserMetadata = Map[String, String]
 
-  lazy val client: S3Client = S3Ops.buildS3Client(config)
-  lazy val presigner = S3Ops.buildPresignerClientV2(config)
   def signUrl(
-                 bucket: Bucket,
-                 url: URI,
+                 bucket: S3Bucket,
+                 key: Key,
                  image: Image,
                  expiration: DateTime = cachableExpiration(),
                  imageType: ImageFileType = Source
                ): String = {
-    // Fix key extraction (use stripPrefix to avoid corrupting relative paths)
-    val key: Key = url.getPath.stripPrefix("/")
-
     val contentDisposition = getContentDisposition(image, imageType, config.shortenDownloadFilename)
 
     val nowMillis = System.currentTimeMillis()
@@ -89,7 +87,7 @@ class S3(config: CommonConfig) extends GridLogging with ContentDisposition with
     val remainingSeconds = Math.max(1, (targetExpirationMillis - nowMillis) / 1000)
 
     val getObjectRequest = GetObjectRequest.builder()
-      .bucket(bucket)
+      .bucket(bucket.name)
       .key(key)
       .responseContentDisposition(contentDisposition)
       .build()
@@ -99,20 +97,17 @@ class S3(config: CommonConfig) extends GridLogging with ContentDisposition with
       .signatureDuration(Duration.ofSeconds(remainingSeconds))
       .build()
 
-    val req = presigner.presignGetObject(getObjectPresignRequest)
+    val req = bucket.presigner.presignGetObject(getObjectPresignRequest)
     req.url().toExternalForm
   }
 
-  def signUrlTony(bucket: Bucket, url: URI, expiration: DateTime = cachableExpiration()): URL = {
-    // get path and remove leading `/`
-    val key: Key = url.getPath.drop(1)
-
+  def signUrlTony(bucket: S3Bucket, key: Key, expiration: DateTime = cachableExpiration()): URL = {
     val nowMillis = System.currentTimeMillis()
     val targetExpirationMillis = expiration.getMillis
     val remainingSeconds = Math.max(1, (targetExpirationMillis - nowMillis) / 1000)
 
     val getObjectRequest = GetObjectRequest.builder()
-      .bucket(bucket)
+      .bucket(bucket.name)
       .key(key)
       .build()
 
@@ -121,104 +116,124 @@ class S3(config: CommonConfig) extends GridLogging with ContentDisposition with
       .signatureDuration(Duration.ofSeconds(remainingSeconds))
       .build()
 
-    val req = presigner.presignGetObject(getObjectPresignRequest)
+    val req = bucket.presigner.presignGetObject(getObjectPresignRequest)
     req.url()
   }
 
-  def getObject(bucket: Bucket, url: URI): ResponseInputStream[GetObjectResponse]= {
-    // get path and remove leading `/`
-    val key: Key = url.getPath.drop(1)
-    client.getObject(GetObjectRequest.builder().key(key).bucket(bucket).build())
+  def presignPutObject(bucket: S3Bucket, putObjectPresignRequest: PutObjectPresignRequest): PresignedPutObjectRequest = {
+    bucket.presigner.presignPutObject(putObjectPresignRequest)
   }
 
-  def getObject(bucket: Bucket, key: String): ResponseInputStream[GetObjectResponse] = {
-    client.getObject(GetObjectRequest.builder().key(key).bucket(bucket).build())
+  def getObject(bucket: S3Bucket, key: String): ResponseInputStream[GetObjectResponse] = {
+    bucket.client.getObject(GetObjectRequest.builder().key(key).bucket(bucket.name).build())
   }
 
-  def getObjectAsString(bucket: Bucket, key: String): Option[String] = {
+  def getObjectAsString(bucket: S3Bucket, key: String): Option[String] = {
     try {
-      val stream = client.getObject(GetObjectRequest.builder().key(key).bucket(bucket).build());
+      val stream = bucket.client.getObject(GetObjectRequest.builder().key(key).bucket(bucket.name).build());
       Some(new String(stream.readAllBytes(), StandardCharsets.UTF_8))
     } catch {
       case e: NoSuchKeyException =>
-        logger.warn(s"Cannot find key: $key in bucket: $bucket")
+        logger.warn(s"Cannot find key: $key in bucket: ${bucket.name}")
         None
     }
   }
 
-  def putString(bucket: String, key: String, fileContents: String) = {
-    client.putObject(PutObjectRequest.builder().bucket(bucket).key(key).build(), RequestBody.fromString(fileContents))
+  def putString(bucket: S3Bucket, key: String, fileContents: String) = {
+    bucket.client.putObject(PutObjectRequest.builder().bucket(bucket.name).key(key).build(), RequestBody.fromString(fileContents))
   }
 
-  def store(bucket: Bucket, id: Key, file: File, mimeType: Option[MimeType], meta: UserMetadata = Map.empty, cacheControl: Option[String] = None)
+  def store(bucket: S3Bucket, id: Key, file: File, mimeType: Option[MimeType], meta: UserMetadata = Map.empty, cacheControl: Option[String] = None)
            (implicit ex: ExecutionContext, logMarker: LogMarker): Future[S3Object] =
     Future {
 
       val fileMarkers = Map(
-        "bucket" -> bucket,
+        "bucket" -> bucket.name,
       )
       val markers = logMarker ++ fileMarkers
 
-      val reqBuilder = PutObjectRequest.builder().key(id).bucket(bucket)
+      val reqBuilder = PutObjectRequest.builder().key(id).bucket(bucket.name)
       cacheControl.foreach(c => reqBuilder.cacheControl(c))
       mimeType.foreach(m => reqBuilder.contentType(m.name))
       reqBuilder.metadata(meta.asJava)
       val req = reqBuilder.build()
 
       Stopwatch(s"S3 client.putObject ($req)"){
-        client.putObject(req, RequestBody.fromFile(file))
+        bucket.client.putObject(req, RequestBody.fromFile(file))
         // once we've completed the PUT read back to ensure that we are returning reality
-        val metadata = client.headObject(
-          HeadObjectRequest.builder().key(id).bucket(bucket).build()
+        val metadata = bucket.client.headObject(
+          HeadObjectRequest.builder().key(id).bucket(bucket.name).build()
         )
 
         S3Object(bucket, id, metadata.contentLength(), S3Metadata(metadata))
       }(markers)
     }
 
-  def storeIfNotPresent(bucket: Bucket, id: Key, file: File, mimeType: Option[MimeType], meta: UserMetadata = Map.empty, cacheControl: Option[String] = None)
+  def storeIfNotPresent(bucket: S3Bucket, id: Key, file: File, mimeType: Option[MimeType], meta: UserMetadata = Map.empty, cacheControl: Option[String] = None)
                        (implicit ex: ExecutionContext, logMarker: LogMarker): Future[S3Object] = {
     Future {
-      Some(client.headObject(
-        HeadObjectRequest.builder().key(id).bucket(bucket).build()
+      Some(bucket.client.headObject(
+        HeadObjectRequest.builder().key(id).bucket(bucket.name).build()
       ))
     }.recover {
       // translate this exception into the object not existing
       case _: NoSuchKeyException => None
     }.flatMap {
       case Some(metadata) =>
-        logger.info(logMarker, s"Skipping storing of S3 file $id as key is already present in bucket $bucket")
+        logger.info(logMarker, s"Skipping storing of S3 file $id as key is already present in bucket ${bucket.name}")
         Future.successful(S3Object(bucket, id, metadata.contentLength(), S3Metadata(metadata)))
       case None =>
         store(bucket, id, file, mimeType, meta, cacheControl)
     }
   }
 
-  def list(bucket: Bucket, prefixDir: String)
+  def list(bucket: S3Bucket, prefixDir: String)
           (implicit ex: ExecutionContext): Future[List[S3Object]] =
     Future {
-      val req = ListObjectsV2Request.builder().bucket(bucket).prefix(s"$prefixDir/").build()
-      val listing = client.listObjectsV2(req)
+      val req = ListObjectsV2Request.builder().bucket(bucket.name).prefix(s"$prefixDir/").build()
+      val listing = bucket.client.listObjectsV2(req)
       val s3Objects = listing.contents().asScala.toList
       s3Objects.map(s3Object => {
         S3Object(bucket, s3Object.key(), size = s3Object.size(), metadata = getMetadata(bucket, s3Object.key()))
       })
     }
 
-  def getMetadata(bucket: Bucket, key: Key): S3Metadata = {
-    val meta = client.headObject(HeadObjectRequest.builder().key(key).bucket(bucket).build())
+  def listPaginating(bucket: S3Bucket, prefixDir: Option[String])
+                    (implicit ex: ExecutionContext): Future[List[S3Object]] =
+    Future {
+      @tailrec
+      def pageThrough(continuationToken: Option[String], accumulated: List[S3Object]): List[S3Object] = {
+        val reqBuilder = prefixDir.map { prefix =>
+          ListObjectsV2Request.builder().bucket(bucket.name).prefix(s"$prefix/")
+        }.getOrElse{
+          ListObjectsV2Request.builder().bucket(bucket.name)
+        }
+        continuationToken.foreach(reqBuilder.continuationToken)
+        val listing = bucket.client.listObjectsV2(reqBuilder.build())
+        val s3Objects = listing.contents().asScala.toList.map(s3Object =>
+          S3Object(bucket, s3Object.key(), size = s3Object.size(), metadata = getMetadata(bucket, s3Object.key()))
+        )
+        val all = accumulated ++ s3Objects
+        if (listing.isTruncated) pageThrough(Some(listing.nextContinuationToken()), all) else all
+      }
+
+      pageThrough(None, Nil)
+    }
+
+  def getMetadata(bucket: S3Bucket, key: Key): S3Metadata = {
+    val meta = bucket.client.headObject(HeadObjectRequest.builder().key(key).bucket(bucket.name).build())
     S3Metadata(meta)
   }
 
-  def syncFindKey(bucket: Bucket, prefixName: String): Option[Key] = {
-    val req = ListObjectsV2Request.builder().bucket(bucket).prefix(s"$prefixName-").build()
-    val objects = client.listObjectsV2(req).contents().asScala.toList
+  def syncFindKey(bucket: S3Bucket, prefixName: String): Option[Key] = {
+    val req = ListObjectsV2Request.builder().bucket(bucket.name).prefix(s"$prefixName-").build()
+    val objects = bucket.client.listObjectsV2(req).contents().asScala.toList
     objects.headOption.map(_.key())
   }
-  def doesObjectExist(bucket: Bucket, key: String) = {
+  def doesObjectExist(bucket: S3Bucket, key: String) = {
     try {
-      client.headObject(
-        HeadObjectRequest.builder().key(key).bucket(bucket).build()
+      bucket.client.headObject(
+        HeadObjectRequest.builder().key(key).bucket(bucket.name).build()
       )
       true
     } catch {
@@ -226,35 +241,79 @@ class S3(config: CommonConfig) extends GridLogging with ContentDisposition with
     }
   }
 
+  def deleteObject(bucket: S3Bucket, key: String): Unit =
+    bucket.client.deleteObject(DeleteObjectRequest.builder().bucket(bucket.name).key(key).build())
+
+  def deleteObjects(bucket: S3Bucket, keys: List[String]): Map[String, Boolean] = {
+    val objects: util.List[ObjectIdentifier] = keys.map { key =>
+      ObjectIdentifier.builder()
+        .key(key)
+        .build()
+    }.asJava
+    val response = bucket.client.deleteObjects(
+      DeleteObjectsRequest.builder().bucket(bucket.name)
+        .delete(Delete.builder().objects(objects).build())
+        .build()
+    )
+    val errorKeys = response.errors().asScala.toList.map(_.key())
+    keys.map { key =>
+      key -> !errorKeys.contains(key)
+    }.toMap
+  }
+
+  def deleteVersion(bucket: S3Bucket, key: String, objectVersion: String): Unit =
+    bucket.client.deleteObject(DeleteObjectRequest.builder().bucket(bucket.name).key(key).versionId(objectVersion).build())
+
+  def copy(key: String, sourceBucket: S3Bucket, destinationBucket: S3Bucket): CopyObjectResponse = {
+    sourceBucket.client.copyObject(
+      CopyObjectRequest.builder()
+        .sourceBucket(sourceBucket.name)
+        .sourceKey(key)
+        .destinationBucket(destinationBucket.name)
+        .destinationKey(key)
+        .build()
+    )
+  }
+
 }
 
-object S3Ops {
+object S3Ops extends GridLogging {
   // TODO make this localstack friendly
   // TODO: Make this region aware - i.e. RegionUtils.getRegion(region).getServiceEndpoint(AmazonS3.ENDPOINT_PREFIX)
   val s3Endpoint = "s3.amazonaws.com"
 
-  def buildS3Client(config: CommonConfig, localstackAware: Boolean = true, maybeRegionOverride: Option[Region] = None): S3Client = {
-    val builder = config.awsLocalEndpoint match {
-      case Some(_) if config.isDev =>
-        S3Client.builder().forcePathStyle(true)
-      case _ => S3Client.builder()
+  def buildS3Client(config: CommonConfig, endpointOverride: Option[String] = None, usesPathStyleURLs: Boolean = false, maybeRegionOverride: Option[Region] = None): S3Client = {
+    val builder = S3Client.builder()
+      .credentialsProvider(config.awsCredentials)
+      .region(maybeRegionOverride.getOrElse(config.awsRegion))
+      .forcePathStyle(usesPathStyleURLs)
+
+    val withEndpoint = endpointOverride match {
+      case Some(endpoint) =>
+        logger.info(s"creating S3 client with endpoint override: $endpoint")
+        builder.endpointOverride(new URI(endpoint))
+      case _ => builder
     }
 
-    config.withAWSCredentials(builder, localstackAware, maybeRegionOverride).build()
+    withEndpoint.build()
   }
 
-  def buildPresignerClientV2(config: CommonConfig, localstackAware: Boolean = true, maybeRegionOverride: Option[Region] = None): S3Presigner = {
+  def buildPresignerClientV2(config: CommonConfig, endpointOverride: Option[String] = None, usesPathStyleURLs: Boolean = false, maybeRegionOverride: Option[Region] = None): S3Presigner = {
     val builder = S3Presigner.builder()
       .credentialsProvider(config.awsCredentials)
-      .region(config.awsRegion)
-
-    config.awsLocalEndpointUri match {
-      case Some(endpoint) if config.isDev => builder.endpointOverride(endpoint)
-        .serviceConfiguration(S3Configuration.builder().pathStyleAccessEnabled(true).build()).build()
-      case _ => builder.build()
-
+      .region(maybeRegionOverride.getOrElse(config.awsRegion))
+      .serviceConfiguration(S3Configuration.builder()
+        .pathStyleAccessEnabled(usesPathStyleURLs)
+        .build())
+
+    val withEndpoint = endpointOverride match {
+      case Some(endpoint) =>
+        logger.info(s"creating S3 presigner with endpoint override: $endpoint")
+        builder.endpointOverride(new URI(endpoint))
+      case _ => builder
     }
 
+    withEndpoint.build()
   }
 
 }
diff --git a/common-lib/src/main/scala/com/gu/mediaservice/lib/aws/S3Bucket.scala b/common-lib/src/main/scala/com/gu/mediaservice/lib/aws/S3Bucket.scala
new file mode 100644
index 00000000000..e3d82b53d55
--- /dev/null
+++ b/common-lib/src/main/scala/com/gu/mediaservice/lib/aws/S3Bucket.scala
@@ -0,0 +1,55 @@
+package com.gu.mediaservice.lib.aws
+
+import com.gu.mediaservice.lib.config.CommonConfig
+import software.amazon.awssdk.regions.Region
+import software.amazon.awssdk.services.s3.S3Client
+import software.amazon.awssdk.services.s3.presigner.S3Presigner
+import java.net.URI
+
+case class S3Bucket(name: String, endPoint: URI, usesPathStyleURLs: Boolean, client: S3Client, presigner: S3Presigner) {
+
+  def objectUrl(key: String): URI = {
+    val bucketBaseURL = bucketURL()
+    new URI(bucketBaseURL.getScheme, bucketBaseURL.getHost, bucketBaseURL.getPath + key, null)
+  }
+
+  def keyFromURL(url: URI): String = {
+    if (usesPathStyleURLs) {
+      url.getPath.drop(name.length + 2)
+    } else {
+      // get path and remove leading `/`
+      url.getPath.drop(1)
+    }
+  }
+
+  def bucketURL(): URI = {
+    if (usesPathStyleURLs) {
+      new URI(endPoint.getScheme, endPoint.getHost, s"/$name/", null)
+    } else {
+      new URI(endPoint.getScheme, s"$name.${endPoint.getHost}", "/", null)
+    }
+  }
+
+}
+
+object S3Bucket {
+
+  /**
+   * Build a bucket that talks to the endpoint implied by the current environment - i.e. the localstack
+   * endpoint (with path style URLs) when running in DEV, otherwise the real AWS S3 endpoint.
+   */
+  def apply(name: String, config: CommonConfig): S3Bucket = {
+    val endpointOverride = config.awsLocalEndpoint
+    val usesPathStyleURLs = endpointOverride.isDefined
+    apply(name, config, endpointOverride, usesPathStyleURLs, None)
+  }
+
+  def apply(name: String, config: CommonConfig, endpointOverride: Option[String], usesPathStyleURLs: Boolean, maybeRegionOverride: Option[Region]): S3Bucket =
+    S3Bucket(
+      name = name,
+      endPoint = new URI(endpointOverride.getOrElse(S3Ops.s3Endpoint)),
+      usesPathStyleURLs = usesPathStyleURLs,
+      client = S3Ops.buildS3Client(config, endpointOverride, usesPathStyleURLs, maybeRegionOverride),
+      presigner = S3Ops.buildPresignerClientV2(config, endpointOverride, usesPathStyleURLs, maybeRegionOverride)
+    )
+}
diff --git a/common-lib/src/main/scala/com/gu/mediaservice/lib/config/CommonConfig.scala b/common-lib/src/main/scala/com/gu/mediaservice/lib/config/CommonConfig.scala
index 70167400d49..bf26b0b039a 100644
--- a/common-lib/src/main/scala/com/gu/mediaservice/lib/config/CommonConfig.scala
+++ b/common-lib/src/main/scala/com/gu/mediaservice/lib/config/CommonConfig.scala
@@ -1,6 +1,6 @@
 package com.gu.mediaservice.lib.config
 
-import com.gu.mediaservice.lib.aws.{AwsClientBuilderUtils, KinesisSenderConfig}
+import com.gu.mediaservice.lib.aws.{AwsClientBuilderUtils, KinesisSenderConfig, S3Bucket}
 import com.gu.mediaservice.model.UsageRightsSpec
 import com.typesafe.config.Config
 import com.typesafe.scalalogging.StrictLogging
@@ -60,13 +60,17 @@ abstract class CommonConfig(resources: GridConfigResources) extends AwsClientBui
 
   lazy val softDeletedMetadataTable: String = string("dynamo.table.softDelete.metadata")
 
+  val imageBucket: S3Bucket = S3Bucket(string("s3.image.bucket"), this)
+  val thumbnailBucket: S3Bucket = S3Bucket(string("s3.thumb.bucket"), this)
+  val imgPublishingBucket: S3Bucket = S3Bucket(string("publishing.image.bucket"), this)
+
   val maybeIngestSqsQueueUrl: Option[String] = stringOpt("sqs.ingest.queue.url")
-  val maybeIngestBucket: Option[String] = stringOpt("s3.ingest.bucket")
-  val maybeFailBucket: Option[String] = stringOpt("s3.fail.bucket")
+  val maybeIngestBucket: Option[S3Bucket] = stringOpt("s3.ingest.bucket").map(S3Bucket(_, this))
+  val maybeFailBucket: Option[S3Bucket] = stringOpt("s3.fail.bucket").map(S3Bucket(_, this))
 
-  val maybeQuarantineBucket: Option[String] = stringOpt("s3.quarantine.bucket")
+  val maybeQuarantineBucket: Option[S3Bucket] = stringOpt("s3.quarantine.bucket").map(S3Bucket(_, this))
 
-  val maybeBucketForUIUploads: Option[String] = maybeQuarantineBucket orElse maybeIngestBucket
+  val maybeBucketForUIUploads: Option[S3Bucket] = maybeQuarantineBucket orElse maybeIngestBucket
 
   val maybeUploadLimitInBytes: Option[Int] = intOpt("upload.limit.mb").map(_ * 1024 * 1024)
 
diff --git a/common-lib/src/test/resources/application.conf b/common-lib/src/test/resources/application.conf
index 05d1db98e6b..92e1d78ac76 100644
--- a/common-lib/src/test/resources/application.conf
+++ b/common-lib/src/test/resources/application.conf
@@ -3,6 +3,9 @@ grid.appName: "test"
 thrall.kinesis.stream.name: "not-used"
 thrall.kinesis.lowPriorityStream.name: "not-used"
 domain.root: "notused.example.com"
+s3.image.bucket: "not-used"
+s3.thumb.bucket: "not-used"
+publishing.image.bucket: "not-used"
 
 image.processors = [
   "com.gu.mediaservice.lib.cleanup.GuardianMetadataCleaners",
diff --git a/cropper/app/CropperComponents.scala b/cropper/app/CropperComponents.scala
index b05381cb06d..85a150fca87 100644
--- a/cropper/app/CropperComponents.scala
+++ b/cropper/app/CropperComponents.scala
@@ -1,4 +1,5 @@
 import com.gu.mediaservice.GridClient
+import com.gu.mediaservice.lib.aws.S3
 import com.gu.mediaservice.lib.imaging.ImageOperations
 import com.gu.mediaservice.lib.management.Management
 import com.gu.mediaservice.lib.play.GridComponents
@@ -13,7 +14,9 @@ class CropperComponents(context: Context) extends GridComponents(context, new Cr
   val store = new CropStore(config)
   val imageOperations = new ImageOperations(context.environment.rootPath.getAbsolutePath)
 
-  val crops = new Crops(config, store, imageOperations, config.imageBucket)
+  private val s3 = new S3(config)
+
+  val crops = new Crops(config, store, imageOperations, config.imageBucket, s3)
   val notifications = new Notifications(config)
 
   private val gridClient = GridClient(config.services, config.services.cropperBaseUri)(wsClient)
diff --git a/cropper/app/lib/CropStore.scala b/cropper/app/lib/CropStore.scala
index 749764337e7..a7a99c02607 100644
--- a/cropper/app/lib/CropStore.scala
+++ b/cropper/app/lib/CropStore.scala
@@ -78,12 +78,13 @@ class CropStore(config: CropperConfig) extends S3ImageStorage(config) with CropS
   def translateImgHost(uri: URI): URI =
     new URI("https", config.imgPublishingHost, uri.getPath, uri.getFragment)
 
-  private def folderForImagesCrops(id: Bucket, instance: Instance) = {
+  private def folderForImagesCrops(id: String, instance: Instance) = {
     instance.id + "/" + id
   }
 
   private def signedCropAssetUrl(uri: URI): URI = {
-    signUrlTony(config.imgPublishingBucket, uri).toURI
+    val key = config.imgPublishingBucket.keyFromURL(uri)
+    signUrlTony(config.imgPublishingBucket, key).toURI
   }
 
 }
diff --git a/cropper/app/lib/CropperConfig.scala b/cropper/app/lib/CropperConfig.scala
index 96d043d60a7..176fdeed941 100644
--- a/cropper/app/lib/CropperConfig.scala
+++ b/cropper/app/lib/CropperConfig.scala
@@ -7,10 +7,6 @@ import java.io.File
 
 
 class CropperConfig(resources: GridConfigResources) extends CommonConfig(resources) {
-  val imageBucket: String = string("s3.image.bucket")
-
-  val imgPublishingBucket = string("publishing.image.bucket")
-
   val canDownloadCrop: Boolean = boolean("canDownloadCrop")
 
   val imgPublishingHost = string("publishing.image.host")
diff --git a/cropper/app/lib/Crops.scala b/cropper/app/lib/Crops.scala
index 6c2be74d21c..32b848b02cc 100644
--- a/cropper/app/lib/Crops.scala
+++ b/cropper/app/lib/Crops.scala
@@ -3,7 +3,7 @@ package lib
 import java.io.File
 import com.gu.mediaservice.lib.metadata.FileMetadataHelper
 import com.gu.mediaservice.lib.Files
-import com.gu.mediaservice.lib.aws.S3
+import com.gu.mediaservice.lib.aws.{S3, S3Bucket}
 import com.gu.mediaservice.lib.imaging.{ExportResult, ImageOperations}
 import com.gu.mediaservice.lib.logging.{GridLogging, LogMarker, Stopwatch}
 import com.gu.mediaservice.model._
@@ -17,7 +17,7 @@ case object InvalidCropRequest extends Exception("Crop request invalid for image
 
 case class MasterCrop(sizing: Future[Asset], file: File, dimensions: Dimensions, aspectRatio: Float)
 
-class Crops(config: CropperConfig, store: CropStore, imageOperations: ImageOperations, imageBucket: String)(implicit ec: ExecutionContext) extends GridLogging {
+class Crops(config: CropperConfig, store: CropStore, imageOperations: ImageOperations, imageBucket: S3Bucket, s3: S3)(implicit ec: ExecutionContext) extends GridLogging {
   import Files._
 
   private val cropQuality = 75d
@@ -26,8 +26,6 @@ class Crops(config: CropperConfig, store: CropStore, imageOperations: ImageOpera
   // We don't overly care about output crop file sizes here, but prefer a fast output, so turn it right down.
   private val pngCropQuality = 1d
 
-  private val s3 = new S3(config)
-
   def outputFilename(source: SourceImage, bounds: Bounds, outputWidth: Int, fileType: MimeType, isMaster: Boolean = false)(implicit instance: Instance): String = {
     val masterString: String = if (isMaster) "master/" else ""
     instance.id + "/" + s"${source.id}/${Crop.getCropId(bounds)}/$masterString$outputWidth${fileType.fileExtension}"
@@ -116,7 +114,8 @@ class Crops(config: CropperConfig, store: CropStore, imageOperations: ImageOpera
     val hasAlpha = apiImage.fileMetadata.colourModelInformation.get("hasAlpha").flatMap(a => Try(a.toBoolean).toOption).getOrElse(true)
     val cropType = Crops.cropType(mimeType, colourType, hasAlpha)
 
-    val secureUrl = s3.signUrlTony(imageBucket, secureFile)
+    val key = imageBucket.keyFromURL(secureFile)
+    val secureUrl = s3.signUrlTony(imageBucket, key)
 
     Stopwatch.async(s"making crop assets for ${apiImage.id} ${Crop.getCropId(source.bounds)}") {
       for {
diff --git a/cropper/test/lib/CropsTest.scala b/cropper/test/lib/CropsTest.scala
index e348b9a4a6e..b852f9c948a 100644
--- a/cropper/test/lib/CropsTest.scala
+++ b/cropper/test/lib/CropsTest.scala
@@ -1,11 +1,14 @@
 package lib
 
+import com.gu.mediaservice.lib.aws.{S3, S3Bucket}
 import com.gu.mediaservice.lib.imaging.ImageOperations
 import com.gu.mediaservice.model._
 import org.scalatest.funspec.AnyFunSpec
 import org.scalatest.matchers.should.Matchers
 import org.scalatestplus.mockito.MockitoSugar
 
+import java.net.URI
+
 class CropsTest extends AnyFunSpec with Matchers with MockitoSugar {
   import scala.concurrent.ExecutionContext.Implicits.global
 
@@ -50,25 +53,26 @@ class CropsTest extends AnyFunSpec with Matchers with MockitoSugar {
   private val source: SourceImage = SourceImage("test", mock[Asset], valid = true, mock[ImageMetadata], mock[FileMetadata])
   private val bounds: Bounds = Bounds(10, 20, 30, 40)
   private val outputWidth = 1234
-  private val imageBucket = "crops-bucket"
+  private val imageBucket = S3Bucket("crops-bucket", new URI("https://s3.amazonaws.com"), usesPathStyleURLs = false, client = null, presigner = null)
+  private val s3 = new S3(config)
 
   it("should should construct a correct address for a master jpg") {
-    val outputFilename = new Crops(config, store, imageOperations, imageBucket)
+    val outputFilename = new Crops(config, store, imageOperations, imageBucket, s3)
       .outputFilename(source, bounds, outputWidth, Jpeg, isMaster = true)
     outputFilename shouldBe "an-instance/test/10_20_30_40/master/1234.jpg"
   }
   it("should should construct a correct address for a non-master jpg") {
-    val outputFilename = new Crops(config, store, imageOperations, imageBucket)
+    val outputFilename = new Crops(config, store, imageOperations, imageBucket, s3)
       .outputFilename(source, bounds, outputWidth, Jpeg)
     outputFilename shouldBe "an-instance/test/10_20_30_40/1234.jpg"
   }
   it("should should construct a correct address for a non-master tiff") {
-    val outputFilename = new Crops(config, store, imageOperations, imageBucket)
+    val outputFilename = new Crops(config, store, imageOperations, imageBucket, s3)
       .outputFilename(source, bounds, outputWidth, Tiff)
     outputFilename shouldBe "an-instance/test/10_20_30_40/1234.tiff"
   }
   it("should should construct a correct address for a non-master png") {
-    val outputFilename = new Crops(config, store, imageOperations, imageBucket)
+    val outputFilename = new Crops(config, store, imageOperations, imageBucket, s3)
       .outputFilename(source, bounds, outputWidth, Png)
     outputFilename shouldBe "an-instance/test/10_20_30_40/1234.png"
   }
diff --git a/image-loader/app/ImageLoaderComponents.scala b/image-loader/app/ImageLoaderComponents.scala
index 9d4a739b263..c6529ecbb15 100644
--- a/image-loader/app/ImageLoaderComponents.scala
+++ b/image-loader/app/ImageLoaderComponents.scala
@@ -1,5 +1,5 @@
 import com.gu.mediaservice.GridClient
-import com.gu.mediaservice.lib.aws.{Bedrock, Embedder, S3Vectors, SimpleSqsMessageConsumer}
+import com.gu.mediaservice.lib.aws.{Bedrock, Embedder, S3, S3Vectors, SimpleSqsMessageConsumer}
 import com.gu.mediaservice.lib.imaging.ImageOperations
 import com.gu.mediaservice.lib.logging.GridLogging
 import com.gu.mediaservice.lib.play.GridComponents
@@ -34,8 +34,10 @@ class ImageLoaderComponents(context: Context) extends GridComponents(context, ne
       new Embedder(new Bedrock(config), new SimpleSqsMessageConsumer(queueUrl, config))
     }
 
+  private val s3 = new S3(config)
+
   val uploader = new Uploader(store, config, imageOperations, notifications, maybeEmbedder, imageProcessor, gridClient, auth)
-  val projector = Projector(config, imageOperations, imageProcessor, auth, maybeEmbedder)
+  val projector = Projector(config, s3, imageOperations, imageProcessor, auth, maybeEmbedder)
   val quarantineUploader: Option[QuarantineUploader] = config.maybeQuarantineBucket.map(_ =>
     new QuarantineUploader(new QuarantineStore(config), config)
   )
diff --git a/image-loader/app/controllers/ImageLoaderController.scala b/image-loader/app/controllers/ImageLoaderController.scala
index 2c85b621cba..be3eda29050 100644
--- a/image-loader/app/controllers/ImageLoaderController.scala
+++ b/image-loader/app/controllers/ImageLoaderController.scala
@@ -1,9 +1,5 @@
 package controllers
 
-import org.apache.pekko.Done
-import org.apache.pekko.stream.Materializer
-import org.apache.pekko.stream.scaladsl.Source
-import software.amazon.awssdk.services.sqs.model.{Message => SQSMessage}
 import com.drew.imaging.ImageProcessingException
 import com.gu.mediaservice.GridClient
 import com.gu.mediaservice.lib.ImageIngestOperations.fileKeyFromId
@@ -12,7 +8,7 @@ import com.gu.mediaservice.lib.argo.model.Link
 import com.gu.mediaservice.lib.auth.Authentication.{MachinePrincipal, OnBehalfOfPrincipal, UserPrincipal}
 import com.gu.mediaservice.lib.auth._
 import com.gu.mediaservice.lib.auth.provider.ApiKeyAuthenticationProvider
-import com.gu.mediaservice.lib.aws.{S3Ops, SimpleSqsMessageConsumer, SqsHelpers}
+import com.gu.mediaservice.lib.aws.{S3Bucket, SimpleSqsMessageConsumer, SqsHelpers}
 import com.gu.mediaservice.lib.config.InstanceForRequest
 import com.gu.mediaservice.lib.events.UsageEvents
 import com.gu.mediaservice.lib.formatting.printDateTime
@@ -38,9 +34,7 @@ import play.api.inject.ApplicationLifecycle
 import play.api.libs.json.Json
 import play.api.libs.ws.WSClient
 import play.api.mvc._
-import software.amazon.awssdk.regions.Region
 import software.amazon.awssdk.services.cloudwatch.model.Dimension
-import software.amazon.awssdk.services.s3.S3Client
 import software.amazon.awssdk.services.s3.model.{GetObjectRequest, HeadObjectRequest, NoSuchKeyException}
 import software.amazon.awssdk.services.sqs.model.{Message => SQSMessage}
 
@@ -460,7 +454,7 @@ class ImageLoaderController(auth: Authentication,
           logger.info(context, "image found")
           Ok(Json.toJson(img)).as(ArgoMediaType)
         case None =>
-          val s3Path = "s3://" + config.imageBucket + "/" + ImageIngestOperations.fileKeyFromId(imageId)
+          val s3Path = "s3://" + config.imageBucket.name + "/" + ImageIngestOperations.fileKeyFromId(imageId)
           logger.info(context, "image not found")
           respondError(NotFound, "image-not-found", s"Could not find image: $imageId in s3 at $s3Path")
       } recover {
@@ -637,10 +631,10 @@ class ImageLoaderController(auth: Authentication,
       }
   }
 
-  lazy val replicaS3: S3Client = S3Ops.buildS3Client(config, maybeRegionOverride = Some(Region.US_WEST_1))
-  def doesObjectExist(bucket: String, key: String) = {
+  // TODO is this a duplicate with S3?
+  def doesObjectExist(bucket: S3Bucket, key: String) = {
     try {
-      replicaS3.headObject(HeadObjectRequest.builder().bucket(bucket).key(key).build())
+      bucket.client.headObject(HeadObjectRequest.builder().bucket(bucket.name).key(key).build())
       true
     } catch {
       case _: NoSuchKeyException => false
@@ -674,8 +668,8 @@ class ImageLoaderController(auth: Authentication,
 
           logger.info(logMarker, s"Restoring image $imageId from replica bucket $replicaBucket (key: $s3Key)")
 
-          val replicaObject = replicaS3.getObject(
-            GetObjectRequest.builder().bucket(replicaBucket).key(s3Key).build()
+          val replicaObject = replicaBucket.client.getObject(
+            GetObjectRequest.builder().bucket(replicaBucket.name).key(s3Key).build()
           )
           val lastModified = replicaObject.response().lastModified()
           val metaMap = replicaObject.response().metadata().asScala.toMap
diff --git a/image-loader/app/lib/ImageLoaderConfig.scala b/image-loader/app/lib/ImageLoaderConfig.scala
index 37d43aa0781..c6436af9ad9 100644
--- a/image-loader/app/lib/ImageLoaderConfig.scala
+++ b/image-loader/app/lib/ImageLoaderConfig.scala
@@ -1,20 +1,21 @@
 package lib
 
+import com.gu.mediaservice.lib.aws.{S3Bucket, S3Ops}
+
 import java.io.File
 import com.gu.mediaservice.lib.cleanup.{ComposedImageProcessor, ImageProcessor, ImageProcessorResources}
 import com.gu.mediaservice.lib.config.{CommonConfig, GridConfigResources, ImageProcessorLoader}
 import com.gu.mediaservice.model._
 import com.typesafe.scalalogging.StrictLogging
 import play.api.inject.ApplicationLifecycle
+import software.amazon.awssdk.regions.Region
 
 import scala.concurrent.duration.FiniteDuration
 
 class ImageLoaderConfig(resources: GridConfigResources) extends CommonConfig(resources) with StrictLogging {
-  val imageBucket: String = string("s3.image.bucket")
-
-  val maybeImageReplicaBucket: Option[String] = stringOpt("s3.image.replicaBucket")
-
-  val thumbnailBucket: String = string("s3.thumb.bucket")
+  val maybeImageReplicaBucket: Option[S3Bucket] = stringOpt("s3.image.replicaBucket").map{ replicaBucketName =>
+    S3Bucket.apply(replicaBucketName, this, None, usesPathStyleURLs = false, maybeRegionOverride = Some(Region.US_WEST_1))
+  }
 
   val lowerEnvironmentSamplingPercentageAsDecimal = intOpt("s3.sampling.percentage").getOrElse(1) / 100.0
   val maybeLowerEnvironmentQueueBucketToSampleInto = stringOpt("s3.sampling.targetBucket")
diff --git a/image-loader/app/lib/ImageLoaderStore.scala b/image-loader/app/lib/ImageLoaderStore.scala
index 455a8337164..39fd800e9ce 100644
--- a/image-loader/app/lib/ImageLoaderStore.scala
+++ b/image-loader/app/lib/ImageLoaderStore.scala
@@ -32,8 +32,7 @@ class ImageLoaderStore(config: ImageLoaderConfig) extends lib.ImageIngestOperati
   }
 
   def getS3Object(key: String)(implicit logMarker: LogMarker): ResponseInputStream[GetObjectResponse] = handleNotFound(key) {
-      client.getObject(
-        GetObjectRequest.builder().bucket(config.maybeIngestBucket.get).key(key).build())
+      getObject(config.maybeIngestBucket.get, key)
   } {
     logger.error(logMarker, s"Attempted to read $key from ingest bucket, but it does not exist.")
   }
@@ -49,8 +48,9 @@ class ImageLoaderStore(config: ImageLoaderConfig) extends lib.ImageIngestOperati
   }
   def generatePreSignedUploadUrl(filename: String, duration: Duration, uploadedBy: String, mediaId: String)(implicit instance: Instance): String = {
 
+    val bucket = config.maybeBucketForUIUploads.get
     val putObjectRequest = PutObjectRequest.builder()
-      .bucket(config.maybeBucketForUIUploads.get).key(s"${instance.id}/$uploadedBy/$filename").metadata(Map(
+      .bucket(bucket.name).key(s"${instance.id}/$uploadedBy/$filename").metadata(Map(
         "media-id" -> mediaId).asJava)
       .build()
     val putObjectPresignRequest =
@@ -59,28 +59,21 @@ class ImageLoaderStore(config: ImageLoaderConfig) extends lib.ImageIngestOperati
         .signatureDuration(duration)
         .build();
 
-    val req = presigner.presignPutObject(putObjectPresignRequest)
+    val req = presignPutObject(bucket, putObjectPresignRequest)
     req.url().toExternalForm
   }
 
-  def moveObjectToFailedBucket(key: String)(implicit logMarker: LogMarker) = handleNotFound(key){
-    client.copyObject(
-      CopyObjectRequest.builder()
-        .sourceBucket(config.maybeIngestBucket.get)  // TODO Naked get - make optional
-        .sourceKey(key)
-        .destinationBucket(config.maybeFailBucket.get)   // TODO Naked get - make optional
-        .destinationKey(key)
-        .build()
-    )
+  def moveObjectToFailedBucket(key: String)(implicit logMarker: LogMarker): Unit = handleNotFound(key){
+    val sourceBucket = config.maybeIngestBucket.get  // TODO Naked get - make optional
+    val destinationBucket = config.maybeFailBucket.get // TODO Naked get - make optional
+    copy(key, sourceBucket, destinationBucket)
     deleteObjectFromIngestBucket(key)
   } {
     logger.warn(logMarker, s"Attempted to copy $key from ingest bucket to fail bucket, but it does not exist.")
   }
 
-  def deleteObjectFromIngestBucket(key: String)(implicit logMarker: LogMarker) = handleNotFound(key) {
-    client.deleteObject(
-      DeleteObjectRequest.builder().bucket(config.maybeIngestBucket.get).key(key).build())
-    ()
+  def deleteObjectFromIngestBucket(key: String)(implicit logMarker: LogMarker): Unit = handleNotFound(key) {
+    deleteObject(config.maybeIngestBucket.get, key)
   } {
     logger.warn(logMarker, s"Attempted to delete $key from ingest bucket, but it does not exist.")
   }
diff --git a/image-loader/app/model/Projector.scala b/image-loader/app/model/Projector.scala
index cc469d886bd..217abb0edd8 100644
--- a/image-loader/app/model/Projector.scala
+++ b/image-loader/app/model/Projector.scala
@@ -2,7 +2,7 @@ package model
 
 import com.gu.mediaservice.lib.ImageIngestOperations.{fileKeyFromId, optimisedPngKeyFromId}
 import com.gu.mediaservice.lib.auth.Authentication
-import com.gu.mediaservice.lib.aws.{Embedder, S3}
+import com.gu.mediaservice.lib.aws.{Embedder, S3, S3Bucket}
 import com.gu.mediaservice.lib.cleanup.ImageProcessor
 import com.gu.mediaservice.lib.config.InstanceForRequest
 import com.gu.mediaservice.lib.imaging.ImageOperations
@@ -29,8 +29,8 @@ object Projector {
 
   import Uploader.toImageUploadOpsCfg
 
-  def apply(config: ImageLoaderConfig, imageOps: ImageOperations, processor: ImageProcessor, auth: Authentication, maybeEmbedder: Option[Embedder])(implicit ec: ExecutionContext): Projector
-  = new Projector(toImageUploadOpsCfg(config), new S3(config), imageOps, processor, auth, maybeEmbedder)
+  def apply(config: ImageLoaderConfig, s3: S3, imageOps: ImageOperations, processor: ImageProcessor, auth: Authentication, maybeEmbedder: Option[Embedder])(implicit ec: ExecutionContext): Projector
+  = new Projector(toImageUploadOpsCfg(config), s3, imageOps, processor, auth, maybeEmbedder)
 }
 
 case class S3FileExtractedMetadata(
@@ -93,9 +93,9 @@ class Projector(config: ImageUploadOpsCfg,
       val s3Key = fileKeyFromId(imageId)
 
         if (!s3.doesObjectExist(config.originalFileBucket, s3Key))
-        throw new NoSuchImageExistsInS3(config.originalFileBucket, s3Key)
+        throw new NoSuchImageExistsInS3(config.originalFileBucket.name, s3Key)
 
-      val s3Source = Stopwatch(s"object exists, getting s3 object at s3://${config.originalFileBucket}/$s3Key to perform Image projection"){
+      val s3Source = Stopwatch(s"object exists, getting s3 object at s3://${config.originalFileBucket.name}/$s3Key to perform Image projection"){
         s3.getObject(config.originalFileBucket, s3Key)
       }(logMarker)
 
@@ -203,13 +203,13 @@ class ImageUploadProjectionOps(config: ImageUploadOpsCfg,
   }
 
   private def fetchFile(
-    bucket: String, key: String, outFile: File
+    bucket: S3Bucket, key: String, outFile: File
   )(implicit ec: ExecutionContext, logMarker: LogMarker): Future[Option[(File, MimeType)]] = {
-    logger.info(logMarker, s"Trying fetch existing image from S3 bucket - $bucket at key $key")
+    logger.info(logMarker, s"Trying fetch existing image from S3 bucket - ${bucket.name} at key $key")
     val doesFileExist = Future { s3.doesObjectExist(bucket, key) } recover { case _ => false }
     doesFileExist.flatMap {
       case false =>
-        logger.warn(logMarker, s"image did not exist in bucket $bucket at key $key")
+        logger.warn(logMarker, s"image did not exist in bucket ${bucket.name} at key $key")
         Future.successful(None) // falls back to creating from original file
       case true =>
         val obj = s3.getObject(bucket, key)
diff --git a/image-loader/app/model/Uploader.scala b/image-loader/app/model/Uploader.scala
index 6b3f722d715..65f58181ec7 100644
--- a/image-loader/app/model/Uploader.scala
+++ b/image-loader/app/model/Uploader.scala
@@ -7,7 +7,7 @@ import com.gu.mediaservice.lib.ImageIngestOperations.fileKeyFromId
 import com.gu.mediaservice.lib._
 import com.gu.mediaservice.lib.argo.ArgoHelpers
 import com.gu.mediaservice.lib.auth.Authentication
-import com.gu.mediaservice.lib.aws.{Embedder, EmbedderMessage, S3Object, UpdateMessage}
+import com.gu.mediaservice.lib.aws.{Embedder, EmbedderMessage, S3Bucket, S3Object, UpdateMessage}
 import com.gu.mediaservice.lib.cleanup.ImageProcessor
 import com.gu.mediaservice.lib.formatting._
 import com.gu.mediaservice.lib.imaging.ImageOperations
@@ -67,8 +67,8 @@ case class ImageUploadOpsCfg(
   thumbWidth: Int,
   thumbQuality: Double,
   transcodedMimeTypes: List[MimeType],
-  originalFileBucket: String,
-  thumbBucket: String
+  originalFileBucket: S3Bucket,
+  thumbBucket: S3Bucket
 )
 
 case class ImageUploadOpsDependencies(
@@ -475,6 +475,7 @@ class Uploader(
       ))
       // TODO: centralise where all these URLs are constructed
     } yield {
+      /*
       config.maybeLowerEnvironmentQueueBucketToSampleInto.foreach { lowerEnvironmentQueueBucket =>
         if (math.random() < config.lowerEnvironmentSamplingPercentageAsDecimal) {
           val mediaId = imageUpload.image.id
@@ -494,7 +495,7 @@ class Uploader(
           }
         }
       }
-
+      */
       UploadStatusUri(s"${config.rootUri(instance)}/uploadStatus/${uploadRequest.imageId}")
     }
 
diff --git a/image-loader/test/scala/lib/ResourceHelpers.scala b/image-loader/test/scala/lib/ResourceHelpers.scala
index 2abae0ce4dd..716235d362f 100644
--- a/image-loader/test/scala/lib/ResourceHelpers.scala
+++ b/image-loader/test/scala/lib/ResourceHelpers.scala
@@ -1,6 +1,9 @@
 package test.lib
 
+import com.gu.mediaservice.lib.aws.S3Bucket
+
 import java.io.File
+import java.net.URI
 
 object ResourceHelpers {
 
@@ -8,4 +11,8 @@ object ResourceHelpers {
     new File(getClass.getResource(s"/$resourcePath").toURI)
   }
 
+  /** A bucket with no working client/presigner, for tests that never touch S3. */
+  def dummyBucket(name: String): S3Bucket =
+    S3Bucket(name, new URI("s3.amazonaws.com"), usesPathStyleURLs = false, client = null, presigner = null)
+
 }
diff --git a/image-loader/test/scala/model/ImageUploadTest.scala b/image-loader/test/scala/model/ImageUploadTest.scala
index 9711519a0cf..874c4b70d7a 100644
--- a/image-loader/test/scala/model/ImageUploadTest.scala
+++ b/image-loader/test/scala/model/ImageUploadTest.scala
@@ -1,7 +1,7 @@
 package model
 
 import com.drew.imaging.ImageProcessingException
-import com.gu.mediaservice.lib.aws.{S3Metadata, S3Object, S3ObjectMetadata}
+import com.gu.mediaservice.lib.aws.{S3Bucket, S3Metadata, S3Object, S3ObjectMetadata}
 import com.gu.mediaservice.lib.cleanup.ImageProcessor
 import com.gu.mediaservice.lib.imaging.ImageOperations
 import com.gu.mediaservice.lib.logging.LogMarker
@@ -32,7 +32,7 @@ class ImageUploadTest extends AsyncFunSuite with Matchers with MockitoSugar {
   private implicit val logMarker: MockLogMarker = new MockLogMarker()
     // For mime type info, see https://github.com/guardian/grid/pull/2568
     val tempDir = new File("/tmp")
-    val mockConfig: ImageUploadOpsCfg = ImageUploadOpsCfg(tempDir, 256, 85d, List(Tiff), "img-bucket", "thumb-bucket")
+    val mockConfig: ImageUploadOpsCfg = ImageUploadOpsCfg(tempDir, 256, 85d, List(Tiff), ResourceHelpers.dummyBucket("img-bucket"), ResourceHelpers.dummyBucket("thumb-bucket"))
 
   /**
     * @todo: I flailed about until I found a path that worked, but
@@ -53,7 +53,7 @@ class ImageUploadTest extends AsyncFunSuite with Matchers with MockitoSugar {
 
     def mockStore = (a: StorableImage) =>
       Future.successful(
-        S3Object("madeupname", "madeupkey", a.file, Some(a.mimeType), None, a.meta, None)
+        mockS3Object
       )
 
     def storeOrProjectOriginalFile: StorableOriginalImage => Future[S3Object] = mockStore
diff --git a/image-loader/test/scala/model/ProjectorTest.scala b/image-loader/test/scala/model/ProjectorTest.scala
index 3c9e27e87ea..b7194928c92 100644
--- a/image-loader/test/scala/model/ProjectorTest.scala
+++ b/image-loader/test/scala/model/ProjectorTest.scala
@@ -41,7 +41,7 @@ class ProjectorTest extends AnyFreeSpec with Matchers with ScalaFutures with Moc
 
   private val imageOperations = new ImageOperations(ctxPath)
 
-  private val config = ImageUploadOpsCfg(new File("/tmp"), 256, 85d, Nil, "img-bucket", "thumb-bucket")
+  private val config = ImageUploadOpsCfg(new File("/tmp"), 256, 85d, Nil, dummyBucket("img-bucket"), dummyBucket("thumb-bucket"))
 
   private val maybeEmbedder = None
 
diff --git a/kahuna/app/lib/KahunaConfig.scala b/kahuna/app/lib/KahunaConfig.scala
index d4b65b73ecf..72ae0766557 100644
--- a/kahuna/app/lib/KahunaConfig.scala
+++ b/kahuna/app/lib/KahunaConfig.scala
@@ -2,6 +2,7 @@ package lib
 
 import com.gu.mediaservice.lib.auth.Permissions.Pinboard
 import com.gu.mediaservice.lib.auth.SimplePermission
+import com.gu.mediaservice.lib.aws.S3Bucket
 import com.gu.mediaservice.lib.config.{CommonConfig, GridConfigResources}
 import com.gu.mediaservice.model.Instance
 import play.api.libs.json._
@@ -51,9 +52,8 @@ class KahunaConfig(resources: GridConfigResources) extends CommonConfig(resource
   val aiSearchResultLimit: Int = intOpt("ai.search.resultLimit").getOrElse(200)
 
   val frameAncestors: Set[String] = getStringSet("security.frameAncestors")
-  val connectSources: Set[String] = getStringSet("security.connectSources") ++ maybeBucketForUIUploads.map { bucket =>
-    if (isDev) "https://localstack.media.local.dev-gutools.co.uk"
-    else s"https://$bucket.s3.$awsRegion.amazonaws.com"
+  val connectSources: Set[String] = getStringSet("security.connectSources") ++ maybeBucketForUIUploads.map { bucketForUIUploads =>
+    bucketForUIUploads.bucketURL().toURL.toExternalForm
   } ++ telemetryUri
   val fontSources: Set[String] = getStringSet("security.fontSources")
   val imageSources: Set[String] = getStringSet("security.imageSources")
diff --git a/media-api/app/MediaApiComponents.scala b/media-api/app/MediaApiComponents.scala
index ed62917800e..b73420d5ffd 100644
--- a/media-api/app/MediaApiComponents.scala
+++ b/media-api/app/MediaApiComponents.scala
@@ -18,9 +18,9 @@ class MediaApiComponents(context: Context) extends GridComponents(context, new M
   val messageSender = new ThrallMessageSender(config.thrallKinesisStreamConfig)
   val mediaApiMetrics = new MediaApiMetrics(config, actorSystem, applicationLifecycle)
 
-  val s3Client = new S3(config)
+  private val s3 = new S3(config)
 
-  val usageQuota = new UsageQuota(config, actorSystem.scheduler)
+  val usageQuota = new UsageQuota(config, actorSystem.scheduler, s3)
   usageQuota.quotaStore.update()
   usageQuota.scheduleUpdates()
   applicationLifecycle.addStopHook(() => Future{usageQuota.stopUpdates()})
@@ -28,12 +28,12 @@ class MediaApiComponents(context: Context) extends GridComponents(context, new M
   val elasticSearch = new ElasticSearch(config, mediaApiMetrics, config.esConfig, () => usageQuota.usageStore.overQuotaAgencies, actorSystem.scheduler, new InstancesClient(config, wsClient))
   // TODO needs to move somewhere more instance aware elasticSearch.ensureIndexExistsAndAliasAssigned()
 
-  val imageResponse = new ImageResponse(config, s3Client, usageQuota)
+  val imageResponse = new ImageResponse(config, s3, usageQuota)
 
   val softDeletedMetadataTable = new SoftDeletedMetadataTable(config)
   val embedder = new Embedder(new Bedrock(config), new SimpleSqsMessageConsumer(config.queueUrl, config))
 
-  val mediaApi = new MediaApi(auth, messageSender, softDeletedMetadataTable, elasticSearch, imageResponse, config, controllerComponents, s3Client, mediaApiMetrics, wsClient, authorisation, embedder, usageEvents)
+  val mediaApi = new MediaApi(auth, messageSender, softDeletedMetadataTable, elasticSearch, imageResponse, config, controllerComponents, s3, mediaApiMetrics, wsClient, authorisation, embedder, usageEvents)
   val suggestionController = new SuggestionController(auth, elasticSearch, controllerComponents)
   val aggController = new AggregationController(auth, elasticSearch, controllerComponents)
   val usageController = new UsageController(auth, config, elasticSearch, usageQuota, controllerComponents)
diff --git a/media-api/app/controllers/MediaApi.scala b/media-api/app/controllers/MediaApi.scala
index ec46f0b043e..e9e8ebd612d 100644
--- a/media-api/app/controllers/MediaApi.scala
+++ b/media-api/app/controllers/MediaApi.scala
@@ -44,7 +44,7 @@ class MediaApi(
                 imageResponse: ImageResponse,
                 config: MediaApiConfig,
                 override val controllerComponents: ControllerComponents,
-                s3Client: S3,
+                s3: S3,
                 mediaApiMetrics: MediaApiMetrics,
                 ws: WSClient,
                 authorisation: Authorisation,
@@ -324,7 +324,8 @@ class MediaApi(
         val maybeResult = for {
           export <- source.exports.find(_.id.contains(exportId))
           asset <- export.assets.find(_.dimensions.exists(_.width == width))
-          s3Res = Try(s3Client.getObject(config.imgPublishingBucket, asset.file))
+          key = config.imgPublishingBucket.keyFromURL(asset.file)
+          s3Res = Try(s3.getObject(config.imgPublishingBucket, key))
           _ = s3Res.failed.foreach { ex =>
             logger.error("Failed to fetch S3 object", ex)
           }
@@ -461,7 +462,8 @@ class MediaApi(
         val apiKey = request.user.accessor
         logger.info(logMarker, s"Download original image: $id from user: ${Authentication.getIdentity(request.user)}")
         mediaApiMetrics.incrementImageDownload(apiKey, mediaApiMetrics.OriginalDownloadType)
-        val s3Object = s3Client.getObject(config.imageBucket, image.source.file)
+        val key = config.imageBucket.keyFromURL(image.source.file)
+        val s3Object = s3.getObject(config.imageBucket, key)
         val file = StreamConverters.fromInputStream(() => s3Object)
         val entity = HttpEntity.Streamed(file, image.source.size, image.source.mimeType.map(_.name))
 
@@ -524,8 +526,9 @@ class MediaApi(
         logger.info(logMarker, s"Download optimised image: $id from user: ${Authentication.getIdentity(request.user)}")
         mediaApiMetrics.incrementImageDownload(apiKey, mediaApiMetrics.OptimisedDownloadType)
 
+        val key = config.imageBucket.keyFromURL(image.optimisedPng.getOrElse(image.source).file)
         val sourceImageUri =
-          new URI(s3Client.signUrl(config.imageBucket, image.optimisedPng.getOrElse(image.source).file, image, imageType = image.optimisedPng match {
+          new URI(s3.signUrl(config.imageBucket, key, image, imageType = image.optimisedPng match {
             case Some(_) => OptimisedPng
             case _ => Source
           }))
diff --git a/media-api/app/lib/ImageResponse.scala b/media-api/app/lib/ImageResponse.scala
index a8cd9df3bda..7934546a7b5 100644
--- a/media-api/app/lib/ImageResponse.scala
+++ b/media-api/app/lib/ImageResponse.scala
@@ -78,12 +78,14 @@ class ImageResponse(config: MediaApiConfig, s3Client: S3, usageQuota: UsageQuota
     val pngFileUri = image.optimisedPng.map(_.file)
 
     val fileUri = image.source.file
+    val imageKey = config.imageBucket.keyFromURL(fileUri)
 
-    val imageUrl = s3Client.signUrl(config.imageBucket, fileUri, image, imageType = Source)
+    val imageUrl = s3Client.signUrl(config.imageBucket, imageKey, image, imageType = Source)
     val pngUrl: Option[String] = pngFileUri
-      .map(s3Client.signUrl(config.imageBucket, _, image, imageType = OptimisedPng))
+      .map(file => s3Client.signUrl(config.imageBucket, config.imageBucket.keyFromURL(file), image, imageType = OptimisedPng))
 
-    def s3SignedThumbUrl = s3Client.signUrl(config.thumbBucket, fileUri, image, imageType = Thumbnail)
+    val thumbKey = config.thumbnailBucket.keyFromURL(fileUri)
+    def s3SignedThumbUrl = s3Client.signUrl(config.thumbnailBucket, thumbKey, image, imageType = Thumbnail)
 
     val thumbUrl = config.cloudFrontDomainThumbBucket
       .map(domain => s"https://$domain${fileUri.getPath}")
diff --git a/media-api/app/lib/MediaApiConfig.scala b/media-api/app/lib/MediaApiConfig.scala
index 5b48ec4a37c..13ea92abaae 100644
--- a/media-api/app/lib/MediaApiConfig.scala
+++ b/media-api/app/lib/MediaApiConfig.scala
@@ -1,5 +1,6 @@
 package lib
 
+import com.gu.mediaservice.lib.aws.S3Bucket
 import com.gu.mediaservice.lib.config.{CommonConfigWithElastic, GridConfigResources}
 import com.gu.mediaservice.lib.elasticsearch.filters
 import com.sksamuel.elastic4s.ElasticApi.{matchPhraseQuery, should}
@@ -14,23 +15,17 @@ import scala.collection.immutable
 import scala.util.Try
 
 case class StoreConfig(
-  storeBucket: String,
+  storeBucket: S3Bucket,
   storeKey: String
 )
 
 class MediaApiConfig(resources: GridConfigResources) extends CommonConfigWithElastic(resources) {
-  val configBucket: String = string("s3.config.bucket")
-  val usageMailBucket: String = string("s3.usagemail.bucket")
+  val configBucket: S3Bucket = S3Bucket(string("s3.config.bucket"), this)
+  val usageMailBucket: S3Bucket = S3Bucket(string("s3.usagemail.bucket"), this)
 
   val quotaStoreKey: String = string("quota.store.key")
   val quotaStoreConfig: StoreConfig = StoreConfig(configBucket, quotaStoreKey)
 
-  //Lazy allows this to be empty and not break things unless used somewhere
-  lazy val imgPublishingBucket = string("publishing.image.bucket")
-
-  val imageBucket: String = string("s3.image.bucket")
-  val thumbBucket: String = string("s3.thumb.bucket")
-
   val cloudFrontDomainThumbBucket: Option[String]   = stringOpt("cloudfront.domain.thumbbucket")
   val cloudFrontPrivateKeyBucket: Option[String]    = stringOpt("cloudfront.private-key.bucket")
   val cloudFrontPrivateKeyBucketKey: Option[String] = stringOpt("cloudfront.private-key.key")
diff --git a/media-api/app/lib/QuotaStore.scala b/media-api/app/lib/QuotaStore.scala
index c79486fbd17..9b380e6f32c 100644
--- a/media-api/app/lib/QuotaStore.scala
+++ b/media-api/app/lib/QuotaStore.scala
@@ -1,15 +1,17 @@
 package lib
 
 import com.gu.mediaservice.lib.BaseStore
+import com.gu.mediaservice.lib.aws.{S3, S3Bucket}
 import play.api.libs.json.Json
 
 import scala.concurrent.ExecutionContext
 
 class QuotaStore(
                   quotaFile: String,
-                  bucket: String,
-                  config: MediaApiConfig
-                )(implicit ec: ExecutionContext) extends BaseStore[String, SupplierUsageQuota](bucket, config)(ec) {
+                  bucket: S3Bucket,
+                  config: MediaApiConfig,
+                  s3: S3
+                )(implicit ec: ExecutionContext) extends BaseStore[String, SupplierUsageQuota](bucket, config, s3)(ec) {
 
   def getQuota: Map[String, SupplierUsageQuota] = store.get()
 
diff --git a/media-api/app/lib/UsageQuota.scala b/media-api/app/lib/UsageQuota.scala
index bcd4e93e212..75762e2d69e 100644
--- a/media-api/app/lib/UsageQuota.scala
+++ b/media-api/app/lib/UsageQuota.scala
@@ -2,6 +2,7 @@ package lib
 
 import org.apache.pekko.actor.Scheduler
 import com.gu.mediaservice.lib.FeatureToggle
+import com.gu.mediaservice.lib.aws.S3
 import com.gu.mediaservice.model.UsageRights
 
 import scala.concurrent.Await
@@ -12,17 +13,19 @@ import scala.util.Try
 case class ImageNotFound() extends Exception("Image not found")
 case class NoUsageQuota() extends Exception("No usage found for this image")
 
-class UsageQuota(config: MediaApiConfig, scheduler: Scheduler) {
+class UsageQuota(config: MediaApiConfig, scheduler: Scheduler, s3: S3) {
   val quotaStore = new QuotaStore(
     config.quotaStoreConfig.storeKey,
     config.quotaStoreConfig.storeBucket,
-    config
+    config,
+    s3
   )
 
   val usageStore = new UsageStore(
     config.usageMailBucket,
     config,
-    quotaStore
+    quotaStore,
+    s3
   )
 
   def scheduleUpdates(): Unit = {
diff --git a/media-api/app/lib/UsageStore.scala b/media-api/app/lib/UsageStore.scala
index 3490455caff..8e229a3e459 100644
--- a/media-api/app/lib/UsageStore.scala
+++ b/media-api/app/lib/UsageStore.scala
@@ -1,6 +1,7 @@
 package lib
 
 import com.gu.mediaservice.lib.BaseStore
+import com.gu.mediaservice.lib.aws.{S3, S3Bucket}
 import com.gu.mediaservice.lib.logging.GridLogging
 import com.gu.mediaservice.model.{Agencies, Agency, UsageRights}
 import com.gu.mediaservice.model.usage.{DigitalUsage, PrintUsage, PublishedUsageStatus, RemovedUsageStatus, UnknownUsageStatus, Usage, UsageStatus, UsageType}
@@ -60,10 +61,11 @@ object UsageStore extends GridLogging {
 }
 
 class UsageStore(
-  bucket: String,
+  bucket: S3Bucket,
   config: MediaApiConfig,
-  quotaStore: QuotaStore
-)(implicit val ec: ExecutionContext) extends BaseStore[String, SupplierUsageStatus](bucket, config) with GridLogging {
+  quotaStore: QuotaStore,
+  s3: S3
+)(implicit val ec: ExecutionContext) extends BaseStore[String, SupplierUsageStatus](bucket, config, s3) with GridLogging {
 
   def getUsageStatusForUsageRights(usageRights: UsageRights): Future[SupplierUsageStatus] = {
     usageRights match {
diff --git a/media-api/test/lib/elasticsearch/Fixtures.scala b/media-api/test/lib/elasticsearch/Fixtures.scala
index f2b5a0e4957..70375788be0 100644
--- a/media-api/test/lib/elasticsearch/Fixtures.scala
+++ b/media-api/test/lib/elasticsearch/Fixtures.scala
@@ -37,6 +37,7 @@ trait Fixtures {
     "es6.url",
     "s3.image.bucket",
     "s3.thumb.bucket",
+    "publishing.image.bucket",
     "grid.stage",
     "grid.appName",
     "instance.service.my",
diff --git a/rest-lib/src/main/scala/com/gu/mediaservice/lib/auth/provider/ApiKeyAuthenticationProvider.scala b/rest-lib/src/main/scala/com/gu/mediaservice/lib/auth/provider/ApiKeyAuthenticationProvider.scala
index e42c25c7180..d315662b740 100644
--- a/rest-lib/src/main/scala/com/gu/mediaservice/lib/auth/provider/ApiKeyAuthenticationProvider.scala
+++ b/rest-lib/src/main/scala/com/gu/mediaservice/lib/auth/provider/ApiKeyAuthenticationProvider.scala
@@ -2,6 +2,7 @@ package com.gu.mediaservice.lib.auth.provider
 import com.gu.mediaservice.lib.auth.Authentication.{MachinePrincipal, Principal}
 import com.gu.mediaservice.lib.auth.provider.ApiKeyAuthenticationProvider.{ApiKeyInstance, KindeIdKey}
 import com.gu.mediaservice.lib.auth.{ApiAccessor, KeyStore}
+import com.gu.mediaservice.lib.aws.{S3, S3Bucket}
 import com.gu.mediaservice.lib.config.InstanceForRequest
 import com.gu.mediaservice.lib.events.UsageEvents
 import com.gu.mediaservice.model.Instance
@@ -25,7 +26,7 @@ class ApiKeyAuthenticationProvider(configuration: Configuration, resources: Auth
   var keyStorePlaceholder: Option[KeyStore] = _
 
   override def initialise(): Unit = {
-    val store = new KeyStore(configuration.get[String]("authKeyStoreBucket"), resources.commonConfig)
+    val store = new KeyStore(S3Bucket(configuration.get[String]("authKeyStoreBucket"), resources.commonConfig), resources.commonConfig, new S3(resources.commonConfig))
     store.scheduleUpdates(resources.actorSystem.scheduler)
     keyStorePlaceholder = Some(store)
   }
diff --git a/rest-lib/src/main/scala/com/gu/mediaservice/lib/guardian/auth/PandaAuthenticationProvider.scala b/rest-lib/src/main/scala/com/gu/mediaservice/lib/guardian/auth/PandaAuthenticationProvider.scala
index d98cefe4efc..ab5b36f80d1 100644
--- a/rest-lib/src/main/scala/com/gu/mediaservice/lib/guardian/auth/PandaAuthenticationProvider.scala
+++ b/rest-lib/src/main/scala/com/gu/mediaservice/lib/guardian/auth/PandaAuthenticationProvider.scala
@@ -161,7 +161,7 @@ class PandaAuthenticationProvider(
       system = providerConfiguration.getOptional[String]("panda.system").getOrElse("media-service"),
       bucketName = providerConfiguration.getOptional[String]("panda.bucketName").getOrElse("pan-domain-auth-settings"),
       settingsFileKey = providerConfiguration.getOptional[String]("panda.settingsFileKey").getOrElse(s"$domain.settings"),
-      s3Client = S3Ops.buildS3Client(resources.commonConfig, localstackAware=resources.commonConfig.useLocalAuth)
+      s3Client = S3Ops.buildS3Client(resources.commonConfig, usesPathStyleURLs = resources.commonConfig.useLocalAuth)
     )
   }
 
diff --git a/rest-lib/src/test/resources/application.conf b/rest-lib/src/test/resources/application.conf
index 930c6a00d57..1bff942991f 100644
--- a/rest-lib/src/test/resources/application.conf
+++ b/rest-lib/src/test/resources/application.conf
@@ -3,3 +3,6 @@ grid.appName: "test"
 thrall.kinesis.stream.name: "not-used"
 thrall.kinesis.lowPriorityStream.name: "not-used"
 domain.root: "notused.example.com"
+s3.image.bucket: "not-used"
+s3.thumb.bucket: "not-used"
+publishing.image.bucket: "not-used"
diff --git a/rest-lib/src/test/scala/com/gu/mediaservice/lib/auth/ApiKeyAuthenticationProviderTest.scala b/rest-lib/src/test/scala/com/gu/mediaservice/lib/auth/ApiKeyAuthenticationProviderTest.scala
index 3ba9231943c..64ef9094330 100644
--- a/rest-lib/src/test/scala/com/gu/mediaservice/lib/auth/ApiKeyAuthenticationProviderTest.scala
+++ b/rest-lib/src/test/scala/com/gu/mediaservice/lib/auth/ApiKeyAuthenticationProviderTest.scala
@@ -3,6 +3,7 @@ package com.gu.mediaservice.lib.auth
 import org.apache.pekko.actor.ActorSystem
 import com.gu.mediaservice.lib.auth.Authentication.MachinePrincipal
 import com.gu.mediaservice.lib.auth.provider.{ApiKeyAuthenticationProvider, Authenticated, AuthenticationProviderResources, Invalid, NotAuthenticated, NotAuthorised}
+import com.gu.mediaservice.lib.aws.{S3, S3Bucket}
 import com.gu.mediaservice.lib.config.{CommonConfig, GridConfigResources}
 import com.gu.mediaservice.lib.events.UsageEvents
 import com.gu.mediaservice.model.Instance
@@ -17,6 +18,7 @@ import play.api.mvc.DefaultControllerComponents
 import play.api.test.{FakeRequest, WsTestClient}
 import play.api.{Configuration, Environment}
 
+import java.net.URI
 import scala.concurrent.ExecutionContext.global
 import scala.concurrent.Future
 
@@ -36,6 +38,7 @@ class ApiKeyAuthenticationProviderTest extends AsyncFreeSpec with Matchers with
   private val providerConfig = Configuration.empty
   private val controllerComponents: DefaultControllerComponents = DefaultControllerComponents(null, null, null, null, null, global)
   private val resources = AuthenticationProviderResources(config, actorSystem, wsClient, controllerComponents, mock[Authorisation], mock[CookieSigner], mock[UsageEvents] )
+  private val s3 = new S3(config)
   private val provider = new ApiKeyAuthenticationProvider(providerConfig, resources) {
     override def initialise(): Unit = { /* do nothing */ }
 
@@ -43,7 +46,7 @@ class ApiKeyAuthenticationProviderTest extends AsyncFreeSpec with Matchers with
       Future.successful(())
     }
 
-    override def keyStore: KeyStore = new KeyStore("not-used", resources.commonConfig) {
+    override def keyStore: KeyStore = new KeyStore(S3Bucket("not-used", new URI("https://s3.amazonaws.com"), usesPathStyleURLs = false, client = null, presigner = null), resources.commonConfig, s3) {
       override def lookupIdentity(key: String)(implicit instance: Instance): Option[ApiAccessor] = {
         key match {
           case "key-chuckle" => Some(ApiAccessor("brothers", Internal))
diff --git a/thrall/app/ThrallComponents.scala b/thrall/app/ThrallComponents.scala
index 9d5f8e4d0c9..0f908c99a41 100644
--- a/thrall/app/ThrallComponents.scala
+++ b/thrall/app/ThrallComponents.scala
@@ -91,7 +91,6 @@ class ThrallComponents(context: Context) extends GridComponents(context, new Thr
 
   val streamRunning: Future[Done] = thrallStreamProcessor.run()
 
-  val s3 = S3Ops.buildS3Client(config)
   val s3Vectors = new S3Vectors(config)
 
   Source.repeat(()).throttle(1, per = 5.minute).map(_ => {
diff --git a/thrall/app/controllers/ReaperController.scala b/thrall/app/controllers/ReaperController.scala
index c4ae8367e12..8e32cb5e78d 100644
--- a/thrall/app/controllers/ReaperController.scala
+++ b/thrall/app/controllers/ReaperController.scala
@@ -19,11 +19,9 @@ import play.api.libs.json.{JsValue, Json, OWrites}
 import play.api.libs.ws.WSClient
 import play.api.mvc.{Action, AnyContent, ControllerComponents}
 import scalaz.NonEmptyList
-import software.amazon.awssdk.services.s3.model.ListObjectsV2Request
 
 import scala.concurrent.duration.DurationInt
 import scala.concurrent.{ExecutionContext, Future}
-import scala.jdk.CollectionConverters._
 import scala.language.postfixOps
 import scala.util.control.NonFatal
 import scala.util.{Failure, Success}
@@ -200,31 +198,27 @@ class ReaperController(
       }.toMap
     }).map(Json.toJson(_))
   }
-  def index = withLoginRedirect {
+  def index: Action[AnyContent] = withLoginRedirectAsync {
     val now = DateTime.now(DateTimeZone.UTC)
     (config.maybeReaperBucket, config.maybeReaperCountPerRun) match {
-    case (None, _) => NotImplemented("'s3.reaper.bucket' not configured in thrall.conf")
-    case (_, None) => NotImplemented("'reaper.countPerRun' not configured in thrall.conf")
+    case (None, _) => Future.successful(NotImplemented("'s3.reaper.bucket' not configured in thrall.conf"))
+    case (_, None) => Future.successful(NotImplemented("'reaper.countPerRun' not configured in thrall.conf"))
     case (Some(reaperBucket), Some(countOfImagesToReap)) =>
-      val recentRecords = List(now, now.minusDays(1), now.minusDays(2)).flatMap { day =>
+      Future.sequence(List(now, now.minusDays(1), now.minusDays(2)).map { day =>
         val s3DirName = s3DirNameFromDate(day)
-        val softDeletes = store.client.listObjectsV2(
-            ListObjectsV2Request.builder().bucket(reaperBucket).prefix(s"soft/$s3DirName/").build()
-        ).contents().asScala.toList
-
-        val hardDeletes = store.client.listObjectsV2(
-          ListObjectsV2Request.builder().bucket(reaperBucket).prefix(s"hard/$s3DirName/").build()
-        ).contents().asScala.toList
-
-        softDeletes ++ hardDeletes
+        for {
+          softDeletes <- store.list(reaperBucket, s"soft/$s3DirName")
+          hardDeletes <- store.list(reaperBucket, s"hard/$s3DirName")
+        } yield softDeletes ++ hardDeletes
+      }).map { recentRecords =>
+        val recentRecordKeys = recentRecords.flatten
+          .filter(_.metadata.objectMetadata.lastModified.exists(_ isAfter now.minusHours(48)))
+          .sortBy(_.metadata.objectMetadata.lastModified.map(_.getMillis))
+          .reverse
+          .map(s3Object => reaperBucket.keyFromURL(s3Object.uri))
+
+        Ok(views.html.reaper(isPaused, INTERVAL.toString(), countOfImagesToReap, recentRecordKeys))
       }
-      val recentRecordKeys = recentRecords
-        .filter(_.lastModified() isAfter now.minusHours(48).toDate.toInstant)
-        .sortBy(_.lastModified())
-        .reverse
-        .map(_.key())
-
-      Ok(views.html.reaper(isPaused, INTERVAL.toString(), countOfImagesToReap, recentRecordKeys))
   }}
 
   def reaperRecord(key: String) = auth { config.maybeReaperBucket match {
diff --git a/thrall/app/lib/ThrallConfig.scala b/thrall/app/lib/ThrallConfig.scala
index 477cfc5d3e5..8518a442051 100644
--- a/thrall/app/lib/ThrallConfig.scala
+++ b/thrall/app/lib/ThrallConfig.scala
@@ -1,6 +1,6 @@
 package lib
 
-import com.gu.mediaservice.lib.aws.AwsClientBuilderUtils
+import com.gu.mediaservice.lib.aws.{AwsClientBuilderUtils, S3Bucket}
 import com.gu.mediaservice.lib.cleanup.ReapableEligibiltyResources
 import com.gu.mediaservice.lib.config.{CommonConfigWithElastic, GridConfigResources, ReapableEligibilityLoader}
 import com.gu.mediaservice.lib.elasticsearch.ReapableEligibility
@@ -56,11 +56,7 @@ object KinesisReceiverConfig {
 }
 
 class ThrallConfig(resources: GridConfigResources) extends CommonConfigWithElastic(resources) {
-  val imageBucket: String = string("s3.image.bucket")
-
-  val thumbnailBucket: String = string("s3.thumb.bucket")
-
-  val maybeReaperBucket: Option[String] = stringOpt("s3.reaper.bucket")
+  val maybeReaperBucket: Option[S3Bucket] = stringOpt("s3.reaper.bucket").map(S3Bucket(_, this))
   val maybeReaperCountPerRun: Option[Int] = intOpt("reaper.countPerRun")
 
   val metadataTopicArn: String = string("indexed.image.sns.topic.arn")