Fix the S3 defects a 10-reviewer audit found in the SDK v2 port

**Silent data loss on multipart upload.** `uploadPart` was handed a `BoundedInputStream`
over a `FileInputStream`, which cannot be reset, so the SDK refused to retry a part after
any transient error ("Request cannot be retried, because the request stream could not be
reset"). The v1 SDK passed the file plus an offset and retried freely. The failure then
landed in a catch that logged and swallowed, `completeMultipartUpload` never ran, and
`put()` carried on setting size and checksum - DSpace recorded a bitstream that S3 does
not hold, with no error anywhere.

Now uses the public `AsyncRequestBody.fromFile(FileRequestBodyConfiguration)` with
`position`/`numBytesToRead`, which is file-backed and therefore retryable; this also
removes the hand-rolled `openPart` helper and its executor. A failed upload now aborts the
multipart upload (parts were previously leaked and billed) and throws instead of returning
normally.

**The multipart loop was never executed more than once.** 50 MB part size against a 23-byte
fixture meant one iteration with offset 0, so the offset arithmetic and last-part handling
this port rewrote had no coverage at all. `uploadPartSizeBytes` is now settable and
`syncStoreUploadsInMultipleParts` uploads 5 MB + 4 KB across a real part boundary, verifying
the object byte for byte.

**A tripwire in `amazonClientBuilderBy` left all 21 S3 tests green.** Every test injects a
ready-made client, so `FunctionalUtils.getDefaultOrBuild` never called the supplier and
`endpointOverride`, `forcePathStyle`, `maxConcurrency` and the part sizes never ran - the
fork's headline delta, configurable `pathStyleAccessEnabled`, was verified only at its
getter. `clientBuilderAppliesEndpointAndPathStyle` builds a client through that method and
uses it against LocalStack. Verified it detects breakage: removing `endpointOverride` turns
it red, restoring it turns it green.

**Content-Disposition had two implementations and the newer one was wrong.** It escaped `"`
but not `\`, so a bitstream named `evil\` terminated the quoted string and swallowed
`filename*`; and it used `URLEncoder` directly, so a space arrived as `+`, which RFC 8187
reads literally - re-introducing the bug vanilla had just fixed. Extracted the correct
implementation to `org.dspace.util.ContentDispositionUtils`, now used by both call sites.

**Enabling direct downloads silently disabled inline preview.** The redirect hardcoded
`attachment`, discarding the disposition the allowlist and the 8 MB threshold had just
computed. `generatePresignedUrl` takes the caller's disposition; `BitstreamRestController`
passes the exact header the non-redirect path would have sent.

**Presigner region could diverge from the client's.** It hardcoded `us-east-1` while the
client leaves the region to the default provider chain when no credentials are configured,
so presigned URLs would be signed for the wrong region on an IAM role outside us-east-1. It
now mirrors `init()` branch for branch.

**A 403 was reported as "bucket absent".** `doesBucketExist` treated everything except
`NoSuchBucketException` as absent, so `init()` would try to create a bucket that already
exists; a least-privilege policy denies that and the assetstore comes up dead. Only 404 and
`NoSuchBucketException` now answer false. `doesObjectExist` likewise stops reporting a 403
or a timeout as "object does not exist" with the cause hidden at DEBUG.

**`s3ChecksumAlgorithm` was dead configuration.** `bitstore.xml` wires
`SyncS3BitStoreService`, which overrides `put()`, so the parent's `putObject` - the only
consumer - never ran. `uploadFluently` now applies it. Also `${assetstore.s3.endpoint:}` is
self-defaulting, matching vanilla.

Tests: S3BitStoreServiceIT 15, ClarinS3BitStoreServiceIT 8, S3DirectDownloadServiceTest 11,
ContentDispositionUtilsTest 9 - all 0 failures, checkstyle 0 violations. Maven does not
recompile edited sources reliably here, so every run used
`-Dmaven.compiler.useIncrementalCompilation=false`.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: milanmajchrak

dspace-api/src/main/java/org/dspace/storage/bitstore/S3BitStoreService.java

@@ -58,6 +58,7 @@
 import software.amazon.awssdk.services.s3.model.ChecksumAlgorithm;
 import software.amazon.awssdk.services.s3.model.HeadObjectResponse;
 import software.amazon.awssdk.services.s3.model.NoSuchBucketException;
+import software.amazon.awssdk.services.s3.model.S3Exception;
 
 /**
  * Asset store using Amazon's Simple Storage Service (S3).
@@ -262,11 +263,18 @@ public boolean doesBucketExist(String bucketName ) {
             s3AsyncClient.headBucket(r -> r.bucket(bucketName)).join();
             return true;
         } catch (CompletionException ce) {
-            if (!(ce.getCause() instanceof NoSuchBucketException)) {
-                log.error("headBucket(" + bucketName + ")", ce.getCause());
+            Throwable cause = ce.getCause();
+            if (cause instanceof NoSuchBucketException
+                    || (cause instanceof S3Exception
+                        && ((S3Exception) cause).statusCode() == HttpStatusCode.NOT_FOUND)) {
+                return false;
             }
 
-            return false;
+            // CLARIN: only a genuinely absent bucket may answer "false". Reporting a 403 as "absent" makes
+            // init() try to create a bucket that already exists, which a least-privilege policy denies -
+            // and the assetstore then comes up dead. The v1 SDK's doesBucketExistV2 drew the same line.
+            log.error("headBucket(" + bucketName + ") failed for a reason other than an absent bucket", cause);
+            throw ce;
         }
     }
 

dspace-api/src/main/java/org/dspace/storage/bitstore/S3DirectDownloadServiceImpl.java

@@ -9,22 +9,25 @@
 
 import java.io.IOException;
 import java.net.URI;
-import java.net.URLEncoder;
-import java.nio.charset.StandardCharsets;
 import java.time.Duration;
+import java.util.concurrent.CompletionException;
 
 import org.apache.commons.lang3.StringUtils;
 import org.dspace.services.ConfigurationService;
 import org.dspace.storage.bitstore.service.S3DirectDownloadService;
+import org.dspace.util.ContentDispositionUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
 import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
+import software.amazon.awssdk.http.HttpStatusCode;
 import software.amazon.awssdk.regions.Region;
 import software.amazon.awssdk.services.s3.S3AsyncClient;
 import software.amazon.awssdk.services.s3.S3Configuration;
 import software.amazon.awssdk.services.s3.model.GetObjectRequest;
+import software.amazon.awssdk.services.s3.model.NoSuchKeyException;
+import software.amazon.awssdk.services.s3.model.S3Exception;
 import software.amazon.awssdk.services.s3.presigner.S3Presigner;
 import software.amazon.awssdk.services.s3.presigner.model.GetObjectPresignRequest;
 
@@ -90,17 +93,28 @@ private S3Presigner buildPresigner() {
         }
         // otherwise fall back to the default provider chain (IAM role / environment), as S3BitStoreService does
 
-        // Mirror S3BitStoreService: an unset or unparseable region falls back to us-east-1
-        Region region = Region.US_EAST_1;
+        // Mirror S3BitStoreService exactly. With explicit credentials it falls back to us-east-1; without
+        // them it leaves the region unset so the default provider chain resolves it. Hardcoding us-east-1
+        // in the no-credentials case would sign against the wrong region on an IAM role outside us-east-1,
+        // and the signature would be rejected.
         String regionName = s3BitStoreService.getAwsRegionName();
-        if (StringUtils.isNotBlank(regionName)) {
+        if (StringUtils.isNotBlank(accessKey) && StringUtils.isNotBlank(secretKey)) {
+            Region region = Region.US_EAST_1;
+            if (StringUtils.isNotBlank(regionName)) {
+                try {
+                    region = Region.of(regionName);
+                } catch (IllegalArgumentException e) {
+                    log.warn("Invalid aws_region: {}", regionName);
+                }
+            }
+            builder.region(region);
+        } else if (StringUtils.isNotBlank(regionName)) {
             try {
-                region = Region.of(regionName);
+                builder.region(Region.of(regionName));
             } catch (IllegalArgumentException e) {
                 log.warn("Invalid aws_region: {}", regionName);
             }
         }
-        builder.region(region);
 
         String endpoint = s3BitStoreService.getEndpoint();
         if (StringUtils.isNotBlank(endpoint)) {
@@ -115,18 +129,40 @@ private S3Presigner buildPresigner() {
 
     /**
      * Whether the object is really there. A URL we cannot verify is never signed.
+     *
+     * The v1 SDK's `doesObjectExist` rethrew anything that was not a 404. There is no v2 equivalent, so
+     * this stays fail-closed - but a 403, an expired credential or a timeout is a misconfiguration the
+     * operator has to see, not a missing object, so it is logged at ERROR rather than DEBUG.
      */
     private boolean doesObjectExist(String bucket, String key) {
         try {
             s3Client.headObject(r -> r.bucket(bucket).key(key)).join();
             return true;
+        } catch (CompletionException e) {
+            Throwable cause = e.getCause();
+            if (cause instanceof NoSuchKeyException
+                    || (cause instanceof S3Exception
+                        && ((S3Exception) cause).statusCode() == HttpStatusCode.NOT_FOUND)) {
+                log.debug("headObject(bucket={}, key={}): object not found", bucket, key);
+            } else {
+                log.error("headObject(bucket={}, key={}) failed for a reason other than a missing object; "
+                        + "refusing to sign a URL", bucket, key, cause);
+            }
+            return false;
         } catch (Exception e) {
-            log.debug("headObject(bucket={}, key={}) failed", bucket, key, e);
+            log.error("headObject(bucket={}, key={}) failed; refusing to sign a URL", bucket, key, e);
             return false;
         }
     }
 
+    @Override
     public String generatePresignedUrl(String bucket, String key, int expirationSeconds, String desiredFilename) {
+        return generatePresignedUrl(bucket, key, expirationSeconds, desiredFilename, null);
+    }
+
+    @Override
+    public String generatePresignedUrl(String bucket, String key, int expirationSeconds, String desiredFilename,
+                                       String contentDispositionOverride) {
         if (desiredFilename == null) {
             log.error("Cannot generate presigned URL – desired filename is null");
             throw new IllegalArgumentException("Desired filename cannot be null");
@@ -140,14 +176,12 @@ public String generatePresignedUrl(String bucket, String key, int expirationSeco
             throw new IllegalArgumentException("Requested S3 object does not exist");
         }
 
-        // Add custom response header for filename - to download the file with the desired name
-        // Remove CRLF and quotes to prevent header injection
-        String safeName = desiredFilename.replaceAll("[\\r\\n\"]", "_");
-        // RFC-5987: percent-encode UTF-8, e.g. filename*=UTF-8''%E2%82%ACrates.txt
-        String encoded = URLEncoder.encode(desiredFilename, StandardCharsets.UTF_8);
-        String contentDisposition = String.format(
-                "attachment; filename=\"%s\"; filename*=UTF-8''%s",
-                safeName, encoded);
+        // Add custom response header for filename - to download the file with the desired name.
+        // The caller passes the disposition it would have served itself, so that redirecting to S3 does not
+        // silently turn an inline preview into a download; falling back to `attachment` is the safe default.
+        String contentDisposition = StringUtils.isNotBlank(contentDispositionOverride)
+                ? contentDispositionOverride
+                : ContentDispositionUtils.build(ContentDispositionUtils.ATTACHMENT, desiredFilename);
 
         try {
             GetObjectRequest getObjectRequest = GetObjectRequest.builder()

dspace-api/src/main/java/org/dspace/storage/bitstore/SyncS3BitStoreService.java

@@ -19,20 +19,19 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CompletionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 
 import org.apache.commons.io.IOUtils;
-import org.apache.commons.io.input.BoundedInputStream;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.dspace.content.Bitstream;
 import org.dspace.core.Utils;
 import org.dspace.services.ConfigurationService;
 import org.springframework.beans.factory.annotation.Autowired;
+import software.amazon.awssdk.core.FileRequestBodyConfiguration;
 import software.amazon.awssdk.core.async.AsyncRequestBody;
 import software.amazon.awssdk.core.exception.SdkException;
+import software.amazon.awssdk.services.s3.model.ChecksumAlgorithm;
 import software.amazon.awssdk.services.s3.model.CompletedMultipartUpload;
 import software.amazon.awssdk.services.s3.model.CompletedPart;
 import software.amazon.awssdk.services.s3.model.UploadPartResponse;
@@ -52,8 +51,12 @@ public class SyncS3BitStoreService extends S3BitStoreService {
 
     /**
      * The uploading file is divided into parts and each part is uploaded separately. The size of the part is 50 MB.
+     *
+     * Settable so the multipart path can actually be exercised by a test: with the 50 MB default and a small
+     * fixture the loop only ever runs once, which leaves the offset arithmetic and the last-part handling
+     * unverified. S3 requires every part except the last to be at least 5 MB.
      */
-    private static final long UPLOAD_FILE_PART_SIZE = 50 * 1024 * 1024; // 50 MB
+    private long uploadPartSizeBytes = 50 * 1024 * 1024; // 50 MB
 
     /**
      * Upload large file by parts - check the checksum of every part
@@ -189,8 +192,15 @@ private void createFileIfNotExist(File localFile) throws IOException {
      * @param scratchFile the file to upload
      */
     private void uploadFluently(String key, File scratchFile) {
-        s3AsyncClient.putObject(r -> r.bucket(getBucketName()).key(key),
-                AsyncRequestBody.fromFile(scratchFile)).join();
+        // `assetstore.s3.s3ChecksumAlgorithm` would otherwise be dead configuration for the fork:
+        // bitstore.xml wires this class, which overrides put(), so the parent's putObject never runs.
+        ChecksumAlgorithm algorithm = getS3ChecksumAlgorithm();
+        s3AsyncClient.putObject(r -> {
+            r.bucket(getBucketName()).key(key);
+            if (algorithm != null) {
+                r.checksumAlgorithm(algorithm);
+            }
+        }, AsyncRequestBody.fromFile(scratchFile)).join();
     }
 
     /**
@@ -217,28 +227,31 @@ private void uploadByParts(String key, File scratchFile) throws IOException {
         // Create a list to hold the ETags for individual parts
         List<CompletedPart> completedParts = new ArrayList<>();
 
-        ExecutorService executor = Executors.newSingleThreadExecutor();
         try {
             // Upload parts
             long fileLength = scratchFile.length();
             long remainingBytes = fileLength;
             int partNumber = 1;
 
             while (remainingBytes > 0) {
-                long bytesToUpload = Math.min(UPLOAD_FILE_PART_SIZE, remainingBytes);
+                long bytesToUpload = Math.min(uploadPartSizeBytes, remainingBytes);
                 long offset = fileLength - remainingBytes;
 
                 // Calculate the checksum for the part
                 String partChecksum = calculatePartChecksum(scratchFile, offset, bytesToUpload, digest);
 
                 final int currentPartNumber = partNumber;
-                UploadPartResponse uploadPartResponse;
-                try (InputStream partStream = openPart(scratchFile, offset, bytesToUpload)) {
-                    // Upload the part
-                    uploadPartResponse = s3AsyncClient.uploadPart(
-                            r -> r.bucket(getBucketName()).key(key).uploadId(uploadId).partNumber(currentPartNumber),
-                            AsyncRequestBody.fromInputStream(partStream, bytesToUpload, executor)).join();
-                }
+                // A file-backed body, not a stream: the SDK re-reads it from the start when it retries a
+                // part. A non-resettable stream makes any transient S3 error permanent
+                // ("Request cannot be retried, because the request stream could not be reset"), which is
+                // what the v1 SDK avoided by taking the file plus an offset.
+                UploadPartResponse uploadPartResponse = s3AsyncClient.uploadPart(
+                        r -> r.bucket(getBucketName()).key(key).uploadId(uploadId).partNumber(currentPartNumber),
+                        AsyncRequestBody.fromFile(FileRequestBodyConfiguration.builder()
+                                .path(scratchFile.toPath())
+                                .position(offset)
+                                .numBytesToRead(bytesToUpload)
+                                .build())).join();
 
                 // Collect the ETag for the part
                 completedParts.add(CompletedPart.builder()
@@ -263,33 +276,42 @@ private void uploadByParts(String key, File scratchFile) throws IOException {
             // Complete the multipart upload
             s3AsyncClient.completeMultipartUpload(r -> r.bucket(getBucketName()).key(key).uploadId(uploadId)
                     .multipartUpload(CompletedMultipartUpload.builder().parts(completedParts).build())).join();
+        } catch (IOException e) {
+            abortQuietly(key, uploadId);
+            throw e;
         } catch (SdkException | CompletionException e) {
-            log.error("Cannot upload the file by parts because: ", e);
-        } finally {
-            executor.shutdown();
+            // This used to be logged and swallowed, which let put() carry on and record a bitstream that
+            // S3 does not actually hold - silent data loss. Abort so the parts are not billed forever.
+            abortQuietly(key, uploadId);
+            throw new IOException("Multipart upload of " + key + " failed", e);
         }
     }
 
     /**
-     * Open a stream over exactly one part of the file, without loading it into memory.
+     * Abort a multipart upload, reporting but not rethrowing - the caller is already failing and the
+     * original cause is the one worth propagating.
      *
-     * @param file the uploading file
-     * @param offset the offset in the file
-     * @param length the length of the part
-     * @return a stream bounded to the requested part
-     * @throws IOException if an I/O error occurs
+     * @param key the bitstream's internalId
+     * @param uploadId the multipart upload to abort
      */
-    private static InputStream openPart(File file, long offset, long length) throws IOException {
-        FileInputStream fis = new FileInputStream(file);
+    private void abortQuietly(String key, String uploadId) {
         try {
-            fis.getChannel().position(offset);
-            return BoundedInputStream.builder().setInputStream(fis).setMaxCount(length).get();
-        } catch (IOException | RuntimeException e) {
-            fis.close();
-            throw e;
+            s3AsyncClient.abortMultipartUpload(
+                    r -> r.bucket(getBucketName()).key(key).uploadId(uploadId)).join();
+        } catch (SdkException | CompletionException e) {
+            log.error("Could not abort multipart upload " + uploadId + " for " + key
+                    + "; its parts will remain until a lifecycle rule removes them", e);
         }
     }
 
+    public long getUploadPartSizeBytes() {
+        return uploadPartSizeBytes;
+    }
+
+    public void setUploadPartSizeBytes(long uploadPartSizeBytes) {
+        this.uploadPartSizeBytes = uploadPartSizeBytes;
+    }
+
     /**
      * Calculate the checksum of the specified part of the file (Multipart upload)
      *

dspace-api/src/main/java/org/dspace/storage/bitstore/service/S3DirectDownloadService.java

@@ -26,4 +26,20 @@ public interface S3DirectDownloadService {
      */
     String generatePresignedUrl(String bucket, String key, int expirationSeconds, String bitstreamName)
             throws UnsupportedEncodingException;
+
+    /**
+     * Generate a presigned URL, serving it with a caller-supplied Content-Disposition.
+     *
+     * Without this the redirect always forced `attachment`, so turning on direct downloads silently
+     * disabled inline preview for every format in `webui.content_disposition_inline`.
+     *
+     * @param bucket                      The S3 bucket name
+     * @param key                         The bitstream path in the S3 bucket
+     * @param expirationSeconds           The number of seconds until the URL expires
+     * @param bitstreamName               The name of the bitstream, used when no override is given
+     * @param contentDispositionOverride  The exact Content-Disposition to serve, or null for `attachment`
+     * @return                            A string containing the presigned URL for direct download access
+     */
+    String generatePresignedUrl(String bucket, String key, int expirationSeconds, String bitstreamName,
+                                String contentDispositionOverride) throws UnsupportedEncodingException;
 }

dspace-api/src/main/java/org/dspace/util/ContentDispositionUtils.java

@@ -0,0 +1,57 @@
+/**
+ * The contents of this file are subject to the license and copyright
+ * detailed in the LICENSE and NOTICE files at the root of the source
+ * tree and available online at
+ *
+ * http://www.dspace.org/license/
+ */
+package org.dspace.util;
+
+import java.net.URLEncoder;
+import java.nio.charset.StandardCharsets;
+
+/**
+ * Builds a `Content-Disposition` header value from a bitstream name.
+ *
+ * The fork had two independent implementations of this, one of them wrong: it escaped `"` but not `\`,
+ * so a bitstream named `evil\` terminated the quoted string early and swallowed the `filename*`
+ * parameter; and it used {@link URLEncoder} directly, which encodes a space as `+`. RFC 8187 treats
+ * `+` as a literal character, so `my report.pdf` arrived as `my+report.pdf`.
+ *
+ * @author Milan Majchrak (dspace at dataquest.sk)
+ */
+public final class ContentDispositionUtils {
+
+    public static final String ATTACHMENT = "attachment";
+    public static final String INLINE = "inline";
+
+    private ContentDispositionUtils() {
+    }
+
+    /**
+     * Build a `Content-Disposition` value carrying both the RFC 6266 ASCII fallback and the RFC 8187
+     * percent-encoded UTF-8 name.
+     *
+     * @param disposition `attachment` or `inline`
+     * @param name        the bitstream name; must not be null
+     * @return the header value
+     */
+    public static String build(String disposition, String name) {
+        if (name == null) {
+            throw new IllegalArgumentException("Bitstream name cannot be null");
+        }
+
+        // RFC 8187 percent-encoding for filename*. URLEncoder is form-encoding, so `+` has to be
+        // converted back to `%20` - a literal `+` in a filename is already encoded as `%2B` by then.
+        String encoded = URLEncoder.encode(name, StandardCharsets.UTF_8)
+                .replace("+", "%20");
+
+        // ASCII fallback for clients that ignore filename*. Non-ASCII becomes `_`; backslash and quote
+        // are escaped, in that order, so the quoted-string cannot be terminated early.
+        String asciiFallback = name.replaceAll("[^\\x20-\\x7E]", "_")
+                .replace("\\", "\\\\")
+                .replace("\"", "\\\"");
+
+        return String.format("%s; filename=\"%s\"; filename*=UTF-8''%s", disposition, asciiFallback, encoded);
+    }
+}