diff --git a/libraries/object-persistor/src/Errors.js b/libraries/object-persistor/src/Errors.js
index 330ffe42b5..0c9dd52f68 100644
--- a/libraries/object-persistor/src/Errors.js
+++ b/libraries/object-persistor/src/Errors.js
@@ -6,6 +6,7 @@ class ReadError extends OError {}
 class SettingsError extends OError {}
 class NotImplementedError extends OError {}
 class AlreadyWrittenError extends OError {}
+class NoKEKMatchedError extends OError {}
 
 module.exports = {
   NotFoundError,
@@ -14,4 +15,5 @@ module.exports = {
   SettingsError,
   NotImplementedError,
   AlreadyWrittenError,
+  NoKEKMatchedError,
 }
diff --git a/libraries/object-persistor/src/PerProjectEncryptedS3Persistor.js b/libraries/object-persistor/src/PerProjectEncryptedS3Persistor.js
index ba9550839f..8e58641591 100644
--- a/libraries/object-persistor/src/PerProjectEncryptedS3Persistor.js
+++ b/libraries/object-persistor/src/PerProjectEncryptedS3Persistor.js
@@ -1,31 +1,62 @@
 // @ts-check
-const Stream = require('stream')
-const { promisify } = require('util')
 const Crypto = require('crypto')
+const Stream = require('stream')
+const fs = require('fs')
+const { promisify } = require('util')
 const { WritableBuffer } = require('@overleaf/stream-utils')
 const { S3Persistor, SSECOptions } = require('./S3Persistor.js')
+const {
+  AlreadyWrittenError,
+  NoKEKMatchedError,
+  NotFoundError,
+  NotImplementedError,
+  ReadError,
+} = require('./Errors')
+const logger = require('@overleaf/logger')
 
 const generateKey = promisify(Crypto.generateKey)
 
 /**
- * @typedef {Object} Settings
- * @property {(bucketName: string, path: string) => {bucketName: string, path: string}} pathToDataEncryptionKeyPath
- * @property {(bucketName: string, path: string) => boolean} pathIsProjectFolder
- * @property {() => Promise<Buffer>} getKeyEncryptionKey
+ * @typedef {import('aws-sdk').AWSError} AWSError
  */
 
-const {
-  NotFoundError,
-  NotImplementedError,
-  AlreadyWrittenError,
-} = require('./Errors')
-const fs = require('fs')
+/**
+ * @typedef {Object} Settings
+ * @property {boolean} automaticallyRotateDEKEncryption
+ * @property {boolean} ignoreErrorsFromDEKReEncryption
+ * @property {(bucketName: string, path: string) => {bucketName: string, path: string}} pathToDataEncryptionKeyPath
+ * @property {(bucketName: string, path: string) => boolean} pathIsProjectFolder
+ * @property {() => Promise<Array<Buffer>>} getKeyEncryptionKeys
+ */
+
+/**
+ * Helper function to make TS happy when accessing error properties
+ * AWSError is not an actual class, so we cannot use instanceof.
+ * @param {any} err
+ * @return {err is AWSError}
+ */
+function isAWSError(err) {
+  return !!err
+}
+
+/**
+ * @param {any} err
+ * @return {boolean}
+ */
+function isForbiddenError(err) {
+  if (!err || !(err instanceof ReadError || err instanceof NotFoundError)) {
+    return false
+  }
+  const cause = err.cause
+  if (!isAWSError(cause)) return false
+  return cause.statusCode === 403
+}
 
 class PerProjectEncryptedS3Persistor extends S3Persistor {
-  /** @type Settings */
+  /** @type {Settings} */
   #settings
-  /** @type Promise<SSECOptions> */
-  #keyEncryptionKeyOptions
+  /** @type {Promise<Array<SSECOptions>>} */
+  #availableKeyEncryptionKeysPromise
 
   /**
    * @param {Settings} settings
@@ -33,13 +64,21 @@ class PerProjectEncryptedS3Persistor extends S3Persistor {
   constructor(settings) {
     super(settings)
     this.#settings = settings
-    this.#keyEncryptionKeyOptions = this.#settings
-      .getKeyEncryptionKey()
-      .then(keyAsBuffer => new SSECOptions(keyAsBuffer))
+    this.#availableKeyEncryptionKeysPromise = this.#settings
+      .getKeyEncryptionKeys()
+      .then(keysAsBuffer => {
+        if (keysAsBuffer.length === 0) throw new Error('no kek provided')
+        return keysAsBuffer.map(buffer => new SSECOptions(buffer))
+      })
   }
 
-  async ensureKeyEncryptionKeyLoaded() {
-    await this.#keyEncryptionKeyOptions
+  async ensureKeyEncryptionKeysLoaded() {
+    await this.#availableKeyEncryptionKeysPromise
+  }
+
+  async #getCurrentKeyEncryptionKey() {
+    const available = await this.#availableKeyEncryptionKeysPromise
+    return available[0]
   }
 
   /**
@@ -48,9 +87,17 @@ class PerProjectEncryptedS3Persistor extends S3Persistor {
    */
   async getDataEncryptionKeySize(bucketName, path) {
     const dekPath = this.#settings.pathToDataEncryptionKeyPath(bucketName, path)
-    return await super.getObjectSize(dekPath.bucketName, dekPath.path, {
-      ssecOptions: await this.#keyEncryptionKeyOptions,
-    })
+    for (const ssecOptions of await this.#availableKeyEncryptionKeysPromise) {
+      try {
+        return await super.getObjectSize(dekPath.bucketName, dekPath.path, {
+          ssecOptions,
+        })
+      } catch (err) {
+        if (isForbiddenError(err)) continue
+        throw err
+      }
+    }
+    throw new NoKEKMatchedError('no kek matched')
   }
 
   /**
@@ -91,7 +138,7 @@ class PerProjectEncryptedS3Persistor extends S3Persistor {
       {
         // Do not overwrite any objects if already created
         ifNoneMatch: '*',
-        ssecOptions: await this.#keyEncryptionKeyOptions,
+        ssecOptions: await this.#getCurrentKeyEncryptionKey(),
       }
     )
     return new SSECOptions(dataEncryptionKey)
@@ -104,11 +151,42 @@ class PerProjectEncryptedS3Persistor extends S3Persistor {
    */
   async #getExistingDataEncryptionKeyOptions(bucketName, path) {
     const dekPath = this.#settings.pathToDataEncryptionKeyPath(bucketName, path)
-    const res = await super.getObjectStream(dekPath.bucketName, dekPath.path, {
-      ssecOptions: await this.#keyEncryptionKeyOptions,
-    })
+    let res
+    let kekIndex = 0
+    for (const ssecOptions of await this.#availableKeyEncryptionKeysPromise) {
+      try {
+        res = await super.getObjectStream(dekPath.bucketName, dekPath.path, {
+          ssecOptions,
+        })
+      } catch (err) {
+        if (isForbiddenError(err)) {
+          kekIndex++
+          continue
+        }
+        throw err
+      }
+    }
+    if (!res) throw new NoKEKMatchedError('no kek matched')
     const buf = new WritableBuffer()
     await Stream.promises.pipeline(res, buf)
+
+    if (kekIndex !== 0 && this.#settings.automaticallyRotateDEKEncryption) {
+      try {
+        await super.sendStream(
+          dekPath.bucketName,
+          dekPath.path,
+          Stream.Readable.from([buf.getContents()]),
+          { ssecOptions: await this.#getCurrentKeyEncryptionKey() }
+        )
+      } catch (err) {
+        if (this.#settings.ignoreErrorsFromDEKReEncryption) {
+          logger.warn({ err, ...dekPath }, 'failed to persist re-encrypted DEK')
+        } else {
+          throw err
+        }
+      }
+    }
+
     return new SSECOptions(buf.getContents())
   }
 
diff --git a/libraries/object-persistor/src/S3Persistor.js b/libraries/object-persistor/src/S3Persistor.js
index 54ff33c940..110414d560 100644
--- a/libraries/object-persistor/src/S3Persistor.js
+++ b/libraries/object-persistor/src/S3Persistor.js
@@ -179,7 +179,8 @@ class S3Persistor extends AbstractPersistor {
             case 200: // full response
             case 206: // partial response
               return resolve(undefined)
-            case 403: // AccessDenied is handled the same as NoSuchKey
+            case 403: // AccessDenied
+              return // handled by stream.on('error') handler below
             case 404: // NoSuchKey
               return reject(new NotFoundError('not found'))
             default:
diff --git a/libraries/object-persistor/test/unit/S3PersistorTests.js b/libraries/object-persistor/test/unit/S3PersistorTests.js
index 60d639fce5..cc785034a2 100644
--- a/libraries/object-persistor/test/unit/S3PersistorTests.js
+++ b/libraries/object-persistor/test/unit/S3PersistorTests.js
@@ -93,6 +93,12 @@ describe('S3PersistorTests', function () {
         setTimeout(() => {
           if (this.err) return ReadStream.emit('error', this.err)
           this.emit('httpHeaders', this.statusCode)
+          if (this.statusCode === 403) {
+            ReadStream.emit('error', S3AccessDeniedError)
+          }
+          if (this.statusCode === 404) {
+            ReadStream.emit('error', S3NotFoundError)
+          }
         })
         return ReadStream
       }
@@ -359,7 +365,7 @@ describe('S3PersistorTests', function () {
       })
 
       it('wraps the error', function () {
-        expect(error.cause).to.exist
+        expect(error.cause).to.equal(S3AccessDeniedError)
       })
 
       it('stores the bucket and key in the error', function () {
diff --git a/services/filestore/test/acceptance/js/FilestoreTests.js b/services/filestore/test/acceptance/js/FilestoreTests.js
index 5254daf9c5..6eef9e585b 100644
--- a/services/filestore/test/acceptance/js/FilestoreTests.js
+++ b/services/filestore/test/acceptance/js/FilestoreTests.js
@@ -32,12 +32,15 @@ process.on('unhandledRejection', e => {
 
 // store settings for multiple backends, so that we can test each one.
 // fs will always be available - add others if they are configured
-const { BackendSettings, s3Config } = require('./TestConfig')
+const { BackendSettings, s3Config, s3SSECConfig } = require('./TestConfig')
 const {
   AlreadyWrittenError,
   NotFoundError,
   NotImplementedError,
+  NoKEKMatchedError,
 } = require('@overleaf/object-persistor/src/Errors')
+const PerProjectEncryptedS3Persistor = require('@overleaf/object-persistor/src/PerProjectEncryptedS3Persistor')
+const crypto = require('crypto')
 
 describe('Filestore', function () {
   this.timeout(1000 * 10)
@@ -1015,12 +1018,19 @@ describe('Filestore', function () {
           expect(dataEncryptionKeySize).to.equal(32)
         })
 
-        let fileId1, fileId2, fileKey1, fileKey2, fileUrl1, fileUrl2
+        let fileId1,
+          fileId2,
+          fileKey1,
+          fileKey2,
+          fileKeyOtherProject,
+          fileUrl1,
+          fileUrl2
         beforeEach('prepare ids', function () {
           fileId1 = new ObjectId().toString()
           fileId2 = new ObjectId().toString()
           fileKey1 = `${projectId}/${fileId1}`
           fileKey2 = `${projectId}/${fileId2}`
+          fileKeyOtherProject = `${new ObjectId().toString()}/${new ObjectId().toString()}`
           fileUrl1 = `${filestoreUrl}/project/${projectId}/file/${fileId1}`
           fileUrl2 = `${filestoreUrl}/project/${projectId}/file/${fileId2}`
         })
@@ -1107,6 +1117,139 @@ describe('Filestore', function () {
           await checkGET2()
         })
 
+        describe('kek rotation', function () {
+          const newKEK = crypto.generateKeySync('aes', { length: 256 }).export()
+          const oldKEK = crypto.generateKeySync('aes', { length: 256 }).export()
+          const migrationStep0 = new PerProjectEncryptedS3Persistor({
+            ...s3SSECConfig(),
+            automaticallyRotateDEKEncryption: false,
+            async getKeyEncryptionKeys() {
+              return [oldKEK] // only old key
+            },
+          })
+          const migrationStep1 = new PerProjectEncryptedS3Persistor({
+            ...s3SSECConfig(),
+            automaticallyRotateDEKEncryption: false,
+            async getKeyEncryptionKeys() {
+              return [oldKEK, newKEK] // new key as fallback
+            },
+          })
+          const migrationStep2 = new PerProjectEncryptedS3Persistor({
+            ...s3SSECConfig(),
+            automaticallyRotateDEKEncryption: true, // <- different compared to partiallyRotated
+            async getKeyEncryptionKeys() {
+              return [newKEK, oldKEK] // old keys as fallback
+            },
+          })
+          const migrationStep3 = new PerProjectEncryptedS3Persistor({
+            ...s3SSECConfig(),
+            automaticallyRotateDEKEncryption: true,
+            async getKeyEncryptionKeys() {
+              return [newKEK] // only new key
+            },
+          })
+
+          async function checkWrites(
+            fileKey,
+            writer,
+            readersSuccess,
+            readersFailed
+          ) {
+            const content = Math.random().toString()
+            await writer.sendStream(
+              Settings.filestore.stores.user_files,
+              fileKey,
+              Stream.Readable.from([content])
+            )
+
+            for (const persistor of readersSuccess) {
+              await TestHelper.expectPersistorToHaveFile(
+                persistor,
+                backendSettings.stores.user_files,
+                fileKey,
+                content
+              )
+            }
+
+            for (const persistor of readersFailed) {
+              await expect(
+                TestHelper.expectPersistorToHaveFile(
+                  persistor,
+                  backendSettings.stores.user_files,
+                  fileKey,
+                  content
+                )
+              ).to.be.rejectedWith(NoKEKMatchedError)
+            }
+          }
+
+          const stages = [
+            {
+              name: 'stage 0 - [old]',
+              prev: migrationStep0,
+              cur: migrationStep0,
+              fail: [migrationStep3],
+            },
+            {
+              name: 'stage 1 - [old,new]',
+              prev: migrationStep0,
+              cur: migrationStep1,
+              fail: [],
+            },
+            {
+              name: 'stage 2 - [new,old]',
+              prev: migrationStep1,
+              cur: migrationStep2,
+              fail: [],
+            },
+            {
+              name: 'stage 3 - [new]',
+              prev: migrationStep2,
+              cur: migrationStep3,
+              fail: [migrationStep0],
+            },
+          ]
+
+          for (const { name, prev, cur, fail } of stages) {
+            describe(name, function () {
+              it('can read old writes', async function () {
+                await checkWrites(fileKey1, prev, [prev, cur], fail)
+                await checkWrites(fileKey2, prev, [prev, cur], fail) // check again after access
+                await checkWrites(fileKeyOtherProject, prev, [prev, cur], fail)
+              })
+              it('can read new writes', async function () {
+                await checkWrites(fileKey1, prev, [prev, cur], fail)
+                await checkWrites(fileKey2, cur, [prev, cur], fail) // check again after access
+                await checkWrites(fileKeyOtherProject, cur, [prev, cur], fail)
+              })
+            })
+          }
+
+          describe('full migration', function () {
+            it('can read old writes if rotated in sequence', async function () {
+              await checkWrites(
+                fileKey1,
+                migrationStep0,
+                [
+                  migrationStep0,
+                  migrationStep1,
+                  migrationStep2, // migrates
+                  migrationStep3,
+                ],
+                []
+              )
+            })
+            it('cannot read/write if not rotated', async function () {
+              await checkWrites(
+                fileKey1,
+                migrationStep0,
+                [migrationStep0],
+                [migrationStep3]
+              )
+            })
+          })
+        })
+
         let s3Client
         before('create s3Client', function () {
           const cfg = s3Config()
diff --git a/services/filestore/test/acceptance/js/TestConfig.js b/services/filestore/test/acceptance/js/TestConfig.js
index ffc94601d3..f5f8b60060 100644
--- a/services/filestore/test/acceptance/js/TestConfig.js
+++ b/services/filestore/test/acceptance/js/TestConfig.js
@@ -26,9 +26,13 @@ function s3Config() {
   }
 }
 
+const S3SSECKeys = [crypto.generateKeySync('aes', { length: 256 }).export()]
+
 function s3SSECConfig() {
   return {
     ...s3Config(),
+    ignoreErrorsFromDEKReEncryption: false,
+    automaticallyRotateDEKEncryption: true,
     pathIsProjectFolder(_bucketName, path) {
       return !!path.match(/^[a-f0-9]+\/$/)
     },
@@ -39,8 +43,8 @@ function s3SSECConfig() {
         path: Path.join(projectFolder, 'dek'),
       }
     },
-    async getKeyEncryptionKey() {
-      return crypto.generateKeySync('aes', { length: 256 }).export()
+    async getKeyEncryptionKeys() {
+      return S3SSECKeys
     },
   }
 }
@@ -177,4 +181,5 @@ checkForUnexpectedTestFile()
 module.exports = {
   BackendSettings,
   s3Config,
+  s3SSECConfig,
 }