diff --git a/README.md b/README.md
index 78cc783..9130160 100644
--- a/README.md
+++ b/README.md
@@ -76,7 +76,9 @@ It's way more work, and way more hassle. This is great for sensitive files, and
 
 ## What this gem _cannot_ do
 
-This gem does not provide an E2E encrypted solution. The file still gets encrypted by your cloud provider, and decrypted by your cloud provider. While it offers a strong protection _at rest_ it does not offer extra protection _in transit._ If you need that level of protection, you may want to look into [S3 client encryption](https://ankane.org/activestorage-s3-encryption) or other similar tech.
+The `EncryptedGCSService` and the `EncryptedS3Service` do not provide an E2E encrypted solution. The file still gets encrypted by your cloud provider, and decrypted by your cloud provider. While it offers a strong protection _at rest_ it does not offer extra protection _in transit._ If you need that level of protection on S3-compatible storage, use the [ClientSideEncryptedS3Service](#clientsideencrypteds3service---s3-compatible-storage-encrypted-in-your-application), which never hands the provider anything but ciphertext.
+
+Nothing here protects you against an attacker who has your running application, since the application is what holds the keys.
 
 ## Encrypted Service implementations
 
@@ -115,6 +117,32 @@ Implementation details:
 
 While S3 allows the `x-amz-server-side-encryption-customer-key-MD5` to be added to the signed URL for PUT, the value of that header gets removed from the signature due to the process called "hoisting" - which occurs during the signing of the URL. So your client _may_ override the encryption key you give it forcibly, by replacing the `x-amz-server-side-encryption-customer-key` and `x-amz-server-side-encryption-customer-key-MD5`. This can produce Blobs encrypted with a key you do not have. If you want to exclude the possibility of this, you need to perform an integrity check on your uploads. The integrity check will fail if the encryption key has been overridden in this manner, and you can then destroy the Blob. This problem has been reported to AWS.
 
+### ClientSideEncryptedS3Service - S3-compatible storage, encrypted in your application
+
+Where the `EncryptedS3Service` asks S3 to encrypt for us (SSE-C), this service encrypts inside your application and hands the bucket ciphertext only. The provider never holds the plaintext, at rest or in transit, and neither does anything between you and it. If your reason for encrypting is that you do not want to trust your storage provider with the contents, this is the service to use.
+
+```yaml
+# storage.yml
+encrypted_s3:
+  service: ClientSideEncryptedS3
+  bucket: my-bucket
+  region: eu-central-1
+  private_url_policy: stream
+```
+
+Implementation details:
+
+* It uses the same encryption scheme as the `EncryptedDiskService`, and the same `block_cipher_kit` code path, rather than introducing a second scheme. The schemes only need an IO to read from, and `ClientSideEncryptedS3Service::SeekableObjectIO` gives them one backed by ranged `GET` requests. So random access - being able to serve a byte range of a large video without downloading all of it - survives the move from a local disk to a bucket.
+* Reads are buffered with a window that starts at 64 KB and doubles up to 5 MB for as long as reads stay sequential, resetting after a seek. A one-byte `download_chunk` costs one small request; a full download quickly reaches the large window.
+* Each object begins with a five byte header: the ASCII `ASEC`, then one byte of scheme version. An object in a bucket has no filename to hang a version on (which is how the `EncryptedDiskService` does it) and asking for object metadata costs a request, so the ciphertext names its own format. Reading an object without that header raises `ActiveStorageEncryption::UnknownCiphertextFormat`, which is also how a blob written by a stock `S3Service` announces itself.
+* `private_url_policy: require_headers` is refused at configuration time. A presigned URL can only ever serve ciphertext, and the client has no key. Use `stream` or `disable`.
+* Direct uploads do not go to the bucket - the browser has no key, so a `PUT` there would store plaintext. They are routed to `EncryptedBlobsController` exactly as the `EncryptedDiskService` routes them: the application receives the plaintext, encrypts it, and puts it in the bucket. No bucket CORS configuration is needed.
+* SSE-C is not used, so S3-compatible providers which do not implement it (R2, Minio, Ceph...) work with this service.
+* The checksum ActiveStorage computes is of the plaintext, so it cannot be sent to S3 as a `Content-MD5` - S3 would be checking it against our ciphertext. The plaintext is digested as it streams into the cipher instead, and the object is deleted if the digests differ.
+* `#compose` downloads, decrypts and re-encrypts, as it does for GCS: the bucket cannot splice objects it cannot read.
+
+Note what streaming decryption can and cannot promise, since it is easy to assume more. GCM verifies its authentication tag only after the whole ciphertext has been read, so a `download` which yields chunks to a block will have yielded tampered plaintext before it raises, and `download_chunk` reads a range with no authentication at all (see the note on random access below). If you need the guarantee that no altered byte reaches a caller, read the blob with the non-block form of `download`, which buffers and raises before it returns anything.
+
 ### EncryptedDiskSevice - Filesystem
 
 Can be used instead of the cloud services in development, or on the server if desired. The service will use AES-256-GCM encryption, with a way to switch to a different/more modern encryption scheme in the future.
diff --git a/lib/active_storage/service/client_side_encrypted_s3_service.rb b/lib/active_storage/service/client_side_encrypted_s3_service.rb
new file mode 100644
index 0000000..cc3bd15
--- /dev/null
+++ b/lib/active_storage/service/client_side_encrypted_s3_service.rb
@@ -0,0 +1,10 @@
+# frozen_string_literal: true
+
+# Needed so that Rails can find our service definition. It will perform the following
+# steps. Given a "ClientSideEncryptedS3" value of the `service:` key in the YAML, it will:
+#
+# * Force-require a file at "active_storage/service/client_side_encrypted_s3", from any path on the $LOAD_PATH
+# * Instantiate a class called "ActiveStorage::Service::ClientSideEncryptedS3Service"
+require_relative "../../active_storage_encryption"
+class ActiveStorage::Service::ClientSideEncryptedS3Service < ActiveStorageEncryption::ClientSideEncryptedS3Service
+end
diff --git a/lib/active_storage_encryption.rb b/lib/active_storage_encryption.rb
index 8c0819d..38cc9e6 100644
--- a/lib/active_storage_encryption.rb
+++ b/lib/active_storage_encryption.rb
@@ -10,6 +10,7 @@ module ActiveStorageEncryption
   autoload :EncryptedDiskService, __dir__ + "/active_storage_encryption/encrypted_disk_service.rb"
   autoload :EncryptedMirrorService, __dir__ + "/active_storage_encryption/encrypted_mirror_service.rb"
   autoload :EncryptedS3Service, __dir__ + "/active_storage_encryption/encrypted_s3_service.rb"
+  autoload :ClientSideEncryptedS3Service, __dir__ + "/active_storage_encryption/client_side_encrypted_s3_service.rb"
   autoload :EncryptedGCSService, __dir__ + "/active_storage_encryption/encrypted_gcs_service.rb"
   autoload :Overrides, __dir__ + "/active_storage_encryption/overrides.rb"
 
@@ -19,6 +20,12 @@ class IncorrectEncryptionKey < ArgumentError
   class StreamingDisabled < ArgumentError
   end
 
+  # Raised when an object in a bucket does not carry the header a client-side encrypting
+  # service writes ahead of its ciphertext - it was written by another service, or is not
+  # encrypted at all.
+  class UnknownCiphertextFormat < StandardError
+  end
+
   class StreamingTokenInvalidOrExpired < ActiveSupport::MessageEncryptor::InvalidMessage
   end
 
diff --git a/lib/active_storage_encryption/client_side_encrypted_s3_service.rb b/lib/active_storage_encryption/client_side_encrypted_s3_service.rb
new file mode 100644
index 0000000..979b1a4
--- /dev/null
+++ b/lib/active_storage_encryption/client_side_encrypted_s3_service.rb
@@ -0,0 +1,277 @@
+# frozen_string_literal: true
+
+require "block_cipher_kit"
+require "active_storage/service/s3_service"
+
+module ActiveStorageEncryption
+  # Stores ActiveStorage blobs on S3-compatible storage, encrypting them inside the application
+  # process. Where `EncryptedS3Service` asks the storage provider to encrypt for us (SSE-C, so the
+  # provider does hold the plaintext at the moment of the write), this service hands the provider
+  # ciphertext and nothing else. The bytes on the wire and the bytes at rest are both encrypted
+  # with a key the provider never sees.
+  #
+  # It reuses the encryption scheme of `EncryptedDiskService` rather than introducing a second one.
+  # That is possible because the schemes only need an IO to read from - see `SeekableObjectIO`,
+  # which turns ranged GET requests into the `read`/`seek`/`pos`/`size` they expect. So random
+  # access - the property that lets a large video be scrubbed rather than downloaded whole -
+  # survives the move from a local disk to a bucket.
+  #
+  # Configure it like so:
+  #
+  #   encrypted_s3:
+  #     service: ClientSideEncryptedS3
+  #     bucket: my-bucket
+  #     region: eu-central-1
+  #     private_url_policy: stream
+  #
+  # Two things behave differently from the SSE-C service, both of them consequences of the app
+  # being the only party that can encrypt or decrypt:
+  #
+  # * A presigned URL can only ever yield ciphertext, so `private_url_policy: require_headers`
+  #   is refused at configuration time. Use `stream` (through the controller in this gem, or
+  #   through one of your own) or `disable`.
+  # * Direct uploads from a browser cannot go to the bucket, since the browser has no key. They
+  #   are routed to `EncryptedBlobsController` instead, exactly as `EncryptedDiskService` does:
+  #   the app receives the plaintext, encrypts it, and puts it in the bucket.
+  #
+  # Bear in mind what streaming decryption can and cannot promise. GCM verifies its authentication
+  # tag only once the whole ciphertext has been read, so a `download` that yields chunks to a block
+  # will have yielded tampered plaintext before it raises, and `download_chunk` reads a range
+  # without any authentication at all (see `V2Scheme`). If you need a guarantee that no altered
+  # byte can reach a caller, read the whole blob with the non-block form of `download`, which
+  # buffers and raises before returning anything.
+  class ClientSideEncryptedS3Service < ActiveStorage::Service::S3Service
+    include ActiveStorageEncryption::PrivateUrlPolicy
+
+    autoload :SeekableObjectIO, __dir__ + "/client_side_encrypted_s3_service/seekable_object_io.rb"
+
+    # Unlike a file on disk, an object in a bucket has no filename we can hang a scheme version
+    # on, and asking the bucket for object metadata costs a request. So the ciphertext names its
+    # own format: a magic string (which also tells us an object was written by this service at
+    # all, rather than by a stock S3 service) followed by one byte of scheme version.
+    CIPHERTEXT_MAGIC_BYTES = "ASEC"
+    CIPHERTEXT_HEADER_BYTE_SIZE = CIPHERTEXT_MAGIC_BYTES.bytesize + 1
+
+    # The version byte matches the scheme version of EncryptedDiskService, so that the same number
+    # always means the same bytes on the wire regardless of where a blob is stored.
+    SCHEME_VERSIONS = {
+      2 => "ActiveStorageEncryption::EncryptedDiskService::V2Scheme"
+    }
+    CURRENT_SCHEME_VERSION = 2
+
+    # This lets the Blob encryption key methods know that this
+    # storage service _must_ use encryption
+    def encrypted? = true
+
+    def initialize(public: false, **options_for_s3_service_and_private_url_policy)
+      raise ArgumentError, "encrypted files cannot be served via a public URL or a CDN" if public
+      super
+      if private_url_policy == :require_headers
+        raise ArgumentError, "private_url_policy: require_headers is not available for #{self.class.name}, " \
+          "because a presigned URL would serve ciphertext which the client has no key to decrypt"
+      end
+    end
+
+    def service_name
+      # ActiveStorage::Service::DiskService => Disk
+      # Overridden because in Rails 8 this is "self.class.name.split("::").third.remove("Service")"
+      self.class.name.split("::").last.remove("Service")
+    end
+
+    def upload(key, io, encryption_key:, checksum: nil, filename: nil, content_type: nil, disposition: nil, custom_metadata: {}, **)
+      instrument :upload, key: key, checksum: checksum do
+        # The checksum ActiveStorage gives us is of the plaintext, so it cannot be handed to S3 as
+        # a Content-MD5 - S3 would be checking it against our ciphertext. We digest the plaintext
+        # as it streams into the cipher instead, which verifies the same thing without reading the
+        # object back out of the bucket afterwards.
+        plaintext_io = checksum ? PlaintextChecksumIO.new(io) : io
+        content_disposition = content_disposition_with(type: disposition, filename: filename) if disposition && filename
+
+        # Build the scheme before opening the upload. An unusable encryption key must raise here,
+        # where the caller can see the reason, rather than inside the block - the SDK would bury
+        # it in a MultipartUploadError, and we would have left a dangling upload behind.
+        scheme = scheme_for(CURRENT_SCHEME_VERSION, encryption_key)
+
+        object_for(key).upload_stream(
+          content_type: content_type,
+          content_disposition: content_disposition,
+          part_size: MINIMUM_UPLOAD_PART_SIZE,
+          metadata: custom_metadata,
+          **upload_options
+        ) do |ciphertext_io|
+          ciphertext_io.binmode
+          ciphertext_io.write(ciphertext_header)
+          scheme.streaming_encrypt(into_ciphertext_io: ciphertext_io, from_plaintext_io: plaintext_io)
+        end
+
+        ensure_integrity_of(key, checksum, plaintext_io) if checksum
+      end
+    end
+
+    def download(key, encryption_key:, &block)
+      if block_given?
+        instrument :streaming_download, key: key do
+          stream(key, encryption_key, &block)
+        end
+      else
+        instrument :download, key: key do
+          (+"").b.tap do |buf|
+            stream(key, encryption_key) { |chunk| buf << chunk }
+          end
+        end
+      end
+    end
+
+    def download_chunk(key, range, encryption_key:)
+      instrument :download_chunk, key: key, range: range do
+        open_ciphertext(key, encryption_key) do |ciphertext_io, scheme|
+          scheme.decrypt_range(from_ciphertext_io: ciphertext_io, range: inclusive(range))
+        end
+      end
+    end
+
+    def compose(source_keys, destination_key, source_encryption_keys:, encryption_key:, filename: nil, content_type: nil, disposition: nil, custom_metadata: {})
+      if source_keys.length != source_encryption_keys.length
+        raise ArgumentError, "With #{source_keys.length} keys to compose there should be exactly as many source_encryption_keys, but got #{source_encryption_keys.length}"
+      end
+      content_disposition = content_disposition_with(type: disposition, filename: filename) if disposition && filename
+      scheme = scheme_for(CURRENT_SCHEME_VERSION, encryption_key)
+
+      object_for(destination_key).upload_stream(
+        content_type: content_type,
+        content_disposition: content_disposition,
+        part_size: MINIMUM_UPLOAD_PART_SIZE,
+        metadata: custom_metadata,
+        **upload_options
+      ) do |ciphertext_io|
+        ciphertext_io.binmode
+        ciphertext_io.write(ciphertext_header)
+        scheme.streaming_encrypt(into_ciphertext_io: ciphertext_io) do |plaintext_writable|
+          source_keys.zip(source_encryption_keys).each do |(source_key, source_encryption_key)|
+            stream(source_key, source_encryption_key) { |chunk| plaintext_writable.write(chunk) }
+          end
+        end
+      end
+    end
+
+    def url_for_direct_upload(key, expires_in:, content_type:, content_length:, checksum:, encryption_key:, custom_metadata: {})
+      # A browser has no encryption key, so a PUT straight to the bucket would store plaintext.
+      # The upload goes to this gem's own controller instead, which encrypts it on the way in.
+      instrument :url, key: key do |payload|
+        upload_token = ActiveStorage.verifier.generate(
+          {
+            key: key,
+            content_type: content_type,
+            content_length: content_length,
+            encryption_key_sha256: Digest::SHA256.base64digest(encryption_key),
+            checksum: checksum,
+            service_name: name
+          },
+          expires_in: expires_in,
+          purpose: :encrypted_put
+        )
+
+        # Unlike the DiskService, an S3 service has no url_options of its own to build absolute
+        # URLs from - the ones ActiveStorage sets for the current request are what we have.
+        url_options = ActiveStorage::Current.url_options
+        raise ArgumentError, "Cannot generate a direct upload URL because ActiveStorage::Current.url_options is not set" if url_options.blank?
+
+        url_helpers = ActiveStorageEncryption::Engine.routes.url_helpers
+        url_helpers.encrypted_blob_put_url(upload_token, **url_options).tap do |generated_url|
+          payload[:url] = generated_url
+        end
+      end
+    end
+
+    def headers_for_direct_upload(key, content_type:, encryption_key:, checksum:, **)
+      {
+        "Content-Type" => content_type,
+        "x-active-storage-encryption-key" => Base64.strict_encode64(encryption_key),
+        "content-md5" => checksum
+      }
+    end
+
+    def headers_for_private_download(key, **)
+      # Nothing to send: the bytes in the bucket are of no use without the key, and the key
+      # never leaves the application.
+      {}
+    end
+
+    private
+
+    def ciphertext_header
+      (CIPHERTEXT_MAGIC_BYTES + CURRENT_SCHEME_VERSION.chr).b
+    end
+
+    def scheme_for(version, encryption_key)
+      scheme_class_name = SCHEME_VERSIONS.fetch(version) do
+        raise ActiveStorageEncryption::UnknownCiphertextFormat, "Unknown encryption scheme version #{version.inspect}"
+      end
+      Object.const_get(scheme_class_name).new(encryption_key.b)
+    end
+
+    # Opens the object, reads the header which tells us how the ciphertext after it was written,
+    # and yields an IO positioned at the start of that ciphertext together with the matching scheme.
+    def open_ciphertext(key, encryption_key)
+      ciphertext_io = SeekableObjectIO.new(object_for(key))
+      header = ciphertext_io.read(CIPHERTEXT_HEADER_BYTE_SIZE)
+      if header.nil? || header.byteslice(0, CIPHERTEXT_MAGIC_BYTES.bytesize) != CIPHERTEXT_MAGIC_BYTES
+        raise ActiveStorageEncryption::UnknownCiphertextFormat,
+          "Object #{key.inspect} in #{name} was not written by #{self.class.name} (no #{CIPHERTEXT_MAGIC_BYTES} header)"
+      end
+
+      scheme = scheme_for(header.getbyte(CIPHERTEXT_MAGIC_BYTES.bytesize), encryption_key)
+      yield ciphertext_io.rebase(CIPHERTEXT_HEADER_BYTE_SIZE), scheme
+    rescue Aws::S3::Errors::NoSuchKey
+      raise ActiveStorage::FileNotFoundError
+    end
+
+    def stream(key, encryption_key, &blk)
+      open_ciphertext(key, encryption_key) do |ciphertext_io, scheme|
+        scheme.streaming_decrypt(from_ciphertext_io: ciphertext_io, &blk)
+      end
+    end
+
+    # ActiveStorage passes exclusive ranges in some places (`0...4.kilobytes` when identifying a
+    # blob) and inclusive ones in others, while the schemes only understand inclusive ranges.
+    def inclusive(range)
+      range.exclude_end? ? (range.begin..(range.end - 1)) : range
+    end
+
+    def ensure_integrity_of(key, checksum, plaintext_io)
+      return if plaintext_io.base64digest == checksum
+
+      delete key
+      raise ActiveStorage::IntegrityError
+    end
+
+    def private_url(key, **options)
+      # :require_headers is refused in the constructor, and :disable raises inside this call,
+      # so streaming through a controller is all that is left.
+      private_url_for_streaming_via_controller(key, **options)
+    end
+
+    def public_url(key, **)
+      raise "This should never be called"
+    end
+
+    # Passes plaintext through to the cipher while digesting it, so that the checksum
+    # ActiveStorage computed over the same bytes can be verified after the upload.
+    class PlaintextChecksumIO
+      def initialize(io)
+        @io = io
+        @digest = OpenSSL::Digest.new("MD5")
+      end
+
+      def read(n_bytes = nil, outbuf = nil)
+        bytes_read = outbuf ? @io.read(n_bytes, outbuf) : @io.read(n_bytes)
+        @digest << bytes_read if bytes_read
+        bytes_read
+      end
+
+      def base64digest
+        @digest.base64digest
+      end
+    end
+  end
+end
diff --git a/lib/active_storage_encryption/client_side_encrypted_s3_service/seekable_object_io.rb b/lib/active_storage_encryption/client_side_encrypted_s3_service/seekable_object_io.rb
new file mode 100644
index 0000000..ab1d006
--- /dev/null
+++ b/lib/active_storage_encryption/client_side_encrypted_s3_service/seekable_object_io.rb
@@ -0,0 +1,144 @@
+# frozen_string_literal: true
+
+# A read-only, seekable IO over an object in an S3-compatible bucket. It exists because the
+# encryption schemes in this gem are IO-agnostic: they need `read`, `pos`, `seek` and `size`,
+# and they do not care whether those are served from a local file or from ranged GET requests.
+# `EncryptedDiskService` hands them a `File`; this class is what lets a bucket take its place.
+#
+# Reads are buffered, because the schemes read in small increments (12 bytes of IV, 16 bytes of
+# auth tag, then cipher blocks) and one HTTP request per such read would be unusable. The buffer
+# window starts small and doubles - up to `maximum_read_ahead` - for as long as reads stay
+# sequential, and resets the moment a read seeks elsewhere. That way a 1-byte `download_chunk`
+# does not pull five megabytes, while a full download quickly reaches the large window it wants.
+class ActiveStorageEncryption::ClientSideEncryptedS3Service::SeekableObjectIO
+  INITIAL_READ_AHEAD_BYTES = 64 * 1024
+  MAXIMUM_READ_AHEAD_BYTES = 5 * 1024 * 1024
+
+  attr_reader :offset
+
+  # @param object[Aws::S3::Object] the object to read from. Only `get(range:)` and `content_length` get used.
+  # @param offset[Integer] the byte offset in the object which this IO presents as its own position 0
+  # @param initial_read_ahead[Integer] the size of the first ranged GET, and of any GET after a seek
+  # @param maximum_read_ahead[Integer] the largest ranged GET this IO will ever issue
+  def initialize(object, offset: 0, initial_read_ahead: INITIAL_READ_AHEAD_BYTES, maximum_read_ahead: MAXIMUM_READ_AHEAD_BYTES)
+    @object = object
+    @offset = offset
+    @initial_read_ahead = initial_read_ahead
+    @maximum_read_ahead = maximum_read_ahead
+    @read_ahead = initial_read_ahead
+    @absolute_pos = offset
+    @buffer = (+"").b
+    @buffer_starts_at = 0
+    @object_byte_size = nil
+  end
+
+  # Presents the bytes from `byte_offset` onwards as the start of this IO, and rewinds to it.
+  # This is how the service skips its own ciphertext header before handing the IO to a scheme -
+  # the scheme then computes its offsets from 0, as it would for a file containing only its own
+  # ciphertext. Any buffered bytes are kept, so skipping the header costs no extra request.
+  #
+  # @param byte_offset[Integer] offset in the object to present as position 0
+  # @return [self]
+  def rebase(byte_offset)
+    @offset = byte_offset
+    @absolute_pos = byte_offset
+    self
+  end
+
+  # @return [Integer] the number of readable bytes, excluding everything before the base offset
+  def size
+    object_byte_size - @offset
+  end
+
+  # @return [Integer] the read position, relative to the base offset
+  def pos
+    @absolute_pos - @offset
+  end
+
+  # @param to_offset[Integer] the offset to seek to, relative to the base offset
+  # @param whence[Integer] one of IO::SEEK_SET, IO::SEEK_CUR or IO::SEEK_END
+  # @return [Integer] 0, as IO#seek does
+  def seek(to_offset, whence = IO::SEEK_SET)
+    absolute_pos = case whence
+    when IO::SEEK_SET then @offset + to_offset
+    when IO::SEEK_CUR then @absolute_pos + to_offset
+    when IO::SEEK_END then @offset + size + to_offset
+    else raise ArgumentError, "Unsupported whence: #{whence.inspect}"
+    end
+    raise Errno::EINVAL, "Cannot seek before the start of the IO" if absolute_pos < @offset
+
+    @absolute_pos = absolute_pos
+    0
+  end
+
+  # Reads at most `n_bytes` bytes, following the semantics of IO#read: at EOF it returns
+  # `nil` for a non-zero read length, and an empty String when reading to EOF.
+  #
+  # @param n_bytes[Integer,nil] how many bytes to read, or nil to read until EOF
+  # @param outbuf[String,nil] a String to read into, as IO#read accepts
+  # @return [String,nil]
+  def read(n_bytes = nil, outbuf = nil)
+    raise ArgumentError, "Negative length #{n_bytes} given" if n_bytes && n_bytes < 0
+
+    read_bytes = (+"").b
+    while n_bytes.nil? || read_bytes.bytesize < n_bytes
+      fill_buffer_at(@absolute_pos) unless buffer_covers?(@absolute_pos)
+      wanted = n_bytes ? (n_bytes - read_bytes.bytesize) : @buffer.bytesize
+      available = @buffer.byteslice(@absolute_pos - @buffer_starts_at, wanted)
+      break if available.nil? || available.empty?
+
+      read_bytes << available
+      @absolute_pos += available.bytesize
+    end
+
+    exhausted = read_bytes.empty? && n_bytes && n_bytes > 0
+    return exhausted ? nil : read_bytes unless outbuf
+
+    outbuf.clear
+    exhausted ? nil : outbuf.replace(read_bytes)
+  end
+
+  private
+
+  def buffer_covers?(absolute_pos)
+    absolute_pos >= @buffer_starts_at && absolute_pos < (@buffer_starts_at + @buffer.bytesize)
+  end
+
+  def fill_buffer_at(absolute_pos)
+    @read_ahead = if sequential_with_buffer?(absolute_pos)
+      [@read_ahead * 2, @maximum_read_ahead].min
+    else
+      @initial_read_ahead
+    end
+
+    @buffer = (+"").b
+    @buffer_starts_at = absolute_pos
+    return if @object_byte_size && absolute_pos >= @object_byte_size
+
+    last_byte_offset = absolute_pos + @read_ahead - 1
+    last_byte_offset = [last_byte_offset, @object_byte_size - 1].min if @object_byte_size
+
+    response = @object.get(range: "bytes=#{absolute_pos}-#{last_byte_offset}")
+    @object_byte_size ||= total_byte_size_from(response.content_range)
+    @buffer = (response.body.read || "").b
+  rescue Aws::S3::Errors::InvalidRange
+    # The object turned out to be shorter than the offset we asked for. This can only happen while
+    # the size is still unknown, since we clamp the requested range once we do know it.
+    @object_byte_size ||= @object.content_length
+  end
+
+  def sequential_with_buffer?(absolute_pos)
+    @buffer.bytesize > 0 && absolute_pos == (@buffer_starts_at + @buffer.bytesize)
+  end
+
+  # The Content-Range response header of a ranged GET reads "bytes 0-63/1024", so a successful
+  # ranged read also tells us the size of the whole object. Learning it this way spares us a HEAD
+  # request, which matters because every scheme begins by reading the head of the ciphertext.
+  def total_byte_size_from(content_range)
+    content_range.to_s.split("/").last.to_i
+  end
+
+  def object_byte_size
+    @object_byte_size ||= @object.content_length
+  end
+end
diff --git a/test/lib/client_side_encrypted_s3_service_against_bucket_test.rb b/test/lib/client_side_encrypted_s3_service_against_bucket_test.rb
new file mode 100644
index 0000000..a7c0057
--- /dev/null
+++ b/test/lib/client_side_encrypted_s3_service_against_bucket_test.rb
@@ -0,0 +1,128 @@
+# frozen_string_literal: true
+
+require "test_helper"
+
+# The in-memory tests in client_side_encrypted_s3_service_test.rb prove the logic of the service.
+# These prove the integration: that a real bucket answers ranged GET requests the way this service
+# needs it to, that multipart uploads of files larger than one part reassemble correctly, and that
+# the round trip survives a real network. They are skipped unless credentials are in the ENV.
+#
+# Point them at an S3-compatible provider by setting S3_ENDPOINT as well - R2 for instance, which
+# is worth testing separately from AWS because it is a different implementation of the same API:
+#
+#   AWS_ACCESS_KEY_ID=... AWS_SECRET_ACCESS_KEY=... \
+#   S3_BUCKET=my-bucket S3_ENDPOINT=https://<account>.eu.r2.cloudflarestorage.com \
+#   bin/rails test test/lib/client_side_encrypted_s3_service_against_bucket_test.rb
+class ActiveStorageEncryption::ClientSideEncryptedS3ServiceAgainstBucketTest < ActiveSupport::TestCase
+  setup do
+    if ENV["AWS_ACCESS_KEY_ID"].blank? || ENV["AWS_SECRET_ACCESS_KEY"].blank? || ENV["S3_BUCKET"].blank?
+      skip "Set AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY and S3_BUCKET to test against a real bucket"
+    end
+
+    @service = ActiveStorageEncryption::ClientSideEncryptedS3Service.new(**config)
+    @service.name = "client_side_encrypted_s3"
+    @written_keys = []
+  end
+
+  teardown do
+    @written_keys&.each { |key| @service.delete(key) }
+  end
+
+  def config
+    {
+      access_key_id: ENV.fetch("AWS_ACCESS_KEY_ID"),
+      secret_access_key: ENV.fetch("AWS_SECRET_ACCESS_KEY"),
+      region: ENV.fetch("S3_REGION", "auto"),
+      bucket: ENV.fetch("S3_BUCKET")
+    }.tap do |options|
+      next if ENV["S3_ENDPOINT"].blank?
+
+      # S3-compatible providers need the endpoint named, and path style addressing. The checksum
+      # settings are there because R2 refuses the CRC32 checksum aws-sdk-s3 sends by default
+      # ("You can only specify one non-default checksum at a time").
+      options.merge!(
+        endpoint: ENV["S3_ENDPOINT"],
+        force_path_style: true,
+        request_checksum_calculation: "when_required",
+        response_checksum_validation: "when_required"
+      )
+    end
+  end
+
+  # The bucket is shared with other runs, so keys carry a per-run prefix
+  def key_for(name)
+    @run_id ||= SecureRandom.base36(10)
+    "#{@run_id}-#{name}".tap { |key| @written_keys << key }
+  end
+
+  def test_round_trip_of_an_object_spanning_several_upload_parts
+    key = key_for("large")
+    encryption_key = Random.bytes(68)
+    plaintext = Random.bytes(11.megabytes + 17) # More than two 5 MB parts
+
+    @service.upload(key, StringIO.new(plaintext), encryption_key: encryption_key, checksum: Digest::MD5.base64digest(plaintext))
+    assert @service.exist?(key)
+
+    readback = (+"").b
+    @service.download(key, encryption_key: encryption_key) { |chunk| readback << chunk }
+    assert_equal plaintext.bytesize, readback.bytesize
+    assert_equal Digest::SHA256.hexdigest(plaintext), Digest::SHA256.hexdigest(readback)
+  end
+
+  def test_ranged_reads_land_on_the_exact_plaintext_bytes
+    key = key_for("ranges")
+    encryption_key = Random.bytes(68)
+    plaintext = Random.bytes(6.megabytes + 1023)
+    @service.upload(key, StringIO.new(plaintext), encryption_key: encryption_key)
+
+    last_offset = plaintext.bytesize - 1
+    ranges = [
+      0..0,                       # single byte at the start
+      100..199,                   # inside the first read-ahead window
+      0...4.kilobytes,            # exclusive, as blob identification asks for it
+      (5.megabytes)..(5.megabytes + 511), # past a part boundary
+      last_offset..last_offset    # single byte at the very end
+    ]
+
+    ranges.each do |range|
+      assert_equal plaintext[range], @service.download_chunk(key, range, encryption_key: encryption_key), "range #{range} did not match"
+    end
+  end
+
+  def test_the_bucket_holds_ciphertext_and_nothing_else
+    key = key_for("ciphertext")
+    encryption_key = Random.bytes(68)
+    plaintext = "the deceased owned a house at 12 Example Street".b * 100
+    @service.upload(key, StringIO.new(plaintext), encryption_key: encryption_key)
+
+    # Read the object the way any holder of the bucket credentials would - without our key
+    stored = @service.client.bucket(config.fetch(:bucket)).object(key).get.body.read.b
+    refute_includes stored, "12 Example Street"
+    assert_equal "ASEC", stored.byteslice(0, 4)
+
+    assert_equal Digest::SHA256.hexdigest(plaintext), Digest::SHA256.hexdigest(@service.download(key, encryption_key: encryption_key))
+  end
+
+  def test_wrong_key_is_refused_and_missing_objects_are_reported
+    key = key_for("wrong-key")
+    encryption_key = Random.bytes(68)
+    @service.upload(key, StringIO.new(Random.bytes(2048)), encryption_key: encryption_key)
+
+    assert_raises(ActiveStorageEncryption::IncorrectEncryptionKey) do
+      @service.download_chunk(key, 0..0, encryption_key: Random.bytes(68))
+    end
+
+    assert_raises(ActiveStorage::FileNotFoundError) do
+      @service.download("#{key}-does-not-exist", encryption_key: encryption_key)
+    end
+  end
+
+  def test_delete_removes_the_object
+    key = key_for("deletable")
+    @service.upload(key, StringIO.new(Random.bytes(1024)), encryption_key: Random.bytes(68))
+    assert @service.exist?(key)
+
+    @service.delete(key)
+    refute @service.exist?(key)
+  end
+end
diff --git a/test/lib/client_side_encrypted_s3_service_test.rb b/test/lib/client_side_encrypted_s3_service_test.rb
new file mode 100644
index 0000000..8b67dc9
--- /dev/null
+++ b/test/lib/client_side_encrypted_s3_service_test.rb
@@ -0,0 +1,278 @@
+# frozen_string_literal: true
+
+require "test_helper"
+require_relative "../support/in_memory_s3"
+
+class ActiveStorageEncryption::ClientSideEncryptedS3ServiceTest < ActiveSupport::TestCase
+  setup do
+    @service = ActiveStorageEncryption::ClientSideEncryptedS3Service.new(bucket: "test-bucket", region: "eu-central-1", stub_responses: true)
+    @service.name = "client_side_encrypted_s3" # Needed for the controllers and service lookup
+    @bucket = InMemoryS3.new(@service.client.client)
+
+    ActiveStorage::Current.url_options = {host: "www.example.com", protocol: "https"}
+  end
+
+  def test_encrypted_question_method
+    assert @service.encrypted?
+  end
+
+  def test_refuses_a_policy_which_would_hand_out_ciphertext
+    error = assert_raises(ArgumentError) do
+      ActiveStorageEncryption::ClientSideEncryptedS3Service.new(bucket: "test-bucket", region: "eu-central-1", stub_responses: true, private_url_policy: :require_headers)
+    end
+    assert_includes error.message, "require_headers"
+  end
+
+  def test_refuses_to_be_public
+    assert_raises(ArgumentError) do
+      ActiveStorageEncryption::ClientSideEncryptedS3Service.new(bucket: "test-bucket", region: "eu-central-1", stub_responses: true, public: true)
+    end
+  end
+
+  def test_the_bucket_never_receives_the_plaintext
+    key = "key-1"
+    encryption_key = Random.bytes(68)
+    plaintext = "the deceased owned a house at 12 Example Street" + generate_random_binary_string
+
+    @service.upload(key, StringIO.new(plaintext), encryption_key: encryption_key)
+
+    stored = @bucket.objects.fetch(key)
+    refute_includes stored, "12 Example Street"
+    refute_equal Digest::SHA256.hexdigest(plaintext), Digest::SHA256.hexdigest(stored)
+    assert_equal "ASEC", stored.byteslice(0, 4)
+    assert_equal 2, stored.getbyte(4) # The scheme version, matching EncryptedDiskService's V2Scheme
+  end
+
+  def test_upload_then_download_using_the_correct_key
+    key = "key-1"
+    encryption_key = Random.bytes(68)
+    plaintext = generate_random_binary_string
+
+    @service.upload(key, StringIO.new(plaintext), encryption_key: encryption_key)
+    assert @service.exist?(key)
+
+    assert_equal Digest::SHA256.hexdigest(plaintext), Digest::SHA256.hexdigest(@service.download(key, encryption_key: encryption_key))
+
+    streamed = (+"").b
+    @service.download(key, encryption_key: encryption_key) { |chunk| streamed << chunk }
+    assert_equal Digest::SHA256.hexdigest(plaintext), Digest::SHA256.hexdigest(streamed)
+  end
+
+  def test_upload_then_download_using_a_key_of_arbitrary_length
+    key = "key-1"
+    encryption_key = Random.new(Minitest.seed).bytes(128)
+    plaintext = generate_random_binary_string
+
+    @service.upload(key, StringIO.new(plaintext), encryption_key: encryption_key)
+    assert_equal Digest::SHA256.hexdigest(plaintext), Digest::SHA256.hexdigest(@service.download(key, encryption_key: encryption_key))
+  end
+
+  def test_upload_requires_a_key_of_sufficient_length
+    assert_raises(ArgumentError) do
+      @service.upload("key-1", StringIO.new(generate_random_binary_string), encryption_key: Random.bytes(12))
+    end
+  end
+
+  def test_download_with_an_incorrect_key_refuses_before_reading_the_body
+    key = "key-1"
+    correct_key, incorrect_key = Random.new(Minitest.seed).bytes(68), Random.new(Minitest.seed + 1).bytes(68)
+    plaintext = generate_random_binary_string
+    @service.upload(key, StringIO.new(plaintext), encryption_key: correct_key)
+
+    assert_raises(ActiveStorageEncryption::IncorrectEncryptionKey) do
+      @service.download(key, encryption_key: incorrect_key) { |chunk| flunk "Plaintext escaped: #{chunk.bytesize} bytes" }
+    end
+
+    assert_raises(ActiveStorageEncryption::IncorrectEncryptionKey) do
+      @service.download_chunk(key, 0..0, encryption_key: incorrect_key)
+    end
+  end
+
+  def test_random_access_reads_the_requested_plaintext_range
+    key = "key-1"
+    encryption_key = Random.bytes(68)
+    plaintext = generate_random_binary_string
+    @service.upload(key, StringIO.new(plaintext), encryption_key: encryption_key)
+
+    inclusive_range = 1234..2345
+    assert_equal plaintext[inclusive_range], @service.download_chunk(key, inclusive_range, encryption_key: encryption_key)
+
+    # Ranges arrive exclusive too - ActiveStorage identifies a blob by reading `0...4.kilobytes`
+    exclusive_range = 0...4.kilobytes
+    assert_equal plaintext[exclusive_range], @service.download_chunk(key, exclusive_range, encryption_key: encryption_key)
+
+    # And the single byte the proxy controller reads to check the key before it starts streaming
+    assert_equal plaintext[0..0], @service.download_chunk(key, 0..0, encryption_key: encryption_key)
+
+    last_byte_offset = plaintext.bytesize - 1
+    assert_equal plaintext[last_byte_offset..], @service.download_chunk(key, last_byte_offset..last_byte_offset, encryption_key: encryption_key)
+  end
+
+  def test_upload_with_a_checksum_verifies_the_plaintext_and_removes_the_object_when_it_differs
+    key = "key-1"
+    encryption_key = Random.bytes(68)
+    plaintext = generate_random_binary_string
+
+    assert_raises(ActiveStorage::IntegrityError) do
+      @service.upload(key, StringIO.new(plaintext), encryption_key: encryption_key, checksum: Digest::MD5.base64digest("something else entirely"))
+    end
+    refute @service.exist?(key)
+
+    assert_nothing_raised do
+      @service.upload(key, StringIO.new(plaintext), encryption_key: encryption_key, checksum: Digest::MD5.base64digest(plaintext))
+    end
+    assert @service.exist?(key)
+  end
+
+  def test_a_tampered_ciphertext_never_returns_plaintext_from_a_buffered_download
+    key = "key-1"
+    encryption_key = Random.bytes(68)
+    plaintext = generate_random_binary_string
+    @service.upload(key, StringIO.new(plaintext), encryption_key: encryption_key)
+
+    flip_one_ciphertext_byte(key)
+
+    assert_raises(OpenSSL::Cipher::CipherError) do
+      @service.download(key, encryption_key: encryption_key)
+    end
+  end
+
+  # The honest limitation of streaming AEAD, and the reason a framed scheme is worth having:
+  # GCM can only verify its tag once the whole ciphertext has been read, so a download which
+  # yields chunks has already handed some of them out by the time the tag fails. Callers who
+  # need the guarantee must use the buffered form above.
+  def test_a_tampered_ciphertext_raises_only_after_yielding_from_a_streaming_download
+    key = "key-1"
+    encryption_key = Random.bytes(68)
+    @service.upload(key, StringIO.new(generate_random_binary_string), encryption_key: encryption_key)
+
+    flip_one_ciphertext_byte(key)
+
+    yielded_chunks = 0
+    assert_raises(OpenSSL::Cipher::CipherError) do
+      @service.download(key, encryption_key: encryption_key) { |_chunk| yielded_chunks += 1 }
+    end
+    assert yielded_chunks > 0, "expected the streaming download to have yielded before failing"
+  end
+
+  def test_composes_objects
+    keys = ["key-1", "key-2"]
+    encryption_keys = [Random.bytes(68), Random.bytes(68)]
+    plaintexts = [generate_random_binary_string, generate_random_binary_string]
+
+    keys.zip(encryption_keys, plaintexts).each do |(key, encryption_key, plaintext)|
+      @service.upload(key, StringIO.new(plaintext), encryption_key: encryption_key)
+    end
+
+    composed_key = "key-3"
+    composed_encryption_key = Random.bytes(68)
+    @service.compose(keys, composed_key, source_encryption_keys: encryption_keys, encryption_key: composed_encryption_key)
+
+    readback = @service.download(composed_key, encryption_key: composed_encryption_key)
+    assert_equal Digest::SHA256.hexdigest(plaintexts.join), Digest::SHA256.hexdigest(readback)
+  end
+
+  def test_compose_refuses_a_mismatched_number_of_keys
+    assert_raises(ArgumentError) do
+      @service.compose(["key-1", "key-2"], "key-3", source_encryption_keys: [Random.bytes(68)], encryption_key: Random.bytes(68))
+    end
+  end
+
+  def test_delete
+    key = "key-1"
+    encryption_key = Random.bytes(68)
+    @service.upload(key, StringIO.new(generate_random_binary_string), encryption_key: encryption_key)
+
+    @service.delete(key)
+    refute @service.exist?(key)
+  end
+
+  def test_downloading_a_missing_object
+    assert_raises(ActiveStorage::FileNotFoundError) do
+      @service.download("no-such-key", encryption_key: Random.bytes(68))
+    end
+    assert_raises(ActiveStorage::FileNotFoundError) do
+      @service.download_chunk("no-such-key", 0..10, encryption_key: Random.bytes(68))
+    end
+  end
+
+  def test_refuses_an_object_which_was_not_written_by_this_service
+    @bucket.objects["plaintext-key"] = "this was put here by a stock S3 service"
+
+    assert_raises(ActiveStorageEncryption::UnknownCiphertextFormat) do
+      @service.download("plaintext-key", encryption_key: Random.bytes(68))
+    end
+  end
+
+  def test_generates_a_streaming_url_and_refuses_one_when_disabled
+    filename = ActiveStorage::Filename.new("temp.bin")
+
+    @service.private_url_policy = :stream
+    url = @service.url("key-1", blob_byte_size: 14, filename: filename, content_type: "binary/octet-stream", disposition: "inline", encryption_key: Random.bytes(32), expires_in: 10.seconds)
+    assert_includes url, "/active-storage-encryption/blob/"
+
+    @service.private_url_policy = :disable
+    assert_raises(ActiveStorageEncryption::StreamingDisabled) do
+      @service.url("key-1", blob_byte_size: 14, filename: filename, content_type: "binary/octet-stream", disposition: "inline", encryption_key: Random.bytes(32), expires_in: 10.seconds)
+    end
+  end
+
+  def test_direct_uploads_are_routed_through_the_application
+    key = "key-1"
+    encryption_key = Random.bytes(68)
+    plaintext = generate_random_binary_string
+    checksum = Digest::MD5.base64digest(plaintext)
+
+    # A browser has no key, so the PUT cannot go to the bucket - it goes to the controller in
+    # this gem, which encrypts what it receives before it reaches the bucket.
+    url = @service.url_for_direct_upload(key, expires_in: 60.seconds, content_type: "binary/octet-stream", content_length: plaintext.bytesize, checksum: checksum, encryption_key: encryption_key)
+    assert_includes url, "/active-storage-encryption/blob/"
+
+    headers = @service.headers_for_direct_upload(key, content_type: "binary/octet-stream", encryption_key: encryption_key, checksum: checksum)
+    assert_equal Base64.strict_encode64(encryption_key), headers["x-active-storage-encryption-key"]
+    assert_equal checksum, headers["content-md5"]
+
+    previous_service = ActiveStorage::Blob.service
+    ActiveStorage::Blob.service = @service # So that the controller can find it
+    uri = URI.parse(url)
+    rack_env = {
+      "SCRIPT_NAME" => "",
+      "PATH_INFO" => uri.path,
+      "QUERY_STRING" => uri.query,
+      "REQUEST_METHOD" => "PUT",
+      "SERVER_NAME" => uri.host,
+      "rack.input" => StringIO.new(plaintext),
+      "CONTENT_LENGTH" => plaintext.bytesize.to_s(10),
+      "CONTENT_TYPE" => "binary/octet-stream",
+      "HTTP_X_ACTIVE_STORAGE_ENCRYPTION_KEY" => Base64.strict_encode64(encryption_key),
+      "HTTP_CONTENT_MD5" => checksum,
+      "action_dispatch.request.parameters" => {"token" => uri.path.split("/").last}
+    }
+    status, _headers, _body = ActiveStorageEncryption::EncryptedBlobsController.action(:update).call(rack_env)
+    assert_equal 204, status
+
+    assert_equal Digest::SHA256.hexdigest(plaintext), Digest::SHA256.hexdigest(@service.download(key, encryption_key: encryption_key))
+  ensure
+    ActiveStorage::Blob.service = previous_service
+  end
+
+  def test_headers_for_private_download_carry_no_key
+    assert_empty @service.headers_for_private_download("key-1", encryption_key: Random.bytes(68))
+  end
+
+  def test_service_name
+    assert_equal "ClientSideEncryptedS3", @service.service_name
+  end
+
+  private
+
+  def flip_one_ciphertext_byte(key)
+    stored = @bucket.objects.fetch(key)
+    offset = stored.bytesize / 2
+    stored.setbyte(offset, stored.getbyte(offset) ^ 0xFF)
+  end
+
+  def generate_random_binary_string(size = 17.kilobytes + 13)
+    Random.bytes(size)
+  end
+end
diff --git a/test/lib/seekable_object_io_test.rb b/test/lib/seekable_object_io_test.rb
new file mode 100644
index 0000000..7057729
--- /dev/null
+++ b/test/lib/seekable_object_io_test.rb
@@ -0,0 +1,179 @@
+# frozen_string_literal: true
+
+require "test_helper"
+
+class ActiveStorageEncryption::SeekableObjectIOTest < ActiveSupport::TestCase
+  SeekableObjectIO = ActiveStorageEncryption::ClientSideEncryptedS3Service::SeekableObjectIO
+
+  # Stands in for an Aws::S3::Object. The IO only uses `get(range:)` and `content_length`, and
+  # this double records what it was asked for so that the tests can assert on the number and the
+  # size of the requests - the buffering is the entire reason this class exists.
+  class FakeObject
+    Response = Struct.new(:body, :content_range)
+
+    attr_reader :requested_ranges, :n_head_requests
+
+    def initialize(bytes)
+      @bytes = bytes.b
+      @requested_ranges = []
+      @n_head_requests = 0
+    end
+
+    def content_length
+      @n_head_requests += 1
+      @bytes.bytesize
+    end
+
+    def get(range:)
+      first, last = range.match(/\Abytes=(\d+)-(\d+)\z/).captures.map(&:to_i)
+      raise Aws::S3::Errors::InvalidRange.new(nil, "Range Not Satisfiable") if first >= @bytes.bytesize
+
+      last = [last, @bytes.bytesize - 1].min
+      @requested_ranges << (first..last)
+      Response.new(StringIO.new(@bytes.byteslice(first, last - first + 1)), "bytes #{first}-#{last}/#{@bytes.bytesize}")
+    end
+  end
+
+  setup do
+    @bytes = Random.new(Minitest.seed).bytes(100 * 1024)
+    @object = FakeObject.new(@bytes)
+  end
+
+  def test_reads_the_entire_object_sequentially
+    io = SeekableObjectIO.new(@object)
+    assert_equal @bytes, io.read
+    assert_equal @bytes.bytesize, io.pos
+  end
+
+  def test_reads_in_increments_and_reassembles_the_object
+    io = SeekableObjectIO.new(@object)
+    reassembled = (+"").b
+    while (chunk = io.read(1024))
+      reassembled << chunk
+    end
+    assert_equal Digest::SHA256.hexdigest(@bytes), Digest::SHA256.hexdigest(reassembled)
+  end
+
+  def test_read_at_eof_follows_io_semantics
+    io = SeekableObjectIO.new(@object)
+    io.seek(@bytes.bytesize)
+
+    assert_nil io.read(1)
+    assert_equal "", io.read
+    assert_equal "", io.read(0)
+  end
+
+  def test_read_into_a_buffer
+    io = SeekableObjectIO.new(@object)
+    buffer = (+"x" * 999)
+
+    assert_same buffer, io.read(64, buffer)
+    assert_equal @bytes.byteslice(0, 64), buffer
+
+    io.seek(0, IO::SEEK_END)
+    assert_nil io.read(1, buffer)
+    assert_equal "", buffer
+  end
+
+  def test_seeks_from_every_whence
+    io = SeekableObjectIO.new(@object)
+
+    io.seek(10)
+    assert_equal @bytes.byteslice(10, 4), io.read(4)
+
+    io.seek(6, IO::SEEK_CUR)
+    assert_equal @bytes.byteslice(20, 4), io.read(4)
+
+    io.seek(-8, IO::SEEK_END)
+    assert_equal @bytes.byteslice(@bytes.bytesize - 8, 8), io.read(8)
+
+    assert_raises(Errno::EINVAL) { io.seek(-1) }
+  end
+
+  def test_random_access_reads_only_what_was_asked_for
+    io = SeekableObjectIO.new(@object, initial_read_ahead: 1024, maximum_read_ahead: 8 * 1024)
+
+    io.seek(50_000)
+    assert_equal @bytes.byteslice(50_000, 10), io.read(10)
+
+    assert_equal 1, @object.requested_ranges.length
+    assert_equal (50_000..51_023), @object.requested_ranges.first
+  end
+
+  def test_read_ahead_grows_while_sequential_and_resets_after_a_seek
+    io = SeekableObjectIO.new(@object, initial_read_ahead: 1024, maximum_read_ahead: 4096)
+
+    io.read(4096) # Spans several windows, each one larger than the last
+    assert_equal [1024, 2048, 4096], @object.requested_ranges.map(&:count).take(3)
+
+    io.seek(70_000)
+    io.read(1)
+    assert_equal 1024, @object.requested_ranges.last.count
+  end
+
+  def test_learns_the_object_size_from_a_ranged_read_without_a_head_request
+    io = SeekableObjectIO.new(@object)
+
+    io.read(16)
+    assert_equal @bytes.bytesize, io.size
+    assert_equal 0, @object.n_head_requests
+  end
+
+  def test_asks_for_the_object_size_when_nothing_has_been_read_yet
+    io = SeekableObjectIO.new(@object)
+
+    assert_equal @bytes.bytesize, io.size
+    assert_equal 1, @object.n_head_requests
+  end
+
+  def test_size_of_an_object_shorter_than_the_first_read
+    short_object = FakeObject.new("hello")
+    io = SeekableObjectIO.new(short_object)
+
+    assert_equal "hello", io.read
+    assert_equal 5, io.size
+  end
+
+  def test_rebase_presents_the_bytes_after_the_offset
+    io = SeekableObjectIO.new(@object)
+    io.read(5) # Reads a header, as the service does
+
+    io.rebase(5)
+    assert_equal 0, io.pos
+    assert_equal @bytes.bytesize - 5, io.size
+    assert_equal @bytes.byteslice(5, 4), io.read(4)
+
+    io.seek(0)
+    assert_equal @bytes.byteslice(5, 4), io.read(4)
+  end
+
+  def test_rebase_keeps_the_buffer_it_already_has
+    io = SeekableObjectIO.new(@object)
+    io.read(5)
+    n_requests_after_header = @object.requested_ranges.length
+
+    io.rebase(5)
+    io.read(4)
+    assert_equal n_requests_after_header, @object.requested_ranges.length
+  end
+
+  # The point of this IO is that the encryption schemes in this gem do not know or care where
+  # their ciphertext comes from. This is that claim, tested: the scheme reads and seeks through
+  # ranged requests exactly as it would through a file.
+  def test_a_scheme_can_decrypt_and_seek_through_this_io
+    encryption_key = Random.new(Minitest.seed).bytes(32)
+    plaintext = Random.new(Minitest.seed).bytes(64 * 1024 + 7)
+    scheme = ActiveStorageEncryption::EncryptedDiskService::V2Scheme.new(encryption_key)
+
+    ciphertext = StringIO.new((+"").b)
+    scheme.streaming_encrypt(into_ciphertext_io: ciphertext, from_plaintext_io: StringIO.new(plaintext))
+    object = FakeObject.new(ciphertext.string)
+
+    readback = (+"").b
+    scheme.streaming_decrypt(from_ciphertext_io: SeekableObjectIO.new(object)) { |chunk| readback << chunk }
+    assert_equal Digest::SHA256.hexdigest(plaintext), Digest::SHA256.hexdigest(readback)
+
+    range = 40_000..40_511
+    assert_equal plaintext[range], scheme.decrypt_range(from_ciphertext_io: SeekableObjectIO.new(object), range: range)
+  end
+end
diff --git a/test/support/in_memory_s3.rb b/test/support/in_memory_s3.rb
new file mode 100644
index 0000000..6619690
--- /dev/null
+++ b/test/support/in_memory_s3.rb
@@ -0,0 +1,99 @@
+# frozen_string_literal: true
+
+# A small in-memory stand-in for an S3 bucket, installed into an Aws::S3::Client through the
+# SDK's own response stubbing. It exists so that the client-side encrypting service can be tested
+# everywhere, rather than only on a machine holding credentials for a real bucket: encryption,
+# ranged decryption and tamper detection are properties of our code, not of the storage provider.
+#
+# It implements only what the service asks of a bucket, and it implements ranged GET requests
+# faithfully, because that is the part the service leans on hardest. Tests against a real bucket
+# still exist alongside it - this double proves the logic, they prove the integration.
+class InMemoryS3
+  attr_reader :objects
+
+  def initialize(client)
+    @objects = {}
+    @parts_per_upload = {}
+    @n_uploads = 0
+    install_into(client)
+  end
+
+  private
+
+  def install_into(client)
+    client.stub_responses(:put_object, ->(context) {
+      @objects[context.params[:key]] = read_body(context.params[:body])
+      {}
+    })
+
+    client.stub_responses(:create_multipart_upload, ->(_context) {
+      upload_id = "upload-#{@n_uploads += 1}"
+      @parts_per_upload[upload_id] = {}
+      {upload_id: upload_id}
+    })
+
+    client.stub_responses(:upload_part, ->(context) {
+      params = context.params
+      @parts_per_upload.fetch(params[:upload_id])[params[:part_number]] = read_body(params[:body])
+      {etag: "\"part-#{params[:part_number]}\""}
+    })
+
+    client.stub_responses(:complete_multipart_upload, ->(context) {
+      params = context.params
+      parts = @parts_per_upload.delete(params[:upload_id]) || {}
+      @objects[params[:key]] = parts.keys.sort.map { |part_number| parts[part_number] }.join.b
+      {}
+    })
+
+    client.stub_responses(:abort_multipart_upload, ->(context) {
+      @parts_per_upload.delete(context.params[:upload_id])
+      {}
+    })
+
+    client.stub_responses(:head_object, ->(context) {
+      bytes = @objects[context.params[:key]]
+      # A raw 404, rather than an error class: Aws::S3::Object#exists? runs a waiter which
+      # decides on the HTTP status, and only a real 404 means "no such object" to it.
+      next {status_code: 404, headers: {}, body: ""} unless bytes
+
+      {content_length: bytes.bytesize}
+    })
+
+    client.stub_responses(:delete_object, ->(context) {
+      @objects.delete(context.params[:key])
+      {}
+    })
+
+    client.stub_responses(:get_object, ->(context) {
+      params = context.params
+      bytes = @objects[params[:key]]
+      next no_such_key_response unless bytes
+      next {body: bytes, content_length: bytes.bytesize} unless params[:range]
+
+      first, last = params[:range].match(/\Abytes=(\d+)-(\d+)\z/).captures.map(&:to_i)
+      next Aws::S3::Errors::InvalidRange.new(nil, "Range Not Satisfiable") if first >= bytes.bytesize
+
+      last = [last, bytes.bytesize - 1].min
+      {
+        body: bytes.byteslice(first, last - first + 1),
+        content_length: last - first + 1,
+        content_range: "bytes #{first}-#{last}/#{bytes.bytesize}"
+      }
+    })
+  end
+
+  def read_body(body)
+    body.is_a?(String) ? body.b : body.read.b
+  end
+
+  # S3 tells a missing key apart from other 404s through the error code in the response body,
+  # and the SDK turns that code into Aws::S3::Errors::NoSuchKey - which is what the service
+  # translates into ActiveStorage::FileNotFoundError.
+  def no_such_key_response
+    {
+      status_code: 404,
+      headers: {},
+      body: "<Error><Code>NoSuchKey</Code><Message>The specified key does not exist.</Message></Error>"
+    }
+  end
+end