本文使用「署名 4.0 国际 (CC BY 4.0)」许可协议，欢迎转载、或重新修改使用，但需要注明来源。 [署名 4.0 国际 (CC BY 4.0)](https://creativecommons.org/licenses/by/4.0/deed.zh)

本文作者: 苏洋

创建时间: 2026年10月05日
统计字数: 23638字
阅读时间: 48分钟阅读
本文链接: https://soulteary.com/2026/10/05/continuing-minio-otterio-early-october-maintenance-notes.html

-----


# 接力 MinIO：OtterIO 十月初维护记录

时隔一个季度，对 OtterIO 进行了一些维护工作，应该对一些有需求的朋友有用，那么就记录下来吧。

## 写在前面

六月整理 OtterIO 时（项目地址 [soulteary/otterio](https://github.com/soulteary/otterio)），我主要处理了代码基线、功能裁剪和部署迁移。当完成 HTTP 路由调整、依赖升级和一轮历史安全问题检查后，项目也就有了容器镜像和二进制发布。

OtterIO 是一个 S3 兼容对象存储服务，基于 MinIO 最后一个 Apache 2.0 release 附近的代码，起点约为 `RELEASE.2021-04-22T15-44-28Z`。项目继续以 Apache License 2.0 分发，保留原始版权与来源说明（[项目说明](https://github.com/soulteary/otterio/blob/be8596f0d69d530586f35366fb2d5c79bdc54399/README.md)）。

当前这轮维护中，一个安全反馈暴露了预签名上传的问题：接收方可能通过附加未签名的请求头，把上传改成服务端复制。分布式存储接口也补上了路径和元数据检查。

** 如果你还在使用六月份发布的软件，记得升级容器镜像的版本。**

本文以北京时间 2026 年 10 月 5 日发布的 [`RELEASE.2026-10-04T22-12-15Z`](https://github.com/soulteary/otterio/releases/tag/RELEASE.2026-10-04T22-12-15Z) 为准，并与六月七日的正式版本对照。版本标签使用 UTC，所以日期仍是十月四日。

之前的整理过程，记录在这三篇文章里：

- [重新审视 MinIO：许可证、归档、社区 fork 与我的 Apache 2.0 基线](https://soulteary.com/2026/06/07/revisiting-minio-license-archive-community-forks-and-my-apache-2-baseline.html)
- [从 MinIO 到 OtterIO：整理一条 Apache 2.0 开源对象存储代码线](https://soulteary.com/2026/06/08/from-minio-to-otterio-curating-an-apache-2-object-storage-codeline.html)
- [把 MinIO 示例迁到 OtterIO：使用、部署与迁移验证](https://soulteary.com/2026/06/09/migrate-minio-examples-to-otterio-usage-deployment-and-verification.html)

## 修复 SigV4 请求头校验

感谢 Act Security 的 Oren Yomtov 报告这个问题，并提供详细分析和复现用例（[问题与修复说明](https://github.com/soulteary/otterio/blob/be8596f0d69d530586f35366fb2d5c79bdc54399/docs/security/sigv4-header-coverage.md)）。

预签名 URL 常用于让用户直接上传对象：应用服务端生成带签名的 URL，指定上传位置和有效期，再把它交给用户。用户凭这个 URL 就能上传，不需要拿到服务端的访问密钥。

受影响的代码只检查签名中列出的请求头，没有拦住额外附加的操作请求头。例如，拿到上传 URL 的人可以加上未签名的 `x-amz-copy-source`，指定一个源对象，让服务端把上传改成复制。服务端随后会使用生成签名的身份所拥有的读取权限，把源对象复制到 URL 指定的位置。

复制目的地仍由签名 URL 限定；要读取复制后的内容，还需要具备目的地的读取权限。六月七日的正式版已确认受影响，使用这个版本的用户需要升级到包含修复的版本。

修复后，服务端会检查请求中的 `x-amz-*` 和 `x-otterio-*` 请求头。决定或修改 S3 操作的请求头必须受到签名保护，如果这些操作请求头未被签名覆盖，服务端会返回 HTTP 403 AccessDenied。这项检查同时用于预签名 URL、通过 Authorization 请求头认证的 SigV4 请求，以及流式上传的初始签名。

这里保留了协议允许的写法。`x-amz-content-sha256` 不一定要列入 `SignedHeaders`，因为它的有效值已经作为 payload hash 参与签名。同一参数也可以同时出现在已签名的 URL 查询参数和请求头里，但两处的值必须完全一致，包括多值的数量和顺序。

如果自己封装了 S3 客户端，需要在生成签名前设置好这些操作请求头。先签名、再附加复制来源等请求头的写法，会被新版拒绝，应调整客户端的签名流程。合法复制仍然可用，源对象的 IAM 权限检查也继续保留。

回归测试使用独立的 S3 SDK 生成签名，再通过实际使用的 Fiber 路由，分别在 FS 和纠删码存储上执行请求。测试确认普通上传和合法复制继续可用，附加未签名复制头的请求会被拒绝，分片上传的数据也不会被这种请求改变。只具备写入权限的签名身份，即使正确签名了复制请求，也不能绕过源对象的读取权限检查（[回归测试](https://github.com/soulteary/otterio/blob/be8596f0d69d530586f35366fb2d5c79bdc54399/cmd/signature-v4-copy_security_test.go)）。

## 检查内部存储接口的路径和元数据

OtterIO 为分布式纠删码部署注册了一组内部 storage REST 接口，供节点之间读写数据。这些接口使用由集群 root 凭据签名的 JWT 进行认证，匿名 S3 请求或有限权限的 S3 访问密钥不足以调用。

即使请求通过认证，传来的路径和元数据仍然需要检查。有内部调用权限的节点也可能发来异常输入，导致操作越过存储目录，或让接收方因畸形元数据而崩溃。

这次参考 SILO SN-2026-002 公告，核查了 OtterIO 继承的同类缺陷，并独立实现相应校验。两条代码线的接口和传输方式不同，修复范围以 OtterIO 的实际实现为准（[内部存储安全说明](https://github.com/soulteary/otterio/blob/be8596f0d69d530586f35366fb2d5c79bdc54399/docs/security/sn-2026-002-storage-hardening.md)）。

### 在拼接路径之前检查输入

路径清理和拼接会消掉原始输入中的部分路径段。如果等到这些处理完成后再检查，可能已经看不出输入里的问题。新版先检查未经规范化的路径，再进行拼接；检查范围也不只包括 URL 参数，还包括元数据请求体中的对象名和数据目录。

重命名时，来源和目标路径都要检查。批量建卷和删除版本时，则先检查整个批次，避免已经修改一部分数据，才发现后面的路径不合法。

删除目录时，还要正确判断目录之间的包含关系。例如，`backup` 和 `backup-old` 名称相近，但后者并不是前者的子目录。新版按路径段判断范围，避免仅凭字符串前缀做出错误判断。已有的嵌套系统目录、目录修复路径和根目录列举等合法用法仍然保留（[存储校验实现](https://github.com/soulteary/otterio/blob/be8596f0d69d530586f35366fb2d5c79bdc54399/cmd/storage-validation.go)）。

这些检查针对的是路径字符串，不能替代操作系统对数据目录的保护，也不能保证防住符号链接或并发文件系统改动造成的目录逃逸。数据目录仍应由服务控制，节点通信也应保持隔离。

### 检查元数据的数值和结构

纠删码参数、分片数组和对象长度会用于索引、大小计算和完整性判断。数组长度不一致可能导致越界访问，无效或溢出的长度可能触发异常内存分配，错误的大小信息还可能把截断的分片误判为健康。新版在使用这些值前增加了校验，并拒绝不支持的 bitrot 校验算法。

仅限制请求体大小还不够。MessagePack 可以用很少的字节声明一个很大的数组或映射，解码器如果直接按声明分配内存，小请求也可能造成大量内存占用。因此，新版在调用正式解码器之前，先检查数组、映射的元素数量和嵌套深度。

对内部缓冲读写和 `FileInfo` 元数据请求体，还增加了单次 **64 MiB** 的大小限制。这项限制不影响 S3 对象大小，大对象继续使用已有的流式读写路径。自行实现内部客户端时，需要拆分过大的缓冲操作和版本删除批次。

回归测试会用有效的节点 JWT 调用内部 REST 接口，并确认正常写入能够成功。在此基础上，再验证异常路径和元数据会被拒绝，确保这些请求确实经过了输入校验（[存储安全回归](https://github.com/soulteary/otterio/blob/be8596f0d69d530586f35366fb2d5c79bdc54399/cmd/storage-security_test.go)）。

升级前应备份元数据，升级后再验证已有对象的读取、修复和复制。新版会把不合法的历史元数据视为损坏数据并拒绝处理，不会自动改写或删除；遇到这类问题，需要调查原因，并从已知良好的副本恢复。

分布式部署需要更新所有节点。

## 新版容器要求显式配置凭据

六月八日那篇文章里，容器启动示例省略了凭据。新版在运行 `server` 和 `gateway` 前，会先检查用户名和密码是否完整配置，并拒绝使用默认密码。因此，旧命令需要补上凭据才能继续使用（[容器升级说明](https://github.com/soulteary/otterio/blob/be8596f0d69d530586f35366fb2d5c79bdc54399/README_DOCKER_SECURITY.md)）。

已经配置自定义凭据的部署，可以继续沿用原值；六月九日文章中的显式凭据示例也仍然适用。升级前需要确认这些值已经传入容器。

隔离的本地演示仍可通过 `OTTERIO_ALLOW_DEFAULT_CREDENTIALS=1` 使用默认凭据，但这个开关不会接受只提供用户名或密码的不完整配置。

新建本地开发或测试环境，可以先生成一组凭据，再传给固定版本的镜像。下面的例子用 OpenSSL 生成密码，S3 和控制台端口都绑定到 `127.0.0.1`：

```bash
mkdir -p ./data

export OTTERIO_ROOT_USER=otterio-admin
export OTTERIO_ROOT_PASSWORD="$(openssl rand -hex 32)"

docker run -d \
  --name otterio \
  -p 127.0.0.1:9000:9000 \
  -p 127.0.0.1:9001:9001 \
  -e OTTERIO_ROOT_USER \
  -e OTTERIO_ROOT_PASSWORD \
  -v "$PWD/data:/data" \
  ghcr.io/soulteary/otterio:RELEASE.2026-10-04T22-12-15Z \
  server --address ":9000" --console-address ":9001" /data
```

启动后，S3 接口地址是 `http://127.0.0.1:9000`，控制台地址是 `http://127.0.0.1:9001`。生成的用户名和密码要保存，后续连接或重建容器时继续使用。上面的密码生成步骤用于新建环境，已有部署升级时应保留原来的凭据。如果容器名或端口已经被占用，需要修改示例中的对应值。

### 凭据文件和目录权限

用户名和密码可以分别从环境变量或文件加载。例如，用户名通过 `OTTERIO_ROOT_USER` 设置，密码通过 `OTTERIO_ROOT_PASSWORD_FILE` 指定的容器内文件加载，这种组合可以正常使用。

如果要求启动时必须读到某个凭据文件，应通过 `_FILE` 指定它在容器内的绝对路径，并确保文件可读、为普通文件且内容非空。同一项配置同时提供非空环境值和已有文件，会报冲突。

新变量 `OTTERIO_ROOT_USER` / `OTTERIO_ROOT_PASSWORD` 优先于旧变量 `OTTERIO_ACCESS_KEY` / `OTTERIO_SECRET_KEY`，但必须成对使用。不能从新、旧两组变量中各取一项，拼成一组凭据。

凭据检查由容器入口脚本执行，直接运行裸机二进制时没有这项强制要求。镜像默认 UID 也没有改变，已有数据卷的归属不会自动调整。

需要以非 root 用户运行时，可以使用[安全 Compose 示例](https://github.com/soulteary/otterio/blob/be8596f0d69d530586f35366fb2d5c79bdc54399/docker-compose.secure.yml)。这份配置指定了非 root UID/GID，使用只读根文件系统，限制临时目录大小，移除 capabilities，并禁止提权。使用前应确保选定的容器用户能够写入数据目录、读取凭据文件。

## 补充 Windows 启动和并发回归测试

六月八日那篇文章里，Windows CI 主要验证代码能否编译。后来社区反馈了目录枚举和服务启动的问题，并提供分析和补丁建议。相应修复完成后，CI 也补上了回归测试（[贡献记录](https://github.com/soulteary/otterio/blob/be8596f0d69d530586f35366fb2d5c79bdc54399/ACKNOWLEDGMENTS.md)）。

现在，Windows CI 会执行选定的目录回归用例，并实际启动编译出来的服务，验证 S3 读写。完整的 Go 测试仍在 Linux 上运行，其中包含依赖 POSIX 文件系统语义的用例（参考[当前 Go workflow](https://github.com/soulteary/otterio/blob/be8596f0d69d530586f35366fb2d5c79bdc54399/.github/workflows/go.yml)）。

这轮还修复了流式响应中的并发读写问题。HTTP 路由继续使用 Fiber v3，处理协程和发送侧需要交接响应头、状态码等信息。新版在首次提交响应时固定一份快照，发送侧只读取这份结果，后续修改不会影响已提交的响应。

响应提交前发生的异常会交回原有错误处理流程，提交后的异常则中断数据流；响应内容仍然边写边传。对应测试覆盖提交前后的异常、部分数据已经写出，以及读取端提前退出等情况（[流式响应实现](https://github.com/soulteary/otterio/blob/be8596f0d69d530586f35366fb2d5c79bdc54399/cmd/fiber_stream_writer.go)）。

源码构建的 Go 最低版本由 1.26 提升到 1.27.1。服务端依赖和前端锁文件一并更新，生成代码及嵌入服务中的控制台产物也已重新构建。直接运行发布镜像或二进制时，无需安装 Go 或前端构建工具（[依赖与兼容记录](https://github.com/soulteary/otterio/blob/be8596f0d69d530586f35366fb2d5c79bdc54399/docs/development/dependency-upgrade-20261004.md)）。

Linux 上的 Race 检查会先列出需要测试的包，再运行这些包中的全部用例，不额外筛选测试。`-count=1` 用来避免复用之前成功的测试结果。如果获取包列表失败，任务会直接停止，避免漏测后仍报告通过。

## 再次核对发布出去的镜像和二进制

源码通过测试之后，我们还要确认发布的是同一份代码。六月已有镜像、二进制和校验和。这次发布会核对待发布的源码提交，实际运行已推送的镜像，并把上传的附件重新下载比对。

本次正式版提供 Linux amd64、arm64、ppc64le，macOS amd64、arm64，以及 Windows amd64 六种系统与架构组合的二进制，附带校验文件和发布清单（[版本下载](https://github.com/soulteary/otterio/releases/tag/RELEASE.2026-10-04T22-12-15Z)）。

### 本地 Docker 构建使用当前源码

原来的 Dockerfile 会在构建时克隆远端仓库，再切到 `main` 编译。当时的策略是，始终以合并到 main 的代码为准。所以，本地修改不会进入这条构建路径，生成的镜像可能与准备测试的源码不同。

现在，本地 Docker 构建直接使用传入的源码上下文，包含本地修改和已提交的控制台资源。来源提交号默认为 `unknown`，可通过 `VCS_REF` 显式标注；这个参数只记录来源，不会切换构建使用的代码。正式发布仍从固定提交编译二进制，再由 `Dockerfile.ci` 打包镜像（[本地源码构建说明](https://github.com/soulteary/otterio/blob/be8596f0d69d530586f35366fb2d5c79bdc54399/docs/development/docker-build.md)）。

### 先验证固定版本再更新稳定标签

现在的发布流程按下面的步骤执行：

1. 确认针对该源码提交运行的 Go、Lint 和 Release checks 三个工作流都已完成且成功；这些工作流由 `main` 分支推送触发。
2. 从这个固定提交构建二进制和版本镜像，暂不更新稳定标签。
3. 按 digest 拉取已经推送到 GHCR 的镜像，在 Linux amd64 上运行启动和 S3 读写冒烟验证。
4. 将二进制、校验文件和发布清单上传到 GitHub 草稿 Release，逐个下载并与原产物比对。比对通过后公开这个版本，仍保留现有的 `latest` 指向。
5. 在串行流程中核对已发布版本的源码提交、各仓库固定版本标签指向的镜像 digest，并比较版本先后，阻止较旧版本覆盖当前的 `latest`。通过检查后，用已验证的 digest 更新稳定镜像标签，最后更新 GitHub latest。

固定版本的公开发布和稳定标签更新共用并发锁，这两类任务会串行执行，避免彼此在核对与更新之间改变版本状态。镜像标签更新后，还会再次核对 digest（[发布工作流](https://github.com/soulteary/otterio/blob/be8596f0d69d530586f35366fb2d5c79bdc54399/.github/workflows/release.yml)）。

新增的 `release-manifest.json` 记录了发布标签、源码提交和镜像 digest，可以用来核对运行中的版本。使用镜像时可以按 digest 固定内容，下载二进制后则用附带的 SHA256 文件校验（[本次发布清单](https://github.com/soulteary/otterio/releases/download/RELEASE.2026-10-04T22-12-15Z/release-manifest.json)）。

清单并不提供发布签名，也不证明可以逐字节复现同一次构建。GitHub、GHCR 和 Docker Hub 的更新也分步完成，不能作为一个原子事务提交。某一步失败后，应先核对各平台已经完成的状态，再恢复对应步骤。

本版的 Go 检查和正式发布流程均已成功执行，包含 Race、平台检查、附件比对和稳定标签更新。发布记录中的镜像运行验证覆盖 Linux amd64，其他镜像架构仍需单独验证（[Go 检查记录](https://github.com/soulteary/otterio/actions/runs/37237904482)、[正式发布记录](https://github.com/soulteary/otterio/actions/runs/37239122357)）。

## 升级后还要验证自己的应用

六月的迁移实验工具会生成随机对象，记录源端大小和 SHA256，再从目标端下载校验，检查两端内容是否一致。升级后仍可沿用这个方法，并结合应用的实际用法检查：

- 验证应用的上传、下载和删除流程，并抽查已有对象能否读取。
- 确认未签名操作头会被拒绝，合法签名复制仍然可用，缺少源对象读取权限时复制失败。
- 验证分片上传，以及超过 64 MiB 对象的流式读写。
- 检查环境变量和 secret 文件是否正确加载、数据目录权限是否匹配，并在实际使用的平台上启动服务。
- 分布式部署还要确认所有节点的版本，并检查已有数据的修复和复制。

[之前的一篇文章里](https://soulteary.com/2026/06/09/migrate-minio-examples-to-otterio-usage-deployment-and-verification.html#%E4%BD%BF%E7%94%A8-go--aws-s3-sdk-v2-%E5%81%9A%E8%BF%81%E7%A7%BB%E9%AA%8C%E8%AF%81)，已经验证过 MinIO 不同衍生版本之间的数据迁移。这次只测试 OtterIO 六月版到十月版的迁移：

```bash
2026/10/05 06:00:06 SYNC SUMMARY
2026/10/05 06:00:06   source endpoints: 1
2026/10/05 06:00:06   objects per source: 12
2026/10/05 06:00:06   expected objects: 12
2026/10/05 06:00:06   uploaded source objects: 12
2026/10/05 06:00:06   verified target objects: 12
2026/10/05 06:00:06   target endpoint: http://otterio-261004:9000
2026/10/05 06:00:06   target bucket: migration-test
2026/10/05 06:00:06   target from/ objects: 12
2026/10/05 06:00:06   target from/otterio-260607/ objects: 12
2026/10/05 06:00:06   result: ALL OBJECTS SYNCED AND VERIFIED
```

### 测试程序

测试使用的环境配置：

```bash
ROOT_USER=admin
ROOT_PASSWORD=otterio123456
```

容器配置文件(`docker-compose.yaml`)：

```yaml
services:
  otterio-261004:
    image: soulteary/otterio:RELEASE.2026-10-04T22-12-15Z
    container_name: otterio-261004
    command: server /data
    environment:
      OTTERIO_ROOT_USER: ${ROOT_USER}
      OTTERIO_ROOT_PASSWORD: ${ROOT_PASSWORD}
    volumes:
      - ./data/otterio-261004:/data
    ports:
      - "9101:9000"
    networks:
      - object-storage-lab
    restart: unless-stopped

  otterio-260607:
    image: soulteary/otterio:RELEASE.2026-06-07T11-32-46Z
    container_name: otterio-260607
    command: server /data
    environment:
      OTTERIO_ROOT_USER: ${ROOT_USER}
      OTTERIO_ROOT_PASSWORD: ${ROOT_PASSWORD}
    volumes:
      - ./data/otterio-260607:/data
    ports:
      - "9102:9000"
    networks:
      - object-storage-lab
    restart: unless-stopped

networks:
  object-storage-lab:
    name: object-storage-lab
    driver: bridge
```

测试程序：

```go
package main

import (
	"bytes"
	"context"
	"crypto/rand"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"flag"
	"fmt"
	"io"
	"log"
	"math/big"
	"mime"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/config"
	awscred "github.com/aws/aws-sdk-go-v2/credentials"
	"github.com/aws/aws-sdk-go-v2/service/s3"
	s3types "github.com/aws/aws-sdk-go-v2/service/s3/types"
)

type Endpoint struct {
	Name      string
	Endpoint  string
	AccessKey string
	SecretKey string
	Region    string
	Client    *s3.Client
}

type ObjectRecord struct {
	SourceName string
	Bucket     string
	Key        string
	Size       int64
	SHA256     string
}

type Options struct {
	Bucket      string
	TargetName  string
	ObjectCount int
	MinSize     int64
	MaxSize     int64
	Cleanup     bool
	ListObjects bool
}

func env(key, fallback string) string {
	if v := strings.TrimSpace(os.Getenv(key)); v != "" {
		return v
	}
	return fallback
}

func parseSize(s string) (int64, error) {
	original := s
	s = strings.TrimSpace(strings.ToLower(s))

	multiplier := int64(1)

	switch {
	case strings.HasSuffix(s, "kb"):
		multiplier = 1024
		s = strings.TrimSuffix(s, "kb")
	case strings.HasSuffix(s, "mb"):
		multiplier = 1024 * 1024
		s = strings.TrimSuffix(s, "mb")
	case strings.HasSuffix(s, "gb"):
		multiplier = 1024 * 1024 * 1024
		s = strings.TrimSuffix(s, "gb")
	case strings.HasSuffix(s, "b"):
		multiplier = 1
		s = strings.TrimSuffix(s, "b")
	}

	var n int64
	_, err := fmt.Sscanf(strings.TrimSpace(s), "%d", &n)
	if err != nil {
		return 0, fmt.Errorf("invalid size %q: %w", original, err)
	}

	if n < 0 {
		return 0, fmt.Errorf("invalid size %q: size must be positive", original)
	}

	return n * multiplier, nil
}

func randomInt64(min, max int64) (int64, error) {
	if max < min {
		return 0, fmt.Errorf("max size must be >= min size")
	}

	if min == max {
		return min, nil
	}

	span := max - min + 1
	if span <= 0 {
		return 0, fmt.Errorf("invalid random range: min=%d max=%d", min, max)
	}

	n, err := rand.Int(rand.Reader, big.NewInt(span))
	if err != nil {
		return 0, err
	}

	return min + n.Int64(), nil
}

func randomBytes(size int64) ([]byte, string, error) {
	if size < 0 {
		return nil, "", fmt.Errorf("size must be positive")
	}

	buf := make([]byte, size)

	if _, err := rand.Read(buf); err != nil {
		return nil, "", err
	}

	sum := sha256.Sum256(buf)

	return buf, hex.EncodeToString(sum[:]), nil
}

func sha256File(path string) (string, int64, error) {
	f, err := os.Open(path)
	if err != nil {
		return "", 0, err
	}
	defer f.Close()

	h := sha256.New()

	n, err := io.Copy(h, f)
	if err != nil {
		return "", 0, err
	}

	return hex.EncodeToString(h.Sum(nil)), n, nil
}

func newS3Client(ctx context.Context, ep Endpoint) (*s3.Client, error) {
	cfg, err := config.LoadDefaultConfig(
		ctx,
		config.WithRegion(ep.Region),
		config.WithCredentialsProvider(
			awscred.NewStaticCredentialsProvider(ep.AccessKey, ep.SecretKey, ""),
		),
	)
	if err != nil {
		return nil, err
	}

	client := s3.NewFromConfig(cfg, func(o *s3.Options) {
		o.BaseEndpoint = aws.String(ep.Endpoint)
		o.UsePathStyle = true
	})

	return client, nil
}

func ensureBucket(ctx context.Context, client *s3.Client, bucket string) error {
	_, err := client.CreateBucket(ctx, &s3.CreateBucketInput{
		Bucket: aws.String(bucket),
	})
	if err == nil {
		return nil
	}

	var owned *s3types.BucketAlreadyOwnedByYou
	if errors.As(err, &owned) {
		return nil
	}

	var exists *s3types.BucketAlreadyExists
	if errors.As(err, &exists) {
		return nil
	}

	msg := err.Error()

	if strings.Contains(msg, "BucketAlreadyOwnedByYou") ||
		strings.Contains(msg, "BucketAlreadyExists") ||
		strings.Contains(msg, "Your previous request to create the named bucket succeeded") {
		return nil
	}

	return err
}

func deleteBucketObjects(ctx context.Context, client *s3.Client, bucket string) error {
	var token *string

	for {
		out, err := client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
			Bucket:            aws.String(bucket),
			ContinuationToken: token,
		})
		if err != nil {
			if strings.Contains(err.Error(), "NoSuchBucket") {
				return nil
			}
			return err
		}

		for _, obj := range out.Contents {
			key := aws.ToString(obj.Key)
			if key == "" {
				continue
			}

			_, err := client.DeleteObject(ctx, &s3.DeleteObjectInput{
				Bucket: aws.String(bucket),
				Key:    aws.String(key),
			})
			if err != nil {
				return fmt.Errorf("delete object %s: %w", key, err)
			}

			log.Printf("[cleanup] deleted: s3://%s/%s", bucket, key)
		}

		if out.IsTruncated == nil || !*out.IsTruncated {
			break
		}

		token = out.NextContinuationToken
	}

	return nil
}

func contentTypeForKey(key string) string {
	ext := filepath.Ext(key)
	if ext == "" {
		return "application/octet-stream"
	}

	if ct := mime.TypeByExtension(ext); ct != "" {
		return ct
	}

	switch ext {
	case ".bin":
		return "application/octet-stream"
	case ".json":
		return "application/json"
	case ".txt":
		return "text/plain; charset=utf-8"
	default:
		return "application/octet-stream"
	}
}

func makeObjectKey(sourceName string, index int) string {
	now := time.Now().UTC().Format("20060102T150405Z")

	patterns := []string{
		"small/%s-%03d-random.txt",
		"medium/%s-%03d-random.bin",
		"nested/year=2026/month=06/day=09/%s-%03d-file.txt",
		"unicode/%s-%03d-中文文件名.txt",
		"unicode/%s-%03d-空 格 文件.txt",
		"metadata/%s-%03d-content-type.json",
	}

	pattern := patterns[(index-1)%len(patterns)]

	return fmt.Sprintf(pattern, now+"-"+sourceName, index)
}

func uploadRandomObjects(ctx context.Context, ep Endpoint, opt Options) ([]ObjectRecord, error) {
	log.Printf("[%s] ensure bucket: %s", ep.Name, opt.Bucket)

	if err := ensureBucket(ctx, ep.Client, opt.Bucket); err != nil {
		return nil, fmt.Errorf("[%s] create bucket: %w", ep.Name, err)
	}

	records := make([]ObjectRecord, 0, opt.ObjectCount)

	for i := 1; i <= opt.ObjectCount; i++ {
		size, err := randomInt64(opt.MinSize, opt.MaxSize)
		if err != nil {
			return nil, err
		}

		key := makeObjectKey(ep.Name, i)

		data, digest, err := randomBytes(size)
		if err != nil {
			return nil, err
		}

		_, err = ep.Client.PutObject(ctx, &s3.PutObjectInput{
			Bucket:       aws.String(opt.Bucket),
			Key:          aws.String(key),
			Body:         bytes.NewReader(data),
			ContentLength: aws.Int64(size),
			ContentType:   aws.String(contentTypeForKey(key)),
			Metadata: map[string]string{
				"source": ep.Name,
				"sha256": digest,
			},
		})
		if err != nil {
			return nil, fmt.Errorf("[%s] put object %s: %w", ep.Name, key, err)
		}

		log.Printf("[%s] uploaded: s3://%s/%s size=%d sha256=%s",
			ep.Name,
			opt.Bucket,
			key,
			size,
			digest,
		)

		records = append(records, ObjectRecord{
			SourceName: ep.Name,
			Bucket:     opt.Bucket,
			Key:        key,
			Size:       size,
			SHA256:     digest,
		})
	}

	return records, nil
}

func downloadToTempFile(ctx context.Context, client *s3.Client, bucket, key string) (string, error) {
	out, err := client.GetObject(ctx, &s3.GetObjectInput{
		Bucket: aws.String(bucket),
		Key:    aws.String(key),
	})
	if err != nil {
		return "", err
	}
	defer out.Body.Close()

	tmp, err := os.CreateTemp("", "s3-migrate-*")
	if err != nil {
		return "", err
	}
	defer tmp.Close()

	if _, err := io.Copy(tmp, out.Body); err != nil {
		_ = os.Remove(tmp.Name())
		return "", err
	}

	return tmp.Name(), nil
}

func uploadFile(
	ctx context.Context,
	client *s3.Client,
	bucket string,
	key string,
	path string,
	sourceName string,
	digest string,
	size int64,
) error {
	f, err := os.Open(path)
	if err != nil {
		return err
	}
	defer f.Close()

	_, err = client.PutObject(ctx, &s3.PutObjectInput{
		Bucket:       aws.String(bucket),
		Key:          aws.String(key),
		Body:         f,
		ContentLength: aws.Int64(size),
		ContentType:   aws.String(contentTypeForKey(key)),
		Metadata: map[string]string{
			"source":        sourceName,
			"source-sha256": digest,
		},
	})

	return err
}

func syncOneObject(ctx context.Context, source Endpoint, target Endpoint, record ObjectRecord, opt Options) error {
	targetKey := fmt.Sprintf("from/%s/%s", source.Name, record.Key)

	tmpPath, err := downloadToTempFile(ctx, source.Client, record.Bucket, record.Key)
	if err != nil {
		return fmt.Errorf("[%s] download %s: %w", source.Name, record.Key, err)
	}
	defer os.Remove(tmpPath)

	digest, size, err := sha256File(tmpPath)
	if err != nil {
		return fmt.Errorf("[%s] sha256 temp file %s: %w", source.Name, record.Key, err)
	}

	if size != record.Size || digest != record.SHA256 {
		return fmt.Errorf(
			"[%s] local verify failed key=%s source_size=%d local_size=%d source_sha=%s local_sha=%s",
			source.Name,
			record.Key,
			record.Size,
			size,
			record.SHA256,
			digest,
		)
	}

	if err := uploadFile(ctx, target.Client, opt.Bucket, targetKey, tmpPath, source.Name, digest, size); err != nil {
		return fmt.Errorf("[otterio] upload %s: %w", targetKey, err)
	}

	log.Printf("[%s -> %s] synced: s3://%s/%s size=%d sha256=%s",
		source.Name,
		target.Name,
		opt.Bucket,
		targetKey,
		size,
		digest,
	)

	return nil
}

func verifyTargetObject(ctx context.Context, target Endpoint, opt Options, record ObjectRecord) error {
	targetKey := fmt.Sprintf("from/%s/%s", record.SourceName, record.Key)

	tmpPath, err := downloadToTempFile(ctx, target.Client, opt.Bucket, targetKey)
	if err != nil {
		return fmt.Errorf("[verify] download target %s: %w", targetKey, err)
	}
	defer os.Remove(tmpPath)

	digest, size, err := sha256File(tmpPath)
	if err != nil {
		return fmt.Errorf("[verify] sha256 target %s: %w", targetKey, err)
	}

	if size != record.Size || digest != record.SHA256 {
		return fmt.Errorf(
			"[verify] mismatch target=%s source_size=%d target_size=%d source_sha=%s target_sha=%s",
			targetKey,
			record.Size,
			size,
			record.SHA256,
			digest,
		)
	}

	log.Printf("[verify] ok: s3://%s/%s size=%d sha256=%s",
		opt.Bucket,
		targetKey,
		size,
		digest,
	)

	return nil
}

func countObjectsWithPrefix(ctx context.Context, client *s3.Client, bucket, prefix string) (int, error) {
	var token *string
	count := 0

	for {
		out, err := client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
			Bucket:            aws.String(bucket),
			Prefix:            aws.String(prefix),
			ContinuationToken: token,
		})
		if err != nil {
			return 0, err
		}

		count += len(out.Contents)

		if out.IsTruncated == nil || !*out.IsTruncated {
			break
		}

		token = out.NextContinuationToken
	}

	return count, nil
}

func listTargetObjects(ctx context.Context, target Endpoint, bucket string) error {
	var token *string
	var keys []string

	for {
		out, err := target.Client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
			Bucket:            aws.String(bucket),
			ContinuationToken: token,
		})
		if err != nil {
			return err
		}

		for _, obj := range out.Contents {
			keys = append(keys, fmt.Sprintf("%s  %d", aws.ToString(obj.Key), aws.ToInt64(obj.Size)))
		}

		if out.IsTruncated == nil || !*out.IsTruncated {
			break
		}

		token = out.NextContinuationToken
	}

	sort.Strings(keys)

	log.Printf("[otterio] object list:")

	for _, key := range keys {
		log.Printf("  %s", key)
	}

	return nil
}

func printSyncSummary(
	ctx context.Context,
	target Endpoint,
	sources []Endpoint,
	opt Options,
	allRecords []ObjectRecord,
	verifiedCount int,
) error {
	expected := len(sources) * opt.ObjectCount

	log.Printf("")
	log.Printf("SYNC SUMMARY")
	log.Printf("  source endpoints: %d", len(sources))
	log.Printf("  objects per source: %d", opt.ObjectCount)
	log.Printf("  expected objects: %d", expected)
	log.Printf("  uploaded source objects: %d", len(allRecords))
	log.Printf("  verified target objects: %d", verifiedCount)
	log.Printf("  target endpoint: %s", target.Endpoint)
	log.Printf("  target bucket: %s", opt.Bucket)

	if len(allRecords) != expected {
		return fmt.Errorf("uploaded object count mismatch: expected=%d actual=%d", expected, len(allRecords))
	}

	if verifiedCount != expected {
		return fmt.Errorf("verified object count mismatch: expected=%d actual=%d", expected, verifiedCount)
	}

	totalFromPrefixCount, err := countObjectsWithPrefix(ctx, target.Client, opt.Bucket, "from/")
	if err != nil {
		return fmt.Errorf("count target objects: %w", err)
	}

	log.Printf("  target from/ objects: %d", totalFromPrefixCount)

	if totalFromPrefixCount != expected {
		return fmt.Errorf("target object count mismatch under from/: expected=%d actual=%d", expected, totalFromPrefixCount)
	}

	for _, source := range sources {
		prefix := fmt.Sprintf("from/%s/", source.Name)

		count, err := countObjectsWithPrefix(ctx, target.Client, opt.Bucket, prefix)
		if err != nil {
			return fmt.Errorf("count target prefix %s: %w", prefix, err)
		}

		log.Printf("  target %s objects: %d", prefix, count)

		if count != opt.ObjectCount {
			return fmt.Errorf("target prefix count mismatch prefix=%s expected=%d actual=%d", prefix, opt.ObjectCount, count)
		}
	}

	log.Printf("  result: ALL OBJECTS SYNCED AND VERIFIED")

	return nil
}

func main() {
	var (
		bucket      string
		objectCount int
		minSizeText string
		maxSizeText string
		cleanup     bool
		listObjects bool
	)

	flag.StringVar(&bucket, "bucket", env("S3_TEST_BUCKET", "migration-test"), "bucket name")
	flag.IntVar(&objectCount, "count", 12, "random object count per source")
	flag.StringVar(&minSizeText, "min-size", "1kb", "min random object size, e.g. 1kb, 1mb")
	flag.StringVar(&maxSizeText, "max-size", "10mb", "max random object size, e.g. 10mb, 100mb")
	flag.BoolVar(&cleanup, "cleanup", false, "delete all objects in test bucket before running")
	flag.BoolVar(&listObjects, "list", false, "list all objects in target bucket after sync")
	flag.Parse()

	minSize, err := parseSize(minSizeText)
	if err != nil {
		log.Fatal(err)
	}

	maxSize, err := parseSize(maxSizeText)
	if err != nil {
		log.Fatal(err)
	}

	if objectCount <= 0 {
		log.Fatal("count must be greater than 0")
	}

	if maxSize < minSize {
		log.Fatal("max-size must be greater than or equal to min-size")
	}

	opt := Options{
		Bucket:      bucket,
		TargetName:  "otterio-261004",
		ObjectCount: objectCount,
		MinSize:     minSize,
		MaxSize:     maxSize,
		Cleanup:     cleanup,
		ListObjects: listObjects,
	}

	accessKey := env("S3_ACCESS_KEY", "admin")
	secretKey := env("S3_SECRET_KEY", "otterio123456")
	region := env("S3_REGION", "us-east-1")

	endpoints := []Endpoint{
		{
			Name:      "otterio-260607",
			Endpoint:  env("OTTERIO_0607_ENDPOINT", "http://127.0.0.1:9102"),
			AccessKey: accessKey,
			SecretKey: secretKey,
			Region:    region,
		},
		{
			Name:      "otterio-261004",
			Endpoint:  env("OTTERIO_1004_ENDPOINT", "http://127.0.0.1:9101"),
			AccessKey: accessKey,
			SecretKey: secretKey,
			Region:    region,
		},
	}

	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute)
	defer cancel()

	for i := range endpoints {
		client, err := newS3Client(ctx, endpoints[i])
		if err != nil {
			log.Fatalf("[%s] create client: %v", endpoints[i].Name, err)
		}

		endpoints[i].Client = client

		log.Printf("[%s] endpoint=%s", endpoints[i].Name, endpoints[i].Endpoint)
	}

	var target Endpoint
	var sources []Endpoint

	for _, ep := range endpoints {
		if ep.Name == opt.TargetName {
			target = ep
		} else {
			sources = append(sources, ep)
		}
	}

	if target.Client == nil {
		log.Fatalf("target endpoint %q not found", opt.TargetName)
	}

	log.Printf("[otterio] ensure target bucket: %s", opt.Bucket)

	if err := ensureBucket(ctx, target.Client, opt.Bucket); err != nil {
		log.Fatalf("[otterio] create bucket: %v", err)
	}

	if opt.Cleanup {
		log.Printf("[cleanup] delete all objects in bucket %s on all endpoints", opt.Bucket)

		for _, ep := range endpoints {
			if err := deleteBucketObjects(ctx, ep.Client, opt.Bucket); err != nil {
				log.Fatalf("[%s] cleanup failed: %v", ep.Name, err)
			}
		}
	}

	var allRecords []ObjectRecord

	for _, source := range sources {
		records, err := uploadRandomObjects(ctx, source, opt)
		if err != nil {
			log.Fatal(err)
		}

		allRecords = append(allRecords, records...)
	}

	for _, source := range sources {
		for _, record := range allRecords {
			if record.SourceName != source.Name {
				continue
			}

			if err := syncOneObject(ctx, source, target, record, opt); err != nil {
				log.Fatal(err)
			}
		}
	}

	verifiedCount := 0

	for _, record := range allRecords {
		if err := verifyTargetObject(ctx, target, opt, record); err != nil {
			log.Fatal(err)
		}

		verifiedCount++
	}

	if opt.ListObjects {
		if err := listTargetObjects(ctx, target, opt.Bucket); err != nil {
			log.Fatal(err)
		}
	}

	if err := printSyncSummary(ctx, target, sources, opt, allRecords, verifiedCount); err != nil {
		log.Fatal(err)
	}
}
```

## 最后

这轮先折腾到这里。有需要的朋友可以更新试试，也欢迎带上版本号、部署方式和复现步骤来提 issue。

朴素的对象存储，还是安安静静地干活就好了。

—EOF